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