Regionlist overhaul: Allocate regionlists asynchronously

regionlists are updated during DLE submit under the logmtx. Thus, we
should avoid synchronous memory allocations.

We change the strategy as follows:

* Memory for the top regionlist (which has one regl embedded) _is_
  allocated synchronously, but with maximum cram to reduce latencies
  at the expense of memory efficiency.

  The case where the allocation does block will not hit us for the
  most critical path in fellow_log_dle_submit(), because we
  pre-allocate there outside the logmtx.

* When we create the top regionlist, we make two asynchronous memory
  allocation requests for our hard-coded size (16KB for prod), one
  crammed and one not. The crammed request is made such that we get
  _any_ memory rather than waiting.

* When we need to extend the regionlist, we should already have an
  allocation available (if not, we need to wait, bad luck). The next
  allocation available is either [1] (uncrammed) left over after the
  previous extension, or [0], which is potentially crammed. If it is
  and we have an uncrammed [1], then we use that and return the
  crammed allocation. If there are no allocations left, we issue the
  next asynchronous request.

src/fellow_regionlist.h

@@ -51,12 +51,16 @@ struct regl {
 	struct buddy_off_extent	arr[];
 };
 
+// [0] crammed
+// [1] not crammed
+BUDDY_REQS(rl_rsv, 2);
+
 struct regionlist {
 	unsigned			magic;
 #define REGIONLIST_MAGIC		0xeb869815
 	size_t				size;
 	VSTAILQ_HEAD(,regl)		head;
-	buddy_t				*membuddy;
+	struct rl_rsv			rsv;
 };
 
 // usage: struct regl_stk(42)
@@ -103,16 +107,14 @@ do {									\
 	regionlist_onlystk_add(name, o, s);				\
 } while(0)
 
-static uint8_t regl_minbits;
-static void __attribute__((constructor))
-init_regl(void)
-{
-	unsigned b =
-	    log2up(2 * sizeof(struct regl) + sizeof(struct regionlist));
+#ifdef DEBUG
+static const int8_t regl_bits = 12;
+#else
+static const int8_t regl_bits = 16;	// XXX 64k reasonable?
+#endif
 
-	assert(b <= UINT8_MAX);
-	regl_minbits = (typeof(regl_minbits))b;
-}
+static uint8_t regl_minbits;
+static int8_t regl_bits_cram;
 
 static int8_t
 regl_cram(const uint8_t bits, int8_t cram)
@@ -131,77 +133,76 @@ regl_cram(const uint8_t bits, int8_t cram)
 	return (cram);
 }
 
+static void __attribute__((constructor))
+init_regl(void)
+{
+	unsigned b =
+	    log2up(2 * sizeof(struct regl) + sizeof(struct regionlist));
+
+	assert(b <= UINT8_MAX);
+	regl_minbits = (typeof(regl_minbits))b;
+	regl_bits_cram = regl_cram(regl_bits, regl_bits);
+}
+
+static void
+regl_rsv(struct buddy_reqs *reqs)
+{
+	BUDDY_REQS_PRI(reqs, FEP_MEM_FREE);
+	AN(buddy_req_page(reqs, regl_bits, regl_bits_cram));
+	AN(buddy_req_page(reqs, regl_bits, 0));
+	(void) buddy_alloc_async(reqs);
+}
+
 static struct regl *
-regl_init(buddy_t *membuddy, struct buddy_ptr_page alloc,
-    struct regionlist **rlp)
+regl_init(const struct buddy_ptr_page alloc, size_t off)
 {
-	struct regionlist *rl;
 	struct regl *r;
 	size_t sz, b;
 
 	sz = (size_t)1 << alloc.bits;
-	assert(sz >= (sizeof *r + sizeof *rl));
+	sz -= off;
+	assert(sz >= 2 * sizeof *r);
 
-	memset(alloc.ptr, 0, sz);
-
-	if (rlp) {
-		rl = alloc.ptr;
-		rl->magic = REGIONLIST_MAGIC;
-		VSTAILQ_INIT(&rl->head);
-		rl->membuddy = membuddy;
-		sz -= sizeof(*rl);
-		*rlp = rl;
-
-		r = (void *)(rl + 1);
-	}
-	else {
-		r = alloc.ptr;
-	}
+	r = (void *)((char *)alloc.ptr + off);
+	memset(r, 0, sz);
 
-	sz -= sizeof(*r);
-
-	r->magic = REGL_MAGIC;
+	sz -= sizeof *r;
 	b = sz / sizeof *r->arr;
 	assert(b <= UINT16_MAX);
+
+	r->magic = REGL_MAGIC;
 	r->space = (typeof(r->space))b;
 	r->alloc = alloc;
+	AZ(r->n);
+	AZ(r->arr[0].off);
+	AZ(r->arr[r->space - 1].off);
 	return (r);
 }
 
