Add first pass at segment compaction

This is the first draft of compaction which has the core mechanics.

Add segment functions to free a segment's segno and to delete the entry
that refers to the given segment.

Add manifest functions that lock the manifest and dirty and delete
manifest entries.  These are used by the compaction thread to atomically
modify the manfiest with the result of a compaction.

Sort the level 0 entries in the manifest by their sequence.  This lets
compaction use the first oldest entry and reading can walk them
backwards to get them in order and not have to sort.  We also more
carefully use the sequence field in the manifest search key to
differentiate between finding high level entries that overlap and
finding specific entries identified by their seq.

Add some fields to the per-super compact_info struct which support
compaction.  We need to know the limit on the number of segments per
level and we record keys per level which tell us which segment to use
next time that level is compacted.

We kick a compaction thread when we add a manifest entry and that brings
the level count over the limit.

scoutfs_manifest_next_compact() is the first meaty function.  The
compaction thread uses this to get all the segments involved in a
compaction.  It does a quick manifest update if the next manifest
candidate doesn't overlap with any sgements in the next level.

The compaction operation itself is a pretty straight forward
read-modify-write operation.  It asks the manifest to give it references
to the segments it'll need, reads them in, iterates over them to count
and copies items in order to output segments, and atomically updates the
manifest.

Now that the manifest can be dirty without any dirty segments we need to
fix the transaction writing function's assumption that everything flows
from dirty segments.  It also has to now lock and unlock the manifest as
it adds the entry for its level 0 segment.

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

kmod/src/Makefile

@@ -2,6 +2,6 @@ obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o
 
 CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include
  
-scoutfs-y += alloc.o bio.o block.o btree.o buddy.o counters.o crc.o dir.o \
-	     filerw.o kvec.o inode.o ioctl.o item.o manifest.o msg.o name.o \
-	     seg.o scoutfs_trace.o super.o trans.o treap.o xattr.o
+scoutfs-y += alloc.o bio.o block.o btree.o buddy.o compact.o counters.o crc.o \
+	     dir.o filerw.o kvec.o inode.o ioctl.o item.o manifest.o msg.o \
+	     name.o seg.o scoutfs_trace.o super.o trans.o treap.o xattr.o

kmod/src/compact.c

