fellow_cache: Handle fcs read errors

src/fellow_cache.c

@@ -3348,36 +3348,38 @@ fellow_cache_obj_unbusy(struct fellow_busy *fbo, enum fcol_state wantlog)
 static void
 fellow_cache_read_complete(struct fellow_cache *fc, void *ptr, int32_t result)
 {
-	enum fcos_state fcos_next = (typeof(fcos_next))FCOS_CHECK;
 	struct fellow_cache_seg *fcs;
 	struct fellow_cache_obj *fco;
+	struct fellow_cache_res fcr;
+	enum fcos_state fcos_next;
 	unsigned refcount;
 
 	CAST_OBJ_NOTNULL(fcs, ptr, FELLOW_CACHE_SEG_MAGIC);
 	assert(FCOS(fcs->state) == FCOS_READING);
 
+	fco = fcs->fco;
+	CHECK_OBJ_NOTNULL(fco, FELLOW_CACHE_OBJ_MAGIC);
+
 	assert(fcs->alloc.size <= INT32_MAX);
-	if (result < (int32_t)fcs->alloc.size) {
-		if (fc->tune->ioerr_obj == 0)
-			FC_WRONG("read error result %d size %d",
-			    result, (int32_t)fcs->alloc.size);
+	if (FC_INJ || result < (int32_t)fcs->alloc.size) {
+		fcr = FCR_IOFAIL("fcs read error");
 		fcos_next = (typeof(fcos_next))FCOS_READFAIL;
-		// XXX handle
-		INCOMPL();
+	}
+	else {
+		fcr = FCR_OK(fco);
+		fcos_next = (typeof(fcos_next))FCOS_CHECK;
 	}
 
 	//lint -e{655} bit-wise operation uses (compatible) enum's
 	fcos_next |= (typeof(fcos_next))FCOS_HIGH(fcs->state);
 
-	fco = fcs->fco;
-	CHECK_OBJ_NOTNULL(fco, FELLOW_CACHE_OBJ_MAGIC);
-
 	struct fellow_lru_chgbatch lcb[1] =
 	    FELLOW_LRU_CHGBATCH_INIT(lcb, fco, 1);
 
 	assert_fcos_transition(fcs->state, fcos_next);
 
 	AZ(pthread_mutex_lock(&fco->mtx));
+	(void) fellow_cache_obj_res(fc, fco, fcr);
 	fellow_cache_seg_transition_locked(lcb, fcs, fcs->state, fcos_next);
 	// io holds a ref on the seg and the fco
 	if (fellow_cache_seg_deref_locked(lcb, fcs))
@@ -6671,6 +6673,9 @@ static void test_fellow_cache_obj_iter_final(
 		} else
 			injcount--;
 
+		/* dumb wait until writes are complete */
+		while (FCO_REFCNT(fco) > 1)
+			usleep(1000);
 		assert(FCO_REFCNT(fco) == 1);
 		fellow_cache_obj_deref(fc, fco);
 	}