leveldb_store.go 6.8 KB

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