From db9f2be728113f5fd19d05de180f41937dc7f7fd Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Thu, 29 Dec 2016 16:06:12 -0800 Subject: [PATCH] Switch to indexed manifest using treap ring The first pass manifest and allocator storage used a simple ring log that was entirely replayed into memory to be used. That risked the manifest being too large to fit in memory, especially with large keys and large volumes. So we move to using an indexed persistent structure that can be read on demand and cached. We use a treap of byte referenced nodoes stored in a circular ring. The code interface is modeled a bit on the in-memory rbtree interface. Except that we can get IO errors and manage allocation so we return data pointers to the item payload istead of item structs and we can return errors. The manifest and allocator are converted over and the old ring code is removed entirely. Signed-off-by: Zach Brown --- kmod/src/Makefile | 2 +- kmod/src/alloc.c | 398 ++++++------ kmod/src/alloc.h | 3 +- kmod/src/format.h | 82 ++- kmod/src/manifest.c | 536 ++++++++-------- kmod/src/manifest.h | 3 +- kmod/src/ring.c | 326 ---------- kmod/src/ring.h | 18 - kmod/src/scoutfs_trace.h | 10 +- kmod/src/seg.c | 2 +- kmod/src/super.c | 7 +- kmod/src/super.h | 4 +- kmod/src/trans.c | 15 +- kmod/src/treap.c | 1271 ++++++++++++++++++++++++++++++++++++++ kmod/src/treap.h | 44 ++ 15 files changed, 1848 insertions(+), 873 deletions(-) delete mode 100644 kmod/src/ring.c delete mode 100644 kmod/src/ring.h create mode 100644 kmod/src/treap.c create mode 100644 kmod/src/treap.h diff --git a/kmod/src/Makefile b/kmod/src/Makefile index bffe9ec6..cb9b29c9 100644 --- a/kmod/src/Makefile +++ b/kmod/src/Makefile @@ -4,4 +4,4 @@ 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 \ - ring.o seg.o scoutfs_trace.o super.o trans.o xattr.o + seg.o scoutfs_trace.o super.o trans.o treap.o xattr.o diff --git a/kmod/src/alloc.c b/kmod/src/alloc.c index dbacf288..709f3b78 100644 --- a/kmod/src/alloc.c +++ b/kmod/src/alloc.c @@ -17,77 +17,128 @@ #include "super.h" #include "format.h" -#include "ring.h" +#include "treap.h" +#include "cmp.h" #include "alloc.h" /* - * scoutfs allocates segments by storing regions of a bitmap in a radix. - * As the regions are modified their index in the radix is marked dirty - * for writeout. + * scoutfs allocates segments by storing regions of a bitmap in treap + * nodes. * - * Frees are tracked in a separate radix. They're only applied to the - * free regions as a transaction is written. The frees can't satisfy - * allocation until they're committed so that we don't overwrite stable - * referenced data. + * Freed segments are recorded in nodes in an rbtree. The frees can't + * satisfy allocation until they're committed to prevent overwriting + * live data so they're only applied to the region nodes as their + * transaction is written. * - * The allocated segments are large enough to be effectively - * independent. We allocate by sweeping a cursor through the volume. - * This gives racing unlocked readers more time to try to sample a stale - * freed segment, when its safe to do so, before it is reallocated and + * We allocate by sweeping a cursor through the volume. This gives + * racing unlocked readers more time to try to sample a stale freed + * segment, when its safe to do so, before it is reallocated and * rewritten and they're forced to retry their racey read. - * - * XXX - * - make sure seg fits in long index - * - frees can delete region, leave non-NULL nul behind for logging */ struct seg_alloc { - spinlock_t lock; - struct radix_tree_root regs; - struct radix_tree_root pending; + struct rw_semaphore rwsem; + struct rb_root pending_root; + struct scoutfs_treap *treap; u64 next_segno; }; #define DECLARE_SEG_ALLOC(sb, name) \ struct seg_alloc *name = SCOUTFS_SB(sb)->seg_alloc -enum { - DIRTY_RADIX_TAG = 0, +struct pending_region { + struct rb_node node; + struct scoutfs_alloc_region reg; }; +static struct pending_region *find_pending(struct rb_root *root, u64 ind) +{ + struct rb_node *node = root->rb_node; + struct pending_region *pend; + + while (node) { + pend = container_of(node, struct pending_region, node); + + if (ind < le64_to_cpu(pend->reg.index)) + node = node->rb_left; + else if (ind > le64_to_cpu(pend->reg.index)) + node = node->rb_right; + else + return pend; + } + + return NULL; +} + +static void insert_pending(struct rb_root *root, struct pending_region *ins) +{ + struct rb_node **node = &root->rb_node; + struct rb_node *parent = NULL; + struct pending_region *pend; + u64 ind = le64_to_cpu(ins->reg.index); + + while (*node) { + parent = *node; + pend = container_of(*node, struct pending_region, node); + + if (ind < le64_to_cpu(pend->reg.index)) + node = &(*node)->rb_left; + else if (ind > le64_to_cpu(pend->reg.index)) + node = &(*node)->rb_right; + else + BUG(); + } + + rb_link_node(&ins->node, parent, node); + rb_insert_color(&ins->node, root); +} + +static bool empty_region(struct scoutfs_alloc_region *reg) +{ + int i; + + for (i = 0; i < ARRAY_SIZE(reg->bits); i++) { + if (reg->bits[i]) + return false; + } + + return true; +} + int scoutfs_alloc_segno(struct super_block *sb, u64 *segno) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; - struct scoutfs_ring_alloc_region *reg; + struct scoutfs_alloc_region *reg; DECLARE_SEG_ALLOC(sb, sal); - unsigned long flags; - unsigned long ind; + u64 ind; int ret; int nr; - spin_lock_irqsave(&sal->lock, flags); + down_write(&sal->rwsem); - /* start by sweeping through the device for the first time */ - if (sal->next_segno == le64_to_cpu(super->alloc_uninit)) { + /* initially sweep through all segments */ + if (super->alloc_uninit != super->total_segs) { + *segno = le64_to_cpu(super->alloc_uninit); + /* done when inc hits total_segs */ le64_add_cpu(&super->alloc_uninit, 1); - *segno = sal->next_segno++; - if (sal->next_segno == le64_to_cpu(super->total_segs)) - sal->next_segno = 0; ret = 0; goto out; } - /* then fall back to the allocator */ + /* but usually search for region nodes */ ind = sal->next_segno >> SCOUTFS_ALLOC_REGION_SHIFT; nr = sal->next_segno & SCOUTFS_ALLOC_REGION_MASK; do { - ret = radix_tree_gang_lookup(&sal->regs, (void **)®, ind, 1); - } while (ret == 0 && ind && (ind = 0, nr = 0, 1)); + reg = scoutfs_treap_lookup_next_dirty(sal->treap, &ind); + } while (reg == NULL && ind && (ind = 0, nr = 0, 1)); - if (ret == 0) { - ret = -ENOSPC; + if (IS_ERR_OR_NULL(reg)) { + if (IS_ERR(reg)) + ret = PTR_ERR(reg); + else + ret = -ENOSPC; goto out; } @@ -98,237 +149,200 @@ int scoutfs_alloc_segno(struct super_block *sb, u64 *segno) goto out; } + ind = le64_to_cpu(reg->index); + clear_bit_le(nr, reg->bits); - radix_tree_tag_set(&sal->regs, ind, DIRTY_RADIX_TAG); + + 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; + } *segno = (ind << SCOUTFS_ALLOC_REGION_SHIFT) + nr; - - /* once this wraps it will never equal alloc_uninit */ sal->next_segno = *segno + 1; - if (sal->next_segno == le64_to_cpu(super->total_segs)) - sal->next_segno = 0; ret = 0; out: - spin_unlock_irqrestore(&sal->lock, flags); + up_write(&sal->rwsem); trace_printk("segno %llu ret %d\n", *segno, ret); return ret; } /* - * Record newly freed sgements in pending regions. These can't be - * applied to the main allocator regions until the next commit so that - * they're not still referenced by the stable tree in event of a crash. - * - * The pending regions are merged into dirty regions for the next commit. + * Record newly freed sgements in pending regions. These are applied to + * treap nodes as the transaction commits. */ int scoutfs_alloc_free(struct super_block *sb, u64 segno) { - struct scoutfs_ring_alloc_region *reg; - struct scoutfs_ring_alloc_region *ins; + struct pending_region *pend; DECLARE_SEG_ALLOC(sb, sal); - unsigned long flags; - unsigned long ind; + u64 ind; int ret; int nr; ind = segno >> SCOUTFS_ALLOC_REGION_SHIFT; nr = segno & SCOUTFS_ALLOC_REGION_MASK; - ins = kzalloc(sizeof(struct scoutfs_ring_alloc_region), GFP_NOFS); - if (!ins) { - ret = -ENOMEM; - goto out; + down_write(&sal->rwsem); + + pend = find_pending(&sal->pending_root, ind); + if (!pend) { + pend = kzalloc(sizeof(struct pending_region), GFP_NOFS); + if (!pend) { + ret = -ENOMEM; + goto out; + } + + pend->reg.index = cpu_to_le64(ind); + insert_pending(&sal->pending_root, pend); } - ins->eh.type = SCOUTFS_RING_ADD_ALLOC; - ins->eh.len = cpu_to_le16(sizeof(struct scoutfs_ring_alloc_region)); - ins->index = cpu_to_le64(ind); - - ret = radix_tree_preload(GFP_NOFS); - if (ret) { - goto out; - } - - spin_lock_irqsave(&sal->lock, flags); - - reg = radix_tree_lookup(&sal->pending, ind); - if (!reg) { - reg = ins; - ins = NULL; - radix_tree_insert(&sal->pending, ind, reg); - } - - set_bit_le(nr, reg->bits); - - spin_unlock_irqrestore(&sal->lock, flags); - radix_tree_preload_end(); + set_bit_le(nr, pend->reg.bits); + ret = 0; out: - kfree(ins); - trace_printk("freeing segno %llu ind %lu nr %d ret %d\n", + up_write(&sal->rwsem); + + trace_printk("freeing segno %llu ind %llu nr %d ret %d\n", segno, ind, nr, ret); return ret; } -/* - * Add a new clean region from the ring. It can be replacing existing - * clean stale entries during replay as we make our way through the - * ring. - */ -int scoutfs_alloc_add(struct super_block *sb, - struct scoutfs_ring_alloc_region *ins) +static void or_region_bits(struct scoutfs_alloc_region *dst, + struct scoutfs_alloc_region *src) +{ + int i; + + for (i = 0; i < ARRAY_SIZE(dst->bits); i++) + dst->bits[i] |= src->bits[i]; +} + +int scoutfs_alloc_has_dirty(struct super_block *sb) { - struct scoutfs_ring_alloc_region *existing; - struct scoutfs_ring_alloc_region *reg; DECLARE_SEG_ALLOC(sb, sal); - unsigned long flags; int ret; - reg = kmalloc(sizeof(struct scoutfs_ring_alloc_region), GFP_NOFS); - if (!reg) { - ret = -ENOMEM; - goto out; - } + down_write(&sal->rwsem); + ret = scoutfs_treap_has_dirty(sal->treap); + up_write(&sal->rwsem); - memcpy(reg, ins, sizeof(struct scoutfs_ring_alloc_region)); - - ret = radix_tree_preload(GFP_NOFS); - if (ret) { - kfree(reg); - goto out; - } - - spin_lock_irqsave(&sal->lock, flags); - - existing = radix_tree_lookup(&sal->regs, le64_to_cpu(reg->index)); - if (existing) - radix_tree_delete(&sal->regs, le64_to_cpu(reg->index)); - radix_tree_insert(&sal->regs, le64_to_cpu(reg->index), reg); - - spin_unlock_irqrestore(&sal->lock, flags); - radix_tree_preload_end(); - - if (existing) - kfree(existing); - - ret = 0; -out: - trace_printk("inserted reg ind %llu ret %d\n", - le64_to_cpu(ins->index), ret); return ret; } /* - * Append all the dirty alloc regions to the end of the ring. First we - * apply the pending frees to create the final set of dirty regions. - * - * This can't fail and always returns 0. + * 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. */ int scoutfs_alloc_dirty_ring(struct super_block *sb) { - struct scoutfs_ring_alloc_region *regs[16]; - struct scoutfs_ring_alloc_region *reg; DECLARE_SEG_ALLOC(sb, sal); - unsigned long start; - unsigned long ind; - int nr; - int i; - int b; + 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; + u64 ind; + int ret; - /* - * Merge pending free regions into dirty regions. If the dirty - * region doesn't exist we can just move the pending region over. - * If it does we or the pending bits in the region. - */ - start = 0; - do { - nr = radix_tree_gang_lookup(&sal->pending, (void **)regs, - start, ARRAY_SIZE(regs)); - for (i = 0; i < nr; i++) { - ind = le64_to_cpu(regs[i]->index); + down_write(&sal->rwsem); - reg = radix_tree_lookup(&sal->regs, ind); - if (!reg) { - radix_tree_insert(&sal->regs, ind, regs[i]); - } else { - for (b = 0; b < ARRAY_SIZE(reg->bits); b++) - reg->bits[i] |= regs[i]->bits[i]; - kfree(regs[i]); - } + while ((node = rb_first(&sal->pending_root))) { + pend = container_of(node, struct pending_region, node); - radix_tree_delete(&sal->pending, ind); - radix_tree_tag_set(&sal->regs, ind, DIRTY_RADIX_TAG); - start = ind + 1; + 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; } - } while (nr); - /* and append all the dirty regions to the ring */ - start = 0; - do { - nr = radix_tree_gang_lookup_tag(&sal->regs, (void **)regs, - start, ARRAY_SIZE(regs), - DIRTY_RADIX_TAG); - for (i = 0; i < nr; i++) { - reg = regs[i]; - ind = le64_to_cpu(reg->index); + reg->index = pend->reg.index; + or_region_bits(reg, &pend->reg); - scoutfs_ring_append(sb, ®->eh); - radix_tree_tag_clear(&sal->regs, ind, DIRTY_RADIX_TAG); - start = ind + 1; - } - } while (nr); + rb_erase(&pend->node, &sal->pending_root); + kfree(pend); + } - return 0; + scoutfs_treap_dirty_ring(sal->treap); + scoutfs_treap_update_root(&super->alloc_treap_root, sal->treap); + ret = 0; +out: + up_write(&sal->rwsem); + return ret; } +static int alloc_treap_compare(void *key, void *data) +{ + u64 *ind = key; + struct scoutfs_alloc_region *reg = data; + + return scoutfs_cmp_u64s(*ind, le64_to_cpu(reg->index)); +} + +static void alloc_treap_fill(void *data, void *fill_arg) +{ + struct scoutfs_alloc_region *reg = data; + u64 *ind = fill_arg; + + memset(reg, 0, sizeof(struct scoutfs_alloc_region)); + reg->index = cpu_to_le64p(ind); +} + +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; /* bits need to be aligned so hosts can use native bitops */ - BUILD_BUG_ON(offsetof(struct scoutfs_ring_alloc_region, bits) & + BUILD_BUG_ON(offsetof(struct scoutfs_alloc_region, bits) & (sizeof(long) - 1)); sal = kzalloc(sizeof(struct seg_alloc), GFP_KERNEL); if (!sal) return -ENOMEM; - sbi->seg_alloc = sal; - spin_lock_init(&sal->lock); - /* inserts preload with _NOFS */ - INIT_RADIX_TREE(&sal->pending, GFP_ATOMIC); - INIT_RADIX_TREE(&sal->regs, GFP_ATOMIC); + 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) { + kfree(sal); + return -ENOMEM; + } + /* XXX read next_segno from super? */ + sbi->seg_alloc = sal; + return 0; } -static void destroy_radix_regs(struct radix_tree_root *radix) -{ - struct scoutfs_ring_alloc_region *regs[16]; - int nr; - int i; - - - do { - nr = radix_tree_gang_lookup(radix, (void **)regs, - 0, ARRAY_SIZE(regs)); - for (i = 0; i < nr; i++) { - radix_tree_delete(radix, le64_to_cpu(regs[i]->index)); - kfree(regs[i]); - } - } while (nr); -} - void scoutfs_alloc_destroy(struct super_block *sb) { DECLARE_SEG_ALLOC(sb, sal); + struct pending_region *pend; + struct rb_node *node; if (sal) { - destroy_radix_regs(&sal->pending); - destroy_radix_regs(&sal->regs); + scoutfs_treap_free(sal->treap); + while ((node = rb_first(&sal->pending_root))) { + pend = container_of(node, struct pending_region, node); + rb_erase(&pend->node, &sal->pending_root); + kfree(pend); + } kfree(sal); } } diff --git a/kmod/src/alloc.h b/kmod/src/alloc.h index 4d3d398b..453d667a 100644 --- a/kmod/src/alloc.h +++ b/kmod/src/alloc.h @@ -6,8 +6,7 @@ struct scoutfs_alloc_region; int scoutfs_alloc_segno(struct super_block *sb, u64 *segno); int scoutfs_alloc_free(struct super_block *sb, u64 segno); -int scoutfs_alloc_add(struct super_block *sb, - struct scoutfs_ring_alloc_region *ins); +int scoutfs_alloc_has_dirty(struct super_block *sb); int scoutfs_alloc_dirty_ring(struct super_block *sb); int scoutfs_alloc_setup(struct super_block *sb); diff --git a/kmod/src/format.h b/kmod/src/format.h index 71027e90..c611ea32 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -51,22 +51,63 @@ struct scoutfs_block_header { __le64 blkno; } __packed; -struct scoutfs_ring_entry_header { - __u8 type; - __le16 len; +struct scoutfs_treap_ref { + __le64 off; + __le64 gen; + __u8 aug_bits; } __packed; -#define SCOUTFS_RING_ADD_MANIFEST 1 -#define SCOUTFS_RING_ADD_ALLOC 2 +/* + * 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) -struct scoutfs_ring_add_manifest { - struct scoutfs_ring_entry_header eh; +/* + * 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 { + __le32 crc; + __le64 off; + __le64 gen; + __le64 prio; + struct scoutfs_treap_ref left; + struct scoutfs_treap_ref right; + __le16 bytes; + u8 data[0]; +} __packed; + +struct scoutfs_treap_root { + struct scoutfs_treap_ref ref; +} __packed; + +/* + * This is absurdly huge. If there was only ever 1 item per segment and + * 2^64 items the tree could get this deep. + */ +#define SCOUTFS_MANIFEST_MAX_LEVEL 20 + +struct scoutfs_manifest { + struct scoutfs_treap_root root; + __le64 level_counts[SCOUTFS_MANIFEST_MAX_LEVEL]; +} __packed; + +struct scoutfs_manifest_entry { __le64 segno; __le64 seq; __le16 first_key_len; __le16 last_key_len; __u8 level; - /* first and last key bytes */ + __u8 keys[0]; } __packed; #define SCOUTFS_ALLOC_REGION_SHIFT 8 @@ -77,27 +118,11 @@ struct scoutfs_ring_add_manifest { * The bits need to be aligned so that the host can use native long * bitops on the bits in memory. */ -struct scoutfs_ring_alloc_region { - struct scoutfs_ring_entry_header eh; +struct scoutfs_alloc_region { __le64 index; - __u8 pad[5]; __le64 bits[SCOUTFS_ALLOC_REGION_BITS / 64]; } __packed; -/* - * This is absurdly huge. If there was only ever 1 item per segment and - * 2^64 items the tree could get this deep. - */ -#define SCOUTFS_MANIFEST_MAX_LEVEL 20 - -/* - * The packed entries in the block are terminated by a header with a 0 length. - */ -struct scoutfs_ring_block { - struct scoutfs_block_header hdr; - struct scoutfs_ring_entry_header entries[0]; -} __packed; - /* * We really want these to be a power of two size so that they're naturally * aligned. This ensures that they won't cross page boundaries and we @@ -315,12 +340,13 @@ struct scoutfs_super_block { __le64 free_blocks; __le64 ring_blkno; __le64 ring_blocks; - __le64 ring_index; - __le64 ring_nr; - __le64 ring_seq; + __le64 ring_tail_block; + __le64 ring_gen; __le64 buddy_blocks; struct scoutfs_buddy_root buddy_root; struct scoutfs_btree_root btree_root; + struct scoutfs_treap_root alloc_treap_root; + struct scoutfs_manifest manifest; } __packed; #define SCOUTFS_ROOT_INO 1 diff --git a/kmod/src/manifest.c b/kmod/src/manifest.c index 4980109c..51ff1b0e 100644 --- a/kmod/src/manifest.c +++ b/kmod/src/manifest.c @@ -14,43 +14,42 @@ #include #include #include +#include #include "super.h" #include "format.h" #include "kvec.h" #include "seg.h" #include "item.h" -#include "ring.h" +#include "treap.h" +#include "cmp.h" #include "manifest.h" #include "scoutfs_trace.h" +/* + * Manifest entries are stored as treap nodes in the ring. + * + * 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 { - spinlock_t lock; - - struct list_head level0_list; - unsigned int level0_nr; - - u8 last_level; - struct rb_root level_roots[SCOUTFS_MANIFEST_MAX_LEVEL + 1]; - - struct list_head dirty_list; + struct rw_semaphore rwsem; + struct scoutfs_treap *treap; + u8 nr_levels; }; #define DECLARE_MANIFEST(sb, name) \ struct manifest *name = SCOUTFS_SB(sb)->manifest -struct manifest_entry { - union { - struct list_head level0_entry; - struct rb_node node; - }; - struct list_head dirty_entry; - - struct scoutfs_ring_add_manifest am; - /* u8 key_bytes[am.first_key_len]; */ - /* u8 val_bytes[am.last_key_len]; */ -}; - /* * A reader uses references to segments copied from a walk of the * manifest. The references are a point in time sample of the manifest. @@ -71,17 +70,31 @@ struct manifest_ref { u16 first_key_len; u16 last_key_len; u8 level; - u8 keys[SCOUTFS_MAX_KEY_SIZE * 2]; + u8 keys[0]; }; -static void init_ment_keys(struct manifest_entry *ment, struct kvec *first, - struct kvec *last) +struct manifest_fill_args { + struct scoutfs_manifest_entry ment; + struct kvec *first; + struct kvec *last; +}; + +struct manifest_search_key { + u64 seq; + struct kvec *key; + u8 level; +}; + +static void init_ment_keys(struct scoutfs_manifest_entry *ment, + struct kvec *first, struct kvec *last) { - scoutfs_kvec_init(first, &ment->am + 1, - le16_to_cpu(ment->am.first_key_len)); - scoutfs_kvec_init(last, (void *)(&ment->am + 1) + - le16_to_cpu(ment->am.first_key_len), - le16_to_cpu(ment->am.last_key_len)); + if (first) + scoutfs_kvec_init(first, ment->keys, + le16_to_cpu(ment->first_key_len)); + if (last) + scoutfs_kvec_init(last, ment->keys + + le16_to_cpu(ment->first_key_len), + le16_to_cpu(ment->last_key_len)); } static void init_ref_keys(struct manifest_ref *ref, struct kvec *first, @@ -95,7 +108,7 @@ static void init_ref_keys(struct manifest_ref *ref, struct kvec *first, } static bool cmp_range_ment(struct kvec *key, struct kvec *end, - struct manifest_entry *ment) + struct scoutfs_manifest_entry *ment) { SCOUTFS_DECLARE_KVEC(first); SCOUTFS_DECLARE_KVEC(last); @@ -105,199 +118,102 @@ static bool cmp_range_ment(struct kvec *key, struct kvec *end, return scoutfs_kvec_cmp_overlap(key, end, first, last); } -static struct manifest_entry *find_ment(struct rb_root *root, struct kvec *key) -{ - struct rb_node *node = root->rb_node; - struct manifest_entry *ment; - int cmp; - - while (node) { - ment = container_of(node, struct manifest_entry, node); - - cmp = cmp_range_ment(key, key, ment); - if (cmp < 0) - node = node->rb_left; - else if (cmp > 0) - node = node->rb_right; - else - return ment; - } - - return NULL; -} - /* - * Insert a new entry into one of the L1+ trees. There should never be - * entries that overlap. + * Insert a new manifest entry in the treap. The treap allocates a new + * node for us and we fill it. */ -static int insert_ment(struct rb_root *root, struct manifest_entry *ins) +int scoutfs_manifest_add(struct super_block *sb, struct kvec *first, + struct kvec *last, u64 segno, u64 seq, u8 level) { - struct rb_node **node = &root->rb_node; - struct rb_node *parent = NULL; - struct manifest_entry *ment; - SCOUTFS_DECLARE_KVEC(key); - SCOUTFS_DECLARE_KVEC(end); - int cmp; - - init_ment_keys(ins, key, end); - - while (*node) { - parent = *node; - ment = container_of(*node, struct manifest_entry, node); - - cmp = cmp_range_ment(key, end, ment); - if (cmp < 0) { - node = &(*node)->rb_left; - } else if (cmp > 0) { - node = &(*node)->rb_right; - } else { - return -EEXIST; - } - } - - rb_link_node(&ins->node, parent, node); - rb_insert_color(&ins->node, root); - - return 0; -} - -static void free_ment(struct manifest_entry *ment) -{ - if (!IS_ERR_OR_NULL(ment)) - kfree(ment); -} - -static int add_ment(struct manifest *mani, struct manifest_entry *ment, - bool dirty) -{ - u8 level = ment->am.level; + 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_fill_args args; + struct manifest_search_key skey; + unsigned key_bytes; + unsigned bytes; int ret; + trace_scoutfs_manifest_add(sb, first, last, segno, seq, level); - trace_printk("adding ment %p level %u\n", ment, level); + key_bytes = scoutfs_kvec_length(first) + scoutfs_kvec_length(last); + bytes = offsetof(struct scoutfs_manifest_entry, keys[key_bytes]); - if (level) { - ret = insert_ment(&mani->level_roots[level], ment); - if (!ret) - mani->last_level = max(mani->last_level, level); + args.ment.segno = cpu_to_le64(segno); + args.ment.seq = cpu_to_le64(seq); + args.ment.first_key_len = cpu_to_le16(scoutfs_kvec_length(first)); + args.ment.last_key_len = cpu_to_le16(scoutfs_kvec_length(last)); + args.ment.level = level; + + args.first = first; + args.last = last; + + skey.key = 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 { - list_add_tail(&ment->level0_entry, &mani->level0_list); - mani->level0_nr++; + mani->nr_levels = max_t(u8, mani->nr_levels, level + 1); + le64_add_cpu(&super->manifest.level_counts[level], 1); ret = 0; } - if (dirty) - list_add_tail(&ment->dirty_entry, &mani->dirty_list); + up_write(&mani->rwsem); return ret; } -static void update_last_level(struct manifest *mani) -{ - int i; - - for (i = mani->last_level; - i > 0 && RB_EMPTY_ROOT(&mani->level_roots[i]); i--) - ; - - mani->last_level = i; -} - -static void remove_ment(struct manifest *mani, struct manifest_entry *ment) -{ - u8 level = ment->am.level; - - if (level) { - rb_erase(&ment->node, &mani->level_roots[level]); - update_last_level(mani); - } else { - list_del_init(&ment->level0_entry); - mani->level0_nr--; - } - - /* XXX more carefully remove dirty ments.. should be exceptional */ - if (!list_empty(&ment->dirty_entry)) - list_del_init(&ment->dirty_entry); -} - -int scoutfs_manifest_add(struct super_block *sb, struct kvec *first, - struct kvec *last, u64 segno, u64 seq, u8 level, - bool dirty) -{ - DECLARE_MANIFEST(sb, mani); - struct manifest_entry *ment; - SCOUTFS_DECLARE_KVEC(ment_first); - SCOUTFS_DECLARE_KVEC(ment_last); - unsigned long flags; - int key_bytes; - int ret; - - trace_scoutfs_manifest_add(sb, first, last, segno, seq, level, dirty); - - key_bytes = scoutfs_kvec_length(first) + scoutfs_kvec_length(last); - ment = kmalloc(sizeof(struct manifest_entry) + key_bytes, GFP_NOFS); - if (!ment) - return -ENOMEM; - - if (level) - RB_CLEAR_NODE(&ment->node); - else - INIT_LIST_HEAD(&ment->level0_entry); - INIT_LIST_HEAD(&ment->dirty_entry); - - ment->am.eh.type = SCOUTFS_RING_ADD_MANIFEST; - ment->am.eh.len = cpu_to_le16(sizeof(struct scoutfs_ring_add_manifest) + - key_bytes); - ment->am.segno = cpu_to_le64(segno); - ment->am.seq = cpu_to_le64(seq); - ment->am.first_key_len = cpu_to_le16(scoutfs_kvec_length(first)); - ment->am.last_key_len = cpu_to_le16(scoutfs_kvec_length(last)); - ment->am.level = level; - - init_ment_keys(ment, ment_first, ment_last); - scoutfs_kvec_memcpy(ment_first, first); - scoutfs_kvec_memcpy(ment_last, last); - - /* XXX think about where to insert level 0 */ - spin_lock_irqsave(&mani->lock, flags); - ret = add_ment(mani, ment, dirty); - spin_unlock_irqrestore(&mani->lock, flags); - if (WARN_ON_ONCE(ret)) /* XXX can this happen? ring corruption? */ - free_ment(ment); - - return ret; -} - -/* - * Grab an allocated ref from the src list, fill it with the details - * from the ment, and add it to the dst list. The ref is added to the - * tail of the dst list so that we maintain the caller's manifest walk - * order. - */ -static void fill_ref_tail(struct list_head *dst, struct list_head *src, - struct manifest_entry *ment) +static int alloc_add_ref(struct list_head *list, + struct scoutfs_manifest_entry *ment) { SCOUTFS_DECLARE_KVEC(ment_first); SCOUTFS_DECLARE_KVEC(ment_last); SCOUTFS_DECLARE_KVEC(first); SCOUTFS_DECLARE_KVEC(last); struct manifest_ref *ref; - - ref = list_first_entry(src, struct manifest_ref, entry); - - ref->segno = le64_to_cpu(ment->am.segno); - ref->seq = le64_to_cpu(ment->am.seq); - ref->level = ment->am.level; - ref->first_key_len = le16_to_cpu(ment->am.first_key_len); - ref->last_key_len = le16_to_cpu(ment->am.last_key_len); + unsigned bytes; init_ment_keys(ment, ment_first, ment_last); - init_ref_keys(ref, first, last); + bytes = scoutfs_kvec_length(ment_first) + + scoutfs_kvec_length(ment_first); + + ref = kmalloc(offsetof(struct manifest_ref, keys[bytes]), GFP_NOFS); + if (!ref) + return -ENOMEM; + + memset(ref, 0, offsetof(struct manifest_ref, keys)); + + ref->segno = le64_to_cpu(ment->segno); + ref->seq = le64_to_cpu(ment->seq); + ref->level = ment->level; + ref->first_key_len = le16_to_cpu(ment->first_key_len); + ref->last_key_len = le16_to_cpu(ment->last_key_len); + + init_ref_keys(ref, first, last); scoutfs_kvec_memcpy(first, ment_first); scoutfs_kvec_memcpy(last, ment_last); - list_move_tail(&ref->entry, dst); + list_add_tail(&ref->entry, list); + + return 0; + +} + +/* 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); } /* @@ -308,78 +224,85 @@ static void fill_ref_tail(struct list_head *dst, struct list_head *src, * of items that we want to search because the level 0 segments can * arbitrarily overlap with each other. * - * We only need to search for the starting key in all the higher order - * levels. They do not overlap so we can iterate through the key space - * in each segment starting with the key. + * We only need to search for the starting key in all the higher levels. + * They do not overlap so we can iterate through the key space in each + * segment starting with the key. */ -static int get_range_refs(struct manifest *mani, struct kvec *key, - struct kvec *end, struct list_head *ref_list) +static int get_range_refs(struct super_block *sb, struct manifest *mani, + struct kvec *key, struct kvec *end, + struct list_head *ref_list) { - struct manifest_entry *ment; + struct scoutfs_manifest_entry *ment; + struct manifest_search_key skey; + SCOUTFS_DECLARE_KVEC(first); + SCOUTFS_DECLARE_KVEC(last); struct manifest_ref *ref; struct manifest_ref *tmp; - struct rb_root *root; - unsigned long flags; - unsigned int total; - unsigned int nr = 0; - LIST_HEAD(alloced); + int cmp; int ret; int i; - trace_printk("getting refs\n"); + down_write(&mani->rwsem); - spin_lock_irqsave(&mani->lock, flags); + /* get level 0 segments that overlap with the missing range */ + ment = scoutfs_treap_first(mani->treap); + while (!IS_ERR_OR_NULL(ment)) { + if (ment->level > 0) + break; - /* allocate enough refs for the of segments */ - total = mani->level0_nr + mani->last_level; - while (nr < total) { - spin_unlock_irqrestore(&mani->lock, flags); + cmp = cmp_range_ment(key, end, ment); + if (cmp < 0) + break; - for (i = nr; i < total; i++) { - ref = kmalloc(sizeof(struct manifest_ref), GFP_NOFS); - if (!ref) { - ret = -ENOMEM; + if (cmp == 0) { + ret = alloc_add_ref(ref_list, ment); + if (ret) goto out; - } - - memset(ref, 0, offsetof(struct manifest_ref, keys)); - list_add(&ref->entry, &alloced); } - nr = total; - spin_lock_irqsave(&mani->lock, flags); + ment = scoutfs_treap_next(mani->treap, ment); + } + if (IS_ERR(ment)) { + ret = PTR_ERR(ment); + goto out; } - /* find all the overlapping level 0 segments */ - list_for_each_entry(ment, &mani->level0_list, level0_entry) { - if (cmp_range_ment(key, end, ment)) - continue; + /* level0s are sorted by key, reverse sort by seq */ + list_sort(NULL, ref_list, cmp_ref_list_seqs); - fill_ref_tail(ref_list, &alloced, ment); + /* get higher level segments that overlap with the starting key */ + for (i = 1; i < mani->nr_levels; i++) { + skey.key = key; + skey.level = i; + + /* 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; + } + + if (ment) { + init_ment_keys(ment, first, last); + ret = alloc_add_ref(ref_list, ment); + if (ret) + goto out; + } } - /* find each segment containing the key at the higher orders */ - for (i = 1; i <= mani->last_level; i++) { - root = &mani->level_roots[i]; - if (RB_EMPTY_ROOT(root)) - continue; - - ment = find_ment(root, key); - if (ment) - fill_ref_tail(ref_list, &alloced, ment); - } - - spin_unlock_irqrestore(&mani->lock, flags); ret = 0; out: + up_write(&mani->rwsem); + if (ret) { - list_splice_init(ref_list, &alloced); - list_for_each_entry_safe(ref, tmp, &alloced, entry) { + list_for_each_entry_safe(ref, tmp, ref_list, entry) { list_del_init(&ref->entry); kfree(ref); } } + trace_printk("ret %d\n", ret); return ret; } @@ -436,7 +359,7 @@ int scoutfs_manifest_read_items(struct super_block *sb, struct kvec *key, trace_printk("reading items\n"); /* get refs on all the segments */ - ret = get_range_refs(mani, key, end, &ref_list); + ret = get_range_refs(sb, mani, key, end, &ref_list); if (ret) return ret; @@ -576,8 +499,13 @@ out: int scoutfs_manifest_has_dirty(struct super_block *sb) { DECLARE_MANIFEST(sb, mani); + int ret; - return !list_empty_careful(&mani->dirty_list); + down_write(&mani->rwsem); + ret = scoutfs_treap_has_dirty(mani->treap); + up_write(&mani->rwsem); + + return ret; } /* @@ -588,33 +516,100 @@ int scoutfs_manifest_has_dirty(struct super_block *sb) int scoutfs_manifest_dirty_ring(struct super_block *sb) { DECLARE_MANIFEST(sb, mani); - struct manifest_entry *ment; - struct manifest_entry *tmp; + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_super_block *super = &sbi->super; - list_for_each_entry_safe(ment, tmp, &mani->dirty_list, dirty_entry) { - scoutfs_ring_append(sb, &ment->am.eh); - list_del_init(&ment->dirty_entry); - } + down_write(&mani->rwsem); + scoutfs_treap_dirty_ring(mani->treap); + scoutfs_treap_update_root(&super->manifest.root, mani->treap); + up_write(&mani->rwsem); return 0; } +/* + * Manifest entries are first sorted by their level. + * + * 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. + * + * Higher level segments don't overlap. There will never be manifest + * entries with the same key at a given level. + */ +static int manifest_treap_compare(void *key, void *data) +{ + struct manifest_search_key *skey = key; + struct scoutfs_manifest_entry *ment = data; + SCOUTFS_DECLARE_KVEC(first); + SCOUTFS_DECLARE_KVEC(last); + + if (skey->level < ment->level) + return -1; + if (skey->level > ment->level) + return 1; + + init_ment_keys(ment, first, NULL); + + 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, NULL, last); + + return scoutfs_kvec_cmp_overlap(skey->key, skey->key, first, last); +} + +static void manifest_treap_fill(void *data, void *arg) +{ + struct scoutfs_manifest_entry *ment = data; + struct manifest_fill_args *args = arg; + SCOUTFS_DECLARE_KVEC(ment_first); + SCOUTFS_DECLARE_KVEC(ment_last); + + *ment = args->ment; + + init_ment_keys(ment, ment_first, ment_last); + scoutfs_kvec_memcpy(ment_first, args->first); + scoutfs_kvec_memcpy(ment_last, args->last); +} + +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 i; mani = kzalloc(sizeof(struct manifest), GFP_KERNEL); if (!mani) return -ENOMEM; - sbi->manifest = mani; - spin_lock_init(&mani->lock); - INIT_LIST_HEAD(&mani->level0_list); - INIT_LIST_HEAD(&mani->dirty_list); - for (i = 0; i < ARRAY_SIZE(mani->level_roots); i++) - mani->level_roots[i] = RB_ROOT; + init_rwsem(&mani->rwsem); + mani->treap = scoutfs_treap_alloc(sb, &manifest_treap_ops, + &super->manifest.root); + if (!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; + break; + } + } + + sbi->manifest = mani; return 0; } @@ -623,30 +618,9 @@ void scoutfs_manifest_destroy(struct super_block *sb) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct manifest *mani = sbi->manifest; - struct manifest_entry *ment; - struct manifest_entry *tmp; - struct rb_node *node; - struct rb_root *root; - int i; - if (!mani) - return; - - for (i = 1; i <= mani->last_level; i++) { - root = &mani->level_roots[i]; - - for (node = rb_first(root); node; ) { - ment = container_of(node, struct manifest_entry, node); - node = rb_next(node); - remove_ment(mani, ment); - free_ment(ment); - } + if (mani) { + scoutfs_treap_free(mani->treap); + kfree(mani); } - - list_for_each_entry_safe(ment, tmp, &mani->level0_list, level0_entry) { - remove_ment(mani, ment); - free_ment(ment); - } - - kfree(mani); } diff --git a/kmod/src/manifest.h b/kmod/src/manifest.h index 9f1477b4..021bdf14 100644 --- a/kmod/src/manifest.h +++ b/kmod/src/manifest.h @@ -2,8 +2,7 @@ #define _SCOUTFS_MANIFEST_H_ int scoutfs_manifest_add(struct super_block *sb, struct kvec *first, - struct kvec *last, u64 segno, u64 seq, u8 level, - bool dirty); + struct kvec *last, u64 segno, u64 seq, u8 level); int scoutfs_manifest_has_dirty(struct super_block *sb); int scoutfs_manifest_dirty_ring(struct super_block *sb); diff --git a/kmod/src/ring.c b/kmod/src/ring.c deleted file mode 100644 index 050fa49c..00000000 --- a/kmod/src/ring.c +++ /dev/null @@ -1,326 +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 "super.h" -#include "format.h" -#include "kvec.h" -#include "bio.h" -#include "manifest.h" -#include "alloc.h" -#include "ring.h" -#include "crc.h" - - -/* - * Right now we're only writing a segment a time. The entries needed to - * write a segment will always be smaller than a segment itself. - * - * XXX This'll get more clever as we can write multiple segments and build - * up dirty entries while processing compaction results. - */ -struct ring_info { - struct page *pages[SCOUTFS_SEGMENT_PAGES]; - struct scoutfs_ring_block *ring; - struct scoutfs_ring_entry_header *next_eh; - unsigned int nr_blocks; - unsigned int space; -}; - -#define DECLARE_RING_INFO(sb, name) \ - struct ring_info *name = SCOUTFS_SB(sb)->ring_info - -/* - * XXX - * - verify blocks - * - could compress - * - have all entry sources dirty at cursors before dirtying - * - advancing cursor updates head as cursor wraps - */ - -/* - * The space calculation when starting a block included a final empty - * entry header. That is zeroed here. - */ -static void finish_block(struct scoutfs_ring_block *ring, unsigned int tail) -{ - memset((char *)ring + SCOUTFS_BLOCK_SIZE - tail, 0, tail); - scoutfs_crc_block(&ring->hdr); -} - -void scoutfs_ring_append(struct super_block *sb, - struct scoutfs_ring_entry_header *eh) -{ - DECLARE_RING_INFO(sb, rinf); - struct scoutfs_ring_block *ring = rinf->ring; - unsigned int len = le16_to_cpu(eh->len); - - if (rinf->space < (len + sizeof(struct scoutfs_ring_entry_header))) { - if (rinf->space) - finish_block(ring, rinf->space); - ring = scoutfs_page_block_address(rinf->pages, rinf->nr_blocks); - rinf->ring = ring; - - memset(ring, 0, sizeof(struct scoutfs_ring_block)); - - rinf->nr_blocks++; - rinf->next_eh = ring->entries; - rinf->space = SCOUTFS_BLOCK_SIZE - - offsetof(struct scoutfs_ring_block, entries); - } - - memcpy(rinf->next_eh, eh, len); - rinf->next_eh = (void *)rinf->next_eh + len; - rinf->space -= len; -} - -static u64 ring_ind_wrap(struct scoutfs_super_block *super, u64 ind) -{ - u64 ring_blocks = le64_to_cpu(super->ring_blocks); - - while (ind >= ring_blocks) - ind -= ring_blocks; - - return ind; -} - -/* - * Submit writes for all the dirty ring blocks that accumulated as dirty - * 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. - */ -int scoutfs_ring_submit_write(struct super_block *sb, - struct scoutfs_bio_completion *comp) -{ - struct scoutfs_super_block *super = &SCOUTFS_SB(sb)->super; - DECLARE_RING_INFO(sb, rinf); - u64 wrapped_blocks; - u64 index_blocks; - u64 index; - - if (!rinf->nr_blocks) - return 0; - - if (rinf->space) - finish_block(rinf->ring, rinf->space); - - /* first and last ring block indexes that will be written */ - index = ring_ind_wrap(super, le64_to_cpu(super->ring_index) + - le64_to_cpu(super->ring_nr)); - index_blocks = min_t(u64, rinf->nr_blocks, - le64_to_cpu(super->ring_blocks) - index); - wrapped_blocks = rinf->nr_blocks - index_blocks; - - if (wrapped_blocks) { - BUILD_BUG_ON(SCOUTFS_BLOCK_SIZE != PAGE_SIZE); - scoutfs_bio_submit_comp(sb, WRITE, rinf->pages + index_blocks, - le64_to_cpu(super->ring_blkno), - wrapped_blocks, comp); - } - - scoutfs_bio_submit_comp(sb, WRITE, rinf->pages, - le64_to_cpu(super->ring_blkno) + index, - index_blocks, comp); - - /* record new tail index in super and reset for next trans */ - le64_add_cpu(&super->ring_nr, rinf->nr_blocks); - rinf->nr_blocks = 0; - rinf->space = 0; - - return 0; -} - -static int read_one_entry(struct super_block *sb, - struct scoutfs_ring_entry_header *eh) -{ - struct scoutfs_ring_alloc_region *reg; - struct scoutfs_ring_add_manifest *am; - SCOUTFS_DECLARE_KVEC(first); - SCOUTFS_DECLARE_KVEC(last); - int ret; - - trace_printk("type %u len %u\n", eh->type, le16_to_cpu(eh->len)); - - switch(eh->type) { - case SCOUTFS_RING_ADD_MANIFEST: - am = container_of(eh, struct scoutfs_ring_add_manifest, eh); - - trace_printk("lens %u %u\n", - le16_to_cpu(am->first_key_len), - le16_to_cpu(am->last_key_len)); - - scoutfs_kvec_init(first, am + 1, - le16_to_cpu(am->first_key_len)); - scoutfs_kvec_init(last, - first[0].iov_base + first[0].iov_len, - le16_to_cpu(am->last_key_len)); - - ret = scoutfs_manifest_add(sb, first, last, - le64_to_cpu(am->segno), - le64_to_cpu(am->seq), am->level, - false); - break; - - case SCOUTFS_RING_ADD_ALLOC: - reg = container_of(eh, struct scoutfs_ring_alloc_region, eh); - ret = scoutfs_alloc_add(sb, reg); - break; - - default: - ret = -EINVAL; - } - - return ret; -} - -static int read_entries(struct super_block *sb, - struct scoutfs_ring_block *ring) -{ - struct scoutfs_ring_entry_header *eh; - int ret = 0; - - for (eh = ring->entries; eh->len; - eh = (void *)eh + le16_to_cpu(eh->len)) { - - ret = read_one_entry(sb, eh); - if (ret) - break; - } - - return ret; -} - - -/* read in a meg at a time */ -#define NR_PAGES DIV_ROUND_UP(1024 * 1024, PAGE_SIZE) -#define NR_BLOCKS (NR_PAGES * SCOUTFS_BLOCKS_PER_PAGE) - -int scoutfs_ring_read(struct super_block *sb) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_super_block *super = &sbi->super; - struct scoutfs_ring_block *ring; - struct page **pages; - struct page *page; - u64 index; - u64 blkno; - u64 part; - u64 seq; - u64 nr; - int ret; - int i; - - /* nr_blocks/pages calc doesn't handle multiple pages per block */ - BUILD_BUG_ON(PAGE_SIZE < SCOUTFS_BLOCK_SIZE); - - pages = kcalloc(NR_PAGES, sizeof(struct page *), GFP_NOFS); - if (!pages) - return -ENOMEM; - - for (i = 0; i < NR_PAGES; i++) { - page = alloc_page(GFP_NOFS); - if (!page) { - ret = -ENOMEM; - goto out; - } - - pages[i] = page; - } - - index = le64_to_cpu(super->ring_index); - nr = le64_to_cpu(super->ring_nr); - seq = le64_to_cpu(super->ring_seq); - - while (nr) { - blkno = le64_to_cpu(super->ring_blkno) + index; - /* XXX min3_t should be a thing */ - part = min3(nr, (u64)NR_BLOCKS, - le64_to_cpu(super->ring_blocks) - index); - - trace_printk("index %llu part %llu\n", index, part); - - ret = scoutfs_bio_read(sb, pages, blkno, part); - if (ret) - goto out; - - /* XXX verify block header */ - - for (i = 0; i < part; i++) { - ring = scoutfs_page_block_address(pages, i); - ret = read_entries(sb, ring); - if (ret) - goto out; - } - - index = ring_ind_wrap(super, index + part); - nr -= part; - } - -out: - for (i = 0; i < NR_PAGES && pages && pages[i]; i++) - __free_page(pages[i]); - kfree(pages); - - return ret; -} - -int scoutfs_ring_setup(struct super_block *sb) -{ - struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct ring_info *rinf; - struct page *page; - int i; - - rinf = kzalloc(sizeof(struct ring_info), GFP_KERNEL); - if (!rinf) - return -ENOMEM; - sbi->ring_info = rinf; - - for (i = 0; i < ARRAY_SIZE(rinf->pages); i++) { - page = alloc_page(GFP_KERNEL); - if (!page) { - while (--i >= 0) - __free_page(rinf->pages[i]); - return -ENOMEM; - } - - rinf->pages[i] = page; - } - - return 0; -} - -void scoutfs_ring_destroy(struct super_block *sb) -{ - DECLARE_RING_INFO(sb, rinf); - int i; - - if (rinf) { - for (i = 0; i < ARRAY_SIZE(rinf->pages); i++) - __free_page(rinf->pages[i]); - - kfree(rinf); - } -} - diff --git a/kmod/src/ring.h b/kmod/src/ring.h deleted file mode 100644 index 94eb84c3..00000000 --- a/kmod/src/ring.h +++ /dev/null @@ -1,18 +0,0 @@ -#ifndef _SCOUTFS_RING_H_ -#define _SCOUTFS_RING_H_ - -#include - -struct scoutfs_bio_completion; - -int scoutfs_ring_read(struct super_block *sb); -void scoutfs_ring_append(struct super_block *sb, - struct scoutfs_ring_entry_header *eh); - -int scoutfs_ring_submit_write(struct super_block *sb, - struct scoutfs_bio_completion *comp); - -int scoutfs_ring_setup(struct super_block *sb); -void scoutfs_ring_destroy(struct super_block *sb); - -#endif diff --git a/kmod/src/scoutfs_trace.h b/kmod/src/scoutfs_trace.h index 8793a544..7b10ae90 100644 --- a/kmod/src/scoutfs_trace.h +++ b/kmod/src/scoutfs_trace.h @@ -349,15 +349,14 @@ DEFINE_EVENT(scoutfs_btree_ranged_op, scoutfs_btree_since, TRACE_EVENT(scoutfs_manifest_add, TP_PROTO(struct super_block *sb, struct kvec *first, - struct kvec *last, u64 segno, u64 seq, u8 level, bool dirty), - TP_ARGS(sb, first, last, segno, seq, level, dirty), + struct kvec *last, u64 segno, u64 seq, u8 level), + TP_ARGS(sb, first, last, segno, seq, level), TP_STRUCT__entry( __dynamic_array(char, first, scoutfs_kvec_key_strlen(first)) __dynamic_array(char, last, scoutfs_kvec_key_strlen(last)) __field(u64, segno) __field(u64, seq) __field(u8, level) - __field(u8, dirty) ), TP_fast_assign( scoutfs_kvec_key_sprintf(__get_dynamic_array(first), first); @@ -365,11 +364,10 @@ TRACE_EVENT(scoutfs_manifest_add, __entry->segno = segno; __entry->seq = seq; __entry->level = level; - __entry->dirty = dirty; ), - TP_printk("first %s last %s segno %llu seq %llu level %u dirty %u", + TP_printk("first %s last %s segno %llu seq %llu level %u", __get_str(first), __get_str(last), __entry->segno, - __entry->seq, __entry->level, __entry->dirty) + __entry->seq, __entry->level) ); TRACE_EVENT(scoutfs_item_lookup, diff --git a/kmod/src/seg.c b/kmod/src/seg.c index 37277096..c16c2e2d 100644 --- a/kmod/src/seg.c +++ b/kmod/src/seg.c @@ -540,7 +540,7 @@ int scoutfs_seg_manifest_add(struct super_block *sb, kvec_from_pages(seg, last, item.key_off, item.key_len); return scoutfs_manifest_add(sb, first, last, le64_to_cpu(sblk->segno), - le64_to_cpu(sblk->max_seq), level, true); + le64_to_cpu(sblk->max_seq), level); } int scoutfs_seg_setup(struct super_block *sb) diff --git a/kmod/src/super.c b/kmod/src/super.c index 5ce1a2b7..00cafba4 100644 --- a/kmod/src/super.c +++ b/kmod/src/super.c @@ -28,12 +28,12 @@ #include "counters.h" #include "trans.h" #include "buddy.h" -#include "ring.h" #include "item.h" #include "manifest.h" #include "seg.h" #include "bio.h" #include "alloc.h" +#include "treap.h" #include "scoutfs_trace.h" static struct kset *scoutfs_kset; @@ -228,8 +228,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) scoutfs_manifest_setup(sb) ?: scoutfs_item_setup(sb) ?: scoutfs_alloc_setup(sb) ?: - scoutfs_ring_setup(sb) ?: - scoutfs_ring_read(sb) ?: + scoutfs_treap_setup(sb) ?: // scoutfs_buddy_setup(sb) ?: scoutfs_setup_trans(sb); if (ret) @@ -269,8 +268,8 @@ 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_ring_destroy(sb); scoutfs_block_destroy(sb); scoutfs_destroy_counters(sb); if (sbi->kset) diff --git a/kmod/src/super.h b/kmod/src/super.h index bb803105..f68a1c66 100644 --- a/kmod/src/super.h +++ b/kmod/src/super.h @@ -12,7 +12,7 @@ struct buddy_info; struct item_cache; struct manifest; struct segment_cache; -struct ring_info; +struct treap_info; struct scoutfs_sb_info { struct super_block *sb; @@ -36,7 +36,7 @@ struct scoutfs_sb_info { struct item_cache *item_cache; struct segment_cache *segment_cache; struct seg_alloc *seg_alloc; - struct ring_info *ring_info; + struct treap_info *treap_info; struct buddy_info *buddy_info; diff --git a/kmod/src/trans.c b/kmod/src/trans.c index 1e23295f..04b1ed52 100644 --- a/kmod/src/trans.c +++ b/kmod/src/trans.c @@ -27,7 +27,7 @@ #include "manifest.h" #include "seg.h" #include "alloc.h" -#include "ring.h" +#include "treap.h" #include "scoutfs_trace.h" /* @@ -96,20 +96,15 @@ void scoutfs_trans_write_func(struct work_struct *work) scoutfs_filerw_free_alloc(sb); #endif - /* - * We only have to check if there are dirty items or manifest - * entries. You can't have dirty alloc regions without having - * changed references to the allocated segments which produces - * dirty manfiest entries. - */ - if (scoutfs_item_dirty_bytes(sb) || scoutfs_manifest_has_dirty(sb)) { + if (scoutfs_item_dirty_bytes(sb) || scoutfs_manifest_has_dirty(sb) || + scoutfs_alloc_has_dirty(sb)) { ret = scoutfs_seg_alloc(sb, &seg) ?: - scoutfs_item_dirty_seg(sb, seg); + scoutfs_item_dirty_seg(sb, seg) ?: scoutfs_seg_manifest_add(sb, seg, 0) ?: scoutfs_manifest_dirty_ring(sb) ?: scoutfs_alloc_dirty_ring(sb) ?: - scoutfs_ring_submit_write(sb, &comp) ?: + scoutfs_treap_submit_write(sb, &comp) ?: scoutfs_seg_submit_write(sb, seg, &comp) ?: scoutfs_bio_wait_comp(sb, &comp) ?: scoutfs_write_dirty_super(sb); diff --git a/kmod/src/treap.c b/kmod/src/treap.c new file mode 100644 index 00000000..81cfe343 --- /dev/null +++ b/kmod/src/treap.c @@ -0,0 +1,1271 @@ +/* + * 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; + 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)); +}; + +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. */ +static u8 node_aug_bits(struct scoutfs_treap *treap, struct treap_node *node) +{ + DECLARE_TREAP_INFO(treap->sb, tinf); + u8 aug_bits = 0; + + if (node->off == tinf->dirty_off) + aug_bits |= SCOUTFS_TREAP_AUG_DIRTY; + + return aug_bits | off_aug_bit(treap, node->off); +} + +/* + * 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) + parent_ref(treap, node)->aug_bits = 0; + + while (node) { + bits = node_aug_bits(treap, node); + ref = parent_ref(treap, node); + 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); +} + +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; + + treap->dirty_bytes += node_ring_bytes(node); + + 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; + parent_ref(treap, node)->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; +} + +/* + * Deletion is easy if the node to delete doesn't have both children. + * We just point its parent at its child, if it has one. If it has both + * children we relocate it in the tree so that it doesn't. We find its + * next least successor which by definition won't have a left child. We + * can swap them and maintain key ordering in the tree. But we have to + * repair balance and augmentation. + */ +int scoutfs_treap_delete(struct scoutfs_treap *treap, void *key) +{ + struct treap_ref *ref = &treap->root_ref; + struct treap_ref *child_ref; + struct treap_node *parent = NULL; + struct treap_node *node = NULL; + struct treap_node *child; + 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; + } + + /* if it has both children find next successor and swap */ + if (node->left.gen && node->right.gen) { + child_ref = &node->right; + child = read_node(treap, node, child_ref, true); + if (IS_ERR(child)) { + ret = PTR_ERR(child); + goto out; + } + + while (child->left.gen) { + child_ref = &child->left; + child = read_node(treap, child, child_ref, true); + if (IS_ERR(child)) { + ret = PTR_ERR(child); + goto out; + } + } + + /* + * ref points to node, child_ref points to the child. + * We swap the node and child's position in the tree by + * updating refs in and out of the nodes. + * + * Repair after deletion catches the aug and prio + * inconsistencies from moving the child up the tree and + * removing the node. + */ + + swap(*ref, *child_ref); + swap(node->parent, child->parent); + swap(node->left, child->left); + swap(node->right, child->right); + if (node->left.node) + node->left.node->parent = node; + if (node->right.node) + node->right.node->parent = node; + if (child->left.node) + child->left.node->parent = child; + if (child->right.node) + child->right.node->parent = child; + + } + + /* 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; +} + +static void *treap_lookup(struct scoutfs_treap *treap, void *key, bool dirty) +{ + 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, dirty); + if (IS_ERR(node)) + break; + + 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 (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, false); +} + +void *scoutfs_treap_lookup_dirty(struct scoutfs_treap *treap, void *key) +{ + return treap_lookup(treap, key, true); +} + +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_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->left.node) + node = parent; + node = parent; + +out: + if (IS_ERR(node)) + return ERR_CAST(node); + if (node) + return node->data; + return NULL; +} + +static void *lookup_next(struct scoutfs_treap *treap, void *key, bool dirty) +{ + struct treap_ref *ref = &treap->root_ref; + struct treap_node *parent = NULL; + struct treap_node *node = NULL; + struct treap_node *next = NULL; + int cmp; + + while (ref->off) { + node = read_node(treap, parent, ref, 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; + } else { + next = node; + break; + } + + parent = node; + node = NULL; + } + + if (IS_ERR(node)) + return ERR_CAST(node); + if (next) + return next->data; + return NULL; +} + +void *scoutfs_treap_lookup_next(struct scoutfs_treap *treap, void *key) +{ + return lookup_next(treap, key, false); +} + +void *scoutfs_treap_lookup_next_dirty(struct scoutfs_treap *treap, void *key) +{ + return lookup_next(treap, key, true); +} + +int scoutfs_treap_has_dirty(struct scoutfs_treap *treap) +{ + return !!(treap->root_ref.aug_bits & SCOUTFS_TREAP_AUG_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 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; + } + } + + treap->dirty_bytes = 0; + 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; +} + +void scoutfs_treap_update_root(struct scoutfs_treap_root *root, + struct scoutfs_treap *treap) +{ + 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; +} + +/* + * 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 new file mode 100644 index 00000000..0265611e --- /dev/null +++ b/kmod/src/treap.h @@ -0,0 +1,44 @@ +#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_update_root(struct scoutfs_treap_root *root, + struct scoutfs_treap *treap); +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_first(struct scoutfs_treap *treap); +void *scoutfs_treap_next(struct scoutfs_treap *treap, void *data); + +int scoutfs_treap_has_dirty(struct scoutfs_treap *treap); +int scoutfs_treap_dirty_ring(struct scoutfs_treap *treap); +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