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.

302 lines
7.2 KiB

4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
4 years ago
  1. // +build rocksdb
  2. package rocksdb
  3. import (
  4. "bytes"
  5. "context"
  6. "crypto/md5"
  7. "fmt"
  8. "io"
  9. "github.com/tecbot/gorocksdb"
  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. func init() {
  16. filer.Stores = append(filer.Stores, &RocksDBStore{})
  17. }
  18. type options struct {
  19. opt *gorocksdb.Options
  20. ro *gorocksdb.ReadOptions
  21. wo *gorocksdb.WriteOptions
  22. }
  23. func (opt *options) init() {
  24. opt.opt = gorocksdb.NewDefaultOptions()
  25. opt.ro = gorocksdb.NewDefaultReadOptions()
  26. opt.wo = gorocksdb.NewDefaultWriteOptions()
  27. }
  28. func (opt *options) close() {
  29. opt.opt.Destroy()
  30. opt.ro.Destroy()
  31. opt.wo.Destroy()
  32. }
  33. type RocksDBStore struct {
  34. path string
  35. db *gorocksdb.DB
  36. options
  37. }
  38. func (store *RocksDBStore) GetName() string {
  39. return "rocksdb"
  40. }
  41. func (store *RocksDBStore) Initialize(configuration weed_util.Configuration, prefix string) (err error) {
  42. dir := configuration.GetString(prefix + "dir")
  43. return store.initialize(dir)
  44. }
  45. func (store *RocksDBStore) initialize(dir string) (err error) {
  46. glog.Infof("filer store rocksdb dir: %s", dir)
  47. if err := weed_util.TestFolderWritable(dir); err != nil {
  48. return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err)
  49. }
  50. store.options.init()
  51. store.opt.SetCreateIfMissing(true)
  52. // reduce write amplification
  53. // also avoid expired data stored in highest level never get compacted
  54. store.opt.SetLevelCompactionDynamicLevelBytes(true)
  55. store.opt.SetCompactionFilter(NewTTLFilter())
  56. // store.opt.SetMaxBackgroundCompactions(2)
  57. store.db, err = gorocksdb.OpenDb(store.opt, dir)
  58. return
  59. }
  60. func (store *RocksDBStore) BeginTransaction(ctx context.Context) (context.Context, error) {
  61. return ctx, nil
  62. }
  63. func (store *RocksDBStore) CommitTransaction(ctx context.Context) error {
  64. return nil
  65. }
  66. func (store *RocksDBStore) RollbackTransaction(ctx context.Context) error {
  67. return nil
  68. }
  69. func (store *RocksDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) {
  70. dir, name := entry.DirAndName()
  71. key := genKey(dir, name)
  72. value, err := entry.EncodeAttributesAndChunks()
  73. if err != nil {
  74. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
  75. }
  76. err = store.db.Put(store.wo, key, value)
  77. if err != nil {
  78. return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
  79. }
  80. // println("saved", entry.FullPath, "chunks", len(entry.Chunks))
  81. return nil
  82. }
  83. func (store *RocksDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) {
  84. return store.InsertEntry(ctx, entry)
  85. }
  86. func (store *RocksDBStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
  87. dir, name := fullpath.DirAndName()
  88. key := genKey(dir, name)
  89. data, err := store.db.Get(store.ro, key)
  90. if data == nil {
  91. return nil, filer_pb.ErrNotFound
  92. }
  93. defer data.Free()
  94. if err != nil {
  95. return nil, fmt.Errorf("get %s : %v", fullpath, err)
  96. }
  97. entry = &filer.Entry{
  98. FullPath: fullpath,
  99. }
  100. err = entry.DecodeAttributesAndChunks(data.Data())
  101. if err != nil {
  102. return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
  103. }
  104. // println("read", entry.FullPath, "chunks", len(entry.Chunks), "data", len(data), string(data))
  105. return entry, nil
  106. }
  107. func (store *RocksDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  108. dir, name := fullpath.DirAndName()
  109. key := genKey(dir, name)
  110. err = store.db.Delete(store.wo, key)
  111. if err != nil {
  112. return fmt.Errorf("delete %s : %v", fullpath, err)
  113. }
  114. return nil
  115. }
  116. func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
  117. directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
  118. batch := gorocksdb.NewWriteBatch()
  119. defer batch.Destroy()
  120. ro := gorocksdb.NewDefaultReadOptions()
  121. defer ro.Destroy()
  122. ro.SetFillCache(false)
  123. iter := store.db.NewIterator(ro)
  124. defer iter.Close()
  125. err = enumerate(iter, directoryPrefix, nil, false, -1, func(key, value []byte) bool {
  126. batch.Delete(key)
  127. return true
  128. })
  129. if err != nil {
  130. return fmt.Errorf("delete list %s : %v", fullpath, err)
  131. }
  132. err = store.db.Write(store.wo, batch)
  133. if err != nil {
  134. return fmt.Errorf("delete %s : %v", fullpath, err)
  135. }
  136. return nil
  137. }
  138. func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int64, fn func(key, value []byte) bool) (err error) {
  139. if len(lastKey) == 0 {
  140. iter.Seek(prefix)
  141. } else {
  142. iter.Seek(lastKey)
  143. if !includeLastKey {
  144. if iter.Valid() {
  145. if bytes.Equal(iter.Key().Data(), lastKey) {
  146. iter.Next()
  147. }
  148. }
  149. }
  150. }
  151. i := int64(0)
  152. for ; iter.Valid(); iter.Next() {
  153. if limit > 0 {
  154. i++
  155. if i > limit {
  156. break
  157. }
  158. }
  159. key := iter.Key().Data()
  160. if !bytes.HasPrefix(key, prefix) {
  161. break
  162. }
  163. ret := fn(key, iter.Value().Data())
  164. if !ret {
  165. break
  166. }
  167. }
  168. if err := iter.Err(); err != nil {
  169. return fmt.Errorf("prefix scan iterator: %v", err)
  170. }
  171. return nil
  172. }
  173. func (store *RocksDBStore) ListDirectoryEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  174. return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", eachEntryFunc)
  175. }
  176. func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath weed_util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) {
  177. directoryPrefix := genDirectoryKeyPrefix(dirPath, prefix)
  178. lastFileStart := directoryPrefix
  179. if startFileName != "" {
  180. lastFileStart = genDirectoryKeyPrefix(dirPath, startFileName)
  181. }
  182. ro := gorocksdb.NewDefaultReadOptions()
  183. defer ro.Destroy()
  184. ro.SetFillCache(false)
  185. iter := store.db.NewIterator(ro)
  186. defer iter.Close()
  187. err = enumerate(iter, directoryPrefix, lastFileStart, includeStartFile, limit, func(key, value []byte) bool {
  188. fileName := getNameFromKey(key)
  189. if fileName == "" {
  190. return true
  191. }
  192. entry := &filer.Entry{
  193. FullPath: weed_util.NewFullPath(string(dirPath), fileName),
  194. }
  195. lastFileName = fileName
  196. // println("list", entry.FullPath, "chunks", len(entry.Chunks))
  197. if decodeErr := entry.DecodeAttributesAndChunks(value); decodeErr != nil {
  198. err = decodeErr
  199. glog.V(0).Infof("list %s : %v", entry.FullPath, err)
  200. return false
  201. }
  202. if !eachEntryFunc(entry) {
  203. return false
  204. }
  205. return true
  206. })
  207. if err != nil {
  208. return lastFileName, fmt.Errorf("prefix list %s : %v", dirPath, err)
  209. }
  210. return lastFileName, err
  211. }
  212. func genKey(dirPath, fileName string) (key []byte) {
  213. key = hashToBytes(dirPath)
  214. key = append(key, []byte(fileName)...)
  215. return key
  216. }
  217. func genDirectoryKeyPrefix(fullpath weed_util.FullPath, startFileName string) (keyPrefix []byte) {
  218. keyPrefix = hashToBytes(string(fullpath))
  219. if len(startFileName) > 0 {
  220. keyPrefix = append(keyPrefix, []byte(startFileName)...)
  221. }
  222. return keyPrefix
  223. }
  224. func getNameFromKey(key []byte) string {
  225. return string(key[md5.Size:])
  226. }
  227. // hash directory, and use last byte for partitioning
  228. func hashToBytes(dir string) []byte {
  229. h := md5.New()
  230. io.WriteString(h, dir)
  231. b := h.Sum(nil)
  232. return b
  233. }
  234. func (store *RocksDBStore) Shutdown() {
  235. store.db.Close()
  236. store.options.close()
  237. }