@@ -0,0 +1,531 @@
+/*
+ * Copyright (C) 2017 Versity Software, Inc.  All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License v2 as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ */
+#include <linux/kernel.h>
+#include <linux/fs.h>
+#include <linux/slab.h>
+
+#include "super.h"
+#include "format.h"
+#include "kvec.h"
+#include "seg.h"
+#include "bio.h"
+#include "cmp.h"
+#include "compact.h"
+#include "manifest.h"
+#include "scoutfs_trace.h"
+
+/*
+ * Compaction is what maintains the exponentially increasing number of
+ * segments in each level of the lsm tree and is what merges duplicate
+ * and deletion keys.
+ *
+ * When the manifest is modified in a way that requires compaction it
+ * kicks the compaction thread.  The compaction thread calls into the
+ * manifest to find the segments that need to be compaction.
+ *
+ * The compaction operation itself always involves a single "upper"
+ * segment at a given level and a limited number of "lower" segments at
+ * the next higher level whose key range intersects with the upper
+ * segment.
+ *
+ * Compaction proceeds by iterating over the items in the upper segment
+ * and items in each of the lower segments in sort order.  The items
+ * from the two input segments are copied into new output segments in
+ * sorted order.  Item space is reclaimed as duplicate or deletion items
+ * are removed.
+ *
+ * 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 {
+	struct super_block *sb;
+	struct workqueue_struct *workq;
+	struct work_struct work;
+};
+
+#define DECLARE_COMPACT_INFO(sb, name) \
+	struct compact_info *name = SCOUTFS_SB(sb)->compact_info
+
+struct compact_seg {
+	struct list_head entry;
+
+	u64 segno;
+	u64 seq;
+	u8 level;
+	SCOUTFS_DECLARE_KVEC(first);
+	struct scoutfs_segment *seg;
+	int pos;
+	int saved_pos;
+};
+
+/*
+ * A compaction request.  It's filled up in scoutfs_compact_add() as
+ * the manifest is wlaked and it finds segments involved in the compaction.
+ */
+struct compact_cursor {
+	struct list_head csegs;
+
+	u8 lower_level;
+
+	struct compact_seg *upper;
+	struct compact_seg *saved_upper;
+	struct compact_seg *lower;
+	struct compact_seg *saved_lower;
+};
+
+static void save_pos(struct compact_cursor *curs)
+{
+	struct compact_seg *cseg;
+
+	list_for_each_entry(cseg, &curs->csegs, entry)
+		cseg->saved_pos = cseg->pos;
+
+	curs->saved_upper = curs->upper;
+	curs->saved_lower = curs->lower;
+}
+
+static void restore_pos(struct compact_cursor *curs)
+{
+	struct compact_seg *cseg;
+
+	list_for_each_entry(cseg, &curs->csegs, entry)
+		cseg->pos = cseg->saved_pos;
+
+	curs->upper = curs->saved_upper;
+	curs->lower = curs->saved_lower;
+}
+
+/*
+ * There's some common patterns with scoutfs_manifest_read_items().. may
+ * want some sharing if it's clean.
+ */
+static int read_segments(struct super_block *sb, struct compact_cursor *curs)
+{
+	struct scoutfs_segment *seg;
+	struct compact_seg *cseg;
+	int ret = 0;
+	int err;
+
+	list_for_each_entry(cseg, &curs->csegs, entry) {
+		seg = scoutfs_seg_submit_read(sb, cseg->segno);
+		if (IS_ERR(seg)) {
+			ret = PTR_ERR(seg);
+			break;
+		}
+
+		cseg->seg = seg;
+	}
+
+	list_for_each_entry(cseg, &curs->csegs, entry) {
+		if (!cseg->seg)
+			break;
+
+		err = scoutfs_seg_wait(sb, cseg->seg);
+		if (err && !ret)
+			ret = err;
+
+		/* XXX verify segs */
+	}
+
+	return ret;
+}
+
+/*
+ * This is synchronous for now.  We're just ensuring that the segments
+ * are stable on disk so that the references to them in the dirty manifest
+ * are safe without having to associate dirty segments and manifest entries.
+ */
+static int write_segments(struct super_block *sb, struct list_head *results)
+{
+	struct scoutfs_bio_completion comp;
+	struct compact_seg *cseg;
+	int ret = 0;
+	int err;
+
+	scoutfs_bio_init_comp(&comp);
+
+	list_for_each_entry(cseg, results, entry) {
+		ret = scoutfs_seg_submit_write(sb, cseg->seg, &comp);
+		if (ret)
+			break;
+	}
+
+	err = scoutfs_bio_wait_comp(sb, &comp);
+	if (err && !ret)
+		ret = err;
+
+	return ret;
+}
+
+static struct compact_seg *next_spos(struct compact_cursor *curs,
+				     struct compact_seg *cseg)
+{
+	if (cseg->entry.next == &curs->csegs)
+		return NULL;
+
+	return list_next_entry(cseg, entry);
+}
+
+/*
+ * Point the caller's key and value kvecs at the next item that should
+ * be copied from the segment's position in the upper and lower
+ * segments.  We use the item that has 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.
+ */
+static bool next_item(struct compact_cursor *curs,
+		      struct kvec *item_key, struct kvec *item_val)
+{
+	struct compact_seg *upper = curs->upper;
+	struct compact_seg *lower = curs->lower;
+	SCOUTFS_DECLARE_KVEC(lower_key);
+	SCOUTFS_DECLARE_KVEC(lower_val);
+	bool found = false;
+	int cmp;
+	int ret;
+
+	if (upper) {
+		ret = scoutfs_seg_item_kvecs(upper->seg, upper->pos,
+					     item_key, item_val);
+		if (ret < 0)
+			upper = NULL;
+	}
+
+	while (lower) {
+		ret = scoutfs_seg_item_kvecs(lower->seg, lower->pos,
+					     lower_key, lower_val);
+		if (ret == 0)
+			break;
+		lower = next_spos(curs, lower);
+	}
+
+	/* we're done if all are empty */
+	if (!upper && !lower) {
+		found = false;
+		goto out;
+	}
+
+	/*
+	 * < 0: return upper, advance upper
+	 * == 0: return upper, advance both
+	 * > 0: return lower, advance lower
+	 */
+	if (upper && lower)
+		cmp = scoutfs_kvec_memcmp(item_key, lower_key);
+	else if (upper)
+		cmp = -1;
+	else
+		cmp = 1;
+
+	if (cmp > 0) {
+		scoutfs_kvec_clone(item_key, lower_key);
+		scoutfs_kvec_clone(item_val, lower_val);
+	}
+
+	if (cmp <= 0)
+		upper->pos++;
+	if (cmp >= 0)
+		lower->pos++;
+
+	found = true;
+out:
+	curs->upper = upper;
+	curs->lower = lower;
+
+	return found;
+}
+
+/*
+ * Figure out how many items and bytes of keys we're going to try and
+ * compact into the next segment.
+ */
+static void count_items(struct super_block *sb, struct compact_cursor *curs,
+			u32 *nr_items, u32 *key_bytes)
+{
+	SCOUTFS_DECLARE_KVEC(item_key);
+	SCOUTFS_DECLARE_KVEC(item_val);
+	u32 total;
+
+	*nr_items = 0;
+	*key_bytes = 0;
+	total = sizeof(struct scoutfs_segment_block);
+
+	while (next_item(curs, item_key, item_val)) {
+
+		total += sizeof(struct scoutfs_segment_item) +
+			 scoutfs_kvec_length(item_key) +
+			 scoutfs_kvec_length(item_val);
+
+		if (total > SCOUTFS_SEGMENT_SIZE)
+			break;
+
+		(*nr_items)++;
+		(*key_bytes) += scoutfs_kvec_length(item_key);
+	}
+}
+
+static void compact_items(struct super_block *sb, struct compact_cursor *curs,
+			  struct scoutfs_segment *seg, u32 nr_items,
+			  u32 key_bytes)
+{
+	SCOUTFS_DECLARE_KVEC(item_key);
+	SCOUTFS_DECLARE_KVEC(item_val);
+
+	next_item(curs, item_key, item_val);
+	scoutfs_seg_first_item(sb, seg, item_key, item_val,
+			       nr_items, key_bytes);
+
+	while (--nr_items && next_item(curs, item_key, item_val))
+		scoutfs_seg_append_item(sb, seg, item_key, item_val);
+}
+
+static int compact_segments(struct super_block *sb,
+			    struct compact_cursor *curs,
+			    struct list_head *results)
+{
+	struct scoutfs_segment *seg;
+	struct compact_seg *cseg;
+	u32 key_bytes;
+	u32 nr_items;
+	int ret;
+
+	for (;;) {
+
+		save_pos(curs);
+		count_items(sb, curs, &nr_items, &key_bytes);
+		restore_pos(curs);
+
+		if (nr_items == 0) {
+			ret = 0;
+			break;
+		}
+
+		cseg = kzalloc(sizeof(struct compact_seg), GFP_NOFS);
+		if (!cseg) {
+			ret = -ENOMEM;
+			break;
+		}
+
+		ret = scoutfs_seg_alloc(sb, &seg);
+		if (ret) {
+			kfree(cseg);
+			break;
+		}
+
+		cseg->level = curs->lower_level;
+		cseg->seg = seg;
+		list_add_tail(&cseg->entry, results);
+
+		compact_items(sb, curs, seg, nr_items, key_bytes);
+	}
+
+	return ret;
+}
+
+static void free_csegs(struct list_head *list)
+{
+	struct compact_seg *cseg;
+	struct compact_seg *tmp;
+
+	list_for_each_entry_safe(cseg, tmp, list, entry) {
+		list_del_init(&cseg->entry);
+		scoutfs_seg_put(cseg->seg);
+		scoutfs_kvec_kfree(cseg->first);
+		kfree(cseg);
+	}
+}
+
+int scoutfs_compact_add(struct super_block *sb, void *data, struct kvec *first,
+			u64 segno, u64 seq, u8 level)
+{
+	struct compact_cursor *curs = data;
+	struct compact_seg *cseg;
+	int ret;
+
+	cseg = kzalloc(sizeof(struct compact_seg), GFP_NOFS);
+	if (!cseg) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	list_add_tail(&cseg->entry, &curs->csegs);
+
+	ret = scoutfs_kvec_dup_flatten(cseg->first, first);
+	if (ret)
+		goto out;
+
+	cseg->segno = segno;
+	cseg->seq = seq;
+	cseg->level = level;
+
+	if (!curs->upper) {
+		curs->upper = cseg;
+	} else if (!curs->lower) {
+		curs->lower = cseg;
+		curs->lower_level = level;
+	}
+
+	ret = 0;
+out:
+	return ret;
+}
+
+/*
+ * Atomically update the manifest.  We lock down the manifest so no one
+ * can use it while we're mucking with it.  We can always delete dirty
+ * treap nodes without failure.  So we first dirty the deletion nodes
+ * before modifying anything.  Then we add and if any of those fail we
+ * can delete the dirty previous additions.  Then we can delete the
+ * dirty existing entries without failure.
+ *
+ * XXX does locking the manifest prevent commits?  I would think so?
+ */
+static int update_manifest(struct super_block *sb, struct compact_cursor *curs,
+			   struct list_head *results)
+{
+	struct compact_seg *cseg;
+	struct compact_seg *until;
+	int ret = 0;
+	int err;
+
+	scoutfs_manifest_lock(sb);
+
+	list_for_each_entry(cseg, &curs->csegs, entry) {
+		ret = scoutfs_manifest_dirty(sb, cseg->first,
+					     cseg->seq, cseg->level);
+		if (ret)
+			goto out;
+	}
+
+	list_for_each_entry(cseg, results, entry) {
+		ret = scoutfs_seg_manifest_add(sb, cseg->seg, cseg->level);
+		if (ret) {
+			until = cseg;
+			list_for_each_entry(cseg, results, entry) {
+				if (cseg == until)
+					break;
+				err = scoutfs_seg_manifest_del(sb, cseg->seg,
+							       cseg->level);
+				BUG_ON(err);
+			}
+			goto out;
+		}
+	}
+
+	list_for_each_entry(cseg, &curs->csegs, entry) {
+		ret = scoutfs_manifest_del(sb, cseg->first,
+					   cseg->seq, cseg->level);
+		BUG_ON(ret);
+	}
+
+out:
+	scoutfs_manifest_unlock(sb);
+
+	return ret;
+}
+
+static int free_result_segnos(struct super_block *sb,
+			      struct list_head *results)
+{
+	struct compact_seg *cseg;
+	int ret = 0;
+	int err;
+
+	list_for_each_entry(cseg, results, entry) {
+		/* XXX failure here would be an inconsistency */
+		err = scoutfs_seg_free_segno(sb, cseg->seg);
+		if (err && !ret)
+			ret = err;
+	}
+
+	return ret;
+}
+
+static void scoutfs_compact_func(struct work_struct *work)
+{
+	struct compact_info *ci = container_of(work, struct compact_info, work);
+	struct super_block *sb = ci->sb;
+	struct compact_cursor curs = {{NULL,}};
+	LIST_HEAD(results);
+	int ret;
+
+	INIT_LIST_HEAD(&curs.csegs);
+
+	ret = scoutfs_manifest_next_compact(sb, (void *)&curs) ?:
+	      read_segments(sb, &curs) ?:
+	      compact_segments(sb, &curs, &results) ?:
+	      write_segments(sb, &results) ?:
+	      update_manifest(sb, &curs, &results);
+
+	if (ret)
+		free_result_segnos(sb, &results);
+
+	free_csegs(&curs.csegs);
+	free_csegs(&results);
+
+	WARN_ON_ONCE(ret);
+	trace_printk("ret %d\n", ret);
+}
+
+void scoutfs_compact_kick(struct super_block *sb)
+{
+	DECLARE_COMPACT_INFO(sb, ci);
+
+	queue_work(ci->workq, &ci->work);
+}
+
+int scoutfs_compact_setup(struct super_block *sb)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct compact_info *ci;
+
+	ci = kzalloc(sizeof(struct compact_info), GFP_KERNEL);
+	if (!ci)
+		return -ENOMEM;
+
+	ci->sb = sb;
+	INIT_WORK(&ci->work, scoutfs_compact_func);
+
+	ci->workq = alloc_workqueue("scoutfs_compact", 0, 1);
+	if (!ci->workq) {
+		kfree(ci);
+		return -ENOMEM;
+	}
+
+	sbi->compact_info = ci;
+
+	return 0;
+}
+
+/*
+ * The system should be idle, there should not be any more manifest
+ * modification which would kick compaction.
+ */
+void scoutfs_compact_destroy(struct super_block *sb)
+{
+	DECLARE_COMPACT_INFO(sb, ci);
+
+	if (ci->workq) {
+		flush_work(&ci->work);
+		destroy_workqueue(ci->workq);
+	}
+}

