fellow_busy: Give the busy object a dowry and return it after taking the pool

src/fellow_cache.c

@@ -732,6 +732,7 @@ struct fellow_busy {
 	struct fellow_cache_seg	*body_seg;
 	struct fellow_body_region	body_region;
 
+	struct buddy_ptr_page		segdowry;
 	struct fbo_segmem		segmem[1];
 
 #define FCO_MAX_REGIONS ((FELLOW_DISK_LOG_BLOCK_ENTRIES - 1) * DLE_REG_NREGION)
@@ -1936,6 +1937,7 @@ fellow_cache_obj_new(
     struct fellow_cache *fc,
     size_t dsk_sz, unsigned nseg_guess,
     struct buddy_ptr_extent *fbo_mem,
+    struct buddy_ptr_page *dowry,
     uint8_t pri)
 {
 	struct fellow_disk_obj *fdo;
@@ -1964,24 +1966,32 @@ fellow_cache_obj_new(
 	mem_sz = sizeof *fco + nseg_guess * sizeof *fcs;
 	asz = (size_t)1 << log2up(dsk_sz);
 
-	reqs = BUDDY_REQS_STK(fc->membuddy, 3);
+	// dowry is largest, so we allocate it first
+	reqs = BUDDY_REQS_STK(fc->membuddy, 4);
 	BUDDY_REQS_PRI(reqs, pri);
-	AN(buddy_req_page(reqs, log2up(mem_sz), 0));
-	AN(buddy_req_extent(reqs, asz, 0));
+	if (dowry != NULL)
+		AN(buddy_req_page(reqs, fc->tune->chunk_exponent + 1, 0));
 	if (fbo_mem != NULL)
 		AN(buddy_req_extent(reqs, sizeof(struct fellow_busy), 0));
+	AN(buddy_req_page(reqs, log2up(mem_sz), 0));
+	AN(buddy_req_extent(reqs, asz, 0));
 
 	u = buddy_alloc_wait(reqs);
-	if (FC_INJ || u != (fbo_mem == NULL ? 2 : 3)) {
+	if (FC_INJ || u != 2 + (dowry ? 1 : 0) + (fbo_mem ? 1 : 0)) {
 		buddy_alloc_wait_done(reqs);
 		fcr = FCR_ALLOCFAIL("fellow_cache_obj_new alloc failed");
 		goto err;
 	}
 
-	fco_mem = buddy_get_ptr_page(reqs, 0);
-	fdo_mem = buddy_get_ptr_extent(reqs, 1);
+	if (dowry != NULL) {
+		assert(u >= 3);
+		*dowry = buddy_get_next_ptr_page(reqs);
+	}
 	if (fbo_mem != NULL)
-		*fbo_mem = buddy_get_ptr_extent(reqs, 2);
+		*fbo_mem = buddy_get_next_ptr_extent(reqs);
+
+	fco_mem = buddy_get_next_ptr_page(reqs);
+	fdo_mem = buddy_get_next_ptr_extent(reqs);
 
 	buddy_alloc_wait_done(reqs);
 
@@ -2049,6 +2059,7 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 	struct fellow_cache_res fcr;
 	struct fellow_busy *fbo;
 	struct buddy_ptr_extent fbo_mem;
+	struct buddy_ptr_page fbo_dowry;
 	struct buddy_off_extent fds_seg;
 	size_t sz, asz, dsk_sz;
 	uint16_t ldsegs;
@@ -2090,7 +2101,8 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 		return (fcr);
 	}
 
-	fcr = fellow_cache_obj_new(fc, dsk_sz, ldsegs, &fbo_mem, FEP_NEW);
+	fcr = fellow_cache_obj_new(fc, dsk_sz, ldsegs, &fbo_mem, &fbo_dowry,
+	    FEP_NEW);
 	if (fcr.status != fcr_ok) {
 		fellow_cache_res_check(fc, fcr);
 		buddy_return1_off_extent(fellow_dskbuddy(fc->ffd), &fds_seg);
@@ -2106,6 +2118,7 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 	AN(fbo);
 	INIT_OBJ(fbo, FELLOW_BUSY_MAGIC);
 	fbo->fbo_mem = fbo_mem;
+	fbo->segdowry = fbo_dowry;
 
 	// disk object in memory
 	fdo = fcs->alloc.ptr;
@@ -2164,6 +2177,7 @@ fellow_busy_obj_alloc(struct fellow_cache *fc,
 
 	BUDDY_POOL_INIT(fbo->segmem, fc->membuddy, FEP_SPC,
 	    fellow_busy_fill_segmem, fc->tune);
+	buddy_return1_ptr_page(fc->membuddy, &fbo->segdowry);
 
 	return (FCR_OK(fbo));
 }
@@ -4764,6 +4778,8 @@ fellow_busy_done(struct fellow_busy *fbo, struct objcore *oc, unsigned inlog)
 	fc = fbo->fc;
 	CHECK_OBJ_NOTNULL(fc, FELLOW_CACHE_MAGIC);
 
+	if (fbo->segdowry.ptr)
+		buddy_return1_ptr_page(fc->membuddy, &fbo->segdowry);
 	BUDDY_POOL_FINI(fbo->segmem);
 
 	// nicer way?
@@ -4895,7 +4911,8 @@ fellow_cache_obj_prepread(struct fellow_cache *fc, fellow_disk_block fdba,
 	if (! PAOK(sz))
 		return (FCR_IOFAIL("bad size"));
 
-	fcr = fellow_cache_obj_new(fc, sz, 0, NULL, crit ? FEP_ITER : FEP_GET);
+	fcr = fellow_cache_obj_new(fc, sz, 0, NULL, NULL,
+	    crit ? FEP_ITER : FEP_GET);
 	if (fcr.status != fcr_ok)
 		return (fcr);