Multi-LRU infrastructure

we allow up to 1 << MAX_NLRU_EXPONENT (64) LRUs. When objects
are created, they get hashed onto LRUs. LRUs never die but
during shutdown.

Consequently, the number of LRUs can be tuned at run time.

src/fellow_cache.c

@@ -85,6 +85,7 @@ fellow_cache_lru_new(struct fellow_cache *fc)
 	lru->fc = fc;
 	AZ(pthread_mutex_init(&lru->lru_mtx, NULL));
 	VTAILQ_INIT(&lru->lru_head);
+
 	return (lru);
 }
 
@@ -103,7 +104,6 @@ fellow_cache_lru_fini(struct fellow_cache_lru **lrup)
 	AZ(pthread_mutex_destroy(&lru->lru_mtx));
 }
 
-
 /*
  * lifetime of a fellow_obj in cache
  *
@@ -684,6 +684,16 @@ struct fellow_busy {
 	struct fellow_busy_io		io[FBO_NIO];
 };
 
+#define MAX_NLRU_EXPONENT 6
+
+struct fellow_cache_lrus {
+	unsigned			magic;
+#define FELLOW_CACHE_LRUS_MAGIC	0xadad56fb
+	uint8_t				exponent;
+	pthread_mutex_t			mtx;
+	struct fellow_cache_lru		*lru[1 << MAX_NLRU_EXPONENT];
+};
+
 struct fellow_cache {
 	unsigned			magic;
 #define FELLOW_CACHE_MAGIC		0xe2f2243e
@@ -696,7 +706,7 @@ struct fellow_cache {
 	pthread_mutex_t			fdb_mtx;
 	struct fellow_cache_fdb_head	fdb_head;
 
-	struct fellow_cache_lru		*lru;
+	struct fellow_cache_lrus	lrus[1];
 
 	pthread_mutex_t			async_mtx;
 	pthread_cond_t			async_cond;
@@ -705,6 +715,66 @@ struct fellow_cache {
 	unsigned			async_idle;
 };
 
+/* ============================================================
+ * multi-LRU
+ */
+
+static void * fellow_cache_lru_thread(struct worker *wrk, void *priv);
+
+static void
+fellow_cache_lrus_init(struct fellow_cache_lrus *lrus)
+{
+
+	INIT_OBJ(lrus, FELLOW_CACHE_LRUS_MAGIC);
+	AZ(pthread_mutex_init(&lrus->mtx, NULL));
+}
+
+static void
+fellow_cache_lrus_fini(struct fellow_cache_lrus *lrus)
+{
+	unsigned u;
+
+	for (u = 0; u < 1 << MAX_NLRU_EXPONENT; u++) {
+		if (lrus->lru[u] == NULL)
+			continue;
+		fellow_cache_lru_fini(&lrus->lru[u]);
+	}
+	AZ(pthread_mutex_destroy(&lrus->mtx));
+}
+
+static struct fellow_cache_lru *
+fellow_cache_get_lru(struct fellow_cache *fc, uint64_t n)
+{
+	struct fellow_cache_lrus *lrus;
+	struct fellow_cache_lru *lru;
+	uint8_t exponent;
+	pthread_t thr;
+	size_t i;
+
+	CHECK_OBJ_NOTNULL(fc, FELLOW_CACHE_MAGIC);
+	lrus = fc->lrus;
+	CHECK_OBJ_NOTNULL(lrus, FELLOW_CACHE_LRUS_MAGIC);
+
+	exponent = lrus->exponent;
+	assert(exponent <= MAX_NLRU_EXPONENT);
+	i = exponent ? fib(n, exponent) : 0;
+
+	lru = lrus->lru[i];
+	if (lru != NULL && lru->lru_thread != 0)
+		return (lru);
+
+	AZ(pthread_mutex_lock(&lrus->mtx));
+	lru = lrus->lru[i];
+	if (lru == NULL) {
+		lru = lrus->lru[i] = fellow_cache_lru_new(fc);
+		WRK_BgThread(&thr, "sfe-mem-lru", fellow_cache_lru_thread, lru);
+		AN(thr);
+	}
+	AZ(pthread_mutex_unlock(&lrus->mtx));
+	AN(lru);
+	return (lru);
+}
+
 /* ============================================================
  * util
  */
