fellow: Add object slimming

src/fellow_cache.c

@@ -886,7 +886,6 @@ fellow_cache_seg_wait_locked(const struct fellow_cache_seg *fcs)
 	AZ(pthread_cond_wait(&fcs->fco->cond, &fcs->fco->mtx));
 }
 
-
 /*
  * free the cached segment, not the fellow_(cache|disk)_seg
  *
@@ -2527,6 +2526,54 @@ fellow_cache_lru_seg_evict_locked(
 	fellow_cache_seg_evict_locked(fcs, alloc);
 }
 
+/* evict all segments of an object */
+void
+fellow_cache_obj_slim(struct fellow_cache *fc, struct fellow_cache_obj *fco)
+{
+	struct buddy_returns *rets =
+	    BUDDY_RETURNS_STK(fc->membuddy, BUDDY_RETURNS_MAX);
+	struct buddy_ptr_extent alloc;
+	struct fellow_cache_seg *fcs;
+	struct fcscursor c;
+
+	fcsc_init(&c, &fco->seglist);
+
+	AZ(pthread_mutex_lock(&fco->mtx));
+
+	/* anything to do at all? */
+	while ((fcs = fcsc_next(&c)) != NULL)
+		if (fcs->alloc.ptr)
+			break;
+
+	if (fcs == NULL || fcs->alloc.ptr == NULL)
+		goto out;
+
+	fcsc_init(&c, &fco->seglist);
+	AZ(pthread_mutex_lock(&fc->lru_mtx));
+	while ((fcs = fcsc_next(&c)) != NULL) {
+		if (fcs->alloc.ptr == NULL)
+			break;
+		while (fcs->state == FCS_READING || fcs->state == FCS_WRITING) {
+			AZ(pthread_mutex_unlock(&fc->lru_mtx));
+
+			(void) fellow_cache_seg_ref_locked(fc, fcs);
+			fellow_cache_seg_wait_locked(fcs);
+			AZ(fellow_cache_seg_deref_locked(fc, fcs));
+
+			AZ(pthread_mutex_lock(&fc->lru_mtx));
+		}
+		if (fcs->refcnt || fcs->state != FCS_INCORE)
+			continue;
+		fellow_cache_lru_seg_evict_locked(fcs, &alloc, fc);
+		AN(buddy_return_ptr_extent(rets, &alloc));
+	}
+	AZ(pthread_mutex_unlock(&fc->lru_mtx));
+
+  out:
+	AZ(pthread_mutex_unlock(&fco->mtx));
+	buddy_return(rets);
+	}
+
 static int
 fellow_cache_lru_work(struct worker *wrk, struct fellow_cache *fc)
 {

src/fellow_cache_storage.h

@@ -22,6 +22,8 @@
 
 void fellow_cache_obj_evict_mutate(struct fellow_cache *fc,
     struct fellow_cache_obj *fco);
+void fellow_cache_obj_slim(struct fellow_cache *fc,
+    struct fellow_cache_obj *fco);
 
 void stvfe_oc_log_removed(struct objcore *oc);
 void stvfe_oc_log_submitted(struct objcore *oc);

src/fellow_storage.c

@@ -387,7 +387,8 @@ sfemem_objslim(struct worker *wrk, struct objcore *memoc)
 	 * body. For now, just free any body segments cached in memory
 	 */
 
-	
+	struct fellow_cache_obj *fco = stvfe_memoc_fco(stv, stvfe, memoc);
+	fellow_cache_obj_slim(stvfe->fc, fco);
 }
 
 static void v_matchproto_(objfree_f)