fellow_cache: Wait for io completions before mutating objects

Fixes #52

src/fellow_cache.c

@@ -5157,6 +5157,21 @@ fellow_disk_obj_delete_submit(struct fellow_fd *ffd,
 	fellow_log_dle_submit(ffd, dles, r);
 }
 
+/*
+ * for drain, wait until object is written before mutating
+ */
+void
+fellow_cache_obj_wait_written(struct fellow_cache_obj *fco)
+{
+	CHECK_OBJ_NOTNULL(fco, FELLOW_CACHE_OBJ_MAGIC);
+
+	AZ(pthread_mutex_lock(&fco->mtx));
+	/* we must not free the object's disk space while it is still writing */
+	while (FCO_STATE(fco) == FCO_WRITING)
+		fellow_cache_seg_wait_locked(FCO_FCS(fco));
+	AZ(pthread_mutex_unlock(&fco->mtx));
+}
+
 /* evict, then delete from log (this order is important to avoid
  * fdb reuse)
  *

src/fellow_cache.h

@@ -52,6 +52,7 @@ fellow_cache_init(struct fellow_fd *, buddy_t *, struct stvfe_tune *,
     fellow_task_run_t, uint64_t *);
 void fellow_cache_fini(struct fellow_cache **);
 
+void fellow_cache_obj_wait_written(struct fellow_cache_obj *fco);
 void fellow_cache_obj_delete(struct fellow_cache *fc,
     struct fellow_cache_obj *fco, const uint8_t hash[32]);
 void fellow_cache_obj_deref(struct fellow_cache *fc,

src/fellow_storage.c

@@ -537,6 +537,7 @@ sfemem_free(struct worker *wrk, struct objcore *memoc)
 		LRU_Remove(memoc);
 
 	if (stvfe->shutdown && wrk->strangelove == (int)STVELOVE_IS_DRAIN) {
+		fellow_cache_obj_wait_written(fco);
 		fellow_cache_obj_deref(stvfe->fc, fco);
 	} else {
 		// fellow_cache_obj_delete() implies deref