diff --git a/kmod/src/Makefile b/kmod/src/Makefile index 68870aa4..3ceb5af4 100644 --- a/kmod/src/Makefile +++ b/kmod/src/Makefile @@ -2,5 +2,5 @@ obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include -scoutfs-y += block.o counters.o crc.o dir.o filerw.o inode.o msg.o \ - scoutfs_trace.o super.o +scoutfs-y += block.o btree.o counters.o crc.o dir.o filerw.o inode.o msg.o \ + scoutfs_trace.o super.o treap.o diff --git a/kmod/src/block.c b/kmod/src/block.c index 51815f9d..fc232e38 100644 --- a/kmod/src/block.c +++ b/kmod/src/block.c @@ -64,6 +64,7 @@ static struct scoutfs_block *alloc_block(struct super_block *sb, u64 blkno) void scoutfs_put_block(struct scoutfs_block *bl) { if (!IS_ERR_OR_NULL(bl) && atomic_dec_and_test(&bl->refcount)) { + trace_printk("freeing bl %p\n", bl); __free_pages(bl->page, SCOUTFS_BLOCK_PAGE_ORDER); kfree(bl); scoutfs_inc_counter(bl->sb, block_mem_free); @@ -153,10 +154,14 @@ struct scoutfs_block *scoutfs_read_block(struct super_block *sb, u64 blkno) spin_lock(&sbi->block_lock); bl = radix_tree_lookup(&sbi->block_radix, blkno); - if (bl && test_bit(SCOUTFS_BLOCK_BIT_ERROR, &bl->bits)) { - radix_tree_delete(&sbi->block_radix, bl->blkno); - scoutfs_put_block(bl); - bl = NULL; + if (bl) { + if (test_bit(SCOUTFS_BLOCK_BIT_ERROR, &bl->bits)) { + radix_tree_delete(&sbi->block_radix, bl->blkno); + scoutfs_put_block(bl); + bl = NULL; + } else { + atomic_inc(&bl->refcount); + } } spin_unlock(&sbi->block_lock); @@ -180,6 +185,7 @@ struct scoutfs_block *scoutfs_read_block(struct super_block *sb, u64 blkno) if (found) { scoutfs_put_block(bl); bl = found; + atomic_inc(&bl->refcount); } else { radix_tree_insert(&sbi->block_radix, blkno, bl); atomic_inc(&bl->refcount); @@ -212,9 +218,116 @@ out: return bl; } +/* + * Return the block pointed to by the caller's reference. + * + * If the reference sequence numbers don't match then we could be racing + * with another writer. We back off and try again. If it happens too + * many times the caller assumes that we've hit persistent corruption + * and returns an error. + * + * XXX how does this race with + * - reads that span transactions? + * - writers creating a new dirty block? + */ +struct scoutfs_block *scoutfs_read_ref(struct super_block *sb, + struct scoutfs_block_ref *ref) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_block_header *hdr; + struct scoutfs_block *bl; + struct scoutfs_block *found; + + bl = scoutfs_read_block(sb, le64_to_cpu(ref->blkno)); + if (!IS_ERR(bl)) { + hdr = bl->data; + + if (WARN_ON_ONCE(hdr->seq != ref->seq)) { + /* XXX hack, make this a function */ + spin_lock(&sbi->block_lock); + found = radix_tree_lookup(&sbi->block_radix, + bl->blkno); + if (found == bl) { + radix_tree_delete(&sbi->block_radix, bl->blkno); + scoutfs_put_block(bl); + } + spin_unlock(&sbi->block_lock); + + scoutfs_put_block(bl); + bl = ERR_PTR(-EAGAIN); + } + } + + return bl; +} + +/* + * Give the caller a dirty block that they can safely modify. If the + * reference refers to a stable clean block then we allocate a new block + * and update the reference. + * + * Blocks are dirtied and modified within a transaction that has a given + * sequence number which we use to determine if the block is currently + * dirty or not. + * + * For now we're using the dirty super block in the sb_info to track + * the dirty seq. That'll be different when we have multiple btrees. + * + * Callers are working in structures that have sufficient locking to + * protect references to the source block. If we've come to dirty it then + * there won't be concurrent users and we can just move it in the cache. + */ +struct scoutfs_block *scoutfs_dirty_ref(struct super_block *sb, + struct scoutfs_block_ref *ref) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_block_header *hdr; + struct scoutfs_block *found; + struct scoutfs_block *bl; + u64 blkno; + int ret; + + bl = scoutfs_read_block(sb, le64_to_cpu(ref->blkno)); + if (IS_ERR(bl) || ref->seq == sbi->super.hdr.seq) + return bl; + + ret = radix_tree_preload(GFP_NOFS); + if (ret) { + scoutfs_put_block(bl); + return ERR_PTR(ret); + } + + /* XXX cheesy */ + blkno = atomic64_inc_return(&sbi->next_blkno); + hdr = bl->data; + + spin_lock(&sbi->block_lock); + + /* XXX don't really like this */ + found = radix_tree_lookup(&sbi->block_radix, bl->blkno); + if (found == bl) { + radix_tree_delete(&sbi->block_radix, bl->blkno); + atomic_dec(&bl->refcount); + } + + bl->blkno = blkno; + hdr->blkno = cpu_to_le64(blkno); + hdr->seq = sbi->super.hdr.seq; + radix_tree_insert(&sbi->block_radix, blkno, bl); + atomic_inc(&bl->refcount); + + spin_unlock(&sbi->block_lock); + radix_tree_preload_end(); + + ref->blkno = hdr->blkno; + ref->seq = hdr->seq; + + return bl; +} + /* * Return a newly allocated metadata block with an updated block header - * to match the current dirty super block. Callers are responsible for + * to match the current dirty seq. Callers are responsible for * serializing access to the block and for zeroing unwritten block * contents. */ @@ -242,6 +355,7 @@ struct scoutfs_block *scoutfs_new_block(struct super_block *sb, u64 blkno) hdr = bl->data; *hdr = sbi->super.hdr; hdr->blkno = cpu_to_le64(blkno); + hdr->seq = sbi->super.hdr.seq; spin_lock(&sbi->block_lock); found = radix_tree_lookup(&sbi->block_radix, blkno); @@ -265,6 +379,21 @@ out: return bl; } +/* + * Allocate a new dirty writable block. The caller must be in a + * transaction so that we can assign the dirty seq. + */ +struct scoutfs_block *scoutfs_alloc_block(struct super_block *sb) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + u64 blkno; + + /* XXX cheesy */ + blkno = atomic64_inc_return(&sbi->next_blkno); + + return scoutfs_new_block(sb, blkno); +} + void scoutfs_calc_hdr_crc(struct scoutfs_block *bl) { struct scoutfs_block_header *hdr = bl->data; diff --git a/kmod/src/block.h b/kmod/src/block.h index c04586a8..dac2b4e8 100644 --- a/kmod/src/block.h +++ b/kmod/src/block.h @@ -23,6 +23,13 @@ struct scoutfs_block { struct scoutfs_block *scoutfs_read_block(struct super_block *sb, u64 blkno); struct scoutfs_block *scoutfs_new_block(struct super_block *sb, u64 blkno); +struct scoutfs_block *scoutfs_alloc_block(struct super_block *sb); + +struct scoutfs_block *scoutfs_read_ref(struct super_block *sb, + struct scoutfs_block_ref *ref); +struct scoutfs_block *scoutfs_dirty_ref(struct super_block *sb, + struct scoutfs_block_ref *ref); + void scoutfs_put_block(struct scoutfs_block *bl); void scoutfs_calc_hdr_crc(struct scoutfs_block *bl); diff --git a/kmod/src/btree.c b/kmod/src/btree.c new file mode 100644 index 00000000..cf3e700e --- /dev/null +++ b/kmod/src/btree.c @@ -0,0 +1,950 @@ +/* + * Copyright (C) 2016 Zach Brown. 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 "super.h" +#include "format.h" +#include "block.h" +#include "key.h" +#include "treap.h" +#include "btree.h" + +/* + * scoutfs stores file system metadata in btrees whose items have fixed + * sized keys and variable length values. + * + * Items are stored as a small header with the key followed by the + * value. New items are appended to the end of the block. Free space + * is not indexed. Deleted items can be reclaimed by walking all the + * items from the front of the block and moving later live items onto + * earlier deleted items. + * + * The items are kept in a treap sorted by their keys. Using a dynamic + * structure keeps the modification costs low. Modifying persistent + * structures avoids translation to and from run-time structures around + * read and write. The treap was chosen because it's very simple to + * implement and has some cool merging and splitting functions that we + * could make use of. The treap has parent pointers so that we can + * perform operations relative to a node without having to keep a record + * of the path down the tree. + * + * Parent blocks in the btree have the same format as leaf blocks. + * There's one key for every child reference instead of having separator + * keys between child references. The key in a child reference contains + * the largest key that may be found in the child subtree. The right + * spine of the tree has maximal keys so that they don't have to be + * updated if we insert an item with a key greater than everything in + * the tree. + * + * Operations are performed in one pass down the tree. This lets us + * cascade locks from the root down to the leaves and avoids having to + * maintain a record of the path down the tree. Splits and merges are + * performed as we descend. + * + * XXX + * - actually free blknos + * - do we want a level in the btree header? seems like we would? + * - validate structures on read? + */ + +/* size of the item with a value of the given length */ +static inline unsigned int val_bytes(unsigned int val_len) +{ + return sizeof(struct scoutfs_btree_item) + val_len; +} + +static inline unsigned int item_bytes(struct scoutfs_btree_item *item) +{ + return val_bytes(le16_to_cpu(item->val_len)); +} + +static inline unsigned int used_total(struct scoutfs_btree_block *bt) +{ + return SCOUTFS_BLOCK_SIZE - sizeof(struct scoutfs_btree_block) - + le16_to_cpu(bt->total_free); +} + +static int cmp_tnode_items(struct scoutfs_treap_node *A, + struct scoutfs_treap_node *B) +{ + struct scoutfs_btree_item *a; + struct scoutfs_btree_item *b; + + a = container_of(A, struct scoutfs_btree_item, tnode); + b = container_of(B, struct scoutfs_btree_item, tnode); + + return scoutfs_key_cmp(&a->key, &b->key); +} + +/* A bunch of wrappers for navigating items through treap nodes. */ + +#define BT_TREAP_KEY_WRAPPER(which) \ +static struct scoutfs_btree_item *bt_##which(struct scoutfs_btree_block *bt, \ + struct scoutfs_key *key) \ +{ \ + struct scoutfs_btree_item dummy = { .key = *key }; \ + struct scoutfs_treap_node *node; \ + \ + node = scoutfs_treap_##which(&bt->treap, cmp_tnode_items, \ + &dummy.tnode); \ + if (!node) \ + return NULL; \ + \ + return container_of(node, struct scoutfs_btree_item, tnode); \ +} + +BT_TREAP_KEY_WRAPPER(lookup) +/* BT_TREAP_KEY_WRAPPER(before) */ +BT_TREAP_KEY_WRAPPER(after) + +#define BT_TREAP_ROOT_WRAPPER(which) \ +static struct scoutfs_btree_item *bt_##which(struct scoutfs_btree_block *bt) \ +{ \ + struct scoutfs_treap_node *node; \ + \ + node = scoutfs_treap_##which(&bt->treap); \ + if (!node) \ + return NULL; \ + \ + return container_of(node, struct scoutfs_btree_item, tnode); \ +} + +BT_TREAP_ROOT_WRAPPER(first) +BT_TREAP_ROOT_WRAPPER(last) + +#define BT_TREAP_NODE_WRAPPER(which) \ +static struct scoutfs_btree_item *bt_##which(struct scoutfs_btree_block *bt, \ + struct scoutfs_btree_item *item)\ +{ \ + struct scoutfs_treap_node *node; \ + \ + node = scoutfs_treap_##which(&bt->treap, &item->tnode); \ + if (!node) \ + return NULL; \ + \ + return container_of(node, struct scoutfs_btree_item, tnode); \ +} + +BT_TREAP_NODE_WRAPPER(next) +BT_TREAP_NODE_WRAPPER(prev) + +static inline struct scoutfs_key *least_key(struct scoutfs_btree_block *bt) +{ + return &bt_first(bt)->key; +} + +static inline struct scoutfs_key *greatest_key(struct scoutfs_btree_block *bt) +{ + return &bt_last(bt)->key; +} + +/* + * Allocate and insert a new item into the block. + * + * The caller has made sure that there's room for everything. + * + * The caller is responsible for initializing the value. + */ +static struct scoutfs_btree_item *create_item(struct scoutfs_btree_block *bt, + struct scoutfs_key *key, + unsigned int val_len) +{ + unsigned int bytes = val_bytes(val_len); + struct scoutfs_btree_item *item; + + item = (void *)((char *)bt + SCOUTFS_BLOCK_SIZE - + le16_to_cpu(bt->tail_free)); + le16_add_cpu(&bt->tail_free, -bytes); + le16_add_cpu(&bt->total_free, -bytes); + le16_add_cpu(&bt->nr_items, 1); + + item->key = *key; + item->val_len = cpu_to_le16(val_len); + + scoutfs_treap_insert(&bt->treap, cmp_tnode_items, &item->tnode); + + return item; +} + +#define MAGIC_DELETED_PARENT cpu_to_le16(1) + +/* + * Delete an item from a btree block. We set the deleted item's parent + * treap offset to a magic value for compaction. + */ +static void delete_item(struct scoutfs_btree_block *bt, + struct scoutfs_btree_item *item) +{ + scoutfs_treap_delete(&bt->treap, &item->tnode); + item->tnode.parent = MAGIC_DELETED_PARENT; + + le16_add_cpu(&bt->total_free, item_bytes(item)); + le16_add_cpu(&bt->nr_items, -1); +} + +/* + * Move items from a source block to a destination block. The caller + * tells us if we're moving from the tail of the source block right to + * the head of the destination block, or vice versa. We stop moving + * once we've moved enough bytes of items. + * + * XXX This could use fancy treap splitting and merging. We don't need + * to go there yet. + */ +static void move_items(struct scoutfs_btree_block *dst, + struct scoutfs_btree_block *src, bool move_right, + int to_move) +{ + struct scoutfs_btree_item *from; + struct scoutfs_btree_item *to; + unsigned int val_len; + + if (move_right) + from = bt_last(src); + else + from = bt_first(src); + + while (from && to_move > 0) { + val_len = le16_to_cpu(from->val_len); + + to = create_item(dst, &from->key, val_len); + memcpy(to->val, from->val, val_len); + + delete_item(src, from); + + if (move_right) + from = bt_prev(src, from); + else + from = bt_next(src, from); + to_move -= item_bytes(to); + } +} + +/* + * As items are deleted they create fragmented free space. Even if we + * indexed free space in the block it could still get sufficiently + * fragmented to force a split on insertion even though the two + * resulting blocks would have less than the minimum space consumed by + * items. + * + * We don't bother implementing free space indexing and addressing that + * corner case. Instead we track the number of total free bytes in the + * block. If free space needed is available in the block but is not + * available at the end of the block then we reclaim the fragmented free + * space by compacting the items. + * + * We move the free space to the tail of the block by walk forward + * through the items in allocated order moving live items back in to + * free space. + * + * Compaction is only attempted during descent as we find a block that + * needs more or less free space. The caller has the parent locked for + * writing and there are no references to the items at this point so + * it's safe to scramble the block contents. + */ +static void compact_items(struct scoutfs_btree_block *bt) +{ + struct scoutfs_btree_item *from = (void *)(bt + 1); + struct scoutfs_btree_item *to = from; + unsigned int bytes; + unsigned int i; + + for (i = 0; i < le16_to_cpu(bt->nr_items); i++) { + bytes = item_bytes(from); + + if (from->tnode.parent != MAGIC_DELETED_PARENT) { + if (from != to) { + memmove(to, from, bytes); + scoutfs_treap_move(&bt->treap, + &from->tnode, + &to->tnode); + } + to = (void *)to + bytes; + } else { + i--; + } + + from = (void *)from + bytes; + } + + bytes = SCOUTFS_BLOCK_SIZE - ((char *)to - (char *)bt); + bt->tail_free = cpu_to_le16(bytes); +} + +/* + * Allocate and initialize a new tree block. The caller adds references + * to it. + */ +static struct scoutfs_block *alloc_tree_block(struct super_block *sb) +{ + struct scoutfs_btree_block *bt; + struct scoutfs_block *bl; + + bl = scoutfs_alloc_block(sb); + if (!IS_ERR(bl)) { + bt = bl->data; + + bt->total_free = cpu_to_le16(SCOUTFS_BLOCK_SIZE - + sizeof(struct scoutfs_btree_block)); + bt->tail_free = bt->total_free; + bt->nr_items = 0; + } + + return bl; +} + +/* + * Allocate a new tree block and point the root at it. The caller + * is responsible for the items in the new root block. + */ +static struct scoutfs_block *grow_tree(struct super_block *sb, + struct scoutfs_btree_root *root) +{ + struct scoutfs_block_header *hdr; + struct scoutfs_block *bl; + + bl = alloc_tree_block(sb); + if (!IS_ERR(bl)) { + hdr = bl->data; + + root->height++; + root->ref.blkno = hdr->blkno; + root->ref.seq = hdr->seq; + } + + return bl; +} + +/* + * Create a new item in the parent which references the child. The caller + * specifies the key in the item that describes the items in the child. + */ +static void create_parent_item(struct scoutfs_btree_block *parent, + struct scoutfs_btree_block *child, + struct scoutfs_key *key) +{ + struct scoutfs_btree_item *item; + struct scoutfs_block_ref ref = { + .blkno = child->hdr.blkno, + .seq = child->hdr.seq, + }; + + item = create_item(parent, key, sizeof(ref)); + memcpy(&item->val, &ref, sizeof(ref)); +} + +/* + * See if we need to split this block while descending for insertion so + * that we have enough space to insert. + * + * Parent blocks need enough space for a new item and child ref if a + * child block splits. Leaf blocks need enough space to insert the new + * item with its value. + * + * We split to the left so that the greatest key in the existing block + * doesn't change so we don't have to update the key in its parent item. + * + * If the search key falls in the new split block then we return it + * to the caller to walk through. + * + * The locking in the case where we add the first parent is a little wonky. + * We're creating a parent block that the walk doesn't know about. It + * holds the tree mutex while we add the parent ref and then will lock + * the child that we return. It's skipping locking the new parent as it + * descends but that's fine. + */ +static struct scoutfs_block *try_split(struct super_block *sb, + struct scoutfs_btree_root *root, + int level, struct scoutfs_key *key, + unsigned int val_len, + struct scoutfs_btree_block *parent, + struct scoutfs_btree_item *par_item, + struct scoutfs_block *right_bl) +{ + struct scoutfs_btree_block *right = right_bl->data; + struct scoutfs_btree_block *left; + struct scoutfs_block *left_bl; + struct scoutfs_block *par_bl = NULL; + unsigned int bytes; + + if (level) + val_len = sizeof(struct scoutfs_block_ref); + bytes = val_bytes(val_len); + + if (le16_to_cpu(right->tail_free) >= bytes) + return right_bl; + + if (le16_to_cpu(right->total_free) >= bytes) { + compact_items(right); + return right_bl; + } + + if (!parent) { + par_bl = grow_tree(sb, root); + if (IS_ERR(par_bl)) { + scoutfs_put_block(right_bl); + return par_bl; + } + + parent = par_bl->data; + } + + left_bl = alloc_tree_block(sb); + if (IS_ERR(left_bl)) { + /* XXX free parent block? */ + scoutfs_put_block(par_bl); + scoutfs_put_block(right_bl); + return left_bl; + } + left = left_bl->data; + + /* only grow the tree once we have the split neighbour */ + if (par_bl) { + struct scoutfs_key ones; + memset(&ones, 0xff, sizeof(ones)); + create_parent_item(parent, right, &ones); + } + + move_items(left, right, false, used_total(right) / 2); + create_parent_item(parent, left, greatest_key(left)); + + if (scoutfs_key_cmp(key, greatest_key(left)) <= 0) { + scoutfs_put_block(right_bl); + right_bl = left_bl; + } else { + scoutfs_put_block(left_bl); + } + + scoutfs_put_block(par_bl); + + return right_bl; +} + +/* + * This is called during descent for deletion when we have a parent and + * might need to merge items from a sibling block if this block has too + * much free space. Eventually we'll be able to fit all of the + * sibling's items in our free space which lets us delete the sibling + * block. + * + * The error handling here is a little weird. We're returning an + * ERR_PTR buffer to match splitting so that the walk can handle errors + * from both easily. We have to unlock and release our buffer to return + * an error. + * + * The caller only has the parent locked. They'll lock whichever + * block we return. + * + * XXX this could more cleverly chose a merge candidate sibling + */ +static struct scoutfs_block *try_merge(struct super_block *sb, + struct scoutfs_btree_root *root, + struct scoutfs_btree_block *parent, + struct scoutfs_btree_item *par_item, + struct scoutfs_block *bl) +{ + struct scoutfs_btree_block *bt = bl->data; + struct scoutfs_btree_block *sib_bt; + struct scoutfs_block *sib_bl; + struct scoutfs_btree_item *sib_item; + int to_move; + bool move_right; + + if (le16_to_cpu(bt->total_free) <= SCOUTFS_BTREE_FREE_LIMIT) + return bl; + + /* move items right into our block if we have a left sibling */ + sib_item = bt_prev(parent, par_item); + if (sib_item) { + move_right = false; + } else { + sib_item = bt_next(parent, par_item); + move_right = true; + } + + sib_bl = scoutfs_dirty_ref(sb, (void *)sib_item->val); + if (IS_ERR(sib_bl)) { + /* XXX do we need to unlock this? don't think so */ + scoutfs_put_block(bl); + return sib_bl; + } + sib_bt = sib_bl->data; + + if (used_total(sib_bt) <= le16_to_cpu(bt->total_free)) + to_move = used_total(sib_bt); + else + to_move = le16_to_cpu(bt->total_free) - + SCOUTFS_BTREE_FREE_LIMIT; + + if (le16_to_cpu(bt->tail_free) < to_move) + compact_items(bt); + + move_items(bt, sib_bt, move_right, to_move); + + /* update our parent's ref if we changed our greatest key */ + if (!move_right) + par_item->key = *greatest_key(bt); + + /* delete an empty sib or update if we changed its greatest key */ + if (sib_bt->nr_items == 0) { + delete_item(parent, sib_item); + /* XXX free sib block */ + } else if (move_right) { + sib_item->key = *greatest_key(sib_bt); + } + + /* and finally shrink the tree if our parent is the root with 1 */ + if (le16_to_cpu(parent->nr_items) == 1) { + root->height--; + root->ref.blkno = bt->hdr.blkno; + root->ref.seq = bt->hdr.seq; + /* XXX free block */ + } + + return bl; +} + +enum { + WALK_INSERT = 1, + WALK_DELETE, + WALK_NEXT, + WALK_PREV, + WALK_DIRTY, +}; + +/* + * Usually we descend to a leaf that contains the key. But if we're + * searching for a next or previous item then we might hit a leaf block + * that contains the key but no items in the direction of the search. + * We might need to ascend and continue the search in the next block. + * + * The caller has just descended to a leaf and is asking us to discover + * this case. We set the parent item and return true to tell the caller + * to read the new parent item's referenced block. + * + * XXX I don't think this can see bts with 0 items? would need to verify? + */ +static bool next_leaf(struct scoutfs_btree_block *parent, + struct scoutfs_btree_item **par_item, + struct scoutfs_btree_block *bt, int op, int level, + struct scoutfs_key *key) +{ + struct scoutfs_btree_item *nei; + + if (level > 0 || !parent || le16_to_cpu(parent->nr_items) < 2) + return false; + + if (op == WALK_NEXT) { + nei = bt_next(parent, *par_item); + if (nei && (scoutfs_key_cmp(key, greatest_key(bt)) > 0)) { + *par_item = nei; + return true; + } + } + + if (op == WALK_PREV) { + nei = bt_prev(parent, *par_item); + if (nei && (scoutfs_key_cmp(key, least_key(bt)) < 0)) { + *par_item = nei; + return true; + } + } + + return false; +} + +/* + * As we descend we lock parent blocks (or the root), then lock the child, + * then unlock the parent. + */ +static void lock_block(struct scoutfs_sb_info *sbi, struct scoutfs_block *bl, + bool dirty) +{ + struct rw_semaphore *rwsem; + + if (bl == NULL) + rwsem = &sbi->btree_rwsem; + else + rwsem = &bl->rwsem; + + if (dirty) + down_write(rwsem); + else + down_read(rwsem); +} + +static void unlock_block(struct scoutfs_sb_info *sbi, struct scoutfs_block *bl, + bool dirty) +{ + struct rw_semaphore *rwsem; + + if (bl == NULL) + rwsem = &sbi->btree_rwsem; + else + rwsem = &bl->rwsem; + + if (dirty) + up_write(rwsem); + else + up_read(rwsem); +} + + +/* + * Return the leaf block that should contain the given key. The caller + * is responsible for searching the leaf block and performing their + * operation. The block is returned locked for either reading or writing + * depending on the operation. + */ +static struct scoutfs_block *btree_walk(struct super_block *sb, + struct scoutfs_key *key, + unsigned int val_len, int op) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_btree_block *parent = NULL; + struct scoutfs_btree_block *bt; + struct scoutfs_btree_root *root; + struct scoutfs_block *par_bl = NULL; + struct scoutfs_block *bl = NULL; + struct scoutfs_btree_item *item = NULL; + struct scoutfs_block_ref *ref; + unsigned int level; + const bool dirty = op == WALK_INSERT || op == WALK_DELETE || + op == WALK_DIRTY; + + lock_block(sbi, par_bl, dirty); + + /* XXX one for now */ + root = &sbi->super.btree_root; + ref = &root->ref; + level = root->height; + + if (!root->height) { + if (op == WALK_INSERT) { + bl = ERR_PTR(-ENOENT); + } else { + bl = grow_tree(sb, root); + if (!IS_ERR(bl)) + lock_block(sbi, bl, dirty); + } + unlock_block(sbi, par_bl, dirty); + return bl; + } + + while (level--) { + /* XXX hmm, need to think about retry */ + if (dirty) { + bl = scoutfs_dirty_ref(sb, ref); + } else { + bl = scoutfs_read_ref(sb, ref); + } + if (IS_ERR(bl)) + break; + bt = bl->data; + + /* see if a search needs to move to the next parent ref */ + if (next_leaf(parent, &item, bt, op, level, key)) { + ref = (void *)item->val; + level++; + scoutfs_put_block(bl); + continue; + } + + if (op == WALK_INSERT) + bl = try_split(sb, root, level, key, val_len, parent, + item, bl); + if ((op == WALK_DELETE) && parent) + bl = try_merge(sb, root, parent, item, bl); + if (IS_ERR(bl)) + break; + + lock_block(sbi, bl, dirty); + + if (!level) + break; + + /* unlock parent before searching so others can use it */ + unlock_block(sbi, par_bl, dirty); + scoutfs_put_block(par_bl); + par_bl = bl; + parent = bt; + + /* there should always be a parent item */ + item = bt_after(parent, key); + if (!item) { + /* current block dropped as parent below */ + bl = ERR_PTR(-EIO); + break; + } + + /* XXX verify sane length */ + ref = (void *)item->val; + } + + unlock_block(sbi, par_bl, dirty); + scoutfs_put_block(par_bl); + + return bl; +} + +static void set_cursor(struct scoutfs_btree_cursor *curs, + struct scoutfs_block *bl, + struct scoutfs_btree_item *item, bool write) +{ + curs->bl = bl; + curs->item = item; + curs->key = &item->key; + curs->val = item->val; + curs->val_len = le16_to_cpu(item->val_len); + curs->write = !!write; +} + +/* + * Point the caller's cursor at the item if it's found. It can't be + * modified. -ENOENT is returned if the key isn't found in the tree. + */ +int scoutfs_btree_lookup(struct super_block *sb, struct scoutfs_key *key, + struct scoutfs_btree_cursor *curs) +{ + struct scoutfs_btree_item *item; + struct scoutfs_block *bl; + int ret; + + BUG_ON(curs->bl); + + bl = btree_walk(sb, key, 0, 0); + if (IS_ERR(bl)) + return PTR_ERR(bl); + + item = bt_lookup(bl->data, key); + if (item) { + set_cursor(curs, bl, item, false); + ret = 0; + } else { + up_read(&bl->rwsem); + scoutfs_put_block(bl); + ret = -ENOENT; + } + + return ret; +} + +/* + * Insert a new item in the tree and point the caller's cursor at it. + * The caller is responsible for setting the value. + * + * -EEXIST is returned if the key is already present in the tree. + * + * XXX this walks the treap twice, which isn't great + */ +int scoutfs_btree_insert(struct super_block *sb, struct scoutfs_key *key, + unsigned int val_len, + struct scoutfs_btree_cursor *curs) +{ + struct scoutfs_btree_item *item; + struct scoutfs_btree_block *bt; + struct scoutfs_block *bl; + int ret; + + BUG_ON(curs->bl); + + bl = btree_walk(sb, key, val_len, WALK_INSERT); + if (IS_ERR(bl)) + return PTR_ERR(bl); + bt = bl->data; + + /* XXX should this return -eexist? */ + item = bt_lookup(bt, key); + if (!item) { + item = create_item(bt, key, val_len); + set_cursor(curs, bl, item, true); + ret = 0; + } else { + up_write(&bl->rwsem); + scoutfs_put_block(bl); + ret = -ENOENT; + } + + return ret; +} + +/* + * Delete an item from the tree. -ENOENT is returned if the key isn't + * found. + */ +int scoutfs_btree_delete(struct super_block *sb, struct scoutfs_key *key) +{ + struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); + struct scoutfs_btree_item *item; + struct scoutfs_btree_block *bt; + struct scoutfs_block *bl; + int ret; + + bl = btree_walk(sb, key, 0, WALK_DELETE); + if (IS_ERR(bl)) + return PTR_ERR(bl); + bt = bl->data; + + item = bt_lookup(bt, key); + if (item) { + delete_item(bt, item); + ret = 0; + + /* XXX this locking is broken.. hold root rwsem? */ + + /* delete the final block in the tree */ + if (bt->nr_items == 0) { + memset(&sbi->super.btree_root, 0, + sizeof(struct scoutfs_btree_root)); + /* XXX free block */ + } + } else { + ret = -ENOENT; + } + + up_write(&bl->rwsem); + scoutfs_put_block(bl); + + return ret; +} + +/* + * The caller initializes the cursor and first and last keys and then + * gets the cursor set to each item within those keys. + * + * The btree walk takes care of advancing past interior leaves that + * don't contain items past the key. Our job is to find the next item + * after the key. If that next item's key is past the caller's last key + * then the iteration is done. + * + * returns 0 if no next, > 0 when curs contains next, < 0 on error + */ +int scoutfs_btree_next(struct super_block *sb, struct scoutfs_key *first, + struct scoutfs_key *last, + struct scoutfs_btree_cursor *curs) +{ + struct scoutfs_btree_block *bt; + struct scoutfs_block *bl; + struct scoutfs_key key = *first; + int ret; + + trace_printk("first "CKF" last "CKF" %d curs "CKF"\n", + CKA(first), CKA(last), + !!curs->bl, CKA(curs->bl ? curs->key : &key)); + + /* find the next item after the cursor, releasing if we're done */ + if (curs->bl) { + key = curs->item->key; + scoutfs_inc_key(&key); + + curs->item = bt_next(curs->bl->data, curs->item); + trace_printk("next %p\n", curs->item); + if (curs->item) + set_cursor(curs, curs->bl, curs->item, curs->write); + else + scoutfs_btree_release(curs); + } + + /* walk the tree to find the key, can be first or later */ + if (!curs->bl) { + bl = btree_walk(sb, &key, 0, WALK_NEXT); + if (IS_ERR(bl)) + return PTR_ERR(bl); + bt = bl->data; + + curs->item = bt_after(bl->data, &key); + trace_printk("after %p\n", curs->item); + if (curs->item) { + set_cursor(curs, bl, curs->item, false); + } else { + up_read(&bl->rwsem); + scoutfs_put_block(bl); + } + } + + /* only return the next item if it's within last */ + if (curs->item && scoutfs_key_cmp(curs->key, last) <= 0) { + ret = 1; + } else { + scoutfs_btree_release(curs); + ret = 0; + } + + trace_printk("ret %d\n", ret); + return ret; +} + +/* + * Ensure that the blocks that lead to the item with the given key are + * dirty. caller can hold a transaction to pin the dirty blocks and + * guarantee that later updates of the item will succeed. + * + * <0 is returned on error, including -ENOENT if the key isn't present. + */ +int scoutfs_btree_dirty(struct super_block *sb, struct scoutfs_key *key) +{ + struct scoutfs_btree_item *item; + struct scoutfs_block *bl; + int ret; + + bl = btree_walk(sb, key, 0, WALK_DIRTY); + if (IS_ERR(bl)) + return PTR_ERR(bl); + + item = bt_lookup(bl->data, key); + if (item) { + ret = 0; + } else { + ret = -ENOENT; + } + + up_write(&bl->rwsem); + scoutfs_put_block(bl); + + return ret; +} + +/* + * For this to be safe the caller has to have pinned the dirty blocks + * for the item in their transaction. + */ +void scoutfs_btree_update(struct super_block *sb, struct scoutfs_key *key, + struct scoutfs_btree_cursor *curs) +{ + struct scoutfs_btree_item *item; + struct scoutfs_block *bl; + + BUG_ON(curs->bl); + + bl = btree_walk(sb, key, 0, WALK_DIRTY); + BUG_ON(IS_ERR(bl)); + + item = bt_lookup(bl->data, key); + BUG_ON(!item); + + set_cursor(curs, bl, item, true); +} + +void scoutfs_btree_release(struct scoutfs_btree_cursor *curs) +{ + if (curs->bl) { + if (curs->write) + up_write(&curs->bl->rwsem); + else + up_read(&curs->bl->rwsem); + scoutfs_put_block(curs->bl); + } + curs->bl = NULL; +} diff --git a/kmod/src/btree.h b/kmod/src/btree.h index d7432313..e902a6bd 100644 --- a/kmod/src/btree.h +++ b/kmod/src/btree.h @@ -8,51 +8,27 @@ struct scoutfs_btree_cursor { /* for callers */ struct scoutfs_key *key; - unsigned val_len; void *val; + u16 val_len; + u16 write:1; }; -static inline int scoutfs_btree_lookup(struct super_block *sb, - struct scoutfs_key *key, - struct scoutfs_btree_cursor *curs) -{ - return -ENOSYS; -} +#define DECLARE_SCOUTFS_BTREE_CURSOR(name) \ + struct scoutfs_btree_cursor name = {NULL,} -static inline int scoutfs_btree_insert(struct super_block *sb, - struct scoutfs_key *key, - unsigned short val_len, - struct scoutfs_btree_cursor *curs) -{ - return -ENOSYS; -} +int scoutfs_btree_lookup(struct super_block *sb, struct scoutfs_key *key, + struct scoutfs_btree_cursor *curs); +int scoutfs_btree_insert(struct super_block *sb, struct scoutfs_key *key, + unsigned int val_len, + struct scoutfs_btree_cursor *curs); +int scoutfs_btree_delete(struct super_block *sb, struct scoutfs_key *key); +int scoutfs_btree_next(struct super_block *sb, struct scoutfs_key *first, + struct scoutfs_key *last, + struct scoutfs_btree_cursor *curs); +int scoutfs_btree_dirty(struct super_block *sb, struct scoutfs_key *key); +void scoutfs_btree_update(struct super_block *sb, struct scoutfs_key *key, + struct scoutfs_btree_cursor *curs); -static inline int scoutfs_btree_dirty(struct super_block *sb, - struct scoutfs_key *key, - unsigned short val_len, - struct scoutfs_btree_cursor *curs) -{ - return -ENOSYS; -} - - -static inline int scoutfs_btree_delete(struct super_block *sb, - struct scoutfs_btree_cursor *curs) -{ - return -ENOSYS; -} - -static inline int scoutfs_btree_next(struct super_block *sb, - struct scoutfs_key *first, - struct scoutfs_key *last, - struct scoutfs_btree_cursor *curs) -{ - return -ENOSYS; -} - -static inline int scoutfs_btree_release(struct scoutfs_btree_cursor *curs) -{ - return -ENOSYS; -} +void scoutfs_btree_release(struct scoutfs_btree_cursor *curs); #endif diff --git a/kmod/src/dir.c b/kmod/src/dir.c index 6a95bfb3..140ae023 100644 --- a/kmod/src/dir.c +++ b/kmod/src/dir.c @@ -176,7 +176,7 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, unsigned int flags) { struct scoutfs_inode_info *si = SCOUTFS_I(dir); - struct scoutfs_btree_cursor curs = {NULL,}; + DECLARE_SCOUTFS_BTREE_CURSOR(curs); struct super_block *sb = dir->i_sb; struct scoutfs_dirent *dent; struct dentry_info *di; @@ -209,6 +209,7 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, h = name_hash(dentry->d_name.name, dentry->d_name.len, h); scoutfs_set_key(&key, scoutfs_ino(dir), SCOUTFS_DIRENT_KEY, h); + scoutfs_btree_release(&curs); ret = scoutfs_btree_lookup(sb, &key, &curs); if (ret == -ENOENT) continue; @@ -274,44 +275,38 @@ static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir) { struct inode *inode = file_inode(file); struct super_block *sb = inode->i_sb; - struct scoutfs_btree_cursor curs = {NULL,}; + DECLARE_SCOUTFS_BTREE_CURSOR(curs); struct scoutfs_dirent *dent; struct scoutfs_key first; struct scoutfs_key last; unsigned int name_len; - int ret = 0; + int ret; u32 pos; if (!dir_emit_dots(file, dirent, filldir)) return 0; + scoutfs_set_key(&first, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, + file->f_pos); scoutfs_set_key(&last, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, SCOUTFS_DIRENT_LAST_POS); - while (file->f_pos <= SCOUTFS_DIRENT_LAST_POS) { - scoutfs_set_key(&first, scoutfs_ino(inode), SCOUTFS_DIRENT_KEY, - file->f_pos); - - ret = scoutfs_btree_next(sb, &first, &last, &curs); - if (ret) - break; - + while ((ret = scoutfs_btree_next(sb, &first, &last, &curs)) > 0) { dent = curs.val; name_len = item_name_len(&curs); pos = scoutfs_key_offset(curs.key); if (filldir(dirent, dent->name, name_len, pos, - le64_to_cpu(dent->ino), dentry_type(dent->type))) + le64_to_cpu(dent->ino), dentry_type(dent->type))) { + ret = 0; break; + } file->f_pos = pos + 1; } scoutfs_btree_release(&curs); - if (ret == -ENOENT) - ret = 0; - return ret; } @@ -320,7 +315,7 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode, { struct super_block *sb = dir->i_sb; struct scoutfs_inode_info *si = SCOUTFS_I(dir); - struct scoutfs_btree_cursor curs = {NULL,}; + DECLARE_SCOUTFS_BTREE_CURSOR(curs); struct inode *inode = NULL; struct scoutfs_dirent *dent; struct dentry_info *di; @@ -413,7 +408,6 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry) struct super_block *sb = dir->i_sb; struct inode *inode = dentry->d_inode; struct timespec ts = current_kernel_time(); - struct scoutfs_btree_cursor curs = {NULL,}; struct dentry_info *di; struct scoutfs_key key; int ret = 0; @@ -432,12 +426,7 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry) scoutfs_set_key(&key, scoutfs_ino(dir), SCOUTFS_DIRENT_KEY, di->hash); - ret = scoutfs_btree_lookup(sb, &key, &curs); - if (ret) - goto out; - - ret = scoutfs_btree_delete(sb, &curs); - scoutfs_btree_release(&curs); + ret = scoutfs_btree_delete(sb, &key); if (ret) goto out; diff --git a/kmod/src/format.h b/kmod/src/format.h index d35f5613..a80e95b0 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -34,6 +34,72 @@ struct scoutfs_block_header { __le64 blkno; } __packed; +/* + * We should be able to make the offset smaller if neither dirents nor + * data items use the full 64 bits. + */ +struct scoutfs_key { + __le64 inode; + u8 type; + __le64 offset; +} __packed; + +/* + * Currently we sort keys by the numeric value of the types, but that + * isn't necessary. We could have an arbitrary sort order. So we don't + * have to stress about cleverly allocating the types. + */ +#define SCOUTFS_INODE_KEY 1 +#define SCOUTFS_DIRENT_KEY 2 +#define SCOUTFS_DATA_KEY 3 + +#define SCOUTFS_MAX_ITEM_LEN 2048 + +/* + * Block references include the sequence number so that we can detect + * readers racing with writers and so that we can tell that we don't + * need to follow a reference when traversing based on seqs. + */ +struct scoutfs_block_ref { + __le64 blkno; + __le64 seq; +} __packed; + +struct scoutfs_treap_root { + __le16 off; +} __packed; + +struct scoutfs_treap_node { + __le16 parent; + __le16 left; + __le16 right; + __le32 prio; +} __packed; + +struct scoutfs_btree_root { + u8 height; + struct scoutfs_block_ref ref; +} __packed; + +struct scoutfs_btree_block { + struct scoutfs_block_header hdr; + struct scoutfs_treap_root treap; + __le16 total_free; + __le16 tail_free; + __le16 nr_items; +} __packed; + +struct scoutfs_btree_item { + struct scoutfs_key key; + struct scoutfs_treap_node tnode; + __le16 val_len; + char val[0]; +} __packed; + +/* Blocks are no more than half free. */ +#define SCOUTFS_BTREE_FREE_LIMIT \ + ((SCOUTFS_BLOCK_SIZE - sizeof(struct scoutfs_btree_block)) / 2) + #define SCOUTFS_UUID_BYTES 16 /* @@ -50,31 +116,11 @@ struct scoutfs_super_block { struct scoutfs_block_header hdr; __le64 id; __u8 uuid[SCOUTFS_UUID_BYTES]; -} __packed; - -/* - * We should be able to make the offset smaller if neither dirents nor - * data items use the full 64 bits. - */ -struct scoutfs_key { - __le64 inode; - u8 type; - __le64 offset; + struct scoutfs_btree_root btree_root; } __packed; #define SCOUTFS_ROOT_INO 1 -/* - * Currently we sort keys by the numeric value of the types, but that - * isn't necessary. We could have an arbitrary sort order. So we don't - * have to stress about cleverly allocating the types. - */ -#define SCOUTFS_INODE_KEY 1 -#define SCOUTFS_DIRENT_KEY 2 -#define SCOUTFS_DATA_KEY 3 - -#define SCOUTFS_MAX_ITEM_LEN 2048 - struct scoutfs_timespec { __le64 sec; __le32 nsec; diff --git a/kmod/src/inode.c b/kmod/src/inode.c index 44b4c3de..8ce039b3 100644 --- a/kmod/src/inode.c +++ b/kmod/src/inode.c @@ -112,7 +112,7 @@ static void load_inode(struct inode *inode, struct scoutfs_inode *cinode) static int scoutfs_read_locked_inode(struct inode *inode) { - struct scoutfs_btree_cursor curs = {NULL,}; + DECLARE_SCOUTFS_BTREE_CURSOR(curs); struct super_block *sb = inode->i_sb; struct scoutfs_key key; int ret; @@ -209,23 +209,20 @@ static void store_inode(struct scoutfs_inode *cinode, struct inode *inode) * * The caller has to prevent sync between dirtying and updating the * inodes. + * + * XXX this will have to do something about variable length inodes */ int scoutfs_dirty_inode_item(struct inode *inode) { struct super_block *sb = inode->i_sb; - struct scoutfs_btree_cursor curs = {NULL,}; struct scoutfs_key key; int ret; scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_INODE_KEY, 0); - ret = scoutfs_btree_dirty(sb, &key, sizeof(struct scoutfs_inode), - &curs); - if (!ret) { - store_inode(curs.val, inode); - scoutfs_btree_release(&curs); + ret = scoutfs_btree_dirty(sb, &key); + if (!ret) trace_scoutfs_dirty_inode(inode); - } return ret; } @@ -240,18 +237,13 @@ int scoutfs_dirty_inode_item(struct inode *inode) */ void scoutfs_update_inode_item(struct inode *inode) { - struct scoutfs_btree_cursor curs = {NULL,}; + DECLARE_SCOUTFS_BTREE_CURSOR(curs); struct super_block *sb = inode->i_sb; struct scoutfs_key key; - int ret; scoutfs_set_key(&key, scoutfs_ino(inode), SCOUTFS_INODE_KEY, 0); - /* XXX maybe just use dirty again? not sure.. */ - ret = scoutfs_btree_dirty(sb, &key, sizeof(struct scoutfs_inode), - &curs); - BUG_ON(ret); - + scoutfs_btree_update(sb, &key, &curs); store_inode(curs.val, inode); scoutfs_btree_release(&curs); trace_scoutfs_update_inode(inode); @@ -265,7 +257,7 @@ struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir, umode_t mode, dev_t rdev) { struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb); - struct scoutfs_btree_cursor curs = {NULL,}; + DECLARE_SCOUTFS_BTREE_CURSOR(curs); struct scoutfs_inode_info *ci; struct scoutfs_key key; struct inode *inode; diff --git a/kmod/src/super.c b/kmod/src/super.c index 0e8a7d60..0921e313 100644 --- a/kmod/src/super.c +++ b/kmod/src/super.c @@ -103,6 +103,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) spin_lock_init(&sbi->block_lock); INIT_RADIX_TREE(&sbi->block_radix, GFP_NOFS); init_waitqueue_head(&sbi->block_wq); + init_rwsem(&sbi->btree_rwsem); /* XXX can have multiple mounts of a device, need mount id */ sbi->kset = kset_create_and_add(sb->s_id, NULL, &scoutfs_kset->kobj); diff --git a/kmod/src/super.h b/kmod/src/super.h index 1d5170f4..c7659844 100644 --- a/kmod/src/super.h +++ b/kmod/src/super.h @@ -18,6 +18,9 @@ struct scoutfs_sb_info { atomic64_t next_ino; atomic64_t next_blkno; + /* XXX there will be a lot more of these :) */ + struct rw_semaphore btree_rwsem; + /* $sysfs/fs/scoutfs/$id/ */ struct kset *kset; diff --git a/kmod/src/treap.c b/kmod/src/treap.c new file mode 100644 index 00000000..229e4daf --- /dev/null +++ b/kmod/src/treap.c @@ -0,0 +1,364 @@ +/* + * 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 "format.h" +#include "treap.h" + +/* + * Implement a simple treap in memory. The caller is responsible for + * allocating and freeing roots and nodes. This only performs the tree + * operations on them. + * + * Node references are stored as byte offsets from the root to the node. + * As long as we have the root the byte offsets or node pointers are + * interchangeable. The code tries to prefer to use pointers to be + * slightly easier to read. + * + * The caller is responsible for locking access to the tree. + */ + +static struct scoutfs_treap_node *off_node(struct scoutfs_treap_root *root, + __le16 off) +{ + if (!off) + return NULL; + + return (void *)root + le16_to_cpu(off); +} + +static __le16 node_off(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node) +{ + if (!node) + return 0; + + return cpu_to_le16((char *)node - (char *)root); +} + +/* + * Walk the tree looking for a node that matches a node in the tree. + * Return the found node or the last node traversed. Set the caller's + * cmp to the comparison between the key and the returned node. The + * caller can ask that we set their pointers to the most recently + * traversed node before or after the returned node. + */ +static struct scoutfs_treap_node *descend(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key, + int *cmp, + struct scoutfs_treap_node **before, + struct scoutfs_treap_node **after) +{ + struct scoutfs_treap_node *node = NULL; + __le16 off = root->off; + + *cmp = -1; + if (before) + *before = NULL; + if (after) + *after = NULL; + + while (off) { + node = off_node(root, off); + *cmp = cmp_func(key, node); + if (*cmp < 0) { + if (after) + *after = node; + off = node->left; + } else if (*cmp > 0) { + if (before) + *before = node; + off = node->right; + } else { + break; + } + } + + return node; +} + +/* + * Link the two nodes together by setting their child and parent pointers + * as needed. Both parent and child can be null. + */ +static void set_links(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *parent, bool left, + struct scoutfs_treap_node *child) +{ + if (!parent) + root->off = node_off(root, child); + else if (left) + parent->left = node_off(root, child); + else + parent->right = node_off(root, child); + + if (child) + child->parent = node_off(root, parent); +} + +/* + * Perform a tree rotation. The node pointer names describe their + * relationships before the rotation. We use the relationship between + * the node and its child to determine the direction of the rotation. + * After the rotation the child will be higher than the node. Only the + * node and child must exist. + * + * Here's a right rotation: + * + * parent parent + * | | + * node child + * / \ / \ + * child a b node + * / \ / \ + * b gr_chi gr_chi a + * + */ +static void rotation(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node, + struct scoutfs_treap_node *child) +{ + struct scoutfs_treap_node *parent = off_node(root, node->parent); + struct scoutfs_treap_node *grand_child; + bool right; + + if (node->left == node_off(root, child)) { + right = true; + grand_child = off_node(root, child->right); + } else { + right = false; + grand_child = off_node(root, child->left); + } + + set_links(root, parent, + parent && (parent->left == node_off(root, node)), child); + set_links(root, node, right, grand_child); + set_links(root, child, !right, node); +} + +/* + * Insertion links a node in at a leaf and then rotates it up the + * tree until its parent has a higher priority. + */ +int scoutfs_treap_insert(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *ins) +{ + struct scoutfs_treap_node *parent; + int cmp; + + ins->prio = cpu_to_le32(get_random_int()); + ins->parent = 0; + ins->left = 0; + ins->right = 0; + + parent = descend(root, cmp_func, ins, &cmp, NULL, NULL); + if (cmp == 0) + return -EEXIST; + + set_links(root, parent, cmp < 0, ins); + + while (ins->parent) { + parent = off_node(root, ins->parent); + if (le32_to_cpu(ins->prio) < le32_to_cpu(parent->prio)) + break; + + rotation(root, parent, ins); + } + + return 0; +} + +/* + * Deletion rotates the node down the tree until it doesn't have two + * children so that it can be unlinked by pointing its parent at its + * child, if it has one. + */ +void scoutfs_treap_delete(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node) +{ + struct scoutfs_treap_node *left; + struct scoutfs_treap_node *right; + struct scoutfs_treap_node *child; + struct scoutfs_treap_node *parent; + + while (node->left && node->right) { + left = off_node(root, node->left); + right = off_node(root, node->right); + + if (le32_to_cpu(left->prio) > le32_to_cpu(right->prio)) + rotation(root, node, left); + else + rotation(root, node, right); + } + + parent = off_node(root, node->parent); + + if (node->left) + child = off_node(root, node->left); + else + child = off_node(root, node->right); + + set_links(root, parent, + parent && parent->left == node_off(root, node), child); +} + +struct scoutfs_treap_node *scoutfs_treap_lookup(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key) +{ + struct scoutfs_treap_node *node; + int cmp; + + node = descend(root, cmp_func, key, &cmp, NULL, NULL); + if (cmp != 0) + return NULL; + + return node; +} + +/* return the first node in the tree */ +struct scoutfs_treap_node *scoutfs_treap_first(struct scoutfs_treap_root *root) +{ + struct scoutfs_treap_node *node = off_node(root, root->off); + + while (node && node->left) + node = off_node(root, node->left); + + return node; +} + +/* return the last node in the tree */ +struct scoutfs_treap_node *scoutfs_treap_last(struct scoutfs_treap_root *root) +{ + struct scoutfs_treap_node *node = off_node(root, root->off); + + while (node && node->right) + node = off_node(root, node->right); + + return node; +} + +/* return the last node whose key is less than or equal to the key */ +struct scoutfs_treap_node *scoutfs_treap_before(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key) +{ + struct scoutfs_treap_node *before; + struct scoutfs_treap_node *node; + int cmp; + + node = descend(root, cmp_func, key, &cmp, &before, NULL); + if (cmp == 0) + return node; + + return before; +} + +/* return the first node whose key is greater than or equal to the key */ +struct scoutfs_treap_node *scoutfs_treap_after(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key) +{ + struct scoutfs_treap_node *after; + struct scoutfs_treap_node *node; + int cmp; + + node = descend(root, cmp_func, key, &cmp, NULL, &after); + if (cmp == 0) + return node; + + return after; +} + +/* + * The usual BST iteration: either the least descendant or the first + * ancestor in the direction of the iteration. + */ +struct scoutfs_treap_node *scoutfs_treap_next(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node) +{ + struct scoutfs_treap_node *parent; + + if (node->right) { + node = off_node(root, node->right); + while (node->left) + node = off_node(root, node->left); + return node; + } + + while ((parent = off_node(root, node->parent)) && + parent->right == node_off(root, node)) { + node = parent; + } + + return parent; +} + +struct scoutfs_treap_node *scoutfs_treap_prev(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node) +{ + struct scoutfs_treap_node *parent; + + if (node->left) { + node = off_node(root, node->left); + while (node->right) + node = off_node(root, node->right); + return node; + } + + while ((parent = off_node(root, node->parent)) && + parent->left == node_off(root, node)) { + node = parent; + } + + return parent; +} + +static void update_relative(struct scoutfs_treap_root *root, __le16 node_off, + __le16 from_off, __le16 to_off) +{ + struct scoutfs_treap_node *node = off_node(root, node_off); + + if (node) { + if (node->parent == from_off) + node->parent = to_off; + else if (node->left == from_off) + node->left = to_off; + else if (node->right == from_off) + node->right = to_off; + } +} + +/* + * A node has moved from one storage location to another. Update the + * nodes that refer to it. The from pointer can only be used to + * determine the old offset. Its contents are undefined. + */ +void scoutfs_treap_move(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *from, + struct scoutfs_treap_node *to) +{ + __le16 from_off = node_off(root, from); + __le16 to_off = node_off(root, to); + + if (root->off == from_off) + root->off = to_off; + else + update_relative(root, to->parent, from_off, to_off); + + update_relative(root, to->left, from_off, to_off); + update_relative(root, to->right, from_off, to_off); +} diff --git a/kmod/src/treap.h b/kmod/src/treap.h new file mode 100644 index 00000000..d02f406c --- /dev/null +++ b/kmod/src/treap.h @@ -0,0 +1,38 @@ +#ifndef _SCOUTFS_TREAP_H_ +#define _SCOUTFS_TREAP_H_ + +#include "format.h" + +typedef int (*scoutfs_treap_cmp_t)(struct scoutfs_treap_node *a, + struct scoutfs_treap_node *b); + +static inline void scoutfs_treap_init(struct scoutfs_treap_root *root) +{ + root->off = 0; +} + +int scoutfs_treap_insert(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *ins); +void scoutfs_treap_delete(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node); +struct scoutfs_treap_node *scoutfs_treap_lookup(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key); +struct scoutfs_treap_node *scoutfs_treap_first(struct scoutfs_treap_root *root); +struct scoutfs_treap_node *scoutfs_treap_last(struct scoutfs_treap_root *root); +struct scoutfs_treap_node *scoutfs_treap_before(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key); +struct scoutfs_treap_node *scoutfs_treap_after(struct scoutfs_treap_root *root, + scoutfs_treap_cmp_t cmp_func, + struct scoutfs_treap_node *key); +struct scoutfs_treap_node *scoutfs_treap_next(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node); +struct scoutfs_treap_node *scoutfs_treap_prev(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *node); +void scoutfs_treap_move(struct scoutfs_treap_root *root, + struct scoutfs_treap_node *from, + struct scoutfs_treap_node *to); + +#endif