fellow_log: rewrite bitfield allocation

The previous implementation used only one BUDDY_REQS, so whenever one
segment allocation was fulfilled, other requests with lower priority
could "get through" and ultimately lead to bfa_alloc() failing to
complete.

By using two BUDDY_REQS, we now make sure to "keep out place in the
priority queue".

We also limit cramming not only by the available bitfield segment slots,
but also by a maximum of 4 (1/16th of the requested size) and yield when
a lower cram does not succeed to motivate LRU more to make room.

This has undergone a _lot_ of testing and has gone through many
iterations, which all have been squashed into this commit.

src/fellow_log.c

@@ -5238,14 +5238,18 @@ fellow_logs_iter(const struct flics *flics, struct flivs *flivs,
  * rewrite
  */
 
+#define BITFMAXSEGS 128
+
 struct bitfalloc {
 	unsigned		magic;
 #define BITFALLOC_MAGIC	0x93a14c7f
 	unsigned		nmem, lmem;
-	size_t			need;	// bits
-	struct buddy_reqs	*reqs;
-	struct bitfs		*bitfs;
-	struct buddy_ptr_extent	*mem;
+	buddy_t			*buddy;
+	union {
+		struct bitfs	bitfs[1];
+		char		bitfsmem[bitfs_size(BITFMAXSEGS)];
+	} u;
+	struct buddy_ptr_extent	mem[BITFMAXSEGS];
 };
 
 static void
@@ -5254,72 +5258,114 @@ bfa_free(struct bitfalloc *bfa)
 	struct buddy_returns *rets;
 	unsigned u;
 
-	CHECK_OBJ(bfa, BITFALLOC_MAGIC);
-	AN(bfa->reqs);
-	rets = BUDDY_RETURNS_STK(bfa->reqs->buddy, BUDDY_RETURNS_MAX);
+	CHECK_OBJ_NOTNULL(bfa, BITFALLOC_MAGIC);
+
+	assert(bfa->nmem <= bfa->lmem);
+
+	rets = BUDDY_RETURNS_STK(bfa->buddy, BUDDY_RETURNS_MAX);
 	for (u = 0; u < bfa->nmem; u++)
 		AN(buddy_return_ptr_extent(rets, &bfa->mem[u]));
+	bfa->nmem = 0;
 	buddy_return(rets);
-	buddy_alloc_async_done(bfa->reqs);
+
 	memset(bfa, 0, sizeof *bfa);
 }
 
 static void
