diff --git a/weed/command/scaffold/filer.toml b/weed/command/scaffold/filer.toml index 67bfea93aa..8b9552e052 100644 --- a/weed/command/scaffold/filer.toml +++ b/weed/command/scaffold/filer.toml @@ -33,12 +33,22 @@ recursive_delete = false # faster than previous leveldb, recommended. enabled = true dir = "./filerldb2" # directory to store level db files +# optional tuning, applied to each of the 8 partitioned stores +# block_cache_size_mb = 32 # sorted-table read cache +# write_buffer_size_mb = 16 # memtable size before flushing to disk +# block_size_kb = 4 # uncompressed block ('page') size inside table files +# compaction_table_size_mb = 2 # target size of generated table files [leveldb3] # similar to leveldb2. # each bucket has its own meta store. enabled = false dir = "./filerldb3" # directory to store level db files +# optional tuning, applied to the default store and each bucket store +# block_cache_size_mb = 32 # sorted-table read cache +# write_buffer_size_mb = 16 # memtable size before flushing to disk +# block_size_kb = 4 # uncompressed block ('page') size inside table files +# compaction_table_size_mb = 2 # target size of generated table files [rocksdb] # local on disk, similar to leveldb diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index c3bb2ac55b..93c587d703 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/weed/filer/leveldb2/leveldb2_store.go @@ -6,6 +6,7 @@ import ( "crypto/md5" "fmt" "io" + "math" "os" "github.com/syndtr/goleveldb/leveldb" @@ -28,6 +29,11 @@ type LevelDB2Store struct { dbs []*leveldb.DB dbCount int ReadOnly bool + + blockCacheSizeMB int + writeBufferSizeMB int + blockSizeKB int + compactionTableSizeMB int } func (store *LevelDB2Store) GetName() string { @@ -36,9 +42,54 @@ func (store *LevelDB2Store) GetName() string { func (store *LevelDB2Store) Initialize(configuration weed_util.Configuration, prefix string) (err error) { dir := configuration.GetString(prefix + "dir") + store.blockCacheSizeMB = configuration.GetInt(prefix + "block_cache_size_mb") + store.writeBufferSizeMB = configuration.GetInt(prefix + "write_buffer_size_mb") + store.blockSizeKB = configuration.GetInt(prefix + "block_size_kb") + store.compactionTableSizeMB = configuration.GetInt(prefix + "compaction_table_size_mb") + if err := store.validateSizeOptions(); err != nil { + return err + } return store.initialize(dir, 8) } +// validateSizeOptions rejects sizes too large to convert to bytes without +// overflowing int on this platform. +func (store *LevelDB2Store) validateSizeOptions() error { + for name, kv := range map[string][2]int{ + "block_cache_size_mb": {store.blockCacheSizeMB, 1 << 20}, + "write_buffer_size_mb": {store.writeBufferSizeMB, 1 << 20}, + "block_size_kb": {store.blockSizeKB, 1 << 10}, + "compaction_table_size_mb": {store.compactionTableSizeMB, 1 << 20}, + } { + if kv[0] > math.MaxInt/kv[1] { + return fmt.Errorf("leveldb2 option %s=%d overflows int", name, kv[0]) + } + } + return nil +} + +func (store *LevelDB2Store) buildOptions() *opt.Options { + opts := &opt.Options{ + BlockCacheCapacity: 32 * 1024 * 1024, // default value is 8MiB + WriteBuffer: 16 * 1024 * 1024, // default value is 4MiB + Filter: filter.NewBloomFilter(8), // false positive rate 0.02 + ReadOnly: store.ReadOnly, + } + if store.blockCacheSizeMB > 0 { + opts.BlockCacheCapacity = store.blockCacheSizeMB * 1024 * 1024 + } + if store.writeBufferSizeMB > 0 { + opts.WriteBuffer = store.writeBufferSizeMB * 1024 * 1024 + } + if store.blockSizeKB > 0 { + opts.BlockSize = store.blockSizeKB * 1024 + } + if store.compactionTableSizeMB > 0 { + opts.CompactionTableSize = store.compactionTableSizeMB * 1024 * 1024 + } + return opts +} + func (store *LevelDB2Store) initialize(dir string, dbCount int) (err error) { glog.Infof("filer store leveldb2 dir: %s", dir) os.MkdirAll(dir, 0755) @@ -46,12 +97,7 @@ func (store *LevelDB2Store) initialize(dir string, dbCount int) (err error) { return fmt.Errorf("Check Level Folder %s Writable: %s", dir, err) } - opts := &opt.Options{ - BlockCacheCapacity: 32 * 1024 * 1024, // default value is 8MiB - WriteBuffer: 16 * 1024 * 1024, // default value is 4MiB - Filter: filter.NewBloomFilter(8), // false positive rate 0.02 - ReadOnly: store.ReadOnly, - } + opts := store.buildOptions() for d := 0; d < dbCount; d++ { dbFolder := fmt.Sprintf("%s/%02d", dir, d) diff --git a/weed/filer/leveldb2/leveldb2_store_test.go b/weed/filer/leveldb2/leveldb2_store_test.go index 5c6a3a4d71..9651711d92 100644 --- a/weed/filer/leveldb2/leveldb2_store_test.go +++ b/weed/filer/leveldb2/leveldb2_store_test.go @@ -2,6 +2,8 @@ package leveldb import ( "context" + "math" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -62,6 +64,93 @@ func TestCreateAndFind(t *testing.T) { } +type fakeConfiguration map[string]interface{} + +func (c fakeConfiguration) GetString(key string) string { + value, _ := c[key].(string) + return value +} + +func (c fakeConfiguration) GetBool(key string) bool { + value, _ := c[key].(bool) + return value +} + +func (c fakeConfiguration) GetInt(key string) int { + value, _ := c[key].(int) + return value +} + +func (c fakeConfiguration) GetStringSlice(key string) []string { + value, _ := c[key].([]string) + return value +} + +func (c fakeConfiguration) SetDefault(key string, value interface{}) { + if _, found := c[key]; !found { + c[key] = value + } +} + +func TestInitializeOptions(t *testing.T) { + store := &LevelDB2Store{} + if err := store.Initialize(fakeConfiguration{ + "leveldb2.dir": t.TempDir(), + "leveldb2.block_cache_size_mb": 64, + "leveldb2.write_buffer_size_mb": 32, + "leveldb2.block_size_kb": 16, + "leveldb2.compaction_table_size_mb": 8, + }, "leveldb2."); err != nil { + t.Fatalf("initialize: %v", err) + } + defer store.Shutdown() + + opts := store.buildOptions() + if opts.BlockCacheCapacity != 64*1024*1024 { + t.Errorf("block cache %d, want 64MiB", opts.BlockCacheCapacity) + } + if opts.WriteBuffer != 32*1024*1024 { + t.Errorf("write buffer %d, want 32MiB", opts.WriteBuffer) + } + if opts.BlockSize != 16*1024 { + t.Errorf("block size %d, want 16KiB", opts.BlockSize) + } + if opts.CompactionTableSize != 8*1024*1024 { + t.Errorf("compaction table size %d, want 8MiB", opts.CompactionTableSize) + } +} + +func TestInitializeDefaultOptions(t *testing.T) { + store := &LevelDB2Store{} + if err := store.Initialize(fakeConfiguration{ + "leveldb2.dir": t.TempDir(), + "leveldb2.block_cache_size_mb": -1, + "leveldb2.write_buffer_size_mb": 0, + }, "leveldb2."); err != nil { + t.Fatalf("initialize: %v", err) + } + defer store.Shutdown() + + opts := store.buildOptions() + if opts.BlockCacheCapacity != 32*1024*1024 { + t.Errorf("block cache %d, want the 32MiB default", opts.BlockCacheCapacity) + } + if opts.WriteBuffer != 16*1024*1024 { + t.Errorf("write buffer %d, want the 16MiB default", opts.WriteBuffer) + } +} + +func TestInitializeRejectsOversizedOption(t *testing.T) { + store := &LevelDB2Store{} + err := store.Initialize(fakeConfiguration{ + "leveldb2.dir": t.TempDir(), + "leveldb2.write_buffer_size_mb": math.MaxInt/(1<<20) + 1, + }, "leveldb2.") + if err == nil || !strings.Contains(err.Error(), "write_buffer_size_mb") { + t.Fatalf("expected overflow rejection, got %v", err) + } +} + func TestEmptyRoot(t *testing.T) { testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) dir := t.TempDir() diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index 564d6eaddd..5e012ba673 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/weed/filer/leveldb3/leveldb3_store.go @@ -6,6 +6,7 @@ import ( "crypto/md5" "fmt" "io" + "math" "os" "strings" "sync" @@ -35,6 +36,11 @@ type LevelDB3Store struct { dbs map[string]*leveldb.DB dbsLock sync.RWMutex ReadOnly bool + + blockCacheSizeMB int + writeBufferSizeMB int + blockSizeKB int + compactionTableSizeMB int } func (store *LevelDB3Store) GetName() string { @@ -43,9 +49,58 @@ func (store *LevelDB3Store) GetName() string { func (store *LevelDB3Store) Initialize(configuration weed_util.Configuration, prefix string) (err error) { dir := configuration.GetString(prefix + "dir") + store.blockCacheSizeMB = configuration.GetInt(prefix + "block_cache_size_mb") + store.writeBufferSizeMB = configuration.GetInt(prefix + "write_buffer_size_mb") + store.blockSizeKB = configuration.GetInt(prefix + "block_size_kb") + store.compactionTableSizeMB = configuration.GetInt(prefix + "compaction_table_size_mb") + if err := store.validateSizeOptions(); err != nil { + return err + } return store.initialize(dir) } +// validateSizeOptions rejects sizes too large to convert to bytes without +// overflowing int on this platform. +func (store *LevelDB3Store) validateSizeOptions() error { + for name, kv := range map[string][2]int{ + "block_cache_size_mb": {store.blockCacheSizeMB, 1 << 20}, + "write_buffer_size_mb": {store.writeBufferSizeMB, 1 << 20}, + "block_size_kb": {store.blockSizeKB, 1 << 10}, + "compaction_table_size_mb": {store.compactionTableSizeMB, 1 << 20}, + } { + if kv[0] > math.MaxInt/kv[1] { + return fmt.Errorf("leveldb3 option %s=%d overflows int", name, kv[0]) + } + } + return nil +} + +func (store *LevelDB3Store) buildOptions(name string) *opt.Options { + opts := &opt.Options{ + BlockCacheCapacity: 32 * 1024 * 1024, // default value is 8MiB + WriteBuffer: 16 * 1024 * 1024, // default value is 4MiB + Filter: filter.NewBloomFilter(8), + ReadOnly: store.ReadOnly, + } + if name != DEFAULT { + opts.BlockCacheCapacity = 16 * 1024 * 1024 // default value is 8MiB + opts.WriteBuffer = 8 * 1024 * 1024 // default value is 4MiB + } + if store.blockCacheSizeMB > 0 { + opts.BlockCacheCapacity = store.blockCacheSizeMB * 1024 * 1024 + } + if store.writeBufferSizeMB > 0 { + opts.WriteBuffer = store.writeBufferSizeMB * 1024 * 1024 + } + if store.blockSizeKB > 0 { + opts.BlockSize = store.blockSizeKB * 1024 + } + if store.compactionTableSizeMB > 0 { + opts.CompactionTableSize = store.compactionTableSizeMB * 1024 * 1024 + } + return opts +} + func (store *LevelDB3Store) initialize(dir string) (err error) { glog.Infof("filer store leveldb3 dir: %s", dir) os.MkdirAll(dir, 0755) @@ -65,21 +120,7 @@ func (store *LevelDB3Store) initialize(dir string) (err error) { } func (store *LevelDB3Store) loadDB(name string) (*leveldb.DB, error) { - bloom := filter.NewBloomFilter(8) // false positive rate 0.02 - opts := &opt.Options{ - BlockCacheCapacity: 32 * 1024 * 1024, // default value is 8MiB - WriteBuffer: 16 * 1024 * 1024, // default value is 4MiB - Filter: bloom, - ReadOnly: store.ReadOnly, - } - if name != DEFAULT { - opts = &opt.Options{ - BlockCacheCapacity: 16 * 1024 * 1024, // default value is 8MiB - WriteBuffer: 8 * 1024 * 1024, // default value is 4MiB - Filter: bloom, - ReadOnly: store.ReadOnly, - } - } + opts := store.buildOptions(name) dbFolder := fmt.Sprintf("%s/%s", store.dir, name) os.MkdirAll(dbFolder, 0755) diff --git a/weed/filer/leveldb3/leveldb3_store_test.go b/weed/filer/leveldb3/leveldb3_store_test.go index 8a3cd2d82e..62896a299f 100644 --- a/weed/filer/leveldb3/leveldb3_store_test.go +++ b/weed/filer/leveldb3/leveldb3_store_test.go @@ -2,6 +2,8 @@ package leveldb import ( "context" + "math" + "strings" "testing" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -62,6 +64,103 @@ func TestCreateAndFind(t *testing.T) { } +type fakeConfiguration map[string]interface{} + +func (c fakeConfiguration) GetString(key string) string { + value, _ := c[key].(string) + return value +} + +func (c fakeConfiguration) GetBool(key string) bool { + value, _ := c[key].(bool) + return value +} + +func (c fakeConfiguration) GetInt(key string) int { + value, _ := c[key].(int) + return value +} + +func (c fakeConfiguration) GetStringSlice(key string) []string { + value, _ := c[key].([]string) + return value +} + +func (c fakeConfiguration) SetDefault(key string, value interface{}) { + if _, found := c[key]; !found { + c[key] = value + } +} + +func TestInitializeOptions(t *testing.T) { + store := &LevelDB3Store{} + if err := store.Initialize(fakeConfiguration{ + "leveldb3.dir": t.TempDir(), + "leveldb3.block_cache_size_mb": 64, + "leveldb3.write_buffer_size_mb": 32, + "leveldb3.block_size_kb": 16, + "leveldb3.compaction_table_size_mb": 8, + }, "leveldb3."); err != nil { + t.Fatalf("initialize: %v", err) + } + defer store.Shutdown() + + for _, name := range []string{DEFAULT, "bucket-a"} { + opts := store.buildOptions(name) + if opts.BlockCacheCapacity != 64*1024*1024 { + t.Errorf("%s block cache %d, want 64MiB", name, opts.BlockCacheCapacity) + } + if opts.WriteBuffer != 32*1024*1024 { + t.Errorf("%s write buffer %d, want 32MiB", name, opts.WriteBuffer) + } + if opts.BlockSize != 16*1024 { + t.Errorf("%s block size %d, want 16KiB", name, opts.BlockSize) + } + if opts.CompactionTableSize != 8*1024*1024 { + t.Errorf("%s compaction table size %d, want 8MiB", name, opts.CompactionTableSize) + } + } +} + +func TestInitializeDefaultOptions(t *testing.T) { + store := &LevelDB3Store{} + if err := store.Initialize(fakeConfiguration{ + "leveldb3.dir": t.TempDir(), + "leveldb3.block_cache_size_mb": -1, + "leveldb3.write_buffer_size_mb": 0, + }, "leveldb3."); err != nil { + t.Fatalf("initialize: %v", err) + } + defer store.Shutdown() + + opts := store.buildOptions(DEFAULT) + if opts.BlockCacheCapacity != 32*1024*1024 { + t.Errorf("block cache %d, want the 32MiB default", opts.BlockCacheCapacity) + } + if opts.WriteBuffer != 16*1024*1024 { + t.Errorf("write buffer %d, want the 16MiB default", opts.WriteBuffer) + } + + bucketOpts := store.buildOptions("bucket-a") + if bucketOpts.BlockCacheCapacity != 16*1024*1024 { + t.Errorf("bucket block cache %d, want the 16MiB default", bucketOpts.BlockCacheCapacity) + } + if bucketOpts.WriteBuffer != 8*1024*1024 { + t.Errorf("bucket write buffer %d, want the 8MiB default", bucketOpts.WriteBuffer) + } +} + +func TestInitializeRejectsOversizedOption(t *testing.T) { + store := &LevelDB3Store{} + err := store.Initialize(fakeConfiguration{ + "leveldb3.dir": t.TempDir(), + "leveldb3.write_buffer_size_mb": math.MaxInt/(1<<20) + 1, + }, "leveldb3.") + if err == nil || !strings.Contains(err.Error(), "write_buffer_size_mb") { + t.Fatalf("expected overflow rejection, got %v", err) + } +} + func TestEmptyRoot(t *testing.T) { testFiler := filer.NewFiler(pb.ServerDiscovery{}, nil, "", "", "", "", "", 255, nil) dir := t.TempDir()