fellow_logcache: prune behind current block once we know to not reverse

src/fellow_log.c

@@ -1090,6 +1090,8 @@ struct fellow_logcache {
 	unsigned				n;
 	unsigned				outstanding;
 	uint8_t				id;
+	// we do not keep blocks in the prune direction if != 0
+	int					prune_direction;
 	struct fellow_fd			*ffd;
 	struct fellow_fd_ioctx_lease		fdil;
 	const struct buddy_off_extent		*region;
@@ -1100,6 +1102,14 @@ struct fellow_logcache {
 	struct flehead				used;
 };
 
+static void
+fellow_logcache_prune_direction(struct fellow_logcache *flc, int direction)
+{
+	AZ(flc->prune_direction);
+	AN(direction);
+	flc->prune_direction = direction;
+}
+
 static void
 fellow_logcache_wait(struct fellow_logcache *flc, unsigned min)
 {
@@ -1523,7 +1533,7 @@ static struct fellow_logcache_r
 fellow_logcache_get(struct fellow_logcache *flc, off_t off, int direction,
     unsigned prefetch)
 {
-	struct fellow_logcache_entry *fle;
+	struct fellow_logcache_entry *fle, *ofle;
 	struct fellow_logcache_r r;
 	unsigned pre;
 
@@ -1568,7 +1578,16 @@ fellow_logcache_get(struct fellow_logcache *flc, off_t off, int direction,
 			//DBG("hit  %zu %s", off, direction < 0 ? "<-" : "->");
 		}
 		assert(fle->off == off);
+
+		ofle = flc->current;
 		flc->current = fle;
+		assert(fle != ofle);
+
+		if (ofle != NULL &&
+		    flc->prune_direction != 0 &&
+		    flc->prune_direction != direction) {
+			fellow_logcache_prune(flc, ofle, flc->prune_direction);
+		}
 	}
 
 	pre = fellow_logcache_prefetch(flc, fle, direction, prefetch);
@@ -3540,6 +3559,7 @@ logbuffer_append(struct fellow_fd *ffd,
   dsk_init:
 	AN(from->id);
 	fellow_logcache_init(flc, ffd, from->id, from->logreg->region);
+	fellow_logcache_prune_direction(flc, -1);
 	AZ(locked);
   dsk:
 	where = NULL;
@@ -5225,8 +5245,10 @@ fellow_logs_iter(const struct flics *flics, struct flivs *flivs,
 		 * always be set, or we are dealing with an old active block
 		 * (head on second flush)
 		 */
-		if (direction >= 0 && logblk->next_off == 0)
+		if (direction >= 0 && logblk->next_off == 0) {
 			direction = -1;
+			fellow_logcache_prune_direction(flc, 1);
+		}
 		else
 			AN(logblk->next_off);