This commit is contained in:
chenqieqie 2021-01-04 18:48:55 +08:00
parent 2ce86f308e
commit 40b3207f18

View file

@ -11,7 +11,7 @@ import (
"github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
weed_util "github.com/chrislusf/seaweedfs/weed/util" weed_util "github.com/chrislusf/seaweedfs/weed/util"
"github.com/tecbot/gorocksdb" rocksdb "github.com/tecbot/gorocksdb"
"io" "io"
) )
@ -19,9 +19,28 @@ func init() {
filer.Stores = append(filer.Stores, &RocksDBStore{}) filer.Stores = append(filer.Stores, &RocksDBStore{})
} }
type options struct {
opt *rocksdb.Options
ro *rocksdb.ReadOptions
wo *rocksdb.WriteOptions
}
func (opt *options) init() {
opt.opt = rocksdb.NewDefaultOptions()
opt.ro = rocksdb.NewDefaultReadOptions()
opt.wo = rocksdb.NewDefaultWriteOptions()
}
func (opt *options) close() {
opt.opt.Destroy()
opt.ro.Destroy()
opt.wo.Destroy()
}
type RocksDBStore struct { type RocksDBStore struct {
path string path string
db *gorocksdb.DB db *rocksdb.DB
options
} }
func (store *RocksDBStore) GetName() string { func (store *RocksDBStore) GetName() string {
@ -38,10 +57,9 @@ func (store *RocksDBStore) initialize(dir string) (err error) {
if err := weed_util.TestFolderWritable(dir); err != nil { if err := weed_util.TestFolderWritable(dir); err != nil {
return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err) return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err)
} }
store.options.init()
options := gorocksdb.NewDefaultOptions() store.opt.SetCreateIfMissing(true)
options.SetCreateIfMissing(true) store.db, err = rocksdb.OpenDb(store.opt, dir)
store.db, err = gorocksdb.OpenDb(options, dir)
return return
} }
@ -65,8 +83,7 @@ func (store *RocksDBStore) InsertEntry(ctx context.Context, entry *filer.Entry)
return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err)
} }
wo := gorocksdb.NewDefaultWriteOptions() err = store.db.Put(store.wo, key, value)
err = store.db.Put(wo, key, value)
if err != nil { if err != nil {
return fmt.Errorf("persisting %s : %v", entry.FullPath, err) return fmt.Errorf("persisting %s : %v", entry.FullPath, err)
@ -85,21 +102,21 @@ func (store *RocksDBStore) UpdateEntry(ctx context.Context, entry *filer.Entry)
func (store *RocksDBStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) { func (store *RocksDBStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) {
dir, name := fullpath.DirAndName() dir, name := fullpath.DirAndName()
key := genKey(dir, name) key := genKey(dir, name)
data, err := store.db.Get(store.ro, key)
ro := gorocksdb.NewDefaultReadOptions()
data, err := store.db.GetBytes(ro, key)
if data == nil { if data == nil {
return nil, filer_pb.ErrNotFound return nil, filer_pb.ErrNotFound
} }
defer data.Free()
if err != nil { if err != nil {
return nil, fmt.Errorf("get %s : %v", entry.FullPath, err) return nil, fmt.Errorf("get %s : %v", fullpath, err)
} }
entry = &filer.Entry{ entry = &filer.Entry{
FullPath: fullpath, FullPath: fullpath,
} }
err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(data)) err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(data.Data()))
if err != nil { if err != nil {
return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err) return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err)
} }
@ -113,8 +130,7 @@ func (store *RocksDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.F
dir, name := fullpath.DirAndName() dir, name := fullpath.DirAndName()
key := genKey(dir, name) key := genKey(dir, name)
wo := gorocksdb.NewDefaultWriteOptions() err = store.db.Delete(store.wo, key)
err = store.db.Delete(wo, key)
if err != nil { if err != nil {
return fmt.Errorf("delete %s : %v", fullpath, err) return fmt.Errorf("delete %s : %v", fullpath, err)
} }
@ -125,10 +141,13 @@ func (store *RocksDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.F
func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) { func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) {
directoryPrefix := genDirectoryKeyPrefix(fullpath, "") directoryPrefix := genDirectoryKeyPrefix(fullpath, "")
batch := new(gorocksdb.WriteBatch) batch := rocksdb.NewWriteBatch()
defer batch.Destroy()
ro := gorocksdb.NewDefaultReadOptions() ro := rocksdb.NewDefaultReadOptions()
defer ro.Destroy()
ro.SetFillCache(false) ro.SetFillCache(false)
iter := store.db.NewIterator(ro) iter := store.db.NewIterator(ro)
defer iter.Close() defer iter.Close()
err = enumerate(iter, directoryPrefix, nil, false, -1, func(key, value []byte) bool { err = enumerate(iter, directoryPrefix, nil, false, -1, func(key, value []byte) bool {
@ -139,8 +158,7 @@ func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath we
return fmt.Errorf("delete list %s : %v", fullpath, err) return fmt.Errorf("delete list %s : %v", fullpath, err)
} }
wo := gorocksdb.NewDefaultWriteOptions() err = store.db.Write(store.wo, batch)
err = store.db.Write(wo, batch)
if err != nil { if err != nil {
return fmt.Errorf("delete %s : %v", fullpath, err) return fmt.Errorf("delete %s : %v", fullpath, err)
@ -149,7 +167,7 @@ func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath we
return nil return nil
} }
func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int, fn func(key, value []byte) bool) error { func enumerate(iter *rocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int, fn func(key, value []byte) bool) error {
if len(lastKey) == 0 { if len(lastKey) == 0 {
iter.Seek(prefix) iter.Seek(prefix)
@ -157,11 +175,7 @@ func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey
iter.Seek(lastKey) iter.Seek(lastKey)
if !includeLastKey { if !includeLastKey {
k := iter.Key() key := iter.Key().Data()
v := iter.Value()
key := k.Data()
defer k.Free()
defer v.Free()
if !bytes.HasPrefix(key, prefix) { if !bytes.HasPrefix(key, prefix) {
return nil return nil
@ -184,21 +198,13 @@ func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey
} }
} }
k := iter.Key() key := iter.Key().Data()
v := iter.Value()
key := k.Data()
value := v.Data()
if !bytes.HasPrefix(key, prefix) { if !bytes.HasPrefix(key, prefix) {
k.Free()
v.Free()
break break
} }
ret := fn(key, value) ret := fn(key, iter.Value().Data())
k.Free()
v.Free()
if !ret { if !ret {
break break
@ -225,8 +231,10 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, ful
lastFileStart = genDirectoryKeyPrefix(fullpath, startFileName) lastFileStart = genDirectoryKeyPrefix(fullpath, startFileName)
} }
ro := gorocksdb.NewDefaultReadOptions() ro := rocksdb.NewDefaultReadOptions()
defer ro.Destroy()
ro.SetFillCache(false) ro.SetFillCache(false)
iter := store.db.NewIterator(ro) iter := store.db.NewIterator(ro)
defer iter.Close() defer iter.Close()
err = enumerate(iter, directoryPrefix, lastFileStart, inclusive, limit, func(key, value []byte) bool { err = enumerate(iter, directoryPrefix, lastFileStart, inclusive, limit, func(key, value []byte) bool {
@ -290,4 +298,5 @@ func hashToBytes(dir string) []byte {
func (store *RocksDBStore) Shutdown() { func (store *RocksDBStore) Shutdown() {
store.db.Close() store.db.Close()
store.options.close()
} }