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,