From b8f59d922c72e1e1b8d4b1ee36bbd89e6e0d9096 Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 19 Nov 2021 14:37:59 -0600 Subject: [PATCH] checkpoint rework/refactoring: logger, async-ish checkpoint Trying to make the checkpoint be asynchronous-at-all, and also allowing it to log. --- rbf/cfg/cfg.go | 6 ++ rbf/db.go | 162 ++++++++++++++++++++++++++++++++++++++---------- rbf/rbf_test.go | 8 +++ 3 files changed, 144 insertions(+), 32 deletions(-) diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go index cc2cf7a8a..671c6fe43 100644 --- a/rbf/cfg/cfg.go +++ b/rbf/cfg/cfg.go @@ -2,6 +2,7 @@ package cfg import ( + "github.com/molecula/featurebase/v2/logger" "github.com/spf13/pflag" ) @@ -35,6 +36,11 @@ type Config struct { // CursorCacheSize is the number of copies of Cursor{} to keep in our // readyCursorCh arena to avoid GC pressure. CursorCacheSize int64 `toml:"cursor-cache-size"` + + // Logger specifies a logger for asynchronous errors, such as + // background checkpoints. It cannot be set from toml. The default is + // to use stderr. + Logger logger.Logger `toml:"-"` } func NewDefaultConfig() *Config { diff --git a/rbf/db.go b/rbf/db.go index 7f27ca418..830e31f5a 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -11,6 +11,7 @@ import ( "syscall" "github.com/benbjohnson/immutable" + "github.com/molecula/featurebase/v2/logger" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/syswrap" ) @@ -38,6 +39,7 @@ type DB struct { pageMap *PageMap // pgno-to-WALID mapping txs map[*Tx]struct{} // active transactions opened bool // true if open + logger logger.Logger // for diagnostics from async things wal []byte // wal mmap walFile *os.File // wal file descriptor @@ -62,6 +64,11 @@ func NewDB(path string, cfg *rbfcfg.Config) *DB { txs: make(map[*Tx]struct{}), pageMap: NewPageMap(), Path: path, + logger: cfg.Logger, + } + if db.logger == nil { + // default to writing to stdout if not told otherwise + db.logger = logger.NewStandardLogger(os.Stderr) } db.haltCond = sync.NewCond(&db.mu) @@ -134,7 +141,7 @@ func (db *DB) Open() (err error) { if err := db.openWAL(); err != nil { return fmt.Errorf("wal open: %w", err) } else if err := db.checkpoint(); err != nil { - return fmt.Errorf("checkpoint: %w", err) + return fmt.Errorf("startup checkpoint: %w", err) } return nil @@ -158,10 +165,12 @@ func (db *DB) openWAL() (err error) { // Determine the number of whole pages in the WAL. var pageN int + var fileSize int64 if fi, err := db.walFile.Stat(); err != nil { return fmt.Errorf("wal stat: %w", err) } else { - pageN = int(fi.Size() / PageSize) + fileSize = fi.Size() + pageN = int(fileSize / PageSize) } // Read backwards through the WAL to find the last valid meta page. @@ -169,14 +178,45 @@ func (db *DB) openWAL() (err error) { if page, err := db.readWALPageAt(pageN - 1); err != nil { return err } else if IsMetaPage(page) { + // We now face a challenge. Probably this is a meta page. + // But consider a sequence of pages written which gets + // interrupted right before the meta page is written. + // If the last page is a bitmap page, it could LOOK LIKE a meta + // page. So we have to check the page before it. If that page + // is a bitmap header, then actually this is a bitmap page, right? + // If that page doesn't exist, of course, we're fine, except + // for the philosophical question of why we wrote a meta page + // when no pages had changed. + if pageN > 1 { + if page, err = db.readWALPageAt(pageN - 2); err != nil { + return err + } + if IsBitmapHeader(page) { + // But wait! + // What if this *is* a meta page, and the page before it is + // actually a *bitmap page* that looks like a bitmap header? And + // so on. + // + // Rather than try to resolve this, in this insanely unlikely + // situation, we read from the beginning which allows us to + // always know what we're seeing, because every bitmap page + // comes *after* a bitmap header page, and thus, we know when + // we might be seeing one. + pageN, err = db.methodicalWALPageN(pageN) + if err != nil { + return err + } + } + } break } } - - // Truncate WAL to the last valid meta page. - if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil { - return fmt.Errorf("wal truncate: %w", err) - } else if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil { + if fileSize != int64(pageN*PageSize) { + if err := db.walFile.Truncate(int64(pageN * PageSize)); err != nil { + return fmt.Errorf("wal truncate: %w", err) + } + } + if _, err := db.walFile.Seek(int64(pageN*PageSize), io.SeekStart); err != nil { return fmt.Errorf("wal seek: %w", err) } db.walPageN = pageN @@ -184,6 +224,27 @@ func (db *DB) openWAL() (err error) { return nil } +// methodicalWALPageN tries to determine the last meta page in a very reliable +// but slow way. This handles the theoretical but hard to imagine creating +// edge case where we have a bitmap page which happens to look like a meta +// page, and the write got interrupted before the meta page got written. +func (db *DB) methodicalWALPageN(pageN int) (lastMeta int, err error) { + for i := 0; i < pageN; i++ { + var page []byte + if page, err = db.readWALPageAt(i); err != nil { + return -1, err + } + switch { + case IsMetaPage(page): + lastMeta = i + case IsBitmapHeader(page): + // skip the bitmap page, which we can't usefully evaluate + i++ + } + } + return lastMeta, nil +} + // checkpoint moves all WAL pages to the main DB file. // Must be called by a write transaction while under db.mu lock. func (db *DB) checkpoint() error { @@ -199,28 +260,53 @@ func (db *DB) checkpoint() error { if db.walPageN == 0 { return nil } + // We might have either a *PageMap or just the file. If we have the file, + // building the PageMap is fairly expensive because it's fancy and immutable. + // If we have the PageMap *or* some other map, that's two different things + // to iterate. If we have the PageMap, building a map from it is relatively + // cheap, so we'll do it that way. + pages := make(map[uint32]int) + if db.pageMap.size == 0 { + // you'd think we're done, but actually this PROBABLY means that + // this is initial startup, and we haven't read the file yet. We scan + // the file for pages, because it turns out most of them probably + // got overwritten. + for i := 0; i < db.walPageN; i++ { + page, err := db.readWALPageAt(i) + if err != nil { + return err + } + + // Determine page number. Meta pages are always on zero & bitmap + // headers specify the page number of the next page in the WAL. + // All other pages have their page number in the page data. + var pgno uint32 + if IsBitmapHeader(page) { + pgno = readPageNo(page) + if page, err = db.readWALPageAt(i + 1); err != nil { + return err + } + i++ // bitmaps in WAL are two pages + } else if !IsMetaPage(page) { + pgno = readPageNo(page) + } + // record where in the file we have this page + pages[pgno] = i + } + } else { + itr := db.pageMap.Iterator() + itr.First() + for k, v, ok := itr.Next(); ok; k, v, ok = itr.Next() { + pages[k] = int(v) + } + } // fmt.Printf("checkpoint: walPageN %d, PageMap size %d\n", db.walPageN, db.pageMap.size) - for i := 0; i < db.walPageN; i++ { - page, err := db.readWALPageAt(i) + for pgno, walID := range pages { + page, err := db.readWALPageAt(walID) if err != nil { return err } - - // Determine page number. Meta pages are always on zero & bitmap - // headers specify the page number of the next page in the WAL. - // All other pages have their page number in the page data. - var pgno uint32 - if IsBitmapHeader(page) { - pgno = readPageNo(page) - if page, err = db.readWALPageAt(i + 1); err != nil { - return err - } - i++ // bitmaps in WAL are two pages - } else if !IsMetaPage(page) { - pgno = readPageNo(page) - } - // Write data to the data file. if err := db.writeDBPage(pgno, page); err != nil { return err @@ -509,23 +595,35 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { // running new tx. func (db *DB) removeTx(tx *Tx) error { db.mu.Lock() - defer db.mu.Unlock() - // Release writer lock if tx is writable. + // release the write lock. we have to do this for now. some day we won't, + // and will want to hold it, but right now we can't be sure we can get it. if tx.writable { tx.db.rwmu.Unlock() } - + // remove ourselves from the list of transactions the db is keeping. delete(tx.db.txs, tx) // Disassociate from db. tx.db = nil - // Write pages from WAL to DB. - // TODO(bbj): Move this to an async goroutine. + // Write pages from WAL to DB. As of this instant, we are the ONLY + // transaction, which means that no transaction has an older version + // of the PageMap than we do, and if we're a write, Commit() already + // updated the page map to our page map. So, if we *can* checkpoint, + // the checkpoint gets spawned asynchronously. We could block writes, + // except doing so will deadlock in a weird way. if len(db.txs) == 0 && db.walSize() > db.cfg.MinWALCheckpointSize { - if err := db.checkpoint(); err != nil { - return fmt.Errorf("checkpoint: %w", err) - } + // We are doing this function with the db lock held, which means + // we're *not* releasing the db lock, even though we're returning. + // This is a weird special case, and probably a bad idea. + go func() { + defer db.mu.Unlock() + if err := db.checkpoint(); err != nil { + db.logger.Errorf("async checkpoint: %w", err) + } + }() + } else { + defer db.mu.Unlock() } return nil diff --git a/rbf/rbf_test.go b/rbf/rbf_test.go index 48f16335a..be3461e37 100644 --- a/rbf/rbf_test.go +++ b/rbf/rbf_test.go @@ -11,6 +11,7 @@ import ( "sort" "testing" + "github.com/molecula/featurebase/v2/logger" "github.com/molecula/featurebase/v2/rbf" rbfcfg "github.com/molecula/featurebase/v2/rbf/cfg" "github.com/molecula/featurebase/v2/testhook" @@ -65,6 +66,13 @@ func NewDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { // MustOpenDB returns a db opened on a temporary file. On error, fail test. func MustOpenDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { tb.Helper() + if len(cfg) == 0 || cfg[0] == nil { + newconf := rbfcfg.NewDefaultConfig() + newconf.Logger = logger.NewLogfLogger(tb) + cfg = []*rbfcfg.Config{newconf} + } else if cfg[0].Logger == nil { + cfg[0].Logger = logger.NewLogfLogger(tb) + } db := NewDB(tb, cfg...) if err := db.Open(); err != nil { tb.Fatal(err)