diff --git a/.gitignore b/.gitignore index 6e553c794..9bf572296 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ release-pilosa-fsck.*.*.tar.gz /log.* /tourna.log.* pilosa +*.dot diff --git a/ctl/server.go b/ctl/server.go index 629aac3dd..1a0754486 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -15,6 +15,7 @@ package ctl import ( + "fmt" "time" "github.com/pilosa/pilosa/v2" @@ -89,7 +90,9 @@ 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", "", 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.") + // Note: the default for --tx must be kept "" empty string. Otherwise we + // cannot detect and honor the PILOSA_TXSRC env var over-ride. + flags.StringVarP(&srv.Config.Txsrc, "txsrc", "", "", fmt.Sprintf("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. The default is: %v. The env var PILOSA_TXSRC is over-ridden by --tx option on the command line.", pilosa.DefaultTxsrc)) // RowcacheOff flags.BoolVarP((&srv.Config.RowcacheOff), "rowcache-off", "", srv.Config.RowcacheOff, "turn off the rowcache for all backends (reduces memory use)") diff --git a/ctl/server_test.go b/ctl/server_test.go index b99a2ed25..81f49a5fd 100644 --- a/ctl/server_test.go +++ b/ctl/server_test.go @@ -35,3 +35,14 @@ func TestBuildServerFlags(t *testing.T) { t.Fatal("log-path flag is required") } } + +func TestServerDefaultTxsrcFlags(t *testing.T) { + cm := &cobra.Command{} + buf := bytes.Buffer{} + stdin, stdout, stderr := GetIO(buf) + Server := server.NewCommand(stdin, stdout, stderr) + BuildServerFlags(cm, Server) + if cm.Flags().Lookup("txsrc").DefValue != "" { + t.Fatal("cannot set the txsrc default in ctl/server.go, otherwise we won't know to let the environment override the lack of --txsrc on the command line. We want explicit command line --txsrc to override the env value.") + } +} diff --git a/executor.go b/executor.go index 505897fed..f17c65e55 100644 --- a/executor.go +++ b/executor.go @@ -1450,6 +1450,7 @@ func executeDistinctShardSet(ctx context.Context, qcx *Qcx, idx *Index, fieldNam if err != nil { return SignedRow{}, errors.Wrap(err, "getting fragment data") } + defer fragData.Close() // We can't grab the containers "for each row" from the set-type field, // because we don't know how many rows there are, and some of them // might be empty, so really, we're going to iterate through the diff --git a/go.mod b/go.mod index 0690aca19..ed42aac5f 100644 --- a/go.mod +++ b/go.mod @@ -14,7 +14,6 @@ require ( github.com/dustin/go-humanize v1.0.0 github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311 github.com/glycerine/lmdb-go v1.9.34 - github.com/glycerine/rbtree v0.0.0-20190406191118-ceb71889d809 github.com/go-ole/go-ole v1.2.4 // indirect github.com/gogo/protobuf v1.2.1 github.com/golang/protobuf v1.3.3 diff --git a/go.sum b/go.sum index 2cb15dd96..ea9c803a4 100644 --- a/go.sum +++ b/go.sum @@ -56,8 +56,6 @@ github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311 h1:AAXH0ZvYIHHqU06A github.com/glycerine/idem v0.0.0-20190127113923-7a8083893311/go.mod h1:B72P/ZM99sNiCmaQJflpmMAF5LsDzStpLdWzn0+Vr2Y= github.com/glycerine/lmdb-go v1.9.34 h1:0lymJjpdelYnIMcNzsKROfIaApt99zhaHtjDJTHjGkE= github.com/glycerine/lmdb-go v1.9.34/go.mod h1:DrPeeTGooMg6B7cjNSP14perptTJzzdBy5YoosthrRs= -github.com/glycerine/rbtree v0.0.0-20190406191118-ceb71889d809 h1:wBr8MeUUS+Xi4oweFspffWBlDw8s1rGmRBwM4fUjxrc= -github.com/glycerine/rbtree v0.0.0-20190406191118-ceb71889d809/go.mod h1:tf1G9WLJXoNEQ5TWYvCSkqsOepuCNCJebECwJ/B/64I= github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go index c91ad8e05..f90179349 100644 --- a/rbf/cfg/cfg.go +++ b/rbf/cfg/cfg.go @@ -44,6 +44,10 @@ type Config struct { // for mmap correctness testing. DoAllocZero bool + + // CursorCacheSize is the number of copies of Cursor{} to keep in our + // readyCursorCh arena to avoid GC pressure. + CursorCacheSize int64 } func NewDefaultConfig() *Config { @@ -53,6 +57,10 @@ func NewDefaultConfig() *Config { MinWALCheckpointSize: DefaultMinWALCheckpointSize, MaxWALCheckpointSize: DefaultMaxWALCheckpointSize, FsyncEnabled: true, + + // CI passed with 20. 50 was too big for CI, even on X-large instances. + // For now we default to 0, which means use sync.Pool. + CursorCacheSize: 0, } } @@ -65,4 +73,6 @@ func (cfg *Config) DefineFlags(flags *pflag.FlagSet) { // renamed from --rbf-fsync to just --fsync because now it applies to all Tx backends. flags.BoolVar(&cfg.FsyncEnabled, "fsync", default0.FsyncEnabled, "enable fsync fully safe flush-to-disk") + flags.Int64Var(&cfg.CursorCacheSize, "rbf-cursor-cache", default0.CursorCacheSize, "how big a Cursor arena to maintain. 0 means use sync.Pool with dynamic sizing. Note that <= 20 is needed to pass CI. Controls the memory footprint of rbf.") + } diff --git a/rbf/cursor.go b/rbf/cursor.go index 8c674ad00..8271c894c 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -1275,3 +1275,21 @@ func (c *Cursor) difference(key uint64, data *roaring.Container) (bool, error) { return false, nil } + +func (c *Cursor) Close() { + if c == nil { + panic("cannot Close nil Cursor") + } + tx := c.tx + c.tx = nil // allow tx to be garbage collected. + + if tx.db.cfg.CursorCacheSize == 0 { + globalCursorSyncPool.Put(c) + return + } + + select { + case tx.db.cursorArenaCh <- c: + case <-tx.db.cursorCleaner.ReqStop.Chan: + } +} diff --git a/rbf/cursor_test.go b/rbf/cursor_test.go index 2dcc20e64..1a72bcd73 100644 --- a/rbf/cursor_test.go +++ b/rbf/cursor_test.go @@ -1024,8 +1024,8 @@ func TestCursor_PlayContainer(t *testing.T) { } cur, _ := tx.Cursor("x") offset := uint64(0) - many(&cur, 0, rbf.ArrayMaxSize+offset) - many(&cur, 65536, rbf.ArrayMaxSize+offset) + many(cur, 0, rbf.ArrayMaxSize+offset) + many(cur, 65536, rbf.ArrayMaxSize+offset) /* many(cur, 2*65536, rbf.ArrayMaxSize+offset) many(cur, 3*65536, rbf.ArrayMaxSize) //+offset) diff --git a/rbf/cursorx.go b/rbf/cursorx.go index 2d032ba64..6b0864195 100644 --- a/rbf/cursorx.go +++ b/rbf/cursorx.go @@ -71,9 +71,15 @@ func (c *Cursor) Rows() ([]uint64, error) { } return rows, err } + func (tx *Tx) FieldViews() []string { - rr, _ := tx.RootRecords() - return rr.sliceOfNames() + records, _ := tx.RootRecords() + a := make([]string, 0, records.Len()) + for itr := records.Iterator(); !itr.Done(); { + name, _ := itr.Next() + a = append(a, name.(string)) + } + return a } func (c *Cursor) DumpKeys() { diff --git a/rbf/db.go b/rbf/db.go index 9f70b18d3..62278c307 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -24,26 +24,34 @@ import ( "syscall" "github.com/benbjohnson/immutable" - "github.com/pilosa/pilosa/v2/syswrap" - + "github.com/glycerine/idem" rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" + "github.com/pilosa/pilosa/v2/syswrap" ) var ( ErrClosed = errors.New("rbf: database closed") ) +// global in the sense that it is shared among all instances +// of rbf.DBs in this process. This is deliberate. +var globalCursorSyncPool = &sync.Pool{ + New: func() interface{} { + return &Cursor{} + }, +} + // DB options like MaxSize, FsyncEnabled, DoAllocZero // can be set before calling DB.Open(). type DB struct { cfg rbfcfg.Config - data []byte // database mmap - file *os.File // database file descriptor - rootRecords *rr // cached root records - pageMap *immutable.Map // pgno-to-WALID mapping - txs map[*Tx]struct{} // active transactions - opened bool // true if open + data []byte // database mmap + file *os.File // database file descriptor + rootRecords *immutable.SortedMap // cached root records + pageMap *immutable.Map // pgno-to-WALID mapping + txs map[*Tx]struct{} // active transactions + opened bool // true if open wal []byte // wal mmap walFile *os.File // wal file descriptor @@ -55,6 +63,9 @@ type DB struct { // Path represents the path to the database file. Path string + + cursorArenaCh chan *Cursor + cursorCleaner *idem.Halter } // NewDB returns a new instance of DB. @@ -68,8 +79,15 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB { txs: make(map[*Tx]struct{}), pageMap: immutable.NewMap(&uint32Hasher{}), Path: path, + + cursorArenaCh: make(chan *Cursor, cfg.CursorCacheSize), + cursorCleaner: idem.NewHalter(), + } + for i := int64(0); i < cfg.CursorCacheSize; i++ { + db.cursorArenaCh <- &Cursor{} } db.haltCond = sync.NewCond(&db.mu) + return db } @@ -254,6 +272,8 @@ func (db *DB) Close() (err error) { db.mu.Lock() defer db.mu.Unlock() + defer db.cursorCleaner.RequestStop() + db.opened = false // Close mmap handle. @@ -330,13 +350,15 @@ func (db *DB) HasData(requireOneHotBit bool) (hasAnyRecords bool, err error) { // Loop over each bitmap and attempt to move to the first cell. // If we can move to a cell then we have at least one record. - for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() { - record := it.Item().(RootRecord) + for itr := records.Iterator(); !itr.Done(); { + name, _ := itr.Next() // Fetch cursor for bitmap. - cur, err := tx.Cursor(record.Name) + cur, err := tx.Cursor(name.(string)) if err != nil { return false, err } + defer cur.Close() + if !requireOneHotBit { return true, nil } @@ -497,9 +519,6 @@ func (db *DB) removeTx(tx *Tx) error { tx.db.rwmu.Unlock() } - db.mu.Lock() - defer db.mu.Unlock() - delete(tx.db.txs, tx) // Disassociate from db. @@ -560,3 +579,23 @@ func (db *DB) readMetaPage() ([]byte, error) { } return db.readDBPage(0) } + +func (db *DB) getCursor(tx *Tx) (c *Cursor) { + if db.cfg.CursorCacheSize == 0 { + c = globalCursorSyncPool.Get().(*Cursor) + c.tx = tx + return + } + + n := len(db.cursorArenaCh) + if n < 10 { + vv("warning, db.cursorArenaCh is low! %v left", n) + } + select { + case c = <-db.cursorArenaCh: + c.tx = tx + return + case <-db.cursorCleaner.ReqStop.Chan: + return nil + } +} diff --git a/rbf/ingest_test.go b/rbf/ingest_test.go index 881bec444..89456b678 100644 --- a/rbf/ingest_test.go +++ b/rbf/ingest_test.go @@ -129,6 +129,7 @@ func TestIngest_lots_of_views(t *testing.T) { } else if err != nil { panic(err) } + defer c.Close() c.Dump("one.bitmap.dot.dump") } } diff --git a/rbf/rbf.go b/rbf/rbf.go index 277f5691a..91718eadc 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -27,7 +27,7 @@ import ( "time" "unsafe" - "github.com/glycerine/rbtree" + "github.com/benbjohnson/immutable" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/shardwidth" ) @@ -177,18 +177,19 @@ func readRootRecords(page []byte) (records []*RootRecord, err error) { // We can return io.ErrShortBuffer in err. If we still have records // to write that don't fit on page, remain will point to the next // record that hasn't yet been written. -func writeRootRecords(page []byte, recit, limit rbtree.Iterator) (remain rbtree.Iterator, err error) { +func writeRootRecords(page []byte, itr *immutable.SortedMapIterator) (err error) { data := page[rootRecordPageHeaderSize:] - for recit != limit { - rec := recit.Item().(RootRecord) - data, err = WriteRootRecord(data, &rec) + for !itr.Done() { + name, pgno := itr.Next() + + data, err = WriteRootRecord(data, &RootRecord{Name: name.(string), Pgno: pgno.(uint32)}) if err != nil { - return recit, err + itr.Seek(name) + return err } - recit = recit.Next() } - return recit, nil + return nil } // Branch & leaf page helpers diff --git a/rbf/tx.go b/rbf/tx.go index 6a36fc3cf..14505dd61 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -23,7 +23,6 @@ import ( "sync" "github.com/benbjohnson/immutable" - "github.com/glycerine/rbtree" "github.com/pilosa/pilosa/v2/hash" "github.com/pilosa/pilosa/v2/roaring" "github.com/pilosa/pilosa/v2/txkey" @@ -34,11 +33,11 @@ var _ = txkey.ToString // Tx represents a transaction. type Tx struct { mu sync.RWMutex - db *DB // parent db - meta [PageSize]byte // copy of current meta page - walID int64 // max WAL ID at start of tx - walPageN int // wal page count - rootRecords *rr // read-only cache of root records + db *DB // parent db + meta [PageSize]byte // copy of current meta page + walID int64 // max WAL ID at start of tx + walPageN int // wal page count + rootRecords *immutable.SortedMap // read-only cache of root records // pageMap holds WAL pages that have not yet been transferred // into the database pages. So it can be empty, if the whole previous @@ -59,70 +58,6 @@ type Tx struct { DeleteEmptyContainer bool } -type rr struct { - tree rbtree.Tree -} - -func newRR() *rr { - return &rr{ - tree: *rbtree.NewTree( - func(a, b rbtree.Item) int { - an := a.(RootRecord).Name - bn := b.(RootRecord).Name - if an == bn { - return 0 - } - if an < bn { - return -1 - } - return 1 - }), - } -} - -func (s *rr) size() int { - return s.tree.Len() -} - -func (s *rr) add(r RootRecord) { - s.tree.Insert(r) -} - -func (s *rr) addAll(recs []*RootRecord) { - for _, r := range recs { - s.add(*r) - } -} - -func (s *rr) remove(it rbtree.Iterator) { - s.tree.DeleteWithIterator(it) -} - -func iterToRootRecord(it rbtree.Iterator) RootRecord { - return it.Item().(RootRecord) -} - -func (s *rr) sliceOfNames() (res []string) { - res = make([]string, s.size()) - - i := 0 - for it := s.tree.Min(); it != s.tree.Limit(); it = it.Next() { - res[i] = it.Item().(RootRecord).Name - i++ - } - return -} - -func (s *rr) find(name string) (r RootRecord, iter rbtree.Iterator, exact bool) { - iter = s.tree.FindGE(RootRecord{Name: name}) - if iter.Limit() { - return - } - r = iter.Item().(RootRecord) - exact = (r.Name == name) - return -} - func (tx *Tx) DBPath() string { return tx.db.Path } @@ -169,13 +104,16 @@ func (tx *Tx) Commit() error { // avoid race detector firing on a write race here // vs the read of rootRecords at db.Begin() tx.db.mu.Lock() + defer tx.db.mu.Unlock() tx.db.rootRecords = tx.rootRecords tx.db.pageMap = tx.pageMap tx.db.walPageN = tx.walPageN - tx.db.mu.Unlock() + return tx.db.removeTx(tx) } // Disconnect transaction from DB. + tx.db.mu.Lock() + defer tx.db.mu.Unlock() return tx.db.removeTx(tx) } @@ -194,6 +132,8 @@ func (tx *Tx) Rollback() { } // Disconnect transaction from DB. + tx.db.mu.Lock() + defer tx.db.mu.Unlock() panicOn(tx.db.removeTx(tx)) } @@ -210,11 +150,11 @@ func (tx *Tx) root(name string) (uint32, error) { return 0, err } - _, it, exactHit := records.find(name) - if !exactHit { + pgno, ok := records.Get(name) + if !ok { return 0, ErrBitmapNotFound } - return iterToRootRecord(it).Pgno, nil + return pgno.(uint32), nil } // BitmapNames returns a list of all bitmap names. @@ -231,7 +171,13 @@ func (tx *Tx) BitmapNames() ([]string, error) { if err != nil { return nil, err } - return records.sliceOfNames(), nil + + a := make([]string, 0, records.Len()) + for itr := records.Iterator(); !itr.Done(); { + k, _ := itr.Next() + a = append(a, k.(string)) + } + return a, nil } // CreateBitmap creates a new empty bitmap with the given name. @@ -258,15 +204,12 @@ func (tx *Tx) createBitmap(name string) error { } // Find btree by name. Exit if already exists. - _, _, exact := records.find(name) - if exact { + if _, ok := records.Get(name); ok { return ErrBitmapExists } - //fmt.Println("CREATE BITMAP", name, index) // Allocate new root page. pgno, err := tx.allocatePgno() - //fmt.Println("CREATE BITMAP @ PGNO", pgno) if err != nil { return err } @@ -281,7 +224,7 @@ func (tx *Tx) createBitmap(name string) error { } // Insert into correct index. - records.add(RootRecord{Name: name, Pgno: pgno}) + records = records.Set(name, pgno) if err := tx.writeRootRecordPages(records); err != nil { return fmt.Errorf("write bitmaps: %w", err) } @@ -305,15 +248,6 @@ func (tx *Tx) createBitmapIfNotExists(name string) error { return nil } -/* -func dump(r []*RootRecord) { - for _, i := range r { - fmt.Println("RECORD", i.Name, i.Pgno) - } - -} -*/ - // DeleteBitmap removes a bitmap with the given name. // Returns an error if the bitmap does not exist. func (tx *Tx) DeleteBitmap(name string) error { @@ -335,21 +269,18 @@ func (tx *Tx) DeleteBitmap(name string) error { } // Find btree by name. Exit if it doesn't exist. - record, it, ok := records.find(name) - + pgno, ok := records.Get(name) if !ok { return fmt.Errorf("bitmap does not exist: %q", name) } - pgno := record.Pgno // Deallocate all pages in the tree. - if err := tx.deallocateTree(pgno); err != nil { + if err := tx.deallocateTree(pgno.(uint32)); err != nil { return err } - records.remove(it) - // Delete from record list & rewrite record pages. + records = records.Delete(name) if err := tx.writeRootRecordPages(records); err != nil { return fmt.Errorf("write bitmaps: %w", err) } @@ -373,24 +304,21 @@ func (tx *Tx) DeleteBitmapsWithPrefix(prefix string) error { if err != nil { return err } - for it := records.tree.Min(); it != records.tree.Limit(); { - record := it.Item().(RootRecord) + + for itr := records.Iterator(); !itr.Done(); { + name, pgno := itr.Next() // Skip bitmaps without matching prefix. - if !strings.HasPrefix(record.Name, prefix) { + if !strings.HasPrefix(name.(string), prefix) { continue } // Deallocate all pages in the tree. - if err := tx.deallocateTree(record.Pgno); err != nil { + if err := tx.deallocateTree(pgno.(uint32)); err != nil { return err } - // as long we've advanced it past delme, we can - // delete delme without affecting it. - delme := it - it = it.Next() - records.remove(delme) + records = records.Delete(name.(string)) } // Rewrite record pages. @@ -422,16 +350,14 @@ func (tx *Tx) RenameBitmap(oldname, newname string) error { } // Find btree by name. Exit if it doesn't exist. - rec, it, exactHit := records.find(oldname) - if !exactHit { + pgno, ok := records.Get(oldname) + if !ok { return fmt.Errorf("bitmap does not exist: %q", oldname) } // Update record name & rewrite record pages. - rec2 := rec - rec2.Name = newname - records.remove(it) - records.add(rec2) + records = records.Delete(oldname) + records = records.Set(newname, pgno) if err := tx.writeRootRecordPages(records); err != nil { return fmt.Errorf("write bitmaps: %w", err) } @@ -440,12 +366,12 @@ func (tx *Tx) RenameBitmap(oldname, newname string) error { } // RootRecords returns a list of root records. -func (tx *Tx) RootRecords() (records *rr, err error) { +func (tx *Tx) RootRecords() (records *immutable.SortedMap, err error) { if tx.rootRecords != nil { return tx.rootRecords, nil } - records = newRR() + records = immutable.NewSortedMap(nil) for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; { page, err := tx.readPage(pgno) if err != nil { @@ -457,7 +383,9 @@ func (tx *Tx) RootRecords() (records *rr, err error) { if err != nil { return nil, err } - records.addAll(a) + for _, rec := range a { + records = records.Set(rec.Name, rec.Pgno) + } // Read next overflow page number. pgno = WalkRootRecordPages(page) @@ -469,7 +397,7 @@ func (tx *Tx) RootRecords() (records *rr, err error) { } // writeRootRecordPages writes a list of root record pages. -func (tx *Tx) writeRootRecordPages(records *rr) (err error) { +func (tx *Tx) writeRootRecordPages(records *immutable.SortedMap) (err error) { // Release all existing root record pages. for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; { @@ -486,7 +414,7 @@ func (tx *Tx) writeRootRecordPages(records *rr) (err error) { } // Exit early if no records exist. - if records.size() == 0 { + if records.Len() == 0 { writeMetaRootRecordPageNo(tx.meta[:], 0) return nil } @@ -499,30 +427,19 @@ func (tx *Tx) writeRootRecordPages(records *rr) (err error) { writeMetaRootRecordPageNo(tx.meta[:], pgno) // Write new root record pages. - limit := records.tree.Limit() - it := records.tree.Min() - for it != limit { - + for itr := records.Iterator(); !itr.Done(); { // Initialize page & write as many records as will fit. page := make([]byte, PageSize) writePageNo(page, pgno) writeFlags(page, PageTypeRootRecord) - // writeRootRecords does it = it.Next() for us after - // each successful write to the page. - it, err = writeRootRecords(page, it, limit) - - switch err { - case nil: - // nothing to do, all the rest of the records fit on the page. - - case io.ErrShortBuffer: + if err := writeRootRecords(page, itr); err == io.ErrShortBuffer { // Allocate next pgno and write overflow if we have remaining records. if pgno, err = tx.allocatePgno(); err != nil { return err } writeRootRecordOverflowPgno(page, pgno) - default: + } else if err != nil { return err } @@ -559,6 +476,8 @@ func (tx *Tx) Add(name string, a ...uint64) (changeCount int, err error) { if err != nil { return 0, err } + defer c.Close() + for _, v := range a { if vchanged, err := c.Add(v); err != nil { return changeCount, err @@ -588,6 +507,8 @@ func (tx *Tx) Remove(name string, a ...uint64) (changeCount int, err error) { } else if err != nil { return 0, err } + defer c.Close() + for _, v := range a { if vchanged, err := c.Remove(v); err != nil { return changeCount, err @@ -615,29 +536,31 @@ func (tx *Tx) Contains(name string, v uint64) (bool, error) { } else if err != nil { return false, err } + defer c.Close() + return c.Contains(v) } // Cursor returns an instance of a cursor this bitmap. -func (tx *Tx) Cursor(name string) (Cursor, error) { +func (tx *Tx) Cursor(name string) (*Cursor, error) { tx.mu.RLock() defer tx.mu.RUnlock() return tx.cursor(name) } -func (tx *Tx) cursor(name string) (Cursor, error) { +func (tx *Tx) cursor(name string) (*Cursor, error) { if tx.db == nil { - return Cursor{}, ErrTxClosed + return nil, ErrTxClosed } else if name == "" { - return Cursor{}, ErrBitmapNameRequired + return nil, ErrBitmapNameRequired } root, err := tx.root(name) if err != nil { - return Cursor{}, err + return nil, err } - c := Cursor{tx: tx} + c := tx.db.getCursor(tx) c.stack.elems[0] = stackElem{pgno: root} return c, nil } @@ -659,6 +582,7 @@ func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) { } else if err != nil { return nil, err } + defer c.Close() other := roaring.NewSliceBitmap() if err := c.First(); err == io.EOF { @@ -698,9 +622,13 @@ func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) { return nil, nil } else if err != nil { return nil, err - } else if exact, err := c.Seek(key); err != nil || !exact { + } + defer c.Close() + + if exact, err := c.Seek(key); err != nil || !exact { return nil, err } + return toContainer(c.cell(), tx), nil } @@ -725,9 +653,13 @@ func (tx *Tx) putContainer(name string, key uint64, ct *roaring.Container) error c, err := tx.cursor(name) if err != nil { return err - } else if _, err := c.Seek(cell.Key); err != nil { + } + defer c.Close() + + if _, err := c.Seek(cell.Key); err != nil { return err } + return c.putLeafCell(cell) } @@ -754,9 +686,13 @@ func (tx *Tx) removeContainer(name string, key uint64) error { return nil } else if err != nil { return err - } else if exact, err := c.Seek(key); err != nil || !exact { + } + defer c.Close() + + if exact, err := c.Seek(key); err != nil || !exact { return err } + return c.deleteLeafCell(key) } @@ -869,10 +805,10 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) { return m, err } - for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() { - record := it.Item().(RootRecord) + for itr := records.Iterator(); !itr.Done(); { + _, pgno := itr.Next() - if err := tx.walkTree(record.Pgno, 0, func(pgno, parent, typ uint32) error { + if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error { m[pgno] = struct{}{} return nil }); err != nil { @@ -1059,6 +995,8 @@ func (tx *Tx) AddRoaring(name string, bm *roaring.Bitmap) (changed bool, err err if err != nil { return false, err } + defer c.Close() + return c.AddRoaring(bm) } @@ -1101,7 +1039,10 @@ func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error return nil } else if err != nil { return err - } else if _, err := c.Seek(highbits(start)); err != nil { + } + defer c.Close() + + if _, err := c.Seek(highbits(start)); err != nil { return err } @@ -1190,7 +1131,10 @@ func (tx *Tx) Count(name string) (uint64, error) { return 0, nil } else if err != nil { return 0, err - } else if err := c.First(); err != nil { + } + defer c.Close() + + if err := c.First(); err != nil { return 0, err } @@ -1216,7 +1160,10 @@ func (tx *Tx) Max(name string) (uint64, error) { return 0, nil } else if err != nil { return 0, err - } else if err := c.Last(); err == io.EOF { + } + defer c.Close() + + if err := c.Last(); err == io.EOF { return 0, nil } else if err != nil { return 0, err @@ -1235,7 +1182,10 @@ func (tx *Tx) Min(name string) (uint64, bool, error) { return 0, false, nil } else if err != nil { return 0, false, err - } else if err := c.First(); err == io.EOF { + } + defer c.Close() + + if err := c.First(); err == io.EOF { return 0, false, nil } else if err != nil { return 0, false, err @@ -1284,6 +1234,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) { } else if err != nil { return 0, err } + defer csr.Close() exact, err := csr.Seek(skey) _ = exact @@ -1354,6 +1305,7 @@ func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bit } else if err != nil { return nil, err } + defer c.Close() other := roaring.NewSliceBitmap() off := highbits(offset) @@ -1386,11 +1338,15 @@ func (tx *Tx) OffsetRange(name string, offset, start, endx uint64) (*roaring.Bit // containerIterator wraps Cursor to implement roaring.ContainerIterator. type containerIterator struct { - cursor Cursor + cursor *Cursor } -// Close is a no-op. It exists to implement the roaring.ContainerIterator interface. -func (itr *containerIterator) Close() {} +// Close must be called when the client is done +// with the containerIterator so that the internal +// Cursor can be recycled. +func (itr *containerIterator) Close() { + itr.cursor.Close() +} // Next moves the iterator to the next container. func (itr *containerIterator) Next() bool { @@ -1428,11 +1384,13 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) { panicOn(err) n := 0 - for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() { - rr := it.Item().(RootRecord) + for itr := records.Iterator(); !itr.Done(); { + name, _ := itr.Next() - c, err := tx.cursor(rr.Name) + c, err := tx.cursor(name.(string)) panicOn(err) + defer c.Close() + err = c.First() // First will rewind to beginning. if err == io.EOF { r += "" @@ -1450,7 +1408,7 @@ func (tx *Tx) DumpString(short bool, shard uint64) (r string) { ckey := cell.Key ct := toContainer(cell, tx) - s := stringOfCkeyCt(ckey, ct, rr.Name, short) + s := stringOfCkeyCt(ckey, ct, name.(string), short) r += s n++ } @@ -1557,6 +1515,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear if err != nil { return changed, rowSet, err } + defer cur.Close() for itrKey, synthC := itr.NextContainer(); synthC != nil; itrKey, synthC = itr.NextContainer() { if rowSize != 0 { @@ -1585,7 +1544,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear changed += nsynth rowSet[currRow] += nsynth - if err := tx.putContainerWithCursor(&cur, itrKey, synthC); err != nil { + if err := tx.putContainerWithCursor(cur, itrKey, synthC); err != nil { return changed, rowSet, err } continue @@ -1606,7 +1565,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear changes := int(existN - newC.N()) changed += changes rowSet[currRow] -= changes - err = tx.putContainerWithCursor(&cur, itrKey, newC) + err = tx.putContainerWithCursor(cur, itrKey, newC) if err != nil { return } @@ -1624,7 +1583,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear // can nsynth be zero? No, because of the continue/invariant above where nsynth > 0 changed += nsynth rowSet[currRow] += nsynth - err = tx.putContainerWithCursor(&cur, itrKey, synthC) + err = tx.putContainerWithCursor(cur, itrKey, synthC) if err != nil { return } @@ -1641,7 +1600,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear changed += changes rowSet[currRow] += changes - err = tx.putContainerWithCursor(&cur, itrKey, newC) + err = tx.putContainerWithCursor(cur, itrKey, newC) if err != nil { panicOn(err) return @@ -1656,7 +1615,6 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear // flush writes the dirty pages & meta page to the WAL. func (tx *Tx) flush() error { w := bufio.NewWriterSize(tx.db.walFile, 65536) - builder := immutable.NewMapBuilder(tx.pageMap) // Write non-bitmap pages to WAL. for _, pgno := range dirtyPageMapKeys(tx.dirtyPages) { @@ -1664,7 +1622,7 @@ func (tx *Tx) flush() error { if err != nil { return fmt.Errorf("write page to wal: %w", err) } - builder.Set(pgno, walID) + tx.pageMap = tx.pageMap.Set(pgno, walID) } // Write bitmap headers & pages to WAL. @@ -1682,7 +1640,7 @@ func (tx *Tx) flush() error { if err != nil { return fmt.Errorf("write bitmap page to wal: %w", err) } - builder.Set(pgno, walID) + tx.pageMap = tx.pageMap.Set(pgno, walID) } // Write meta page to WAL. @@ -1690,7 +1648,7 @@ func (tx *Tx) flush() error { if err != nil { return fmt.Errorf("write meta page to wal: %w", err) } - builder.Set(uint32(0), walID) + tx.pageMap = tx.pageMap.Set(uint32(0), walID) // Flush & sync WAL. if err := w.Flush(); err != nil { @@ -1699,9 +1657,6 @@ func (tx *Tx) flush() error { return fmt.Errorf("sync wal: %w", err) } - // Save page map for new WAL pages. - tx.pageMap = builder.Map() - return nil } @@ -1828,10 +1783,10 @@ func (tx *Tx) PageInfos() ([]PageInfo, error) { return nil, err } - for it := records.tree.Min(); it != records.tree.Limit(); it = it.Next() { - record := it.Item().(RootRecord) + for itr := records.Iterator(); !itr.Done(); { + name, pgno := itr.Next() - if err := tx.walkPageInfo(infos, record.Pgno, record.Name); err != nil { + if err := tx.walkPageInfo(infos, pgno.(uint32), name.(string)); err != nil { return nil, err } } diff --git a/rbf/tx_test.go b/rbf/tx_test.go index 27ac4d4f5..77f69b40d 100644 --- a/rbf/tx_test.go +++ b/rbf/tx_test.go @@ -361,11 +361,11 @@ func TestTx_CursorCrashArray(t *testing.T) { } //setArray(t, 0, 2379, &c) //setArray(t, 1, 2337, &c) - setArray(t, 32, 1216, &c) - setArray(t, 33, 1195, &c) - setArray(t, 48, 1186, &c) - setArray(t, 49, 1223, &c) - setArray(t, 50, 1223, &c) + setArray(t, 32, 1216, c) + setArray(t, 33, 1195, c) + setArray(t, 48, 1186, c) + setArray(t, 49, 1223, c) + setArray(t, 50, 1223, c) } @@ -385,8 +385,8 @@ func TestTx_CursorCrashBitmap(t *testing.T) { if err != nil { t.Fatal(err) } - setArray(t, 0, 22510, &c) - setArray(t, 1, 23584, &c) + setArray(t, 0, 22510, c) + setArray(t, 1, 23584, c) } func setArray(tb testing.TB, key, num int, c *rbf.Cursor) { @@ -488,6 +488,30 @@ func TestTx_Dump(t *testing.T) { } } +func TestTx_CreateBitmap(t *testing.T) { + t.Run("Bulk", func(t *testing.T) { + db := MustOpenDB(t) + defer MustCloseDB(t, db) + + tx, err := db.Begin(true) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + + if err := tx.CreateBitmap(fmt.Sprintf("%4000x", 0)); err != nil { + t.Fatal(err) + } else if err := tx.CreateBitmap(fmt.Sprintf("%4000x", 1)); err != nil { + t.Fatal(err) + } else if err := tx.CreateBitmap(fmt.Sprintf("%4000x", 2)); err != nil { + t.Fatal(err) + } + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + }) +} + func rbfName(index, field, view string, shard uint64) string { return string(txkey.Prefix(index, field, view, shard)) } diff --git a/stattx.go b/stattx.go index 383ad0d78..af9fff542 100644 --- a/stattx.go +++ b/stattx.go @@ -19,6 +19,7 @@ import ( "io" "math" "os" + "runtime" "sort" "sync" "time" @@ -108,6 +109,11 @@ func (c *callStats) report() (r string) { for i := range lines { r += lines[i].Line } + + var m1 runtime.MemStats + runtime.ReadMemStats(&m1) + r += fmt.Sprintf("\n m1.TotalAlloc = %v\n", m1.TotalAlloc) + return } diff --git a/tx.go b/tx.go index 61c2dc212..aa7cde450 100644 --- a/tx.go +++ b/tx.go @@ -109,6 +109,7 @@ type Tx interface { // ContainerIterator must not have side-effects. blueGreenTx will // call it at the very beginning of commit to verify db contents. // + // citer.Close() must be called when the client is done using it. ContainerIterator(index, field, view string, shard uint64, ckey uint64) (citer roaring.ContainerIterator, found bool, err error) // RoaringBitmap retreives the roaring.Bitmap for the entire shard.