fellow_storage: Improve error handling for _getattr()

src/fellow_cache.c

@@ -6095,56 +6095,67 @@ fellow_cache_obj_delete(struct fellow_cache *fc,
 	buddy_return(rets);
 }
 
-// XXX status
-void *
+struct fellow_cache_res
 fellow_cache_obj_getattr(struct fellow_cache *fc,
     struct fellow_cache_obj *fco,
     enum obj_attr attr, size_t *len)
 {
+	struct fellow_cache_res fcr;
 	struct fellow_disk_obj *fdo;
+	const char *err;
 
 	CHECK_OBJ_NOTNULL(fco, FELLOW_CACHE_OBJ_MAGIC);
 	fdo = FCO_FDO(fco);
 	CHECK_OBJ_NOTNULL(fdo, FELLOW_DISK_OBJ_MAGIC);
 	AN(len);
 
+	fcr = fco->fcr;
+	if (fcr.status != fcr_ok)
+		return (fcr);
+
+	/* Auxiliary attributes:
+	 *
+	 * ref is never returned and just ignored by
+	 * fellow_cache_seg_auxattr_free()
+	 */
+
 	switch (attr) {
-		/* Fixed size attributes */
 #define FDO_FIXATTR(U, l, s, vs)					\
 	case OA_##U:							\
 		*len = vs;						\
-		return (fdo->fa_##l);
-#include "tbl/fellow_obj_attr.h"
-
-		/* Variable size attributes */
+		fcr = FCR_OK(fdo->fa_##l);				\
+		break;
 #define FDO_VARATTR(U, l)						\
 	case OA_##U:							\
-		if (fdo->va_##l.aoff == 0)				\
-			return (NULL);					\
+		if (fdo->va_##l.aoff == 0) {				\
+			fcr = FCR_OK(NULL);				\
+			break;						\
+		}							\
 		*len = fdo->va_##l.alen;				\
-		return ((uint8_t *)fdo + fdo->va_##l.aoff);
-#include "tbl/fellow_obj_attr.h"
-
-		/* Auxiliary attributes
-		 * ref is never returned and just ignored by
-		 * fellow_cache_seg_auxattr_free()
-		 */
+		fcr = FCR_OK((uint8_t *)fdo + fdo->va_##l.aoff);	\
+		break;
 #define FDO_AUXATTR(U, l)						\
 	case OA_##U: {							\
 		struct fellow_cache_seg *fcs = &fco->aa_##l##_seg;	\
-		if (fcs->disk_seg->seg.size == 0)			\
-			return (NULL);					\
+		if (fcs->disk_seg->seg.size == 0) {			\
+			fcr = FCR_OK(NULL);				\
+			break;						\
+		}							\
 		fellow_cache_seg_auxattr_ref_in(fc, fcs);		\
-		XXXAZ(fellow_cache_seg_check(fcs));			\
+		err = fellow_cache_seg_check(fcs);			\
+		if (err != NULL) {					\
+			fcr = FCR_IOFAIL(err);				\
+			break;						\
+		}							\
 		*len = fcs->u.fcs_len;					\
-		return (fcs->alloc.ptr);				\
+		fcr = FCR_OK(fcs->alloc.ptr);				\
+		break;							\
 	}
 #include "tbl/fellow_obj_attr.h"
-
 	default:
-		break;
+		WRONG("Unsupported OBJ_ATTR");
 	}
-	WRONG("Unsupported OBJ_ATTR");
+	return (fellow_cache_obj_res(fc, fco, fcr));
 }
 
 void *
@@ -6412,15 +6423,16 @@ static const char * const oatest[OA__MAX] = {
 
 void t_getattr(struct fellow_cache *fc, struct fellow_cache_obj *fco)
 {
+	struct fellow_cache_res fcr;
 	unsigned u;
 	size_t sz;
-	const char *v;
 
 	for (u = 0; u < OA__MAX; u++) {
-		v = fellow_cache_obj_getattr(fc, fco, u, &sz);
-		AN(v);
+		fcr = fellow_cache_obj_getattr(fc, fco, u, &sz);
+		assert(fcr.status == fcr_ok);
+		AN(fcr.r.ptr);
 		assert(sz == strlen(oatest[u]));
-		AZ(strncmp(v, oatest[u], sz));
+		AZ(strncmp(fcr.r.ptr, oatest[u], sz));
 	}
 }
 
@@ -6981,7 +6993,7 @@ t_busyobj(unsigned chksum, struct fellow_cache *fc)
 	AN(fco->fcsl->next);
 	// fixattr always return a pointer
 	for (u = OA_VARY; u < OA__MAX; u++)
-		AZ(fellow_cache_obj_getattr(fc, fco, u, &sz));
+		AZ(fellow_cache_obj_getattr(fc, fco, u, &sz).r.ptr);
 	for (u = 0; u < OA__MAX; u++)
 		AN(fellow_busy_setattr(fbo, u, strlen(oatest[u]), oatest[u]));
 	t_getattr(fc, fco);
@@ -7290,7 +7302,7 @@ t_cache(const char *fn, unsigned chksum)
 	AN(fco->fcsl->next);
 	// fixattr always return a pointer
 	for (u = OA_VARY; u < OA__MAX; u++)
-		AZ(fellow_cache_obj_getattr(fc, fco, u, &sz));
+		AZ(fellow_cache_obj_getattr(fc, fco, u, &sz).r.ptr);
 	for (u = 0; u < OA__MAX; u++)
 		AN(fellow_busy_setattr(fbo, u, strlen(oatest[u]), oatest[u]));
 	t_getattr(fc, fco);

src/fellow_cache.h

@@ -64,7 +64,8 @@ struct fellow_cache_res fellow_cache_obj_get(struct fellow_cache *fc,
 struct fellow_cache_res
 fellow_cache_obj_iter(struct fellow_cache *fc, struct fellow_cache_obj *fco,
     void *priv, objiterate_f func, int final);
-void *fellow_cache_obj_getattr(struct fellow_cache *fc,
+struct fellow_cache_res
+fellow_cache_obj_getattr(struct fellow_cache *fc,
     struct fellow_cache_obj *fco,
     enum obj_attr attr, size_t *len);
 int fellow_cache_obj_lru_touch(struct fellow_cache_obj *fco);

src/fellow_storage.c

@@ -504,6 +504,7 @@ stvfe_fcr_handle(struct worker *wrk, struct objcore *oc,
     const struct stevedore *stv, const struct stvfe *stvfe,
     struct fellow_cache_res fcr)
 {
+
 	if (fcr.status == fcr_ok)
 		return;
 
@@ -524,6 +525,7 @@ stvfe_fcr_handle_iter(struct worker *wrk, struct objcore *oc,
     const struct stevedore *stv, const struct stvfe *stvfe,
     struct fellow_cache_res fcr)
 {
+
 	if (fcr.status == fcr_ok)
 		return (fcr.r.integer);
 
@@ -1108,12 +1110,13 @@ sfemem_getattr(struct worker *wrk, struct objcore *memoc, enum obj_attr attr,
 	const struct stevedore *stv = oc_stv(wrk, memoc);
 	const struct stvfe *stvfe = stv_stvfe(stv);
 	struct fellow_cache_obj *fco = stvfe_memoc_fco(stv, stvfe, memoc);
+	struct fellow_cache_res fcr;
 	size_t len;
-	void *r;
 
 	assert_memstv(stvfe, stv);
 
-	r = fellow_cache_obj_getattr(stvfe->fc, fco, attr, &len);
+	fcr = fellow_cache_obj_getattr(stvfe->fc, fco, attr, &len);
+	stvfe_fcr_handle(wrk, memoc, stv, stvfe, fcr);
 
 	if (slen)
 		*slen = (ssize_t)len;
@@ -1121,7 +1124,7 @@ sfemem_getattr(struct worker *wrk, struct objcore *memoc, enum obj_attr attr,
 	if (attr == OA_VARY)
 		(void) fellow_cache_obj_lru_touch(fco);
 
-	return (r);
+	return (fcr.status == fcr_ok ? fcr.r.ptr : NULL);
 }
 
 static const void * v_matchproto_(objgetattr_f)
@@ -1132,17 +1135,18 @@ sfedsk_getattr_tomem(struct worker *wrk, struct objcore *dskoc,
 	const struct stvfe *stvfe = stv_stvfe(stv);
 	struct fcoc fcoc = stvfe_dskoc_fco(wrk, stv, stvfe, dskoc,
 	    attr == OA_VARY ? 1 : 0);
+	struct fellow_cache_res fcr;
 	size_t len;
-	void *r;
 
 	if (fcoc.fco == NULL)
 		return (NULL);
 
-	r = fellow_cache_obj_getattr(stvfe->fc, fcoc.fco, attr, &len);
+	fcr = fellow_cache_obj_getattr(stvfe->fc, fcoc.fco, attr, &len);
+	stvfe_fcr_handle(wrk, dskoc, stv, stvfe, fcr);
 	fcoc_fini(stvfe->fc, &fcoc);
 	if (slen)
 		*slen = (ssize_t)len;
-	return (r);
+	return (fcr.status == fcr_ok ? fcr.r.ptr : NULL);
 }
 
 /* we do not want to pull in the fco just to get the xid, just return 0 */