You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

296 lines
8.3 KiB

7 years ago
12 years ago
12 years ago
10 years ago
  1. package storage
  2. import (
  3. "fmt"
  4. "strconv"
  5. "strings"
  6. "github.com/chrislusf/seaweedfs/weed/glog"
  7. "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
  8. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  9. )
  10. const (
  11. MAX_TTL_VOLUME_REMOVAL_DELAY = 10 // 10 minutes
  12. )
  13. /*
  14. * A VolumeServer contains one Store
  15. */
  16. type Store struct {
  17. Ip string
  18. Port int
  19. PublicUrl string
  20. Locations []*DiskLocation
  21. dataCenter string //optional informaton, overwriting master setting if exists
  22. rack string //optional information, overwriting master setting if exists
  23. connected bool
  24. VolumeSizeLimit uint64 //read from the master
  25. Client master_pb.Seaweed_SendHeartbeatClient
  26. NeedleMapType NeedleMapType
  27. }
  28. func (s *Store) String() (str string) {
  29. str = fmt.Sprintf("Ip:%s, Port:%d, PublicUrl:%s, dataCenter:%s, rack:%s, connected:%v, volumeSizeLimit:%d", s.Ip, s.Port, s.PublicUrl, s.dataCenter, s.rack, s.connected, s.VolumeSizeLimit)
  30. return
  31. }
  32. func NewStore(port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int, needleMapKind NeedleMapType) (s *Store) {
  33. s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl, NeedleMapType: needleMapKind}
  34. s.Locations = make([]*DiskLocation, 0)
  35. for i := 0; i < len(dirnames); i++ {
  36. location := NewDiskLocation(dirnames[i], maxVolumeCounts[i])
  37. location.loadExistingVolumes(needleMapKind)
  38. s.Locations = append(s.Locations, location)
  39. }
  40. return
  41. }
  42. func (s *Store) AddVolume(volumeListString string, collection string, needleMapKind NeedleMapType, replicaPlacement string, ttlString string, preallocate int64) error {
  43. rt, e := NewReplicaPlacementFromString(replicaPlacement)
  44. if e != nil {
  45. return e
  46. }
  47. ttl, e := ReadTTL(ttlString)
  48. if e != nil {
  49. return e
  50. }
  51. for _, range_string := range strings.Split(volumeListString, ",") {
  52. if strings.Index(range_string, "-") < 0 {
  53. id_string := range_string
  54. id, err := NewVolumeId(id_string)
  55. if err != nil {
  56. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", id_string)
  57. }
  58. e = s.addVolume(VolumeId(id), collection, needleMapKind, rt, ttl, preallocate)
  59. } else {
  60. pair := strings.Split(range_string, "-")
  61. start, start_err := strconv.ParseUint(pair[0], 10, 64)
  62. if start_err != nil {
  63. return fmt.Errorf("Volume Start Id %s is not a valid unsigned integer!", pair[0])
  64. }
  65. end, end_err := strconv.ParseUint(pair[1], 10, 64)
  66. if end_err != nil {
  67. return fmt.Errorf("Volume End Id %s is not a valid unsigned integer!", pair[1])
  68. }
  69. for id := start; id <= end; id++ {
  70. if err := s.addVolume(VolumeId(id), collection, needleMapKind, rt, ttl, preallocate); err != nil {
  71. e = err
  72. }
  73. }
  74. }
  75. }
  76. return e
  77. }
  78. func (s *Store) DeleteCollection(collection string) (e error) {
  79. for _, location := range s.Locations {
  80. e = location.DeleteCollectionFromDiskLocation(collection)
  81. if e != nil {
  82. return
  83. }
  84. }
  85. return
  86. }
  87. func (s *Store) findVolume(vid VolumeId) *Volume {
  88. for _, location := range s.Locations {
  89. if v, found := location.FindVolume(vid); found {
  90. return v
  91. }
  92. }
  93. return nil
  94. }
  95. func (s *Store) findFreeLocation() (ret *DiskLocation) {
  96. max := 0
  97. for _, location := range s.Locations {
  98. currentFreeCount := location.MaxVolumeCount - location.VolumesLen()
  99. if currentFreeCount > max {
  100. max = currentFreeCount
  101. ret = location
  102. }
  103. }
  104. return ret
  105. }
  106. func (s *Store) addVolume(vid VolumeId, collection string, needleMapKind NeedleMapType, replicaPlacement *ReplicaPlacement, ttl *TTL, preallocate int64) error {
  107. if s.findVolume(vid) != nil {
  108. return fmt.Errorf("Volume Id %d already exists!", vid)
  109. }
  110. if location := s.findFreeLocation(); location != nil {
  111. glog.V(0).Infof("In dir %s adds volume:%v collection:%s replicaPlacement:%v ttl:%v",
  112. location.Directory, vid, collection, replicaPlacement, ttl)
  113. if volume, err := NewVolume(location.Directory, collection, vid, needleMapKind, replicaPlacement, ttl, preallocate); err == nil {
  114. location.SetVolume(vid, volume)
  115. return nil
  116. } else {
  117. return err
  118. }
  119. }
  120. return fmt.Errorf("No more free space left")
  121. }
  122. func (s *Store) Status() []*VolumeInfo {
  123. var stats []*VolumeInfo
  124. for _, location := range s.Locations {
  125. location.RLock()
  126. for k, v := range location.volumes {
  127. s := &VolumeInfo{
  128. Id: VolumeId(k),
  129. Size: v.ContentSize(),
  130. Collection: v.Collection,
  131. ReplicaPlacement: v.ReplicaPlacement,
  132. Version: v.Version(),
  133. FileCount: v.nm.FileCount(),
  134. DeleteCount: v.nm.DeletedCount(),
  135. DeletedByteCount: v.nm.DeletedSize(),
  136. ReadOnly: v.readOnly,
  137. Ttl: v.Ttl}
  138. stats = append(stats, s)
  139. }
  140. location.RUnlock()
  141. }
  142. sortVolumeInfos(stats)
  143. return stats
  144. }
  145. func (s *Store) SetDataCenter(dataCenter string) {
  146. s.dataCenter = dataCenter
  147. }
  148. func (s *Store) SetRack(rack string) {
  149. s.rack = rack
  150. }
  151. func (s *Store) CollectHeartbeat() *master_pb.Heartbeat {
  152. var volumeMessages []*master_pb.VolumeInformationMessage
  153. maxVolumeCount := 0
  154. var maxFileKey NeedleId
  155. for _, location := range s.Locations {
  156. maxVolumeCount = maxVolumeCount + location.MaxVolumeCount
  157. location.Lock()
  158. for k, v := range location.volumes {
  159. if maxFileKey < v.nm.MaxFileKey() {
  160. maxFileKey = v.nm.MaxFileKey()
  161. }
  162. if !v.expired(s.VolumeSizeLimit) {
  163. volumeMessage := &master_pb.VolumeInformationMessage{
  164. Id: uint32(k),
  165. Size: uint64(v.Size()),
  166. Collection: v.Collection,
  167. FileCount: uint64(v.nm.FileCount()),
  168. DeleteCount: uint64(v.nm.DeletedCount()),
  169. DeletedByteCount: v.nm.DeletedSize(),
  170. ReadOnly: v.readOnly,
  171. ReplicaPlacement: uint32(v.ReplicaPlacement.Byte()),
  172. Version: uint32(v.Version()),
  173. Ttl: v.Ttl.ToUint32(),
  174. }
  175. volumeMessages = append(volumeMessages, volumeMessage)
  176. } else {
  177. if v.exiredLongEnough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
  178. location.deleteVolumeById(v.Id)
  179. glog.V(0).Infoln("volume", v.Id, "is deleted.")
  180. } else {
  181. glog.V(0).Infoln("volume", v.Id, "is expired.")
  182. }
  183. }
  184. }
  185. location.Unlock()
  186. }
  187. return &master_pb.Heartbeat{
  188. Ip: s.Ip,
  189. Port: uint32(s.Port),
  190. PublicUrl: s.PublicUrl,
  191. MaxVolumeCount: uint32(maxVolumeCount),
  192. MaxFileKey: NeedleIdToUint64(maxFileKey),
  193. DataCenter: s.dataCenter,
  194. Rack: s.rack,
  195. Volumes: volumeMessages,
  196. }
  197. }
  198. func (s *Store) Close() {
  199. for _, location := range s.Locations {
  200. location.Close()
  201. }
  202. }
  203. func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
  204. if v := s.findVolume(i); v != nil {
  205. if v.readOnly {
  206. err = fmt.Errorf("Volume %d is read only", i)
  207. return
  208. }
  209. // TODO: count needle size ahead
  210. if MaxPossibleVolumeSize >= v.ContentSize()+uint64(size) {
  211. size, err = v.writeNeedle(n)
  212. } else {
  213. err = fmt.Errorf("Volume Size Limit %d Exceeded! Current size is %d", s.VolumeSizeLimit, v.ContentSize())
  214. }
  215. return
  216. }
  217. glog.V(0).Infoln("volume", i, "not found!")
  218. err = fmt.Errorf("Volume %d not found!", i)
  219. return
  220. }
  221. func (s *Store) updateMaster() {
  222. if s.Client != nil {
  223. if e := s.Client.Send(s.CollectHeartbeat()); e != nil {
  224. glog.V(0).Infoln("error when reporting size:", e)
  225. }
  226. }
  227. }
  228. func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
  229. if v := s.findVolume(i); v != nil && !v.readOnly {
  230. return v.deleteNeedle(n)
  231. }
  232. return 0, nil
  233. }
  234. func (s *Store) ReadVolumeNeedle(i VolumeId, n *Needle) (int, error) {
  235. if v := s.findVolume(i); v != nil {
  236. return v.readNeedle(n)
  237. }
  238. return 0, fmt.Errorf("Volume %d not found!", i)
  239. }
  240. func (s *Store) GetVolume(i VolumeId) *Volume {
  241. return s.findVolume(i)
  242. }
  243. func (s *Store) HasVolume(i VolumeId) bool {
  244. v := s.findVolume(i)
  245. return v != nil
  246. }
  247. func (s *Store) MountVolume(i VolumeId) error {
  248. for _, location := range s.Locations {
  249. if found := location.LoadVolume(i, s.NeedleMapType); found == true {
  250. s.updateMaster()
  251. return nil
  252. }
  253. }
  254. return fmt.Errorf("Volume %d not found on disk", i)
  255. }
  256. func (s *Store) UnmountVolume(i VolumeId) error {
  257. for _, location := range s.Locations {
  258. if err := location.UnloadVolume(i); err == nil {
  259. s.updateMaster()
  260. return nil
  261. }
  262. }
  263. return fmt.Errorf("Volume %d not found on disk", i)
  264. }
  265. func (s *Store) DeleteVolume(i VolumeId) error {
  266. for _, location := range s.Locations {
  267. if error := location.deleteVolumeById(i); error == nil {
  268. s.updateMaster()
  269. return nil
  270. }
  271. }
  272. return fmt.Errorf("Volume %d not found on disk", i)
  273. }