diff --git a/kmod/src/compact.c b/kmod/src/compact.c index b6e0965d..a0f951cd 100644 --- a/kmod/src/compact.c +++ b/kmod/src/compact.c @@ -49,10 +49,6 @@ * Once the compaction is completed the manifest is updated to remove * the input segments and add the output segments. Here segment space * is reclaimed when the input items fit in fewer output segments. - * - * XXX today we only know how to skip duplicate individual items. We'll - * need to know how to skip lower based on upper range deletion items - * and to combine incremental update items. */ struct compact_info { @@ -85,6 +81,7 @@ struct compact_cursor { struct list_head csegs; u8 lower_level; + u8 last_level; struct compact_seg *upper; struct compact_seg *saved_upper; @@ -193,22 +190,25 @@ static struct compact_seg *next_spos(struct compact_cursor *curs, * the lowest key or the upper if they're the same. We advance the * cursor past the item that is returned. * - * XXX this will get fancier as we get range deletion items and incremental - * update items. + * XXX this will get fancier as we get range deletion items and + * incremental update items. */ static int next_item(struct super_block *sb, struct compact_cursor *curs, - struct scoutfs_key_buf *item_key, struct kvec *item_val) + struct scoutfs_key_buf *item_key, struct kvec *item_val, + u8 *item_flags) { struct compact_seg *upper = curs->upper; struct compact_seg *lower = curs->lower; struct scoutfs_key_buf lower_key; SCOUTFS_DECLARE_KVEC(lower_val); + u8 lower_flags; int cmp; int ret; +retry: if (upper) { ret = scoutfs_seg_item_ptrs(upper->seg, upper->pos, - item_key, item_val); + item_key, item_val, item_flags); if (ret < 0) upper = NULL; } @@ -219,7 +219,8 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs, goto out; ret = scoutfs_seg_item_ptrs(lower->seg, lower->pos, - &lower_key, lower_val); + &lower_key, lower_val, + &lower_flags); if (ret == 0) break; lower = next_spos(curs, lower); @@ -246,6 +247,7 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs, if (cmp > 0) { scoutfs_key_clone(item_key, &lower_key); scoutfs_kvec_clone(item_val, lower_val); + *item_flags = lower_flags; } if (cmp <= 0) @@ -253,6 +255,16 @@ static int next_item(struct super_block *sb, struct compact_cursor *curs, if (cmp >= 0) lower->pos++; + /* + * Deletion items make their way down all the levels, replacing + * all the duplicate items that they find. When we're + * compacting to the last level we can remove them by retrying + * the search after we've advanced past them. + */ + if ((curs->lower_level == curs->last_level) && + ((*item_flags) & SCOUTFS_ITEM_FLAG_DELETION)) + goto retry; + ret = 1; out: curs->upper = upper; @@ -273,12 +285,13 @@ static int count_items(struct super_block *sb, struct compact_cursor *curs, 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)) > 0) { + while ((ret = next_item(sb, curs, &item_key, item_val, &flags)) > 0) { items++; keys += item_key.key_len; @@ -300,21 +313,22 @@ static int compact_items(struct super_block *sb, struct compact_cursor *curs, { struct scoutfs_key_buf item_key; SCOUTFS_DECLARE_KVEC(item_val); + u8 flags; int ret; - ret = next_item(sb, curs, &item_key, item_val); + 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, + 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); + ret = next_item(sb, curs, &item_key, item_val, &flags); if (ret <= 0) break; - scoutfs_seg_append_item(sb, seg, &item_key, item_val); + scoutfs_seg_append_item(sb, seg, &item_key, item_val, flags); } out: @@ -344,6 +358,12 @@ static int compact_segments(struct super_block *sb, /* * We can just move the upper segment down a level if it * doesn't intersect any lower segments. + * + * XXX we can't do this if the segment we're moving has + * deletion items. We need to copy the non-deletion items + * and drop the deletion items in that case. To do that + * we'll need the manifest to count the number of deletion + * and non-deletion items. */ if (upper && upper->pos == 0 && (!lower || @@ -383,11 +403,14 @@ static int compact_segments(struct super_block *sb, /* * We can skip a lower segment if there's no upper segment * or the next upper item is past the last in the lower. + * + * XXX this will need to test for intersection with range + * deletion items. */ if (lower && lower->pos == 0 && (!upper || (!scoutfs_seg_item_ptrs(upper->seg, upper->pos, - &upper_next, NULL) && + &upper_next, NULL, NULL) && scoutfs_key_compare(&upper_next, lower->last) > 0))) { curs->lower = next_spos(curs, lower); @@ -447,6 +470,25 @@ static int compact_segments(struct super_block *sb, return ret; } +/* + * Manifest walking is providing the details of the overall compaction + * operation. It'll then add all the segments involved. + */ +void scoutfs_compact_describe(struct super_block *sb, void *data, + u8 upper_level, u8 last_level) +{ + struct compact_cursor *curs = data; + + curs->lower_level = upper_level + 1; + curs->last_level = last_level; +} + +/* + * Add a segment involved in the compaction operation. + * + * XXX Today we know that the caller is always adding only one upper segment + * and is then possibly adding all the lower overlapping segments. + */ int scoutfs_compact_add(struct super_block *sb, void *data, struct scoutfs_key_buf *first, struct scoutfs_key_buf *last, u64 segno, u64 seq, @@ -468,12 +510,10 @@ int scoutfs_compact_add(struct super_block *sb, void *data, cseg->seq = seq; cseg->level = level; - if (!curs->upper) { + if (!curs->upper) curs->upper = cseg; - } else if (!curs->lower) { + else if (!curs->lower) curs->lower = cseg; - curs->lower_level = level; - } ret = 0; out: diff --git a/kmod/src/compact.h b/kmod/src/compact.h index 5241ff11..d3654fd3 100644 --- a/kmod/src/compact.h +++ b/kmod/src/compact.h @@ -3,6 +3,8 @@ void scoutfs_compact_kick(struct super_block *sb); +void scoutfs_compact_describe(struct super_block *sb, void *data, + u8 upper_level, u8 last_level); int scoutfs_compact_add(struct super_block *sb, void *data, struct scoutfs_key_buf *first, struct scoutfs_key_buf *last, u64 segno, u64 seq, diff --git a/kmod/src/counters.h b/kmod/src/counters.h index a473523d..86787796 100644 --- a/kmod/src/counters.h +++ b/kmod/src/counters.h @@ -27,6 +27,7 @@ EXPAND_COUNTER(item_create) \ EXPAND_COUNTER(item_lookup_hit) \ EXPAND_COUNTER(item_lookup_miss) \ + EXPAND_COUNTER(item_delete) \ EXPAND_COUNTER(item_range_hit) \ EXPAND_COUNTER(item_range_miss) \ EXPAND_COUNTER(item_range_insert) diff --git a/kmod/src/format.h b/kmod/src/format.h index 901d4a66..a2931169 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -137,9 +137,12 @@ struct scoutfs_segment_item { __le32 val_off; __le16 key_len; __le16 val_len; - __u8 padding[12]; + __u8 padding[11]; + __u8 flags; } __packed; +#define SCOUTFS_ITEM_FLAG_DELETION (1 << 0) + /* * Each large segment starts with a segment block that describes the * rest of the blocks that make up the segment. diff --git a/kmod/src/item.c b/kmod/src/item.c index 85c4942b..b3300598 100644 --- a/kmod/src/item.c +++ b/kmod/src/item.c @@ -33,6 +33,11 @@ * that are completely described by the items. This lets it return * negative lookups cache hits for items that don't exist without having * to constantly perform expensive segment searches. + * + * Deletions are recorded with items in the rbtree which record the key + * of the deletion. They're removed once they're written to a level0 + * segment. While they're present in the cache we have to be careful to + * clobber them in creation and skip them in lookups. */ struct item_cache { @@ -57,7 +62,9 @@ struct cached_item { struct rb_node node; struct list_head entry; }; + long dirty; + unsigned deletion:1; struct scoutfs_key_buf *key; @@ -71,6 +78,38 @@ struct cached_range { struct scoutfs_key_buf *end; }; +static u8 item_flags(struct cached_item *item) +{ + return item->deletion ? SCOUTFS_ITEM_FLAG_DELETION : 0; +} + +static void free_item(struct super_block *sb, struct cached_item *item) +{ + if (!IS_ERR_OR_NULL(item)) { + scoutfs_key_free(sb, item->key); + scoutfs_kvec_kfree(item->val); + kfree(item); + } +} + +static struct cached_item *alloc_item(struct super_block *sb, + struct scoutfs_key_buf *key, + struct kvec *val) +{ + struct cached_item *item; + + item = kzalloc(sizeof(struct cached_item), GFP_NOFS); + if (item) { + item->key = scoutfs_key_dup(sb, key); + if (!item->key || scoutfs_kvec_dup_flatten(item->val, val)) { + free_item(sb, item); + item = NULL; + } + } + + return item; +} + /* * Walk the item rbtree and return the item found and the next and * prev items. @@ -105,6 +144,14 @@ static struct cached_item *walk_items(struct rb_root *root, return NULL; } +/* + * Look for the item with the given key. Callers of this are looking + * for existing items. They would just return -ENOENT from a deletion + * item if we gave it to them so we return null for deletion items. + * Callers that would remove a deletion item before inserting a new + * version of the item do so by having insert_item() replace existing + * deleted items on their behalf. + */ static struct cached_item *find_item(struct super_block *sb, struct rb_root *root, struct scoutfs_key_buf *key) @@ -115,6 +162,9 @@ static struct cached_item *find_item(struct super_block *sb, item = walk_items(root, key, &prev, &next); + if (item && item->deletion) + item = NULL; + if (item) scoutfs_inc_counter(sb, item_lookup_hit); else @@ -224,11 +274,64 @@ static const struct rb_augment_callbacks scoutfs_item_rb_cb = { }; /* - * Try to insert the given item. If there's already an item with the - * insertion key then return -EEXIST. + * The caller has changed an item's dirty bit. Its child dirty bits are + * still consistent. But its parent's bits might need to be updated. + * Its bits are consistent so we don't propagate from the node itself + * because it would immediately terminate. */ -static int insert_item(struct rb_root *root, struct cached_item *ins) +static void update_dirty_parents(struct cached_item *item) { + scoutfs_item_rb_propagate(rb_parent(&item->node), NULL); +} + +static void mark_item_dirty(struct item_cache *cac, + struct cached_item *item) +{ + if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node))) + return; + + if (item->dirty & ITEM_DIRTY) + return; + + item->dirty |= ITEM_DIRTY; + cac->nr_dirty_items++; + cac->dirty_key_bytes += item->key->key_len; + cac->dirty_val_bytes += scoutfs_kvec_length(item->val); + + update_dirty_parents(item); +} + +static void clear_item_dirty(struct item_cache *cac, + struct cached_item *item) +{ + if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node))) + return; + + if (!(item->dirty & ITEM_DIRTY)) + return; + + item->dirty &= ~ITEM_DIRTY; + cac->nr_dirty_items--; + cac->dirty_key_bytes -= item->key->key_len; + cac->dirty_val_bytes -= scoutfs_kvec_length(item->val); + + WARN_ON_ONCE(cac->nr_dirty_items < 0 || cac->dirty_key_bytes < 0 || + cac->dirty_val_bytes < 0); + + update_dirty_parents(item); +} + +/* + * Try to insert the given item. If there's already a non-deletion item + * with the insertion key then return -EEXIST. An existing deletion + * item is replaced and freed. + * + * The caller is responsible for marking the newly inserted item dirty. + */ +static int insert_item(struct super_block *sb, struct item_cache *cac, + struct cached_item *ins) +{ + struct rb_root *root = &cac->items; struct rb_node **node = &root->rb_node; struct rb_node *parent = NULL; struct cached_item *item; @@ -248,7 +351,13 @@ static int insert_item(struct rb_root *root, struct cached_item *ins) item->dirty |= RIGHT_DIRTY; node = &(*node)->rb_right; } else { - return -EEXIST; + if (!item->deletion) + return -EEXIST; + + clear_item_dirty(cac, item); + rb_replace_node(&item->node, &ins->node, root); + free_item(sb, item); + return 0; } } @@ -443,6 +552,43 @@ int scoutfs_item_lookup_exact(struct super_block *sb, return ret; } +/* + * Find the next item to return from the "_next" item interface. It's the + * next item from the key that isn't a deletion item and is within the + * bounds of the end of the cache and the caller's last key. + */ +static struct cached_item *item_for_next(struct rb_root *root, + struct scoutfs_key_buf *key, + struct scoutfs_key_buf *range_end, + struct scoutfs_key_buf *last) +{ + struct cached_item *item; + struct rb_node *node; + + /* limit by the lesser of the two */ + if (scoutfs_key_compare(range_end, last) < 0) + last = range_end; + + item = next_item(root, key); + while (item) { + if (scoutfs_key_compare(item->key, last) > 0) { + item = NULL; + break; + } + + if (!item->deletion) + break; + + node = rb_next(&item->node); + if (node) + item = container_of(node, struct cached_item, node); + else + item = NULL; + } + + return item; +} + /* * Return the next item starting with the given key, returning the last * key at the most. @@ -490,10 +636,8 @@ int scoutfs_item_next(struct super_block *sb, struct scoutfs_key_buf *key, /* see if we have a usable item in cache and before last */ cached = check_range(sb, &cac->ranges, key, range_end); - if (cached && (item = next_item(&cac->items, key)) && - scoutfs_key_compare(item->key, range_end) <= 0 && - scoutfs_key_compare(item->key, last) <= 0) { - + if (cached && (item = item_for_next(&cac->items, key, + range_end, last))) { scoutfs_key_copy(key, item->key); if (val) ret = scoutfs_kvec_memcpy(val, item->val); @@ -565,81 +709,6 @@ int scoutfs_item_next_same_min(struct super_block *sb, return ret; } -static void free_item(struct super_block *sb, struct cached_item *item) -{ - if (!IS_ERR_OR_NULL(item)) { - scoutfs_key_free(sb, item->key); - scoutfs_kvec_kfree(item->val); - kfree(item); - } -} - -/* - * The caller has changed an item's dirty bit. Its child dirty bits are - * still consistent. But its parent's bits might need to be updated. - * Its bits are consistent so we don't propagate from the node itself - * because it would immediately terminate. - */ -static void update_dirty_parents(struct cached_item *item) -{ - scoutfs_item_rb_propagate(rb_parent(&item->node), NULL); -} - -static void mark_item_dirty(struct item_cache *cac, - struct cached_item *item) -{ - if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node))) - return; - - if (item->dirty & ITEM_DIRTY) - return; - - item->dirty |= ITEM_DIRTY; - cac->nr_dirty_items++; - cac->dirty_key_bytes += item->key->key_len; - cac->dirty_val_bytes += scoutfs_kvec_length(item->val); - - update_dirty_parents(item); -} - -static void clear_item_dirty(struct item_cache *cac, - struct cached_item *item) -{ - if (WARN_ON_ONCE(RB_EMPTY_NODE(&item->node))) - return; - - if (!(item->dirty & ITEM_DIRTY)) - return; - - item->dirty &= ~ITEM_DIRTY; - cac->nr_dirty_items--; - cac->dirty_key_bytes -= item->key->key_len; - cac->dirty_val_bytes -= scoutfs_kvec_length(item->val); - - WARN_ON_ONCE(cac->nr_dirty_items < 0 || cac->dirty_key_bytes < 0 || - cac->dirty_val_bytes < 0); - - update_dirty_parents(item); -} - -static struct cached_item *alloc_item(struct super_block *sb, - struct scoutfs_key_buf *key, - struct kvec *val) -{ - struct cached_item *item; - - item = kzalloc(sizeof(struct cached_item), GFP_NOFS); - if (item) { - item->key = scoutfs_key_dup(sb, key); - if (!item->key || scoutfs_kvec_dup_flatten(item->val, val)) { - free_item(sb, item); - item = NULL; - } - } - - return item; -} - /* * Create a new dirty item in the cache. Returns -EEXIST if an item * already exists with the given key. @@ -660,7 +729,7 @@ int scoutfs_item_create(struct super_block *sb, struct scoutfs_key_buf *key, return -ENOMEM; spin_lock_irqsave(&cac->lock, flags); - ret = insert_item(&cac->items, item); + ret = insert_item(sb, cac, item); if (!ret) { scoutfs_inc_counter(sb, item_create); mark_item_dirty(cac, item); @@ -745,7 +814,7 @@ int scoutfs_item_insert_batch(struct super_block *sb, struct list_head *list, list_for_each_entry_safe(item, tmp, list, entry) { list_del(&item->entry); - if (insert_item(&cac->items, item)) + if (insert_item(sb, cac, item)) list_add(&item->entry, list); } @@ -871,12 +940,61 @@ out: } /* - * XXX how nice, it'd just creates a cached deletion item. It doesn't - * have to read. + * Delete an existing item with the given key. + * + * If a non-deletion item is present then we mark it dirty and deleted + * and free it's value. + * + * Returns -ENOENT if an item doesn't exist at the key. This forces us + * to read the item before creating a deletion item for it. XXX If we + * relaxed this we'd need to see if callers make use of -ENOENT and if + * there are any ways for userspace to overwhelm the system with + * deletion items for items that didn't exist in the first place. */ int scoutfs_item_delete(struct super_block *sb, struct scoutfs_key_buf *key) { - return WARN_ON_ONCE(-EINVAL); + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct item_cache *cac = sbi->item_cache; + struct scoutfs_key_buf *end; + struct cached_item *item; + SCOUTFS_DECLARE_KVEC(del_val); + unsigned long flags; + int ret; + + scoutfs_kvec_init_null(del_val); + + end = scoutfs_key_alloc(sb, SCOUTFS_MAX_KEY_SIZE); + if (!end) { + ret = -ENOMEM; + goto out; + } + + do { + spin_lock_irqsave(&cac->lock, flags); + + item = find_item(sb, &cac->items, key); + if (item) { + scoutfs_kvec_swap(item->val, del_val); + item->deletion = 1; + mark_item_dirty(cac, item); + scoutfs_inc_counter(sb, item_delete); + ret = 0; + } else if (check_range(sb, &cac->ranges, key, end)) { + ret = -ENOENT; + } else { + ret = -ENODATA; + } + + spin_unlock_irqrestore(&cac->lock, flags); + + } while (ret == -ENODATA && + (ret = scoutfs_manifest_read_items(sb, key, end)) == 0); + + scoutfs_key_free(sb, end); + scoutfs_kvec_kfree(del_val); +out: + trace_printk("ret %d\n", ret); + return ret; } /* @@ -1020,28 +1138,46 @@ static void count_seg_items(struct item_cache *cac, u32 *nr_items, * segments and can be partially visible if we only write the first * segment. We probably want to throttle trans enters once we have as * many dirty items as our atomic segment updates can write. + * + * XXX this first/append pattern will go away once we can write a stream + * of items to a segment without needing to know the item count to + * find the starting key and value offsets. */ 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; - struct cached_item *item; + struct cached_item *item = NULL; + struct cached_item *del; u32 key_bytes; u32 nr_items; count_seg_items(cac, &nr_items, &key_bytes); - item = first_dirty(cac->items.rb_node); - if (item) { - scoutfs_seg_first_item(sb, seg, item->key, item->val, - nr_items, key_bytes); - clear_item_dirty(cac, item); - nr_items--; - } + /* remember nr_items is passed to _first_item */ + while (nr_items) { + + 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)); + } - while (nr_items-- && (item = next_dirty(item))) { - scoutfs_seg_append_item(sb, seg, item->key, item->val); clear_item_dirty(cac, item); + + del = item; + item = next_dirty(item); + + if (del->deletion) { + rb_erase(&del->node, &cac->items); + free_item(sb, del); + } + + nr_items--; } return 0; diff --git a/kmod/src/manifest.c b/kmod/src/manifest.c index 98ac6234..f8c4edb5 100644 --- a/kmod/src/manifest.c +++ b/kmod/src/manifest.c @@ -459,6 +459,8 @@ int scoutfs_manifest_read_items(struct super_block *sb, struct manifest_ref *tmp; LIST_HEAD(ref_list); LIST_HEAD(batch); + u8 found_flags = 0; + u8 item_flags; int found_ctr; bool found; int ret = 0; @@ -534,7 +536,8 @@ int scoutfs_manifest_read_items(struct super_block *sb, * caller's end. */ ret = scoutfs_seg_item_ptrs(ref->seg, ref->pos, - &item_key, item_val); + &item_key, item_val, + &item_flags); if (ret < 0 || scoutfs_key_compare(&item_key, end) > 0){ ref->pos = -1; continue; @@ -554,6 +557,7 @@ int scoutfs_manifest_read_items(struct super_block *sb, /* remember new least key */ scoutfs_key_clone(&found_key, &item_key); scoutfs_kvec_clone(found_val, item_val); + found_flags = item_flags; ref->found_ctr = ++found_ctr; found = true; } @@ -566,15 +570,22 @@ int scoutfs_manifest_read_items(struct super_block *sb, } /* + * Add the next found item to the batch if it's not a + * deletion item. We still need to use their key to + * remember the end of the batch for negative caching. + * * If we fail to add an item we're done. If we already - * have items it's not a failure and the end of the cached - * range is the last successfully added item. + * have items it's not a failure and the end of the + * cached range is the last successfully added item. */ - ret = scoutfs_item_add_batch(sb, &batch, &found_key, found_val); - if (ret) { - if (n > 0) - ret = 0; - break; + if (!(found_flags & SCOUTFS_ITEM_FLAG_DELETION)) { + ret = scoutfs_item_add_batch(sb, &batch, &found_key, + found_val); + if (ret) { + if (n > 0) + ret = 0; + break; + } } /* the last successful key determines range end until run out */ @@ -699,6 +710,8 @@ int scoutfs_manifest_next_compact(struct super_block *sb, void *data) goto out; } + scoutfs_compact_describe(sb, data, level, mani->nr_levels - 1); + /* find the oldest level 0 or the next higher order level by key */ if (level == 0) { ment = scoutfs_treap_first(mani->treap); @@ -737,7 +750,7 @@ 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(mani->treap, &skey); + over = scoutfs_treap_lookup_next(mani->treap, &skey); /* and add a fanout's worth of lower overlapping segments */ for (i = 0; i < SCOUTFS_MANIFEST_FANOUT; i++) { diff --git a/kmod/src/seg.c b/kmod/src/seg.c index 4687b7af..b343cb5e 100644 --- a/kmod/src/seg.c +++ b/kmod/src/seg.c @@ -409,7 +409,8 @@ static void kvec_from_pages(struct scoutfs_segment *seg, } int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos, - struct scoutfs_key_buf *key, struct kvec *val) + struct scoutfs_key_buf *key, struct kvec *val, + u8 *flags) { struct scoutfs_segment_block *sblk = off_ptr(seg, 0); struct scoutfs_segment_item *item; @@ -425,6 +426,8 @@ int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos, if (val) kvec_from_pages(seg, val, le32_to_cpu(item->val_off), le16_to_cpu(item->val_len)); + if (flags) + *flags = item->flags; return 0; } @@ -446,7 +449,7 @@ static int find_key_pos(struct scoutfs_segment *seg, while (start < end) { pos = start + (end - start) / 2; - scoutfs_seg_item_ptrs(seg, pos, &key, NULL); + scoutfs_seg_item_ptrs(seg, pos, &key, NULL, NULL); cmp = scoutfs_key_compare(search, &key); if (cmp < 0) @@ -512,9 +515,11 @@ static u32 align_key_off(struct scoutfs_segment *seg, u32 key_off, u32 len) * * This should never fail because any item must always fit in a segment. */ -void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment *seg, +void scoutfs_seg_first_item(struct super_block *sb, + struct scoutfs_segment *seg, struct scoutfs_key_buf *key, struct kvec *val, - unsigned int nr_items, unsigned int key_bytes) + u8 flags, unsigned int nr_items, + unsigned int key_bytes) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); struct scoutfs_super_block *super = &sbi->super; @@ -543,15 +548,17 @@ void scoutfs_seg_first_item(struct super_block *sb, struct scoutfs_segment *seg, 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); + 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) + struct scoutfs_key_buf *key, struct kvec *val, + u8 flags) { struct scoutfs_segment_block *sblk = off_ptr(seg, 0); struct scoutfs_segment_item *item; @@ -578,11 +585,12 @@ void scoutfs_seg_append_item(struct super_block *sb, 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; trace_printk("item %u offs key %u val %u\n", pos, key_off, val_off); - scoutfs_seg_item_ptrs(seg, pos, &item_key, item_val); + scoutfs_seg_item_ptrs(seg, pos, &item_key, item_val, NULL); scoutfs_key_copy(&item_key, key); scoutfs_kvec_memcpy(item_val, val); } diff --git a/kmod/src/seg.h b/kmod/src/seg.h index e43b2268..15c6834f 100644 --- a/kmod/src/seg.h +++ b/kmod/src/seg.h @@ -13,7 +13,8 @@ int scoutfs_seg_wait(struct super_block *sb, struct scoutfs_segment *seg); int scoutfs_seg_find_pos(struct scoutfs_segment *seg, struct scoutfs_key_buf *key); int scoutfs_seg_item_ptrs(struct scoutfs_segment *seg, int pos, - struct scoutfs_key_buf *key, struct kvec *val); + struct scoutfs_key_buf *key, struct kvec *val, + u8 *flags); void scoutfs_seg_get(struct scoutfs_segment *seg); void scoutfs_seg_put(struct scoutfs_segment *seg); @@ -22,12 +23,15 @@ int scoutfs_seg_alloc(struct super_block *sb, struct scoutfs_segment **seg_ret); 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, +void scoutfs_seg_first_item(struct super_block *sb, + struct scoutfs_segment *seg, struct scoutfs_key_buf *key, struct kvec *val, - unsigned int nr_items, unsigned int key_bytes); + u8 flags, unsigned int nr_items, + unsigned int key_bytes); void scoutfs_seg_append_item(struct super_block *sb, struct scoutfs_segment *seg, - struct scoutfs_key_buf *key, struct kvec *val); + struct scoutfs_key_buf *key, struct kvec *val, + u8 flags); int scoutfs_seg_manifest_add(struct super_block *sb, struct scoutfs_segment *seg, u8 level); int scoutfs_seg_manifest_del(struct super_block *sb,