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.

204 lines
6.0 KiB

12 years ago
12 years ago
  1. package storage
  2. import (
  3. "encoding/json"
  4. "errors"
  5. "io/ioutil"
  6. "log"
  7. "net/url"
  8. "code.google.com/p/weed-fs/go/util"
  9. "strconv"
  10. "strings"
  11. )
  12. type Store struct {
  13. volumes map[VolumeId]*Volume
  14. dir string
  15. Port int
  16. Ip string
  17. PublicUrl string
  18. MaxVolumeCount int
  19. masterNode string
  20. connected bool
  21. volumeSizeLimit uint64 //read from the master
  22. }
  23. func NewStore(port int, ip, publicUrl, dirname string, maxVolumeCount int) (s *Store) {
  24. s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl, dir: dirname, MaxVolumeCount: maxVolumeCount}
  25. s.volumes = make(map[VolumeId]*Volume)
  26. s.loadExistingVolumes()
  27. log.Println("Store started on dir:", dirname, "with", len(s.volumes), "volumes")
  28. return
  29. }
  30. func (s *Store) AddVolume(volumeListString string, replicationType string) error {
  31. rt, e := NewReplicationTypeFromString(replicationType)
  32. if e != nil {
  33. return e
  34. }
  35. for _, range_string := range strings.Split(volumeListString, ",") {
  36. if strings.Index(range_string, "-") < 0 {
  37. id_string := range_string
  38. id, err := NewVolumeId(id_string)
  39. if err != nil {
  40. return errors.New("Volume Id " + id_string + " is not a valid unsigned integer!")
  41. }
  42. e = s.addVolume(VolumeId(id), rt)
  43. } else {
  44. pair := strings.Split(range_string, "-")
  45. start, start_err := strconv.ParseUint(pair[0], 10, 64)
  46. if start_err != nil {
  47. return errors.New("Volume Start Id" + pair[0] + " is not a valid unsigned integer!")
  48. }
  49. end, end_err := strconv.ParseUint(pair[1], 10, 64)
  50. if end_err != nil {
  51. return errors.New("Volume End Id" + pair[1] + " is not a valid unsigned integer!")
  52. }
  53. for id := start; id <= end; id++ {
  54. if err := s.addVolume(VolumeId(id), rt); err != nil {
  55. e = err
  56. }
  57. }
  58. }
  59. }
  60. return e
  61. }
  62. func (s *Store) addVolume(vid VolumeId, replicationType ReplicationType) (err error) {
  63. if s.volumes[vid] != nil {
  64. return errors.New("Volume Id " + vid.String() + " already exists!")
  65. }
  66. log.Println("In dir", s.dir, "adds volume =", vid, ", replicationType =", replicationType)
  67. s.volumes[vid], err = NewVolume(s.dir, vid, replicationType)
  68. return err
  69. }
  70. func (s *Store) CheckCompactVolume(volumeIdString string, garbageThresholdString string) (error, bool) {
  71. vid, err := NewVolumeId(volumeIdString)
  72. if err != nil {
  73. return errors.New("Volume Id " + volumeIdString + " is not a valid unsigned integer!"), false
  74. }
  75. garbageThreshold, e := strconv.ParseFloat(garbageThresholdString, 32)
  76. if e != nil {
  77. return errors.New("garbageThreshold " + garbageThresholdString + " is not a valid float number!"), false
  78. }
  79. return nil, garbageThreshold < s.volumes[vid].garbageLevel()
  80. }
  81. func (s *Store) CompactVolume(volumeIdString string) error {
  82. vid, err := NewVolumeId(volumeIdString)
  83. if err != nil {
  84. return errors.New("Volume Id " + volumeIdString + " is not a valid unsigned integer!")
  85. }
  86. return s.volumes[vid].compact()
  87. }
  88. func (s *Store) CommitCompactVolume(volumeIdString string) error {
  89. vid, err := NewVolumeId(volumeIdString)
  90. if err != nil {
  91. return errors.New("Volume Id " + volumeIdString + " is not a valid unsigned integer!")
  92. }
  93. return s.volumes[vid].commitCompact()
  94. }
  95. func (s *Store) loadExistingVolumes() {
  96. if dirs, err := ioutil.ReadDir(s.dir); err == nil {
  97. for _, dir := range dirs {
  98. name := dir.Name()
  99. if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
  100. base := name[:len(name)-len(".dat")]
  101. if vid, err := NewVolumeId(base); err == nil {
  102. if s.volumes[vid] == nil {
  103. if v, e := NewVolume(s.dir, vid, CopyNil); e == nil {
  104. s.volumes[vid] = v
  105. log.Println("In dir", s.dir, "read volume =", vid, "replicationType =", v.ReplicaType, "version =", v.Version(), "size =", v.Size())
  106. }
  107. }
  108. }
  109. }
  110. }
  111. }
  112. }
  113. func (s *Store) Status() []*VolumeInfo {
  114. var stats []*VolumeInfo
  115. for k, v := range s.volumes {
  116. s := new(VolumeInfo)
  117. s.Id, s.Size, s.RepType, s.Version, s.FileCount, s.DeleteCount, s.DeletedByteCount =
  118. VolumeId(k), v.ContentSize(), v.ReplicaType, v.Version(), v.nm.fileCounter, v.nm.deletionCounter, v.nm.deletionByteCounter
  119. stats = append(stats, s)
  120. }
  121. return stats
  122. }
  123. type JoinResult struct {
  124. VolumeSizeLimit uint64
  125. }
  126. func (s *Store) SetMaster(mserver string) {
  127. s.masterNode = mserver
  128. }
  129. func (s *Store) Join() error {
  130. stats := new([]*VolumeInfo)
  131. for k, v := range s.volumes {
  132. s := new(VolumeInfo)
  133. s.Id, s.Size, s.RepType, s.Version, s.FileCount, s.DeleteCount, s.DeletedByteCount =
  134. VolumeId(k), uint64(v.Size()), v.ReplicaType, v.Version(), v.nm.fileCounter, v.nm.deletionCounter, v.nm.deletionByteCounter
  135. *stats = append(*stats, s)
  136. }
  137. bytes, _ := json.Marshal(stats)
  138. values := make(url.Values)
  139. if !s.connected {
  140. values.Add("init", "true")
  141. }
  142. values.Add("port", strconv.Itoa(s.Port))
  143. values.Add("ip", s.Ip)
  144. values.Add("publicUrl", s.PublicUrl)
  145. values.Add("volumes", string(bytes))
  146. values.Add("maxVolumeCount", strconv.Itoa(s.MaxVolumeCount))
  147. jsonBlob, err := util.Post("http://"+s.masterNode+"/dir/join", values)
  148. if err != nil {
  149. return err
  150. }
  151. var ret JoinResult
  152. if err := json.Unmarshal(jsonBlob, &ret); err != nil {
  153. return err
  154. }
  155. s.volumeSizeLimit = ret.VolumeSizeLimit
  156. s.connected = true
  157. return nil
  158. }
  159. func (s *Store) Close() {
  160. for _, v := range s.volumes {
  161. v.Close()
  162. }
  163. }
  164. func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
  165. if v := s.volumes[i]; v != nil {
  166. size, err = v.write(n)
  167. if err != nil && s.volumeSizeLimit < v.ContentSize()+uint64(size) && s.volumeSizeLimit >= v.ContentSize() {
  168. log.Println("volume", i, "size is", v.ContentSize(), "close to", s.volumeSizeLimit)
  169. s.Join()
  170. }
  171. return
  172. }
  173. log.Println("volume", i, "not found!")
  174. return
  175. }
  176. func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
  177. if v := s.volumes[i]; v != nil {
  178. return v.delete(n)
  179. }
  180. return 0, nil
  181. }
  182. func (s *Store) Read(i VolumeId, n *Needle) (int, error) {
  183. if v := s.volumes[i]; v != nil {
  184. return v.read(n)
  185. }
  186. return 0, errors.New("Not Found")
  187. }
  188. func (s *Store) GetVolume(i VolumeId) *Volume {
  189. return s.volumes[i]
  190. }
  191. func (s *Store) HasVolume(i VolumeId) bool {
  192. _, ok := s.volumes[i]
  193. return ok
  194. }