fellow_cache_obj: Add dowry for fco reads as well

src/fellow_cache.c

@@ -626,6 +626,7 @@ struct fellow_cache_obj {
 	uint64_t			ntouched;
 
 	struct buddy_ptr_page		fco_mem;	// nil == embedded
+	struct buddy_ptr_page		fco_dowry;
 
 	pthread_mutex_t			mtx;
 	pthread_cond_t			cond;	// signal ready
@@ -1567,6 +1568,8 @@ fellow_cache_obj_free(const struct fellow_cache *fc,
 
 	fellow_cache_seg_free(fc, fcs, 0);
 
+	if (fco->fco_dowry.bits)
+		buddy_return1_ptr_page(fc->membuddy, &fco->fco_dowry);
 	TAKE(mem, fco->fco_mem);
 	if (mem.bits) {
 		assert(mem.ptr == fco);
@@ -3817,6 +3820,7 @@ fellow_cache_obj_readahead(
 	struct fellow_cache_seg *fcs;
 	struct fcscursor c;
 	unsigned ran, ckp, u, n, needdisk, ion;
+	struct buddy_ptr_page dowry;
 
 	/* jump target for need == 1 and sync allocation */
   again:
@@ -3934,6 +3938,7 @@ fellow_cache_obj_readahead(
 	AN(FCO_REFCNT(fco));
 	FCO_REFCNT(fco) += ion;
 	fellow_cache_lru_chgbatch_apply(lcb);
+	TAKE(dowry, fco->fco_dowry);
 	AZ(pthread_mutex_unlock(&fco->mtx));
 
 	if (ckp < ran)
@@ -3952,6 +3957,8 @@ fellow_cache_obj_readahead(
 	*ranp = ran;
 
 	uint8_t rdy = buddy_alloc_async(&reqs[NREQS]);
+	if (dowry.bits)
+		buddy_return1_ptr_page(fc->membuddy, &dowry);
 	buddy_alloc_async_done(&reqs[OREQS]);
 	buddy_return(rets);
 
@@ -4905,19 +4912,22 @@ fellow_cache_obj_prepread(struct fellow_cache *fc, fellow_disk_block fdba,
 	struct fellow_cache_res fcr;
 	struct fellow_cache_seg *fcs;
 	struct fellow_cache_obj *fco;
+	struct buddy_ptr_page dowry = buddy_ptr_page_nil;
 	size_t sz;
 
 	sz = fdb_size(fdba);
 	if (! PAOK(sz))
 		return (FCR_IOFAIL("bad size"));
 
-	fcr = fellow_cache_obj_new(fc, sz, 0, NULL, NULL,
+	fcr = fellow_cache_obj_new(fc, sz, 0, NULL,
+	    crit ? NULL : &dowry,
 	    crit ? FEP_ITER : FEP_GET);
 	if (fcr.status != fcr_ok)
 		return (fcr);
 
 	CAST_OBJ_NOTNULL(fco, fcr.r.ptr, FELLOW_CACHE_OBJ_MAGIC);
 	fco->fdb = fdba;
+	fco->fco_dowry = dowry;
 
 	assert(fco->logstate == FCOL_DUNNO);
 	fco->logstate = FCOL_INLOG;
@@ -5032,6 +5042,7 @@ struct objcore **ocp, uintptr_t priv2, unsigned crit)
 	struct fellow_cache_seg *fcs;
 	struct fellow_disk_obj *fdo;
 	struct fellow_disk_seglist *fdsl;
+	struct buddy_ptr_page dowry = buddy_ptr_page_nil;
 	fellow_disk_block fdba;
 	unsigned oref = 0;
 	const char *err;
@@ -5171,8 +5182,12 @@ struct objcore **ocp, uintptr_t priv2, unsigned crit)
 	if (fcs->refcnt > 1)
 		AZ(pthread_cond_broadcast(&fco->cond));
 	assert_cache_seg_consistency(fcs);
+	if (fco->seglist.fdsl && fco->seglist.fdsl->nsegs == 0)
+		TAKE(dowry, fco->fco_dowry);
 	AZ(pthread_mutex_unlock(&fco->mtx));
 
+	if (dowry.bits)
+		buddy_return1_ptr_page(fc->membuddy, &dowry);
 	return (FCR_OK(fco));
   err:
 	if (ocp) {