Introduce new bitfield alloc for logs rewrite

src/fellow_log.c

@@ -46,6 +46,7 @@
 #include "buddy.h"
 #include "foreign/qdef.h"
 #include "vqueue.h"
+#include "bitf_segmentation.h"
 
 #include "foreign/vend.h"
 
@@ -5212,6 +5213,83 @@ fellow_logs_iter(const struct flics *flics, struct flivs *flivs,
 	return (ret);
 }
 
+/*
+ * struct to maintain state on incremental allocation of the bit field for log
+ * rewrite
+ */
+
+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;
+};
+
+static void
+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);
+	for (u = 0; u < bfa->nmem; u++)
+		AN(buddy_return_ptr_extent(rets, &bfa->mem[u]));
+	buddy_return(rets);
+	buddy_alloc_async_done(bfa->reqs);
+	memset(bfa, 0, sizeof *bfa);
+}
+
+static void
+bfa_alloc(struct bitfalloc *bfa)
+{
+	struct buddy_ptr_extent mem;
+	size_t bits;
+
+	CHECK_OBJ(bfa, BITFALLOC_MAGIC);
+
+	if (bfa->need == 0)
+		return;
+
+  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;
+
+		bits = sz_nbits(mem.size);
+		if (bits > bfa->need)
+			bits = bfa->need;
+
+		bitfs_add(bfa->bitfs, bitf_init(mem.ptr, bits, mem.size));
+		buddy_alloc_async_done(bfa->reqs);
+		bfa->need -= bits;
+	}
+	if (bfa->need > 0 && bfa->reqs->n == 0) {
+		BUDDY_REQS_PRI(bfa->reqs, FEP_MEM_REWR);
+		AN(buddy_req_extent(bfa->reqs, bitf_sz(bfa->need),
+			log2down(bfa->lmem - bfa->nmem)));
+		(void) buddy_alloc_async(bfa->reqs);
+		goto bfa_again;
+	}
+}
+
+static void
+bfa_alloc_wait(struct bitfalloc *bfa)
+{
+
+	bfa_alloc(bfa);
+	while (bfa->need > 0) {
+		(void) buddy_alloc_wait(bfa->reqs);
+		bfa_alloc(bfa);
+	}
+}
+
 /*
  * current logbuffer must be LBUF_MEM or LBUF_OPEN
  *
@@ -5284,10 +5362,22 @@ 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;
+	bfa_alloc(bfa);
+
 	AZ(pthread_mutex_lock(&ffd->logmtx));
 	if (fellow_logs_rewriting(ffd)) {
 		AZ(pthread_mutex_unlock(&ffd->logmtx));
 		ffd->diag("... (raced)\n");
+		bfa_free(bfa);
 		return;
 	}
 	ffd->rewriting = 1;
@@ -5338,6 +5428,7 @@ 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);
@@ -5420,11 +5511,13 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 	if (logblk_off == 0 && pendblk_off == 0) {
 		DBG("---- no logs pending %zu log %zu",
 		    pendblk_off, logblk_off);
+		bfa_free(bfa);
 	}
 	else {
 		// the actual rewrite happens outside the logmtx:
 		// ffd->logbuf is usable as LBUF_PEND
 
+		bfa_alloc_wait(bfa);
 		struct bitf *bitf;
 		struct buddy_ptr_extent bitfmem;
 		size_t nblocks;
@@ -5468,6 +5561,7 @@ fellow_logs_rewrite(struct fellow_fd *ffd,
 			    ffd->logreg[empty_region(log_info->region)].region,
 			    logblk_off);
 
+		bfa_free(bfa);
 		buddy_return1_ptr_extent(ffd->membuddy, &bitfmem);
 
 		// does not work for vcl mode and test cases