-bfa_alloc(struct bitfalloc *bfa)
+bfa_alloc(struct bitfalloc *bfa, size_t need, buddy_t *membuddy)
 {
+	struct buddy_reqs *reqs, *hold;
 	struct buddy_ptr_extent mem;
-	size_t bits;
+	struct bitfs *b;
+	size_t sz, bits;
 	int8_t cram;
 	unsigned c;
-	size_t sz;
+	uint8_t u;
 
-	CHECK_OBJ(bfa, BITFALLOC_MAGIC);
+	AN(bfa);
+	INIT_OBJ(bfa, BITFALLOC_MAGIC);
+	b = bitfs_init(bfa->u.bitfsmem, sizeof bfa->u.bitfsmem);
+	assert(b == bfa->u.bitfs);
+	bfa->lmem = sizeof bfa->mem / sizeof *bfa->mem;
+	bfa->buddy = membuddy;
 
-	if (bfa->need == 0)
-		return;
+	reqs = BUDDY_REQS_STK(bfa->buddy, 1);
+	hold = BUDDY_REQS_STK(bfa->buddy, 1);
+	BUDDY_REQS_PRI(reqs, FEP_MEM_REWR);
+	BUDDY_REQS_PRI(hold, FEP_MEM_REWR - 1);
 
-  bfa_again:
-	if (buddy_alloc_async_ready(bfa->reqs)) {
-		assert(bfa->nmem < bfa->lmem);
-		mem = buddy_get_ptr_extent(bfa->reqs, 0);
-		AN(mem.ptr);
-		bfa->mem[bfa->nmem++] = mem;
+	sz = bitf_sz(need, BITF_NOIDX);
 
-		bits = sz_nbits(mem.size, BITF_NOIDX);
-		if (bits > bfa->need)
-			bits = bfa->need;
+	AZ(bfa->nmem);
+	AZ(bfa->u.bitfs->nbits);
 
-		bitfs_add(bfa->bitfs,
-			  bitf_init(mem.ptr, bits, mem.size, BITF_NOIDX));
-		buddy_alloc_async_done(bfa->reqs);
-		bfa->need -= bits;
-	}
-	if (bfa->need > 0 && bfa->reqs->n == 0) {
-		sz = bitf_sz(bfa->need, BITF_NOIDX);
+	cram = 0;
 
-		// do not cram to below the minimum bitf size
+	while (need > 0) {
+		// limit by available segments
 		c = log2down((size_t)bfa->lmem - bfa->nmem);
 		assert(c < INT8_MAX);
-		cram = (int8_t)c;
+
+		// 1/16th of the requested size should be small enough to be
+		// fulfillable with a decent amount of LRU. flw
+		if (c > 4)
+			c = 4;
+		if (cram > (int8_t)c)
+			cram = (int8_t)c;
+
+		// limit by minimal size
 		cram = buddy_cramlimit_bits(sz, cram,
 		    (int8_t)log2up(bitf_sz((size_t)1, BITF_NOIDX)));
 
-		BUDDY_REQS_PRI(bfa->reqs, FEP_MEM_REWR);
-		AN(buddy_req_extent(bfa->reqs, sz, cram));
-		(void) buddy_alloc_async(bfa->reqs);
-		goto bfa_again;
-	}
-}
+		buddy_alloc_async_done(reqs);
+		AN(buddy_req_extent(reqs, sz, cram));
+		(void) buddy_alloc_async(reqs);
+
+		// if the hold has memory or we have a new sz (at cram == 0),
+		// return to transfer to reqs
+		if (cram == 0 || buddy_alloc_async_ready(hold))
+			buddy_alloc_async_done(hold);
+		if (hold->n == 0) {
+			AN(buddy_req_extent(hold, sz, 0));
+			(void) buddy_alloc_async(hold);
+		}
 
-static void
-bfa_alloc_wait(struct bitfalloc *bfa)
-{
+		u = buddy_alloc_async_ready(reqs);
 
-	bfa_alloc(bfa);
-	while (bfa->need > 0) {
-		(void) buddy_alloc_async_wait(bfa->reqs);
-		bfa_alloc(bfa);
+		if (u == 0) {
+			(void) sched_yield();
+			u = buddy_alloc_async_ready(reqs);
+		}
+
+		if (u == 0 && cram == (typeof(cram))c)
+			u = buddy_alloc_async_wait(reqs);
+
+		if (u == 0) {
+			cram++;
+			continue;
+		}
+
+		// got an allocation
+		assert(bfa->nmem < bfa->lmem);
+		mem = buddy_get_ptr_extent(reqs, 0);
+		AN(mem.ptr);
+		bfa->mem[bfa->nmem++] = mem;
+
+		bits = sz_nbits(mem.size, BITF_NOIDX);
+		if (bits > need)
+			bits = need;
+
+		bitfs_add(bfa->u.bitfs, bitf_init(mem.ptr, bits,
+		    mem.size, BITF_NOIDX));
+		need -= bits;
+
+		// next allocation might be of smaller size, reset cram
+		sz = bitf_sz(need, BITF_NOIDX);
+		cram = 0;
 	}
+
+	buddy_alloc_async_done(reqs);
+	buddy_alloc_async_done(hold);
 }
 
 /*
@@ -5394,16 +5440,6 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 	for (u = 0; u < LOGREGIONS; u++)
 		CHECK_OBJ(&ffd->logreg[u], FELLOW_LOG_REGION_MAGIC);
 
-#define BITFMAXSEGS 16
-	char bfamem[bitfs_size(BITFMAXSEGS)];
-	struct bitfalloc bfa[1];
-	INIT_OBJ(bfa, BITFALLOC_MAGIC);
-	bfa->need = ffd->size >> MIN_FELLOW_BITS;
-	bfa->reqs = BUDDY_REQS_STK(ffd->membuddy, 1);
-	bfa->bitfs = bitfs_init(bfamem, sizeof bfamem);
-	bfa->mem = (struct buddy_ptr_extent[BITFMAXSEGS]){{0}};
-	bfa->lmem = BITFMAXSEGS;
-
 	//flexelint
 	memset(lbuf_pend, 0, sizeof lbuf_pend);
 
@@ -5417,7 +5453,9 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 		logregion_reset(logreg);
 		logbuffer_init(ffd, lbuf_pend, logreg);
 	}
-	bfa_alloc(bfa);
+
+	struct bitfalloc bfa[1];
+	bfa_alloc(bfa, ffd->size >> MIN_FELLOW_BITS, ffd->membuddy);
 
 	AZ(pthread_mutex_lock(&ffd->logmtx));
 	if (fellow_logs_rewriting(ffd)) {
@@ -5470,7 +5508,6 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 		logbuffer_take(ffd->logbuf, lbuf_pend);
 
 		AZ(pthread_mutex_unlock(&ffd->logmtx));
-		bfa_alloc(bfa);
 
 		// close current buffer
 		assert(lbuf_save->state == LBUF_OPEN);
@@ -5559,8 +5596,6 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 		// the actual rewrite happens outside the logmtx:
 		// ffd->logbuf is usable as LBUF_PEND
 
-		bfa_alloc_wait(bfa);
-
 #define BFA_DIAG
 #ifdef BFA_DIAG_VERBOSE
 		for (u = 0; u < bfa->nmem; u++) {
@@ -5576,7 +5611,7 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 			.magic = FLIVS_MAGIC,
 			.oob = 0,
 			.lbuf = lbuf,
-			.bitfs = bfa->bitfs,
+			.bitfs = bfa->u.bitfs,
 			.tofree = tofree,
 			.ban_dles = {{0}},
 			.fdct = {{0}},