Optimize FCO LRU eviction

src/fellow_cache.c

@@ -305,6 +305,13 @@ struct fellow_cache_seg {
 	unsigned			fco_infdb:1;
 	unsigned			lcb_add:1;
 	unsigned			lcb_remove:1;
+	/*
+	 * during LRU (fco mutate path): the FCO is already
+	 * removed from the LRU, but fcs_onlru is still 1
+	 * for fcs consistency. If mutation fails, the FCO
+	 * is put back on LRU for real
+	 */
+	unsigned			fco_lru_mutate:1;
 
 	/*
 	 * for FCO, protected by fdb_mtx
@@ -1433,7 +1440,7 @@ fellow_cache_obj_free(const struct fellow_cache *fc,
  *
  * state must be FCO_INCORE, object is on LRU, must be referenced
  *
- * lru and fco mtx are held
+ * fco mtx is held
  *
  * similar in structure to fellow_cache_obj_deref(), but
  * - keep reference
@@ -1441,7 +1448,7 @@ fellow_cache_obj_free(const struct fellow_cache *fc,
  */
 
 void
-fellow_cache_obj_evict_mutate(struct fellow_cache_lru *lru,
+fellow_cache_obj_evict_mutate(const struct fellow_cache_lru *lru,
     struct fellow_cache_obj *fco)
 {
 	struct fellow_cache_seg *fcs;
@@ -1474,9 +1481,14 @@ fellow_cache_obj_evict_mutate(struct fellow_cache_lru *lru,
 	AN(fco->oc);
 	fco->oc = NULL;
 
-	// lru
-	VTAILQ_REMOVE(&lru->lru_head, fcs, lru_list);
+	/* finish lru transaction which has been started in
+	 * fellow_cache_lru_work(): the fcs is already removed
+	 * from LRU, we just finish the flags
+	 */
+
+	AN(fcs->fco_lru_mutate);
 	AN(fcs->fcs_onlru);
+	fcs->fco_lru_mutate = 0;
 	fcs->fcs_onlru = 0;
 
 	assert_cache_seg_consistency(fcs);
@@ -2998,6 +3010,9 @@ fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 	 *   - fcs != NULL
 	 * - or foreach finishes, then fcs == NULL and the mtx
 	 *   needs to be unlocked
+	 *
+	 * this is safe because the oc holds a ref on the fco, so
+	 * the LRU state of the fco does not change
 	 */
 
 	CHECK_OBJ_NOTNULL(lru, FELLOW_CACHE_LRU_MAGIC);
@@ -3029,15 +3044,37 @@ fellow_cache_lru_work(struct worker *wrk, struct fellow_cache_lru *lru)
 		if (fcs->state == FCO_INCORE) {
 			oc = fco->oc;
 			CHECK_OBJ_NOTNULL(oc, OBJCORE_MAGIC);
+			AZ(fcs->fco_lru_mutate);
+			fcs->fco_lru_mutate = 1;
+
+			VTAILQ_REMOVE(&lru->lru_head, fcs, lru_list);
+			AZ(pthread_mutex_unlock(&lru->lru_mtx));
+
 			r = stvfe_mutate(wrk, lru, oc);
+			if (r) {
+				// success
+				AZ(fcs->fco_lru_mutate);
+			}
+			else {
+				// will be put back on LRU below
+				AN(fcs->fcs_onlru);
+				AN(fcs->fco_lru_mutate);
+				fcs->fco_lru_mutate = 0;
+			}
 			AZ(pthread_mutex_unlock(&fco->mtx));
 			if (r) {
 				/* mutate was successful
 				 * VSC_C_main->n_lru_nuked++; // XXX per lru ?
 				 */
-				AZ(pthread_mutex_unlock(&lru->lru_mtx));
 				break;
 			}
+			AZ(pthread_mutex_lock(&lru->lru_mtx));
+			// mutate has failed
+			AN(fcs->fcs_onlru);
+			AZ(fcs->fco_lru_mutate);
+			VTAILQ_INSERT_TAIL(&lru->lru_head, fcs, lru_list);
+			// re-start because we let go of lru_mtx
+			fcss = VTAILQ_FIRST(&lru->lru_head);
 			oc = NULL;
 			fco = NULL;
 			continue;

src/fellow_cache_storage.h

@@ -21,7 +21,7 @@
  */
 
 struct fellow_cache_lru;
-void fellow_cache_obj_evict_mutate(struct fellow_cache_lru *lru,
+void fellow_cache_obj_evict_mutate(const struct fellow_cache_lru *lru,
     struct fellow_cache_obj *fco);
 void fellow_cache_obj_slim(const struct fellow_cache *fc,
     struct fellow_cache_obj *fco);

src/fellow_storage.c

@@ -1809,7 +1809,6 @@ stvfe_mutate(struct worker *wrk, struct fellow_cache_lru *lru,
 		AN(oc->stobj->priv);
 		AN(oc->stobj->priv2);
 
-		// XXX LRU with change, reduce LRU mtx scope
 		fellow_cache_obj_evict_mutate(lru, oc->stobj->priv);
 
 		oc->stobj->priv = NULL;