@@ -1766,7 +1836,7 @@ fellow_cache_seglists_load(const struct fellow_cache *fc,
  */
 
 static struct fellow_cache_res
-fellow_cache_obj_new(const struct fellow_cache *fc,
+fellow_cache_obj_new(struct fellow_cache *fc,
     size_t dsk_sz, unsigned nseg_guess, uint8_t pri)
 {
 	struct fellow_disk_obj *fdo;
@@ -1831,7 +1901,7 @@ fellow_cache_obj_new(const struct fellow_cache *fc,
 	INIT_OBJ(fco, FELLOW_CACHE_OBJ_MAGIC);
 	DBG("fco %p", fco);
 	fco->fco_mem = fco_mem;
-	fco->lru = fc->lru;
+	fco->lru = fellow_cache_get_lru(fc, (uintptr_t)fco);
 
 	AZ(pthread_mutex_init(&fco->mtx, NULL));
 	AZ(pthread_cond_init(&fco->cond, NULL));
@@ -3122,22 +3192,22 @@ reserve_fill(struct buddy_ptr_page *r, const struct buddy_reqs *reqs, uint8_t n)
 	}
 }
 
-void *
+static void *
 fellow_cache_lru_thread(struct worker *wrk, void *priv)
 {
 	struct fellow_cache_lru *lru;
 	struct vsl_log vsl;
 	struct buddy_reqs *reqs;
-	struct fellow_cache *fc;
+	const struct fellow_cache *fc;
 	buddy_t *buddy;
 	struct buddy_ptr_page *r = NULL;
 	unsigned i, filled = 0, nr = 0, rc, cb;
 	uint8_t n;
 	size_t sz;
 
-	CAST_OBJ_NOTNULL(fc, priv, FELLOW_CACHE_MAGIC);
-	lru = fc->lru;
-	CHECK_OBJ_NOTNULL(lru, FELLOW_CACHE_LRU_MAGIC);
+	CAST_OBJ_NOTNULL(lru, priv, FELLOW_CACHE_LRU_MAGIC);
+	fc = lru->fc;
+	CHECK_OBJ_NOTNULL(fc, FELLOW_CACHE_MAGIC);
 
 	buddy = fc->membuddy;
 	CHECK_OBJ(buddy, BUDDY_MAGIC);
@@ -3235,7 +3305,7 @@ fellow_cache_lru_thread(struct worker *wrk, void *priv)
 	free(r);
 	return (NULL);
 }
-#endif
+#endif // TEST_DRIVER
 
 /* returns if moved */
 int
@@ -4330,7 +4400,7 @@ fellow_cache_obj_fini(const struct fellow_cache_obj *fco)
 }
 
 static struct fellow_cache_res
-fellow_cache_obj_prepread(const struct fellow_cache *fc, fellow_disk_block fdba,
+fellow_cache_obj_prepread(struct fellow_cache *fc, fellow_disk_block fdba,
     unsigned crit)
 {
 	struct fellow_cache_res fcr;
@@ -4958,7 +5028,7 @@ fellow_cache_init(struct fellow_fd *ffd, buddy_t *membuddy,
 	fc->tune = tune;
 	fc->running = 1;
 
-	fc->lru = fellow_cache_lru_new(fc);
+	fellow_cache_lrus_init(fc->lrus);
 
 	AZ(pthread_mutex_init(&fc->fdb_mtx, NULL));
 	VRBT_INIT(&fc->fdb_head);
@@ -4980,8 +5050,7 @@ fellow_cache_fini(struct fellow_cache **fcp)
 
 	fc->running = 0;
 	buddy_wait_kick(fc->membuddy);
-	fellow_cache_lru_fini(&fc->lru);
-	AZ(fc->lru);
+	fellow_cache_lrus_fini(fc->lrus);
 
 	assert(VRBT_EMPTY(&fc->fdb_head));
 
@@ -4996,6 +5065,21 @@ fellow_cache_fini(struct fellow_cache **fcp)
 #include "vsha256.h"
 #include "fellow_testenv.h"
 
+int
+stvfe_mutate(struct worker *wrk, struct fellow_cache_lru *lru,
+    struct objcore *oc)
+{
+	WRONG("no lru during test");
+	return (1);
+}
+
+static void *
+fellow_cache_lru_thread(struct worker *wrk, void *priv)
+{
+	(void) wrk;
+	(void) priv;
+	return (NULL);
+}
 #ifdef DEBUG
 const char * const filename = "/tmp/fellowfile.cachetest";
 #else
@@ -5374,7 +5458,7 @@ t_lcb(struct fellow_cache *fc)
 	struct fellow_cache_seg fcs[nfcs];
 
 	INIT_OBJ(fco, FELLOW_CACHE_OBJ_MAGIC);
-	fco->lru = fc->lru;
+	fco->lru = fellow_cache_lru_new(fc);
 
 	for (i = 0; i < nfcs; i++) {
 		INIT_OBJ(&fcs[i], FELLOW_CACHE_SEG_MAGIC);

src/fellow_cache.h

@@ -81,6 +81,3 @@ void fellow_busy_obj_trimstore(struct fellow_busy *fbo);
 void fellow_busy_done(struct fellow_busy *fbo, struct objcore *oc, unsigned);
 void *fellow_busy_setattr(struct fellow_busy *fbo,
     enum obj_attr attr, size_t len, const void *ptr);
-
-void *
-fellow_cache_lru_thread(struct worker *wrk, void *priv);

src/fellow_storage.c

@@ -2312,7 +2312,6 @@ sfe_open_scope(struct stevedore *stv)
 	struct sfe_resurrect_priv sfer[1];
 	struct sfe_open_worker_priv sow[1];
 	const char *err = NULL;
-	pthread_t mem_lru_thread;
 
 	ASSERT_CLI();
 	CAST_OBJ_NOTNULL(stvfe, stv->priv, STVFE_MAGIC);
@@ -2398,9 +2397,6 @@ sfe_open_scope(struct stevedore *stv)
 		goto err;
 	}
 
-	WRK_BgThread(&mem_lru_thread, "sfe-mem-lru",
-	    fellow_cache_lru_thread, stvfe->fc);
-	AN(mem_lru_thread);
 	WRK_BgThread(&stvfe->dsk_lru_thread, "sfe-dsk-lru",
 	    sfedsk_lru_thread, stvfe->dskstv);