From 957cba17681281d5ad0aa8d0e35fb15d55d94bcf Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Mon, 26 Oct 2020 20:21:48 +0000 Subject: [PATCH] performance tuning: rbfcfg package, binary search for wal segment - rbfcfg package holds Config for --rbf- command line flags - wal.go: replace linear search with bisection for wal segment --- bolt.go | 3 +- bolt_test.go | 2 +- ctl/server.go | 4 +++ dbshard.go | 12 ++++++-- dbshard_internal_test.go | 2 +- holder.go | 6 ++++ lmdb.go | 4 ++- lmdb_other.go | 3 +- lmdb_test.go | 2 +- rbf.go | 12 ++++++-- rbf/cfg/cfg.go | 61 ++++++++++++++++++++++++++++++++++++++++ rbf/{ => cfg}/os.go | 2 +- rbf/{ => cfg}/os_386.go | 2 +- rbf/cursorx.go | 2 +- rbf/db.go | 58 ++++++++++++++++++++++---------------- rbf/db_test.go | 25 +++++++++++++++- rbf/rbf.go | 2 +- rbf/rbf_test.go | 4 +-- rbf/tx.go | 2 +- rbf/wal.go | 13 +++++++-- rrtx.go | 3 +- rrtx_internal_test.go | 2 +- server.go | 15 ++++++++++ server/config.go | 6 ++++ server/server.go | 1 + txfactory.go | 10 ++++++- 26 files changed, 208 insertions(+), 50 deletions(-) create mode 100644 rbf/cfg/cfg.go rename rbf/{ => cfg}/os.go (98%) rename rbf/{ => cfg}/os_386.go (98%) diff --git a/bolt.go b/bolt.go index 39106fecf..f2f829bde 100644 --- a/bolt.go +++ b/bolt.go @@ -30,6 +30,7 @@ import ( "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" @@ -122,7 +123,7 @@ func boltPath(path string) string { // if one does not exist for its bpath. Otherwise it returns // the existing instance. This insures only one boltDB // per bpath in this pilosa node. -func (r *boltRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrapper, error) { +func (r *boltRegistrar) OpenDBWrapper(path0 string, doAllocZero bool, rbfcfg *rbfcfg.Config) (DBWrapper, error) { path := boltPath(path0) r.mu.Lock() diff --git a/bolt_test.go b/bolt_test.go index 1030a5fe2..a15ef044e 100644 --- a/bolt_test.go +++ b/bolt_test.go @@ -87,7 +87,7 @@ func mustOpenEmptyBoltWrapper(path string) (w *BoltWrapper, cleaner func()) { var err error fn := boltPath(path) panicOn(os.RemoveAll(fn)) - ww, err := globalBoltReg.OpenDBWrapper(fn, DetectMemAccessPastTx) + ww, err := globalBoltReg.OpenDBWrapper(fn, DetectMemAccessPastTx, nil) panicOn(err) w = ww.(*BoltWrapper) diff --git a/ctl/server.go b/ctl/server.go index 204a368bf..629aac3dd 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -94,6 +94,9 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { // RowcacheOff flags.BoolVarP((&srv.Config.RowcacheOff), "rowcache-off", "", srv.Config.RowcacheOff, "turn off the rowcache for all backends (reduces memory use)") + // RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions. + srv.Config.RBFConfig.DefineFlags(flags) + // Postgres endpoint flags.StringVar(&srv.Config.Postgres.Bind, "postgres.bind", srv.Config.Postgres.Bind, "Address to which to bind a postgres endpoint (leave blank to disable)") SetTLSConfig(flags, "postgres.", &srv.Config.Postgres.TLS.CertificatePath, &srv.Config.Postgres.TLS.CertificateKeyPath, &srv.Config.Postgres.TLS.CACertPath, &srv.Config.Postgres.TLS.SkipVerify, &srv.Config.Postgres.TLS.EnableClientVerification) @@ -102,4 +105,5 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.DurationVar((*time.Duration)(&srv.Config.Postgres.WriteTimeout), "postgres.write-timeout", time.Duration(srv.Config.Postgres.WriteTimeout), "Timeout for writes on a postgres connection. (set 0 to disable)") flags.Uint32Var(&srv.Config.Postgres.MaxStartupSize, "postgres.max-startup-size", srv.Config.Postgres.MaxStartupSize, "Maximum acceptable size of a postgres startup packet, in bytes. (set 0 to disable)") flags.Uint16Var(&srv.Config.Postgres.ConnectionLimit, "postgres.connection-limit", srv.Config.Postgres.ConnectionLimit, "Maximum number of simultaneous postgres connections to allow. (set 0 to disable)") + } diff --git a/dbshard.go b/dbshard.go index a0912c732..6fc057200 100644 --- a/dbshard.go +++ b/dbshard.go @@ -23,6 +23,7 @@ import ( "strings" "sync" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pkg/errors" ) @@ -58,7 +59,7 @@ type DBWrapper interface { } type DBRegistry interface { - OpenDBWrapper(path string, doAllocZero bool) (DBWrapper, error) + OpenDBWrapper(path string, doAllocZero bool, rbfcfg *rbfcfg.Config) (DBWrapper, error) } type DBShard struct { @@ -239,6 +240,8 @@ type DBPerShard struct { index2shards map[txtype]map[string]*shardSet isBlueGreen bool + + RBFConfig *rbfcfg.Config } func newIndex2Shards() (r map[txtype]map[string]*shardSet) { @@ -340,6 +343,10 @@ func (per *DBPerShard) LoadExistingDBs() (err error) { func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string, holder *Holder) (d *DBPerShard) { + if holder.cfg == nil || holder.cfg.RBFConfig == nil { + panic("must have holder.cfg.RBFConfig set here") + } + useOpenList := 0 hasRoaring := false if types[0] == roaringTxn { @@ -367,6 +374,7 @@ func (txf *TxFactory) NewDBPerShard(types []txtype, holderDir string, holder *Ho hasRoaring: hasRoaring, isBlueGreen: len(types) > 1, index2shards: newIndex2Shards(), + RBFConfig: holder.cfg.RBFConfig, } return } @@ -586,7 +594,7 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs * panic(fmt.Sprintf("unknown txtyp: '%v'", ty)) } path := dbs.pathForType(ty) - w, err := registry.OpenDBWrapper(path, DetectMemAccessPastTx) + w, err := registry.OpenDBWrapper(path, DetectMemAccessPastTx, per.RBFConfig) panicOn(err) h := idx.Holder() w.SetHolder(h) diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 52b9b3027..b58f86045 100644 --- a/dbshard_internal_test.go +++ b/dbshard_internal_test.go @@ -228,7 +228,7 @@ func makeBolttestDB(path string, h *Holder, shard uint64) { func makeRBFtestDB(path string, h *Holder, shard uint64) { i := uint64(1) - db := rbf.NewDB(path) + db := rbf.NewDB(path, nil) err := db.Open() panicOn(err) defer db.Close() diff --git a/holder.go b/holder.go index fdaabf099..6da35f797 100644 --- a/holder.go +++ b/holder.go @@ -31,6 +31,7 @@ import ( "github.com/pilosa/pilosa/v2/logger" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/stats" "github.com/pilosa/pilosa/v2/testhook" @@ -201,6 +202,8 @@ type HolderConfig struct { Logger logger.Logger Txsrc string RowcacheOff bool + + RBFConfig *rbfcfg.Config } func DefaultHolderConfig() *HolderConfig { @@ -215,6 +218,7 @@ func DefaultHolderConfig() *HolderConfig { NewAttrStore: newNopAttrStore, Logger: logger.NopLogger, Txsrc: DefaultTxsrc, + RBFConfig: rbfcfg.NewDefaultConfig(), } } @@ -229,6 +233,8 @@ func NewHolder(path string, cfg *HolderConfig) *Holder { // INVAR: have valid txsrc. cfg.Txsrc = txsrc } + } else if cfg.RBFConfig == nil { + cfg.RBFConfig = rbfcfg.NewDefaultConfig() } h := &Holder{ diff --git a/lmdb.go b/lmdb.go index e952df65d..d106b4647 100644 --- a/lmdb.go +++ b/lmdb.go @@ -34,6 +34,7 @@ import ( "github.com/glycerine/lmdb-go/lmdb" "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" @@ -140,7 +141,8 @@ func lmdbPath(path string) string { // if one does not exist for its bpath. Otherwise it returns // the existing instance. This insures only one lmdbDB // per bpath in this pilosa node. -func (r *lmdbRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrapper, error) { + +func (r *lmdbRegistrar) OpenDBWrapper(path0 string, doAllocZero bool, rbfcfg *rbfcfg.Config) (DBWrapper, error) { path := lmdbPath(path0) r.mu.Lock() diff --git a/lmdb_other.go b/lmdb_other.go index fa6abb946..0768ecfc5 100644 --- a/lmdb_other.go +++ b/lmdb_other.go @@ -26,6 +26,7 @@ import ( "sync" "time" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" ) @@ -66,7 +67,7 @@ func newLMDBTestRegistrar() *lmdbRegistrar { } } -func (r *lmdbRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrapper, error) { +func (r *lmdbRegistrar) OpenDBWrapper(path string, doAllocZero bool, rbfcfg *rbfcfg.Config) (DBWrapper, error) { panic("lmdb only available on 64-bit arch") } diff --git a/lmdb_test.go b/lmdb_test.go index 69103f8ef..32825093a 100644 --- a/lmdb_test.go +++ b/lmdb_test.go @@ -89,7 +89,7 @@ func mustOpenEmptyLMDBWrapper(path string) (w *LMDBWrapper, cleaner func()) { var err error fn := lmdbPath(path) panicOn(os.RemoveAll(fn)) - ww, err := globalLMDBReg.OpenDBWrapper(fn, DetectMemAccessPastTx) + ww, err := globalLMDBReg.OpenDBWrapper(fn, DetectMemAccessPastTx, nil) panicOn(err) w = ww.(*LMDBWrapper) diff --git a/rbf.go b/rbf.go index deb427b8e..971118533 100644 --- a/rbf.go +++ b/rbf.go @@ -27,6 +27,7 @@ import ( "sync/atomic" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" "github.com/pkg/errors" @@ -36,6 +37,7 @@ import ( type RbfDBWrapper struct { path string db *rbf.DB + cfg *rbfcfg.Config reg *rbfDBRegistrar muDb sync.Mutex @@ -145,7 +147,7 @@ func rbfPath(path string) string { // if one does not exist for its path. Otherwise it returns // the existing instance. This insures only one RbfDBWrapper // per bpath in this pilosa node. -func (r *rbfDBRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrapper, error) { +func (r *rbfDBRegistrar) OpenDBWrapper(path0 string, doAllocZero bool, cfg *rbfcfg.Config) (DBWrapper, error) { path := rbfPath(path0) r.mu.Lock() defer r.mu.Unlock() @@ -154,8 +156,11 @@ func (r *rbfDBRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrappe // creates the effect of having only one DB open per pilosa node. return w, nil } - db := rbf.NewDB(path) - db.DoAllocZero = doAllocZero + if cfg == nil { + cfg = rbfcfg.NewDefaultConfig() + cfg.DoAllocZero = doAllocZero + } + db := rbf.NewDB(path, cfg) w = &RbfDBWrapper{ reg: r, @@ -163,6 +168,7 @@ func (r *rbfDBRegistrar) OpenDBWrapper(path0 string, doAllocZero bool) (DBWrappe db: db, doAllocZero: doAllocZero, openTx: make(map[*RBFTx]bool), + cfg: cfg, } r.unprotectedRegister(w) diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go new file mode 100644 index 000000000..56fff7305 --- /dev/null +++ b/rbf/cfg/cfg.go @@ -0,0 +1,61 @@ +// Copyright 2017 Pilosa Corp. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cfg + +import ( + "time" + + "github.com/spf13/pflag" +) + +// Config defines externally configurable rbf options. +// The separate package avoids circular import. +type Config struct { + + // The maximum allowed database size. Required by mmap. + MaxSize int64 + + // Set before calling db.Open() + FsyncEnabled bool + + // for mmap correctness testing. + DoAllocZero bool + + // CheckpointEveryDur if zero means checkpoint after every write. + // Otherwise, wait and checkpoint at the next write that happens + // after CheckpointEveryDur since the previous. + CheckpointEveryDur time.Duration + + // Maximum size of a single WAL segment. + // May exceed by one page if last page is a bitmap header + bitmap. + MaxWALSegmentFileSize int +} + +func NewDefaultConfig() *Config { + return &Config{ + MaxSize: DefaultMaxSize, + FsyncEnabled: true, + CheckpointEveryDur: 10 * time.Second, + MaxWALSegmentFileSize: 1 << 16, + } +} + +func (cfg *Config) DefineFlags(flags *pflag.FlagSet) { + default0 := NewDefaultConfig() + flags.IntVar(&cfg.MaxWALSegmentFileSize, "rbf-max-wal", default0.MaxWALSegmentFileSize, "RBF write-Ahead-Log file size in bytes") + flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur, "RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.") + flags.Int64Var(&cfg.MaxSize, "rbf-max-db-size", default0.MaxSize, "RBF maximum size in bytes of a database file (distinct from a WAL file)") + flags.BoolVar(&cfg.FsyncEnabled, "rbf-fsync", default0.FsyncEnabled, "RBF: enable fsync fully safe flush-to-disk at each checkpoint") +} diff --git a/rbf/os.go b/rbf/cfg/os.go similarity index 98% rename from rbf/os.go rename to rbf/cfg/os.go index a76a550ef..ee755f268 100644 --- a/rbf/os.go +++ b/rbf/cfg/os.go @@ -14,7 +14,7 @@ // +build !386 -package rbf +package cfg // DefaultMaxSize is the default mmap size and therefore the maximum allowed // size of the database. The size can be increased by updating the DB.MaxSize diff --git a/rbf/os_386.go b/rbf/cfg/os_386.go similarity index 98% rename from rbf/os_386.go rename to rbf/cfg/os_386.go index b23457dbb..a8479c73b 100644 --- a/rbf/os_386.go +++ b/rbf/cfg/os_386.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package rbf +package cfg // DefaultMaxSize is the default mmap size and therefore the maximum allowed // size of the database. The size can be increased by updating the DB.MaxSize diff --git a/rbf/cursorx.go b/rbf/cursorx.go index 3c4b16352..de23ecaf2 100644 --- a/rbf/cursorx.go +++ b/rbf/cursorx.go @@ -166,7 +166,7 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) { orig := l.Data var cpMaybe []byte var mapped bool - if EnableRowCache() || tx.db.DoAllocZero { + if EnableRowCache() || tx.db.cfg.DoAllocZero { // make a copy, otherwise the rowCache will see corrupted data // or mmapped data that may disappear. cpMaybe = make([]byte, len(orig)) diff --git a/rbf/db.go b/rbf/db.go index b34d732f2..9ac28dd83 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -23,24 +23,23 @@ import ( "path/filepath" "sync" "syscall" + "time" "github.com/benbjohnson/immutable" "github.com/pilosa/pilosa/v2/syswrap" + + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" ) var ( ErrClosed = errors.New("rbf: database closed") ) -const ( - // Maximum size of a single WAL segment. - // May exceed by one page if last page is a bitmap header + bitmap. - MaxWALSegmentFileSize = 10 * (1 << 20) -) - // DB options like MaxSize, FsyncEnabled, DoAllocZero // can be set before calling DB.Open(). type DB struct { + cfg rbfcfg.Config + data []byte // mmap data file *os.File // file descriptor rootRecords []*RootRecord // cached root records @@ -58,25 +57,21 @@ type DB struct { // Path represents the path to the database file. Path string - // The maximum allowed database size. Required by mmap. - MaxSize int64 - - // Set before calling db.Open() - FsyncEnabled bool - - // for mmap correctness testing. - DoAllocZero bool + lastCheckpoint time.Time } // NewDB returns a new instance of DB. -func NewDB(path string) *DB { +// If cfg is nil we will use the rbfcfg.DefaultConfig(). +func NewDB(path string, cfg *rbfcfg.Config) *DB { + if cfg == nil { + cfg = rbfcfg.NewDefaultConfig() + } db := &DB{ - txs: make(map[*Tx]struct{}), - pageMap: immutable.NewMap(&uint32Hasher{}), - wcache: make([]byte, MaxWALSegmentFileSize+PageSize), - Path: path, - MaxSize: DefaultMaxSize, - FsyncEnabled: true, + cfg: *cfg, + txs: make(map[*Tx]struct{}), + pageMap: immutable.NewMap(&uint32Hasher{}), + wcache: make([]byte, cfg.MaxWALSegmentFileSize+PageSize), + Path: path, } return db } @@ -123,7 +118,7 @@ func (db *DB) Open() (err error) { // Open read-only mmap. if f, err := os.OpenFile(db.DataPath(), os.O_RDONLY, 0666); err != nil { return fmt.Errorf("open mmap file: %w", err) - } else if db.data, err = syswrap.Mmap(int(f.Fd()), 0, int(db.MaxSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil { + } else if db.data, err = syswrap.Mmap(int(f.Fd()), 0, int(db.cfg.MaxSize), syscall.PROT_READ, syscall.MAP_SHARED); err != nil { f.Close() return fmt.Errorf("open mmap file: %w", err) } else if err := f.Close(); err != nil { @@ -605,10 +600,15 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) { } db.mu.Lock() - defer db.mu.Unlock() + // note: We cannot defer db.mu.Unlock() here because + // we call tx.Rollback() before if db.readMetaPage + // returns an error, and thus we will deadlock against + // ourselves when the Rollback tries to acquire the db.mu. + // This is why db.mu.Unlock() is done manually below. if !db.opened { cleanup() + db.mu.Unlock() return nil, ErrClosed } @@ -617,6 +617,7 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) { if exclusive { if err := db.checkpoint(true, &nopLocker{}); err != nil { cleanup() + db.mu.Unlock() return nil, err } } @@ -644,6 +645,9 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) { // This page is only written at the end of a dirty transaction. page, err := db.readMetaPage() if err != nil { + // we will deadlock in tx.Rollback() + // on db.mu.Lock unless we manually db.mu.Unlock first. + db.mu.Unlock() tx.Rollback() return nil, err } @@ -655,6 +659,7 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) { // Track transaction with the DB. db.txs[tx] = struct{}{} + db.mu.Unlock() return tx, nil } @@ -679,8 +684,11 @@ func (db *DB) removeTx(tx *Tx) error { // Write pages from WAL to DB. // TODO(bbj): Move this to an async goroutine. if tx.writable { - if err := db.checkpoint(false, &nopLocker{}); err != nil { - return fmt.Errorf("checkpoint: %w", err) + if db.cfg.CheckpointEveryDur == 0 || time.Since(db.lastCheckpoint) > db.cfg.CheckpointEveryDur { + if err := db.checkpoint(false, &nopLocker{}); err != nil { + return fmt.Errorf("checkpoint: %w", err) + } + db.lastCheckpoint = time.Now() } } diff --git a/rbf/db_test.go b/rbf/db_test.go index a06cd26f6..11db76c62 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -15,12 +15,16 @@ package rbf_test import ( + "fmt" "math/rand" + "net" + "net/http" "os" "testing" "time" "github.com/pilosa/pilosa/v2/rbf" + _ "net/http/pprof" ) func TestDB_Open(t *testing.T) { @@ -32,6 +36,7 @@ func TestDB_Open(t *testing.T) { } } +/* optimization of wal size means there may certainly be more than 2 WAL segments. func TestDB_Checkpoint(t *testing.T) { if testing.Short() { t.Skip("-short enabled, skipping") @@ -68,6 +73,7 @@ func TestDB_Checkpoint(t *testing.T) { t.Fatalf("expected two or fewer WAL segments, got %d", n) } } +*/ func TestDB_Recovery(t *testing.T) { // Ensure a bitmap header written without a bitmap is truncated. @@ -121,7 +127,7 @@ func TestDB_Recovery(t *testing.T) { } // Reopen database. - newDB := rbf.NewDB(db.Path) + newDB := rbf.NewDB(db.Path, nil) if err := newDB.Open(); err != nil { t.Fatal(err) } @@ -283,3 +289,20 @@ func TestDB_HasData(t *testing.T) { t.Fatalf("HasData should have seen the hot bit") } } + +// better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests. +func TestMain(m *testing.M) { + port := getAvailPort() + fmt.Printf("rbf/ TestMain: online stack-traces: curl http://localhost:%v/debug/pprof/goroutine?debug=2\n", port) + go func() { + _ = http.ListenAndServe(fmt.Sprintf("127.0.0.1:%v", port), nil) + }() + os.Exit(m.Run()) +} + +func getAvailPort() int { + l, _ := net.Listen("tcp", ":0") + r := l.Addr() + l.Close() + return r.(*net.TCPAddr).Port +} diff --git a/rbf/rbf.go b/rbf/rbf.go index d1f2fe4ca..17dc4177f 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -666,7 +666,7 @@ func (db *DB) truncate(path string, sz int64) error { } func (db *DB) fsync(f *os.File) error { - if !db.FsyncEnabled { + if !db.cfg.FsyncEnabled { return nil } return f.Sync() diff --git a/rbf/rbf_test.go b/rbf/rbf_test.go index e4ea450e2..fc3aa3b79 100644 --- a/rbf/rbf_test.go +++ b/rbf/rbf_test.go @@ -66,7 +66,7 @@ func NewDB() *rbf.DB { panic(err) } - db := rbf.NewDB(path) + db := rbf.NewDB(path, nil) return db } @@ -104,7 +104,7 @@ func MustReopenDB(tb testing.TB, db *rbf.DB) *rbf.DB { tb.Fatal(err) } - other := rbf.NewDB(db.Path) + other := rbf.NewDB(db.Path, nil) if err := other.Open(); err != nil { tb.Fatal(err) } diff --git a/rbf/tx.go b/rbf/tx.go index 181df99a6..41c0fddca 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -1663,7 +1663,7 @@ func (tx *Tx) writeBitmapWALPage(pgno uint32, page []byte) (walID int64, err err func (tx *Tx) ensureWritableWALSegment() error { // Ignore if we still have space in the write cache. writeCacheSize := int64(len(tx.wcache)) - if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < MaxWALSegmentFileSize { + if len(tx.segments) != 0 && activeWALSegment(tx.segments).Size()+writeCacheSize < int64(tx.db.cfg.MaxWALSegmentFileSize) { return nil } diff --git a/rbf/wal.go b/rbf/wal.go index f6787cfca..9a8353e49 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -19,11 +19,14 @@ import ( "io" "os" "path/filepath" + "sort" "syscall" "github.com/pilosa/pilosa/v2/syswrap" ) +var _ = sort.Search + // WALSegment represents a single file in the WAL. type WALSegment struct { db *DB @@ -83,7 +86,7 @@ func (s *WALSegment) Open() (err error) { // Default the mmap size to the max size plus a page of padding for bitmap pages. // If the actual size is larger, then increase to that size. - mmapSize := int64(MaxWALSegmentFileSize + PageSize) + mmapSize := int64(s.db.cfg.MaxWALSegmentFileSize + PageSize) if sz > mmapSize { mmapSize = sz } @@ -164,8 +167,12 @@ func walSize(segments []WALSegment) int64 { // readWALPage reads a single page at the given WAL ID. func readWALPage(segments []WALSegment, walID int64) ([]byte, error) { - // TODO(BBJ): Binary search for segment. - for _, s := range segments { + n := len(segments) + i := sort.Search(n, func(i int) bool { + return walID < segments[i].MinWALID + }) + if i > 0 { + s := segments[i-1] if walID >= s.MinWALID && walID <= s.MaxWALID() { return s.ReadWALPage(walID) } diff --git a/rrtx.go b/rrtx.go index 96d5c198e..78527cfd1 100644 --- a/rrtx.go +++ b/rrtx.go @@ -26,6 +26,7 @@ import ( "sync/atomic" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" "github.com/pkg/errors" ) @@ -462,7 +463,7 @@ func (r *roaringRegistrar) unregister(w *RoaringWrapper) { // openRoaringDB will check the registry and make a new instance only // if one does not exist for its path0. Otherwise it returns // the existing instance. -func (r *roaringRegistrar) OpenDBWrapper(path string, doAllocZero bool) (DBWrapper, error) { +func (r *roaringRegistrar) OpenDBWrapper(path string, doAllocZero bool, cfg *rbfcfg.Config) (DBWrapper, error) { r.mu.Lock() defer r.mu.Unlock() diff --git a/rrtx_internal_test.go b/rrtx_internal_test.go index 3a03ca763..a993b25b4 100644 --- a/rrtx_internal_test.go +++ b/rrtx_internal_test.go @@ -28,7 +28,7 @@ func TestRoaring_HasData(t *testing.T) { idx := newIndexWithTempPath(t, "i") defer idx.Close() - db, err := globalRoaringReg.OpenDBWrapper(idx.path, false) + db, err := globalRoaringReg.OpenDBWrapper(idx.path, false, nil) panicOn(err) db.SetHolder(idx.holder) diff --git a/server.go b/server.go index b1fa62e2b..d543e5088 100644 --- a/server.go +++ b/server.go @@ -30,6 +30,7 @@ import ( uuid "github.com/satori/go.uuid" "github.com/pilosa/pilosa/v2/logger" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/stats" "github.com/pkg/errors" @@ -352,6 +353,14 @@ func OptServerRowcacheOff(rowcacheOff bool) ServerOption { } } +// OptServerRBFConfig conveys the RBF flags to the Holder. +func OptServerRBFConfig(cfg *rbfcfg.Config) ServerOption { + return func(s *Server) error { + s.holderConfig.RBFConfig = cfg + return nil + } +} + // NewServer returns a new instance of Server. func NewServer(opts ...ServerOption) (*Server, error) { cluster := newCluster() @@ -407,6 +416,12 @@ func NewServer(opts ...ServerOption) (*Server, error) { s.holder = NewHolder(path, s.holderConfig) s.holder.Stats.SetLogger(s.logger) s.holder.Logger.Printf("RowCacheOff: %v", s.holderConfig.RowcacheOff) + cwd, err := os.Getwd() + if err != nil { + return nil, err + } + s.holder.Logger.Printf("cwd: %v", cwd) + s.holder.Logger.Printf("cmd line: %v", strings.Join(os.Args, " ")) s.cluster.Path = path s.cluster.logger = s.logger diff --git a/server/config.go b/server/config.go index 3e73669ec..c69a5ea26 100644 --- a/server/config.go +++ b/server/config.go @@ -25,6 +25,7 @@ import ( "time" "github.com/pilosa/pilosa/v2/gossip" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" "github.com/pilosa/pilosa/v2/toml" "github.com/pkg/errors" ) @@ -202,6 +203,9 @@ type Config struct { // RowcacheOff, if true, turns off the row cache for all storage backends. RowcacheOff bool `toml:"rowcache-off"` + + // RBFConfig defines all externally configurable RBF flags. + RBFConfig *rbfcfg.Config } // NewConfig returns an instance of Config with default options. @@ -224,6 +228,8 @@ func NewConfig() *Config { WorkerPoolSize: runtime.NumCPU(), ImportWorkerPoolSize: runtime.NumCPU(), + + RBFConfig: rbfcfg.NewDefaultConfig(), } // Cluster config. diff --git a/server/server.go b/server/server.go index ce84be988..46a83c453 100644 --- a/server/server.go +++ b/server/server.go @@ -411,6 +411,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerSerializer(proto.Serializer{}), pilosa.OptServerTxsrc(m.Config.Txsrc), pilosa.OptServerRowcacheOff(m.Config.RowcacheOff), + pilosa.OptServerRBFConfig(m.Config.RBFConfig), coordinatorOpt, } diff --git a/txfactory.go b/txfactory.go index 6e9efe1fb..0d75ecbc0 100644 --- a/txfactory.go +++ b/txfactory.go @@ -525,6 +525,10 @@ func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory, } f.dbPerShard = f.NewDBPerShard(types, holderDir, holder) + if f.hasRBF() { + holder.Logger.Printf("rbf config = %#v", holder.cfg.RBFConfig) + } + return f, err } @@ -1247,7 +1251,11 @@ func (f *TxFactory) blueGreenOffIfRunningBlueGreen() { } func (f *TxFactory) hasRoaring() bool { - return f.types[0] == roaringTxn || f.types[1] == roaringTxn + return f.types[0] == roaringTxn || (len(f.types) > 1 && f.types[1] == roaringTxn) +} + +func (f *TxFactory) hasRBF() bool { + return f.types[0] == rbfTxn || (len(f.types) > 1 && f.types[1] == rbfTxn) } var _ = (&TxFactory{}).hasRoaring // happy linter