Merge pull request #1034 from jaten-molecula/rbf_checkpoint_rb

performance tuning: rbfcfg package, binary search for wal segment

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()

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)
 

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)")
+
 }

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)

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()

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{

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()

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")
 }
 

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)
 

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)
 

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")
+}

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

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

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))

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()
 		}
 	}
 

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
+}

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()

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)
 	}

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
 	}
 

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)
 		}

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()

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)
 

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

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.

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,
 	}
 

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