Prioritize object memory allocation for OA_VARY

... which happens potentially under the cache lock

src/fellow_cache.c

@@ -4054,7 +4054,8 @@ fellow_cache_obj_fini(const struct fellow_cache_obj *fco)
 }
 
 static struct fellow_cache_res
-fellow_cache_obj_prepread(const struct fellow_cache *fc, fellow_disk_block fdba)
+fellow_cache_obj_prepread(const struct fellow_cache *fc, fellow_disk_block fdba,
+    unsigned crit)
 {
 	struct fellow_cache_res fcr;
 	struct fellow_cache_seg *fcs;
@@ -4065,7 +4066,7 @@ fellow_cache_obj_prepread(const struct fellow_cache *fc, fellow_disk_block fdba)
 	if (! PAOK(sz))
 		return (FCR_IOFAIL("bad size"));
 
-	fcr = fellow_cache_obj_new(fc, sz, 0, FEP_GET);
+	fcr = fellow_cache_obj_new(fc, sz, 0, crit ? FEP_ITER : FEP_GET);
 	if (fcr.status != fcr_ok)
 		return (fcr);
 
@@ -4176,7 +4177,7 @@ fellow_cache_obj_deref(struct fellow_cache *fc, struct fellow_cache_obj *fco)
  */
 struct fellow_cache_res
 fellow_cache_obj_get(struct fellow_cache *fc,
-    struct objcore **ocp, uintptr_t priv2)
+struct objcore **ocp, uintptr_t priv2, unsigned crit)
 {
 	struct fellow_cache_res fcr;
 	struct fellow_cache_obj *fco, *nfco;
@@ -4198,7 +4199,7 @@ fellow_cache_obj_get(struct fellow_cache *fc,
 	 * we wasted some work and free the extra object again
 	 *
 	 */
-	fcr =  fellow_cache_obj_prepread(fc, fdba);
+	fcr =  fellow_cache_obj_prepread(fc, fdba, crit);
 	if (fcr.status != fcr_ok) {
 		fellow_cache_res_check(fc, fcr);
 		return (fcr);
@@ -4257,7 +4258,7 @@ fellow_cache_obj_get(struct fellow_cache *fc,
 		case FCO_EVICT:
 			// race, retry
 			fellow_cache_obj_deref(fc, fco);
-			return (fellow_cache_obj_get(fc, ocp, priv2));
+			return (fellow_cache_obj_get(fc, ocp, priv2, 0));
 		default:
 			WRONG("fco state");
 		}
@@ -4904,7 +4905,7 @@ test_fellow_cache_obj_get(struct fellow_cache *fc, uintptr_t priv2,
 	// baseline, no injection
 	fc_inj_reset();
 	toc = ocmem;
-	fcr = fellow_cache_obj_get(fc, &toc, priv2);
+	fcr = fellow_cache_obj_get(fc, &toc, priv2, 0);
 	if (fcr.status != fcr_ok) {
 		AN(toc);
 		return (NULL);
@@ -4928,7 +4929,7 @@ test_fellow_cache_obj_get(struct fellow_cache *fc, uintptr_t priv2,
 	while (injcount) {
 		fc_inj_set(injcount);
 		toc = ocmem;
-		fcr = fellow_cache_obj_get(fc, &toc, priv2);
+		fcr = fellow_cache_obj_get(fc, &toc, priv2, 0);
 		AN(toc);
 		assert(fcr.status != fcr_ok);
 		DBG("inj %d err %s", injcount, fcr.r.err);
@@ -4939,7 +4940,7 @@ test_fellow_cache_obj_get(struct fellow_cache *fc, uintptr_t priv2,
 
 	fc_inj_reset();
 	toc = ocmem;
-	fcr = fellow_cache_obj_get(fc, &toc, priv2);
+	fcr = fellow_cache_obj_get(fc, &toc, priv2, 0);
 	DBG("fco2 %p ref %u", fco, FCO_REFCNT(fco));
 	assert(toc == oc1);
 	CAST_OBJ_NOTNULL(fco, fcr.r.ptr, FELLOW_CACHE_OBJ_MAGIC);
@@ -4988,7 +4989,7 @@ static void test_fellow_cache_obj_iter_final(
 	AN(injcount);
 	while (injcount) {
 		toc = ocmem;
-		fcr = fellow_cache_obj_get(fc, &toc, priv2);
+		fcr = fellow_cache_obj_get(fc, &toc, priv2, 0);
 		assert(fcr.status == fcr_ok);
 		fco = fcr.r.ptr;
 
@@ -5007,7 +5008,7 @@ static void test_fellow_cache_obj_iter_final(
 	// get back the original number of references
 	for (u = 0; u < refcnt; u++) {
 		toc = ocmem;
-		fcr = fellow_cache_obj_get(fc, &toc, priv2);
+		fcr = fellow_cache_obj_get(fc, &toc, priv2, 0);
 		assert(fcr.status == fcr_ok);
 		assert((toc == NULL) == (u == 0));
 		fco = fcr.r.ptr;

src/fellow_cache.h

@@ -58,7 +58,7 @@ void fellow_cache_obj_deref(struct fellow_cache *fc,
     struct fellow_cache_obj *fco);
 
 struct fellow_cache_res fellow_cache_obj_get(struct fellow_cache *fc,
-    struct objcore **ocp, uintptr_t priv2);
+    struct objcore **ocp, uintptr_t priv2, unsigned crit);
 struct fellow_cache_res
 fellow_cache_obj_iter(struct fellow_cache *fc, struct fellow_cache_obj *fco,
     void *priv, objiterate_f func, int final);

src/fellow_storage.c

@@ -426,7 +426,7 @@ fcoc_fini(struct fellow_cache *fc, struct fcoc *fcoc) {
 static struct fcoc
 stvfe_dskoc_fco(struct worker *wrk,
     const struct stevedore *stv, const struct stvfe *stvfe,
-    struct objcore *oc)
+    struct objcore *oc, unsigned crit)
 {
 	struct stvfe_wait_entry *we;
 	struct fellow_cache_res fcr;
@@ -466,7 +466,7 @@ stvfe_dskoc_fco(struct worker *wrk,
 
 	refoc = oc;
 
-	fcr = fellow_cache_obj_get(stvfe->fc, &refoc, oc->stobj->priv2);
+	fcr = fellow_cache_obj_get(stvfe->fc, &refoc, oc->stobj->priv2, crit);
 	if (fcr.status != fcr_ok) {
 		VSLb(wrk->vsl, SLT_Error, "%s %s: %s",
 		    stv->name, stv->ident,
@@ -622,7 +622,8 @@ sfedsk_objfree(struct worker *wrk, struct objcore *dskoc)
 		wrk->stats->n_vampireobject--;
 		stvfe->stats->c_dsk_obj_free_thin++;
 	}
-	else if ((fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc)).fco != NULL) {
+	else if (
+	    (fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc, 0)).fco != NULL) {
 		/* attempt to read to free space */
 		fellow_cache_obj_delete(stvfe->fc, fcoc.fco, oh->digest);
 		fcoc_fini(stvfe->fc, &fcoc);
@@ -659,7 +660,7 @@ sfedsk_objslim(struct worker *wrk, struct objcore *dskoc)
 		return;
 
 	/* turn into memoc for proper delete during LRU */
-	fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc);
+	fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc, 0);
 	fcoc_fini(stvfe->fc, &fcoc);
 
 	stvfe->stats->c_dsk_obj_free_get++;
@@ -843,7 +844,7 @@ sfedsk_iterator(struct worker *wrk, struct objcore *dskoc,
 {
 	const struct stevedore *stv = oc_stv(wrk, dskoc);
 	const struct stvfe *stvfe = stv_stvfe(stv);
-	struct fcoc fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc);
+	struct fcoc fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc, 0);
 	struct fellow_cache_res fcr;
 
 	if (fcoc.fco == NULL)
@@ -1039,7 +1040,8 @@ sfedsk_getattr_tomem(struct worker *wrk, struct objcore *dskoc,
 {
 	const struct stevedore *stv = oc_stv(wrk, dskoc);
 	const struct stvfe *stvfe = stv_stvfe(stv);
-	struct fcoc fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc);
+	struct fcoc fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc,
+	    attr == OA_VARY ? 1 : 0);
 	size_t len;
 	void *r;