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.

154 lines
4.4 KiB

6 years ago
6 years ago
7 years ago
6 years ago
6 years ago
  1. package storage
  2. import (
  3. "fmt"
  4. "os"
  5. "path/filepath"
  6. "github.com/syndtr/goleveldb/leveldb/errors"
  7. "github.com/syndtr/goleveldb/leveldb/opt"
  8. "github.com/chrislusf/seaweedfs/weed/storage/idx"
  9. "github.com/syndtr/goleveldb/leveldb"
  10. "github.com/chrislusf/seaweedfs/weed/glog"
  11. "github.com/chrislusf/seaweedfs/weed/storage/needle_map"
  12. . "github.com/chrislusf/seaweedfs/weed/storage/types"
  13. "github.com/chrislusf/seaweedfs/weed/util"
  14. )
  15. type LevelDbNeedleMap struct {
  16. baseNeedleMapper
  17. dbFileName string
  18. db *leveldb.DB
  19. }
  20. func NewLevelDbNeedleMap(dbFileName string, indexFile *os.File, opts *opt.Options) (m *LevelDbNeedleMap, err error) {
  21. m = &LevelDbNeedleMap{dbFileName: dbFileName}
  22. m.indexFile = indexFile
  23. if !isLevelDbFresh(dbFileName, indexFile) {
  24. glog.V(0).Infof("Start to Generate %s from %s", dbFileName, indexFile.Name())
  25. generateLevelDbFile(dbFileName, indexFile)
  26. glog.V(0).Infof("Finished Generating %s from %s", dbFileName, indexFile.Name())
  27. }
  28. glog.V(1).Infof("Opening %s...", dbFileName)
  29. if m.db, err = leveldb.OpenFile(dbFileName, opts); err != nil {
  30. if errors.IsCorrupted(err) {
  31. m.db, err = leveldb.RecoverFile(dbFileName, opts)
  32. }
  33. return
  34. }
  35. glog.V(1).Infof("Loading %s...", indexFile.Name())
  36. mm, indexLoadError := newNeedleMapMetricFromIndexFile(indexFile)
  37. if indexLoadError != nil {
  38. return nil, indexLoadError
  39. }
  40. m.mapMetric = *mm
  41. return
  42. }
  43. func isLevelDbFresh(dbFileName string, indexFile *os.File) bool {
  44. // normally we always write to index file first
  45. dbLogFile, err := os.Open(filepath.Join(dbFileName, "LOG"))
  46. if err != nil {
  47. return false
  48. }
  49. defer dbLogFile.Close()
  50. dbStat, dbStatErr := dbLogFile.Stat()
  51. indexStat, indexStatErr := indexFile.Stat()
  52. if dbStatErr != nil || indexStatErr != nil {
  53. glog.V(0).Infof("Can not stat file: %v and %v", dbStatErr, indexStatErr)
  54. return false
  55. }
  56. return dbStat.ModTime().After(indexStat.ModTime())
  57. }
  58. func generateLevelDbFile(dbFileName string, indexFile *os.File) error {
  59. db, err := leveldb.OpenFile(dbFileName, nil)
  60. if err != nil {
  61. return err
  62. }
  63. defer db.Close()
  64. return idx.WalkIndexFile(indexFile, func(key NeedleId, offset Offset, size uint32) error {
  65. if !offset.IsZero() && size != TombstoneFileSize {
  66. levelDbWrite(db, key, offset, size)
  67. } else {
  68. levelDbDelete(db, key)
  69. }
  70. return nil
  71. })
  72. }
  73. func (m *LevelDbNeedleMap) Get(key NeedleId) (element *needle_map.NeedleValue, ok bool) {
  74. bytes := make([]byte, NeedleIdSize)
  75. NeedleIdToBytes(bytes[0:NeedleIdSize], key)
  76. data, err := m.db.Get(bytes, nil)
  77. if err != nil || len(data) != OffsetSize+SizeSize {
  78. return nil, false
  79. }
  80. offset := BytesToOffset(data[0:OffsetSize])
  81. size := util.BytesToUint32(data[OffsetSize : OffsetSize+SizeSize])
  82. return &needle_map.NeedleValue{Key: key, Offset: offset, Size: size}, true
  83. }
  84. func (m *LevelDbNeedleMap) Put(key NeedleId, offset Offset, 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 := m.appendToIndexFile(key, offset, size); err != nil {
  92. return fmt.Errorf("cannot write to indexfile %s: %v", m.indexFile.Name(), err)
  93. }
  94. return levelDbWrite(m.db, key, offset, size)
  95. }
  96. func levelDbWrite(db *leveldb.DB, key NeedleId, offset Offset, size uint32) error {
  97. bytes := needle_map.ToBytes(key, offset, size)
  98. if err := db.Put(bytes[0:NeedleIdSize], bytes[NeedleIdSize:NeedleIdSize+OffsetSize+SizeSize], nil); err != nil {
  99. return fmt.Errorf("failed to write leveldb: %v", err)
  100. }
  101. return nil
  102. }
  103. func levelDbDelete(db *leveldb.DB, key NeedleId) error {
  104. bytes := make([]byte, NeedleIdSize)
  105. NeedleIdToBytes(bytes, key)
  106. return db.Delete(bytes, nil)
  107. }
  108. func (m *LevelDbNeedleMap) Delete(key NeedleId, offset Offset) error {
  109. if oldNeedle, ok := m.Get(key); ok {
  110. m.logDelete(oldNeedle.Size)
  111. }
  112. // write to index file first
  113. if err := m.appendToIndexFile(key, offset, TombstoneFileSize); err != nil {
  114. return err
  115. }
  116. return levelDbDelete(m.db, key)
  117. }
  118. func (m *LevelDbNeedleMap) Close() {
  119. indexFileName := m.indexFile.Name()
  120. if err := m.indexFile.Sync(); err != nil {
  121. glog.Warningf("sync file %s failed: %v", indexFileName, err)
  122. }
  123. if err := m.indexFile.Close(); err != nil {
  124. glog.Warningf("close index file %s failed: %v", indexFileName, err)
  125. }
  126. if err := m.db.Close(); err != nil {
  127. glog.Warningf("close levelDB failed: %v", err)
  128. }
  129. }
  130. func (m *LevelDbNeedleMap) Destroy() error {
  131. m.Close()
  132. os.Remove(m.indexFile.Name())
  133. return os.RemoveAll(m.dbFileName)
  134. }