Add deletion items

So far we were only able to add items to the segments.  To support
deletion we have to insert deletion items and then remove them and the
item they reference when their segments are compacted.

As callers attempt to delete items from the item cache we replace the
existing item with a deletion marker with the key but no value.  Now
that there are deletion items in the cache we have to teach the other
item cache operations to skip them.  There's some noise in the patch
from moving functions around so that item insertion can free a deletion
item it finds.

The deletion items are written out to the segment as usual except now
the in-segment item struct has a flag to mark a deletion item and the
deletion item is removed from the cache once its written to the segment.

Item reading knows to skip deletion items and not add them back into
the cache.

Compaction proceeds as usual for most of the levels with the deletion
item clobbering any older higher level items with the same key.
Eventually the deletion item itself is removed by skipping over it when
compacting to the largest final level.  We support this by adding a
little call that describes the max level of the tree at the time the
compaction starts so that compaction can tell when it should skip
copying the deletion item to the final lower level.

All of this is for deletion of items with a precise key.  In the future
we'll expand the deletion items so that they can reference a contiguous
range of keys.

Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/compact.c

@@ -49,10 +49,6 @@
  * Once the compaction is completed the manifest is updated to remove
  * the input segments and add the output segments.  Here segment space
  * is reclaimed when the input items fit in fewer output segments.
