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.

267 lines
7.7 KiB

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