From 5e0e9ac12eeceada5bfe83e1273226f03c101e75 Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Mon, 10 Apr 2017 10:09:52 -0700 Subject: [PATCH] Move to much simpler manifest/alloc storage Using the treap to be able to incrementally read and write the manifest and allocation storage from all nodes wasn't quite ready for prime time. The biggest problem is that invalidating cached nodes which are the target of native pointers, either for consistency or memory pressure, is problematic. This was getting in the way of adding shared support as readers and writers try to use as much of their treap caches as they can. There were other serious problems that we'd run into eventually: memory pressure from duplicate caching in native nodes and the page cache, small IOs from reading a page at a time, the risk of pathologically imbalanced treaps, and the ring being corrupted if the migration balancing doesn't work (the model assumed you could always dirty an individual node in a transaction, you have to dirty all the parents in each new transaction). Let's back off to a much simpler mechanism while we build the rest of the system around it. We can revisit aggressively optimizing this when it's our worst problem. We'll store the indexes that the manifest server needs in simple preallocated rings with log entries. The server has to read the index in its entirety into a native rbtree before it can work on it. We won't access the physical ring from mounts anymore, they'll send messages to the server. The ring callers are now working with a pinned tree in memory so the interface can be a bit simpler. By storing the indexes in their own rings the code and write path become a lot simper: we have an IO submission path for each index instead of "dirtying" calls per index and then a writing call. All this is much more robust and much less likely to get in our way as we stand up the rest of the system around it. Signed-off-by: Zach Brown --- kmod/src/Makefile | 4 +- kmod/src/alloc.c | 92 +-- kmod/src/alloc.h | 5 +- kmod/src/compact.c | 11 +- kmod/src/format.h | 54 +- kmod/src/manifest.c | 186 +++--- kmod/src/manifest.h | 5 +- kmod/src/ring.c | 803 ++++++++++++++++++++++++++ kmod/src/ring.h | 55 ++ kmod/src/super.c | 4 - kmod/src/super.h | 2 - kmod/src/trans.c | 9 +- kmod/src/treap.c | 1349 ------------------------------------------- kmod/src/treap.h | 47 -- 14 files changed, 1027 insertions(+), 1599 deletions(-) create mode 100644 kmod/src/ring.c create mode 100644 kmod/src/ring.h delete mode 100644 kmod/src/treap.c delete mode 100644 kmod/src/treap.h diff --git a/kmod/src/Makefile b/kmod/src/Makefile index 1448caa6..ecb5f39b 100644 --- a/kmod/src/Makefile +++ b/kmod/src/Makefile @@ -3,5 +3,5 @@ obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include scoutfs-y += alloc.o bio.o compact.o counters.o data.o dir.o kvec.o inode.o \ - ioctl.o item.o key.o lock.o manifest.o msg.o net.o seg.o \ - scoutfs_trace.o super.o trans.o treap.o xattr.o + ioctl.o item.o key.o lock.o manifest.o msg.o net.o ring.o seg.o \ + scoutfs_trace.o super.o trans.o xattr.o diff --git a/kmod/src/alloc.c b/kmod/src/alloc.c index 5be9d490..fa6676d1 100644 --- a/kmod/src/alloc.c +++ b/kmod/src/alloc.c @@ -17,13 +17,13 @@ #include "super.h" #include "format.h" -#include "treap.h" +#include "ring.h" #include "cmp.h" #include "alloc.h" #include "counters.h" /* - * scoutfs allocates segments by storing regions of a bitmap in treap + * scoutfs allocates segments by storing regions of a bitmap in ring * nodes. * * Freed segments are recorded in nodes in an rbtree. The frees can't @@ -40,7 +40,7 @@ struct seg_alloc { struct rw_semaphore rwsem; struct rb_root pending_root; - struct scoutfs_treap *treap; + struct scoutfs_ring_info ring; u64 next_segno; }; @@ -132,7 +132,7 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno) nr = sal->next_segno & SCOUTFS_ALLOC_REGION_MASK; do { - reg = scoutfs_treap_lookup_next_dirty(sal->treap, &ind); + reg = scoutfs_ring_lookup_next(&sal->ring, &ind); } while (reg == NULL && ind && (ind = 0, nr = 0, 1)); if (IS_ERR_OR_NULL(reg)) { @@ -143,6 +143,8 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno) goto out; } + scoutfs_ring_dirty(&sal->ring, reg); + nr = find_next_bit_le(reg->bits, SCOUTFS_ALLOC_REGION_BITS, nr); if (nr >= SCOUTFS_ALLOC_REGION_BITS) { /* XXX corruption? shouldn't find empty regions */ @@ -154,12 +156,8 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno) clear_bit_le(nr, reg->bits); - if (empty_region(reg)) { - ret = scoutfs_treap_delete(sal->treap, &ind); - /* XXX figure out what to do about this inconsistency */ - if (WARN_ON_ONCE(ret)) - goto out; - } + if (empty_region(reg)) + scoutfs_ring_delete(&sal->ring, reg); *segno = (ind << SCOUTFS_ALLOC_REGION_SHIFT) + nr; sal->next_segno = *segno + 1; @@ -178,7 +176,7 @@ out: /* * Record newly freed sgements in pending regions. These are applied to - * treap nodes as the transaction commits. + * ring nodes as the transaction commits. */ int scoutfs_alloc_free(struct super_block *sb, u64 segno) { @@ -234,7 +232,8 @@ int scoutfs_alloc_has_dirty(struct super_block *sb) int ret; down_write(&sal->rwsem); - ret = scoutfs_treap_has_dirty(sal->treap); + ret = !!(scoutfs_ring_has_dirty(&sal->ring) || + !RB_EMPTY_ROOT(&sal->pending_root)); up_write(&sal->rwsem); return ret; @@ -242,13 +241,12 @@ int scoutfs_alloc_has_dirty(struct super_block *sb) /* * First we apply the pending frees to create the final set of dirty - * region nodes and then ask the treap to write them to ring pages. + * region nodes and then ask the ring to write them to the ring. */ -int scoutfs_alloc_dirty_ring(struct super_block *sb) +int scoutfs_alloc_submit_write(struct super_block *sb, + struct scoutfs_bio_completion *comp) { DECLARE_SEG_ALLOC(sb, sal); - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_super_block *super = &sbi->super; struct scoutfs_alloc_region *reg; struct pending_region *pend; struct rb_node *node; @@ -262,30 +260,41 @@ int scoutfs_alloc_dirty_ring(struct super_block *sb) ind = le64_to_cpu(pend->reg.index); - reg = scoutfs_treap_lookup_dirty(sal->treap, &ind); - if (!reg) - reg = scoutfs_treap_insert(sal->treap, &ind, - sizeof(struct scoutfs_alloc_region), - &ind); - if (IS_ERR(reg)) { - ret = PTR_ERR(reg); - goto out; + reg = scoutfs_ring_lookup(&sal->ring, &ind); + if (!reg) { + reg = scoutfs_ring_insert(&sal->ring, &ind, + sizeof(struct scoutfs_alloc_region)); + if (!reg) { + ret = -ENOMEM; + goto out; + } + + memset(reg, 0, sizeof(struct scoutfs_alloc_region)); + reg->index = cpu_to_le64(ind); } - reg->index = pend->reg.index; or_region_bits(reg, &pend->reg); + scoutfs_ring_dirty(&sal->ring, reg); rb_erase(&pend->node, &sal->pending_root); kfree(pend); } - scoutfs_treap_dirty_ring(sal->treap, &super->alloc_treap_root); - ret = 0; + ret = scoutfs_ring_submit_write(sb, &sal->ring, comp); out: up_write(&sal->rwsem); return ret; } +void scoutfs_alloc_write_complete(struct super_block *sb) +{ + DECLARE_SEG_ALLOC(sb, sal); + + down_write(&sal->rwsem); + scoutfs_ring_write_complete(&sal->ring); + up_write(&sal->rwsem); +} + /* * Return the number of blocks free for statfs. */ @@ -303,7 +312,7 @@ u64 scoutfs_alloc_bfree(struct super_block *sb) return bfree; } -static int alloc_treap_compare(void *key, void *data) +static int alloc_ring_compare_key(void *key, void *data) { u64 *ind = key; struct scoutfs_alloc_region *reg = data; @@ -311,25 +320,20 @@ static int alloc_treap_compare(void *key, void *data) return scoutfs_cmp_u64s(*ind, le64_to_cpu(reg->index)); } -static void alloc_treap_fill(void *data, void *fill_arg) +static int alloc_ring_compare_data(void *A, void *B) { - struct scoutfs_alloc_region *reg = data; - u64 *ind = fill_arg; + struct scoutfs_alloc_region *a = A; + struct scoutfs_alloc_region *b = B; - memset(reg, 0, sizeof(struct scoutfs_alloc_region)); - reg->index = cpu_to_le64p(ind); + return scoutfs_cmp_u64s(le64_to_cpu(a->index), le64_to_cpu(b->index)); } -static struct scoutfs_treap_ops alloc_treap_ops = { - .compare = alloc_treap_compare, - .fill = alloc_treap_fill, -}; - int scoutfs_alloc_setup(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; struct seg_alloc *sal; + int ret; /* bits need to be aligned so hosts can use native bitops */ BUILD_BUG_ON(offsetof(struct scoutfs_alloc_region, bits) & @@ -341,11 +345,13 @@ int scoutfs_alloc_setup(struct super_block *sb) init_rwsem(&sal->rwsem); sal->pending_root = RB_ROOT; - sal->treap = scoutfs_treap_alloc(sb, &alloc_treap_ops, - &super->alloc_treap_root); - if (!sal->treap) { + scoutfs_ring_init(&sal->ring, &super->alloc_ring, + alloc_ring_compare_key, alloc_ring_compare_data); + + ret = scoutfs_ring_load(sb, &sal->ring); + if (ret) { kfree(sal); - return -ENOMEM; + return ret; } /* XXX read next_segno from super? */ @@ -362,7 +368,7 @@ void scoutfs_alloc_destroy(struct super_block *sb) struct rb_node *node; if (sal) { - scoutfs_treap_free(sal->treap); + scoutfs_ring_destroy(&sal->ring); while ((node = rb_first(&sal->pending_root))) { pend = container_of(node, struct pending_region, node); rb_erase(&pend->node, &sal->pending_root); diff --git a/kmod/src/alloc.h b/kmod/src/alloc.h index 2a400e64..bb185d90 100644 --- a/kmod/src/alloc.h +++ b/kmod/src/alloc.h @@ -2,12 +2,15 @@ #define _SCOUTFS_ALLOC_H_ struct scoutfs_alloc_region; +struct scoutfs_bio_completion; int scoutfs_alloc_segno(struct super_block *sb, u64 *segno); int scoutfs_alloc_free(struct super_block *sb, u64 segno); int scoutfs_alloc_has_dirty(struct super_block *sb); -int scoutfs_alloc_dirty_ring(struct super_block *sb); +int scoutfs_alloc_submit_write(struct super_block *sb, + struct scoutfs_bio_completion *comp); +void scoutfs_alloc_write_complete(struct super_block *sb); u64 scoutfs_alloc_bfree(struct super_block *sb); int scoutfs_alloc_setup(struct super_block *sb); diff --git a/kmod/src/compact.c b/kmod/src/compact.c index cce6715f..cae9d7f6 100644 --- a/kmod/src/compact.c +++ b/kmod/src/compact.c @@ -522,11 +522,12 @@ out: /* * 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. + * can use it while we're mucking with it. While the current ring can + * always delete without failure we will probably have a manifest + * storage layer eventually that could return errors on deletion. We + * also also have corrupted something and try to delete an entry that + * doesn't exist. So we use an initial dirtying step to ensure that our + * later deletions succeed. * * XXX does locking the manifest prevent commits? I would think so? */ diff --git a/kmod/src/format.h b/kmod/src/format.h index 0fbe3c40..25cb3878 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -50,43 +50,30 @@ struct scoutfs_block_header { __le64 blkno; } __packed; -struct scoutfs_treap_ref { - __le64 off; - __le64 gen; - __u8 aug_bits; +struct scoutfs_ring_entry { + __le16 data_len; + __u8 flags; + __u8 data[0]; } __packed; -/* - * The lesser and greater bits are persistent on disk so that we can migrate - * nodes from the older half of the ring. - * - * The dirty bit is only used for in-memory nodes. - */ -#define SCOUTFS_TREAP_AUG_LESSER (1 << 0) -#define SCOUTFS_TREAP_AUG_GREATER (1 << 1) -#define SCOUTFS_TREAP_AUG_HALVES (SCOUTFS_TREAP_AUG_LESSER | \ - SCOUTFS_TREAP_AUG_GREATER) -#define SCOUTFS_TREAP_AUG_DIRTY (1 << 2) +#define SCOUTFS_RING_ENTRY_FLAG_DELETION (1 << 0) -/* - * Treap nodes are stored at byte offset in the ring of blocks described - * by the super block. Each reference contains the off and gen that it - * will find in the node for verification. Each node has the header - * and data payload covered by a crc. - */ -struct scoutfs_treap_node { +struct scoutfs_ring_block { __le32 crc; - __le64 off; - __le64 gen; - __le64 prio; - struct scoutfs_treap_ref left; - struct scoutfs_treap_ref right; - __le16 bytes; - u8 data[0]; + __le32 pad; + __le64 fsid; + __le64 seq; + __le64 block; + __le32 nr_entries; + struct scoutfs_ring_entry entries[0]; } __packed; -struct scoutfs_treap_root { - struct scoutfs_treap_ref ref; +struct scoutfs_ring_descriptor { + __le64 blkno; + __le64 total_blocks; + __le64 first_block; + __le64 first_seq; + __le64 nr_blocks; } __packed; /* @@ -98,7 +85,7 @@ struct scoutfs_treap_root { #define SCOUTFS_MANIFEST_FANOUT 10 struct scoutfs_manifest { - struct scoutfs_treap_root root; + struct scoutfs_ring_descriptor ring; __le64 level_counts[SCOUTFS_MANIFEST_MAX_LEVEL]; } __packed; @@ -246,6 +233,7 @@ struct scoutfs_symlink_key { #define SCOUTFS_UUID_BYTES 16 + /* * The ring fields describe the statically allocated ring log. The * head and tail indexes are logical 4k blocks offsets inside the ring. @@ -264,7 +252,7 @@ struct scoutfs_super_block { __le64 ring_tail_block; __le64 ring_gen; __le64 next_seg_seq; - struct scoutfs_treap_root alloc_treap_root; + struct scoutfs_ring_descriptor alloc_ring; struct scoutfs_manifest manifest; } __packed; diff --git a/kmod/src/manifest.c b/kmod/src/manifest.c index fd2aeda8..6c9e72cb 100644 --- a/kmod/src/manifest.c +++ b/kmod/src/manifest.c @@ -20,7 +20,7 @@ #include "kvec.h" #include "seg.h" #include "item.h" -#include "treap.h" +#include "ring.h" #include "cmp.h" #include "compact.h" #include "manifest.h" @@ -29,24 +29,17 @@ #include "scoutfs_trace.h" /* - * Manifest entries are stored as treap nodes in the ring. + * Manifest entries are stored in ring nodes. * * They're sorted first by level then by their first key. This enables * the primary searches based on key value for looking up items in * segments via the manifest. - * - * The treap also supports augmented searches. We get callbacks as the - * tree structure which lets us maintain data in nodes that describe - * subtrees to accelerate searches. We will record the max sequence - * numbers in subtrees for all the seq queries. We'll probably also - * have bits that direct us towards segments that contain deletion items - * for prioritized compaction. */ struct manifest { struct rw_semaphore rwsem; seqcount_t seqcount; - struct scoutfs_treap *treap; + struct scoutfs_ring_info ring; u8 nr_levels; /* calculated on mount, const thereafter */ @@ -81,12 +74,6 @@ struct manifest_ref { struct scoutfs_key_buf *last; }; -struct manifest_fill_args { - struct scoutfs_manifest_entry ment; - struct scoutfs_key_buf *first; - struct scoutfs_key_buf *last; -}; - /* * Seq is only specified for operations that differentiate between * segments with identical items by their sequence number. @@ -174,7 +161,7 @@ static void add_level_count(struct super_block *sb, struct manifest *mani, } /* - * Insert a new manifest entry in the treap. The treap allocates a new + * Insert a new manifest entry in the ring. The ring allocates a new * node for us and we fill it. * * This must be called with the manifest lock held. @@ -188,40 +175,38 @@ int scoutfs_manifest_add(struct super_block *sb, struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; struct scoutfs_manifest_entry *ment; - struct manifest_fill_args args; + struct scoutfs_key_buf ment_first; + struct scoutfs_key_buf ment_last; struct manifest_search_key skey; unsigned key_bytes; unsigned bytes; - int ret; trace_scoutfs_manifest_add(sb, first, last, segno, seq, level); key_bytes = first->key_len + last->key_len; bytes = offsetof(struct scoutfs_manifest_entry, keys[key_bytes]); - args.ment.segno = cpu_to_le64(segno); - args.ment.seq = cpu_to_le64(seq); - args.ment.first_key_len = cpu_to_le16(first->key_len); - args.ment.last_key_len = cpu_to_le16(last->key_len); - args.ment.level = level; - - args.first = first; - args.last = last; - skey.key = first; skey.level = level; skey.seq = seq; - 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); - add_level_count(sb, mani, super, level, 1); - ret = 0; - } + ment = scoutfs_ring_insert(&mani->ring, &skey, bytes); + if (!ment) + return -ENOMEM; - return ret; + ment->segno = cpu_to_le64(segno); + ment->seq = cpu_to_le64(seq); + ment->first_key_len = cpu_to_le16(first->key_len); + ment->last_key_len = cpu_to_le16(last->key_len); + ment->level = level; + + init_ment_keys(ment, &ment_first, &ment_last); + scoutfs_key_copy(&ment_first, first); + scoutfs_key_copy(&ment_last, last); + + mani->nr_levels = max_t(u8, mani->nr_levels, level + 1); + add_level_count(sb, mani, super, level, 1); + return 0; } /* @@ -238,11 +223,11 @@ int scoutfs_manifest_dirty(struct super_block *sb, skey.level = level; skey.seq = seq; - ment = scoutfs_treap_lookup_dirty(mani->treap, &skey); - if (IS_ERR(ment)) - return PTR_ERR(ment); + ment = scoutfs_ring_lookup(&mani->ring, &skey); if (!ment) return -ENOENT; + + scoutfs_ring_dirty(&mani->ring, ment); return 0; } @@ -255,18 +240,20 @@ int scoutfs_manifest_del(struct super_block *sb, struct scoutfs_key_buf *first, 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 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) - add_level_count(sb, mani, super, level, -1ULL); + ment = scoutfs_ring_lookup(&mani->ring, &skey); + if (!ment) + return -ENOENT; - return ret; + scoutfs_ring_delete(&mani->ring, ment); + add_level_count(sb, mani, super, level, -1ULL); + return 0; } /* @@ -363,19 +350,15 @@ static int get_range_refs(struct super_block *sb, struct manifest *mani, /* get level 0 segments that overlap with the missing range */ skey.level = 0; skey.seq = ~0ULL; - ment = scoutfs_treap_lookup_prev(mani->treap, &skey); - while (!IS_ERR_OR_NULL(ment)) { + ment = scoutfs_ring_lookup_prev(&mani->ring, &skey); + while (ment) { if (cmp_range_ment(key, end, ment) == 0) { ret = alloc_add_ref(sb, ref_list, ment); if (ret) goto out; } - ment = scoutfs_treap_prev(mani->treap, ment); - } - if (IS_ERR(ment)) { - ret = PTR_ERR(ment); - goto out; + ment = scoutfs_ring_prev(&mani->ring, ment); } /* get higher level segments that overlap with the starting key */ @@ -386,12 +369,7 @@ static int get_range_refs(struct super_block *sb, struct manifest *mani, /* XXX should use level counts to skip searches */ - ment = scoutfs_treap_lookup(mani->treap, &skey); - if (IS_ERR(ment)) { - ret = PTR_ERR(ment); - goto out; - } - + ment = scoutfs_ring_lookup(&mani->ring, &skey); if (ment) { init_ment_keys(ment, &first, &last); ret = alloc_add_ref(sb, ref_list, ment); @@ -625,28 +603,32 @@ int scoutfs_manifest_has_dirty(struct super_block *sb) int ret; down_write(&mani->rwsem); - ret = scoutfs_treap_has_dirty(mani->treap); + ret = scoutfs_ring_has_dirty(&mani->ring); up_write(&mani->rwsem); return ret; } -/* - * Append the dirty manifest entries to the end of the ring. - * - * This returns 0 but can't fail. - */ -int scoutfs_manifest_dirty_ring(struct super_block *sb) +int scoutfs_manifest_submit_write(struct super_block *sb, + struct scoutfs_bio_completion *comp) { DECLARE_MANIFEST(sb, mani); - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_super_block *super = &sbi->super; + int ret; down_write(&mani->rwsem); - scoutfs_treap_dirty_ring(mani->treap, &super->manifest.root); + ret = scoutfs_ring_submit_write(sb, &mani->ring, comp); up_write(&mani->rwsem); - return 0; + return ret; +} + +void scoutfs_manifest_write_complete(struct super_block *sb) +{ + DECLARE_MANIFEST(sb, mani); + + down_write(&mani->rwsem); + scoutfs_ring_write_complete(&mani->ring); + up_write(&mani->rwsem); } u64 scoutfs_manifest_level_count(struct super_block *sb, u8 level) @@ -714,23 +696,19 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data) /* 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 = scoutfs_ring_first(&mani->ring); + if (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); + ment = scoutfs_ring_lookup_next(&mani->ring, &skey); if (ment == NULL || ment->level != level) { scoutfs_key_set_min(skey.key); - ment = scoutfs_treap_lookup_next(mani->treap, &skey); + ment = scoutfs_ring_lookup_next(&mani->ring, &skey); } } - if (IS_ERR(ment)) { - ret = PTR_ERR(ment); - goto out; - } if (ment == NULL || ment->level != level) { /* XXX shouldn't be possible */ ret = 0; @@ -750,14 +728,10 @@ 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_next(mani->treap, &skey); + over = scoutfs_ring_lookup_next(&mani->ring, &skey); /* and add a fanout's worth of lower overlapping segments */ for (i = 0; i < SCOUTFS_MANIFEST_FANOUT; i++) { - if (IS_ERR(over)) { - ret = PTR_ERR(over); - goto out; - } if (!over || over->level != (ment->level + 1)) break; @@ -773,7 +747,7 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data) if (ret) goto out; - over = scoutfs_treap_next(mani->treap, over); + over = scoutfs_ring_next(&mani->ring, over); } /* record the next key to start from */ @@ -787,7 +761,7 @@ out: } /* - * Manifest entries for all levels are stored in a single treap. + * Manifest entries for all levels are stored in a single ring. * * First they're sorted by their level. * @@ -806,7 +780,7 @@ out: * 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) +static int manifest_ring_compare_key(void *key, void *data) { struct manifest_search_key *skey = key; struct scoutfs_manifest_entry *ment = data; @@ -842,32 +816,27 @@ out: return cmp; } -static void manifest_treap_fill(void *data, void *arg) +static int manifest_ring_compare_data(void *a, void *b) { - struct scoutfs_manifest_entry *ment = data; - struct manifest_fill_args *args = arg; - struct scoutfs_key_buf ment_first; - struct scoutfs_key_buf ment_last; + struct manifest_search_key skey; + struct scoutfs_manifest_entry *ment = a; + struct scoutfs_key_buf key; - *ment = args->ment; + init_ment_keys(ment, &key, NULL); - init_ment_keys(ment, &ment_first, &ment_last); - scoutfs_key_copy(&ment_first, args->first); - scoutfs_key_copy(&ment_last, args->last); + skey.seq = le64_to_cpu(ment->seq); + skey.key = &key; + skey.level = ment->level; + + return manifest_ring_compare_key(&skey, b); } -static struct scoutfs_treap_ops manifest_treap_ops = { - .compare = manifest_treap_compare, - .fill = manifest_treap_fill, - /* update aug when we track left and right max seq */ -}; - - 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); @@ -876,12 +845,13 @@ int scoutfs_manifest_setup(struct super_block *sb) init_rwsem(&mani->rwsem); seqcount_init(&mani->seqcount); - - mani->treap = scoutfs_treap_alloc(sb, &manifest_treap_ops, - &super->manifest.root); - if (!mani->treap) { + scoutfs_ring_init(&mani->ring, &super->manifest.ring, + manifest_ring_compare_key, + manifest_ring_compare_data); + ret = scoutfs_ring_load(sb, &mani->ring); + if (ret) { kfree(mani); - return -ENOMEM; + return ret; } for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++) { @@ -890,7 +860,7 @@ int scoutfs_manifest_setup(struct super_block *sb) if (!mani->compact_keys[i]) { while (--i >= 0) scoutfs_key_free(sb, mani->compact_keys[i]); - scoutfs_treap_free(mani->treap); + scoutfs_ring_destroy(&mani->ring); kfree(mani); return -ENOMEM; } @@ -925,7 +895,7 @@ void scoutfs_manifest_destroy(struct super_block *sb) int i; if (mani) { - scoutfs_treap_free(mani->treap); + scoutfs_ring_destroy(&mani->ring); for (i = 0; i < ARRAY_SIZE(mani->compact_keys); i++) scoutfs_key_free(sb, mani->compact_keys[i]); kfree(mani); diff --git a/kmod/src/manifest.h b/kmod/src/manifest.h index d788aeaf..25cf236a 100644 --- a/kmod/src/manifest.h +++ b/kmod/src/manifest.h @@ -2,6 +2,7 @@ #define _SCOUTFS_MANIFEST_H_ struct scoutfs_key_buf; +struct scoutfs_bio_completion; int scoutfs_manifest_add(struct super_block *sb, struct scoutfs_key_buf *first, @@ -12,7 +13,9 @@ int scoutfs_manifest_dirty(struct super_block *sb, int scoutfs_manifest_del(struct super_block *sb, struct scoutfs_key_buf *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_submit_write(struct super_block *sb, + struct scoutfs_bio_completion *comp); +void scoutfs_manifest_write_complete(struct super_block *sb); int scoutfs_manifest_lock(struct super_block *sb); int scoutfs_manifest_unlock(struct super_block *sb); diff --git a/kmod/src/ring.c b/kmod/src/ring.c new file mode 100644 index 00000000..00809b19 --- /dev/null +++ b/kmod/src/ring.c @@ -0,0 +1,803 @@ +/* + * 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 +#include +#include +#include +#include + +#include "super.h" +#include "format.h" +#include "bio.h" +#include "ring.h" + +/* + * scoutfs stores the persistent indexes for the server in a simple log + * entries in a preallocated ring of blocks. + * + * The index is read from the log and loaded in to an rbtree in memory. + * Callers then lock around operations that work on the rbtrees. Dirty + * and deleted nodes are tracked and are eventually copied to pages that + * are written to the tail of the log. + * + * This has the great benefit of updating an index with very few (often + * one) contiguous block writes with low write amplification. + * + * This has the significant cost of requiring reading the indexes in to + * memory before doing any work and then having to hold them resident. + * This is fine for now but we'll have to address these latency and + * capacity limitations before too long. + * + * Callers are entirely responsible for locking. + */ + +/* + * XXX + * - deletion entries could be smaller if we understood keys + * - shouldn't be too hard to compress + */ + +/* + * @block records the logical ring index of the block that contained the + * node. As we commit a ring update we can look at the clean list to + * find the first block that we have to read out of the ring. This + * helps minimize the active region of the ring. + * + * @in_ring is used to mark nodes that were present in the ring and + * which need deletion entries written to the ring before they can be + * freed. + */ +struct ring_node { + struct rb_node rb_node; + struct list_head head; + u64 block; + + u16 data_len; + + u8 dirty:1, + deleted:1, + in_ring:1; + + /* data is packed but callers perform native long bitops */ + u8 data[0] __aligned(__alignof__(long)); +}; + +static struct ring_node *data_rnode(void *data) +{ + return data ? container_of(data, struct ring_node, data) : NULL; +} + +static void *rnode_data(struct ring_node *rnode) +{ + return rnode ? rnode->data : NULL; +} + +static unsigned total_entry_bytes(unsigned data_len) +{ + return offsetof(struct scoutfs_ring_entry, data[data_len]); +} + +/* + * Each time we mark a node dirty we also dirty the oldest clean entry. + * This ensures that we never overwrite stable data. + * + * Picture a ring of blocks where the first half of the ring is full of + * existing entries. Imagine that we continuously update a set of + * entries that make up a single block. Each new update block + * invalidates the previous update block but it advances through the + * ring while the old entries are sitting idle in the first half. + * Eventually the new update blocks wrap around and clobber the old + * blocks. + * + * Now instead imagine that each time we dirty an entry in this set of + * constantly changing entries that we also go and dirty the earliest + * existing entry in the ring. Now each update is a block of the + * useless updating entries and a block of old entries that have been + * migrated. Each time we write two blocks to the ring we migrate one + * block from the start of the ring. Now by the time we fill the second + * half of the ring we've reclaimed half of the first half of the ring. + * + * So we size the ring to fit 4x the largest possible index. Now we're + * sure that we'll be able to fully migrate the index from the first + * half of the ring into the second half before it wraps around and + * starts overwriting the first. + */ +static void mark_node_dirty(struct scoutfs_ring_info *ring, + struct ring_node *rnode, bool migrate) +{ + struct ring_node *pos; + long total; + + if (!rnode || rnode->dirty) + return; + + list_move_tail(&rnode->head, &ring->dirty_list); + rnode->dirty = 1; + ring->dirty_bytes += total_entry_bytes(rnode->data_len); + + if (migrate) { + total = total_entry_bytes(rnode->data_len); + + list_for_each_entry_safe(rnode, pos, &ring->clean_list, head) { + mark_node_dirty(ring, rnode, false); + total -= total_entry_bytes(rnode->data_len); + if (total < 0) + break; + } + } +} + +static void mark_node_clean(struct scoutfs_ring_info *ring, + struct ring_node *rnode) +{ + if (!rnode || !rnode->dirty) + return; + + list_move_tail(&rnode->head, &ring->clean_list); + rnode->dirty = 0; + ring->dirty_bytes -= total_entry_bytes(rnode->data_len); +} + +static void free_node(struct scoutfs_ring_info *ring, + struct ring_node *rnode) +{ + if (rnode) { + mark_node_clean(ring, rnode); + + if (!list_empty(&rnode->head)) + list_del_init(&rnode->head); + if (!RB_EMPTY_NODE(&rnode->rb_node)) + rb_erase(&rnode->rb_node, &ring->rb_root); + + kfree(rnode); + } +} + +/* + * Walk the tree and return the last node traversed. cmp gives the + * caller the comparison between their key and the returned node. The + * caller can provide either their key or another nodes data to compare + * with during descent. If we're asked to insert we replace any node we + * find in the key's place. + */ +static struct ring_node *ring_rb_walk(struct scoutfs_ring_info *ring, + void *key, void *data, + struct ring_node *ins, + int *cmp) +{ + struct rb_node **node = &ring->rb_root.rb_node; + struct rb_node *parent = NULL; + struct ring_node *found = NULL; + struct ring_node *rnode; + + /* only provide one or the other */ + BUG_ON(!!key == !!data); + + while (*node) { + parent = *node; + rnode = container_of(*node, struct ring_node, rb_node); + + if (key) + *cmp = ring->compare_key(key, &rnode->data); + else + *cmp = ring->compare_data(data, &rnode->data); + + if (*cmp < 0) { + node = &(*node)->rb_left; + } else if (*cmp > 0) { + node = &(*node)->rb_right; + } else { + found = rnode; + break; + } + } + + if (ins) { + if (found) { + rb_replace_node(&found->rb_node, &ins->rb_node, + &ring->rb_root); + RB_CLEAR_NODE(&found->rb_node); + free_node(ring, found); + } else { + rb_link_node(&ins->rb_node, parent, node); + rb_insert_color(&ins->rb_node, &ring->rb_root); + } + found = ins; + } + + return found; +} + +static struct ring_node *ring_rb_entry(struct rb_node *node) +{ + return node ? rb_entry(node, struct ring_node, rb_node) : NULL; +} + +/* return the next node, skipping deleted */ +static struct ring_node *ring_rb_next(struct ring_node *rnode) +{ + do { + if (rnode) + rnode = ring_rb_entry(rb_next(&rnode->rb_node)); + } while (rnode && rnode->deleted); + + return rnode; +} + +/* return the prev node, skipping deleted */ +static struct ring_node *ring_rb_prev(struct ring_node *rnode) +{ + do { + if (rnode) + rnode = ring_rb_entry(rb_prev(&rnode->rb_node)); + } while (rnode && rnode->deleted); + + return rnode; +} + +/* return the first node, skipping deleted */ +static struct ring_node *ring_rb_first(struct scoutfs_ring_info *ring) +{ + struct ring_node *rnode; + + rnode = ring_rb_entry(rb_first(&ring->rb_root)); + if (rnode && rnode->deleted) + rnode = ring_rb_next(rnode); + return rnode; +} + +static struct ring_node *alloc_node(unsigned data_len) +{ + struct ring_node *rnode; + + rnode = kzalloc(offsetof(struct ring_node, data[data_len]), GFP_NOFS); + if (rnode) { + RB_CLEAR_NODE(&rnode->rb_node); + INIT_LIST_HEAD(&rnode->head); + rnode->data_len = data_len; + } + + return rnode; +} + +/* + * Insert a new node. This will replace any existing node which could + * be in any state. + */ +void *scoutfs_ring_insert(struct scoutfs_ring_info *ring, void *key, + unsigned data_len) +{ + struct ring_node *rnode; + int cmp; + + rnode = alloc_node(data_len); + if (!rnode) + return NULL; + + ring_rb_walk(ring, key, NULL, rnode, &cmp); + /* just put it on a list, dirtying moves it to dirty */ + list_add_tail(&rnode->head, &ring->dirty_list); + mark_node_dirty(ring, rnode, true); + + return rnode->data; +} + +void *scoutfs_ring_first(struct scoutfs_ring_info *ring) +{ + return rnode_data(ring_rb_first(ring)); +} + +void *scoutfs_ring_lookup(struct scoutfs_ring_info *ring, void *key) +{ + struct ring_node *rnode; + int cmp; + + rnode = ring_rb_walk(ring, key, NULL, NULL, &cmp); + if (rnode && (cmp || rnode->deleted)) + rnode = NULL; + + return rnode_data(rnode); +} + +void *scoutfs_ring_lookup_next(struct scoutfs_ring_info *ring, void *key) +{ + struct ring_node *rnode; + int cmp; + + rnode = ring_rb_walk(ring, key, NULL, NULL, &cmp); + if (rnode && (cmp > 1 || rnode->deleted)) + rnode = ring_rb_next(rnode); + + return rnode_data(rnode); +} + +void *scoutfs_ring_lookup_prev(struct scoutfs_ring_info *ring, void *key) +{ + struct ring_node *rnode; + int cmp; + + rnode = ring_rb_walk(ring, key, NULL, NULL, &cmp); + if (rnode && (cmp < 1 || rnode->deleted)) + rnode = ring_rb_prev(rnode); + + return rnode_data(rnode); +} + +void *scoutfs_ring_next(struct scoutfs_ring_info *ring, void *data) +{ + return rnode_data(ring_rb_next(data_rnode(data))); +} + +void *scoutfs_ring_prev(struct scoutfs_ring_info *ring, void *data) +{ + return rnode_data(ring_rb_prev(data_rnode(data))); +} + +/* + * Calculate the most blocks we could have to use to store a given number + * of bytes of entries. At worst each block has a header and leaves one + * less than the max manifest entry unused. + */ +static unsigned most_blocks(unsigned long bytes) +{ + unsigned long space; + + space = SCOUTFS_BLOCK_SIZE - + sizeof(struct scoutfs_ring_block) - + (sizeof(struct scoutfs_manifest_entry) + + (2 * SCOUTFS_MAX_KEY_SIZE) - 1); + + return DIV_ROUND_UP(bytes, space); +} + +static u64 wrap_ring_block(struct scoutfs_ring_descriptor *rdesc, u64 block) +{ + if (block >= le64_to_cpu(rdesc->total_blocks)) + block -= le64_to_cpu(rdesc->total_blocks); + + /* XXX callers should have verified on load */ + BUG_ON(block >= le64_to_cpu(rdesc->total_blocks)); + + return block; +} + +static u64 calc_first_dirty_block(struct scoutfs_ring_descriptor *rdesc) +{ + return wrap_ring_block(rdesc, le64_to_cpu(rdesc->first_block) + + le64_to_cpu(rdesc->nr_blocks)); +} + +static __le32 rblk_crc(struct scoutfs_ring_block *rblk) +{ + unsigned long skip = (char *)(&rblk->crc + 1) - (char *)rblk; + + return cpu_to_le32(crc32c(~0, (char *)rblk + skip, + SCOUTFS_BLOCK_SIZE - skip)); +} + +/* + * This is called after the caller has copied all the dirty nodes into + * blocks in pages for writing. We might be able to dirty a few more + * clean nodes to fill up the end of the last dirty block to keep the + * ring blocks densely populated. + */ +static void fill_last_dirty_block(struct scoutfs_ring_info *ring, + unsigned space) +{ + struct ring_node *rnode; + struct ring_node *pos; + unsigned tot; + + list_for_each_entry_safe(rnode, pos, &ring->clean_list, head) { + + tot = total_entry_bytes(rnode->data_len); + if (tot > space) + break; + + mark_node_dirty(ring, rnode, false); + space -= tot; + } +} + +void scoutfs_ring_dirty(struct scoutfs_ring_info *ring, void *data) +{ + struct ring_node *rnode; + + rnode = data_rnode(data); + if (rnode) + mark_node_dirty(ring, rnode, true); +} + +/* + * Delete the given node. This can free the node so the caller cannot + * use the data after calling this. + * + * If the node previously existed in the ring then we have to save it and + * write a deletion entry before freeing it. + */ +void scoutfs_ring_delete(struct scoutfs_ring_info *ring, void *data) +{ + struct ring_node *rnode = data_rnode(data); + + BUG_ON(rnode->deleted); + + if (rnode->in_ring) { + rnode->deleted = 1; + mark_node_dirty(ring, rnode, true); + } else { + free_node(ring, rnode); + } +} + +static struct scoutfs_ring_block *block_in_pages(struct page **pages, + unsigned i) +{ + return page_address(pages[i / SCOUTFS_BLOCKS_PER_PAGE]) + + ((i % SCOUTFS_BLOCKS_PER_PAGE) << SCOUTFS_BLOCK_SHIFT); +} + +static int load_ring_block(struct scoutfs_ring_info *ring, + struct scoutfs_ring_block *rblk) +{ + struct scoutfs_ring_entry *rent; + struct ring_node *rnode; + unsigned data_len; + unsigned i; + int ret = 0; + int cmp; + + rent = rblk->entries; + for (i = 0; i < le32_to_cpu(rblk->nr_entries); i++) { + + /* XXX verify fields? */ + data_len = le16_to_cpu(rent->data_len); + + if (rent->flags & SCOUTFS_RING_ENTRY_FLAG_DELETION) { + rnode = ring_rb_walk(ring, NULL, rent->data, NULL, + &cmp); + if (rnode && cmp == 0) + free_node(ring, rnode); + } else { + rnode = alloc_node(data_len); + if (!rnode) { + ret = -ENOMEM; + break; + } + + rnode->block = le64_to_cpu(rblk->block); + rnode->in_ring = 1; + memcpy(rnode->data, rent->data, data_len); + + ring_rb_walk(ring, NULL, rnode->data, rnode, &cmp); + list_add_tail(&rnode->head, &ring->clean_list); + } + + rent = (void *)&rent->data[data_len]; + } + + return ret; +} + +/* + * Read the ring entries into rb nodes with nice large synchronous reads. + */ +#define LOAD_BYTES (4 * 1024 * 1024) +#define LOAD_BLOCKS DIV_ROUND_UP(LOAD_BYTES, SCOUTFS_BLOCK_SIZE) +#define LOAD_PAGES DIV_ROUND_UP(LOAD_BYTES, PAGE_SIZE) +int scoutfs_ring_load(struct super_block *sb, struct scoutfs_ring_info *ring) +{ + struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super; + struct scoutfs_ring_descriptor *rdesc = ring->rdesc; + struct scoutfs_ring_block *rblk; + struct page **pages; + unsigned read_nr; + unsigned i; + __le32 crc; + u64 block; + u64 total; + u64 seq; + u64 nr; + int ret; + + pages = kcalloc(LOAD_PAGES, sizeof(struct page *), GFP_NOFS); + if (!pages) + return -ENOMEM; + + for (i = 0; i < LOAD_PAGES; i++) { + pages[i] = alloc_page(GFP_NOFS); + if (!pages[i]) { + ret = -ENOMEM; + goto out; + } + } + + block = le64_to_cpu(rdesc->first_block); + seq = le64_to_cpu(rdesc->first_seq); + total = le64_to_cpu(rdesc->total_blocks); + nr = le64_to_cpu(rdesc->nr_blocks); + + while (nr) { + read_nr = min3(nr, (u64)LOAD_BLOCKS, total - block); + + ret = scoutfs_bio_read(sb, pages, le64_to_cpu(rdesc->blkno) + + block, read_nr); + if (ret) + goto out; + + for (i = 0; i < read_nr; i++) { + rblk = block_in_pages(pages, i); + crc = rblk_crc(rblk); + + if (rblk->fsid != super->hdr.fsid || + le64_to_cpu(rblk->block) != (block + i) || + le64_to_cpu(rblk->seq) != (seq + i) || + rblk->crc != crc) { + ret = -EIO; + goto out; + } + + ret = load_ring_block(ring, rblk); + if (ret) + goto out; + } + + block = wrap_ring_block(rdesc, block + read_nr); + seq += read_nr; + nr -= read_nr; + } + ret = 0; + +out: + for (i = 0; pages && i < LOAD_PAGES && pages[i]; i++) + __free_page(pages[i]); + kfree(pages); + + if (ret) + scoutfs_ring_destroy(ring); + + return ret; +} + +static struct ring_node *first_dirty_node(struct scoutfs_ring_info *ring) +{ + return list_first_entry_or_null(&ring->dirty_list, struct ring_node, + head); +} + +static struct ring_node *next_dirty_node(struct scoutfs_ring_info *ring, + struct ring_node *rnode) +{ + if (rnode->head.next == &ring->dirty_list) + return NULL; + + return list_next_entry(rnode, head); +} + +static void ring_free_pages(struct scoutfs_ring_info *ring) +{ + unsigned i; + + if (!ring->pages) + return; + + for (i = 0; i < ring->nr_pages; i++) { + if (ring->pages[i]) + __free_page(ring->pages[i]); + } + + kfree(ring->pages); + + ring->pages = NULL; + ring->nr_pages = 0; +} + +int scoutfs_ring_has_dirty(struct scoutfs_ring_info *ring) +{ + return !!ring->dirty_bytes; +} + +int scoutfs_ring_submit_write(struct super_block *sb, + struct scoutfs_ring_info *ring, + struct scoutfs_bio_completion *comp) +{ + struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super; + struct scoutfs_ring_descriptor *rdesc = ring->rdesc; + struct scoutfs_ring_block *rblk; + struct scoutfs_ring_entry *rent; + struct ring_node *rnode; + struct ring_node *next; + struct page **pages; + unsigned nr_blocks; + unsigned nr_pages; + unsigned i; + u64 blkno; + u64 block; + u64 first; + u64 last; + u64 seq; + u64 nr; + u8 *end; + int ret; + + if (ring->dirty_bytes == 0) + return 0; + + nr_blocks = most_blocks(ring->dirty_bytes); + nr_pages = DIV_ROUND_UP(nr_blocks, SCOUTFS_BLOCKS_PER_PAGE); + + ring_free_pages(ring); + + pages = kcalloc(nr_pages, sizeof(struct page *), GFP_NOFS); + if (!pages) + return -ENOMEM; + + ring->pages = pages; + ring->nr_pages = nr_pages; + + for (i = 0; i < nr_pages; i++) { + pages[i] = alloc_page(GFP_NOFS | __GFP_ZERO); + if (!pages[i]) { + ret = -ENOMEM; + goto out; + } + } + + block = ring->first_dirty_block; + seq = ring->first_dirty_seq; + rnode = first_dirty_node(ring); + + for (i = 0; rnode && i < nr_blocks; i++) { + + rblk = block_in_pages(pages, i); + end = (u8 *)rblk + SCOUTFS_BLOCK_SIZE; + + rblk->fsid = super->hdr.fsid; + rblk->seq = cpu_to_le64(seq); + rblk->block = cpu_to_le64(block); + + rent = rblk->entries; + + while (rnode && &rent->data[rnode->data_len] <= end) { + + rent->data_len = cpu_to_le16(rnode->data_len); + if (rnode->deleted) + rent->flags = SCOUTFS_RING_ENTRY_FLAG_DELETION; + memcpy(rent->data, rnode->data, rnode->data_len); + + le32_add_cpu(&rblk->nr_entries, 1); + + rnode->block = block; + + rent = (void *)&rent->data[le16_to_cpu(rent->data_len)]; + + next = next_dirty_node(ring, rnode); + if (!next) { + fill_last_dirty_block(ring, (char *)end - + (char *)rent); + next = next_dirty_node(ring, rnode); + } + rnode = next; + } + + rblk->crc = rblk_crc(rblk); + + block = wrap_ring_block(rdesc, block + 1); + seq++; + } + + /* update the number of blocks we actually filled */ + nr_blocks = i; + + /* point the descriptor at the new active region of the ring */ + rnode = list_first_entry_or_null(&ring->clean_list, struct ring_node, + head); + if (rnode) + first = rnode->block; + else + first = ring->first_dirty_block; + + last = wrap_ring_block(rdesc, ring->first_dirty_block + nr_blocks); + + if (first < last) + nr = last - first; + else + nr = last + le64_to_cpu(rdesc->total_blocks) - first; + + rdesc->first_block = cpu_to_le64(first); + rdesc->first_seq = cpu_to_le64(ring->first_dirty_seq); + rdesc->nr_blocks = cpu_to_le64(nr); + + /* the contig dirty blocks in pages might wrap around ring */ + blkno = le64_to_cpu(rdesc->blkno) + ring->first_dirty_block; + nr = min_t(u64, nr_blocks, + le64_to_cpu(rdesc->total_blocks) - ring->first_dirty_block); + + scoutfs_bio_submit_comp(sb, WRITE, pages, blkno, nr, comp); + + if (nr != nr_blocks) { + pages += nr / SCOUTFS_BLOCKS_PER_PAGE; + blkno = le64_to_cpu(rdesc->blkno); + nr = nr_blocks - nr; + + scoutfs_bio_submit_comp(sb, WRITE, pages, blkno, nr, comp); + } + + ret = 0; + +out: + if (ret) + ring_free_pages(ring); + + return ret; +} + +void scoutfs_ring_write_complete(struct scoutfs_ring_info *ring) +{ + struct ring_node *rnode; + struct ring_node *pos; + + list_for_each_entry_safe(rnode, pos, &ring->dirty_list, head) { + if (rnode->deleted) { + free_node(ring, rnode); + } else { + mark_node_clean(ring, rnode); + rnode->in_ring = 1; + } + } + + ring_free_pages(ring); + + ring->dirty_bytes = 0; + ring->first_dirty_block = calc_first_dirty_block(ring->rdesc); + ring->first_dirty_seq = le64_to_cpu(ring->rdesc->first_seq) + + le64_to_cpu(ring->rdesc->nr_blocks); +} + +void scoutfs_ring_init(struct scoutfs_ring_info *ring, + struct scoutfs_ring_descriptor *rdesc, + scoutfs_ring_cmp_t compare_key, + scoutfs_ring_cmp_t compare_data) +{ + ring->rdesc = rdesc; + ring->compare_key = compare_key; + ring->compare_data = compare_data; + ring->rb_root = RB_ROOT; + INIT_LIST_HEAD(&ring->clean_list); + INIT_LIST_HEAD(&ring->dirty_list); + ring->dirty_bytes = 0; + ring->first_dirty_block = calc_first_dirty_block(rdesc); + ring->first_dirty_seq = le64_to_cpu(rdesc->first_seq) + + le64_to_cpu(rdesc->nr_blocks); + ring->pages = NULL; + ring->nr_pages = 0; +} + +void scoutfs_ring_destroy(struct scoutfs_ring_info *ring) +{ + struct ring_node *rnode; + struct ring_node *pos; + + /* XXX we don't really have a coherent forced dirty unmount story */ + WARN_ON_ONCE(!list_empty(&ring->dirty_list)); + + list_splice_init(&ring->dirty_list, &ring->clean_list); + + list_for_each_entry_safe(rnode, pos, &ring->clean_list, head) { + list_del_init(&rnode->head); + kfree(rnode); + } + + ring_free_pages(ring); + scoutfs_ring_init(ring, ring->rdesc, ring->compare_key, + ring->compare_data); +} diff --git a/kmod/src/ring.h b/kmod/src/ring.h new file mode 100644 index 00000000..a9341092 --- /dev/null +++ b/kmod/src/ring.h @@ -0,0 +1,55 @@ +#ifndef _SCOUTFS_RING_H_ +#define _SCOUTFS_RING_H_ + +struct scoutfs_bio_completion; + +typedef int (*scoutfs_ring_cmp_t)(void *a, void *b); + +struct scoutfs_ring_info { + struct scoutfs_ring_descriptor *rdesc; + + scoutfs_ring_cmp_t compare_key; + scoutfs_ring_cmp_t compare_data; + + struct rb_root rb_root; + + struct list_head clean_list; + struct list_head dirty_list; + + unsigned long dirty_bytes; + u64 first_dirty_block; + u64 first_dirty_seq; + + struct page **pages; + unsigned long nr_pages; +}; + +void scoutfs_ring_init(struct scoutfs_ring_info *ring, + struct scoutfs_ring_descriptor *rdesc, + scoutfs_ring_cmp_t compare_key, + scoutfs_ring_cmp_t compare_data); + +int scoutfs_ring_load(struct super_block *sb, struct scoutfs_ring_info *ring); + +void *scoutfs_ring_insert(struct scoutfs_ring_info *ring, void *key, + unsigned data_len); + +void *scoutfs_ring_first(struct scoutfs_ring_info *ring); +void *scoutfs_ring_lookup(struct scoutfs_ring_info *ring, void *key); +void *scoutfs_ring_lookup_next(struct scoutfs_ring_info *ring, void *key); +void *scoutfs_ring_lookup_prev(struct scoutfs_ring_info *ring, void *key); + +void *scoutfs_ring_next(struct scoutfs_ring_info *ring, void *rdata); +void *scoutfs_ring_prev(struct scoutfs_ring_info *ring, void *rdata); +void scoutfs_ring_dirty(struct scoutfs_ring_info *ring, void *rdata); +void scoutfs_ring_delete(struct scoutfs_ring_info *ring, void *rdata); + +int scoutfs_ring_has_dirty(struct scoutfs_ring_info *ring); +int scoutfs_ring_submit_write(struct super_block *sb, + struct scoutfs_ring_info *ring, + struct scoutfs_bio_completion *comp); +void scoutfs_ring_write_complete(struct scoutfs_ring_info *ring); + +void scoutfs_ring_destroy(struct scoutfs_ring_info *ring); + +#endif diff --git a/kmod/src/super.c b/kmod/src/super.c index 15c0778e..b15c9339 100644 --- a/kmod/src/super.c +++ b/kmod/src/super.c @@ -32,7 +32,6 @@ #include "seg.h" #include "bio.h" #include "alloc.h" -#include "treap.h" #include "compact.h" #include "data.h" #include "lock.h" @@ -218,8 +217,6 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) scoutfs_item_setup(sb) ?: scoutfs_data_setup(sb) ?: scoutfs_alloc_setup(sb) ?: - scoutfs_treap_setup(sb) ?: -// scoutfs_buddy_setup(sb) ?: scoutfs_compact_setup(sb) ?: scoutfs_setup_trans(sb) ?: scoutfs_lock_setup(sb) ?: @@ -265,7 +262,6 @@ static void scoutfs_kill_sb(struct super_block *sb) scoutfs_item_destroy(sb); scoutfs_alloc_destroy(sb); scoutfs_manifest_destroy(sb); - scoutfs_treap_destroy(sb); scoutfs_seg_destroy(sb); scoutfs_destroy_counters(sb); if (sbi->kset) diff --git a/kmod/src/super.h b/kmod/src/super.h index e4f48514..458345fd 100644 --- a/kmod/src/super.h +++ b/kmod/src/super.h @@ -10,7 +10,6 @@ struct scoutfs_counters; struct item_cache; struct manifest; struct segment_cache; -struct treap_info; struct compact_info; struct data_info; struct lock_info; @@ -27,7 +26,6 @@ struct scoutfs_sb_info { struct item_cache *item_cache; struct segment_cache *segment_cache; struct seg_alloc *seg_alloc; - struct treap_info *treap_info; struct compact_info *compact_info; struct data_info *data_info; diff --git a/kmod/src/trans.c b/kmod/src/trans.c index d596bf68..aa927fb2 100644 --- a/kmod/src/trans.c +++ b/kmod/src/trans.c @@ -25,7 +25,7 @@ #include "manifest.h" #include "seg.h" #include "alloc.h" -#include "treap.h" +#include "ring.h" #include "compact.h" #include "counters.h" #include "scoutfs_trace.h" @@ -115,14 +115,15 @@ void scoutfs_trans_write_func(struct work_struct *work) } if (scoutfs_manifest_has_dirty(sb) || scoutfs_alloc_has_dirty(sb)) { - ret = scoutfs_manifest_dirty_ring(sb) ?: - scoutfs_alloc_dirty_ring(sb) ?: - scoutfs_treap_submit_write(sb, &comp) ?: + ret = scoutfs_manifest_submit_write(sb, &comp) ?: + scoutfs_alloc_submit_write(sb, &comp) ?: scoutfs_bio_wait_comp(sb, &comp) ?: scoutfs_write_dirty_super(sb); if (ret) goto out; + scoutfs_manifest_write_complete(sb); + scoutfs_alloc_write_complete(sb); advance = true; } diff --git a/kmod/src/treap.c b/kmod/src/treap.c deleted file mode 100644 index b6346df3..00000000 --- a/kmod/src/treap.c +++ /dev/null @@ -1,1349 +0,0 @@ -/* - * Copyright (C) 2016 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 -#include -#include -#include -#include -#include - -#include "super.h" -#include "format.h" -#include "kvec.h" -#include "bio.h" -#include "treap.h" -#include "scoutfs_trace.h" - -/* - * scoutfs builds a consistent file system out of segments by describing - * them all with the manifest. Typically the manifest will fit in - * memory but in the pathological case it can be much larger. Our task - * is to index the manifest such that the pathological case is possible - * but the typical case isn't unreasonably penalized by the IO cost of - * maintaining the index. - * - * We chose to index the manifest by storing entries in treap nodes in a - * static ring. Updates are large contiguous writes to the ring with - * low amplification. Incremental updates can similarly read-ahead - * large chunks of the ring. Entirely cold reads end up issuing lots of - * small dependent random IOs. - * - * The nodes in the ring are loaded into native copies in memory. - * Having native allocated nodes lets us do things that would be - * unreasonable if we only traversed persistent structures in cached - * blocks: pointers to nodes in memory instead of indirecting through - * block cache lookups, parent pointers for trivial iteration but which - * would would rule out cow updates, and per-node lru tracking so that - * we can reclaim from the leaves of the tree up to the root without - * false pinning based on which nodes happen to share blocks. - * - * As nodes are modified or inserted they're marked dirty. Eventually - * all the dirty nodes are written to the tail of the ring. We ensure - * that new nodes written at the tail never overwrite old live nodes by - * using a large ring and constantly also migrating old nodes in the - * ring to the tail. - * - * Nodes don't span 4k blocks so there will always be at least a node - * struct's worth of blank space in each block, more typically half the - * average item length, and at worst the max item length. - * - * The tree is augmented to enable searches by more than the primary - * sort keys of the tree. The treap itself maintains augmentation in - * memory to track dirty nodes and in the persistent nodes to track old - * nodes for migration. Callers get callbacks to maintain their own - * augmentation in the node payloads. - * - * Each dirty node gets a generation number that is incremented for each - * version of the tree that is written to the tail of the ring. This - * lets traverse cached nodes without needing strong cache coherence - * with other node writers. With the byte offset and generation of root - * node we can traverse our cached nodes and retry the walk when our - * nodes are stale. - * - * XXX - * - add lru list, nodes to tail during walk, shrink from head - * - stale walking needs work: restart walk, get new root sample - * - lru would need to reclaim nodes orphaned by new root ref walk - */ - -/* - * We preallocate sufficient pages to write all the treap nodes to write - * a transactoin. - * - * XXX Today we only ever write a l0 segment or update the manifest and - * allocator for a single compaction. Those events are *well* less than - * the number of pages that make up a large segment. We'll want this to - * be more careful in the future as we batch up updates from lots of - * writers. - */ -struct treap_info { - /* static, derived from the super */ - u64 last_ring_off; - - /* temporarily assigned to each dirty node */ - u64 dirty_off; - u64 dirty_gen; - - /* used to write nodes to the ring */ - struct page *pages[SCOUTFS_SEGMENT_PAGES]; - u64 pages_off; - u64 ring_off; - unsigned int nr_blocks; - unsigned block_space; -}; - -#define DECLARE_TREAP_INFO(sb, name) \ - struct treap_info *name = SCOUTFS_SB(sb)->treap_info - -struct treap_ref { - struct treap_node *node; - u64 off; - u64 gen; - u8 aug_bits; -}; - -struct scoutfs_treap { - struct super_block *sb; - struct scoutfs_super_block *super; - struct scoutfs_treap_ops *ops; - struct treap_ref root_ref; - bool dirty; - u64 dirty_bytes; -}; - -/* - * The in-memory node differs in that it uses native endian fields, has - * a parent pointer, and (will some day have) an lru for reclaiming from - * the leaves up. - * - * The data is long aligned so that callers can use native longs to - * manipulate bitmaps in the data. - */ -struct treap_node { - u64 off; - u64 gen; - u64 prio; - u16 bytes; - - struct treap_node *parent; - - struct treap_ref left; - struct treap_ref right; - - u8 data[0] __aligned(sizeof(long)); -}; - -#if 0 -static void print_treap_node(struct treap_ref *ref, u64 loc) -{ - struct treap_node *node = ref->node; - - if (!node) - return; - - printk("loc %llx node %p: off %llu gen %llu prio %016llx bytes %u\n", - loc, node, node->off, node->gen, node->prio, node->bytes); - printk(" left: off %llu gen %llu aug %u node %p\n", - node->left.off, node->left.gen, node->left.aug_bits, - node->left.node); - printk(" right: off %llu gen %llu aug %u node %p\n", - node->right.off, node->right.gen, node->right.aug_bits, - node->right.node); - - print_treap_node(&node->left, (loc << 4) | 1); - print_treap_node(&node->right, (loc << 4) | 2); -} -#endif - -static struct treap_ref *parent_ref(struct scoutfs_treap *treap, - struct treap_node *node) -{ - if (!node->parent) - return &treap->root_ref; - if (node->parent->left.node == node) - return &node->parent->left; - return &node->parent->right; -} - -static u8 off_aug_bit(struct scoutfs_treap *treap, u64 off) -{ - u64 blocks = le64_to_cpu(treap->super->ring_blocks); - u64 mid = (blocks << SCOUTFS_BLOCK_SHIFT) / 2; - - return off < mid ? SCOUTFS_TREAP_AUG_LESSER : - SCOUTFS_TREAP_AUG_GREATER; -} - -static u8 old_aug_bit(struct scoutfs_treap *treap) -{ - DECLARE_TREAP_INFO(treap->sb, tinf); - - return off_aug_bit(treap, tinf->dirty_off) ^ SCOUTFS_TREAP_AUG_HALVES; -} - -/* - * Return the aug bits that'll be used to refer to the given node. - * We calculate the bits for the node itself and then or those with the - * bits in its references to its children. - */ -static u8 node_aug_bits(struct scoutfs_treap *treap, struct treap_node *node) -{ - DECLARE_TREAP_INFO(treap->sb, tinf); - - return (node->off == tinf->dirty_off ? SCOUTFS_TREAP_AUG_DIRTY : 0) | - off_aug_bit(treap, node->off) | - node->left.aug_bits | - node->right.aug_bits; -} - -/* - * Update the treap augmentation until its back in sync. We can be - * called with a null node to repair a non-existing parent and we just - * have to clear the root aug_bits in that case. - */ -static void update_internal_aug(struct scoutfs_treap *treap, - struct treap_node *node) -{ - struct treap_ref *ref; - u8 bits; - - if (!node) - treap->root_ref.aug_bits = 0; - - while (node) { - bits = node_aug_bits(treap, node); - ref = parent_ref(treap, node); - trace_printk("node %p bits %x parent %p ref bits %x\n", - node, bits, node->parent, ref->aug_bits); - if (ref->aug_bits == bits) - break; - ref->aug_bits = bits; - node = node->parent; - } -} - -static bool ops_update_aug(struct scoutfs_treap *treap, - struct treap_node *parent, struct treap_node *node) -{ - if (!treap->ops->update_aug) - return false; - - return treap->ops->update_aug(parent->data, parent->left.node == node, - node->data); -} - -/* - * Update the tree's augmentation stored in the data payloads. The caller - * sets the left or right aug in the parent to match the node. - */ -static void update_data_aug(struct scoutfs_treap *treap, - struct treap_node *node) -{ - struct treap_node *parent; - - while (node && (parent = node->parent)) { - if (!ops_update_aug(treap, parent, node)) - break; - node = node->parent; - } -} - -/* - * G G - * | | - * P N - * / -> \ - * N P - * \ / - * - * parent->left = node->right; - * node->right = parent; - * grand->(left|right) = node - * - * The rotation has the following effect on augmentation: - * - parent ref's aug bits have the same population, no change - * - node left's unchanged - * - parent right's unchanged - * - parent's left just set to the node's right - * - node right's recalculated based on parent - */ -static void rotate_right(struct scoutfs_treap *treap, - struct treap_node *parent, struct treap_node *node) -{ - struct treap_ref *grand_ref; - struct treap_node *grand; - - /* get grandparent ref before clobbering parent */ - grand = parent->parent; - if (grand) { - if (grand->left.node == parent) - grand_ref = &grand->left; - else - grand_ref = &grand->right; - } else { - grand_ref = &treap->root_ref; - } - - /* parent rotates down and points to node's child */ - parent->left = node->right; - if (parent->left.node) - parent->left.node->parent = parent; - - /* node rotates up and points to parent */ - node->right.node = parent; - node->right.off = parent->off; - node->right.gen = parent->gen; - node->right.aug_bits = node_aug_bits(treap, parent); - parent->parent = node; - - /* grand parent points to node */ - grand_ref->node = node; - grand_ref->off = node->off; - grand_ref->gen = node->gen; - grand_ref->aug_bits = node_aug_bits(treap, node); - node->parent = grand; - - ops_update_aug(treap, node, parent); -} - -/* see above: swap left/right */ -static void rotate_left(struct scoutfs_treap *treap, - struct treap_node *parent, struct treap_node *node) -{ - struct treap_ref *grand_ref; - struct treap_node *grand; - - grand = parent->parent; - if (grand) { - if (grand->right.node == parent) - grand_ref = &grand->right; - else - grand_ref = &grand->left; - } else { - grand_ref = &treap->root_ref; - } - - parent->right = node->left; - if (parent->right.node) - parent->right.node->parent = parent; - - node->left.node = parent; - node->left.off = parent->off; - node->left.gen = parent->gen; - node->left.aug_bits = node_aug_bits(treap, parent); - parent->parent = node; - - grand_ref->node = node; - grand_ref->off = node->off; - grand_ref->gen = node->gen; - grand_ref->aug_bits = node_aug_bits(treap, node); - node->parent = grand; - - ops_update_aug(treap, node, parent); -} - -/* - * Rebalance the tree by rotating the parent and child as long as the - * child has a higher random priority. - */ -static void rebalance(struct scoutfs_treap *treap, struct treap_node *node) -{ - struct treap_node *parent; - - while (node && (parent = node->parent) && node->prio > parent->prio) { - if (parent->left.node == node) - rotate_right(treap, parent, node); - else - rotate_left(treap, parent, node); - } -} - - -/* - * The caller has mucked with a node. We make sure all of our internal - * augmentation, the op data's augmentation, and the treap prio balance - * is repaired. - */ -static void repair(struct scoutfs_treap *treap, struct treap_node *node) -{ - update_internal_aug(treap, node); - update_data_aug(treap, node); - rebalance(treap, node); - - trace_printk("treap %p root aug %x\n", - treap, treap->root_ref.aug_bits); -} - -static struct treap_node *alloc_node(u16 bytes) -{ - struct treap_node *node; - - node = kmalloc(offsetof(struct treap_node, data[bytes]), GFP_NOFS); - if (node) - memset(node, 0, offsetof(struct treap_node, data)); - - return node; -} - -/* - * bytes in the persistent ring taken up by a node with the given number - * of data bytes. - */ -static unsigned node_ring_bytes(struct treap_node *node) -{ - return offsetof(struct scoutfs_treap_node, data[node->bytes]); -} - -static bool dirty_node(struct scoutfs_treap *treap, struct treap_node *node) -{ - DECLARE_TREAP_INFO(treap->sb, tinf); - - return node->off == tinf->dirty_off; -} - -/* - * Ensure that the given node is dirty. If it isn't we need to mark it - * dirty and augment the tree. Transaction limits and preallocation - * make sure that we always have resources to write nodes that are - * dirtied. - * - * When we dirty old nodes we temporarily set their offset to the - * current half of the ring so that they won't show up in augmented - * searches for old nodes. - */ -static bool mark_node_dirty(struct scoutfs_treap *treap, struct treap_ref *ref, - struct treap_node *node) -{ - DECLARE_TREAP_INFO(treap->sb, tinf); - - if (dirty_node(treap, node)) - return false; - - trace_printk("node %p off %llu gen %llu now dirty\n", - node, node->off, node->gen); - - treap->dirty_bytes += node_ring_bytes(node); - treap->dirty = true; - - node->off = tinf->dirty_off; - node->gen = tinf->dirty_gen; - ref->off = node->off; - ref->gen = node->gen; - repair(treap, node); - - return true; -} - -static int dirty_old_nodes(struct scoutfs_treap *treap, unsigned old_target, - unsigned dirty_limit); - -static struct scoutfs_treap_node *read_ring_node(struct scoutfs_treap *treap, - u64 off) -{ - struct address_space *mapping = treap->sb->s_bdev->bd_inode->i_mapping; - struct scoutfs_treap_node *tnode = NULL; - struct page *page = NULL; - unsigned pg_off; - unsigned bytes; - pgoff_t pg_ind; - int ret; - - off += le64_to_cpu(treap->super->ring_blkno) << SCOUTFS_BLOCK_SHIFT; - pg_ind = off >> PAGE_CACHE_SHIFT; - pg_off = off & ~PAGE_CACHE_MASK; - - if (pg_off + sizeof(struct scoutfs_treap_node) > PAGE_CACHE_SIZE) { - ret = -EIO; - goto out; - } - -retry: - page = find_or_create_page(mapping, pg_ind, GFP_NOFS); - if (!page) { - ret = -ENOMEM; - goto out; - } - - tnode = page_address(page) + pg_off; - - if (PageUptodate(page)) { - unlock_page(page); - ret = 0; - goto out; - } - - ClearPageError(page); - ret = mapping->a_ops->readpage(NULL, page); - if (ret) { - if (ret == AOP_TRUNCATED_PAGE) { - page_cache_release(page); - goto retry; - } - goto out; - } - - wait_on_page_locked(page); - if (!PageUptodate(page)) { - if (page->mapping != mapping) { - page_cache_release(page); - goto retry; - } - ret = -EIO; - goto out; - } else { - ret = 0; - } - - bytes = le16_to_cpu(tnode->bytes); - - if (pg_off + offsetof(struct scoutfs_treap_node, data[bytes]) > - PAGE_CACHE_SIZE) { - ret = -EIO; - } - -out: - if (ret) { - if (page) - page_cache_release(page); - return ERR_PTR(ret); - } - - return tnode; -} - -static void release_ring_node(struct scoutfs_treap_node *tnode) -{ - if (!IS_ERR_OR_NULL(tnode)) - page_cache_release(virt_to_page(tnode)); -} - -/* - * We write to ring blocks from preallocated private pages with bios but read - * through the bdev page cache. Invalidate the blocks we're about to write - * so we'll read them later. - */ -static void invalidate_blocks(struct super_block *sb, u64 blkno, u64 nr) -{ - struct address_space *mapping = sb->s_bdev->bd_inode->i_mapping; - loff_t lstart = blkno << SCOUTFS_BLOCK_SHIFT; - loff_t lend = lstart + (nr << SCOUTFS_BLOCK_SHIFT) - 1; - - truncate_inode_pages_range(mapping, lstart, lend); -} - -static void invalidate_ring_block(struct scoutfs_treap *treap, u64 off) -{ - invalidate_blocks(treap->sb, le64_to_cpu(treap->super->ring_blkno) + - (off >> SCOUTFS_BLOCK_SHIFT), 1); -} - -static __le32 tnode_crc(struct scoutfs_treap_node *tnode) -{ - u16 bytes = le16_to_cpu(tnode->bytes); - unsigned skip = sizeof(tnode->crc); - - return cpu_to_le32(crc32c(~0, (void *)tnode + skip, - offsetof(struct scoutfs_treap_node, - data[bytes]) - skip)); -} - -/* - * Give the caller the node pointed to by their reference. If the node - * isn't already in the tree then we link it in and update augmentation. - * - * XXX what's the consequence of failing to also dirty old ring nodes? - * The ring gets out of balance but we do nothing about it. - */ -static struct treap_node *read_node(struct scoutfs_treap *treap, - struct treap_node *parent, - struct treap_ref *ref, bool dirty) -{ - struct scoutfs_treap_node *tnode = NULL; - struct treap_node *node = NULL; - unsigned retries = 3; - u16 bytes; - int ret; - - if (ref->node) { - node = ref->node; - ret = 0; - goto out; - } - -retry: - tnode = read_ring_node(treap, ref->off); - if (IS_ERR(tnode)) { - ret = PTR_ERR(tnode); - goto out; - } - - if (tnode->crc != tnode_crc(tnode) || - le64_to_cpu(tnode->off) != ref->off || - le64_to_cpu(tnode->gen) != ref->gen) { - invalidate_ring_block(treap, ref->off); - if (retries--) { - /* XXX restart search, not just this read */ - release_ring_node(tnode); - goto retry; - } else { - ret = -EIO; - goto out; - } - } - - bytes = le16_to_cpu(tnode->bytes); - - node = alloc_node(bytes); - if (!node) { - ret = -ENOMEM; - goto out; - } - - node->off = le64_to_cpu(tnode->off); - node->gen = le64_to_cpu(tnode->gen); - node->prio = le64_to_cpu(tnode->prio); - node->left.off = le64_to_cpu(tnode->left.off); - node->left.gen = le64_to_cpu(tnode->left.gen); - node->left.aug_bits = tnode->left.aug_bits; - node->right.off = le64_to_cpu(tnode->right.off); - node->right.gen = le64_to_cpu(tnode->right.gen); - node->right.aug_bits = tnode->right.aug_bits; - node->bytes = bytes; - memcpy(node->data, tnode->data, bytes); - - node->parent = parent; - ref->node = node; - ret = 0; -out: - release_ring_node(tnode); - if (!ret && dirty && mark_node_dirty(treap, ref, node)) - ret = dirty_old_nodes(treap, node_ring_bytes(node), 0); - if (ret) - return ERR_PTR(ret); - - return node; -} - -/* - * Find nodes in the older half of the ring and mark them dirty. Stop - * when we don't have any more older nodes, after dirtying enough old - * nodes, or before dirtying too many nodes. - */ -static int dirty_old_nodes(struct scoutfs_treap *treap, unsigned old_target, - unsigned dirty_limit) -{ - u8 bit = old_aug_bit(treap); - struct treap_node *parent; - struct treap_node *node; - struct treap_ref *ref; - unsigned dirty = 0; - unsigned old = 0; - unsigned bytes; - int ret = 0; - -restart: - parent = NULL; - ref = &treap->root_ref; - - while (ref->aug_bits & bit) { - node = read_node(treap, parent, ref, false); - if (IS_ERR(node)) { - ret = PTR_ERR(node); - break; - } - - bytes = node_ring_bytes(node); - - if (!dirty_node(treap, node) && dirty_limit) { - dirty += bytes; - if (dirty > dirty_limit) - break; - } - - if (old_target && off_aug_bit(treap, node->off) == bit) - old += bytes; - - /* sets dirty, sets current half aug bit, repairs */ - mark_node_dirty(treap, ref, node); - - if (old_target && old >= old_target) - break; - - if (node->left.aug_bits & bit) - ref = &node->left; - else if (node->right.aug_bits & bit) - ref = &node->right; - else - goto restart; - } - - return ret; -} - -/* - * Return the dirty node identified by the given key, creating it if it - * doesn't exist. - * - * Returns ERR -EEXIST if a node already exists at the given key. - */ -void *scoutfs_treap_insert(struct scoutfs_treap *treap, void *key, u16 bytes, - void *fill_arg) -{ - struct treap_ref *ref = &treap->root_ref; - struct treap_node *parent = NULL; - struct treap_node *node = NULL; - int cmp; - - while (ref->gen) { - node = read_node(treap, parent, ref, true); - if (IS_ERR(node)) - goto out; - - cmp = treap->ops->compare(key, node->data); - if (cmp < 0) { - ref = &node->left; - } else if (cmp > 0) { - ref = &node->right; - } else { - node = ERR_PTR(-EEXIST); - goto out; - } - - parent = node; - node = NULL; - } - - node = alloc_node(bytes); - if (!node) { - node = ERR_PTR(-ENOMEM); - goto out; - } - - node->parent = parent; - node->bytes = bytes; - get_random_bytes_arch(&node->prio, sizeof(node->prio)); - - ref->node = node; - - /* filling here instead of in caller for aug update in repair */ - treap->ops->fill(node->data, fill_arg); - - /* sets off and gen and repairs */ - mark_node_dirty(treap, ref, node); -out: - if (IS_ERR(node)) - return ERR_CAST(node); - - return node->data; -} - -/* - * Delete a node with the given key. - * - * It's easy when the node doesn't have two children. We remove the - * node and point it's parent ref at either of the child's refs that - * might have been populated. - * - * Deletion's a little tricker when we have both children. We could - * find an ancestor and swap but that's fiddly to get right with all our - * rich node pointers. Instead we can reuse rotation to rotate the node - * down until it doesn't have both children. - */ -int scoutfs_treap_delete(struct scoutfs_treap *treap, void *key) -{ - struct treap_ref *ref = &treap->root_ref; - struct treap_node *parent = NULL; - struct treap_node *node = NULL; - struct treap_ref *child_ref; - struct treap_node *left; - struct treap_node *right; - int cmp; - int ret; - - /* find node to delete */ - while (ref->gen) { - node = read_node(treap, parent, ref, true); - if (IS_ERR(node)) { - ret = PTR_ERR(node); - goto out; - } - - cmp = treap->ops->compare(key, node->data); - if (cmp < 0) - ref = &node->left; - else if (cmp > 0) - ref = &node->right; - else - break; - - parent = node; - node = NULL; - } - - if (!node) { - ret = -ENOENT; - goto out; - } - - /* - * Rotate the node down with its higher priority child until it - * doesn't have both children. Dirtying tries to repair which - * can try to repair priority imbalance with rotation so we swap - * priorities first. Unfortunately we need to read both - * children to get their priorities but we only try to dirty the - * rotation child. It's messy but dirtying both can double - * write amplification. - */ - while (node->left.gen && node->right.gen) { - left = read_node(treap, node, &node->left, false); - right = read_node(treap, node, &node->right, false); - if (IS_ERR(left) || IS_ERR(right)) { - ret = IS_ERR(left) ? PTR_ERR(left) : PTR_ERR(right); - goto out; - } - - if (left->prio > right->prio) { - left = read_node(treap, node, &node->left, true); - if (IS_ERR(left)) { - ret = IS_ERR(left); - goto out; - } - swap(node->prio, left->prio); - rotate_right(treap, node, left); - } else { - right = read_node(treap, node, &node->right, true); - if (IS_ERR(right)) { - ret = IS_ERR(right); - goto out; - } - swap(node->prio, right->prio); - rotate_left(treap, node, right); - } - - parent = node->parent; - ref = parent_ref(treap, node); - } - - /* delete the node, might have to point parent at child */ - if (node->left.gen) - child_ref = &node->left; - else - child_ref = &node->right; - - *ref = *child_ref; - if (ref->node) - ref->node->parent = parent; - - if (dirty_node(treap, node)) - treap->dirty_bytes -= node_ring_bytes(node); - - kfree(node); - - repair(treap, parent); - ret = 0; -out: - return ret; -} - -enum { - LU_DIRTY, - LU_NEXT, - LU_PREV, -}; - -static void *treap_lookup(struct scoutfs_treap *treap, void *key, int flags) -{ - struct treap_ref *ref = &treap->root_ref; - struct treap_node *parent = NULL; - struct treap_node *node = NULL; - struct treap_node *prev = NULL; - struct treap_node *next = NULL; - int cmp; - - while (ref->gen) { - node = read_node(treap, parent, ref, flags & LU_DIRTY); - if (IS_ERR(node)) - break; - - cmp = treap->ops->compare(key, node->data); - if (cmp < 0) { - ref = &node->left; - next = node; - } else if (cmp > 0) { - ref = &node->right; - prev = node; - } else { - break; - } - - parent = node; - node = NULL; - } - - if (!node && (flags & LU_PREV) && prev) - node = prev; - else if (!node && (flags & LU_NEXT) && next) - node = next; - - if (IS_ERR(node)) - return ERR_CAST(node); - if (node) - return node->data; - return NULL; -} - -void *scoutfs_treap_lookup(struct scoutfs_treap *treap, void *key) -{ - return treap_lookup(treap, key, 0); -} - -void *scoutfs_treap_lookup_dirty(struct scoutfs_treap *treap, void *key) -{ - return treap_lookup(treap, key, LU_DIRTY); -} - -void *scoutfs_treap_lookup_next(struct scoutfs_treap *treap, void *key) -{ - return treap_lookup(treap, key, LU_NEXT); -} - -void *scoutfs_treap_lookup_next_dirty(struct scoutfs_treap *treap, void *key) -{ - return treap_lookup(treap, key, LU_NEXT | LU_DIRTY); -} - -void *scoutfs_treap_lookup_prev(struct scoutfs_treap *treap, void *key) -{ - return treap_lookup(treap, key, LU_PREV); -} - -void *scoutfs_treap_lookup_prev_dirty(struct scoutfs_treap *treap, void *key) -{ - return treap_lookup(treap, key, LU_PREV | LU_DIRTY); -} - -void *scoutfs_treap_first(struct scoutfs_treap *treap) -{ - struct treap_ref *ref = &treap->root_ref; - struct treap_node *parent = NULL; - struct treap_node *node = NULL; - - while (ref->gen) { - node = read_node(treap, parent, ref, false); - if (IS_ERR(node)) - break; - - ref = &node->left; - parent = node; - } - - if (IS_ERR(node)) - return ERR_CAST(node); - if (node) - return node->data; - return NULL; -} - -void *scoutfs_treap_last(struct scoutfs_treap *treap) -{ - struct treap_ref *ref = &treap->root_ref; - struct treap_node *parent = NULL; - struct treap_node *node = NULL; - - while (ref->gen) { - node = read_node(treap, parent, ref, false); - if (IS_ERR(node)) - break; - - ref = &node->right; - parent = node; - } - - if (IS_ERR(node)) - return ERR_CAST(node); - if (node) - return node->data; - return NULL; -} - -void *scoutfs_treap_next(struct scoutfs_treap *treap, void *data) -{ - struct treap_node *node = container_of(data, struct treap_node, data); - struct treap_node *parent; - - if (node->right.gen) { - node = read_node(treap, node, &node->right, false); - if (IS_ERR(node)) - goto out; - - while (node->left.gen) { - node = read_node(treap, node, &node->left, false); - if (IS_ERR(node)) - goto out; - } - - goto out; - } - - while (((parent = node->parent)) && node == parent->right.node) - node = parent; - node = parent; - -out: - if (IS_ERR(node)) - return ERR_CAST(node); - if (node) - return node->data; - return NULL; -} - -void *scoutfs_treap_prev(struct scoutfs_treap *treap, void *data) -{ - struct treap_node *node = container_of(data, struct treap_node, data); - struct treap_node *parent; - - if (node->left.gen) { - node = read_node(treap, node, &node->left, false); - if (IS_ERR(node)) - goto out; - - while (node->right.gen) { - node = read_node(treap, node, &node->right, false); - if (IS_ERR(node)) - goto out; - } - - goto out; - } - - while (((parent = node->parent)) && node == parent->left.node) - node = parent; - node = parent; - -out: - if (IS_ERR(node)) - return ERR_CAST(node); - if (node) - return node->data; - return NULL; -} - -int scoutfs_treap_has_dirty(struct scoutfs_treap *treap) -{ - return treap->dirty; -} - -static void *pages_off_ptr(struct treap_info *tinf) -{ - return page_address(tinf->pages[tinf->pages_off >> PAGE_SHIFT]) + - (tinf->pages_off % ~PAGE_MASK); -} - -/* - * The dirty offset is carefully chosen so that it will consider dirty - * nodes part of the current half of the ring but is an offset that will - * never be actually written. That way it is overwritten as dirty nodes - * are copied to the ring and get their final offset and aren't considered - * dirty. Nodes never span blocks so we set the dirty offset to the final - * byte of the next block in the ring. - */ -static void init_writer(struct treap_info *tinf, - struct scoutfs_super_block *super) -{ - tinf->ring_off = le64_to_cpu(super->ring_tail_block) << - SCOUTFS_BLOCK_SHIFT; - tinf->pages_off = 0; - tinf->block_space = 0; - tinf->nr_blocks = 0; - - tinf->dirty_gen = le64_to_cpu(super->ring_gen) + 1; - tinf->dirty_off = tinf->ring_off + SCOUTFS_BLOCK_MASK; -} - -static void try_zero_block_tail(struct treap_info *tinf) -{ - if (tinf->block_space != SCOUTFS_BLOCK_SIZE) - memset(pages_off_ptr(tinf), 0, tinf->block_space); -} - -/* - * Copy the node to the page at the next free tail offset. The - * in-memory node's offset is set to its final ring offset and its - * parent ref is updated. Thus it will no longer have the magic dirty - * offset and won't be considered dirty by the tree augmentation. - */ -static void copy_node_to_ring(struct scoutfs_treap *treap, - struct treap_node *node) -{ - DECLARE_TREAP_INFO(treap->sb, tinf); - struct scoutfs_treap_node *tnode; - u32 bytes = node_ring_bytes(node); - u32 skip; - - if (tinf->block_space < bytes) { - try_zero_block_tail(tinf); - - skip = ALIGN(tinf->ring_off, SCOUTFS_BLOCK_SIZE) - - tinf->ring_off; - tinf->ring_off += skip; - tinf->pages_off += skip; - - tinf->block_space = SCOUTFS_BLOCK_SIZE; - tinf->nr_blocks++; - - /* see if we're wrapping */ - if (tinf->ring_off == tinf->last_ring_off) - tinf->ring_off = 0; - } - - node->off = tinf->ring_off; - parent_ref(treap, node)->off = node->off; - - tnode = pages_off_ptr(tinf); - tinf->ring_off += bytes; - tinf->pages_off += bytes; - tinf->block_space -= bytes; - - tnode->off = cpu_to_le64(node->off); - tnode->gen = cpu_to_le64(node->gen); - tnode->prio = cpu_to_le64(node->prio); - tnode->left.off = cpu_to_le64(node->left.off); - tnode->left.gen = cpu_to_le64(node->left.gen); - tnode->left.aug_bits = node->left.aug_bits; - tnode->right.off = cpu_to_le64(node->right.off); - tnode->right.gen = cpu_to_le64(node->right.gen); - tnode->right.aug_bits = node->right.aug_bits; - tnode->bytes = cpu_to_le16(node->bytes); - memcpy(tnode->data, node->data, node->bytes); - - tnode->crc = tnode_crc(tnode); -} - -/* - * Copy the currently dirty nodes into preallocated pages for writing. - * - * We can consider the nodes clean as we copy them to the pages. The - * caller is responsible for ensuring forward progress or aborting. - * - * As nodes are copied to the pages they are assigned their final offset - * in the ring. We have to update their parent refs with the new - * offset. (We also could have them cross a half ring, getting new off - * aug bits that bubble up). - * - * All that means that we copy from the leaves up to the root so that we - * capture the modifications to parents as we copy children. - * - * This is called for multiple treaps before the ring is written. - */ -int scoutfs_treap_dirty_ring(struct scoutfs_treap *treap, - struct scoutfs_treap_root *root) -{ - struct treap_node *node; - unsigned bytes; - int ret; - - /* first fill final partial block with old nodes */ - bytes = SCOUTFS_BLOCK_SIZE - (treap->dirty_bytes & SCOUTFS_BLOCK_MASK); - if (bytes != SCOUTFS_BLOCK_SIZE) { - ret = dirty_old_nodes(treap, 0, bytes); - if (ret) - goto out; - } - - node = treap->root_ref.node; - while (node) { - /* follow dirty links first */ - if (node->left.aug_bits & SCOUTFS_TREAP_AUG_DIRTY) { - node = node->left.node; - } else if (node->right.aug_bits & SCOUTFS_TREAP_AUG_DIRTY) { - node = node->right.node; - } else { - /* node doesn't have dirty children, append if dirty */ - if (dirty_node(treap, node)) { - copy_node_to_ring(treap, node); - repair(treap, node); - } - - /* ascend back up through parents */ - node = node->parent; - } - } - - /* point the persistent super root at the treap we wrote to the ring */ - root->ref.off = cpu_to_le64(treap->root_ref.off); - root->ref.gen = cpu_to_le64(treap->root_ref.gen); - root->ref.aug_bits = treap->root_ref.aug_bits; - - treap->dirty_bytes = 0; - treap->dirty = false; - ret = 0; -out: - return ret; -} - -/* - * Submit writes for all the dirty nodes that have been copied into the - * preallocated pages. - * entries were appended. The dirty ring blocks are contiguous in the - * page array but can wrap in the block ring on disk. - * - * If it wraps then we submit the earlier fragment at the head of the - * ring first. - * - * The wrapped fragment starts at some block offset in the page array. - * The hacky page array math only works when our fixed 4k block size == - * page_size. To fix it we'd add a offset block to the bio submit loop - * which could add an initial partial page vec to the bios. - * - * XXX figure out where to write. I guess we have a write ring block - * in the super? - */ -int scoutfs_treap_submit_write(struct super_block *sb, - struct scoutfs_bio_completion *comp) -{ - struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super; - DECLARE_TREAP_INFO(sb, tinf); - u64 head_blocks; - u64 tail_blocks; - u64 blkno; - u64 tail; - - if (!tinf->nr_blocks) - return 0; - - try_zero_block_tail(tinf); - - tail = le64_to_cpu(super->ring_tail_block); - tail_blocks = min_t(u64, tinf->nr_blocks, - le64_to_cpu(super->ring_blocks) - tail); - - head_blocks = tinf->nr_blocks - tail_blocks; - - if (head_blocks) { - BUILD_BUG_ON(SCOUTFS_BLOCK_SIZE != PAGE_SIZE); - invalidate_blocks(sb, le64_to_cpu(super->ring_blkno), - head_blocks); - scoutfs_bio_submit_comp(sb, WRITE, tinf->pages + tail_blocks, - le64_to_cpu(super->ring_blkno), - head_blocks, comp); - } - - blkno = le64_to_cpu(super->ring_blkno) + tail; - invalidate_blocks(sb, blkno, tail_blocks); - scoutfs_bio_submit_comp(sb, WRITE, tinf->pages, blkno, tail_blocks, - comp); - - /* record new tail index in super and reset for next trans */ - super->ring_tail_block = cpu_to_le64(tail + tail_blocks); - if (super->ring_tail_block == super->ring_blocks) - super->ring_tail_block = cpu_to_le64(head_blocks); - - super->ring_gen = cpu_to_le64(tinf->dirty_gen); - - init_writer(tinf, super); - - return 0; -} - -struct scoutfs_treap *scoutfs_treap_alloc(struct super_block *sb, - struct scoutfs_treap_ops *ops, - struct scoutfs_treap_root *root) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_treap *treap; - - treap = kzalloc(sizeof(struct scoutfs_treap), GFP_NOFS); - if (treap) { - treap->sb = sb; - treap->super = &sbi->super; - treap->ops = ops; - treap->root_ref.off = le64_to_cpu(root->ref.off); - treap->root_ref.gen = le64_to_cpu(root->ref.gen); - treap->root_ref.aug_bits = root->ref.aug_bits; - } - - return treap; -} - -/* - * Free all the allocated nodes in the treap and clear the root. - */ -void scoutfs_treap_free(struct scoutfs_treap *treap) -{ - struct treap_node *node = treap->root_ref.node; - struct treap_node *fre; - - while (node) { - if (node->left.node) { - node = node->left.node; - node->parent->left.node = NULL; - } if (node->right.node) { - node = node->right.node; - node->parent->right.node = NULL; - } else { - fre = node; - node = node->parent; - kfree(fre); - } - } - - kfree(treap); -} - -int scoutfs_treap_setup(struct super_block *sb) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_super_block *super = &sbi->super; - struct treap_info *tinf; - struct page *page; - int i; - - BUILD_BUG_ON(offsetof(struct treap_node, data) & (sizeof(long) - 1)); - - tinf = kzalloc(sizeof(struct treap_info), GFP_KERNEL); - if (!tinf) - return -ENOMEM; - - tinf->last_ring_off = le64_to_cpu(super->ring_blocks) << - SCOUTFS_BLOCK_SHIFT; - init_writer(tinf, super); - - for (i = 0; i < ARRAY_SIZE(tinf->pages); i++) { - page = alloc_page(GFP_KERNEL); - if (!page) { - while (--i >= 0) - __free_page(tinf->pages[i]); - kfree(tinf); - return -ENOMEM; - } - - tinf->pages[i] = page; - } - - sbi->treap_info = tinf; - - return 0; -} - -void scoutfs_treap_destroy(struct super_block *sb) -{ - DECLARE_TREAP_INFO(sb, tinf); - int i; - - if (tinf) { - for (i = 0; i < ARRAY_SIZE(tinf->pages); i++) - __free_page(tinf->pages[i]); - - kfree(tinf); - } -} diff --git a/kmod/src/treap.h b/kmod/src/treap.h deleted file mode 100644 index 497d742a..00000000 --- a/kmod/src/treap.h +++ /dev/null @@ -1,47 +0,0 @@ -#ifndef _SCOUTFS_TREAP_H_ -#define _SCOUTFS_TREAP_H_ - -struct scoutfs_bio_completion; - -/* - * The runtime root that's used by operations. It's loaded and stored - * from the persistent root in the super block as transactions are written. - */ -struct scoutfs_treap; - -struct scoutfs_treap_ops { - int (*compare)(void *key, void *data); - void (*fill)(void *data, void *fill_arg); - bool (*update_aug)(void *parent_data, bool left, void *node_data); -}; - -struct scoutfs_treap *scoutfs_treap_alloc(struct super_block *sb, - struct scoutfs_treap_ops *ops, - struct scoutfs_treap_root *root); -void scoutfs_treap_free(struct scoutfs_treap *treap); - -void *scoutfs_treap_insert(struct scoutfs_treap *treap, void *key, u16 bytes, - void *fill_arg); -int scoutfs_treap_delete(struct scoutfs_treap *treap, void *key); -void *scoutfs_treap_lookup(struct scoutfs_treap *treap, void *key); -void *scoutfs_treap_lookup_dirty(struct scoutfs_treap *treap, void *key); -void *scoutfs_treap_lookup_next(struct scoutfs_treap *treap, void *key); -void *scoutfs_treap_lookup_next_dirty(struct scoutfs_treap *treap, void *key); -void *scoutfs_treap_lookup_prev(struct scoutfs_treap *treap, void *key); -void *scoutfs_treap_lookup_prev_dirty(struct scoutfs_treap *treap, void *key); - -void *scoutfs_treap_first(struct scoutfs_treap *treap); -void *scoutfs_treap_last(struct scoutfs_treap *treap); -void *scoutfs_treap_next(struct scoutfs_treap *treap, void *data); -void *scoutfs_treap_prev(struct scoutfs_treap *treap, void *data); - -int scoutfs_treap_has_dirty(struct scoutfs_treap *treap); -int scoutfs_treap_dirty_ring(struct scoutfs_treap *treap, - struct scoutfs_treap_root *root); -int scoutfs_treap_submit_write(struct super_block *sb, - struct scoutfs_bio_completion *comp); - -int scoutfs_treap_setup(struct super_block *sb); -void scoutfs_treap_destroy(struct super_block *sb); - -#endif