- *
- * XXX today we only know how to skip duplicate individual items.  We'll
- * need to know how to skip lower based on upper range deletion items
- * and to combine incremental update items.
  */
 
 struct compact_info {
@@ -85,6 +81,7 @@ struct compact_cursor {
 	struct list_head csegs;
 
 	u8 lower_level;
+	u8 last_level;
 
 	struct compact_seg *upper;
 	struct compact_seg *saved_upper;
@@ -193,22 +190,25 @@ static struct compact_seg *next_spos(struct compact_cursor *curs,
  * the lowest key or the upper if they're the same.  We advance the
  * cursor past the item that is returned.
  *
- * XXX this will get fancier as we get range deletion items and incremental
- * update items.
+ * XXX this will get fancier as we get range deletion items and
+ * incremental update items.
  */
 static int next_item(struct super_block *sb, struct compact_cursor *curs,
-		     struct scoutfs_key_buf *item_key, struct kvec *item_val)
+		     struct scoutfs_key_buf *item_key, struct kvec *item_val,
+		     u8 *item_flags)
 {
 	struct compact_seg *upper = curs->upper;
 	struct compact_seg *lower = curs->lower;
 	struct scoutfs_key_buf lower_key;
 	SCOUTFS_DECLARE_KVEC(lower_val);
+	u8 lower_flags;
 	int cmp;
 	int ret;
 
+retry:
 	if (upper) {
 		ret = scoutfs_seg_item_ptrs(upper->seg, upper->pos,
-					    item_key, item_val);
+					    item_key, item_val, item_flags);
 		if (ret < 0)
 			upper = NULL;
 	}
@@ -219,7 +219,8 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs,
 			goto out;
 
 		ret = scoutfs_seg_item_ptrs(lower->seg, lower->pos,
-					    &lower_key, lower_val);
+					    &lower_key, lower_val,
+					    &lower_flags);
 		if (ret == 0)
 			break;
 		lower = next_spos(curs, lower);
@@ -246,6 +247,7 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs,
 	if (cmp > 0) {
 		scoutfs_key_clone(item_key, &lower_key);
 		scoutfs_kvec_clone(item_val, lower_val);
+		*item_flags = lower_flags;
 	}
 
 	if (cmp <= 0)
@@ -253,6 +255,16 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs,
 	if (cmp >= 0)
 		lower->pos++;
 
+	/*
+	 * Deletion items make their way down all the levels, replacing
+	 * all the duplicate items that they find.  When we're
+	 * compacting to the last level we can remove them by retrying
+	 * the search after we've advanced past them.
+	 */
+	if ((curs->lower_level == curs->last_level) &&
+	    ((*item_flags) & SCOUTFS_ITEM_FLAG_DELETION))
+		goto retry;
+
 	ret = 1;
 out:
 	curs->upper = upper;
@@ -273,12 +285,13 @@ static int count_items(struct super_block *sb, struct compact_cursor *curs,
 	u32 items = 0;
 	u32 keys = 0;
 	u32 vals = 0;
+	u8 flags;
 	int ret;
 
 	*nr_items = 0;
 	*key_bytes = 0;
 
-	while ((ret = next_item(sb, curs, &item_key, item_val)) > 0) {
+	while ((ret = next_item(sb, curs, &item_key, item_val, &flags)) > 0) {
 
 		items++;
 		keys += item_key.key_len;
@@ -300,21 +313,22 @@ static int compact_items(struct super_block *sb, struct compact_cursor *curs,
 {
 	struct scoutfs_key_buf item_key;
 	SCOUTFS_DECLARE_KVEC(item_val);
+	u8 flags;
 	int ret;
 
-	ret = next_item(sb, curs, &item_key, item_val);
+	ret = next_item(sb, curs, &item_key, item_val, &flags);
 	if (ret <= 0)
 		goto out;
 
-	scoutfs_seg_first_item(sb, seg, &item_key, item_val,
+	scoutfs_seg_first_item(sb, seg, &item_key, item_val, flags,
 			       nr_items, key_bytes);
 
 	while (--nr_items) {
-		ret = next_item(sb, curs, &item_key, item_val);
+		ret = next_item(sb, curs, &item_key, item_val, &flags);
 		if (ret <= 0)
 			break;
 
-		scoutfs_seg_append_item(sb, seg, &item_key, item_val);
+		scoutfs_seg_append_item(sb, seg, &item_key, item_val, flags);
 	}
 
 out:
@@ -344,6 +358,12 @@ static int compact_segments(struct super_block *sb,
 		/*
 		 * We can just move the upper segment down a level if it
 		 * doesn't intersect any lower segments.
+		 *
+		 * XXX we can't do this if the segment we're moving has
+		 * deletion items.  We need to copy the non-deletion items
+		 * and drop the deletion items in that case.  To do that
+		 * we'll need the manifest to count the number of deletion
+		 * and non-deletion items.
 		 */
 		if (upper && upper->pos == 0 &&
 		    (!lower ||
@@ -383,11 +403,14 @@ static int compact_segments(struct super_block *sb,
 		/*
 		 * We can skip a lower segment if there's no upper segment
 		 * or the next upper item is past the last in the lower.
+		 *
+		 * XXX this will need to test for intersection with range
+		 * deletion items.
 		 */
 		if (lower && lower->pos == 0 &&
 		    (!upper ||
 		     (!scoutfs_seg_item_ptrs(upper->seg, upper->pos,
-					     &upper_next, NULL) &&
+					     &upper_next, NULL, NULL) &&
 		      scoutfs_key_compare(&upper_next, lower->last) > 0))) {
 
 			curs->lower = next_spos(curs, lower);
@@ -447,6 +470,25 @@ static int compact_segments(struct super_block *sb,
 	return ret;
 }
 
+/*
+ * Manifest walking is providing the details of the overall compaction
+ * operation.  It'll then add all the segments involved.
+ */
+void scoutfs_compact_describe(struct super_block *sb, void *data,
+			      u8 upper_level, u8 last_level)
+{
+	struct compact_cursor *curs = data;
+
+	curs->lower_level = upper_level + 1;
+	curs->last_level = last_level;
+}
+
+/*
+ * Add a segment involved in the compaction operation.
+ *
+ * XXX Today we know that the caller is always adding only one upper segment
+ * and is then possibly adding all the lower overlapping segments.
+ */
 int scoutfs_compact_add(struct super_block *sb, void *data,
 			struct scoutfs_key_buf *first,
 			struct scoutfs_key_buf *last, u64 segno, u64 seq,
@@ -468,12 +510,10 @@ int scoutfs_compact_add(struct super_block *sb, void *data,
 	cseg->seq = seq;
 	cseg->level = level;
 
-	if (!curs->upper) {
+	if (!curs->upper)
 		curs->upper = cseg;
-	} else if (!curs->lower) {
+	else if (!curs->lower)
 		curs->lower = cseg;
-		curs->lower_level = level;
-	}
 
 	ret = 0;
 out:

kmod/src/compact.h

@@ -3,6 +3,8 @@
 
 void scoutfs_compact_kick(struct super_block *sb);
 
+void scoutfs_compact_describe(struct super_block *sb, void *data,
+			      u8 upper_level, u8 last_level);
 int scoutfs_compact_add(struct super_block *sb, void *data,
 			struct scoutfs_key_buf *first,
 			struct scoutfs_key_buf *last, u64 segno, u64 seq,

kmod/src/counters.h

@@ -27,6 +27,7 @@
 	EXPAND_COUNTER(item_create)		\
 	EXPAND_COUNTER(item_lookup_hit)		\
 	EXPAND_COUNTER(item_lookup_miss)	\
+	EXPAND_COUNTER(item_delete)		\
 	EXPAND_COUNTER(item_range_hit)		\
 	EXPAND_COUNTER(item_range_miss)		\
 	EXPAND_COUNTER(item_range_insert)

kmod/src/format.h

@@ -137,9 +137,12 @@ struct scoutfs_segment_item {
 	__le32 val_off;
 	__le16 key_len;
 	__le16 val_len;
-	__u8 padding[12];
+	__u8 padding[11];
+	__u8 flags;
 } __packed;
 
+#define SCOUTFS_ITEM_FLAG_DELETION (1 << 0)
+
 /*
  * Each large segment starts with a segment block that describes the
  * rest of the blocks that make up the segment.

kmod/src/item.c

@@ -33,6 +33,11 @@
  * that are completely described by the items.  This lets it return
  * negative lookups cache hits for items that don't exist without having
  * to constantly perform expensive segment searches.
+ *
+ * Deletions are recorded with items in the rbtree which record the key
+ * of the deletion.  They're removed once they're written to a level0
+ * segment.  While they're present in the cache we have to be careful to
+ * clobber them in creation and skip them in lookups.
  */
 
 struct item_cache {
@@ -57,7 +62,9 @@ struct cached_item {
 		struct rb_node node;
 		struct list_head entry;
 	};
+
 	long dirty;
+	unsigned deletion:1;
 
 	struct scoutfs_key_buf *key;
 
@@ -71,6 +78,38 @@ struct cached_range {
 	struct scoutfs_key_buf *end;
 };
 
+static u8 item_flags(struct cached_item *item)
+{
+	return item->deletion ? SCOUTFS_ITEM_FLAG_DELETION : 0;
+}
+
+static void free_item(struct super_block *sb, struct cached_item *item)
+{
+	if (!IS_ERR_OR_NULL(item)) {
+		scoutfs_key_free(sb, item->key);
+		scoutfs_kvec_kfree(item->val);
+		kfree(item);
+	}
+}
+
+static struct cached_item *alloc_item(struct super_block *sb,
+				      struct scoutfs_key_buf *key,
+				      struct kvec *val)
+{
+	struct cached_item *item;
+
+	item = kzalloc(sizeof(struct cached_item), GFP_NOFS);
+	if (item) {
+		item->key = scoutfs_key_dup(sb, key);
+		if (!item->key || scoutfs_kvec_dup_flatten(item->val, val)) {
+			free_item(sb, item);
+			item = NULL;
+		}
+	}
+
+	return item;
+}
+
 /*
  * Walk the item rbtree and return the item found and the next and
  * prev items.
@@ -105,6 +144,14 @@ static struct cached_item *walk_items(struct rb_root *root,
 	return NULL;
 }
 
+/*
+ * Look for the item with the given key.  Callers of this are looking
+ * for existing items.  They would just return -ENOENT from a deletion
+ * item if we gave it to them so we return null for deletion items.
+ * Callers that would remove a deletion item before inserting a new
+ * version of the item do so by having insert_item() replace existing
+ * deleted items on their behalf.
+ */
 static struct cached_item *find_item(struct super_block *sb,
 				     struct rb_root *root,
 				     struct scoutfs_key_buf *key)
@@ -115,6 +162,9 @@ static struct cached_item *find_item(struct super_block *sb,
 
 	item = walk_items(root, key, &prev, &next);
 
+	if (item && item->deletion)
+		item = NULL;
+
 	if (item)
 		scoutfs_inc_counter(sb, item_lookup_hit);
 	else
@@ -224,11 +274,64 @@ static const struct rb_augment_callbacks scoutfs_item_rb_cb  = {
 };
 
 /*
- * Try to insert the given item.  If there's already an item with the
- * insertion key then return -EEXIST.
+ * The caller has changed an item's dirty bit.  Its child dirty bits are
+ * still consistent.  But its parent's bits might need to be updated.
+ * Its bits are consistent so we don't propagate from the node itself
+ * because it would immediately terminate.
  */
-static int insert_item(struct rb_root *root, struct cached_item *ins)
+static void update_dirty_parents(struct cached_item *item)
 {
+	scoutfs_item_rb_propagate(rb_parent(&item->node), NULL);
+}
+
+static void mark_item_dirty(struct item_cache *cac,
+			    struct cached_item *item)
+{
+	if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node)))
+		return;
+
+	if (item->dirty & ITEM_DIRTY)
+		return;
+
+	item->dirty |= ITEM_DIRTY;
+	cac->nr_dirty_items++;
+	cac->dirty_key_bytes += item->key->key_len;
+	cac->dirty_val_bytes += scoutfs_kvec_length(item->val);
+
+	update_dirty_parents(item);
+}
+
+static void clear_item_dirty(struct item_cache *cac,
+			     struct cached_item *item)
+{
+	if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node)))
+		return;
+
+	if (!(item->dirty & ITEM_DIRTY))
+		return;
+
+	item->dirty &= ~ITEM_DIRTY;
+	cac->nr_dirty_items--;
+	cac->dirty_key_bytes -= item->key->key_len;
+	cac->dirty_val_bytes -= scoutfs_kvec_length(item->val);
+
+	WARN_ON_ONCE(cac->nr_dirty_items < 0 || cac->dirty_key_bytes < 0 ||
+		     cac->dirty_val_bytes < 0);
+
+	update_dirty_parents(item);
+}
+
+/*
+ * Try to insert the given item.  If there's already a non-deletion item
+ * with the insertion key then return -EEXIST.  An existing deletion
+ * item is replaced and freed.
+ *
+ * The caller is responsible for marking the newly inserted item dirty.
+ */
+static int insert_item(struct super_block *sb, struct item_cache *cac,
+		       struct cached_item *ins)
+{
+	struct rb_root *root = &cac->items;
 	struct rb_node **node = &root->rb_node;
 	struct rb_node *parent = NULL;
 	struct cached_item *item;
@@ -248,7 +351,13 @@ static int insert_item(struct rb_root *root, struct cached_item *ins)
 				item->dirty |= RIGHT_DIRTY;
 			node = &(*node)->rb_right;
 		} else {
-			return -EEXIST;
+			if (!item->deletion)
+				return -EEXIST;
+
+			clear_item_dirty(cac, item);
+			rb_replace_node(&item->node, &ins->node, root);
+			free_item(sb, item);
+			return 0;
 		}
 	}
 
@@ -443,6 +552,43 @@ int scoutfs_item_lookup_exact(struct super_block *sb,
 	return ret;
 }
 
+/*
+ * Find the next item to return from the "_next" item interface.  It's the
+ * next item from the key that isn't a deletion item and is within the
+ * bounds of the end of the cache and the caller's last key.
+ */
+static struct cached_item *item_for_next(struct rb_root *root,
+				         struct scoutfs_key_buf *key,
+					 struct scoutfs_key_buf *range_end,
+					 struct scoutfs_key_buf *last)
+{
+	struct cached_item *item;
+	struct rb_node *node;
+
+	/* limit by the lesser of the two */
+	if (scoutfs_key_compare(range_end, last) < 0)
+		last = range_end;
+
+	item = next_item(root, key);
+	while (item) {
+		if (scoutfs_key_compare(item->key, last) > 0) {
+			item = NULL;
+			break;
+		}
+
+		if (!item->deletion)
+			break;
+
+		node = rb_next(&item->node);
+		if (node)
+			item = container_of(node, struct cached_item, node);
+		else
+			item = NULL;
+	}
+
+	return item;
+}
+
 /*
  * Return the next item starting with the given key, returning the last
  * key at the most.
@@ -490,10 +636,8 @@ int scoutfs_item_next(struct super_block *sb, struct scoutfs_key_buf *key,
 		/* see if we have a usable item in cache and before last */
 		cached = check_range(sb, &cac->ranges, key, range_end);
 
-		if (cached && (item = next_item(&cac->items, key)) &&
-		    scoutfs_key_compare(item->key, range_end) <= 0 &&
-		    scoutfs_key_compare(item->key, last) <= 0) {
-
+		if (cached && (item = item_for_next(&cac->items, key,
+						    range_end, last))) {
 			scoutfs_key_copy(key, item->key);
 			if (val)
 				ret = scoutfs_kvec_memcpy(val, item->val);
@@ -565,81 +709,6 @@ int scoutfs_item_next_same_min(struct super_block *sb,
 	return ret;
 }
 
-static void free_item(struct super_block *sb, struct cached_item *item)
-{
-	if (!IS_ERR_OR_NULL(item)) {
-		scoutfs_key_free(sb, item->key);
-		scoutfs_kvec_kfree(item->val);
-		kfree(item);
-	}
-}
-
-/*
- * The caller has changed an item's dirty bit.  Its child dirty bits are
- * still consistent.  But its parent's bits might need to be updated.
- * Its bits are consistent so we don't propagate from the node itself
- * because it would immediately terminate.
- */
-static void update_dirty_parents(struct cached_item *item)
-{
-	scoutfs_item_rb_propagate(rb_parent(&item->node), NULL);
-}
-
-static void mark_item_dirty(struct item_cache *cac,
-			    struct cached_item *item)
-{
-	if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node)))
-		return;
-
-	if (item->dirty & ITEM_DIRTY)
-		return;
-
-	item->dirty |= ITEM_DIRTY;
-	cac->nr_dirty_items++;
-	cac->dirty_key_bytes += item->key->key_len;
-	cac->dirty_val_bytes += scoutfs_kvec_length(item->val);
-
-	update_dirty_parents(item);
-}
-
-static void clear_item_dirty(struct item_cache *cac,
-			     struct cached_item *item)
-{
-	if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node)))
-		return;
-
-	if (!(item->dirty & ITEM_DIRTY))
-		return;
-
-	item->dirty &= ~ITEM_DIRTY;
-	cac->nr_dirty_items--;
-	cac->dirty_key_bytes -= item->key->key_len;
-	cac->dirty_val_bytes -= scoutfs_kvec_length(item->val);
-
-	WARN_ON_ONCE(cac->nr_dirty_items < 0 || cac->dirty_key_bytes < 0 ||
-		     cac->dirty_val_bytes < 0);
-
-	update_dirty_parents(item);
-}
-
-static struct cached_item *alloc_item(struct super_block *sb,
-				      struct scoutfs_key_buf *key,
-				      struct kvec *val)
-{
-	struct cached_item *item;
-
-	item = kzalloc(sizeof(struct cached_item), GFP_NOFS);
-	if (item) {
-		item->key = scoutfs_key_dup(sb, key);
-		if (!item->key || scoutfs_kvec_dup_flatten(item->val, val)) {
-			free_item(sb, item);
-			item = NULL;
-		}
-	}
-
-	return item;
-}
-
 /*
  * Create a new dirty item in the cache.  Returns -EEXIST if an item
  * already exists with the given key.
@@ -660,7 +729,7 @@ int scoutfs_item_create(struct super_block *sb, struct scoutfs_key_buf *key,
 		return -ENOMEM;
 
 	spin_lock_irqsave(&cac->lock, flags);
-	ret = insert_item(&cac->items, item);
+	ret = insert_item(sb, cac, item);
 	if (!ret) {
 		scoutfs_inc_counter(sb, item_create);
 		mark_item_dirty(cac, item);
@@ -745,7 +814,7 @@ int scoutfs_item_insert_batch(struct super_block *sb, struct list_head *list,
 
 	list_for_each_entry_safe(item, tmp, list, entry) {
 		list_del(&item->entry);
-		if (insert_item(&cac->items, item))
+		if (insert_item(sb, cac, item))
 			list_add(&item->entry, list);
 	}
 
@@ -871,12 +940,61 @@ out:
 }
 
 /*
- * XXX how nice, it'd just creates a cached deletion item.  It doesn't
- * have to read.
+ * Delete an existing item with the given key.
+ *
+ * If a non-deletion item is present then we mark it dirty and deleted
+ * and free it's value.
+ *
+ * Returns -ENOENT if an item doesn't exist at the key.  This forces us
+ * to read the item before creating a deletion item for it.  XXX If we
+ * relaxed this we'd need to see if callers make use of -ENOENT and if
+ * there are any ways for userspace to overwhelm the system with
+ * deletion items for items that didn't exist in the first place.
  */
 int scoutfs_item_delete(struct super_block *sb, struct scoutfs_key_buf *key)
 {
-	return WARN_ON_ONCE(-EINVAL);
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct item_cache *cac = sbi->item_cache;
+	struct scoutfs_key_buf *end;
+	struct cached_item *item;
+	SCOUTFS_DECLARE_KVEC(del_val);
+	unsigned long flags;
+	int ret;
+
+	scoutfs_kvec_init_null(del_val);
+
+	end = scoutfs_key_alloc(sb, SCOUTFS_MAX_KEY_SIZE);
+	if (!end) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	do {
+		spin_lock_irqsave(&cac->lock, flags);
+
+		item = find_item(sb, &cac->items, key);
+		if (item) {
+			scoutfs_kvec_swap(item->val, del_val);
+			item->deletion = 1;
+			mark_item_dirty(cac, item);
+			scoutfs_inc_counter(sb, item_delete);
+			ret = 0;
+		} else if (check_range(sb, &cac->ranges, key, end)) {
+			ret = -ENOENT;
+		} else {
+			ret = -ENODATA;
+		}
+
+		spin_unlock_irqrestore(&cac->lock, flags);
+
+	} while (ret == -ENODATA &&
+		 (ret = scoutfs_manifest_read_items(sb, key, end)) == 0);
+
+	scoutfs_key_free(sb, end);
+	scoutfs_kvec_kfree(del_val);
+out:
+	trace_printk("ret %d\n", ret);
+	return ret;
 }
 
 /*
@@ -1020,28 +1138,46 @@ static void count_seg_items(struct item_cache *cac, u32 *nr_items,
  * segments and can be partially visible if we only write the first
  * segment.  We probably want to throttle trans enters once we have as
  * many dirty items as our atomic segment updates can write.
+ *
+ * XXX this first/append pattern will go away once we can write a stream
+ * of items to a segment without needing to know the item count to
+ * find the starting key and value offsets.
  */
 int scoutfs_item_dirty_seg(struct super_block *sb, struct scoutfs_segment *seg)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct item_cache *cac = sbi->item_cache;
-	struct cached_item *item;
+	struct cached_item *item = NULL;
+	struct cached_item *del;
 	u32 key_bytes;
 	u32 nr_items;
 
 	count_seg_items(cac, &nr_items, &key_bytes);
 
-	item = first_dirty(cac->items.rb_node);
-	if (item) {
-		scoutfs_seg_first_item(sb, seg, item->key, item->val,
-				       nr_items, key_bytes);
-		clear_item_dirty(cac, item);
-		nr_items--;
-	}
+	/* remember nr_items is passed to _first_item */
+	while (nr_items) {
+
+		if (!item) {
+			item = first_dirty(cac->items.rb_node);
+			scoutfs_seg_first_item(sb, seg, item->key, item->val,
+					       item_flags(item), nr_items,
+					       key_bytes);
+		} else {
+			scoutfs_seg_append_item(sb, seg, item->key, item->val,
+						item_flags(item));
+		}
 
-	while (nr_items-- && (item = next_dirty(item))) {
-		scoutfs_seg_append_item(sb, seg, item->key, item->val);
 		clear_item_dirty(cac, item);
+
+		del = item;
+		item = next_dirty(item);
+
+		if (del->deletion) {
+			rb_erase(&del->node, &cac->items);
+			free_item(sb, del);
+		}
+
+		nr_items--;
 	}
 
 	return 0;

kmod/src/manifest.c

@@ -459,6 +459,8 @@ int scoutfs_manifest_read_items(struct super_block *sb,
 	struct manifest_ref *tmp;
 	LIST_HEAD(ref_list);
 	LIST_HEAD(batch);
+	u8 found_flags = 0;
+	u8 item_flags;
 	int found_ctr;
 	bool found;
 	int ret = 0;
@@ -534,7 +536,8 @@ int scoutfs_manifest_read_items(struct super_block *sb,
 			 * caller's end.
 			 */
 			ret = scoutfs_seg_item_ptrs(ref->seg, ref->pos,
-						    &item_key, item_val);
+						    &item_key, item_val,
+						    &item_flags);
 			if (ret < 0 || scoutfs_key_compare(&item_key, end) > 0){
 				ref->pos = -1;
 				continue;
@@ -554,6 +557,7 @@ int scoutfs_manifest_read_items(struct super_block *sb,
 			/* remember new least key */
 			scoutfs_key_clone(&found_key, &item_key);
 			scoutfs_kvec_clone(found_val, item_val);
+			found_flags = item_flags;
 			ref->found_ctr = ++found_ctr;
 			found = true;
 		}
@@ -566,15 +570,22 @@ int scoutfs_manifest_read_items(struct super_block *sb,
 		}
 
 		/*
+		 * Add the next found item to the batch if it's not a
+		 * deletion item.  We still need to use their key to
+		 * remember the end of the batch for negative caching.
+		 *
 		 * If we fail to add an item we're done.  If we already
-		 * have items it's not a failure and the end of the cached
-		 * range is the last successfully added item.
+		 * have items it's not a failure and the end of the
+		 * cached range is the last successfully added item.
 		 */
-		ret = scoutfs_item_add_batch(sb, &batch, &found_key, found_val);
-		if (ret) {
-			if (n > 0)
-				ret = 0;
-			break;
+		if (!(found_flags & SCOUTFS_ITEM_FLAG_DELETION)) {
+			ret = scoutfs_item_add_batch(sb, &batch, &found_key,
+						     found_val);
+			if (ret) {
+				if (n > 0)
+					ret = 0;
+				break;
+			}
 		}
 
 		/* the last successful key determines range end until run out */
@@ -699,6 +710,8 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data)
 		goto out;
 	}
 
+	scoutfs_compact_describe(sb, data, level, mani->nr_levels - 1);
+
 	/* find the oldest level 0 or the next higher order level by key */
 	if (level == 0) {
 		ment = scoutfs_treap_first(mani->treap);
@@ -737,7 +750,7 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data)
 	skey.key = &ment_first;
 	skey.level = level + 1;
 	skey.seq = 0;
-	over = scoutfs_treap_lookup(mani->treap, &skey);
+	over = scoutfs_treap_lookup_next(mani->treap, &skey);
 
 	/* and add a fanout's worth of lower overlapping segments */
 	for (i = 0; i < SCOUTFS_MANIFEST_FANOUT; i++) {

kmod/src/seg.c

@@ -409,7 +409,8 @@ static void kvec_from_pages(struct scoutfs_segment *seg,
 }
 
 int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos,
-			  struct scoutfs_key_buf *key, struct kvec *val)
+			  struct scoutfs_key_buf *key, struct kvec *val,
+			  u8 *flags)
 {
 	struct scoutfs_segment_block *sblk = off_ptr(seg, 0);
 	struct scoutfs_segment_item *item;
@@ -425,6 +426,8 @@ int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos,
 	if (val)
 		kvec_from_pages(seg, val, le32_to_cpu(item->val_off),
 				le16_to_cpu(item->val_len));
+	if (flags)
+		*flags = item->flags;
 
 	return 0;
 }
@@ -446,7 +449,7 @@ static int find_key_pos(struct scoutfs_segment *seg,
 
 	while (start < end) {
 		pos = start + (end - start) / 2;
-		scoutfs_seg_item_ptrs(seg, pos, &key, NULL);
+		scoutfs_seg_item_ptrs(seg, pos, &key, NULL, NULL);
 
 		cmp = scoutfs_key_compare(search, &key);
 		if (cmp < 0)
@@ -512,9 +515,11 @@ static u32 align_key_off(struct scoutfs_segment *seg, u32 key_off, u32 len)
  *
  * This should never fail because any item must always fit in a segment.
  */
-void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment *seg,
+void scoutfs_seg_first_item(struct super_block *sb,
+			    struct scoutfs_segment *seg,
 			    struct scoutfs_key_buf *key, struct kvec *val,
-			    unsigned int nr_items, unsigned int key_bytes)
+			    u8 flags, unsigned int nr_items,
+			    unsigned int key_bytes)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
@@ -543,15 +548,17 @@ void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment *seg,
 	item->val_off = cpu_to_le32(val_off);
 	item->key_len = cpu_to_le16(key->key_len);
 	item->val_len = cpu_to_le16(scoutfs_kvec_length(val));
+	item->flags = flags;
 
-	scoutfs_seg_item_ptrs(seg, 0, &item_key, item_val);
+	scoutfs_seg_item_ptrs(seg, 0, &item_key, item_val, NULL);
 	scoutfs_key_copy(&item_key, key);
 	scoutfs_kvec_memcpy(item_val, val);
 }
 
 void scoutfs_seg_append_item(struct super_block *sb,
 			     struct scoutfs_segment *seg,
-			     struct scoutfs_key_buf *key, struct kvec *val)
+			     struct scoutfs_key_buf *key, struct kvec *val,
+			     u8 flags)
 {
 	struct scoutfs_segment_block *sblk = off_ptr(seg, 0);
 	struct scoutfs_segment_item *item;
@@ -578,11 +585,12 @@ void scoutfs_seg_append_item(struct super_block *sb,
 	item->val_off = cpu_to_le32(val_off);
 	item->key_len = cpu_to_le16(key->key_len);
 	item->val_len = cpu_to_le16(scoutfs_kvec_length(val));
+	item->flags = flags;
 
 	trace_printk("item %u offs key %u val %u\n",
 		     pos, key_off, val_off);
 
-	scoutfs_seg_item_ptrs(seg, pos, &item_key, item_val);
+	scoutfs_seg_item_ptrs(seg, pos, &item_key, item_val, NULL);
 	scoutfs_key_copy(&item_key, key);
 	scoutfs_kvec_memcpy(item_val, val);
 }

kmod/src/seg.h

@@ -13,7 +13,8 @@ int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg);
 int scoutfs_seg_find_pos(struct scoutfs_segment *seg,
 			 struct scoutfs_key_buf *key);
 int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos,
-			  struct scoutfs_key_buf *key, struct kvec *val);
+			  struct scoutfs_key_buf *key, struct kvec *val,
+			  u8 *flags);
 
 void scoutfs_seg_get(struct scoutfs_segment *seg);
 void scoutfs_seg_put(struct scoutfs_segment *seg);
@@ -22,12 +23,15 @@ int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret);
 int scoutfs_seg_free_segno(struct super_block *sb,
 			   struct scoutfs_segment *seg);
 bool scoutfs_seg_fits_single(u32 nr_items, u32 key_bytes, u32 val_bytes);
-void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment *seg,
+void scoutfs_seg_first_item(struct super_block *sb,
+			    struct scoutfs_segment *seg,
 			    struct scoutfs_key_buf *key, struct kvec *val,
-			    unsigned int nr_items, unsigned int key_bytes);
+			    u8 flags, unsigned int nr_items,
+			    unsigned int key_bytes);
 void scoutfs_seg_append_item(struct super_block *sb,
 			     struct scoutfs_segment *seg,
-			     struct scoutfs_key_buf *key, struct kvec *val);
+			     struct scoutfs_key_buf *key, struct kvec *val,
+			     u8 flags);
 int scoutfs_seg_manifest_add(struct super_block *sb,
 			     struct scoutfs_segment *seg, u8 level);
 int scoutfs_seg_manifest_del(struct super_block *sb,