kmod/src/compact.h

@@ -0,0 +1,12 @@
+#ifndef _SCOUTFS_COMPACT_H_
+#define _SCOUTFS_COMPACT_H_
+
+void scoutfs_compact_kick(struct super_block *sb);
+
+int scoutfs_compact_add(struct super_block *sb, void *data, struct kvec *first,
+			u64 segno, u64 seq, u8 level);
+
+int scoutfs_compact_setup(struct super_block *sb);
+void scoutfs_compact_destroy(struct super_block *sb);
+
+#endif

kmod/src/format.h

@@ -96,6 +96,8 @@ struct scoutfs_treap_root {
  */
 #define SCOUTFS_MANIFEST_MAX_LEVEL 20
 
+#define SCOUTFS_MANIFEST_FANOUT 10
+
 struct scoutfs_manifest {
 	struct scoutfs_treap_root root;
 	__le64 level_counts[SCOUTFS_MANIFEST_MAX_LEVEL];

kmod/src/manifest.c

@@ -14,7 +14,6 @@
 #include <linux/module.h>
 #include <linux/fs.h>
 #include <linux/slab.h>
-#include <linux/list_sort.h>
 
 #include "super.h"
 #include "format.h"
@@ -23,6 +22,7 @@
 #include "item.h"
 #include "treap.h"
 #include "cmp.h"
+#include "compact.h"
 #include "manifest.h"
 #include "scoutfs_trace.h"
 
@@ -45,6 +45,11 @@ struct manifest {
 	struct rw_semaphore rwsem;
 	struct scoutfs_treap *treap;
 	u8 nr_levels;
+
+	/* calculated on mount, const thereafter */
+	u64 level_limits[SCOUTFS_MANIFEST_MAX_LEVEL + 1];
+
+	SCOUTFS_DECLARE_KVEC(compact_keys[SCOUTFS_MANIFEST_MAX_LEVEL + 1]);
 };
 
 #define DECLARE_MANIFEST(sb, name) \
@@ -79,6 +84,10 @@ struct manifest_fill_args {
 	struct kvec *last;
 };
 
+/*
+ * Seq is only specified for operations that differentiate between
+ * segments with identical items by their sequence number.
+ */
 struct manifest_search_key {
 	u64 seq;
 	struct kvec *key;
@@ -121,6 +130,8 @@ static bool cmp_range_ment(struct kvec *key, struct kvec *end,
 /*
  * Insert a new manifest entry in the treap.  The treap allocates a new
  * node for us and we fill it.
+ *
+ * This must be called with the manifest lock held.
  */
 int scoutfs_manifest_add(struct super_block *sb, struct kvec *first,
 			 struct kvec *last, u64 segno, u64 seq, u8 level)
@@ -153,22 +164,92 @@ int scoutfs_manifest_add(struct super_block *sb, struct kvec *first,
 	skey.level = level;
 	skey.seq = seq;
 
-	down_write(&mani->rwsem);
-
 	ment = scoutfs_treap_insert(mani->treap, &skey, bytes, &args);
 	if (IS_ERR(ment)) {
 		ret = PTR_ERR(ment);
 	} else {
 		mani->nr_levels = max_t(u8, mani->nr_levels, level + 1);
 		le64_add_cpu(&super->manifest.level_counts[level], 1);
+
+		if (le64_to_cpu(super->manifest.level_counts[level]) >
+		    mani->level_limits[level])
+			scoutfs_compact_kick(sb);
+
 		ret = 0;
 	}
 
-	up_write(&mani->rwsem);
+	return ret;
+}
+
+/*
+ * This must be called with the manifest lock held.
+ */
+int scoutfs_manifest_dirty(struct super_block *sb, struct kvec *first, u64 seq,
+			   u8 level)
+{
+	DECLARE_MANIFEST(sb, mani);
+	struct scoutfs_manifest_entry *ment;
+	struct manifest_search_key skey;
+
+	skey.key = first;
+	skey.level = level;
+	skey.seq = seq;
+
+	ment = scoutfs_treap_lookup_dirty(mani->treap, &skey);
+	if (IS_ERR(ment))
+		return PTR_ERR(ment);
+	if (!ment)
+		return -ENOENT;
+	return 0;
+}
+
+/*
+ * This must be called with the manifest lock held.
+ */
+int scoutfs_manifest_del(struct super_block *sb, struct kvec *first, u64 seq,
+			 u8 level)
+{
+	DECLARE_MANIFEST(sb, mani);
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct scoutfs_super_block *super = &sbi->super;
+	struct manifest_search_key skey;
+	int ret;
+
+	skey.key = first;
+	skey.level = level;
+	skey.seq = seq;
+
+	ret = scoutfs_treap_delete(mani->treap, &skey);
+	if (ret == 0)
+		le64_add_cpu(&super->manifest.level_counts[level], -1ULL);
 
 	return ret;
 }
 
+/*
+ * XXX This feels pretty gross, but it's a simple way to give compaction
+ * atomic updates.  It'll go away once compactions go to the trouble of
+ * communicating their atomic results in a message instead of a series
+ * of function calls.
+ */
+int scoutfs_manifest_lock(struct super_block *sb)
+{
+	DECLARE_MANIFEST(sb, mani);
+
+	down_write(&mani->rwsem);
+
+	return 0;
+}
+
+int scoutfs_manifest_unlock(struct super_block *sb)
+{
+	DECLARE_MANIFEST(sb, mani);
+
+	up_write(&mani->rwsem);
+
+	return 0;
+}
+
 static int alloc_add_ref(struct list_head *list,
 			 struct scoutfs_manifest_entry *ment)
 {
@@ -206,16 +287,6 @@ static int alloc_add_ref(struct list_head *list,
 
 }
 
-/* sort level 0 segments of the list from greatest to least seq */
-static int cmp_ref_list_seqs(void *priv, struct list_head *A,
-			     struct list_head *B)
-{
-	struct manifest_ref *a = list_entry(A, struct manifest_ref, entry);
-	struct manifest_ref *b = list_entry(B, struct manifest_ref, entry);
-
-	return -scoutfs_cmp_u64s(a->seq, b->seq);
-}
-
 /*
  * Get refs on all the segments in the manifest that we'll need to
  * search to populate the cache with the given range.
@@ -238,42 +309,34 @@ static int get_range_refs(struct super_block *sb, struct manifest *mani,
 	SCOUTFS_DECLARE_KVEC(last);
 	struct manifest_ref *ref;
 	struct manifest_ref *tmp;
-	int cmp;
 	int ret;
 	int i;
 
 	down_write(&mani->rwsem);
 
 	/* get level 0 segments that overlap with the missing range */
-	ment = scoutfs_treap_first(mani->treap);
+	skey.level = 0;
+	skey.seq = ~0ULL;
+	ment = scoutfs_treap_lookup_prev(mani->treap, &skey);
 	while (!IS_ERR_OR_NULL(ment)) {
-		if (ment->level > 0)
-			break;
-
-		cmp = cmp_range_ment(key, end, ment);
-		if (cmp < 0)
-			break;
-
-		if (cmp == 0) {
+		if (cmp_range_ment(key, end, ment) == 0) {
 			ret = alloc_add_ref(ref_list, ment);
 			if (ret)
 				goto out;
 		}
 
-		ment = scoutfs_treap_next(mani->treap, ment);
+		ment = scoutfs_treap_prev(mani->treap, ment);
 	}
 	if (IS_ERR(ment)) {
 		ret = PTR_ERR(ment);
 		goto out;
 	}
 
-	/* level0s are sorted by key, reverse sort by seq */
-	list_sort(NULL, ref_list, cmp_ref_list_seqs);
-
 	/* get higher level segments that overlap with the starting key */
 	for (i = 1; i < mani->nr_levels; i++) {
 		skey.key = key;
 		skey.level = i;
+		skey.seq = 0;
 
 		/* XXX should use level counts to skip searches */
 
@@ -528,16 +591,182 @@ int scoutfs_manifest_dirty_ring(struct super_block *sb)
 }
 
 /*
- * Manifest entries are first sorted by their level.
+ * Give the caller the segments that will be involved in the next
+ * compaction.
  *
- * Level 0 segments can arbitrarily overlap.  Their manifest entries are
- * sorted by their first key so that searches can iterate over the
- * entries until first shows that no more segments can overlap.  We then
- * sort by the sequence so that we can manage entries that have
- * identical keys.
+ * For now we have a simple candidate search.  We only initiate
+ * compaction when a level has exceeded its exponentially increasing
+ * limit on the number of segments.  Once we have a level we use keys at
+ * each level to chose the next segment.  This results in a pattern
+ * where clock hands sweep through each level.  The hands wrap much
+ * faster on the higher levels.
  *
- * Higher level segments don't overlap.  There will never be manifest
- * entries with the same key at a given level.
+ * If the candidate segment doesn't overlap with any higher level
+ * segments then just move it down a level.
+ *
+ * If the candidate does overlap then we add all the segments to the
+ * compaction caller's data and let it do its thing.  It'll allocate and
+ * free segments and update the manifest.
+ *
+ * XXX this will get a lot more clever:
+ *  - ensuring concurrent compactions don't overlap
+ *  - prioritize segments with deletion or incremental records
+ *  - prioritize partial segments
+ *  - maybe compact segments by age in a given level
+ */
+int scoutfs_manifest_next_compact(struct super_block *sb, void *data)
+{
+	DECLARE_MANIFEST(sb, mani);
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct scoutfs_super_block *super = &sbi->super;
+	struct scoutfs_manifest_entry *ment;
+	struct scoutfs_manifest_entry *over;
+	struct manifest_search_key skey;
+	SCOUTFS_DECLARE_KVEC(ment_first);
+	SCOUTFS_DECLARE_KVEC(ment_last);
+	SCOUTFS_DECLARE_KVEC(over_first);
+	SCOUTFS_DECLARE_KVEC(over_last);
+	int level;
+	int err;
+	int ret;
+	int i;
+
+	down_write(&mani->rwsem);
+
+	for (level = mani->nr_levels - 1; level >= 0; level--) {
+		if (le64_to_cpu(super->manifest.level_counts[level]) >=
+		    mani->level_limits[level])
+			break;
+	}
+
+	if (level < 0) {
+		ret = 0;
+		goto out;
+	}
+
+	/* find the oldest level 0 or the next higher order level by key */
+	if (level == 0) {
+		ment = scoutfs_treap_first(mani->treap);
+		if (!IS_ERR_OR_NULL(ment) && ment->level)
+			ment = NULL;
+	} else {
+		skey.key = mani->compact_keys[level];
+		skey.level = level;
+		skey.seq = 0;
+		ment = scoutfs_treap_lookup_next(mani->treap, &skey);
+		if (ment == NULL && scoutfs_kvec_length(skey.key)) {
+			/* XXX ugh, these kvecs are the worst */
+			scoutfs_kvec_init(skey.key,
+					  skey.key[0].iov_base, 0);
+			ment = scoutfs_treap_lookup_next(mani->treap, &skey);
+		}
+	}
+	if (IS_ERR(ment)) {
+		ret = PTR_ERR(ment);
+		goto out;
+	}
+	if (ment == NULL || ment->level != level) {
+		/* XXX shouldn't be possible */
+		ret = 0;
+		goto out;
+	}
+
+	init_ment_keys(ment, ment_first, ment_last);
+
+	/* find first overlapping at the next level */
+	skey.key = ment_first;
+	skey.level = level + 1;
+	skey.seq = 0;
+	over = scoutfs_treap_lookup(mani->treap, &skey);
+	if (IS_ERR(over)) {
+		ret = PTR_ERR(over);
+		goto out;
+	}
+
+	/* if there's no overlap we can just move it down a level */
+	if (!over) {
+		ret = scoutfs_manifest_add(sb, ment_first, ment_last,
+					   le64_to_cpu(ment->segno),
+					   le64_to_cpu(ment->seq),
+					   ment->level + 1);
+		if (ret)
+			goto out;
+
+		ret = scoutfs_manifest_del(sb, ment_first,
+					   le64_to_cpu(ment->seq),
+					   ment->level);
+		if (ret) {
+			err = scoutfs_manifest_del(sb, ment_first,
+						   le64_to_cpu(ment->seq),
+						   ment->level + 1);
+			BUG_ON(err);
+			goto out;
+		}
+
+		goto done;
+	}
+
+	/* add the upper input segment */
+	ret = scoutfs_compact_add(sb, data, ment_first,
+				  le64_to_cpu(ment->segno),
+				  le64_to_cpu(ment->seq), level);
+	if (ret)
+		goto out;
+
+	/* add a fanout's worth of lower overlapping segments */
+	init_ment_keys(over, over_first, over_last);
+	for (i = 0; i < SCOUTFS_MANIFEST_FANOUT; i++) {
+		ret = scoutfs_compact_add(sb, data, over_first,
+					  le64_to_cpu(over->segno),
+					  le64_to_cpu(over->seq), level + 1);
+		if (ret)
+			goto out;
+
+		over = scoutfs_treap_next(mani->treap, over);
+		if (IS_ERR(over)) {
+			ret = PTR_ERR(over);
+			goto out;
+		}
+		if (!over || over->level != (ment->level + 1))
+			break;
+
+		init_ment_keys(over, over_first, over_last);
+		if (scoutfs_kvec_cmp_overlap(ment_first, ment_last,
+					     over_first, over_last) != 0)
+			break;
+	}
+
+done:
+	/* record the next key to start from, not exact */
+	scoutfs_kvec_init_key(mani->compact_keys[level]);
+	scoutfs_kvec_memcpy_truncate(mani->compact_keys[level], ment_last);
+	scoutfs_kvec_be_inc(mani->compact_keys[level]);
+
+	ret = 0;
+out:
+	up_write(&mani->rwsem);
+	return ret;
+}
+
+/*
+ * Manifest entries for all levels are stored in a single treap.
+ *
+ * First they're sorted by their level.
+ *
+ * Level 0 segments can contain any items which overlap so they are
+ * sorted by their sequence number.  Compaction can find the first node
+ * and reading walks backwards through level 0 to get them from newest
+ * to oldest to resolve matching items.
+ *
+ * Higher level segments don't overlap.  They are sorted by their first
+ * key.
+ *
+ * Searching comparisons are different than insertion and deletion
+ * comparisons for higher level segments.  Searches want to find the
+ * segment that intersects with a given key.  Insertions and deletions
+ * want to operate on the segment with a specific first key and sequence
+ * number.  We tell the difference by the presence of a sequence number.
+ * A segment will never have a seq of 0.
  */
 static int manifest_treap_compare(void *key, void *data)
 {
@@ -545,21 +774,34 @@ static int manifest_treap_compare(void *key, void *data)
 	struct scoutfs_manifest_entry *ment = data;
 	SCOUTFS_DECLARE_KVEC(first);
 	SCOUTFS_DECLARE_KVEC(last);
+	int cmp;
 
-	if (skey->level < ment->level)
-		return -1;
-	if (skey->level > ment->level)
-		return 1;
+	if (skey->level < ment->level) {
+		cmp = -1;
+		goto out;
+	}
+	if (skey->level > ment->level) {
+		cmp = 1;
+		goto out;
+	}
 
-	init_ment_keys(ment, first, NULL);
+	if (skey->level == 0) {
+		cmp = scoutfs_cmp_u64s(skey->seq, le64_to_cpu(ment->seq));
+		goto out;
+	}
 
-	if (skey->level == 0)
-		return scoutfs_kvec_memcmp(skey->key, first) ?:
-		       scoutfs_cmp_u64s(skey->seq, le64_to_cpu(ment->seq));
+	init_ment_keys(ment, first, last);
 
-	init_ment_keys(ment, NULL, last);
+	if (skey->seq == 0) {
+		cmp = scoutfs_kvec_cmp_overlap(skey->key, skey->key,
+					       first, last);
+	} else {
+		cmp = scoutfs_kvec_memcmp(skey->key, first) ?:
+		      scoutfs_cmp_u64s(skey->seq, le64_to_cpu(ment->seq));
+	}
 
-	return scoutfs_kvec_cmp_overlap(skey->key, skey->key, first, last);
+out:
+	return cmp;
 }
 
 static void manifest_treap_fill(void *data, void *arg)
@@ -588,6 +830,7 @@ int scoutfs_manifest_setup(struct super_block *sb)
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super = &sbi->super;
 	struct manifest *mani;
+	int ret;
 	int i;
 
 	mani = kzalloc(sizeof(struct manifest), GFP_KERNEL);
@@ -602,6 +845,17 @@ int scoutfs_manifest_setup(struct super_block *sb)
 		return -ENOMEM;
 	}
 
+	for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++) {
+		ret = scoutfs_kvec_alloc_key(mani->compact_keys[i]);
+		if (ret) {
+			while (--i >= 0)
+				scoutfs_kvec_kfree(mani->compact_keys[i]);
+			scoutfs_treap_free(mani->treap);
+			kfree(mani);
+			return -ENOMEM;
+		}
+	}
+
 	for (i = ARRAY_SIZE(super->manifest.level_counts) - 1; i >= 0; i--) {
 		if (super->manifest.level_counts[i]) {
 			mani->nr_levels = i + 1;
@@ -609,6 +863,14 @@ int scoutfs_manifest_setup(struct super_block *sb)
 		}
 	}
 
+	/* always trigger a compaction if there's a single l0 segment? */
+	mani->level_limits[0] = 0;
+	mani->level_limits[1] = SCOUTFS_MANIFEST_FANOUT;
+	for (i = 2; i < ARRAY_SIZE(mani->level_limits); i++) {
+		mani->level_limits[i] = mani->level_limits[i - 1] *
+					SCOUTFS_MANIFEST_FANOUT;
+	}
+
 	sbi->manifest = mani;
 
 	return 0;
@@ -618,9 +880,12 @@ void scoutfs_manifest_destroy(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct manifest *mani = sbi->manifest;
+	int i;
 
 	if (mani) {
 		scoutfs_treap_free(mani->treap);
+		for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++)
+			scoutfs_kvec_kfree(mani->compact_keys[i]);
 		kfree(mani);
 	}
 }

kmod/src/manifest.h

@@ -3,12 +3,21 @@
 
 int scoutfs_manifest_add(struct super_block *sb, struct kvec *first,
 			 struct kvec *last, u64 segno, u64 seq, u8 level);
+int scoutfs_manifest_dirty(struct super_block *sb, struct kvec *first, u64 seq,
+			   u8 level);
+int scoutfs_manifest_del(struct super_block *sb, struct kvec *first, u64 seq,
+			 u8 level);
 int scoutfs_manifest_has_dirty(struct super_block *sb);
 int scoutfs_manifest_dirty_ring(struct super_block *sb);
 
+int scoutfs_manifest_lock(struct super_block *sb);
+int scoutfs_manifest_unlock(struct super_block *sb);
+
 int scoutfs_manifest_read_items(struct super_block *sb, struct kvec *key,
 				struct kvec *until);
 
+int scoutfs_manifest_next_compact(struct super_block *sb, void *data);
+
 int scoutfs_manifest_setup(struct super_block *sb);
 void scoutfs_manifest_destroy(struct super_block *sb);
 

kmod/src/seg.c

@@ -246,6 +246,16 @@ out:
 
 }
 
+/*
+ * This just frees the segno for the given seg.  It's gross but
+ * symmetrical with only being able to allocate segnos by allocating a
+ * seg.  We'll probably have to do better.
+ */
+int scoutfs_seg_free_segno(struct super_block *sb, struct scoutfs_segment *seg)
+{
+	return scoutfs_alloc_free(sb, seg->segno);
+}
+
 /*
  * The bios submitted by this don't have page references themselves.  If
  * this succeeds then the caller must call _wait before putting their
@@ -546,6 +556,19 @@ int scoutfs_seg_manifest_add(struct super_block *sb,
 				    le64_to_cpu(sblk->seq), level);
 }
 
+int scoutfs_seg_manifest_del(struct super_block *sb,
+			     struct scoutfs_segment *seg, u8 level)
+{
+	struct scoutfs_segment_block *sblk = off_ptr(seg, 0);
+	struct native_item item;
+	SCOUTFS_DECLARE_KVEC(first);
+
+	load_item(seg, 0, &item);
+	kvec_from_pages(seg, first, item.key_off, item.key_len);
+
+	return scoutfs_manifest_del(sb, first, le64_to_cpu(sblk->seq), level);
+}
+
 int scoutfs_seg_setup(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);

kmod/src/seg.h

@@ -16,6 +16,8 @@ int scoutfs_seg_item_kvecs(struct scoutfs_segment *seg, int pos,
 void scoutfs_seg_put(struct scoutfs_segment *seg);
 
 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);
 void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment *seg,
 			    struct kvec *key, struct kvec *val,
 			    unsigned int nr_items, unsigned int key_bytes);
@@ -24,6 +26,8 @@ void scoutfs_seg_append_item(struct super_block *sb,
 			     struct kvec *key, struct kvec *val);
 int scoutfs_seg_manifest_add(struct super_block *sb,
 			     struct scoutfs_segment *seg, u8 level);
+int scoutfs_seg_manifest_del(struct super_block *sb,
+			     struct scoutfs_segment *seg, u8 level);
 
 int scoutfs_seg_submit_write(struct super_block *sb,
 			     struct scoutfs_segment *seg,

kmod/src/super.c

@@ -34,6 +34,7 @@
 #include "bio.h"
 #include "alloc.h"
 #include "treap.h"
+#include "compact.h"
 #include "scoutfs_trace.h"
 
 static struct kset *scoutfs_kset;
@@ -230,6 +231,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	      scoutfs_alloc_setup(sb) ?:
 	      scoutfs_treap_setup(sb) ?:
 //	      scoutfs_buddy_setup(sb) ?:
+	      scoutfs_compact_setup(sb) ?:
 	      scoutfs_setup_trans(sb);
 	if (ret)
 		return ret;
@@ -261,6 +263,7 @@ static void scoutfs_kill_sb(struct super_block *sb)
 
 	kill_block_super(sb);
 	if (sbi) {
+		scoutfs_compact_destroy(sb);
 		scoutfs_shutdown_trans(sb);
 		scoutfs_buddy_destroy(sb);
 		if (sbi->block_shrinker.shrink == scoutfs_block_shrink)

kmod/src/super.h

@@ -13,6 +13,7 @@ struct item_cache;
 struct manifest;
 struct segment_cache;
 struct treap_info;
+struct compact_info;
 
 struct scoutfs_sb_info {
 	struct super_block *sb;
@@ -37,6 +38,7 @@ struct scoutfs_sb_info {
 	struct segment_cache *segment_cache;
 	struct seg_alloc *seg_alloc;
 	struct treap_info *treap_info;
+	struct compact_info *compact_info;
 
 	struct buddy_info *buddy_info;
 

kmod/src/trans.c

@@ -84,6 +84,7 @@ void scoutfs_trans_write_func(struct work_struct *work)
 	struct scoutfs_segment *seg;
 	bool advance = false;
 	int ret = 0;
+	int err;
 
 	scoutfs_bio_init_comp(&comp);
 	sbi->trans_task = NULL;
@@ -96,24 +97,44 @@ void scoutfs_trans_write_func(struct work_struct *work)
 	scoutfs_filerw_free_alloc(sb);
 #endif
 
-	if (scoutfs_item_dirty_bytes(sb) || scoutfs_manifest_has_dirty(sb) ||
-	    scoutfs_alloc_has_dirty(sb)) {
-
+	/*
+	 * XXX this needs serious work to handle errors.
+	 */
+	while (scoutfs_item_dirty_bytes(sb)) {
+		advance = true;
+		seg = NULL;
 		ret = scoutfs_seg_alloc(sb, &seg) ?:
 		      scoutfs_item_dirty_seg(sb, seg) ?:
+		      scoutfs_manifest_lock(sb) ?:
 		      scoutfs_seg_manifest_add(sb, seg, 0) ?:
-		      scoutfs_manifest_dirty_ring(sb) ?:
-		      scoutfs_alloc_dirty_ring(sb) ?:
-		      scoutfs_treap_submit_write(sb, &comp) ?:
-		      scoutfs_seg_submit_write(sb, seg, &comp) ?:
-		      scoutfs_bio_wait_comp(sb, &comp) ?:
-		      scoutfs_write_dirty_super(sb);
-		BUG_ON(ret);
-
+		      scoutfs_manifest_unlock(sb) ?:
+		      scoutfs_seg_submit_write(sb, seg, &comp);
 		scoutfs_seg_put(seg);
-		advance = true;
+		if (ret)
+			goto out;
 	}
 
+	if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) {
+		advance = true;
+		ret = scoutfs_manifest_dirty_ring(sb) ?:
+		      scoutfs_alloc_dirty_ring(sb) ?:
+		      scoutfs_treap_submit_write(sb, &comp);
+		if (ret)
+			goto out;
+	}
+
+out:
+	err = scoutfs_bio_wait_comp(sb, &comp) ?:
+	      scoutfs_write_dirty_super(sb);
+	if (err && !ret)
+		ret = err;
+
+	/* XXX this all needs serious work for dealing with errors */
+	WARN_ON_ONCE(ret);
+
+	if (advance && ret)
+		advance = false;
+
 	spin_lock(&sbi->trans_write_lock);
 	if (advance)
 		scoutfs_advance_dirty_super(sb);