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()