leveldb_store.go 6.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241
  1. package leveldb
  2. import (
  3. "bytes"
  4. "context"
  5. "fmt"
  6. "github.com/syndtr/goleveldb/leveldb"
  7. leveldb_errors "github.com/syndtr/goleveldb/leveldb/errors"
  8. "github.com/syndtr/goleveldb/leveldb/opt"
  9. leveldb_util "github.com/syndtr/goleveldb/leveldb/util"
  10. "github.com/chrislusf/seaweedfs/weed/filer"
  11. "github.com/chrislusf/seaweedfs/weed/glog"
  12. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  13. weed_util "github.com/chrislusf/seaweedfs/weed/util"
  14. )
  15. const (
  16. DIR_FILE_SEPARATOR = byte(0x00)
  17. )
  18. func init() {
  19. filer.Stores = append(filer.Stores, &LevelDBStore{})
  20. }
  21. type LevelDBStore struct {
  22. db *leveldb.DB
  23. }
  24. func (store *LevelDBStore) GetName() string {
  25. return "leveldb"
  26. }
  27. func (store *LevelDBStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
  28. dir := configuration.GetString(prefix + "dir")
  29. return store.initialize(dir)
  30. }
  31. func (store *LevelDBStore) initialize(dir string) (err error) {
  32. glog.Infof("filer store dir: %s", dir)
  33. if err := weed_util.TestFolderWritable(dir); err != nil {
  34. return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err)
  35. }
  36. opts := &opt.Options{
  37. BlockCacheCapacity: 32 * 1024 * 1024, // default value is 8MiB
  38. WriteBuffer: 16 * 1024 * 1024, // default value is 4MiB
  39. CompactionTableSizeMultiplier: 10,
  40. }
  41. if store.db, err = leveldb.OpenFile(dir, opts); err != nil {
  42. if leveldb_errors.IsCorrupted(err) {
  43. store.db, err = leveldb.RecoverFile(dir, opts)
  44. }
  45. if err != nil {
  46. glog.Infof("filer store open dir %s: %v", dir, err)
  47. return
  48. }
  49. }
  50. return
  51. }
  52. func (store *LevelDBStore) BeginTransaction(ctx context.Context) (context.Context, error) {
  53. return ctx, nil
  54. }
  55. func (store *LevelDBStore) CommitTransaction(ctx context.Context) error {
  56. return nil
  57. }
  58. func (store *LevelDBStore) RollbackTransaction(ctx context.Context) error {
  59. return nil
  60. }
  61. func (store *LevelDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  62. key := genKey(entry.DirAndName())
  63. value, err := entry.EncodeAttributesAndChunks()
  64. if err != nil {
  65. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  66. }
  67. if len(entry.Chunks) > 50 {
  68. value = weed_util.MaybeGzipData(value)
  69. }
  70. err = store.db.Put(key, value, nil)
  71. if err != nil {
  72. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  73. }
  74. // println("saved", entry.FullPath, "chunks", len(entry.Chunks))
  75. return nil
  76. }
  77. func (store *LevelDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  78. return store.InsertEntry(ctx, entry)
  79. }
  80. func (store *LevelDBStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
  81. key := genKey(fullpath.DirAndName())
  82. data, err := store.db.Get(key, nil)
  83. if err == leveldb.ErrNotFound {
  84. return nil, filer_pb.ErrNotFound
  85. }
  86. if err != nil {
  87. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  88. }
  89. entry = &filer.Entry{
  90. FullPath: fullpath,
  91. }
  92. err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData((data)))
  93. if err != nil {
  94. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  95. }
  96. // println("read", entry.FullPath, "chunks", len(entry.Chunks), "data", len(data), string(data))
  97. return entry, nil
  98. }
  99. func (store *LevelDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  100. key := genKey(fullpath.DirAndName())
  101. err = store.db.Delete(key, nil)
  102. if err != nil {
  103. return fmt.Errorf("delete %s : %v", fullpath, err)
  104. }
  105. return nil
  106. }
  107. func (store *LevelDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  108. batch := new(leveldb.Batch)
  109. directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
  110. iter := store.db.NewIterator(&leveldb_util.Range{Start: directoryPrefix}, nil)
  111. for iter.Next() {
  112. key := iter.Key()
  113. if !bytes.HasPrefix(key, directoryPrefix) {
  114. break
  115. }
  116. fileName := getNameFromKey(key)
  117. if fileName == "" {
  118. continue
  119. }
  120. batch.Delete([]byte(genKey(string(fullpath), fileName)))
  121. }
  122. iter.Release()
  123. err = store.db.Write(batch, nil)
  124. if err != nil {
  125. return fmt.Errorf("delete %s : %v", fullpath, err)
  126. }
  127. return nil
  128. }
  129. func (store *LevelDBStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  130. return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
  131. }
  132. func (store *LevelDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  133. directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
  134. lastFileStart := directoryPrefix
  135. if startFileName != "" {
  136. lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
  137. }
  138. iter := store.db.NewIterator(&leveldb_util.Range{Start: lastFileStart}, nil)
  139. for iter.Next() {
  140. key := iter.Key()
  141. if !bytes.HasPrefix(key, directoryPrefix) {
  142. break
  143. }
  144. fileName := getNameFromKey(key)
  145. if fileName == "" {
  146. continue
  147. }
  148. if fileName == startFileName && !includeStartFile {
  149. continue
  150. }
  151. lastFileName = fileName
  152. limit--
  153. if limit < 0 {
  154. break
  155. }
  156. entry := &filer.Entry{
  157. FullPath: weed_util.NewFullPath(string(dirPath), fileName),
  158. }
  159. if decodeErr := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(iter.Value())); decodeErr != nil {
  160. err = decodeErr
  161. glog.V(0).Infof("list %s : %v", entry.FullPath, err)
  162. break
  163. }
  164. if !eachEntryFunc(entry) {
  165. break
  166. }
  167. }
  168. iter.Release()
  169. return lastFileName, err
  170. }
  171. func genKey(dirPath, fileName string) (key []byte) {
  172. key = []byte(dirPath)
  173. key = append(key, DIR_FILE_SEPARATOR)
  174. key = append(key, []byte(fileName)...)
  175. return key
  176. }
  177. func genDirectoryKeyPrefix(fullpath weed_util.FullPath, startFileName string) (keyPrefix []byte) {
  178. keyPrefix = []byte(string(fullpath))
  179. keyPrefix = append(keyPrefix, DIR_FILE_SEPARATOR)
  180. if len(startFileName) > 0 {
  181. keyPrefix = append(keyPrefix, []byte(startFileName)...)
  182. }
  183. return keyPrefix
  184. }
  185. func getNameFromKey(key []byte) string {
  186. sepIndex := len(key) - 1
  187. for sepIndex >= 0 && key[sepIndex] != DIR_FILE_SEPARATOR {
  188. sepIndex--
  189. }
  190. return string(key[sepIndex+1:])
  191. }
  192. func (store *LevelDBStore) Shutdown() {
  193. store.db.Close()
  194. }