-static struct regl *
-regl_alloc_bits(buddy_t *membuddy, const uint8_t bits, int8_t cram,
-    struct regionlist **rlp)
+static struct regionlist *
+regionlist_alloc(buddy_t *membuddy)
 {
 	struct buddy_ptr_page alloc;
+	struct regionlist *rl;
+	struct regl *r;
+	size_t sz;
 
-	alloc = buddy_alloc1_ptr_page_wait(membuddy, FEP_MEM_FREE, bits,
-	    regl_cram(bits, cram));
+	alloc = buddy_alloc1_ptr_page_wait(membuddy, FEP_MEM_FREE,
+	    regl_bits, regl_bits_cram);
 	AN(alloc.ptr);
 
-	return (regl_init(membuddy, alloc, rlp));
-}
+	sz = (size_t)1 << alloc.bits;
+	assert(sz >= (2 * sizeof *r + sizeof *rl));
 
-#ifdef DEBUG
-static const int8_t regl_bits = 12;
-#else
-static const int8_t regl_bits = 16;	// XXX 64k reasonable?
-#endif
+	rl = alloc.ptr;
+	INIT_OBJ(rl, REGIONLIST_MAGIC);
+	VSTAILQ_INIT(&rl->head);
 
-static struct regl *
-regl_alloc(const struct regionlist *rl)
-{
-	return (regl_alloc_bits(rl->membuddy, regl_bits, regl_bits, NULL));
-}
+	BUDDY_REQS_INIT(&rl->rsv, membuddy);
+	regl_rsv(&rl->rsv.reqs);
 
-static struct regionlist *
-regionlist_alloc(buddy_t *membuddy)
-{
-	struct regionlist *rl;
-	struct regl *r;
-
-	r = regl_alloc_bits(membuddy, regl_bits, regl_bits, &rl);
+	r = regl_init(alloc, sizeof(*rl));
 	AN(r);
-	AN(rl);
 
 	VSTAILQ_INSERT_HEAD(&rl->head, r, list);
 	return (rl);
@@ -226,6 +227,7 @@ regionlist_append(struct regionlist *to, struct regionlist **fromp)
 
 	CHECK_OBJ_NOTNULL(to, REGIONLIST_MAGIC);
 	TAKE_OBJ_NOTNULL(from, fromp, REGIONLIST_MAGIC);
+	buddy_alloc_wait_done(&from->rsv.reqs);
 	to->size += from->size;
 	VSTAILQ_CONCAT(&to->head, &from->head);
 }
@@ -233,12 +235,37 @@ regionlist_append(struct regionlist *to, struct regionlist **fromp)
 static struct regl *
 regionlist_extend(struct regionlist *rl)
 {
+	struct buddy_ptr_page alloc;
+	struct buddy_reqs *reqs;
 	struct regl *regl;
+	unsigned n;
 
 	CHECK_OBJ_NOTNULL(rl, REGIONLIST_MAGIC);
+	reqs = &rl->rsv.reqs;
+
+	n = buddy_alloc_async_ready(reqs);
+	if (n == 0)
+		n = buddy_alloc_async_wait(reqs);
+	AN(n);
+	alloc = buddy_get_next_ptr_page(reqs);
+	AN(alloc.ptr);
+	if (buddy_reqs_next_ready(reqs) &&
+	    alloc.bits < regl_minbits) {
+		// the first request was crammed and we have an uncrammed
+		buddy_return1_ptr_page(reqs->buddy, &alloc);
+		alloc = buddy_get_next_ptr_page(reqs);
+		AN(alloc.ptr);
+	}
 
-	regl = regl_alloc(rl);
+	regl = regl_init(alloc, (size_t)0);
 	VSTAILQ_INSERT_TAIL(&rl->head, regl, list);
+
+	// issue the next async req
+	if (! buddy_reqs_next_ready(reqs)) {
+		buddy_alloc_wait_done(reqs);
+		regl_rsv(reqs);
+	}
+
 	return (regl);
 }
 
@@ -255,6 +282,7 @@ regionlist_add(struct regionlist *rl,
 	CHECK_OBJ_NOTNULL(regl, REGL_MAGIC);
 
 	while (n) {
+		AN(regl->space);
 		if (regl->n == regl->space)
 			regl = regionlist_extend(rl);
 
@@ -283,6 +311,7 @@ static void
 regionlist_free(struct regionlist **rlp, buddy_t *dskbuddy)
 {
 	struct buddy_returns *dskret, *memret;
+	struct buddy *membuddy;
 	struct regionlist *rl;
 	struct regl *regl, *save;
 	size_t size, tot;
@@ -290,8 +319,12 @@ regionlist_free(struct regionlist **rlp, buddy_t *dskbuddy)
 
 	TAKE_OBJ_NOTNULL(rl, rlp, REGIONLIST_MAGIC);
 
+	membuddy = rl->rsv.reqs.buddy;
+	AN(membuddy);
+	buddy_alloc_wait_done(&rl->rsv.reqs);
+
 	dskret = BUDDY_RETURNS_STK(dskbuddy, BUDDY_RETURNS_MAX);
-	memret = BUDDY_RETURNS_STK(rl->membuddy, BUDDY_RETURNS_MAX);
+	memret = BUDDY_RETURNS_STK(membuddy, BUDDY_RETURNS_MAX);
 	tot = rl->size;
 
 	VSTAILQ_FOREACH_SAFE(regl, &rl->head, list, save) {