diff --git a/rbf/db.go b/rbf/db.go index 21596d144..cfa5ede43 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -278,8 +278,6 @@ func (db *DB) checkpoint() (err error) { }() if !db.opened { return nil - } else if len(db.txs) > 0 { - return nil // skip if transactions open } // Check if there are any WAL pages, if not do nothing as @@ -364,6 +362,8 @@ func (db *DB) checkpoint() (err error) { // the rwmu and update the metadata about the WAL. releaseLock = false // fmt.Printf("checkpoint mostly done, waiting for Tx cleanup...\n") + db.walPageN = 0 + db.pageMap = NewPageMap() db.afterCurrentTx(func() { // fmt.Printf("truncating WAL\n") defer db.rwmu.Unlock() @@ -374,8 +374,6 @@ func (db *DB) checkpoint() (err error) { } else if _, err = db.walFile.Seek(0, io.SeekStart); err != nil { db.logger.Errorf("seek wal file: %w", err) } - db.walPageN = 0 - db.pageMap = NewPageMap() // fmt.Printf("checkpoint actually done\n") }) return nil @@ -589,19 +587,22 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) { if db.isDead != nil { err := db.isDead cleanup() - db.mu.Unlock() return nil, err } - // Wait for WAL size to be below threshold. - for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize { - if db.isDead != nil { - err := db.isDead - cleanup() - db.mu.Unlock() - return nil, err + // Wait for WAL size to be below threshold, if we're going to write. + // Reads don't care. + if writable { + for int64(db.walPageN*PageSize) > db.cfg.MaxWALCheckpointSize { + if db.isDead != nil { + err := db.isDead + cleanup() + return nil, err + } + // This implicitly releases db.mu.Lock and comes back with it + // held again. + db.haltCond.Wait() } - db.haltCond.Wait() } tx := &Tx{ @@ -775,6 +776,9 @@ func (db *DB) baseWALID() int64 { // readWALPageByID reads a WAL page by WAL ID. func (db *DB) readWALPageByID(id int64) ([]byte, error) { + if id == db.baseWALID() { + fmt.Printf("id %d oops\n", id) + } return db.readWALPageAt(int(id - db.baseWALID() - 1)) } diff --git a/rbf/db_test.go b/rbf/db_test.go index 5c8504b1b..c0eb3b3c7 100644 --- a/rbf/db_test.go +++ b/rbf/db_test.go @@ -291,7 +291,8 @@ func TestDB_MultiTx(t *testing.T) { time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) - for i := 0; i < rand.Intn(1000); i++ { + n := rand.Intn(500) + 500 + for i := 0; i < n; i++ { v := rand.Intn(1 << 20) if _, err := tx.Contains("x", uint64(v)); err != nil { return err @@ -316,7 +317,8 @@ func TestDB_MultiTx(t *testing.T) { } defer tx.Rollback() - for j := 0; j < rand.Intn(100); j++ { + n := rand.Intn(90) + 10 + for j := 0; j < n; j++ { v := rand.Intn(1 << 20) if _, err := tx.Add("x", uint64(v)); err != nil { t.Fatal(err) diff --git a/rbf/tx.go b/rbf/tx.go index 2cf7420db..bfb8fc00c 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -102,6 +102,8 @@ func (tx *Tx) Commit() error { // If any pages have been written, ensure we write a new meta page with // the commit flag to mark the end of the transaction. + tx.db.mu.Lock() + defer tx.db.mu.Unlock() if tx.dirty() { if err := tx.flush(); err != nil { return err @@ -118,12 +120,9 @@ func (tx *Tx) Commit() error { // the lock, because we need removeTx to grab the lock to // work, but if it wants to checkpoint, it wants to be able to return // to us here and still be holding the lock. - tx.db.mu.Lock() tx.db.rootRecords = tx.rootRecords tx.db.pageMap = tx.pageMap tx.db.walPageN = tx.walPageN - tx.db.mu.Unlock() - return tx.db.removeTx(tx) } // Disconnect transaction from DB.