From adcd5adb0293a46b307a564c10cd7de93f2a31af Mon Sep 17 00:00:00 2001 From: Seebs Date: Mon, 10 Jan 2022 10:33:01 -0600 Subject: [PATCH 1/7] improve the sync.Pool used for pages, avoid excess page allocations for WAL Several changes. One is, we don't provide a `New` for pagePool, which allows allocPage to check whether a page was returned, and thus, zero pages which were found in the pool, or make new pages, but never zero pages it just created with make. We then also make many more things which were making pages use the pool. Reuse the same page allocation for multiple header pages dumped into the WAL; the bitmap header pages aren't stashed in our page map, they're only written to the disk, so we don't need to make a new page each time, we can just make one new page for the whole batch. Internally in the pool, we pool pointers to [PageSize]byte, rather than slices. sync.Pool needs pointer-like things. To store a pointer to a slice, you have to heap-allocate the slice, also. So, instead of heap-allocating copies of these slices, we just use pointers to the raw data. --- rbf/cursor.go | 6 +++--- rbf/db.go | 29 +++++++++++++++++------------ rbf/tx.go | 13 ++++++++++--- 3 files changed, 30 insertions(+), 18 deletions(-) diff --git a/rbf/cursor.go b/rbf/cursor.go index 2419bc5f8..22d4427e6 100644 --- a/rbf/cursor.go +++ b/rbf/cursor.go @@ -526,7 +526,7 @@ func (c *Cursor) putLeafCellFast(in leafCell, isInsert bool) (err error) { } // Write page header. - dst := allocPage() // make([]byte, PageSize) + dst := allocPage() writePageNo(dst, readPageNo(src)) writeFlags(dst, PageTypeLeaf) writeCellN(dst, dstCellN) @@ -616,7 +616,7 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) { cells = cells[:len(cells)-1] // Write cells to page. - buf := make([]byte, PageSize) + buf := allocPage() writePageNo(buf[:], elem.pgno) writeFlags(buf[:], PageTypeLeaf) writeCellN(buf[:], len(cells)) @@ -800,7 +800,7 @@ func (c *Cursor) deleteBranchCell(stackIndex int, key uint64) (err error) { return err } - buf := make([]byte, PageSize) + buf := allocPage() copy(buf, target) writePageNo(buf[:], elem.pgno) diff --git a/rbf/db.go b/rbf/db.go index 97de950e6..1d693fac6 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -11,6 +11,7 @@ import ( "sort" "sync" "syscall" + "unsafe" "github.com/benbjohnson/immutable" "github.com/molecula/featurebase/v2/logger" @@ -560,7 +561,7 @@ func (db *DB) init() error { // initMetaPage initializes the meta page. func (db *DB) initMetaPage() error { - page := make([]byte, PageSize) + page := allocPage() writeMetaMagic(page) writeMetaPageN(page, 3) writeMetaRootRecordPageNo(page, 1) @@ -572,7 +573,7 @@ func (db *DB) initMetaPage() error { // initRootRecordPage initializes the initial root record page. func (db *DB) initRootRecordPage() error { - page := make([]byte, PageSize) + page := allocPage() writePageNo(page, 1) writeFlags(page, PageTypeRootRecord) _, err := db.file.WriteAt(page, 1*PageSize) @@ -582,7 +583,7 @@ func (db *DB) initRootRecordPage() error { // initFreelistPage initializes the initial freelist btree page. func (db *DB) initFreelistPage() error { - page := make([]byte, PageSize) + page := allocPage() writePageNo(page, 2) writeFlags(page, PageTypeLeaf) _, err := db.file.WriteAt(page, 2*PageSize) @@ -829,18 +830,22 @@ type DebugInfo struct { // Shared pool for in-memory database pages. // These are used before being flushed to disk. -var pagePool = &sync.Pool{ - New: func() interface{} { - page := make([]byte, PageSize) - return &page - }, -} +var pagePool = &sync.Pool{} func allocPage() []byte { - page := pagePool.Get().(*[]byte) - return *page + existing := pagePool.Get() + if existing == nil { + return make([]byte, PageSize) + } + // zero the existing page before returning it + page := existing.(*[PageSize]byte)[:] + for i := range page { + page[i] = 0 + } + return page } func freePage(page []byte) { - pagePool.Put(&page) + data := (*[PageSize]byte)(unsafe.Pointer(&page[0])) + pagePool.Put(data) } diff --git a/rbf/tx.go b/rbf/tx.go index aef2b395c..b2a1a5185 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -236,7 +236,7 @@ func (tx *Tx) createBitmap(name string) error { } // Write root page. - page := make([]byte, PageSize) + page := allocPage() writePageNo(page, pgno) writeFlags(page, PageTypeLeaf) writeCellN(page, 0) @@ -449,7 +449,7 @@ func (tx *Tx) writeRootRecordPages(records *immutable.SortedMap) (err error) { // Write new root record pages. for itr := records.Iterator(); !itr.Done(); { // Initialize page & write as many records as will fit. - page := make([]byte, PageSize) + page := allocPage() writePageNo(page, pgno) writeFlags(page, PageTypeRootRecord) @@ -1800,9 +1800,16 @@ func (tx *Tx) flush() error { } // Write bitmap headers & pages to WAL. + // + // We need to write a bitmap header before each such page. We only allocate + // one header, and we reuse it, because each write is flushing it out to + // disk, and it doesn't get stored in-memory. + var hdr []byte + if len(tx.dirtyBitmapPages) > 0 { + hdr = allocPage() + } for _, pgno := range dirtyPageMapKeys(tx.dirtyBitmapPages) { // Write header page. - hdr := make([]byte, PageSize) writePageNo(hdr[:], pgno) writeFlags(hdr[:], PageTypeBitmapHeader) if _, err := tx.writeToWAL(w, hdr); err != nil { From 112abcb549e86135cf8abd55b2e756ff4b93b8dd Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 7 Jan 2022 15:59:49 -0600 Subject: [PATCH 2/7] use stable cursor for freelist operations The Cursor datatype is quite large, and allocating them constantly for ops is extremely expensive. To avoid this, we create a single stable cursor that lives in the DB, and can be used for freelist modifications. Since the freelist is only ever modified once at a time, this should be safe. We also don't fully zero it between operations, we just reset the relevant parts. --- rbf/db.go | 23 ++++++++++++++++++ rbf/tx.go | 73 ++++++++++++++++++++++++++----------------------------- 2 files changed, 58 insertions(+), 38 deletions(-) diff --git a/rbf/db.go b/rbf/db.go index 1d693fac6..f2eeceea3 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -69,6 +69,8 @@ type DB struct { // Path represents the path to the database file. Path string + + freelistCursor Cursor // cursor to reuse for freelist operations } // NewDB returns a new instance of DB. @@ -808,6 +810,11 @@ func (db *DB) readMetaPage() ([]byte, error) { return db.readDBPage(0) } +// getCursor returns a cursor which has not been zeroed. The only thing +// a caller should need to do is set c.stack's top correctly (it should be +// 0, and the [0] elem should be the root page to start on). +// +// TODO: Should this do anything about c.buffered? func (db *DB) getCursor(tx *Tx) *Cursor { c := cursorSyncPool.Get().(*Cursor) c.tx = tx @@ -828,6 +835,22 @@ type DebugInfo struct { Txs []*TxDebugInfo `json:"txs"` } +// when we want a cursor to access a free list, we are always doing this in +// a context specific to a write transaction, of which any DB can only have +// one at a time, and the operations modifying the free list don't recurse, +// because that would corrupt the list (see tx.freelistCleanup for the hairy +// details), which means that there is only ever one cursor being used for the +// free list, but also we use that cursor very often, and if we have to allocate +// it or zero it we end up with a lot of excess allocations and zeroing. +func (db *DB) getFreelistCursor(tx *Tx) *Cursor { + c := &db.freelistCursor + c.tx = tx + c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])} + c.stack.top = 0 + c.buffered = false + return c +} + // Shared pool for in-memory database pages. // These are used before being flushed to disk. var pagePool = &sync.Pool{} diff --git a/rbf/tx.go b/rbf/tx.go index b2a1a5185..4fa8f441e 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -984,6 +984,10 @@ func (tx *Tx) walkTree(pgno, parent uint32, fn func(pgno, parent, typ uint32, er // about that removing things from the free list, because the add logic // already just uses new pages rather than trying to use the free list // when it knows the free list is involved. +// +// Because this is expected to be used in a defer, instead of returning an +// error, it will set the error it got the address of to a new error if it +// encounters one and there wasn't one already. func (tx *Tx) freelistCleanup(outErr *error) { defer func() { // no matter what, we're done with this after this, but we still @@ -994,8 +998,7 @@ func (tx *Tx) freelistCleanup(outErr *error) { if len(tx.pendingFreelistAdds) == 0 { return } - c := Cursor{tx: tx} - c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])} + c := tx.db.getFreelistCursor(tx) for len(tx.pendingFreelistAdds) > 0 { var pass []uint32 pass, tx.pendingFreelistAdds = tx.pendingFreelistAdds, nil @@ -1006,7 +1009,7 @@ func (tx *Tx) freelistCleanup(outErr *error) { } return } else if !changed { - vprint.PanicOn(fmt.Sprintf("rbf.Tx.freePgno(): double free: %d", tx.pendingFreelistAdds)) + vprint.PanicOn(fmt.Sprintf("rbf.Tx.freelistCleanup(): double free: %d", pass)) } } } @@ -1015,44 +1018,25 @@ func (tx *Tx) freelistCleanup(outErr *error) { // allocatePgno returns a page number for a new available page. This page may be // pulled from the free list or, if no free pages are available, it will be // created by extending the file size. +// +// allocatePgno uses the freelist cursor (a shared db-wide thing), and sets +// the "modifyingFreelist" flag while it's running. If for some reason a +// modification to the freelist would require a new allocation or free, +// allocations always just create a new page, and frees are processed later +// by a separate call through a deferred tx.freelistCleanup(). func (tx *Tx) allocatePgno() (_ uint32, outErr error) { if tx.modifyingFreelist { return tx.allocateNewPgno(), nil } - // Attempt to find page in freelist. - pgno, err := tx.nextFreelistPageNo() - - if err != nil { - return 0, err - } else if pgno != 0 { - tx.modifyingFreelist = true - defer tx.freelistCleanup(&outErr) - c := Cursor{tx: tx} - c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])} - if changed, err := c.Remove(uint64(pgno)); err != nil { - return 0, err - } else if !changed { - vprint.PanicOn(fmt.Sprintf("tx.Tx.allocatePgno(): double alloc: %d", pgno)) - } - return pgno, nil - } - // no freelist pages, fall back - return tx.allocateNewPgno(), nil -} - -// allocateNewPgno requests a new page unconditionally, ignoring the free list. -func (tx *Tx) allocateNewPgno() uint32 { - // Increment the total page count by one and return the last page. - pgno := readMetaPageN(tx.meta[:]) - writeMetaPageN(tx.meta[:], pgno+1) - return pgno -} - -func (tx *Tx) nextFreelistPageNo() (uint32, error) { - c := Cursor{tx: tx} - c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])} + // this serves as a precaution against double-use of the freelist cursor + // used database-wide. we don't have actual synchronization here because + // only one write Tx should exist at once and it's not safe to use its + // write-capable ops concurrently anyway. + tx.modifyingFreelist = true + defer tx.freelistCleanup(&outErr) + c := tx.db.getFreelistCursor(tx) if err := c.First(); err == io.EOF { - return 0, nil + return tx.allocateNewPgno(), nil } else if err != nil { return 0, err } @@ -1067,17 +1051,30 @@ func (tx *Tx) nextFreelistPageNo() (uint32, error) { v := cell.firstValue(tx) pgno := uint32((cell.Key << 16) | uint64(v)) + + if changed, err := c.Remove(uint64(pgno)); err != nil { + return 0, err + } else if !changed { + vprint.PanicOn(fmt.Sprintf("tx.Tx.allocatePgno(): double alloc: %d", pgno)) + } return pgno, nil } +// allocateNewPgno requests a new page unconditionally, ignoring the free list. +func (tx *Tx) allocateNewPgno() uint32 { + // Increment the total page count by one and return the last page. + pgno := readMetaPageN(tx.meta[:]) + writeMetaPageN(tx.meta[:], pgno+1) + return pgno +} + // deallocate releases a page number to the freelist. func (tx *Tx) freePgno(pgno uint32) (outErr error) { if tx.modifyingFreelist { tx.pendingFreelistAdds = append(tx.pendingFreelistAdds, pgno) return nil } - c := Cursor{tx: tx} - c.stack.elems[0] = stackElem{pgno: readMetaFreelistPageNo(tx.meta[:])} + c := tx.db.getFreelistCursor(tx) tx.modifyingFreelist = true defer tx.freelistCleanup(&outErr) From 719a30e1289c6c089fcd42cd8025902d5bb97f91 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 19 Jan 2022 13:21:44 -0600 Subject: [PATCH 3/7] shorten MultiTx test The MultiTx test runs for a fairly long time but doesn't add much value running that much longer, and there's no reason it should take more than half the time we spend on this entire directory. --- rbf/db_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rbf/db_test.go b/rbf/db_test.go index f4238e7e1..65d84cff6 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -339,7 +339,7 @@ func TestDB_MultiTx(t *testing.T) { } // Continuously set/clear bits while readers are executing. - for i := 0; i < 1000; i++ { + for i := 0; i < 100; i++ { func() { tx, err := db.Begin(true) if err != nil { From 37507db4ac4ff1a5410aadf230b1519acfede169 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 19 Jan 2022 13:13:05 -0600 Subject: [PATCH 4/7] use array containers instead of individual bitwise adds This affects TestTx_Remove, TestTx_DeallocateToFreeList, and TestTx_RecreateBitmap, all of which were adding hundreds of thousands of individual bits, or more, and all of which work just as well and produce the same behavior using largeish containers. This reduces race-detector-test runtime from about 20 minutes to a couple. --- rbf/tx_test.go | 79 +++++++++++++++++++++++++------------------------- 1 file changed, 40 insertions(+), 39 deletions(-) diff --git a/rbf/tx_test.go b/rbf/tx_test.go index 7a726c2ed..14626ce00 100644 --- a/rbf/tx_test.go +++ b/rbf/tx_test.go @@ -248,6 +248,27 @@ func TestTx_DeallocateTree(t *testing.T) { } } +func arraySizedChunk() []uint16 { + v := make([]uint16, rbf.ArrayMaxSize) + for i := range v { + v[i] = uint16(i) + } + return v +} + +var convenientPrepopulatedArray = arraySizedChunk() + +// populateBitmapWithArrays +func populateBitmapWithArrays(tb testing.TB, tx *rbf.Tx, n int, name string) { + c := roaring.NewContainerArray(convenientPrepopulatedArray) + for i := 0; i < n; i++ { + err := tx.PutContainer(name, uint64(i), c) + if err != nil { + tb.Fatal(err) + } + } +} + func TestTx_RecreateBitmap(t *testing.T) { db := MustOpenDB(t) defer MustCloseDB(t, db) @@ -258,14 +279,8 @@ func TestTx_RecreateBitmap(t *testing.T) { if err := tx.CreateBitmap("x"); err != nil { t.Fatal(err) } - const N = 825000 - slots := make([]uint64, N) - for i := range slots { - slots[i] = uint64(i) << 20 - } - if _, err := tx.Add("x", slots...); err != nil { - t.Fatal(err) - } + const N = 825 + populateBitmapWithArrays(t, tx, N, "x") err := tx.Commit() if err != nil { t.Fatal(err) @@ -291,9 +306,7 @@ func TestTx_RecreateBitmap(t *testing.T) { if err := tx.CreateBitmap("x"); err != nil { t.Fatal(err) } - if _, err := tx.Add("x", slots...); err != nil { - t.Fatal(err) - } + populateBitmapWithArrays(t, tx, N, "x") err = tx.Commit() if err != nil { t.Fatal(err) @@ -374,20 +387,14 @@ func TestTx_DeallocateToFreeList(t *testing.T) { if err = tx.CreateBitmap("y"); err != nil { t.Fatal(err) } - const N = 12274831 - slots := make([]uint64, N) - for i := range slots { - slots[i] = uint64(i) << 10 - } - bm := roaring.NewBitmap(slots...) - if _, err = tx.AddRoaring("x", bm); err != nil { - t.Fatal(err) - } + // Insert large array values. + populateBitmapWithArrays(t, tx, 4080, "x") + if err = tx.Check(); err != nil { t.Fatal(err) } for i := 0; i < 500; i++ { - if _, err := tx.Add("y", uint64(i)<<16); err != nil { + if _, err := tx.Add("y", uint64(i)<<16+32768); err != nil { t.Fatal(err) } } @@ -426,9 +433,8 @@ func TestTx_DeallocateToFreeList(t *testing.T) { if err := tx.CreateBitmap("x"); err != nil { t.Fatal(err) } - if _, err := tx.AddRoaring("x", bm); err != nil { - t.Fatal(err) - } + populateBitmapWithArrays(t, tx, 4080, "x") + if err = tx.Check(); err != nil { t.Fatal(err) } @@ -451,17 +457,7 @@ func TestTx_Remove(t *testing.T) { } // Insert large array values. - var values []uint64 - for i := 0; i < 1000; i++ { - for j := 0; j < rbf.ArrayMaxSize; j++ { - v := uint64((i << 16) + j) - values = append(values, v) - - if _, err := tx.Add("x", v); err != nil { - t.Fatalf("Add(%d) err=%q", v, err) - } - } - } + populateBitmapWithArrays(t, tx, 500, "x") if err := tx.Commit(); err != nil { t.Fatal(err) @@ -471,12 +467,17 @@ func TestTx_Remove(t *testing.T) { defer tx.Rollback() // Remove all array values. - for _, i := range rand.Perm(len(values)) { - v := values[i] - if _, err := tx.Remove("x", v); err != nil { - t.Fatalf("Remove(%d) err=%q", v, err) + for i := 0; i < 500; i++ { + err := tx.RemoveContainer("x", uint64(i)) + if err != nil { + t.Fatal(err) } } + // This triggered a different panic without the relevant patch. + err := tx.RemoveContainer("x", 500) + if err != nil { + t.Fatal(err) + } if err := tx.Commit(); err != nil { t.Fatal(err) From 2dce518a243c4006c75ae7210e2e33e0283ec794 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 19 Jan 2022 14:44:47 -0600 Subject: [PATCH 5/7] retry other etcd ErrTimeout variants etcd can return more detailed ErrTimeout variants in rare cases, and we want to retry on those too. --- etcd/embed.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/etcd/embed.go b/etcd/embed.go index db3bc34ea..a7f1ba073 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -210,7 +210,7 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) { e.cli = cli e.cliMu.Unlock() break - case etcdserver.ErrTimeout: + case etcdserver.ErrTimeout, etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost, etcdserver.ErrTimeoutLeaderTransfer: // sporadic timeouts are concerning but not necessarily fatal // and can usually be retried. elapsed := time.Since(start) From 749dcd69703351179a2341c2872f53a9374ec347 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 19 Jan 2022 16:57:47 -0600 Subject: [PATCH 6/7] retry on etcd timeout errors --- etcd/embed.go | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/etcd/embed.go b/etcd/embed.go index a7f1ba073..a74ae1c7d 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -210,6 +210,13 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) { e.cli = cli e.cliMu.Unlock() break + default: + msg := err.Error() + if !strings.HasPrefix(msg, "etcdserver: request timed out") { + // not a known error, also not a wrapped timeout + return errors.Wrap(err, "non-retryable error") + } + fallthrough // treat this as being like a timeout error case etcdserver.ErrTimeout, etcdserver.ErrTimeoutDueToLeaderFail, etcdserver.ErrTimeoutDueToConnectionLost, etcdserver.ErrTimeoutLeaderTransfer: // sporadic timeouts are concerning but not necessarily fatal // and can usually be retried. @@ -225,9 +232,6 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) { // from spamming these. time.Sleep(100 * time.Millisecond) break - default: - // nil, or an error we don't know about - return errors.Wrap(err, "non-retryable error") } } // if we got here, we got a total of three of some combination of From fb118959856f192118afa8fa0115abdaa60c1e99 Mon Sep 17 00:00:00 2001 From: Seebs Date: Wed, 19 Jan 2022 17:09:57 -0600 Subject: [PATCH 7/7] oops handle nil --- etcd/embed.go | 2 ++ 1 file changed, 2 insertions(+) diff --git a/etcd/embed.go b/etcd/embed.go index a74ae1c7d..0167f6b42 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -210,6 +210,8 @@ func (e *Etcd) retryClient(fn func(cli *clientv3.Client) error) (err error) { e.cli = cli e.cliMu.Unlock() break + case nil: + return nil default: msg := err.Error() if !strings.HasPrefix(msg, "etcdserver: request timed out") {