From ea3732fa62f220d04eec91d81236a8cf6cc502c9 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Fri, 30 Oct 2020 17:12:05 -0600 Subject: [PATCH 1/2] Fix WAL ID not found panic. This commit changes the checkpointing to determine a minimum WAL ID for readers and a max ID based on the writer. Pages are checkpointed from the WAL up to the writer's max WAL ID but segments are removed only up to the reader's minimum WAL ID. This ensures that WAL pages are not removed out from under current read transactions. --- go.sum | 1 + rbf/db.go | 85 +++++++++++++++++++++++++++++++++---------------- rbf/db_test.go | 78 +++++++++++++++++++++++++++++++++++++++++++++ rbf/rbf_test.go | 13 +++++--- 4 files changed, 145 insertions(+), 32 deletions(-) diff --git a/go.sum b/go.sum index ea9c803a4..e18eaa9eb 100644 --- a/go.sum +++ b/go.sum @@ -161,6 +161,7 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= +github.com/pilosa/pilosa v1.4.1 h1:zSNyS/MqXTfRDNRBBNCApsOu2oUnTEj2uymCaqYSNx8= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= diff --git a/rbf/db.go b/rbf/db.go index 1e192ef18..c65c77bcd 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -238,10 +238,9 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { return err } walID := readMetaWALID(page) - maxCheckpointedWALID := walID // Determine the high water mark for WAL pages that can be copied. - minActiveWALID := db.minActiveWALID() + writerWALID := db.writerWALID() // Loop over each transaction walID++ @@ -257,7 +256,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { // Loop over pages in the transaction. for ; walID <= metaWALID; walID++ { - canCheckpoint := exclusive || minActiveWALID == 0 || walID < minActiveWALID + canCheckpoint := exclusive || writerWALID == 0 || walID < writerWALID page, err := readWALPage(segments, walID) if err != nil { @@ -293,11 +292,6 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { if err := db.writeDBPage(pgno, page); err != nil { return err } - - // Track highest WALID that has been checkpointed back to disk. - if IsMetaPage(page) { - maxCheckpointedWALID = walID - } } } @@ -305,21 +299,23 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { if err := db.fsync(db.file); err != nil { return fmt.Errorf("db file sync: %w", err) } + mu.Lock() + db.pageMap = pageMap + mu.Unlock() // Remove WAL segments that have been checkpointed. - if maxCheckpointedWALID != 0 { - for _, segment := range segments { - if segment.MaxWALID() > maxCheckpointedWALID { - break - } + minPageMapWALID := db.minPageMapWALID() + for _, segment := range segments { + if minPageMapWALID != 0 && segment.MaxWALID() > minPageMapWALID { + break + } - if err := func() error { - mu.Lock() - defer mu.Unlock() - return db.removeWALSegment(segment.Path) - }(); err != nil { - return err - } + if err := func() error { + mu.Lock() + defer mu.Unlock() + return db.removeWALSegment(segment.Path) + }(); err != nil { + return err } } @@ -331,7 +327,6 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { mu.Unlock() } - db.pageMap = pageMap return nil } @@ -361,16 +356,38 @@ func (db *DB) removeWALSegment(path string) error { return nil } -// minActiveWALID returns the lowest WAL ID in use by any active transaction. -// Returns 0 if no transactions are active. -func (db *DB) minActiveWALID() int64 { - var walID int64 +// minPageMapWALID returns the lowest WAL ID referenced by an active page map. +func (db *DB) minPageMapWALID() int64 { + // Use the db's page map because that is the state of the map when the + // writer transaction started. We can't use the writer transaction's map + // because it can change. + min := pageMapMinWALID(db.pageMap) + for tx := range db.txs { - if walID == 0 || walID > tx.walID { - walID = tx.walID + // If a write transaction is active, ensure the min is at least the starting WAL. + if tx.writable { + if min == 0 || tx.walID < min { + min = tx.walID + } + continue + } + + // Record the min WAL ID referenced by the reader's page map. + if walID := pageMapMinWALID(tx.pageMap); min == 0 || walID < min { + min = walID } } - return walID + return min +} + +// writerWALID returns the starting WAL ID of the active writer tx. +func (db *DB) writerWALID() int64 { + for tx := range db.txs { + if tx.writable { + return tx.walID + } + } + return 0 } // Close closes the database. @@ -738,3 +755,15 @@ type nopLocker struct{} func (*nopLocker) Lock() {} func (*nopLocker) Unlock() {} + +// pageMapMinWALID returns the lowest WAL ID +func pageMapMinWALID(m *immutable.Map) int64 { + var min int64 + for itr := m.Iterator(); !itr.Done(); { + _, v := itr.Next() + if walID := v.(int64); min == 0 || walID < min { + min = walID + } + } + return min +} diff --git a/rbf/db_test.go b/rbf/db_test.go index 11db76c62..a70ad1d6a 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -15,6 +15,7 @@ package rbf_test import ( + "context" "fmt" "math/rand" "net" @@ -24,6 +25,8 @@ import ( "time" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" + "golang.org/x/sync/errgroup" _ "net/http/pprof" ) @@ -290,6 +293,81 @@ func TestDB_HasData(t *testing.T) { } } +// Ensures the DB can continuously write while readers are executing. +func TestDB_MultiTx(t *testing.T) { + if testing.Short() { + t.Skip("-short enabled, skipping") + } + + cfg := rbfcfg.NewDefaultConfig() + cfg.CheckpointEveryDur = 1 * time.Millisecond + db := MustOpenDB(t, cfg) + defer MustCloseDB(t, db) + + // Run multiple readers in separate goroutines. + ctx, cancel := context.WithCancel(context.Background()) + g, ctx := errgroup.WithContext(ctx) + for i := 0; i < 10; i++ { + g.Go(func() error { + for { + if ctx.Err() != nil { + return nil // cancelled, return no error + } else if err := testDB_MultiTx_reader(db); err != nil { + return err + } + } + }) + } + + // Continuously set/clear bits while readers are executing. + for i := 0; i < 1000; i++ { + func() { + tx, err := db.Begin(true) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback() + + for j := 0; j < rand.Intn(10); j++ { + v := rand.Intn(1 << 20) + if _, err := tx.Add("x", uint64(v)); err != nil { + t.Fatal(err) + } + } + + if err := tx.Commit(); err != nil { + t.Fatal(err) + } + }() + } + + // Stop readers & wait. + cancel() + if err := g.Wait(); err != nil { + t.Fatal(err) + } +} + +// testDB_MultiTx_reader checks if a bitmap contains a random set of bits. +func testDB_MultiTx_reader(db *rbf.DB) error { + tx, err := db.Begin(false) + if err != nil { + return err + } + defer tx.Rollback() + + time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) + + for i := 0; i < rand.Intn(1000); i++ { + v := rand.Intn(1 << 20) + if _, err := tx.Contains("x", uint64(v)); err != nil { + return err + } + } + + return nil +} + // better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests. func TestMain(m *testing.M) { port := getAvailPort() diff --git a/rbf/rbf_test.go b/rbf/rbf_test.go index fc3aa3b79..f628bb04c 100644 --- a/rbf/rbf_test.go +++ b/rbf/rbf_test.go @@ -26,6 +26,7 @@ import ( "testing" "github.com/pilosa/pilosa/v2/rbf" + rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg" ) var quickCheckN *int = flag.Int("quickchecks", 10, "The number of iterations for each quickcheck") @@ -60,20 +61,24 @@ func TestReadWriteRootRecord(t *testing.T) { } // NewDB returns a new instance of DB with a temporary path. -func NewDB() *rbf.DB { +func NewDB(cfg ...*rbfcfg.Config) *rbf.DB { path, err := ioutil.TempDir("", "") if err != nil { panic(err) } - db := rbf.NewDB(path, nil) + var cfg0 *rbfcfg.Config + if len(cfg) > 0 { + cfg0 = cfg[0] + } + db := rbf.NewDB(path, cfg0) return db } // MustOpenDB returns a db opened on a temporary file. On error, fail test. -func MustOpenDB(tb testing.TB) *rbf.DB { +func MustOpenDB(tb testing.TB, cfg ...*rbfcfg.Config) *rbf.DB { tb.Helper() - db := NewDB() + db := NewDB(cfg...) if err := db.Open(); err != nil { tb.Fatal(err) } From 731a1ef25efd3789f1b95eeeb4b1f3f7d373a7d7 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Sun, 1 Nov 2020 10:03:31 -0700 Subject: [PATCH 2/2] Refactor RBF WAL to only only checkpoint-in-full. --- go.sum | 1 - rbf/db.go | 158 ++++++------------------------------------------- rbf/db_test.go | 45 +++++++------- 3 files changed, 39 insertions(+), 165 deletions(-) diff --git a/go.sum b/go.sum index e18eaa9eb..ea9c803a4 100644 --- a/go.sum +++ b/go.sum @@ -161,7 +161,6 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181 github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU= github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I= -github.com/pilosa/pilosa v1.4.1 h1:zSNyS/MqXTfRDNRBBNCApsOu2oUnTEj2uymCaqYSNx8= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= diff --git a/rbf/db.go b/rbf/db.go index c65c77bcd..d37d0ff46 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -146,7 +146,7 @@ func (db *DB) Open() (err error) { // Open write-ahead log & checkpoint to the end since no transactions are open. if err := db.openWALSegments(); err != nil { return fmt.Errorf("wal open: %w", err) - } else if err := db.checkpoint(true, &nopLocker{}); err != nil { + } else if err := db.checkpoint(true); err != nil { return fmt.Errorf("checkpoint: %w", err) } @@ -204,32 +204,13 @@ func (db *DB) updateWALSegment(s WALSegment) { db.segments = segments } -// Checkpoint copies pages from WAL segments into the main DB file. This can -// only copy pages that aren't in use by an active transaction. The page map -// is rebuilt as well for all WAL pages still in use. -// -// If exclusive is true, all WAL writes are flushed to disk. -func (db *DB) Checkpoint() error { - return db.checkpoint(false, &db.mu) -} - // checkpoint moves WAL segments to the main DB file. -// -// Note that mu should db.mu when called through DB.Checkpoint() but it -// can be &nopLocker if called under lock. The external API will be used -// to periodically checkpoint outside of a transaction and the locking -// must be used only in the beginning (to obtain the segment list) and at -// the end (when removing old segments from the list). If the entire function -// were to obtain a lock then it would block all new read & write transactions. -func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { - // Obtain a snapshot of WAL segments at the start. - mu.Lock() - opened := db.opened - segments := db.segments - mu.Unlock() - - if !opened { +// Must be called by a write transaction while under db.mu lock. +func (db *DB) checkpoint(exclusive bool) error { + if !db.opened { return nil + } else if len(db.txs) > 0 { + return nil // skip if transactions open } // Determine last checkpointed WAL ID. @@ -239,15 +220,11 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { } walID := readMetaWALID(page) - // Determine the high water mark for WAL pages that can be copied. - writerWALID := db.writerWALID() - // Loop over each transaction walID++ - pageMap := immutable.NewMap(&uint32Hasher{}) for { // Determine last page of transaction. - metaWALID, err := findNextWALMetaPage(segments, walID) + metaWALID, err := findNextWALMetaPage(db.segments, walID) if err == io.EOF { break } else if err != nil { @@ -256,9 +233,7 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { // Loop over pages in the transaction. for ; walID <= metaWALID; walID++ { - canCheckpoint := exclusive || writerWALID == 0 || walID < writerWALID - - page, err := readWALPage(segments, walID) + page, err := readWALPage(db.segments, walID) if err != nil { return err } @@ -274,16 +249,10 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { pgno = readPageNo(page) } - // If we can no longer checkpoint, map the page number to the WAL page. - if !canCheckpoint { - pageMap = pageMap.Set(pgno, walID) - continue - } - // Ensure we actually read the bitmap data in when we checkpoint. // NOTE: The walID variable is incremented above in the pgno check. if isBitmapHeader { - if page, err = readWALPage(segments, walID); err != nil { + if page, err = readWALPage(db.segments, walID); err != nil { return err } } @@ -299,97 +268,21 @@ func (db *DB) checkpoint(exclusive bool, mu sync.Locker) error { if err := db.fsync(db.file); err != nil { return fmt.Errorf("db file sync: %w", err) } - mu.Lock() - db.pageMap = pageMap - mu.Unlock() + db.pageMap = immutable.NewMap(&uint32Hasher{}) // Remove WAL segments that have been checkpointed. - minPageMapWALID := db.minPageMapWALID() - for _, segment := range segments { - if minPageMapWALID != 0 && segment.MaxWALID() > minPageMapWALID { - break - } - - if err := func() error { - mu.Lock() - defer mu.Unlock() - return db.removeWALSegment(segment.Path) - }(); err != nil { + for _, segment := range db.segments { + if err := segment.Close(); err != nil { + return err + } else if err := os.Remove(segment.Path); err != nil { return err } } - - // Ensure all segments are flushed and there is no remapped pages. - if exclusive { - mu.Lock() - assert(len(db.segments) == 0) - assert(pageMap.Len() == 0) - mu.Unlock() - } + db.segments = nil return nil } -// removeWALSegment closes and deletes the segment with the given path. -// -// The DB's segment list is entirely replaced so that transactions with -// a reference to the old list can continue to use it without a lock. -func (db *DB) removeWALSegment(path string) error { - newSegments := make([]WALSegment, 0, len(db.segments)) - for _, segment := range db.segments { - // Close and remove if path matches. - if segment.Path == path { - if err := segment.Close(); err != nil { - return err - } else if err := os.Remove(segment.Path); err != nil { - return err - } - continue - } - - // Otherwise append to new slice of segments. - newSegments = append(newSegments, segment) - } - - // Replace entire slice of segments. - db.segments = newSegments - return nil -} - -// minPageMapWALID returns the lowest WAL ID referenced by an active page map. -func (db *DB) minPageMapWALID() int64 { - // Use the db's page map because that is the state of the map when the - // writer transaction started. We can't use the writer transaction's map - // because it can change. - min := pageMapMinWALID(db.pageMap) - - for tx := range db.txs { - // If a write transaction is active, ensure the min is at least the starting WAL. - if tx.writable { - if min == 0 || tx.walID < min { - min = tx.walID - } - continue - } - - // Record the min WAL ID referenced by the reader's page map. - if walID := pageMapMinWALID(tx.pageMap); min == 0 || walID < min { - min = walID - } - } - return min -} - -// writerWALID returns the starting WAL ID of the active writer tx. -func (db *DB) writerWALID() int64 { - for tx := range db.txs { - if tx.writable { - return tx.walID - } - } - return 0 -} - // Close closes the database. func (db *DB) Close() (err error) { // TODO(bbj): Add wait group to hang until last Tx is complete. @@ -632,7 +525,7 @@ func (db *DB) begin(writable, exclusive bool) (_ *Tx, err error) { // Flush all WAL writes to disk before an exclusive writer so that we can // work directly with the on-disk database. if exclusive { - if err := db.checkpoint(true, &nopLocker{}); err != nil { + if err := db.checkpoint(true); err != nil { cleanup() db.mu.Unlock() return nil, err @@ -709,7 +602,7 @@ func (db *DB) removeTx(tx *Tx) error { // checkpoint eagerly. if tx.writable { if db.cfg.CheckpointEveryDur == 0 || time.Since(db.lastCheckpoint) > db.cfg.CheckpointEveryDur { - if err := db.checkpoint(false, &nopLocker{}); err != nil { + if err := db.checkpoint(false); err != nil { return fmt.Errorf("checkpoint: %w", err) } db.lastCheckpoint = time.Now() @@ -750,20 +643,3 @@ func (db *DB) readMetaPage() ([]byte, error) { } return db.readDBPage(0) } - -type nopLocker struct{} - -func (*nopLocker) Lock() {} -func (*nopLocker) Unlock() {} - -// pageMapMinWALID returns the lowest WAL ID -func pageMapMinWALID(m *immutable.Map) int64 { - var min int64 - for itr := m.Iterator(); !itr.Done(); { - _, v := itr.Next() - if walID := v.(int64); min == 0 || walID < min { - min = walID - } - } - return min -} diff --git a/rbf/db_test.go b/rbf/db_test.go index a70ad1d6a..00425c13b 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -307,14 +307,32 @@ func TestDB_MultiTx(t *testing.T) { // Run multiple readers in separate goroutines. ctx, cancel := context.WithCancel(context.Background()) g, ctx := errgroup.WithContext(ctx) - for i := 0; i < 10; i++ { + for i := 0; i < 4; i++ { g.Go(func() error { for { if ctx.Err() != nil { return nil // cancelled, return no error - } else if err := testDB_MultiTx_reader(db); err != nil { + } else if err := func() error { + tx, err := db.Begin(false) + if err != nil { + return err + } + defer tx.Rollback() + + time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) + + for i := 0; i < rand.Intn(1000); i++ { + v := rand.Intn(1 << 20) + if _, err := tx.Contains("x", uint64(v)); err != nil { + return err + } + } + return nil + }(); err != nil { return err } + + time.Sleep(time.Duration(rand.Intn(int(100 * time.Millisecond)))) } }) } @@ -328,11 +346,12 @@ func TestDB_MultiTx(t *testing.T) { } defer tx.Rollback() - for j := 0; j < rand.Intn(10); j++ { + for j := 0; j < rand.Intn(100); j++ { v := rand.Intn(1 << 20) if _, err := tx.Add("x", uint64(v)); err != nil { t.Fatal(err) } + } if err := tx.Commit(); err != nil { @@ -348,26 +367,6 @@ func TestDB_MultiTx(t *testing.T) { } } -// testDB_MultiTx_reader checks if a bitmap contains a random set of bits. -func testDB_MultiTx_reader(db *rbf.DB) error { - tx, err := db.Begin(false) - if err != nil { - return err - } - defer tx.Rollback() - - time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) - - for i := 0; i < rand.Intn(1000); i++ { - v := rand.Intn(1 << 20) - if _, err := tx.Contains("x", uint64(v)); err != nil { - return err - } - } - - return nil -} - // better diagnosis of deadlocks/hung situations versus just really slow "Quick" tests. func TestMain(m *testing.M) { port := getAvailPort()