fellow_cache_lru: Solve resume properly with a marker

When we hit an FCO for which we could not lock the varnish object (oc),
we inserted that FCO at the end of the LRU list, and restarted at the
first. But if that FCO was the only one, we would find ourselves in an
infinite loop.

We now put a marker segment on the LRU for the time we need to let go of
the lru lock and continue at that marker.

src/fellow_cache.c

@@ -86,6 +86,7 @@ struct fellow_cache_lru {
 	pthread_mutex_t			lru_mtx;
 	struct fellow_cache_lru_head	lru_head;
 	pthread_t			lru_thread;
+	struct fellow_cache_seg		*resume;
 };
 
 static struct fellow_cache_lru *
@@ -3327,7 +3328,7 @@ fellow_cache_obj_slim(const struct fellow_cache *fc,
 static int
 fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 {
-	struct fellow_cache_seg *fcs, *fcss;
+	struct fellow_cache_seg *fcs, *fcss, *resume;
 	struct fellow_cache_obj *fco;
 	struct objcore *oc;
 	struct buddy_ptr_extent alloc;
@@ -3349,6 +3350,8 @@ fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 	 */
 
 	CHECK_OBJ_NOTNULL(lru, FELLOW_CACHE_LRU_MAGIC);
+	resume = lru->resume;
+	CHECK_OBJ(resume, FELLOW_CACHE_SEG_MAGIC);
 
 	struct buddy_returns *rets =
 	    BUDDY_RETURNS_STK(lru->fc->membuddy, BUDDY_RETURNS_MAX);
@@ -3357,6 +3360,11 @@ fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 	oc = NULL;
 	fco = NULL;
 	AZ(pthread_mutex_lock(&lru->lru_mtx));
+	if (resume->fcs_onlru) {
+		VTAILQ_REMOVE(&lru->lru_head, resume, lru_list);
+		resume->fcs_onlru = 0;
+	}
+
 	//lint -e{850} loop variable modified in body
 	VTAILQ_FOREACH_SAFE(fcs, &lru->lru_head, lru_list, fcss) {
 		assert(fcs != fcss);
@@ -3404,6 +3412,11 @@ fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 
 			AN(lru->n);
 			lru->n--;
+
+			AZ(resume->fcs_onlru);
+			VTAILQ_INSERT_AFTER(&lru->lru_head,
+			    fcs, resume, lru_list);
+			resume->fcs_onlru = 1;
 			VTAILQ_REMOVE(&lru->lru_head, fcs, lru_list);
 			AZ(pthread_mutex_unlock(&lru->lru_mtx));
 
@@ -3427,13 +3440,16 @@ fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 			AN(fcs->fco_lru_mutate);
 			fcs->fco_lru_mutate = 0;
 
-			VTAILQ_INSERT_TAIL(&lru->lru_head, fcs, lru_list);
+			AN(resume->fcs_onlru);
+			VTAILQ_INSERT_BEFORE(resume, fcs, lru_list);
 			lru->n++;
 
 			AZ(pthread_mutex_unlock(&fco->mtx));
 
-			// re-start because we let go of lru_mtx
-			fcss = VTAILQ_FIRST(&lru->lru_head);
+			fcss = VTAILQ_NEXT(resume, lru_list);
+			VTAILQ_REMOVE(&lru->lru_head, resume, lru_list);
+			resume->fcs_onlru = 0;
+
 			oc = NULL;
 			continue;
 		}
@@ -3516,6 +3532,7 @@ fellow_cache_lru_thread(struct worker *wrk, void *priv)
 #endif
 	struct buddy_reqs *reqs;
 	const struct fellow_cache *fc;
+	struct fellow_cache_seg resume[1];
 	buddy_t *buddy;
 	struct buddy_ptr_page *r = NULL;
 	unsigned i, filled = 0, nr = 0, rc, cb;
@@ -3538,6 +3555,10 @@ fellow_cache_lru_thread(struct worker *wrk, void *priv)
 	reqs = BUDDY_REQS_STK(buddy, BUDDY_REQS_MAX);
 	BUDDY_REQS_PRI(reqs, FEP_RESERVE);
 
+	INIT_OBJ(resume, FELLOW_CACHE_SEG_MAGIC);
+	AZ(lru->resume);
+	lru->resume = resume;
+
 #ifdef LRU_NOISE
 	AZ(wrk->vsl);
 	wrk->vsl = &vsl;
@@ -3620,6 +3641,14 @@ fellow_cache_lru_thread(struct worker *wrk, void *priv)
 				(void)usleep(10*1000);
 			}
 	}
+	AN(lru->resume);
+	AZ(pthread_mutex_lock(&lru->lru_mtx));
+	if (lru->resume->fcs_onlru) {
+		VTAILQ_REMOVE(&lru->lru_head, lru->resume, lru_list);
+		lru->resume->fcs_onlru = 0;
+	}
+	AZ(pthread_mutex_unlock(&lru->lru_mtx));
+	lru->resume = NULL;
 	reserve_free(buddy, r, filled);
 	free(r);
 	return (NULL);