From 627b50d89df662aa4395998a4069dd6142ead7f4 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Mon, 31 Aug 2020 08:46:53 -0600 Subject: [PATCH] misc import optimizations --- rbf/cursor.go | 10 +++++----- rbf/db.go | 1 - rbf/rbf.go | 14 +++++++------- rbf/tx.go | 24 ++++++++++++++++++------ rbf/wal.go | 8 ++++++-- 5 files changed, 36 insertions(+), 21 deletions(-) diff --git a/rbf/cursor.go b/rbf/cursor.go index 5045565d2..bdd67e0ad 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -429,10 +429,10 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) { // Initialize a new root if we are currently the root page. if c.stack.index == 0 { - assert(newRoot, "leaf write must be root when stack at root") + assert(newRoot) // leaf write must be root when stack at root return c.writeRoot(origPgno, parents) } - assert(!newRoot, "leaf write must NOT be root when stack not at root") + assert(!newRoot) // leaf write must NOT be root when stack not at root // Otherwise update existing parent. return c.putBranchCells(c.stack.index-1, parents) @@ -553,10 +553,10 @@ func (c *Cursor) putBranchCells(stackIndex int, newCells []branchCell) (err erro // Initialize a new root if we are currently the root page. if stackIndex == 0 { - assert(newRoot, "branch write must be root when stack at root") + assert(newRoot) // branch write must be root when stack at root return c.writeRoot(origPgno, parents) } - assert(!newRoot, "branch write must NOT be root when stack at root") + assert(!newRoot) // branch write must NOT be root when stack at root // Otherwise update existing parent. return c.putBranchCells(stackIndex-1, parents) @@ -817,7 +817,7 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) { c.buffered = true for c.stack.index = 0; ; c.stack.index++ { elem := &c.stack.elems[c.stack.index] - assert(elem.pgno != 0, "cursor should never point to page zero (meta)") + assert(elem.pgno != 0) // cursor should never point to page zero (meta) buf, err := c.tx.readPage(elem.pgno) if err != nil { diff --git a/rbf/db.go b/rbf/db.go index 8816fdedd..fcb5f86a9 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -377,7 +377,6 @@ func (db *DB) WALPageN() int64 { // SyncWAL flushes the active segment to disk. func (db *DB) SyncWAL() error { - if s := db.ActiveWALSegment(); s != nil { return s.Sync() } diff --git a/rbf/rbf.go b/rbf/rbf.go index 728f530dc..f152e3499 100644 --- a/rbf/rbf.go +++ b/rbf/rbf.go @@ -434,12 +434,12 @@ func (c *leafCell) countRange(start, end int32) (n int) { func readLeafCellKey(page []byte, i int) uint64 { offset := readCellOffset(page, i) - assert(offset < len(page), "cell read beyond page size: offset %d >= page size %d", offset, len(page)) + assert(offset < len(page)) // cell read beyond page size return *(*uint64)(unsafe.Pointer(&page[offset])) } func readLeafCell(page []byte, i int) leafCell { - assert(i < readCellN(page), "cell index %d exceeds cell count %d", i, readCellN(page)) + assert(i < readCellN(page)) // cell index exceeds cell count offset := readCellOffset(page, i) buf := page[offset:] @@ -487,7 +487,7 @@ func writeLeafCell(page []byte, i, offset int, cell leafCell) { *(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Type) *(*uint16)(unsafe.Pointer(&page[offset+12])) = uint16(cell.N) *(*uint16)(unsafe.Pointer(&page[offset+14])) = uint16(cell.BitN) - assert(offset+16+len(cell.Data) <= PageSize, "leaf cell write extends beyond page: offset %d + len(cell.Data)(%v) + 16 == %v > page size %d", offset, len(cell.Data), offset+16+len(cell.Data), PageSize) + assert(offset+16+len(cell.Data) <= PageSize) // leaf cell write extends beyond page copy(page[offset+16:], cell.Data) } @@ -513,8 +513,8 @@ func readBranchCellKey(page []byte, i int) uint64 { } func readBranchCell(page []byte, i int) branchCell { - assert(i >= 0, "branch cell index must be zero or greater: index=%d", i) - assert(i < readCellN(page), "branch cell index %d must less than cell count %d", i, readCellN(page)) + assert(i >= 0) // branch cell index must be zero or greater + assert(i < readCellN(page)) // branch cell index must less than cell count offset := readCellOffset(page, i) var cell branchCell @@ -625,9 +625,9 @@ func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) { } } -func assert(condition bool, format string, args ...interface{}) { +func assert(condition bool) { if !condition { - panic(fmt.Sprintf("assertion failed: "+format, args...)) + panic("assertion failed") } } diff --git a/rbf/tx.go b/rbf/tx.go index 9135ef752..b884e70ac 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -604,7 +604,10 @@ func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) { func (tx *Tx) Container(name string, key uint64) (*roaring.Container, error) { tx.mu.RLock() defer tx.mu.RUnlock() + return tx.container(name, key) +} +func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) { if tx.db == nil { return nil, ErrTxClosed } else if name == "" { @@ -626,9 +629,12 @@ func (tx *Tx) Container(name string, key uint64) (*roaring.Container, error) { func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error { tx.mu.Lock() defer tx.mu.Unlock() + return tx.putContainer(name, key, ct) +} +func (tx *Tx) putContainer(name string, key uint64, ct *roaring.Container) error { if tx.DeleteEmptyContainer && ct.N() == 0 { - return tx.RemoveContainer(name, key) + return tx.removeContainer(name, key) } cell := ConvertToLeafArgs(key, ct) @@ -650,7 +656,10 @@ func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error func (tx *Tx) RemoveContainer(name string, key uint64) error { tx.mu.Lock() defer tx.mu.Unlock() + return tx.removeContainer(name, key) +} +func (tx *Tx) removeContainer(name string, key uint64) error { c, err := tx.cursor(name) if err != nil { return err @@ -1414,6 +1423,9 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear return } + tx.mu.Lock() + defer tx.mu.Unlock() + if err = tx.createBitmapIfNotExists(name); err != nil { return } @@ -1438,7 +1450,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear } // INVAR: nsynth > 0 - oldC, err = tx.Container(name, itrKey) + oldC, err = tx.container(name, itrKey) panicOn(err) if err != nil { return @@ -1454,7 +1466,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear changed += nsynth rowSet[currRow] += nsynth - err = tx.PutContainer(name, itrKey, synthC) + err = tx.putContainer(name, itrKey, synthC) if err != nil { return } @@ -1476,7 +1488,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear changes := int(existN - newC.N()) changed += changes rowSet[currRow] -= changes - err = tx.PutContainer(name, itrKey, newC) + err = tx.putContainer(name, itrKey, newC) if err != nil { return } @@ -1494,7 +1506,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.PutContainer(name, itrKey, synthC) + err = tx.putContainer(name, itrKey, synthC) if err != nil { return } @@ -1511,7 +1523,7 @@ func (tx *Tx) ImportRoaringBits(name string, itr roaring.RoaringIterator, clear changed += changes rowSet[currRow] += changes - err = tx.PutContainer(name, itrKey, newC) + err = tx.putContainer(name, itrKey, newC) if err != nil { panicOn(err) return diff --git a/rbf/wal.go b/rbf/wal.go index aee4b07a3..fe7fcd40f 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -154,6 +154,7 @@ func (s *WALSegment) closeForWrite() error { if err := s.sync(); err != nil { return err } + s.writeCache = nil // Close underlying file writer. if s.w != nil { @@ -190,7 +191,7 @@ func (s *WALSegment) ReadWALPage(walID int64) ([]byte, error) { // WriteWALPage writes a single page to the WAL segment and returns its WAL identifier. func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err error) { - assert(len(page) == PageSize, "invalid page size: %d", len(page)) + assert(len(page) == PageSize) // invalid page size s.mu.Lock() defer s.mu.Unlock() @@ -212,6 +213,9 @@ func (s *WALSegment) WriteWALPage(page []byte, isMeta bool) (walID int64, err er } // Append write to write buffer & increment page count. + if s.writeCache == nil { + s.writeCache = make([]byte, 0, MaxWALSegmentFileSize+PageSize) + } s.writeCache = append(s.writeCache, page...) s.pageN++ @@ -229,7 +233,7 @@ func (s *WALSegment) flush() error { if _, err := s.w.WriteAt(s.writeCache, int64((s.pageN*PageSize)-len(s.writeCache))); err != nil { return fmt.Errorf("wal segment write: %w", err) } - s.writeCache = nil + s.writeCache = s.writeCache[:0] return nil }