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