From 8d2ad048fa929e9535d200980e6921c86ac632c3 Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Thu, 22 Oct 2020 14:17:52 +0000 Subject: [PATCH] pilosa server --norowcache disables the row cache - this can lessen memory pressure - certain backends may not need it - enables performance benchmarking and tuning --- bolt.go | 2 +- ctl/server.go | 6 +++++- holder.go | 9 ++++++++- lmdb.go | 2 +- rbf.go | 2 +- rbf/cursorx.go | 21 ++++++++++++++++++--- rrtx.go | 2 +- server.go | 10 ++++++++++ server/config.go | 7 +++++-- server/server.go | 1 + 10 files changed, 51 insertions(+), 11 deletions(-) diff --git a/bolt.go b/bolt.go index 9db81fdca..39106fecf 100644 --- a/bolt.go +++ b/bolt.go @@ -441,7 +441,7 @@ func (tx *BoltTx) Type() string { } func (tx *BoltTx) UseRowCache() bool { - return rbf.EnableRowCache + return rbf.EnableRowCache() } // Pointer gives us a memory address for the underlying transaction for debugging. diff --git a/ctl/server.go b/ctl/server.go index 221f3918f..204a368bf 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -17,6 +17,7 @@ package ctl import ( "time" + "github.com/pilosa/pilosa/v2" "github.com/pilosa/pilosa/v2/server" "github.com/spf13/cobra" ) @@ -88,7 +89,10 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.IntVar(&srv.Config.Profile.MutexFraction, "profile.mutex-fraction", srv.Config.Profile.MutexFraction, "Sampling fraction for mutex contention profiling. Sample 1/ of events.") // Transactional storage engine - flags.StringVarP(&srv.Config.Txsrc, "tx", "", "", "transaction/storage to use: one of roaring, rbf, badger, rbf_roaring, roaring_rbf, badger_roaring, roaring_badger, badger_rbf, or rbf_badger (default roaring)") + flags.StringVarP(&srv.Config.Txsrc, "tx", "", pilosa.DefaultTxsrc, "transaction/storage to use: one of roaring, rbf, bolt, lmdb, or a blue-green setup: rbf_roaring, roaring_rbf, bolt_roaring, roaring_bolt, bolt_rbf, etc.") + + // RowcacheOff + flags.BoolVarP((&srv.Config.RowcacheOff), "rowcache-off", "", srv.Config.RowcacheOff, "turn off the rowcache for all backends (reduces memory use)") // 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)") diff --git a/holder.go b/holder.go index bece8baf2..fdaabf099 100644 --- a/holder.go +++ b/holder.go @@ -30,6 +30,7 @@ import ( "time" "github.com/pilosa/pilosa/v2/logger" + "github.com/pilosa/pilosa/v2/rbf" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/stats" "github.com/pilosa/pilosa/v2/testhook" @@ -137,6 +138,9 @@ type HolderOpts struct { // Txsrc controls the tx/storage engine we instatiate. Set by // server.go OptServerTxsrc Txsrc string + + // RowcacheOff, if true, turns off the row cache for all storage backends. + RowcacheOff bool } func (h *Holder) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) { @@ -196,6 +200,7 @@ type HolderConfig struct { NewAttrStore func(string) AttrStore Logger logger.Logger Txsrc string + RowcacheOff bool } func DefaultHolderConfig() *HolderConfig { @@ -243,7 +248,7 @@ func NewHolder(path string, cfg *HolderConfig) *Holder { OpenTransactionStore: cfg.OpenTransactionStore, translationSyncer: cfg.TranslationSyncer, Logger: cfg.Logger, - Opts: HolderOpts{Txsrc: cfg.Txsrc}, + Opts: HolderOpts{Txsrc: cfg.Txsrc, RowcacheOff: cfg.RowcacheOff}, SnapshotQueue: defaultSnapshotQueue, @@ -254,6 +259,8 @@ func NewHolder(path string, cfg *HolderConfig) *Holder { indexes: make(map[string]*Index), } + rbf.SetRowcacheOn(!cfg.RowcacheOff) + txf, err := NewTxFactory(cfg.Txsrc, path, h) panicOn(err) h.txf = txf diff --git a/lmdb.go b/lmdb.go index dbe539897..e952df65d 100644 --- a/lmdb.go +++ b/lmdb.go @@ -551,7 +551,7 @@ func (tx *LMDBTx) Type() string { } func (tx *LMDBTx) UseRowCache() bool { - return rbf.EnableRowCache + return rbf.EnableRowCache() } // Pointer gives us a memory address for the underlying transaction for debugging. diff --git a/rbf.go b/rbf.go index 0e5b1ec2a..deb427b8e 100644 --- a/rbf.go +++ b/rbf.go @@ -449,7 +449,7 @@ func (tx *RBFTx) UseRowCache() bool { // the rowCache without first making a copy. // So we only use the rowCache if the copy is // enabled. - return rbf.EnableRowCache + return rbf.EnableRowCache() } // rbfName returns a NULL-separated key used for identifying bitmap maps in RBF. diff --git a/rbf/cursorx.go b/rbf/cursorx.go index e196f5c7f..3c4b16352 100644 --- a/rbf/cursorx.go +++ b/rbf/cursorx.go @@ -19,6 +19,7 @@ import ( "io" "math" "os" + "sync/atomic" "github.com/pilosa/pilosa/v2/roaring" "github.com/pkg/errors" @@ -26,7 +27,21 @@ import ( // if enableRowCache, then we must not return mmap-ed memory // directly, but only a copy. -const EnableRowCache = true +var enableRowcache int64 = 1 + +// SetEnableRowCache should only be called in NewHolder before +// all other reads. +func SetRowcacheOn(on bool) { + if on { + atomic.StoreInt64(&enableRowcache, 1) + } else { + atomic.StoreInt64(&enableRowcache, 0) + } +} + +func EnableRowCache() bool { + return atomic.LoadInt64(&enableRowcache) == 1 +} //probably should just implement the container interface // but for now i'll do it @@ -151,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.DoAllocZero { // make a copy, otherwise the rowCache will see corrupted data // or mmapped data that may disappear. cpMaybe = make([]byte, len(orig)) @@ -168,7 +183,7 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) { case ContainerTypeBitmapPtr: _, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe)) cloneMaybe := bm - if EnableRowCache { + if EnableRowCache() { cloneMaybe = make([]uint64, len(bm)) copy(cloneMaybe, bm) } diff --git a/rrtx.go b/rrtx.go index 8e254496e..96d5c198e 100644 --- a/rrtx.go +++ b/rrtx.go @@ -63,7 +63,7 @@ func (tx *RoaringTx) Dump(short bool, shard uint64) { } func (tx *RoaringTx) UseRowCache() bool { - return rbf.EnableRowCache + return rbf.EnableRowCache() } func (tx *RoaringTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) { diff --git a/server.go b/server.go index be0a41aa3..70123465a 100644 --- a/server.go +++ b/server.go @@ -343,6 +343,15 @@ func OptServerTxsrc(txsrc string) ServerOption { } } +// OptServerRowcacheOff is a functional option on Server +// used to turn off the row cache. +func OptServerRowcacheOff(rowcacheOff bool) ServerOption { + return func(s *Server) error { + s.holderConfig.RowcacheOff = rowcacheOff + return nil + } +} + // NewServer returns a new instance of Server. func NewServer(opts ...ServerOption) (*Server, error) { cluster := newCluster() @@ -397,6 +406,7 @@ 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) s.cluster.Path = path s.cluster.logger = s.logger diff --git a/server/config.go b/server/config.go index 3e8575c81..3e73669ec 100644 --- a/server/config.go +++ b/server/config.go @@ -190,8 +190,8 @@ type Config struct { // Txsrc determines which Tx implementation the holder/Index will use; one // of the available transactional-storage engines. Choices are listed // in the string constants below. Should be one of - // "roaring","badger", "rbf", "badger_roaring", "roaring_badger", "rbf_roaring", - // "roaring_rbf", "badger_rbf", "rbf_badger", or any later addition. The + // "roaring","bolt", "rbf", "bolt_roaring", "roaring_bolt", "rbf_roaring", + // "roaring_rbf", "bolt_rbf", "rbf_bolt", or any later addition. The // engines with _ underscore indicate use of a blueGreenTx with a comparison // of values back from each Tx method, and a panic if they differ. This // is an effective test for consistency. If "rbf_roaring" is specified, then @@ -199,6 +199,9 @@ type Config struct { // If "roaring_rbf" is chosen, then the RBF values are the ones actually // returned from the blueGreenTx. Txsrc string `toml:"txsrc"` + + // RowcacheOff, if true, turns off the row cache for all storage backends. + RowcacheOff bool `toml:"rowcache-off"` } // NewConfig returns an instance of Config with default options. diff --git a/server/server.go b/server/server.go index f172f5647..ce84be988 100644 --- a/server/server.go +++ b/server/server.go @@ -410,6 +410,7 @@ func (m *Command) SetupServer() error { pilosa.OptServerClusterName(m.Config.Cluster.Name), pilosa.OptServerSerializer(proto.Serializer{}), pilosa.OptServerTxsrc(m.Config.Txsrc), + pilosa.OptServerRowcacheOff(m.Config.RowcacheOff), coordinatorOpt, }