From bbae13e41581d8fa9dca2e5df7823d15ce94afe9 Mon Sep 17 00:00:00 2001 From: qieqieplus Date: Tue, 5 Jan 2021 17:47:59 +0800 Subject: [PATCH 1/2] impl: TTL per entry for rocksdb; fix package name --- weed/filer/rocksdb/rocksdb_store.go | 40 +++++++++++++++++----------- weed/filer/rocksdb/rocksdb_ttl.go | 41 +++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 16 deletions(-) create mode 100644 weed/filer/rocksdb/rocksdb_ttl.go diff --git a/weed/filer/rocksdb/rocksdb_store.go b/weed/filer/rocksdb/rocksdb_store.go index ca6391386..9e058a96d 100644 --- a/weed/filer/rocksdb/rocksdb_store.go +++ b/weed/filer/rocksdb/rocksdb_store.go @@ -7,12 +7,14 @@ import ( "context" "crypto/md5" "fmt" + "io" + + "github.com/tecbot/gorocksdb" + "github.com/chrislusf/seaweedfs/weed/filer" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" weed_util "github.com/chrislusf/seaweedfs/weed/util" - rocksdb "github.com/tecbot/gorocksdb" - "io" ) func init() { @@ -20,15 +22,15 @@ func init() { } type options struct { - opt *rocksdb.Options - ro *rocksdb.ReadOptions - wo *rocksdb.WriteOptions + opt *gorocksdb.Options + ro *gorocksdb.ReadOptions + wo *gorocksdb.WriteOptions } func (opt *options) init() { - opt.opt = rocksdb.NewDefaultOptions() - opt.ro = rocksdb.NewDefaultReadOptions() - opt.wo = rocksdb.NewDefaultWriteOptions() + opt.opt = gorocksdb.NewDefaultOptions() + opt.ro = gorocksdb.NewDefaultReadOptions() + opt.wo = gorocksdb.NewDefaultWriteOptions() } func (opt *options) close() { @@ -39,7 +41,7 @@ func (opt *options) close() { type RocksDBStore struct { path string - db *rocksdb.DB + db *gorocksdb.DB options } @@ -59,7 +61,13 @@ func (store *RocksDBStore) initialize(dir string) (err error) { } store.options.init() store.opt.SetCreateIfMissing(true) - store.db, err = rocksdb.OpenDb(store.opt, dir) + // reduce write amplification + // also avoid expired data stored in highest level never get compacted + store.opt.SetLevelCompactionDynamicLevelBytes(true) + store.opt.SetCompactionFilter(NewTTLFilter()) + // store.opt.SetMaxBackgroundCompactions(2) + + store.db, err = gorocksdb.OpenDb(store.opt, dir) return } @@ -116,7 +124,7 @@ func (store *RocksDBStore) FindEntry(ctx context.Context, fullpath weed_util.Ful entry = &filer.Entry{ FullPath: fullpath, } - err = entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(data.Data())) + err = entry.DecodeAttributesAndChunks(data.Data()) if err != nil { return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err) } @@ -141,10 +149,10 @@ func (store *RocksDBStore) DeleteEntry(ctx context.Context, fullpath weed_util.F func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath weed_util.FullPath) (err error) { directoryPrefix := genDirectoryKeyPrefix(fullpath, "") - batch := rocksdb.NewWriteBatch() + batch := gorocksdb.NewWriteBatch() defer batch.Destroy() - ro := rocksdb.NewDefaultReadOptions() + ro := gorocksdb.NewDefaultReadOptions() defer ro.Destroy() ro.SetFillCache(false) @@ -167,7 +175,7 @@ func (store *RocksDBStore) DeleteFolderChildren(ctx context.Context, fullpath we return nil } -func enumerate(iter *rocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int, fn func(key, value []byte) bool) error { +func enumerate(iter *gorocksdb.Iterator, prefix, lastKey []byte, includeLastKey bool, limit int, fn func(key, value []byte) bool) error { if len(lastKey) == 0 { iter.Seek(prefix) @@ -231,7 +239,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, ful lastFileStart = genDirectoryKeyPrefix(fullpath, startFileName) } - ro := rocksdb.NewDefaultReadOptions() + ro := gorocksdb.NewDefaultReadOptions() defer ro.Destroy() ro.SetFillCache(false) @@ -251,7 +259,7 @@ func (store *RocksDBStore) ListDirectoryPrefixedEntries(ctx context.Context, ful } // println("list", entry.FullPath, "chunks", len(entry.Chunks)) - if decodeErr := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(value)); decodeErr != nil { + if decodeErr := entry.DecodeAttributesAndChunks(value); decodeErr != nil { err = decodeErr glog.V(0).Infof("list %s : %v", entry.FullPath, err) return false diff --git a/weed/filer/rocksdb/rocksdb_ttl.go b/weed/filer/rocksdb/rocksdb_ttl.go new file mode 100644 index 000000000..1fc239c0c --- /dev/null +++ b/weed/filer/rocksdb/rocksdb_ttl.go @@ -0,0 +1,41 @@ +//+build rocksdb + +package rocksdb + +import ( + "time" + + "github.com/tecbot/gorocksdb" + + "github.com/chrislusf/seaweedfs/weed/filer" +) + +type TTLFilter struct { + skipLevel0 bool +} + +func NewTTLFilter() gorocksdb.CompactionFilter { + return &TTLFilter{ + skipLevel0: true, + } +} + +func (t *TTLFilter) Filter(level int, key, val []byte) (remove bool, newVal []byte) { + // decode could be slow, causing write stall + // level >0 sst can run compaction in parallel + if t.skipLevel0 && level == 0 { + return false, val + } + entry := filer.Entry{} + if err := entry.DecodeAttributesAndChunks(val); err == nil { + if entry.TtlSec == 0 || + entry.Crtime.Add(time.Duration(entry.TtlSec)*time.Second).Before(time.Now()) { + return false, val + } + } + return true, nil +} + +func (t *TTLFilter) Name() string { + return "TTLFilter" +} From b2e50f602f23f919b6837b1207056cf1ebc7fc94 Mon Sep 17 00:00:00 2001 From: qieqieplus Date: Tue, 5 Jan 2021 19:36:40 +0800 Subject: [PATCH 2/2] fix time comparison --- weed/filer/rocksdb/rocksdb_ttl.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/weed/filer/rocksdb/rocksdb_ttl.go b/weed/filer/rocksdb/rocksdb_ttl.go index 1fc239c0c..98918b5d7 100644 --- a/weed/filer/rocksdb/rocksdb_ttl.go +++ b/weed/filer/rocksdb/rocksdb_ttl.go @@ -29,7 +29,7 @@ func (t *TTLFilter) Filter(level int, key, val []byte) (remove bool, newVal []by entry := filer.Entry{} if err := entry.DecodeAttributesAndChunks(val); err == nil { if entry.TtlSec == 0 || - entry.Crtime.Add(time.Duration(entry.TtlSec)*time.Second).Before(time.Now()) { + entry.Crtime.Add(time.Duration(entry.TtlSec)*time.Second).After(time.Now()) { return false, val } }