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.

300 lines
7.0 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
  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 int, fn func(key, value []byte) bool) 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 := 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, fullpath weed_util.FullPath, startFileName string, inclusive bool,
  174. limit int) (entries []*filer.Entry, err error) {
  175. return store.ListDirectoryPrefixedEntries(ctx, fullpath, startFileName, inclusive, limit, "")
  176. }
  177. func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, fullpath weed_util.FullPath, startFileName string, inclusive bool, limit int, prefix string) (entries []*filer.Entry, err error) {
  178. directoryPrefix := genDirectoryKeyPrefix(fullpath, prefix)
  179. lastFileStart := directoryPrefix
  180. if startFileName != "" {
  181. lastFileStart = genDirectoryKeyPrefix(fullpath, startFileName)
  182. }
  183. ro := gorocksdb.NewDefaultReadOptions()
  184. defer ro.Destroy()
  185. ro.SetFillCache(false)
  186. iter := store.db.NewIterator(ro)
  187. defer iter.Close()
  188. err = enumerate(iter, directoryPrefix, lastFileStart, inclusive, limit, func(key, value []byte) bool {
  189. fileName := getNameFromKey(key)
  190. if fileName == "" {
  191. return true
  192. }
  193. entry := &filer.Entry{
  194. FullPath: weed_util.NewFullPath(string(fullpath), fileName),
  195. }
  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. entries = append(entries, entry)
  203. return true
  204. })
  205. if err != nil {
  206. return entries, fmt.Errorf("prefix list %s : %v", fullpath, err)
  207. }
  208. return entries, err
  209. }
  210. func genKey(dirPath, fileName string) (key []byte) {
  211. key = hashToBytes(dirPath)
  212. key = append(key, []byte(fileName)...)
  213. return key
  214. }
  215. func genDirectoryKeyPrefix(fullpath weed_util.FullPath, startFileName string) (keyPrefix []byte) {
  216. keyPrefix = hashToBytes(string(fullpath))
  217. if len(startFileName) > 0 {
  218. keyPrefix = append(keyPrefix, []byte(startFileName)...)
  219. }
  220. return keyPrefix
  221. }
  222. func getNameFromKey(key []byte) string {
  223. return string(key[md5.Size:])
  224. }
  225. // hash directory, and use last byte for partitioning
  226. func hashToBytes(dir string) []byte {
  227. h := md5.New()
  228. io.WriteString(h, dir)
  229. b := h.Sum(nil)
  230. return b
  231. }
  232. func (store *RocksDBStore) Shutdown() {
  233. store.db.Close()
  234. store.options.close()
  235. }