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.

379 lines
11 KiB

12 years ago
12 years ago
  1. package storage
  2. import (
  3. proto "code.google.com/p/goprotobuf/proto"
  4. "code.google.com/p/weed-fs/go/glog"
  5. "code.google.com/p/weed-fs/go/operation"
  6. "code.google.com/p/weed-fs/go/util"
  7. "encoding/json"
  8. "errors"
  9. "fmt"
  10. "io/ioutil"
  11. "math/rand"
  12. "strconv"
  13. "strings"
  14. )
  15. type DiskLocation struct {
  16. Directory string
  17. MaxVolumeCount int
  18. volumes map[VolumeId]*Volume
  19. }
  20. func (mn *DiskLocation) reset() {
  21. }
  22. type MasterNodes struct {
  23. nodes []string
  24. lastNode int
  25. }
  26. func NewMasterNodes(bootstrapNode string) (mn *MasterNodes) {
  27. mn = &MasterNodes{nodes: []string{bootstrapNode}, lastNode: -1}
  28. return
  29. }
  30. func (mn *MasterNodes) reset() {
  31. if len(mn.nodes) > 1 && mn.lastNode > 0 {
  32. mn.lastNode = -mn.lastNode
  33. }
  34. }
  35. func (mn *MasterNodes) findMaster() (string, error) {
  36. if len(mn.nodes) == 0 {
  37. return "", errors.New("No master node found!")
  38. }
  39. if mn.lastNode < 0 {
  40. for _, m := range mn.nodes {
  41. if masters, e := operation.ListMasters(m); e == nil {
  42. if len(masters) == 0 {
  43. continue
  44. }
  45. mn.nodes = masters
  46. mn.lastNode = rand.Intn(len(mn.nodes))
  47. glog.V(2).Info("current master node is :", mn.nodes[mn.lastNode])
  48. break
  49. }
  50. }
  51. }
  52. if mn.lastNode < 0 {
  53. return "", errors.New("No master node avalable!")
  54. }
  55. return mn.nodes[mn.lastNode], nil
  56. }
  57. type Store struct {
  58. Port int
  59. Ip string
  60. PublicUrl string
  61. Locations []*DiskLocation
  62. dataCenter string //optional informaton, overwriting master setting if exists
  63. rack string //optional information, overwriting master setting if exists
  64. connected bool
  65. volumeSizeLimit uint64 //read from the master
  66. masterNodes *MasterNodes
  67. }
  68. func NewStore(port int, ip, publicUrl string, dirnames []string, maxVolumeCounts []int) (s *Store) {
  69. s = &Store{Port: port, Ip: ip, PublicUrl: publicUrl}
  70. s.Locations = make([]*DiskLocation, 0)
  71. for i := 0; i < len(dirnames); i++ {
  72. location := &DiskLocation{Directory: dirnames[i], MaxVolumeCount: maxVolumeCounts[i]}
  73. location.volumes = make(map[VolumeId]*Volume)
  74. location.loadExistingVolumes()
  75. s.Locations = append(s.Locations, location)
  76. }
  77. return
  78. }
  79. func (s *Store) AddVolume(volumeListString string, collection string, replicaPlacement string) error {
  80. rt, e := NewReplicaPlacementFromString(replicaPlacement)
  81. if e != nil {
  82. return e
  83. }
  84. for _, range_string := range strings.Split(volumeListString, ",") {
  85. if strings.Index(range_string, "-") < 0 {
  86. id_string := range_string
  87. id, err := NewVolumeId(id_string)
  88. if err != nil {
  89. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", id_string)
  90. }
  91. e = s.addVolume(VolumeId(id), collection, rt)
  92. } else {
  93. pair := strings.Split(range_string, "-")
  94. start, start_err := strconv.ParseUint(pair[0], 10, 64)
  95. if start_err != nil {
  96. return fmt.Errorf("Volume Start Id %s is not a valid unsigned integer!", pair[0])
  97. }
  98. end, end_err := strconv.ParseUint(pair[1], 10, 64)
  99. if end_err != nil {
  100. return fmt.Errorf("Volume End Id %s is not a valid unsigned integer!", pair[1])
  101. }
  102. for id := start; id <= end; id++ {
  103. if err := s.addVolume(VolumeId(id), collection, rt); err != nil {
  104. e = err
  105. }
  106. }
  107. }
  108. }
  109. return e
  110. }
  111. func (s *Store) DeleteCollection(collection string) (e error) {
  112. for _, location := range s.Locations {
  113. for k, v := range location.volumes {
  114. if v.Collection == collection {
  115. e = v.Destroy()
  116. if e != nil {
  117. return
  118. }
  119. delete(location.volumes, k)
  120. }
  121. }
  122. }
  123. return
  124. }
  125. func (s *Store) findVolume(vid VolumeId) *Volume {
  126. for _, location := range s.Locations {
  127. if v, found := location.volumes[vid]; found {
  128. return v
  129. }
  130. }
  131. return nil
  132. }
  133. func (s *Store) findFreeLocation() (ret *DiskLocation) {
  134. max := 0
  135. for _, location := range s.Locations {
  136. currentFreeCount := location.MaxVolumeCount - len(location.volumes)
  137. if currentFreeCount > max {
  138. max = currentFreeCount
  139. ret = location
  140. }
  141. }
  142. return ret
  143. }
  144. func (s *Store) addVolume(vid VolumeId, collection string, replicaPlacement *ReplicaPlacement) error {
  145. if s.findVolume(vid) != nil {
  146. return fmt.Errorf("Volume Id %d already exists!", vid)
  147. }
  148. if location := s.findFreeLocation(); location != nil {
  149. glog.V(0).Infoln("In dir", location.Directory, "adds volume =", vid, ", collection =", collection, ", replicaPlacement =", replicaPlacement)
  150. if volume, err := NewVolume(location.Directory, collection, vid, replicaPlacement); err == nil {
  151. location.volumes[vid] = volume
  152. return nil
  153. } else {
  154. return err
  155. }
  156. }
  157. return fmt.Errorf("No more free space left")
  158. }
  159. func (s *Store) CheckCompactVolume(volumeIdString string, garbageThresholdString string) (error, bool) {
  160. vid, err := NewVolumeId(volumeIdString)
  161. if err != nil {
  162. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString), false
  163. }
  164. garbageThreshold, e := strconv.ParseFloat(garbageThresholdString, 32)
  165. if e != nil {
  166. return fmt.Errorf("garbageThreshold %s is not a valid float number!", garbageThresholdString), false
  167. }
  168. if v := s.findVolume(vid); v != nil {
  169. glog.V(3).Infoln(vid, "garbage level is", v.garbageLevel())
  170. return nil, garbageThreshold < v.garbageLevel()
  171. }
  172. return fmt.Errorf("volume id %d is not found during check compact!", vid), false
  173. }
  174. func (s *Store) CompactVolume(volumeIdString string) error {
  175. vid, err := NewVolumeId(volumeIdString)
  176. if err != nil {
  177. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString)
  178. }
  179. if v := s.findVolume(vid); v != nil {
  180. return v.Compact()
  181. }
  182. return fmt.Errorf("volume id %d is not found during compact!", vid)
  183. }
  184. func (s *Store) CommitCompactVolume(volumeIdString string) error {
  185. vid, err := NewVolumeId(volumeIdString)
  186. if err != nil {
  187. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString)
  188. }
  189. if v := s.findVolume(vid); v != nil {
  190. return v.commitCompact()
  191. }
  192. return fmt.Errorf("volume id %d is not found during commit compact!", vid)
  193. }
  194. func (s *Store) FreezeVolume(volumeIdString string) error {
  195. vid, err := NewVolumeId(volumeIdString)
  196. if err != nil {
  197. return fmt.Errorf("Volume Id %s is not a valid unsigned integer!", volumeIdString)
  198. }
  199. if v := s.findVolume(vid); v != nil {
  200. if v.readOnly {
  201. return fmt.Errorf("Volume %s is already read-only", volumeIdString)
  202. }
  203. return v.freeze()
  204. }
  205. return fmt.Errorf("volume id %d is not found during freeze!", vid)
  206. }
  207. func (l *DiskLocation) loadExistingVolumes() {
  208. if dirs, err := ioutil.ReadDir(l.Directory); err == nil {
  209. for _, dir := range dirs {
  210. name := dir.Name()
  211. if !dir.IsDir() && strings.HasSuffix(name, ".dat") {
  212. collection := ""
  213. base := name[:len(name)-len(".dat")]
  214. i := strings.Index(base, "_")
  215. if i > 0 {
  216. collection, base = base[0:i], base[i+1:]
  217. }
  218. if vid, err := NewVolumeId(base); err == nil {
  219. if l.volumes[vid] == nil {
  220. if v, e := NewVolume(l.Directory, collection, vid, nil); e == nil {
  221. l.volumes[vid] = v
  222. glog.V(0).Infoln("data file", l.Directory+"/"+name, "replicaPlacement =", v.ReplicaPlacement, "version =", v.Version(), "size =", v.Size())
  223. }
  224. }
  225. }
  226. }
  227. }
  228. }
  229. glog.V(0).Infoln("Store started on dir:", l.Directory, "with", len(l.volumes), "volumes", "max", l.MaxVolumeCount)
  230. }
  231. func (s *Store) Status() []*VolumeInfo {
  232. var stats []*VolumeInfo
  233. for _, location := range s.Locations {
  234. for k, v := range location.volumes {
  235. s := &VolumeInfo{Id: VolumeId(k), Size: v.ContentSize(),
  236. Collection: v.Collection,
  237. ReplicaPlacement: v.ReplicaPlacement,
  238. Version: v.Version(),
  239. FileCount: v.nm.FileCount(),
  240. DeleteCount: v.nm.DeletedCount(),
  241. DeletedByteCount: v.nm.DeletedSize(),
  242. ReadOnly: v.readOnly}
  243. stats = append(stats, s)
  244. }
  245. }
  246. return stats
  247. }
  248. func (s *Store) SetDataCenter(dataCenter string) {
  249. s.dataCenter = dataCenter
  250. }
  251. func (s *Store) SetRack(rack string) {
  252. s.rack = rack
  253. }
  254. func (s *Store) SetBootstrapMaster(bootstrapMaster string) {
  255. s.masterNodes = NewMasterNodes(bootstrapMaster)
  256. }
  257. func (s *Store) Join() (masterNode string, e error) {
  258. masterNode, e = s.masterNodes.findMaster()
  259. if e != nil {
  260. return
  261. }
  262. var volumeMessages []*operation.VolumeInformationMessage
  263. maxVolumeCount := 0
  264. var maxFileKey uint64
  265. for _, location := range s.Locations {
  266. maxVolumeCount = maxVolumeCount + location.MaxVolumeCount
  267. for k, v := range location.volumes {
  268. volumeMessage := &operation.VolumeInformationMessage{
  269. Id: proto.Uint32(uint32(k)),
  270. Size: proto.Uint64(uint64(v.Size())),
  271. Collection: proto.String(v.Collection),
  272. FileCount: proto.Uint64(uint64(v.nm.FileCount())),
  273. DeleteCount: proto.Uint64(uint64(v.nm.DeletedCount())),
  274. DeletedByteCount: proto.Uint64(v.nm.DeletedSize()),
  275. ReadOnly: proto.Bool(v.readOnly),
  276. ReplicaPlacement: proto.Uint32(uint32(v.ReplicaPlacement.Byte())),
  277. Version: proto.Uint32(uint32(v.Version())),
  278. }
  279. volumeMessages = append(volumeMessages, volumeMessage)
  280. if maxFileKey < v.nm.MaxFileKey() {
  281. maxFileKey = v.nm.MaxFileKey()
  282. }
  283. }
  284. }
  285. joinMessage := &operation.JoinMessage{
  286. IsInit: proto.Bool(!s.connected),
  287. Ip: proto.String(s.Ip),
  288. Port: proto.Uint32(uint32(s.Port)),
  289. PublicUrl: proto.String(s.PublicUrl),
  290. MaxVolumeCount: proto.Uint32(uint32(maxVolumeCount)),
  291. MaxFileKey: proto.Uint64(maxFileKey),
  292. DataCenter: proto.String(s.dataCenter),
  293. Rack: proto.String(s.rack),
  294. Volumes: volumeMessages,
  295. }
  296. data, err := proto.Marshal(joinMessage)
  297. if err != nil {
  298. return "", err
  299. }
  300. jsonBlob, err := util.PostBytes("http://"+masterNode+"/dir/join", data)
  301. if err != nil {
  302. s.masterNodes.reset()
  303. return "", err
  304. }
  305. var ret operation.JoinResult
  306. if err := json.Unmarshal(jsonBlob, &ret); err != nil {
  307. return masterNode, err
  308. }
  309. if ret.Error != "" {
  310. return masterNode, errors.New(ret.Error)
  311. }
  312. s.volumeSizeLimit = ret.VolumeSizeLimit
  313. s.connected = true
  314. return
  315. }
  316. func (s *Store) Close() {
  317. for _, location := range s.Locations {
  318. for _, v := range location.volumes {
  319. v.Close()
  320. }
  321. }
  322. }
  323. func (s *Store) Write(i VolumeId, n *Needle) (size uint32, err error) {
  324. if v := s.findVolume(i); v != nil {
  325. if v.readOnly {
  326. err = fmt.Errorf("Volume %d is read only!", i)
  327. return
  328. } else {
  329. if MaxPossibleVolumeSize >= v.ContentSize()+uint64(size) {
  330. size, err = v.write(n)
  331. } else {
  332. err = fmt.Errorf("Volume Size Limit %d Exceeded! Current size is %d", s.volumeSizeLimit, v.ContentSize())
  333. }
  334. if s.volumeSizeLimit < v.ContentSize()+3*uint64(size) {
  335. glog.V(0).Infoln("volume", i, "size", v.ContentSize(), "will exceed limit", s.volumeSizeLimit)
  336. if _, e := s.Join(); e != nil {
  337. glog.V(0).Infoln("error when reporting size:", e)
  338. }
  339. }
  340. }
  341. return
  342. }
  343. glog.V(0).Infoln("volume", i, "not found!")
  344. err = fmt.Errorf("Volume %d not found!", i)
  345. return
  346. }
  347. func (s *Store) Delete(i VolumeId, n *Needle) (uint32, error) {
  348. if v := s.findVolume(i); v != nil && !v.readOnly {
  349. return v.delete(n)
  350. }
  351. return 0, nil
  352. }
  353. func (s *Store) Read(i VolumeId, n *Needle) (int, error) {
  354. if v := s.findVolume(i); v != nil {
  355. return v.read(n)
  356. }
  357. return 0, fmt.Errorf("Volume %v not found!", i)
  358. }
  359. func (s *Store) GetVolume(i VolumeId) *Volume {
  360. return s.findVolume(i)
  361. }
  362. func (s *Store) HasVolume(i VolumeId) bool {
  363. v := s.findVolume(i)
  364. return v != nil
  365. }