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.

181 lines
4.4 KiB

  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "github.com/boltdb/bolt"
  6. "github.com/chrislusf/weed-fs/go/glog"
  7. "github.com/chrislusf/weed-fs/go/util"
  8. )
  9. type BoltDbNeedleMap struct {
  10. dbFileName string
  11. indexFile *os.File
  12. db *bolt.DB
  13. mapMetric
  14. }
  15. var boltdbBucket = []byte("weed")
  16. func NewBoltDbNeedleMap(dbFileName string, indexFile *os.File) (m *BoltDbNeedleMap, err error) {
  17. m = &BoltDbNeedleMap{indexFile: indexFile, dbFileName: dbFileName}
  18. if !isBoltDbFresh(dbFileName, indexFile) {
  19. glog.V(1).Infof("Start to Generate %s from %s", dbFileName, indexFile.Name())
  20. generateBoltDbFile(dbFileName, indexFile)
  21. glog.V(1).Infof("Finished Generating %s from %s", dbFileName, indexFile.Name())
  22. }
  23. glog.V(1).Infof("Opening %s...", dbFileName)
  24. if m.db, err = bolt.Open(dbFileName, 0644, nil); err != nil {
  25. return
  26. }
  27. glog.V(1).Infof("Loading %s...", indexFile.Name())
  28. nm, indexLoadError := LoadNeedleMap(indexFile)
  29. if indexLoadError != nil {
  30. return nil, indexLoadError
  31. }
  32. m.mapMetric = nm.mapMetric
  33. return
  34. }
  35. func isBoltDbFresh(dbFileName string, indexFile *os.File) bool {
  36. // normally we always write to index file first
  37. dbLogFile, err := os.Open(dbFileName)
  38. if err != nil {
  39. return false
  40. }
  41. defer dbLogFile.Close()
  42. dbStat, dbStatErr := dbLogFile.Stat()
  43. indexStat, indexStatErr := indexFile.Stat()
  44. if dbStatErr != nil || indexStatErr != nil {
  45. glog.V(0).Infof("Can not stat file: %v and %v", dbStatErr, indexStatErr)
  46. return false
  47. }
  48. return dbStat.ModTime().After(indexStat.ModTime())
  49. }
  50. func generateBoltDbFile(dbFileName string, indexFile *os.File) error {
  51. db, err := bolt.Open(dbFileName, 0644, nil)
  52. if err != nil {
  53. return err
  54. }
  55. defer db.Close()
  56. return WalkIndexFile(indexFile, func(key uint64, offset, size uint32) error {
  57. if offset > 0 {
  58. boltDbWrite(db, key, offset, size)
  59. } else {
  60. boltDbDelete(db, key)
  61. }
  62. return nil
  63. })
  64. }
  65. func (m *BoltDbNeedleMap) Get(key uint64) (element *NeedleValue, ok bool) {
  66. bytes := make([]byte, 8)
  67. var data []byte
  68. util.Uint64toBytes(bytes, key)
  69. err := m.db.View(func(tx *bolt.Tx) error {
  70. bucket := tx.Bucket(boltdbBucket)
  71. if bucket == nil {
  72. return fmt.Errorf("Bucket %q not found!", boltdbBucket)
  73. }
  74. data = bucket.Get(bytes)
  75. return nil
  76. })
  77. if err != nil || len(data) != 8 {
  78. return nil, false
  79. }
  80. offset := util.BytesToUint32(data[0:4])
  81. size := util.BytesToUint32(data[4:8])
  82. return &NeedleValue{Key: Key(key), Offset: offset, Size: size}, true
  83. }
  84. func (m *BoltDbNeedleMap) Put(key uint64, offset uint32, size uint32) error {
  85. var oldSize uint32
  86. if oldNeedle, ok := m.Get(key); ok {
  87. oldSize = oldNeedle.Size
  88. }
  89. m.logPut(key, oldSize, size)
  90. // write to index file first
  91. if err := appendToIndexFile(m.indexFile, key, offset, size); err != nil {
  92. return fmt.Errorf("cannot write to indexfile %s: %v", m.indexFile.Name(), err)
  93. }
  94. return boltDbWrite(m.db, key, offset, size)
  95. }
  96. func boltDbWrite(db *bolt.DB,
  97. key uint64, offset uint32, size uint32) error {
  98. bytes := make([]byte, 16)
  99. util.Uint64toBytes(bytes[0:8], key)
  100. util.Uint32toBytes(bytes[8:12], offset)
  101. util.Uint32toBytes(bytes[12:16], size)
  102. return db.Update(func(tx *bolt.Tx) error {
  103. bucket, err := tx.CreateBucketIfNotExists(boltdbBucket)
  104. if err != nil {
  105. return err
  106. }
  107. err = bucket.Put(bytes[0:8], bytes[8:16])
  108. if err != nil {
  109. return err
  110. }
  111. return nil
  112. })
  113. }
  114. func boltDbDelete(db *bolt.DB, key uint64) error {
  115. bytes := make([]byte, 8)
  116. util.Uint64toBytes(bytes, key)
  117. return db.Update(func(tx *bolt.Tx) error {
  118. bucket, err := tx.CreateBucketIfNotExists(boltdbBucket)
  119. if err != nil {
  120. return err
  121. }
  122. err = bucket.Delete(bytes)
  123. if err != nil {
  124. return err
  125. }
  126. return nil
  127. })
  128. }
  129. func (m *BoltDbNeedleMap) Delete(key uint64) error {
  130. if oldNeedle, ok := m.Get(key); ok {
  131. m.logDelete(oldNeedle.Size)
  132. }
  133. // write to index file first
  134. if err := appendToIndexFile(m.indexFile, key, 0, 0); err != nil {
  135. return err
  136. }
  137. return boltDbDelete(m.db, key)
  138. }
  139. func (m *BoltDbNeedleMap) Close() {
  140. m.db.Close()
  141. }
  142. func (m *BoltDbNeedleMap) Destroy() error {
  143. m.Close()
  144. os.Remove(m.indexFile.Name())
  145. return os.Remove(m.dbFileName)
  146. }
  147. func (m *BoltDbNeedleMap) ContentSize() uint64 {
  148. return m.FileByteCounter
  149. }
  150. func (m *BoltDbNeedleMap) DeletedSize() uint64 {
  151. return m.DeletionByteCounter
  152. }
  153. func (m *BoltDbNeedleMap) FileCount() int {
  154. return m.FileCounter
  155. }
  156. func (m *BoltDbNeedleMap) DeletedCount() int {
  157. return m.DeletionCounter
  158. }
  159. func (m *BoltDbNeedleMap) MaxFileKey() uint64 {
  160. return m.MaximumFileKey
  161. }