scoutfs: warn on stale cached segments

We've yet to really wire up the eventual consistency of btree ring
blocks and segments.  The btree block reading code has had a warning
that fires if it sees stale blocks for a long time (which we've yet to
hit) but we have no such warning in the segment.  If we hit stale
segments we could have very unpredictable results.  So let's add a quick
warning to highlight the case to save us heartache if we hit it before
implementing full retrying.

Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/compact.c

@@ -153,7 +153,7 @@ static int read_segment(struct super_block *sb, struct compact_seg *cseg)
 	} else {
 		cseg->seg = seg;
 		scoutfs_inc_counter(sb, compact_segment_read);
-		ret = scoutfs_seg_wait(sb, cseg->seg);
+		ret = scoutfs_seg_wait(sb, cseg->seg, cseg->segno, cseg->seq);
 	}
 
 	/* XXX verify read segment metadata */

kmod/src/manifest.c

@@ -633,7 +633,7 @@ static int read_items(struct super_block *sb, struct scoutfs_key_buf *key,
 		if (!ref->seg)
 			break;
 
-		err = scoutfs_seg_wait(sb, ref->seg);
+		err = scoutfs_seg_wait(sb, ref->seg, ref->segno, ref->seq);
 		if (err && !ret)
 			ret = err;
 	}

kmod/src/seg.c

@@ -55,6 +55,14 @@ enum {
 	SF_END_IO = 0,
 };
 
+static void *off_ptr(struct scoutfs_segment *seg, u32 off)
+{
+	unsigned int pg = off >> PAGE_SHIFT;
+	unsigned int pg_off = off & ~PAGE_MASK;
+
+	return page_address(seg->pages[pg]) + pg_off;
+}
+
 static struct scoutfs_segment *alloc_seg(struct super_block *sb, u64 segno)
 {
 	struct scoutfs_segment *seg;
@@ -349,28 +357,51 @@ int scoutfs_seg_submit_write(struct super_block *sb,
 	return 0;
 }
 
-int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg)
+/*
+ * Wait for IO on the segment to complete.  In the cached read fast path
+ * the bit is already set by the reads that populated the cache.
+ *
+ * The caller provides the segno and seq from their segment reference to
+ * validate that we found the version of the segment that they were
+ * looking for.  If we find an old cached version we return -ESTALE and
+ * the caller has to retry its reference to find the current segment for
+ * its operation.  (Typically by getting a new manifest btree root and
+ * searching for keys in the manifest.)
+ *
+ * XXX drop stale segments from the cache
+ * XXX none of the callers perform that retry today.
+ */
+int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg,
+		     u64 segno, u64 seq)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct segment_cache *cac = sbi->segment_cache;
+	struct scoutfs_segment_block *sblk = off_ptr(seg, 0);
 	int ret;
 
 	ret = wait_event_interruptible(cac->waitq,
 				       test_bit(SF_END_IO, &seg->flags));
-	if (!ret)
+	if (ret)
+		goto out;
+
+	if (seg->err) {
 		ret = seg->err;
+		goto out;
+	}
 
+	sblk = off_ptr(seg, 0);
+
+	if (WARN_ON_ONCE(segno != le64_to_cpu(sblk->segno)) ||
+	    WARN_ON_ONCE(seq != le64_to_cpu(sblk->seq))) {
+		    ret = -ESTALE;
+		    goto out;
+	}
+
+	ret = 0;
+out:
 	return ret;
 }
 
-static void *off_ptr(struct scoutfs_segment *seg, u32 off)
-{
-	unsigned int pg = off >> PAGE_SHIFT;
-	unsigned int pg_off = off & ~PAGE_MASK;
-
-	return page_address(seg->pages[pg]) + pg_off;
-}
-
 static void kvec_from_pages(struct scoutfs_segment *seg,
 			    struct kvec *kvec, u32 off, u16 len)
 {

kmod/src/seg.h

@@ -20,7 +20,8 @@ struct scoutfs_segment {
 
 struct scoutfs_segment *scoutfs_seg_submit_read(struct super_block *sb,
 						u64 segno);
-int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg);
+int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg,
+		     u64 segno, u64 seq);
 
 int scoutfs_seg_find_off(struct scoutfs_segment *seg,
 			 struct scoutfs_key_buf *key);