leveldb2_store.go 7.2 KB

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