fellow_cache: Support the first cache seglist (fcsl) as dynamic

This allows us to shrink the fellow_cache_obj allocation for the
fellow_obj_get (from disk) case from 8KB to 512 bytes.

The root cause for the massively oversized fco allocation was that
the nseg_guess heuristic could not take into account the wsl (size of
the actual object), so it had to assume that all of the disk object's
size was taken up for segments in the disk object embedded seglist.

src/fellow_cache.c

@@ -650,11 +650,12 @@ struct fellow_cache_obj {
 	struct fellow_cache_seg		aa_##l##_seg;
 #include "tbl/fellow_obj_attr.h"
 
-	struct fellow_cache_seglist	seglist;
+	struct fellow_cache_seglist	*fcsl;
+	struct fellow_cache_seglist	fcsl_embed;
 };
 
 #define fellow_cache_obj_size(fco)		\
-	(sizeof *fco + (fco)->seglist.lsegs * sizeof *(fco)->seglist.segs)
+	(sizeof *fco + (fco)->fcsl_embed.lsegs * sizeof *(fco)->fcsl_embed.segs)
 
 #define FCR_OK(p)\
 	(struct fellow_cache_res){.r.ptr = (p), .status = fcr_ok}
@@ -1440,8 +1441,6 @@ fellow_cache_seglist_free(struct buddy_returns *memret,
 	struct buddy_ptr_extent e;
 	unsigned u, segs;
 
-	AN(fcsl);
-
 	while (fcsl != NULL) {
 		CHECK_OBJ(fcsl, FELLOW_CACHE_SEGLIST_MAGIC);
 
@@ -1531,7 +1530,7 @@ fellow_seglist_regions(const struct fellow_fd *ffd,
 
 static unsigned
 fellow_obj_regions(const struct fellow_fd *ffd,
-    struct fellow_cache_obj *fco,
+    const struct fellow_cache_obj *fco,
     struct buddy_off_extent region[FCO_MAX_REGIONS])
 {
 	struct fellow_cache_seglist *fcsl;
@@ -1540,7 +1539,7 @@ fellow_obj_regions(const struct fellow_fd *ffd,
 
 	CHECK_OBJ_NOTNULL(fco, FELLOW_CACHE_OBJ_MAGIC);
 
-	n = fellow_seglist_regions(ffd, &fco->seglist, region, 0);
+	n = fellow_seglist_regions(ffd, fco->fcsl, region, 0);
 	assert(n <= FCO_MAX_REGIONS);
 	DBG("seglist_regions %u", n);
 #define FDO_AUXATTR(U, l)						\
@@ -1550,7 +1549,7 @@ fellow_obj_regions(const struct fellow_fd *ffd,
 	DBG("+auxattr %u", n);
 	assert(n <= FCO_MAX_REGIONS);
 
-	fcsl = &fco->seglist;
+	fcsl = fco->fcsl;
 	fdsl = fcsl->fdsl;
 
 	while ((fcsl = fcsl->next) != NULL) {
@@ -1594,7 +1593,7 @@ fellow_cache_obj_free(const struct fellow_cache *fc,
 	fellow_cache_lru_chgbatch_apply(lcb);
 
 	DBG("fco %p", fco);
-	fellow_cache_seglist_free(memret, &fco->seglist);
+	fellow_cache_seglist_free(memret, fco->fcsl);
 
 #define FDO_AUXATTR(U, l)				\
 	fellow_cache_seg_auxattr_free(memret, &fco->aa_##l##_seg);
@@ -1738,12 +1737,12 @@ fellow_cache_seglist_associate(
 {
 	struct fellow_cache_seg *fcs;
 	struct fellow_disk_seg *fds;
-	unsigned u;
+	unsigned max, u;
 
 	CHECK_OBJ_NOTNULL(fcsl, FELLOW_CACHE_SEGLIST_MAGIC);
 	CHECK_OBJ_NOTNULL(fdsl, FELLOW_DISK_SEGLIST_MAGIC);
 
-	assert(fcsl->lsegs >= fdsl->lsegs);
+	assert(fcsl->lsegs >= fdsl->nsegs);
 	AZ(fcsl->fdsl);
 	fcsl->fdsl = fdsl;
 
@@ -1753,7 +1752,10 @@ fellow_cache_seglist_associate(
 
 		fellow_cache_seg_associate(fcs, fds, state);
 	}
-	for (; u < fdsl->lsegs; u++) {
+	max = fdsl->lsegs;
+	if (fcsl->lsegs < max)
+		max = fcsl->lsegs;
+	for (; u < max; u++) {
 		fcs = &fcsl->segs[u];
 		fds = &fdsl->segs[u];
 
@@ -1995,16 +1997,8 @@ fellow_cache_obj_new(
 
 	DBG("arg dsk_sz %zu nseg_guess %u", dsk_sz, nseg_guess);
 
-	if (nseg_guess == 0) {
-		asz = sizeof(struct fellow_disk_obj) +
-		    sizeof(struct fellow_disk_seglist);
-		assert(dsk_sz > asz);
-		ssz = (dsk_sz - asz) / sizeof(struct fellow_disk_seg);
-		assert(ssz < UINT_MAX);
-		nseg_guess = (unsigned)ssz;
-		if (nseg_guess == 0)
-			nseg_guess = 1;
-	}
+	if (nseg_guess == 0)
+		nseg_guess = 1;
 
 	mem_sz = (size_t)1 << log2up(sizeof *fco + nseg_guess * sizeof *fcs);
 	asz = (size_t)1 << log2up(dsk_sz);
@@ -2054,7 +2048,7 @@ fellow_cache_obj_new(
 
 	ssz = fco_mem.size - sizeof *fco;
 
-	(void) fellow_cache_seglist_init(&fco->seglist, ssz, fco);
+	(void) fellow_cache_seglist_init(&fco->fcsl_embed, ssz, fco);
 
 #define FDO_AUXATTR(U, l)						\
 	fellow_cache_seg_init(&fco->aa_##l##_seg, fco, FCS_INIT);
@@ -2235,12 +2229,13 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 
 	fdsl = fellow_disk_seglist_init(fellow_disk_obj_fdsl(fdo),
 	    ldsegs, fc->tune->hash_obj);
-	fellow_cache_seglist_associate(&fco->seglist, fdsl, FCS_USABLE);
-//	DBG("fdsl lsegs %u fcsl lsegs %u", fdsl->lsegs, fco->seglist.lsegs);
+	fellow_cache_seglist_associate(&fco->fcsl_embed, fdsl, FCS_USABLE);
+	fco->fcsl = &fco->fcsl_embed;
+//	DBG("fdsl lsegs %u fcsl lsegs %u", fdsl->lsegs, fco->fcsl->lsegs);
 
 	fbo->fc = fc;
 	fbo->fco = fco;
-	fbo->body_seglist = &fco->seglist;
+	fbo->body_seglist = fco->fcsl;
 	AZ(fbo->body_seg);
 
 	AN(fbo->body_seglist);
@@ -2839,19 +2834,24 @@ fellow_cache_obj_trim(const struct fellow_cache *fc,
 {
 	struct fellow_cache_seglist *fcsl;
 	struct fellow_disk_seglist *fdsl;
+	uint16_t n;
 
 	CHECK_OBJ_NOTNULL(fc, FELLOW_CACHE_MAGIC);
 	CHECK_OBJ_NOTNULL(fco, FELLOW_CACHE_OBJ_MAGIC);
-	fcsl = &fco->seglist;
+	fcsl = &fco->fcsl_embed;
 	CHECK_OBJ(fcsl, FELLOW_CACHE_SEGLIST_MAGIC);
 	fdsl = fcsl->fdsl;
-	CHECK_OBJ_NOTNULL(fdsl, FELLOW_DISK_SEGLIST_MAGIC);
+	CHECK_OBJ_ORNULL(fdsl, FELLOW_DISK_SEGLIST_MAGIC);
+	if (fdsl == NULL)
+		n = 0;
+	else
+		n = fdsl->nsegs;
 
 	AZ(fcsl->fcsl_sz);
 	AZ(fcsl->fdsl_sz);
 
-	assert(fcsl->lsegs >= fdsl->nsegs);
-	fcsl->lsegs = fdsl->nsegs;
+	assert(fcsl->lsegs >= n);
+	fcsl->lsegs = n;
 	buddy_trim1_ptr_extent(fc->membuddy, &fco->fco_mem,
 	    fellow_cache_obj_size(fco));
 }
@@ -3328,7 +3328,7 @@ fellow_cache_obj_slim(const struct fellow_cache *fc,
 	struct fcscursor c;
 	unsigned ref;
 
-	fcsc_init(&c, &fco->seglist);
+	fcsc_init(&c, fco->fcsl);
 
 	lru = fco->lru;
 	CHECK_OBJ_NOTNULL(lru, FELLOW_CACHE_LRU_MAGIC);
@@ -3343,7 +3343,7 @@ fellow_cache_obj_slim(const struct fellow_cache *fc,
 	if (fcs == NULL || fcs->alloc.ptr == NULL)
 		goto out;
 
-	fcsc_init(&c, &fco->seglist);
+	fcsc_init(&c, fco->fcsl);
 	AZ(pthread_mutex_lock(&lru->lru_mtx));
 	while ((fcs = FCSC_NEXT(&c)) != NULL) {
 		if (fcs->alloc.ptr == NULL)
@@ -4222,7 +4222,7 @@ fellow_cache_obj_iter(struct fellow_cache *fc, struct fellow_cache_obj *fco,
 	fcoid->priv = priv;
 	fcoid->memret = BUDDY_RETURNS_STK(fc->membuddy, 1);
 
-	fcsc_init(&c, &fco->seglist);
+	fcsc_init(&c, fco->fcsl);
 	rac = c;
 
 	flags = final ? OBJ_ITER_FLUSH : 0;
@@ -4726,13 +4726,13 @@ fellow_busy_obj_trim_seglists(struct fellow_busy *fbo)
 	// only to be called if anything to trim at all
 	AZ(fbo->body_seglist->fdsl->nsegs);
 
-	fcsl = &fco->seglist;
-	CHECK_OBJ(fcsl, FELLOW_CACHE_SEGLIST_MAGIC);
-
 	// not for the embedded seglist
-	if (fbo->body_seglist == fcsl)
+	if (fbo->body_seglist == &fco->fcsl_embed)
 		return;
 
+	fcsl = fco->fcsl;
+	CHECK_OBJ_NOTNULL(fcsl, FELLOW_CACHE_SEGLIST_MAGIC);
+
 	// find the previous seglist
 	while (fcsl->next != fbo->body_seglist) {
 		fcsl = fcsl->next;
@@ -4966,6 +4966,7 @@ fellow_busy_log_submit(const struct fellow_busy *fbo)
 void
 fellow_busy_done(struct fellow_busy *fbo, struct objcore *oc, unsigned inlog)
 {
+	struct fellow_cache_seglist *fcsl;
 	struct fellow_cache_obj *fco;
 	struct fellow_disk_obj *fdo;
 	struct fellow_cache *fc;
@@ -4991,13 +4992,17 @@ fellow_busy_done(struct fellow_busy *fbo, struct objcore *oc, unsigned inlog)
 	if (fbo->body_seg != NULL)
 		fellow_busy_obj_trimstore(fbo);
 
-	if (fco->seglist.next) {
-		AN(fco->seglist.fdsl);
+	// the first disk seglist is always embedded
+	fcsl = fco->fcsl;
+	assert(fcsl == &fco->fcsl_embed);
+	AN(fcsl);
+	if (fcsl->next) {
+		AN(fcsl->fdsl);
 		fellow_cache_seglists_write(fbo,
-		    fco->seglist.next, fco->seglist.fdsl->next);
+		    fcsl->next, fcsl->fdsl->next);
 	}
 
-	fellow_disk_seglist_fini(fco->seglist.fdsl);
+	fellow_disk_seglist_fini(fcsl->fdsl);
 
 	AZ(fdo->fdo_flags);
 	AZ(fco->oc);
@@ -5233,6 +5238,7 @@ struct objcore **ocp, uintptr_t priv2, unsigned crit)
 	struct fellow_cache_seg *fcs;
 	struct fellow_disk_obj *fdo;
 	struct fellow_disk_seglist *fdsl;
+	struct buddy_ptr_extent fcsl_mem = buddy_ptr_extent_nil;
 	struct buddy_ptr_page dowry = buddy_ptr_page_nil;
 	fellow_disk_block fdba;
 	unsigned oref = 0;
@@ -5343,10 +5349,30 @@ struct objcore **ocp, uintptr_t priv2, unsigned crit)
 
 	fdsl = fellow_disk_obj_fdsl(fdo);
 	assert(PAOK(fdsl));
+	CHECK_OBJ_NOTNULL(fdsl, FELLOW_DISK_SEGLIST_MAGIC);
+
+	// the embedded fcsl may or may not fit
+	if (fco->fcsl_embed.lsegs >= fdsl->nsegs)
+		fco->fcsl = &fco->fcsl_embed;
+	else {
+		// dup fellow_cache_seglists_load()
+		fcsl_mem = buddy_alloc1_ptr_extent_wait(fc->membuddy, FEP_META,
+		    fellow_cache_seglist_size(fco->fcsl, fdsl->nsegs), 0);
+		if (FC_INJ || fcsl_mem.ptr == NULL) {
+			err = FC_ERRSTR("first disk seglist fcsl alloc failed");
+			goto err;
+		}
+		fco->fcsl = fellow_cache_seglist_init(fcsl_mem.ptr,
+		    fcsl_mem.size - sizeof *fco->fcsl, fco);
+		AN(fco->fcsl);
+		fco->fcsl->fcsl_sz = fcsl_mem.size;
+		fcsl_mem = buddy_ptr_extent_nil;
+	}
+
 	// XXX load of folow-up seglists could be async
 	// should not be a common case though, we try to make the first
 	// seglist fit
-	err = fellow_cache_seglists_load(fc, fco, &fco->seglist, fdsl);
+	err = fellow_cache_seglists_load(fc, fco, fco->fcsl, fdsl);
 	if (err != NULL)
 		goto err;
 
@@ -5375,7 +5401,7 @@ struct objcore **ocp, uintptr_t priv2, unsigned crit)
 	if (fcs->refcnt > 1)
 		AZ(pthread_cond_broadcast(&fco->cond));
 	assert_cache_seg_consistency(fcs);
-	if (fco->seglist.fdsl && fco->seglist.fdsl->nsegs == 0)
+	if (fco->fcsl->fdsl && fco->fcsl->fdsl->next.size == 0)
 		TAKE(dowry, fco->fco_dowry);
 	AZ(pthread_mutex_unlock(&fco->mtx));
 
@@ -5396,6 +5422,8 @@ struct objcore **ocp, uintptr_t priv2, unsigned crit)
 	fellow_cache_seg_transition_locked_notincore(FCO_FCS(fco), FCO_READFAIL);
 	fellow_cache_obj_deref(fc, fco);
 
+	if (fcsl_mem.ptr)
+		buddy_return1_ptr_extent(fc->membuddy, &fcsl_mem);
 	if (strstr(err, "alloc"))
 		fcr = FCR_ALLOCERR(err);
 	else
@@ -6077,7 +6105,7 @@ static void test_fellow_cache_obj_iter_final(
 	VSHA256_Final(h2, &sha256ctx);
 	AZ(memcmp(h1, h2, sizeof *h1));
 
-	fcsc_init(&c, &(*fcop)->seglist);
+	fcsc_init(&c, (*fcop)->fcsl);
 	if ((fcs = FCSC_NEXT(&c)) != NULL) {
 		while (fcs->state == FCS_READING || fcs->state == FCS_WRITING)
 			usleep(100);
@@ -6404,7 +6432,7 @@ t_busyobj(unsigned chksum, struct fellow_cache *fc)
 		test_fellow_cache_obj_iter_thread_f, iter_thr));
 	DBG("concurrent test_iter thread %p", (void *)iter_thr->thr);
 
-	u = fco->seglist.fdsl->lsegs + 72 * 2;
+	u = fco->fcsl->fdsl->lsegs + 72 * 2;
 	XXX_LIMIT_LDSEGS = 1;
 	while (u-- > 0) {
 		void *p;
@@ -6430,7 +6458,7 @@ t_busyobj(unsigned chksum, struct fellow_cache *fc)
 	tiw->done = 1;
 	test_iter_extend(tiw, 0);
 	// if this fails, u is too low
-	AN(fco->seglist.next);
+	AN(fco->fcsl->next);
 	// fixattr always return a pointer
 	for (u = OA_VARY; u < OA__MAX; u++)
 		AZ(fellow_cache_obj_getattr(fc, fco, u, &sz));
@@ -6492,7 +6520,7 @@ t_cache(const char *fn, unsigned chksum)
 
 	// canary so size increase does not happen unnoticed
 	sz  = sizeof(struct fellow_cache_obj);
-	assert(sz <= 352);
+	assert(sz <= 360);
 
 	AZ(stvfe_tune_init(tune, memsz, dsksz, objsize_hint));
 	tune->hash_obj = chksum;
@@ -6722,7 +6750,7 @@ t_cache(const char *fn, unsigned chksum)
 	VSHA256_Init(&sha256ctx);
 	fbo = fellow_busy_obj_alloc(fc, &fco, &priv2, 1234).r.ptr;
 	CHECK_OBJ_NOTNULL(fbo, FELLOW_BUSY_MAGIC);
-	u = fco->seglist.fdsl->lsegs + 72 * 2;
+	u = fco->fcsl->fdsl->lsegs + 72 * 2;
 	while (u-- > 0) {
 		void *p;
 		sz = 1234;
@@ -6739,7 +6767,7 @@ t_cache(const char *fn, unsigned chksum)
 	VSHA256_Final(h1, &sha256ctx);
 	fellow_busy_obj_trimstore(fbo);
 	// if this fails, u is too low
-	AN(fco->seglist.next);
+	AN(fco->fcsl->next);
 	// fixattr always return a pointer
 	for (u = OA_VARY; u < OA__MAX; u++)
 		AZ(fellow_cache_obj_getattr(fc, fco, u, &sz));