From 70c7178e6a58a7df639253087236a351909035fa Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Wed, 21 Jun 2017 14:39:00 -0700 Subject: [PATCH] scoutfs: index segment items with skip list We want to be able to read a region of items from a segment by searching for the key that starts the item. In the first version of the segment format we find a key by performing a binary search across an array of offsets that point to the items. Unfortunately the current format requires that we know the number of items before we start writing. With thousands of items per segment it's a little bonkers to ask compaction to walk through all the items twice. Worse still, we didn't want the item offset array entries to span pages so they're rounded up to a power of two after having seqs and offsets and lengths. This makes them surprisingly large and sometimes they can consume up to 60% (!) of a segment. We know that we're inserting in sort order so it's very easy to build an index as we insert. Skip lists give us a nice simple way to ensure o(log n) lookups with only an average of two links per node. CPU use is greatly reduced by removing a full redundant item walk and we know use up almost all of the space in segments. There's still little gaps at the ends of blocks as item's still won't cross block boundaries. Most of this change is safely mechanical. The big difference is in how the compaction loop is built. It used to count the items before hand. It would never try to append when out of segments and writing would stop after the exact number of items. Now it discovers its out of items by allocating and trying to append and finding that there's no more work to do. It required rethinking the loop exit and segment allocation and stopping conditions. Signed-off-by: Zach Brown --- kmod/src/compact.c | 152 +++++------------- kmod/src/format.h | 34 ++-- kmod/src/item.c | 64 +------- kmod/src/manifest.c | 17 +- kmod/src/seg.c | 382 +++++++++++++++++++++++++------------------- kmod/src/seg.h | 15 +- 6 files changed, 308 insertions(+), 356 deletions(-) diff --git a/kmod/src/compact.c b/kmod/src/compact.c index b129b931..9bcefa30 100644 --- a/kmod/src/compact.c +++ b/kmod/src/compact.c @@ -70,8 +70,7 @@ struct compact_seg { struct scoutfs_key_buf *first; struct scoutfs_key_buf *last; struct scoutfs_segment *seg; - int pos; - int saved_pos; + int off; bool part_of_move; }; @@ -90,12 +89,12 @@ struct compact_cursor { u8 last_level; struct compact_seg *upper; - struct compact_seg *saved_upper; struct compact_seg *lower; - struct compact_seg *saved_lower; bool sticky; struct compact_seg *last_lower; + + __le32 *links[SCOUTFS_MAX_SKIP_LINKS]; }; static void free_cseg(struct super_block *sb, struct compact_seg *cseg) @@ -140,28 +139,6 @@ static void free_cseg_list(struct super_block *sb, struct list_head *list) } } -static void save_pos(struct compact_cursor *curs) -{ - struct compact_seg *cseg; - - list_for_each_entry(cseg, &curs->csegs, entry) - cseg->saved_pos = cseg->pos; - - curs->saved_upper = curs->upper; - curs->saved_lower = curs->lower; -} - -static void restore_pos(struct compact_cursor *curs) -{ - struct compact_seg *cseg; - - list_for_each_entry(cseg, &curs->csegs, entry) - cseg->pos = cseg->saved_pos; - - curs->upper = curs->saved_upper; - curs->lower = curs->saved_lower; -} - static int read_segment(struct super_block *sb, struct compact_seg *cseg) { struct scoutfs_segment *seg; @@ -216,7 +193,7 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs, retry: if (upper) { - ret = scoutfs_seg_item_ptrs(upper->seg, upper->pos, + ret = scoutfs_seg_item_ptrs(upper->seg, upper->off, item_key, item_val, item_flags); if (ret < 0) upper = NULL; @@ -227,7 +204,7 @@ retry: if (ret) goto out; - ret = scoutfs_seg_item_ptrs(lower->seg, lower->pos, + ret = scoutfs_seg_item_ptrs(lower->seg, lower->off, &lower_key, lower_val, &lower_flags); if (ret == 0) @@ -273,9 +250,9 @@ retry: } if (cmp <= 0) - upper->pos++; + upper->off = scoutfs_seg_next_off(upper->seg, upper->off); if (cmp >= 0) - lower->pos++; + lower->off = scoutfs_seg_next_off(lower->seg, lower->off); /* * Deletion items make their way down all the levels, replacing @@ -296,64 +273,38 @@ out: } /* - * Figure out how many items and bytes of keys we're going to try and - * compact into the next segment. + * Walk the input segments for items and append them to the output segment. + * Items can exist in the input segments but not be written to the output + * segment, for example if they're deletions. The output segment can be + * full. + * + * Return -errno if something went wrong, then 1 or 0 indicating items written. */ -static int count_items(struct super_block *sb, struct compact_cursor *curs, - u32 *nr_items, u32 *key_bytes) -{ - struct scoutfs_key_buf item_key; - SCOUTFS_DECLARE_KVEC(item_val); - u32 items = 0; - u32 keys = 0; - u32 vals = 0; - u8 flags; - int ret; - - *nr_items = 0; - *key_bytes = 0; - - while ((ret = next_item(sb, curs, &item_key, item_val, &flags)) > 0) { - - items++; - keys += item_key.key_len; - vals += scoutfs_kvec_length(item_val); - - if (!scoutfs_seg_fits_single(items, keys, vals)) - break; - - *nr_items = items; - *key_bytes = keys; - } - - return ret; -} - static int compact_items(struct super_block *sb, struct compact_cursor *curs, - struct scoutfs_segment *seg, u32 nr_items, - u32 key_bytes) + struct scoutfs_segment *seg) { struct scoutfs_key_buf item_key; SCOUTFS_DECLARE_KVEC(item_val); + int has_next; + int ret = 0; u8 flags; - int ret; - ret = next_item(sb, curs, &item_key, item_val, &flags); - if (ret <= 0) - goto out; - - scoutfs_seg_first_item(sb, seg, &item_key, item_val, flags, - nr_items, key_bytes); - - while (--nr_items) { - ret = next_item(sb, curs, &item_key, item_val, &flags); - if (ret <= 0) + for (;;) { + has_next = next_item(sb, curs, &item_key, item_val, &flags); + if (has_next <= 0) { + if (has_next < 0) + ret = has_next; break; - scoutfs_seg_append_item(sb, seg, &item_key, item_val, flags); + } + + if (scoutfs_seg_append_item(sb, seg, &item_key, item_val, flags, + curs->links)) + ret = 1; + else + break; } -out: return ret; } @@ -367,15 +318,14 @@ static int compact_segments(struct super_block *sb, struct compact_seg *upper; struct compact_seg *lower; unsigned next_segno = 0; - u32 key_bytes; - u32 nr_items; - int ret; + int ret = 0; scoutfs_inc_counter(sb, compact_operations); if (curs->sticky) scoutfs_inc_counter(sb, compact_sticky_upper); - for (;;) { + while (curs->upper || curs->lower) { + upper = curs->upper; lower = curs->lower; @@ -389,7 +339,7 @@ static int compact_segments(struct super_block *sb, * XXX We should have metadata in the manifest to tell * us that there's no deletion items in the segment. */ - if (upper && upper->pos == 0 && !lower && !curs->sticky && + if (upper && upper->off == 0 && !lower && !curs->sticky && ((upper->level + 1) < curs->last_level)) { /* @@ -417,9 +367,9 @@ static int compact_segments(struct super_block *sb, cseg->part_of_move = true; curs->upper = NULL; - upper = NULL; scoutfs_inc_counter(sb, compact_segment_moved); + break; } /* we're going to need its next key */ @@ -444,17 +394,6 @@ static int compact_segments(struct super_block *sb, if (ret) break; - save_pos(curs); - ret = count_items(sb, curs, &nr_items, &key_bytes); - restore_pos(curs); - if (ret < 0) - break; - - if (nr_items == 0) { - ret = 0; - break; - } - /* no cseg keys, manifest update uses seg item keys */ cseg = kzalloc(sizeof(struct compact_seg), GFP_NOFS); if (!cseg) { @@ -466,11 +405,19 @@ static int compact_segments(struct super_block *sb, curs->segnos[next_segno] = 0; next_segno++; + /* + * Compaction can free all the remaining items resulting + * in an empty output segment. We just free it in that + * case. + */ ret = scoutfs_seg_alloc(sb, cseg->segno, &seg); - if (ret) { + if (ret == 0) + ret = compact_items(sb, curs, seg); + if (ret < 1) { next_segno--; curs->segnos[next_segno] = cseg->segno; kfree(cseg); + scoutfs_seg_put(seg); break; } @@ -489,21 +436,6 @@ static int compact_segments(struct super_block *sb, cseg->seg = seg; list_add_tail(&cseg->entry, results); - ret = compact_items(sb, curs, seg, nr_items, key_bytes); - if (ret < 0) - break; - - /* - * Clear lower after we've consumed it so that sticky - * compaction can decide to write the rest of the items - * into the upper level. We decide that it's done by - * testing the pos that next_item() is going to try. - */ - if (curs->sticky && curs->lower == curs->last_lower && - scoutfs_seg_item_ptrs(curs->lower->seg, curs->lower->pos, - NULL, NULL, NULL) < 0) - curs->lower = NULL; - /* start a complete segment write now, we'll wait later */ ret = scoutfs_seg_submit_write(sb, seg, comp); if (ret) diff --git a/kmod/src/format.h b/kmod/src/format.h index c511d51a..6ef16073 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -112,19 +112,29 @@ struct scoutfs_alloc_region { } __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 - * can use pointers to them in the page vecs that make up segments without - * funny business. + * The max number of links defines the max number of entries that we can + * index in o(log n) and the static list head storage size in the + * segment block. We always pay the static storage cost, which is tiny, + * and we can look at the number of items to know the greatest number of + * links and skip most of the initial 0 links. + */ +#define SCOUTFS_MAX_SKIP_LINKS 32 + +/* + * Items are packed into segments and linked together in a skip list. + * Each item's header, links, key, and value are stored contiguously. + * They're not allowed to cross a block boundary. */ struct scoutfs_segment_item { - __le64 seq; - __le32 key_off; - __le32 val_off; __le16 key_len; __le16 val_len; - __u8 padding[11]; __u8 flags; + __u8 nr_links; + __le32 skip_links[0]; + /* + * u8 key_bytes[key_len] + * u8 val_bytes[val_len] + */ } __packed; #define SCOUTFS_ITEM_FLAG_DELETION (1 << 0) @@ -138,11 +148,11 @@ struct scoutfs_segment_block { __le32 _padding; __le64 segno; __le64 seq; + __le32 last_item_off; + __le32 total_bytes; __le32 nr_items; - __le32 _moar_pads; - struct scoutfs_segment_item items[0]; - /* packed keys */ - /* packed vals */ + __le32 skip_links[SCOUTFS_MAX_SKIP_LINKS]; + /* packed items */ } __packed; /* diff --git a/kmod/src/item.c b/kmod/src/item.c index 6135b0eb..977ae63d 100644 --- a/kmod/src/item.c +++ b/kmod/src/item.c @@ -1550,39 +1550,6 @@ bool scoutfs_item_dirty_fits_single(struct super_block *sb, u32 nr_items, return fits; } -/* - * Find the initial sorted dirty items that will fit in a segment. Give - * the caller the number of items and the total bytes of their keys. - */ -static void count_seg_items(struct item_cache *cac, u32 *nr_items, - u32 *key_bytes) -{ - struct cached_item *item; - u32 items = 0; - u32 keys = 0; - u32 vals = 0; - - *nr_items = 0; - *key_bytes = 0; - - for (item = first_dirty(cac->items.rb_node); item; - item = next_dirty(item)) { - - items++; - keys += item->key->key_len; - vals += scoutfs_kvec_length(item->val); - - if (!scoutfs_seg_fits_single(items, keys, vals)) - break; - - *nr_items = items; - *key_bytes = keys; - - trace_printk("counted item %p nr %u keys %u\n", - item, items, keys); - } -} - /* * Fill the given segment with sorted dirty items. * @@ -1597,33 +1564,20 @@ int scoutfs_item_dirty_seg(struct super_block *sb, struct scoutfs_segment *seg) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct item_cache *cac = sbi->item_cache; + __le32 *links[SCOUTFS_MAX_SKIP_LINKS]; struct cached_item *item = NULL; struct cached_item *del; unsigned long flags; - u32 key_bytes; - u32 nr_items; + bool appended; spin_lock_irqsave(&cac->lock, flags); - count_seg_items(cac, &nr_items, &key_bytes); - - /* remember nr_items is passed to _first_item */ - while (nr_items) { - - trace_printk("copying item %p nr %u keys %u\n", - item, nr_items, key_bytes); - - if (!item) { - item = first_dirty(cac->items.rb_node); - scoutfs_seg_first_item(sb, seg, item->key, item->val, - item_flags(item), nr_items, - key_bytes); - } else { - scoutfs_seg_append_item(sb, seg, item->key, item->val, - item_flags(item)); - } - - key_bytes -= item->key->key_len; + item = first_dirty(cac->items.rb_node); + while (item) { + appended = scoutfs_seg_append_item(sb, seg, item->key, item->val, + item_flags(item), links); + /* trans reservation should have limited dirty */ + BUG_ON(!appended); clear_item_dirty(sb, cac, item); @@ -1632,8 +1586,6 @@ int scoutfs_item_dirty_seg(struct super_block *sb, struct scoutfs_segment *seg) if (del->deletion) erase_item(sb, cac, del); - - nr_items--; } spin_unlock_irqrestore(&cac->lock, flags); diff --git a/kmod/src/manifest.c b/kmod/src/manifest.c index 51e20802..df6ba43f 100644 --- a/kmod/src/manifest.c +++ b/kmod/src/manifest.c @@ -67,7 +67,7 @@ struct manifest_ref { u64 seq; struct scoutfs_segment *seg; int found_ctr; - int pos; + int off; u8 level; struct scoutfs_key_buf *first; @@ -542,7 +542,7 @@ int scoutfs_manifest_read_items(struct super_block *sb, /* start from the next item from the key in each segment */ list_for_each_entry(ref, &ref_list, entry) - ref->pos = scoutfs_seg_find_pos(ref->seg, key); + ref->off = scoutfs_seg_find_off(ref->seg, key); /* * Find the limit of the range we can safely walk. We have all @@ -567,9 +567,9 @@ int scoutfs_manifest_read_items(struct super_block *sb, found = false; found_ctr++; - /* find the next least key from the pos in each segment */ + /* find the next least key from the off in each segment */ list_for_each_entry_safe(ref, tmp, &ref_list, entry) { - if (ref->pos == -1) + if (ref->off < 0) continue; /* @@ -578,12 +578,12 @@ int scoutfs_manifest_read_items(struct super_block *sb, * items or if the next item is past the keys * that our segments can see. */ - ret = scoutfs_seg_item_ptrs(ref->seg, ref->pos, + ret = scoutfs_seg_item_ptrs(ref->seg, ref->off, &item_key, item_val, &item_flags); if (ret < 0 || - scoutfs_key_compare(&item_key, &seg_end) > 0){ - ref->pos = -1; + scoutfs_key_compare(&item_key, &seg_end) > 0) { + ref->off = -1; continue; } @@ -645,7 +645,8 @@ int scoutfs_manifest_read_items(struct super_block *sb, /* advance all the positions that had the found key */ list_for_each_entry(ref, &ref_list, entry) { if (ref->found_ctr == found_ctr) - ref->pos++; + ref->off = scoutfs_seg_next_off(ref->seg, + ref->off); } ret = 0; diff --git a/kmod/src/seg.c b/kmod/src/seg.c index 1441d152..8437d98b 100644 --- a/kmod/src/seg.c +++ b/kmod/src/seg.c @@ -265,6 +265,10 @@ int scoutfs_seg_alloc(struct super_block *sb, u64 segno, /* reads shouldn't wait for this */ set_bit(SF_END_IO, &seg->flags); + /* zero the block header so the caller knows to initialize */ + memset(page_address(seg->pages[0]), 0, + sizeof(struct scoutfs_segment_block)); + /* XXX always remove existing segs, is that necessary? */ spin_lock_irqsave(&cac->lock, flags); @@ -371,22 +375,6 @@ static void *off_ptr(struct scoutfs_segment *seg, u32 off) return page_address(seg->pages[pg]) + pg_off; } -static u32 pos_off(u32 pos) -{ - /* items need of be a power of two */ - BUILD_BUG_ON(!is_power_of_2(sizeof(struct scoutfs_segment_item))); - /* and the first item has to be naturally aligned */ - BUILD_BUG_ON(offsetof(struct scoutfs_segment_block, items) % - sizeof(struct scoutfs_segment_item)); - - return offsetof(struct scoutfs_segment_block, items[pos]); -} - -static void *pos_ptr(struct scoutfs_segment *seg, u32 pos) -{ - return off_ptr(seg, pos_off(pos)); -} - static void kvec_from_pages(struct scoutfs_segment *seg, struct kvec *kvec, u32 off, u16 len) { @@ -401,118 +389,225 @@ static void kvec_from_pages(struct scoutfs_segment *seg, off_ptr(seg, off + first), len - first); } -int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos, +static u32 item_bytes(u8 nr_links, u16 key_len, u16 val_len) +{ + return offsetof(struct scoutfs_segment_item, skip_links[nr_links]) + + key_len + val_len; +} + +static inline int item_key_off(struct scoutfs_segment_item *item, int item_off) +{ + return item_off + item_bytes(item->nr_links, 0, 0); +} + +static inline void *item_key_ptr(struct scoutfs_segment_item *item) +{ + return (void *)item + item_bytes(item->nr_links, 0, 0); +} + +static inline int item_val_off(struct scoutfs_segment_item *item, int item_off) +{ + return item_key_off(item, item_off) + le16_to_cpu(item->key_len); +} + +static void item_ptrs(struct scoutfs_segment *seg, int off, + struct scoutfs_key_buf *key, struct kvec *val) +{ + struct scoutfs_segment_item *item = off_ptr(seg, off); + + if (key) + scoutfs_key_init(key, item_key_ptr(item), + le16_to_cpu(item->key_len)); + if (val) + kvec_from_pages(seg, val, item_val_off(item, off), + le16_to_cpu(item->val_len)); +} + +static void first_last_keys(struct scoutfs_segment *seg, + struct scoutfs_key_buf *first, + struct scoutfs_key_buf *last) +{ + struct scoutfs_segment_block *sblk = off_ptr(seg, 0); + + item_ptrs(seg, sizeof(struct scoutfs_segment_block), first, NULL); + item_ptrs(seg, le32_to_cpu(sblk->last_item_off), last, NULL); +} + +static int check_caller_off(struct scoutfs_segment_block *sblk, int off) +{ + if (off >= 0 && off < sizeof(struct scoutfs_segment_block)) + off = sizeof(struct scoutfs_segment_block); + + if (off > le32_to_cpu(sblk->last_item_off)) + off = -ENOENT; + + return off; +} + +/* + * Give the caller the key and value of the item at the given offset. + * + * Negative offsets are sticky errors and offsets outside the used bytes + * in the segment return -ENOENT; + * + * All other offsets must be initial values less than the segment header + * size, notably including 0, or returned from _next_off(). + */ +int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int off, struct scoutfs_key_buf *key, struct kvec *val, u8 *flags) { struct scoutfs_segment_block *sblk = off_ptr(seg, 0); struct scoutfs_segment_item *item; - if (pos < 0 || pos >= le32_to_cpu(sblk->nr_items)) - return -ENOENT; + off = check_caller_off(sblk, off); + if (off < 0) + return off; - item = pos_ptr(seg, pos); + item_ptrs(seg, off, key, val); - if (key) - scoutfs_key_init(key, off_ptr(seg, le32_to_cpu(item->key_off)), - le16_to_cpu(item->key_len)); - if (val) - kvec_from_pages(seg, val, le32_to_cpu(item->val_off), - le16_to_cpu(item->val_len)); - if (flags) + if (flags) { + item = off_ptr(seg, off); *flags = item->flags; + } return 0; } /* - * Find the first item array position whose key is >= the search key. - * This can return the number of positions if the key is greater than - * all the keys. + * Return the number of links that the *next* added node should have. + * We're appending in order so we can use the low bits of the node count + * to get an ideal distribution of the number of links to enable (log n) + * searching: of links in each node. Half of the nodes will have 1 + * links, a quarter will have 2, an eighth will have 3, and so on. */ -static int find_key_pos(struct scoutfs_segment *seg, - struct scoutfs_key_buf *search) +static u8 skip_next_nr(u32 nr_items) { - struct scoutfs_segment_block *sblk = off_ptr(seg, 0); - struct scoutfs_key_buf key; - unsigned int start = 0; - unsigned int end = le32_to_cpu(sblk->nr_items); - unsigned int pos = 0; - int cmp; - - while (start < end) { - pos = start + (end - start) / 2; - scoutfs_seg_item_ptrs(seg, pos, &key, NULL, NULL); - - cmp = scoutfs_key_compare(search, &key); - if (cmp < 0) - end = pos; - else if (cmp > 0) - start = ++pos; - else - break; - } - - return pos; + return ffs(nr_items + 1); } -int scoutfs_seg_find_pos(struct scoutfs_segment *seg, +/* The highest 1-based set bit is the max number of links any node can have */ +static u8 skip_most_nr(u32 nr_items) +{ + return fls(nr_items); +} + +/* + * Find offset of the first item in the segment whose key is greater + * than or equal to the search key. -ENOENT is returned if there's no + * item that matches. + * + * This is a standard skip list search from the segment block through + * the items. Follow high less frequent links while the key is greater + * than the items and descend down to lower more frequent links when the + * search key is less. + */ +int scoutfs_seg_find_off(struct scoutfs_segment *seg, struct scoutfs_key_buf *key) { - return find_key_pos(seg, key); + struct scoutfs_segment_block *sblk = off_ptr(seg, 0); + struct scoutfs_segment_item *item; + struct scoutfs_key_buf item_key; + __le32 *links; + int cmp; + int ret; + int i; + int off; + + links = sblk->skip_links; + ret = -ENOENT; + for (i = skip_most_nr(le32_to_cpu(sblk->nr_items)) - 1; i >= 0; i--) { + if (links[i] == 0) + continue; + + off = le32_to_cpu(links[i]); + item = off_ptr(seg, off); + scoutfs_key_init(&item_key, item_key_ptr(item), + le16_to_cpu(item->key_len)); + + cmp = scoutfs_key_compare(key, &item_key); + if (cmp == 0) { + ret = off; + break; + } + + if (cmp > 0) { + links = item->skip_links; + i++; + } else { + ret = off; + } + } + + return ret; } /* - * Keys are aligned to the next block boundary if they'd cross a block - * boundary. To find the first value offset we have to assume that - * there will be a worst case key alignment at every block boundary. + * Return the offset of the next item after the current item. The input offset + * must be a valid offset from _find_off(). */ -static u32 first_val_off(u32 nr_items, u32 key_bytes) +int scoutfs_seg_next_off(struct scoutfs_segment *seg, int off) { - u32 key_padding = SCOUTFS_MAX_KEY_SIZE - 1; - u32 partial_block = SCOUTFS_BLOCK_SIZE - key_padding; - u32 first_key_off = pos_off(nr_items); - u32 block_off = first_key_off & SCOUTFS_BLOCK_MASK; - u32 total_padding = ((block_off + key_bytes) / partial_block) * - key_padding; + struct scoutfs_segment_block *sblk = off_ptr(seg, 0); + struct scoutfs_segment_item *item; - return first_key_off + key_bytes + total_padding; + off = check_caller_off(sblk, off); + if (off > 0) { + item = off_ptr(seg, off); + off = le32_to_cpu(item->skip_links[0]); + if (off == 0) + off = -ENOENT; + } + return off; } /* - * Returns true if the given number of items with the given total byte - * counts of keys and values fits inside a single segment. + * Returns true if the given item population will fit in a single + * segment. + * + * We don't have items cross block boundaries. It would be too + * expensive to maintain packing of sorted dirty items in bins. Instead + * we assume that we'll lose the worst case largest possible item on every + * block transition. This will almost never be the case. This causes us + * to lose around 15% of space for level 0 segment writes. + * + * Our pattern of item link counts ensures that there will always be fewer + * than two links per item. We assume the worst case items have the + * max number of links. */ bool scoutfs_seg_fits_single(u32 nr_items, u32 key_bytes, u32 val_bytes) { - return (first_val_off(nr_items, key_bytes) + val_bytes) + u32 header = sizeof(struct scoutfs_segment_block); + u32 items = nr_items * item_bytes(2, 0, 0); + u32 item_pad = item_bytes(skip_most_nr(nr_items), SCOUTFS_MAX_KEY_SIZE, + SCOUTFS_MAX_VAL_SIZE) - 1; + u32 padding = (SCOUTFS_SEGMENT_SIZE / SCOUTFS_BLOCK_SIZE) * item_pad; + + return (header + items + key_bytes + val_bytes + padding) <= SCOUTFS_SEGMENT_SIZE; } -static u32 align_key_off(struct scoutfs_segment *seg, u32 key_off, u32 len) +static u32 align_item_off(struct scoutfs_segment *seg, u32 item_off, u32 bytes) { - u32 space = SCOUTFS_BLOCK_SIZE - (key_off & SCOUTFS_BLOCK_MASK); + u32 space = SCOUTFS_BLOCK_SIZE - (item_off & SCOUTFS_BLOCK_MASK); - if (len > space) { - memset(off_ptr(seg, key_off), 0, space); - return key_off + space; + if (bytes > space) { + memset(off_ptr(seg, item_off), 0, space); + return item_off + space; } - return key_off; + return item_off; } + /* - * Store the first item in the segment. The caller knows the number - * of items and bytes of keys that determine where the keys and values - * start. Future items are appended by looking at the last item. - * - * This should never fail because any item must always fit in a segment. + * Append an item to the segment. The caller always appends items that + * have been sorted by their keys. They may not know how many will fit. + * We return true if we appended and false if the segment was full. */ -void scoutfs_seg_first_item(struct super_block *sb, - struct scoutfs_segment *seg, - struct scoutfs_key_buf *key, struct kvec *val, - u8 flags, unsigned int nr_items, - unsigned int key_bytes) +bool scoutfs_seg_append_item(struct super_block *sb, struct scoutfs_segment *seg, + struct scoutfs_key_buf *key, struct kvec *val, + u8 flags, __le32 **links) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; @@ -520,82 +615,66 @@ void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment_item *item; struct scoutfs_key_buf item_key; SCOUTFS_DECLARE_KVEC(item_val); - u32 key_off; - u32 val_off; + u8 nr_links; + u32 val_len; + u32 bytes; + u32 off; + int i; - /* XXX the segment block header is a mess, be better */ - sblk->segno = cpu_to_le64(seg->segno); - sblk->seq = super->next_seg_seq; - le64_add_cpu(&super->next_seg_seq, 1); + val_len = scoutfs_kvec_length(val); - key_off = align_key_off(seg, pos_off(nr_items), key->key_len); - val_off = first_val_off(nr_items, key_bytes); + /* initialize the segment and skip links as the first item is appended */ + if (sblk->nr_items == 0) { + /* XXX the segment block header is a mess, be better */ + sblk->segno = cpu_to_le64(seg->segno); + sblk->seq = super->next_seg_seq; + le64_add_cpu(&super->next_seg_seq, 1); + sblk->total_bytes = cpu_to_le32(sizeof(*sblk)); - sblk->nr_items = cpu_to_le32(1); - - trace_printk("first item offs key %u val %u\n", key_off, val_off); - - item = pos_ptr(seg, 0); - item->seq = cpu_to_le64(1); - item->key_off = cpu_to_le32(key_off); - item->val_off = cpu_to_le32(val_off); - item->key_len = cpu_to_le16(key->key_len); - item->val_len = cpu_to_le16(scoutfs_kvec_length(val)); - item->flags = flags; - - scoutfs_seg_item_ptrs(seg, 0, &item_key, item_val, NULL); - scoutfs_key_copy(&item_key, key); - scoutfs_kvec_memcpy(item_val, val); -} - -void scoutfs_seg_append_item(struct super_block *sb, - struct scoutfs_segment *seg, - struct scoutfs_key_buf *key, struct kvec *val, - u8 flags) -{ - struct scoutfs_segment_block *sblk = off_ptr(seg, 0); - struct scoutfs_segment_item *item; - struct scoutfs_segment_item *prev; - struct scoutfs_key_buf item_key; - SCOUTFS_DECLARE_KVEC(item_val); - u32 key_off; - u32 val_off; - u32 pos; - - pos = le32_to_cpu(sblk->nr_items); - sblk->nr_items = cpu_to_le32(pos + 1); + for (i = 0; i < SCOUTFS_MAX_SKIP_LINKS; i++) + links[i] = &sblk->skip_links[i]; + } /* * It's very bad data corruption if we write out of order items * to a segment. It'll mislead the key search during read and * stop it from finding its items. */ - if (pos) { - scoutfs_seg_item_ptrs(seg, pos - 1, &item_key, NULL, NULL); + off = le32_to_cpu(sblk->last_item_off); + if (off) { + item_ptrs(seg, off, &item_key, NULL); BUG_ON(scoutfs_key_compare(key, &item_key) <= 0); } - prev = pos_ptr(seg, pos - 1); - item = pos_ptr(seg, pos); + nr_links = skip_next_nr(le32_to_cpu(sblk->nr_items)); + bytes = item_bytes(nr_links, key->key_len, val_len); + off = align_item_off(seg, le32_to_cpu(sblk->total_bytes), bytes); - key_off = le32_to_cpu(prev->key_off) + le16_to_cpu(prev->key_len); - val_off = le32_to_cpu(prev->val_off) + le16_to_cpu(prev->val_len); + if ((off + bytes) > SCOUTFS_SEGMENT_SIZE) + return false; - key_off = align_key_off(seg, key_off, key->key_len); + sblk->last_item_off = cpu_to_le32(off); + sblk->total_bytes = cpu_to_le32(off + bytes); + le32_add_cpu(&sblk->nr_items, 1); - item->seq = cpu_to_le64(1); - item->key_off = cpu_to_le32(key_off); - item->val_off = cpu_to_le32(val_off); + item = off_ptr(seg, off); item->key_len = cpu_to_le16(key->key_len); - item->val_len = cpu_to_le16(scoutfs_kvec_length(val)); + item->val_len = cpu_to_le16(val_len); item->flags = flags; - trace_printk("item %u offs key %u val %u\n", - pos, key_off, val_off); + /* point the previous skip links at our appended item */ + item->nr_links = nr_links; + for (i = 0; i < nr_links; i++) { + item->skip_links[i] = 0; + *links[i] = cpu_to_le32(off); + links[i] = &item->skip_links[i]; + } - scoutfs_seg_item_ptrs(seg, pos, &item_key, item_val, NULL); + item_ptrs(seg, off, &item_key, item_val); scoutfs_key_copy(&item_key, key); scoutfs_kvec_memcpy(item_val, val); + + return true; } /* @@ -605,17 +684,10 @@ int scoutfs_seg_manifest_add(struct super_block *sb, struct scoutfs_segment *seg, u8 level) { struct scoutfs_segment_block *sblk = off_ptr(seg, 0); - struct scoutfs_segment_item *item; struct scoutfs_key_buf first; struct scoutfs_key_buf last; - item = pos_ptr(seg, 0); - scoutfs_key_init(&first, off_ptr(seg, le32_to_cpu(item->key_off)), - le16_to_cpu(item->key_len)); - - item = pos_ptr(seg, le32_to_cpu(sblk->nr_items) - 1); - scoutfs_key_init(&last, off_ptr(seg, le32_to_cpu(item->key_off)), - le16_to_cpu(item->key_len)); + first_last_keys(seg, &first, &last); return scoutfs_manifest_add(sb, &first, &last, le64_to_cpu(sblk->segno), le64_to_cpu(sblk->seq), level); @@ -625,12 +697,9 @@ int scoutfs_seg_manifest_del(struct super_block *sb, struct scoutfs_segment *seg, u8 level) { struct scoutfs_segment_block *sblk = off_ptr(seg, 0); - struct scoutfs_segment_item *item; struct scoutfs_key_buf first; - item = pos_ptr(seg, 0); - scoutfs_key_init(&first, off_ptr(seg, le32_to_cpu(item->key_off)), - le16_to_cpu(item->key_len)); + first_last_keys(seg, &first, NULL); return scoutfs_manifest_del(sb, &first, le64_to_cpu(sblk->seq), level); } @@ -644,17 +713,10 @@ scoutfs_seg_manifest_entry(struct super_block *sb, struct scoutfs_segment *seg, u8 level) { struct scoutfs_segment_block *sblk = off_ptr(seg, 0); - struct scoutfs_segment_item *item; struct scoutfs_key_buf first; struct scoutfs_key_buf last; - item = pos_ptr(seg, 0); - scoutfs_key_init(&first, off_ptr(seg, le32_to_cpu(item->key_off)), - le16_to_cpu(item->key_len)); - - item = pos_ptr(seg, le32_to_cpu(sblk->nr_items) - 1); - scoutfs_key_init(&last, off_ptr(seg, le32_to_cpu(item->key_off)), - le16_to_cpu(item->key_len)); + first_last_keys(seg, &first, &last); return scoutfs_manifest_alloc_entry(sb, &first, &last, le64_to_cpu(sblk->segno), diff --git a/kmod/src/seg.h b/kmod/src/seg.h index 5ce0c076..fbe11d88 100644 --- a/kmod/src/seg.h +++ b/kmod/src/seg.h @@ -10,9 +10,10 @@ struct scoutfs_segment *scoutfs_seg_submit_read(struct super_block *sb, u64 segno); int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg); -int scoutfs_seg_find_pos(struct scoutfs_segment *seg, +int scoutfs_seg_find_off(struct scoutfs_segment *seg, struct scoutfs_key_buf *key); -int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos, +int scoutfs_seg_next_off(struct scoutfs_segment *seg, int off); +int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int off, struct scoutfs_key_buf *key, struct kvec *val, u8 *flags); @@ -24,15 +25,9 @@ int scoutfs_seg_alloc(struct super_block *sb, u64 segno, int scoutfs_seg_free_segno(struct super_block *sb, struct scoutfs_segment *seg); bool scoutfs_seg_fits_single(u32 nr_items, u32 key_bytes, u32 val_bytes); -void scoutfs_seg_first_item(struct super_block *sb, - struct scoutfs_segment *seg, - struct scoutfs_key_buf *key, struct kvec *val, - u8 flags, unsigned int nr_items, - unsigned int key_bytes); -void scoutfs_seg_append_item(struct super_block *sb, - struct scoutfs_segment *seg, +bool scoutfs_seg_append_item(struct super_block *sb, struct scoutfs_segment *seg, struct scoutfs_key_buf *key, struct kvec *val, - u8 flags); + u8 flags, __le32 **links); int scoutfs_seg_manifest_add(struct super_block *sb, struct scoutfs_segment *seg, u8 level); int scoutfs_seg_manifest_del(struct super_block *sb,