fellow_busy: Introduce dsk dowry

src/fellow_cache.c

@@ -736,6 +736,7 @@ struct fellow_busy {
 	struct fellow_cache_seg	*body_seg;
 	struct fellow_body_region	body_region;
 	struct fellow_bbrr		bbrr;
+	struct buddy_off_extent	segdskdowry;
 
 	struct buddy_ptr_page		segdowry;
 	struct fbo_segmem		segmem[1];
@@ -746,7 +747,7 @@ struct fellow_busy {
 	struct buddy_off_extent	region[FCO_MAX_REGIONS];
 	unsigned			nregion;
 	uint8_t				grown;
-#define FBO_NIO 94
+#define FBO_NIO 91
 	/* protected by fco mtx */
 	uint8_t				io_idx;
 	uint8_t				io_outstanding;
@@ -1977,7 +1978,7 @@ fellow_cache_obj_new(
 	reqs = BUDDY_REQS_STK(fc->membuddy, 4);
 	BUDDY_REQS_PRI(reqs, pri);
 	if (dowry != NULL)
-		AN(buddy_req_page(reqs, fc->tune->chunk_exponent + 1, 0));
+		AN(buddy_req_page(reqs, fc->tune->chunk_exponent, 0));
 	if (fbo_mem != NULL)
 		AN(buddy_req_extent(reqs, sizeof(struct fellow_busy), 0));
 	AN(buddy_req_page(reqs, log2up(mem_sz), 0));
@@ -2066,11 +2067,13 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 	struct fellow_cache_res fcr;
 	struct fellow_busy *fbo;
 	struct buddy *dskbuddy;
+	struct buddy_reqs *reqs;
 	struct buddy_ptr_extent fbo_mem;
 	struct buddy_ptr_page fbo_dowry;
-	struct buddy_off_extent fds_seg;
+	struct buddy_off_extent fds_seg, dskdowry;
 	size_t sz, asz, dsk_sz;
 	uint16_t ldsegs;
+	unsigned u;
 
 	// XXX peek for known content-length=
 
@@ -2099,13 +2102,42 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 	// uring has signed 32bit length
 	assert(dsk_sz <= FIO_MAX);
 
+	// XXX delay to allow new objects during FP_INIT ?
 	dskbuddy = fellow_dskbuddy(fc->ffd);
 
-	// XXX delay to allow new objects during FP_INIT ?
+	/*
+	 * cram == 0:
+	 * the dskdowry acts as an admission ticket to creating an
+	 * expensive busy object: We need to motivate dsk LRU to
+	 * start makeing room if it needs to before we allocate
+	 * memory which LRU would need
+	 */
+	reqs = BUDDY_REQS_STK(dskbuddy, 2);
+	BUDDY_REQS_PRI(reqs, FEP_NEW);
+
+	sz = (size_t)1 << (fc->tune->chunk_exponent);
+	AN(buddy_req_extent(reqs, sz, 0));
+	AN(buddy_req_extent(reqs, dsk_sz, 0));
+
+	u = buddy_alloc_wait(reqs);
+	if (u == 0) {
+		buddy_alloc_wait_done(reqs);
+		fcr = FCR_ALLOCFAIL("dsk dowry and fds");
+		fellow_cache_res_check(fc, fcr);
+		return (fcr);
+	}
+	assert(u == 2);
+
 	fds_seg = FC_INJ ? buddy_off_extent_nil :
-	    buddy_alloc1_off_extent_wait(dskbuddy,
-	    FEP_NEW, dsk_sz, 0);
+	    buddy_get_off_extent(reqs, 1);
+
+	dskdowry = FC_INJ ? buddy_off_extent_nil :
+	    buddy_get_off_extent(reqs, 0);
+	buddy_alloc_wait_done(reqs);
+
 	if (fds_seg.off < 0) {
+		if (dskdowry.off >= 0)
+			buddy_return1_off_extent(dskbuddy, &dskdowry);
 		fcr = FCR_ALLOCFAIL("fds region");
 		fellow_cache_res_check(fc, fcr);
 		return (fcr);
@@ -2116,6 +2148,8 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 	if (fcr.status != fcr_ok) {
 		fellow_cache_res_check(fc, fcr);
 		buddy_return1_off_extent(dskbuddy, &fds_seg);
+		if (dskdowry.off >= 0)
+			buddy_return1_off_extent(dskbuddy, &dskdowry);
 		return (fcr);
 	}
 	fco = fcr.r.ptr;
@@ -2130,6 +2164,7 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 	fbo->fbo_mem = fbo_mem;
 	fbo->segdowry = fbo_dowry;
 	BUDDY_REQS_INIT(&fbo->bbrr, dskbuddy);
+	fbo->segdskdowry = dskdowry;
 
 	// disk object in memory
 	fdo = fcs->alloc.ptr;
@@ -4833,6 +4868,10 @@ fellow_busy_done(struct fellow_busy *fbo, struct objcore *oc, unsigned inlog)
 
 	if (fbo->segdowry.ptr)
 		buddy_return1_ptr_page(fc->membuddy, &fbo->segdowry);
+	if (fbo->segdskdowry.off >= 0) {
+		buddy_return1_off_extent(fellow_dskbuddy(fc->ffd),
+		    &fbo->segdskdowry);
+	}
 	BUDDY_POOL_FINI(fbo->segmem);
 	buddy_alloc_wait_done(&fbo->bbrr.reqs);