diff --git a/rbf/cfg/cfg.go b/rbf/cfg/cfg.go index 56fff7305..0e519569a 100644 --- a/rbf/cfg/cfg.go +++ b/rbf/cfg/cfg.go @@ -55,7 +55,16 @@ func NewDefaultConfig() *Config { func (cfg *Config) DefineFlags(flags *pflag.FlagSet) { default0 := NewDefaultConfig() flags.IntVar(&cfg.MaxWALSegmentFileSize, "rbf-max-wal", default0.MaxWALSegmentFileSize, "RBF write-Ahead-Log file size in bytes") - flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur, "RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.") + + // TODO: make delayed checkpointing work. Currently + // wal.go readWALPage() can fail to locate some pages + // when checkpointing does not happen after every Commit. + // Once that is done we can return to trying to checkpoint + // after some duration. + //flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", default0.CheckpointEveryDur, "RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.") + + flags.DurationVar(&cfg.CheckpointEveryDur, "rbf-checkpoint-dur", 0, "RBF checkpoint on the next write that occurs this long or more after the previous write. 0 means checkpoint after every write.") + flags.Int64Var(&cfg.MaxSize, "rbf-max-db-size", default0.MaxSize, "RBF maximum size in bytes of a database file (distinct from a WAL file)") flags.BoolVar(&cfg.FsyncEnabled, "rbf-fsync", default0.FsyncEnabled, "RBF: enable fsync fully safe flush-to-disk at each checkpoint") } diff --git a/rbf/db.go b/rbf/db.go index 9ac28dd83..17fcf77df 100644 --- a/rbf/db.go +++ b/rbf/db.go @@ -683,6 +683,11 @@ func (db *DB) removeTx(tx *Tx) error { // Write pages from WAL to DB. // TODO(bbj): Move this to an async goroutine. + // TODO(jea): Make the time-based checkpointing work at all, and update the + // comment in cfg/cfg.go for CheckpointEveryDur. Seems that + // wal.go readWALPage() can receive a request for a walID that + // comes before the segments it is passed if we do not + // 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 { diff --git a/rbf/wal.go b/rbf/wal.go index 9a8353e49..9304eed0d 100644 --- a/rbf/wal.go +++ b/rbf/wal.go @@ -171,13 +171,32 @@ func readWALPage(segments []WALSegment, walID int64) ([]byte, error) { i := sort.Search(n, func(i int) bool { return walID < segments[i].MinWALID }) + minWALID, maxWALID := int64(-1), int64(-1) if i > 0 { s := segments[i-1] - if walID >= s.MinWALID && walID <= s.MaxWALID() { + minWALID = s.MinWALID + maxWALID = s.MaxWALID() + if walID >= minWALID && walID <= maxWALID { return s.ReadWALPage(walID) } } - return nil, fmt.Errorf("cannot find segment containing WAL page: %d", walID) + // ok, we're about to error, which should never happen. + // So we can afford to provide detailed diagnostics. + + // Report min and max WALID over all supplied segments. + for _, s := range segments { + if minWALID < 0 || s.MinWALID < minWALID { + minWALID = s.MinWALID + } + max := s.MaxWALID() + if maxWALID < 0 || max > maxWALID { + maxWALID = max + } + } + // show all the current segments too. + detail := WALSegmentsAsString(segments) + + return nil, fmt.Errorf("cannot find segment containing WAL page: %d; over all supplied segments, minWALID=%v, maxWALID=%v; detail='%v'", walID, minWALID, maxWALID, detail) } func findNextWALMetaPage(segments []WALSegment, walID int64) (metaWALID int64, err error) { @@ -263,6 +282,14 @@ func DumpWALSegments(segments []WALSegment) { } } +func WALSegmentsAsString(segments []WALSegment) (r string) { + r = fmt.Sprintf("WAL (%d segments)\n", len(segments)) + for i, s := range segments { + r += fmt.Sprintf("[%d] WALIDs=(%d-%d) PageN=%d\n", i, s.MinWALID, s.MaxWALID(), s.PageN) + } + return +} + // FormatWALSegmentPath returns a path for a WAL segment using a WAL ID. func FormatWALSegmentPath(walID int64) string { return fmt.Sprintf("%016x.wal", walID)