Merge pull request #1039 from jaten-molecula/rbf_logging

pilosa server --rbf-checkpoint-dur to 0 by default

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")
 }

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 {

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)