diff options
Diffstat (limited to 'libbcache/btree_update.c')
| -rw-r--r-- | libbcache/btree_update.c | 2343 |
1 files changed, 2343 insertions, 0 deletions
diff --git a/libbcache/btree_update.c b/libbcache/btree_update.c new file mode 100644 index 00000000..95406a44 --- /dev/null +++ b/libbcache/btree_update.c @@ -0,0 +1,2343 @@ + +#include "bcache.h" +#include "alloc.h" +#include "bkey_methods.h" +#include "btree_cache.h" +#include "btree_gc.h" +#include "btree_update.h" +#include "btree_io.h" +#include "btree_iter.h" +#include "btree_locking.h" +#include "buckets.h" +#include "extents.h" +#include "journal.h" +#include "keylist.h" +#include "super.h" + +#include <linux/random.h> +#include <linux/sort.h> +#include <trace/events/bcache.h> + +static void btree_interior_update_updated_root(struct cache_set *, + struct btree_interior_update *, + enum btree_id); + +/* Calculate ideal packed bkey format for new btree nodes: */ + +void __bch_btree_calc_format(struct bkey_format_state *s, struct btree *b) +{ + struct bkey_packed *k; + struct bset_tree *t; + struct bkey uk; + + bch_bkey_format_add_pos(s, b->data->min_key); + + for_each_bset(b, t) + for (k = btree_bkey_first(b, t); + k != btree_bkey_last(b, t); + k = bkey_next(k)) + if (!bkey_whiteout(k)) { + uk = bkey_unpack_key(b, k); + bch_bkey_format_add_key(s, &uk); + } +} + +static struct bkey_format bch_btree_calc_format(struct btree *b) +{ + struct bkey_format_state s; + + bch_bkey_format_init(&s); + __bch_btree_calc_format(&s, b); + + return bch_bkey_format_done(&s); +} + +static size_t btree_node_u64s_with_format(struct btree *b, + struct bkey_format *new_f) +{ + struct bkey_format *old_f = &b->format; + + /* stupid integer promotion rules */ + ssize_t delta = + (((int) new_f->key_u64s - old_f->key_u64s) * + (int) b->nr.packed_keys) + + (((int) new_f->key_u64s - BKEY_U64s) * + (int) b->nr.unpacked_keys); + + BUG_ON(delta + b->nr.live_u64s < 0); + + return b->nr.live_u64s + delta; +} + +/** + * btree_node_format_fits - check if we could rewrite node with a new format + * + * This assumes all keys can pack with the new format -- it just checks if + * the re-packed keys would fit inside the node itself. + */ +bool bch_btree_node_format_fits(struct cache_set *c, struct btree *b, + struct bkey_format *new_f) +{ + size_t u64s = btree_node_u64s_with_format(b, new_f); + + return __set_bytes(b->data, u64s) < btree_bytes(c); +} + +/* Btree node freeing/allocation: */ + +/* + * We're doing the index update that makes @b unreachable, update stuff to + * reflect that: + * + * Must be called _before_ btree_interior_update_updated_root() or + * btree_interior_update_updated_btree: + */ +static void bch_btree_node_free_index(struct cache_set *c, struct btree *b, + enum btree_id id, struct bkey_s_c k, + struct bucket_stats_cache_set *stats) +{ + struct btree_interior_update *as; + struct pending_btree_node_free *d; + + mutex_lock(&c->btree_interior_update_lock); + + for_each_pending_btree_node_free(c, as, d) + if (!bkey_cmp(k.k->p, d->key.k.p) && + bkey_val_bytes(k.k) == bkey_val_bytes(&d->key.k) && + !memcmp(k.v, &d->key.v, bkey_val_bytes(k.k))) + goto found; + + BUG(); +found: + d->index_update_done = true; + + /* + * Btree nodes are accounted as freed in cache_set_stats when they're + * freed from the index: + */ + stats->s[S_COMPRESSED][S_META] -= c->sb.btree_node_size; + stats->s[S_UNCOMPRESSED][S_META] -= c->sb.btree_node_size; + + /* + * We're dropping @k from the btree, but it's still live until the + * index update is persistent so we need to keep a reference around for + * mark and sweep to find - that's primarily what the + * btree_node_pending_free list is for. + * + * So here (when we set index_update_done = true), we're moving an + * existing reference to a different part of the larger "gc keyspace" - + * and the new position comes after the old position, since GC marks + * the pending free list after it walks the btree. + * + * If we move the reference while mark and sweep is _between_ the old + * and the new position, mark and sweep will see the reference twice + * and it'll get double accounted - so check for that here and subtract + * to cancel out one of mark and sweep's markings if necessary: + */ + + /* + * bch_mark_key() compares the current gc pos to the pos we're + * moving this reference from, hence one comparison here: + */ + if (gc_pos_cmp(c->gc_pos, gc_phase(GC_PHASE_PENDING_DELETE)) < 0) { + struct bucket_stats_cache_set tmp = { 0 }; + + bch_mark_key(c, bkey_i_to_s_c(&d->key), + -c->sb.btree_node_size, true, b + ? gc_pos_btree_node(b) + : gc_pos_btree_root(id), + &tmp, 0); + /* + * Don't apply tmp - pending deletes aren't tracked in + * cache_set_stats: + */ + } + + mutex_unlock(&c->btree_interior_update_lock); +} + +static void __btree_node_free(struct cache_set *c, struct btree *b, + struct btree_iter *iter) +{ + trace_bcache_btree_node_free(c, b); + + BUG_ON(b == btree_node_root(c, b)); + BUG_ON(b->ob); + BUG_ON(!list_empty(&b->write_blocked)); + + six_lock_write(&b->lock); + + if (btree_node_dirty(b)) + bch_btree_complete_write(c, b, btree_current_write(b)); + clear_btree_node_dirty(b); + + mca_hash_remove(c, b); + + mutex_lock(&c->btree_cache_lock); + list_move(&b->list, &c->btree_cache_freeable); + mutex_unlock(&c->btree_cache_lock); + + /* + * By using six_unlock_write() directly instead of + * btree_node_unlock_write(), we don't update the iterator's sequence + * numbers and cause future btree_node_relock() calls to fail: + */ + six_unlock_write(&b->lock); +} + +void bch_btree_node_free_never_inserted(struct cache_set *c, struct btree *b) +{ + struct open_bucket *ob = b->ob; + + b->ob = NULL; + + __btree_node_free(c, b, NULL); + + bch_open_bucket_put(c, ob); +} + +void bch_btree_node_free_inmem(struct btree_iter *iter, struct btree *b) +{ + bch_btree_iter_node_drop_linked(iter, b); + + __btree_node_free(iter->c, b, iter); + + bch_btree_iter_node_drop(iter, b); +} + +static void bch_btree_node_free_ondisk(struct cache_set *c, + struct pending_btree_node_free *pending) +{ + struct bucket_stats_cache_set stats = { 0 }; + + BUG_ON(!pending->index_update_done); + + bch_mark_key(c, bkey_i_to_s_c(&pending->key), + -c->sb.btree_node_size, true, + gc_phase(GC_PHASE_PENDING_DELETE), + &stats, 0); + /* + * Don't apply stats - pending deletes aren't tracked in + * cache_set_stats: + */ +} + +void btree_open_bucket_put(struct cache_set *c, struct btree *b) +{ + bch_open_bucket_put(c, b->ob); + b->ob = NULL; +} + +static struct btree *__bch_btree_node_alloc(struct cache_set *c, + bool use_reserve, + struct disk_reservation *res, + struct closure *cl) +{ + BKEY_PADDED(k) tmp; + struct open_bucket *ob; + struct btree *b; + unsigned reserve = use_reserve ? 0 : BTREE_NODE_RESERVE; + + mutex_lock(&c->btree_reserve_cache_lock); + if (c->btree_reserve_cache_nr > reserve) { + struct btree_alloc *a = + &c->btree_reserve_cache[--c->btree_reserve_cache_nr]; + + ob = a->ob; + bkey_copy(&tmp.k, &a->k); + mutex_unlock(&c->btree_reserve_cache_lock); + goto mem_alloc; + } + mutex_unlock(&c->btree_reserve_cache_lock); + +retry: + /* alloc_sectors is weird, I suppose */ + bkey_extent_init(&tmp.k); + tmp.k.k.size = c->sb.btree_node_size, + + ob = bch_alloc_sectors(c, &c->btree_write_point, + bkey_i_to_extent(&tmp.k), + res->nr_replicas, + use_reserve ? RESERVE_BTREE : RESERVE_NONE, + cl); + if (IS_ERR(ob)) + return ERR_CAST(ob); + + if (tmp.k.k.size < c->sb.btree_node_size) { + bch_open_bucket_put(c, ob); + goto retry; + } +mem_alloc: + b = mca_alloc(c); + + /* we hold cannibalize_lock: */ + BUG_ON(IS_ERR(b)); + BUG_ON(b->ob); + + bkey_copy(&b->key, &tmp.k); + b->key.k.size = 0; + b->ob = ob; + + return b; +} + +static struct btree *bch_btree_node_alloc(struct cache_set *c, + unsigned level, enum btree_id id, + struct btree_reserve *reserve) +{ + struct btree *b; + + BUG_ON(!reserve->nr); + + b = reserve->b[--reserve->nr]; + + BUG_ON(mca_hash_insert(c, b, level, id)); + + set_btree_node_accessed(b); + set_btree_node_dirty(b); + + bch_bset_init_first(b, &b->data->keys); + memset(&b->nr, 0, sizeof(b->nr)); + b->data->magic = cpu_to_le64(bset_magic(&c->disk_sb)); + SET_BSET_BTREE_LEVEL(&b->data->keys, level); + + bch_btree_build_aux_trees(b); + + bch_check_mark_super(c, &b->key, true); + + trace_bcache_btree_node_alloc(c, b); + return b; +} + +struct btree *__btree_node_alloc_replacement(struct cache_set *c, + struct btree *b, + struct bkey_format format, + struct btree_reserve *reserve) +{ + struct btree *n; + + n = bch_btree_node_alloc(c, b->level, b->btree_id, reserve); + + n->data->min_key = b->data->min_key; + n->data->max_key = b->data->max_key; + n->data->format = format; + + btree_node_set_format(n, format); + + bch_btree_sort_into(c, n, b); + + btree_node_reset_sib_u64s(n); + + n->key.k.p = b->key.k.p; + trace_bcache_btree_node_alloc_replacement(c, b, n); + + return n; +} + +struct btree *btree_node_alloc_replacement(struct cache_set *c, + struct btree *b, + struct btree_reserve *reserve) +{ + struct bkey_format new_f = bch_btree_calc_format(b); + + /* + * The keys might expand with the new format - if they wouldn't fit in + * the btree node anymore, use the old format for now: + */ + if (!bch_btree_node_format_fits(c, b, &new_f)) + new_f = b->format; + + return __btree_node_alloc_replacement(c, b, new_f, reserve); +} + +static void bch_btree_set_root_inmem(struct cache_set *c, struct btree *b, + struct btree_reserve *btree_reserve) +{ + struct btree *old = btree_node_root(c, b); + + /* Root nodes cannot be reaped */ + mutex_lock(&c->btree_cache_lock); + list_del_init(&b->list); + mutex_unlock(&c->btree_cache_lock); + + mutex_lock(&c->btree_root_lock); + btree_node_root(c, b) = b; + mutex_unlock(&c->btree_root_lock); + + if (btree_reserve) { + /* + * New allocation (we're not being called because we're in + * bch_btree_root_read()) - do marking while holding + * btree_root_lock: + */ + struct bucket_stats_cache_set stats = { 0 }; + + bch_mark_key(c, bkey_i_to_s_c(&b->key), + c->sb.btree_node_size, true, + gc_pos_btree_root(b->btree_id), + &stats, 0); + + if (old) + bch_btree_node_free_index(c, NULL, old->btree_id, + bkey_i_to_s_c(&old->key), + &stats); + bch_cache_set_stats_apply(c, &stats, &btree_reserve->disk_res, + gc_pos_btree_root(b->btree_id)); + } + + bch_recalc_btree_reserve(c); +} + +static void bch_btree_set_root_ondisk(struct cache_set *c, struct btree *b) +{ + struct btree_root *r = &c->btree_roots[b->btree_id]; + + mutex_lock(&c->btree_root_lock); + + BUG_ON(b != r->b); + bkey_copy(&r->key, &b->key); + r->level = b->level; + r->alive = true; + + mutex_unlock(&c->btree_root_lock); +} + +/* + * Only for cache set bringup, when first reading the btree roots or allocating + * btree roots when initializing a new cache set: + */ +void bch_btree_set_root_initial(struct cache_set *c, struct btree *b, + struct btree_reserve *btree_reserve) +{ + BUG_ON(btree_node_root(c, b)); + + bch_btree_set_root_inmem(c, b, btree_reserve); + bch_btree_set_root_ondisk(c, b); +} + +/** + * bch_btree_set_root - update the root in memory and on disk + * + * To ensure forward progress, the current task must not be holding any + * btree node write locks. However, you must hold an intent lock on the + * old root. + * + * Note: This allocates a journal entry but doesn't add any keys to + * it. All the btree roots are part of every journal write, so there + * is nothing new to be done. This just guarantees that there is a + * journal write. + */ +static void bch_btree_set_root(struct btree_iter *iter, struct btree *b, + struct btree_interior_update *as, + struct btree_reserve *btree_reserve) +{ + struct cache_set *c = iter->c; + struct btree *old; + + trace_bcache_btree_set_root(c, b); + BUG_ON(!b->written); + + old = btree_node_root(c, b); + + /* + * Ensure no one is using the old root while we switch to the + * new root: + */ + btree_node_lock_write(old, iter); + + bch_btree_set_root_inmem(c, b, btree_reserve); + + btree_interior_update_updated_root(c, as, iter->btree_id); + + /* + * Unlock old root after new root is visible: + * + * The new root isn't persistent, but that's ok: we still have + * an intent lock on the new root, and any updates that would + * depend on the new root would have to update the new root. + */ + btree_node_unlock_write(old, iter); +} + +static struct btree *__btree_root_alloc(struct cache_set *c, unsigned level, + enum btree_id id, + struct btree_reserve *reserve) +{ + struct btree *b = bch_btree_node_alloc(c, level, id, reserve); + + b->data->min_key = POS_MIN; + b->data->max_key = POS_MAX; + b->data->format = bch_btree_calc_format(b); + b->key.k.p = POS_MAX; + + btree_node_set_format(b, b->data->format); + bch_btree_build_aux_trees(b); + + six_unlock_write(&b->lock); + + return b; +} + +void bch_btree_reserve_put(struct cache_set *c, struct btree_reserve *reserve) +{ + bch_disk_reservation_put(c, &reserve->disk_res); + + mutex_lock(&c->btree_reserve_cache_lock); + + while (reserve->nr) { + struct btree *b = reserve->b[--reserve->nr]; + + six_unlock_write(&b->lock); + + if (c->btree_reserve_cache_nr < + ARRAY_SIZE(c->btree_reserve_cache)) { + struct btree_alloc *a = + &c->btree_reserve_cache[c->btree_reserve_cache_nr++]; + + a->ob = b->ob; + b->ob = NULL; + bkey_copy(&a->k, &b->key); + } else { + bch_open_bucket_put(c, b->ob); + b->ob = NULL; + } + + __btree_node_free(c, b, NULL); + + six_unlock_intent(&b->lock); + } + + mutex_unlock(&c->btree_reserve_cache_lock); + + mempool_free(reserve, &c->btree_reserve_pool); +} + +static struct btree_reserve *__bch_btree_reserve_get(struct cache_set *c, + unsigned nr_nodes, + unsigned flags, + struct closure *cl) +{ + struct btree_reserve *reserve; + struct btree *b; + struct disk_reservation disk_res = { 0, 0 }; + unsigned sectors = nr_nodes * c->sb.btree_node_size; + int ret, disk_res_flags = BCH_DISK_RESERVATION_GC_LOCK_HELD| + BCH_DISK_RESERVATION_METADATA; + + if (flags & BTREE_INSERT_NOFAIL) + disk_res_flags |= BCH_DISK_RESERVATION_NOFAIL; + + /* + * This check isn't necessary for correctness - it's just to potentially + * prevent us from doing a lot of work that'll end up being wasted: + */ + ret = bch_journal_error(&c->journal); + if (ret) + return ERR_PTR(ret); + + if (bch_disk_reservation_get(c, &disk_res, sectors, disk_res_flags)) + return ERR_PTR(-ENOSPC); + + BUG_ON(nr_nodes > BTREE_RESERVE_MAX); + + /* + * Protects reaping from the btree node cache and using the btree node + * open bucket reserve: + */ + ret = mca_cannibalize_lock(c, cl); + if (ret) { + bch_disk_reservation_put(c, &disk_res); + return ERR_PTR(ret); + } + + reserve = mempool_alloc(&c->btree_reserve_pool, GFP_NOIO); + + reserve->disk_res = disk_res; + reserve->nr = 0; + + while (reserve->nr < nr_nodes) { + b = __bch_btree_node_alloc(c, flags & BTREE_INSERT_USE_RESERVE, + &disk_res, cl); + if (IS_ERR(b)) { + ret = PTR_ERR(b); + goto err_free; + } + + reserve->b[reserve->nr++] = b; + } + + mca_cannibalize_unlock(c); + return reserve; +err_free: + bch_btree_reserve_put(c, reserve); + mca_cannibalize_unlock(c); + trace_bcache_btree_reserve_get_fail(c, nr_nodes, cl); + return ERR_PTR(ret); +} + +struct btree_reserve *bch_btree_reserve_get(struct cache_set *c, + struct btree *b, + unsigned extra_nodes, + unsigned flags, + struct closure *cl) +{ + unsigned depth = btree_node_root(c, b)->level - b->level; + unsigned nr_nodes = btree_reserve_required_nodes(depth) + extra_nodes; + + return __bch_btree_reserve_get(c, nr_nodes, flags, cl); + +} + +int bch_btree_root_alloc(struct cache_set *c, enum btree_id id, + struct closure *writes) +{ + struct closure cl; + struct btree_reserve *reserve; + struct btree *b; + + closure_init_stack(&cl); + + while (1) { + /* XXX haven't calculated capacity yet :/ */ + reserve = __bch_btree_reserve_get(c, 1, 0, &cl); + if (!IS_ERR(reserve)) + break; + + if (PTR_ERR(reserve) == -ENOSPC) + return PTR_ERR(reserve); + + closure_sync(&cl); + } + + b = __btree_root_alloc(c, 0, id, reserve); + + bch_btree_node_write(c, b, writes, SIX_LOCK_intent, -1); + + bch_btree_set_root_initial(c, b, reserve); + btree_open_bucket_put(c, b); + six_unlock_intent(&b->lock); + + bch_btree_reserve_put(c, reserve); + + return 0; +} + +static void bch_insert_fixup_btree_ptr(struct btree_iter *iter, + struct btree *b, + struct bkey_i *insert, + struct btree_node_iter *node_iter, + struct disk_reservation *disk_res) +{ + struct cache_set *c = iter->c; + struct bucket_stats_cache_set stats = { 0 }; + struct bkey_packed *k; + struct bkey tmp; + + if (bkey_extent_is_data(&insert->k)) + bch_mark_key(c, bkey_i_to_s_c(insert), + c->sb.btree_node_size, true, + gc_pos_btree_node(b), &stats, 0); + + while ((k = bch_btree_node_iter_peek_all(node_iter, b)) && + !btree_iter_pos_cmp_packed(b, &insert->k.p, k, false)) + bch_btree_node_iter_advance(node_iter, b); + + /* + * If we're overwriting, look up pending delete and mark so that gc + * marks it on the pending delete list: + */ + if (k && !bkey_cmp_packed(b, k, &insert->k)) + bch_btree_node_free_index(c, b, iter->btree_id, + bkey_disassemble(b, k, &tmp), + &stats); + + bch_cache_set_stats_apply(c, &stats, disk_res, gc_pos_btree_node(b)); + + bch_btree_bset_insert_key(iter, b, node_iter, insert); + set_btree_node_dirty(b); +} + +/* Inserting into a given leaf node (last stage of insert): */ + +/* Handle overwrites and do insert, for non extents: */ +bool bch_btree_bset_insert_key(struct btree_iter *iter, + struct btree *b, + struct btree_node_iter *node_iter, + struct bkey_i *insert) +{ + const struct bkey_format *f = &b->format; + struct bkey_packed *k; + struct bset_tree *t; + unsigned clobber_u64s; + + EBUG_ON(btree_node_just_written(b)); + EBUG_ON(bset_written(b, btree_bset_last(b))); + EBUG_ON(bkey_deleted(&insert->k) && bkey_val_u64s(&insert->k)); + EBUG_ON(bkey_cmp(bkey_start_pos(&insert->k), b->data->min_key) < 0 || + bkey_cmp(insert->k.p, b->data->max_key) > 0); + BUG_ON(insert->k.u64s > bch_btree_keys_u64s_remaining(iter->c, b)); + + k = bch_btree_node_iter_peek_all(node_iter, b); + if (k && !bkey_cmp_packed(b, k, &insert->k)) { + BUG_ON(bkey_whiteout(k)); + + t = bch_bkey_to_bset(b, k); + + if (bset_unwritten(b, bset(b, t)) && + bkey_val_u64s(&insert->k) == bkeyp_val_u64s(f, k)) { + BUG_ON(bkey_whiteout(k) != bkey_whiteout(&insert->k)); + + k->type = insert->k.type; + memcpy_u64s(bkeyp_val(f, k), &insert->v, + bkey_val_u64s(&insert->k)); + return true; + } + + insert->k.needs_whiteout = k->needs_whiteout; + + btree_keys_account_key_drop(&b->nr, t - b->set, k); + + if (t == bset_tree_last(b)) { + clobber_u64s = k->u64s; + + /* + * If we're deleting, and the key we're deleting doesn't + * need a whiteout (it wasn't overwriting a key that had + * been written to disk) - just delete it: + */ + if (bkey_whiteout(&insert->k) && !k->needs_whiteout) { + bch_bset_delete(b, k, clobber_u64s); + bch_btree_node_iter_fix(iter, b, node_iter, t, + k, clobber_u64s, 0); + return true; + } + + goto overwrite; + } + + k->type = KEY_TYPE_DELETED; + bch_btree_node_iter_fix(iter, b, node_iter, t, k, + k->u64s, k->u64s); + + if (bkey_whiteout(&insert->k)) { + reserve_whiteout(b, t, k); + return true; + } else { + k->needs_whiteout = false; + } + } else { + /* + * Deleting, but the key to delete wasn't found - nothing to do: + */ + if (bkey_whiteout(&insert->k)) + return false; + + insert->k.needs_whiteout = false; + } + + t = bset_tree_last(b); + k = bch_btree_node_iter_bset_pos(node_iter, b, t); + clobber_u64s = 0; +overwrite: + bch_bset_insert(b, node_iter, k, insert, clobber_u64s); + if (k->u64s != clobber_u64s || bkey_whiteout(&insert->k)) + bch_btree_node_iter_fix(iter, b, node_iter, t, k, + clobber_u64s, k->u64s); + return true; +} + +static void __btree_node_flush(struct journal *j, struct journal_entry_pin *pin, + unsigned i) +{ + struct cache_set *c = container_of(j, struct cache_set, journal); + struct btree_write *w = container_of(pin, struct btree_write, journal); + struct btree *b = container_of(w, struct btree, writes[i]); + + six_lock_read(&b->lock); + /* + * Reusing a btree node can race with the journal reclaim code calling + * the journal pin flush fn, and there's no good fix for this: we don't + * really want journal_pin_drop() to block until the flush fn is no + * longer running, because journal_pin_drop() is called from the btree + * node write endio function, and we can't wait on the flush fn to + * finish running in mca_reap() - where we make reused btree nodes ready + * to use again - because there, we're holding the lock this function + * needs - deadlock. + * + * So, the b->level check is a hack so we don't try to write nodes we + * shouldn't: + */ + if (!b->level) + bch_btree_node_write(c, b, NULL, SIX_LOCK_read, i); + six_unlock_read(&b->lock); +} + +static void btree_node_flush0(struct journal *j, struct journal_entry_pin *pin) +{ + return __btree_node_flush(j, pin, 0); +} + +static void btree_node_flush1(struct journal *j, struct journal_entry_pin *pin) +{ + return __btree_node_flush(j, pin, 1); +} + +void bch_btree_journal_key(struct btree_insert *trans, + struct btree_iter *iter, + struct bkey_i *insert) +{ + struct cache_set *c = trans->c; + struct journal *j = &c->journal; + struct btree *b = iter->nodes[0]; + struct btree_write *w = btree_current_write(b); + + EBUG_ON(iter->level || b->level); + EBUG_ON(!trans->journal_res.ref && + test_bit(JOURNAL_REPLAY_DONE, &j->flags)); + + if (!journal_pin_active(&w->journal)) + bch_journal_pin_add(j, &w->journal, + btree_node_write_idx(b) == 0 + ? btree_node_flush0 + : btree_node_flush1); + + if (trans->journal_res.ref) { + u64 seq = trans->journal_res.seq; + bool needs_whiteout = insert->k.needs_whiteout; + + /* + * have a bug where we're seeing an extent with an invalid crc + * entry in the journal, trying to track it down: + */ + BUG_ON(bkey_invalid(c, b->btree_id, bkey_i_to_s_c(insert))); + + /* ick */ + insert->k.needs_whiteout = false; + bch_journal_add_keys(j, &trans->journal_res, + b->btree_id, insert); + insert->k.needs_whiteout = needs_whiteout; + + if (trans->journal_seq) + *trans->journal_seq = seq; + btree_bset_last(b)->journal_seq = cpu_to_le64(seq); + } + + if (!btree_node_dirty(b)) + set_btree_node_dirty(b); +} + +static enum btree_insert_ret +bch_insert_fixup_key(struct btree_insert *trans, + struct btree_insert_entry *insert) +{ + struct btree_iter *iter = insert->iter; + + BUG_ON(iter->level); + + if (bch_btree_bset_insert_key(iter, + iter->nodes[0], + &iter->node_iters[0], + insert->k)) + bch_btree_journal_key(trans, iter, insert->k); + + trans->did_work = true; + return BTREE_INSERT_OK; +} + +static void verify_keys_sorted(struct keylist *l) +{ +#ifdef CONFIG_BCACHE_DEBUG + struct bkey_i *k; + + for_each_keylist_key(l, k) + BUG_ON(bkey_next(k) != l->top && + bkey_cmp(k->k.p, bkey_next(k)->k.p) >= 0); +#endif +} + +static void btree_node_lock_for_insert(struct btree *b, struct btree_iter *iter) +{ + struct cache_set *c = iter->c; + + btree_node_lock_write(b, iter); + + if (btree_node_just_written(b) && + bch_btree_post_write_cleanup(c, b)) + bch_btree_iter_reinit_node(iter, b); + + /* + * If the last bset has been written, or if it's gotten too big - start + * a new bset to insert into: + */ + if (want_new_bset(c, b)) + bch_btree_init_next(c, b, iter); +} + +/* Asynchronous interior node update machinery */ + +struct btree_interior_update * +bch_btree_interior_update_alloc(struct cache_set *c) +{ + struct btree_interior_update *as; + + as = mempool_alloc(&c->btree_interior_update_pool, GFP_NOIO); + memset(as, 0, sizeof(*as)); + closure_init(&as->cl, &c->cl); + as->c = c; + as->mode = BTREE_INTERIOR_NO_UPDATE; + + bch_keylist_init(&as->parent_keys, as->inline_keys, + ARRAY_SIZE(as->inline_keys)); + + mutex_lock(&c->btree_interior_update_lock); + list_add(&as->list, &c->btree_interior_update_list); + mutex_unlock(&c->btree_interior_update_lock); + + return as; +} + +static void btree_interior_update_free(struct closure *cl) +{ + struct btree_interior_update *as = container_of(cl, struct btree_interior_update, cl); + + mempool_free(as, &as->c->btree_interior_update_pool); +} + +static void btree_interior_update_nodes_reachable(struct closure *cl) +{ + struct btree_interior_update *as = + container_of(cl, struct btree_interior_update, cl); + struct cache_set *c = as->c; + unsigned i; + + bch_journal_pin_drop(&c->journal, &as->journal); + + mutex_lock(&c->btree_interior_update_lock); + + for (i = 0; i < as->nr_pending; i++) + bch_btree_node_free_ondisk(c, &as->pending[i]); + as->nr_pending = 0; + + mutex_unlock(&c->btree_interior_update_lock); + + mutex_lock(&c->btree_interior_update_lock); + list_del(&as->list); + mutex_unlock(&c->btree_interior_update_lock); + + closure_wake_up(&as->wait); + + closure_return_with_destructor(cl, btree_interior_update_free); +} + +static void btree_interior_update_nodes_written(struct closure *cl) +{ + struct btree_interior_update *as = + container_of(cl, struct btree_interior_update, cl); + struct cache_set *c = as->c; + struct btree *b; + + if (bch_journal_error(&c->journal)) { + /* XXX what? */ + } + + /* XXX: missing error handling, damnit */ + + /* check for journal error, bail out if we flushed */ + + /* + * We did an update to a parent node where the pointers we added pointed + * to child nodes that weren't written yet: now, the child nodes have + * been written so we can write out the update to the interior node. + */ +retry: + mutex_lock(&c->btree_interior_update_lock); + switch (as->mode) { + case BTREE_INTERIOR_NO_UPDATE: + BUG(); + case BTREE_INTERIOR_UPDATING_NODE: + /* The usual case: */ + b = READ_ONCE(as->b); + + if (!six_trylock_read(&b->lock)) { + mutex_unlock(&c->btree_interior_update_lock); + six_lock_read(&b->lock); + six_unlock_read(&b->lock); + goto retry; + } + + BUG_ON(!btree_node_dirty(b)); + closure_wait(&btree_current_write(b)->wait, cl); + + list_del(&as->write_blocked_list); + + if (list_empty(&b->write_blocked)) + bch_btree_node_write(c, b, NULL, SIX_LOCK_read, -1); + six_unlock_read(&b->lock); + break; + + case BTREE_INTERIOR_UPDATING_AS: + /* + * The btree node we originally updated has been freed and is + * being rewritten - so we need to write anything here, we just + * need to signal to that btree_interior_update that it's ok to make the + * new replacement node visible: + */ + closure_put(&as->parent_as->cl); + + /* + * and then we have to wait on that btree_interior_update to finish: + */ + closure_wait(&as->parent_as->wait, cl); + break; + + case BTREE_INTERIOR_UPDATING_ROOT: + /* b is the new btree root: */ + b = READ_ONCE(as->b); + + if (!six_trylock_read(&b->lock)) { + mutex_unlock(&c->btree_interior_update_lock); + six_lock_read(&b->lock); + six_unlock_read(&b->lock); + goto retry; + } + + BUG_ON(c->btree_roots[b->btree_id].as != as); + c->btree_roots[b->btree_id].as = NULL; + + bch_btree_set_root_ondisk(c, b); + + /* + * We don't have to wait anything anything here (before + * btree_interior_update_nodes_reachable frees the old nodes + * ondisk) - we've ensured that the very next journal write will + * have the pointer to the new root, and before the allocator + * can reuse the old nodes it'll have to do a journal commit: + */ + six_unlock_read(&b->lock); + } + mutex_unlock(&c->btree_interior_update_lock); + + continue_at(cl, btree_interior_update_nodes_reachable, system_wq); +} + +/* + * We're updating @b with pointers to nodes that haven't finished writing yet: + * block @b from being written until @as completes + */ +static void btree_interior_update_updated_btree(struct cache_set *c, + struct btree_interior_update *as, + struct btree *b) +{ + mutex_lock(&c->btree_interior_update_lock); + + BUG_ON(as->mode != BTREE_INTERIOR_NO_UPDATE); + BUG_ON(!btree_node_dirty(b)); + + as->mode = BTREE_INTERIOR_UPDATING_NODE; + as->b = b; + list_add(&as->write_blocked_list, &b->write_blocked); + + mutex_unlock(&c->btree_interior_update_lock); + + bch_journal_wait_on_seq(&c->journal, as->journal_seq, &as->cl); + + continue_at(&as->cl, btree_interior_update_nodes_written, + system_freezable_wq); +} + +static void btree_interior_update_updated_root(struct cache_set *c, + struct btree_interior_update *as, + enum btree_id btree_id) +{ + struct btree_root *r = &c->btree_roots[btree_id]; + + mutex_lock(&c->btree_interior_update_lock); + + BUG_ON(as->mode != BTREE_INTERIOR_NO_UPDATE); + + /* + * Old root might not be persistent yet - if so, redirect its + * btree_interior_update operation to point to us: + */ + if (r->as) { + BUG_ON(r->as->mode != BTREE_INTERIOR_UPDATING_ROOT); + + r->as->b = NULL; + r->as->mode = BTREE_INTERIOR_UPDATING_AS; + r->as->parent_as = as; + closure_get(&as->cl); + } + + as->mode = BTREE_INTERIOR_UPDATING_ROOT; + as->b = r->b; + r->as = as; + + mutex_unlock(&c->btree_interior_update_lock); + + bch_journal_wait_on_seq(&c->journal, as->journal_seq, &as->cl); + + continue_at(&as->cl, btree_interior_update_nodes_written, + system_freezable_wq); +} + +static void interior_update_flush(struct journal *j, struct journal_entry_pin *pin) +{ + struct btree_interior_update *as = + container_of(pin, struct btree_interior_update, journal); + + bch_journal_flush_seq_async(j, as->journal_seq, NULL); +} + +/* + * @b is being split/rewritten: it may have pointers to not-yet-written btree + * nodes and thus outstanding btree_interior_updates - redirect @b's + * btree_interior_updates to point to this btree_interior_update: + */ +void bch_btree_interior_update_will_free_node(struct cache_set *c, + struct btree_interior_update *as, + struct btree *b) +{ + struct btree_interior_update *p, *n; + struct pending_btree_node_free *d; + struct bset_tree *t; + + /* + * Does this node have data that hasn't been written in the journal? + * + * If so, we have to wait for the corresponding journal entry to be + * written before making the new nodes reachable - we can't just carry + * over the bset->journal_seq tracking, since we'll be mixing those keys + * in with keys that aren't in the journal anymore: + */ + for_each_bset(b, t) + as->journal_seq = max(as->journal_seq, bset(b, t)->journal_seq); + + /* + * Does this node have unwritten data that has a pin on the journal? + * + * If so, transfer that pin to the btree_interior_update operation - + * note that if we're freeing multiple nodes, we only need to keep the + * oldest pin of any of the nodes we're freeing. We'll release the pin + * when the new nodes are persistent and reachable on disk: + */ + bch_journal_pin_add_if_older(&c->journal, + &b->writes[0].journal, + &as->journal, interior_update_flush); + bch_journal_pin_add_if_older(&c->journal, + &b->writes[1].journal, + &as->journal, interior_update_flush); + + mutex_lock(&c->btree_interior_update_lock); + + /* + * Does this node have any btree_interior_update operations preventing + * it from being written? + * + * If so, redirect them to point to this btree_interior_update: we can + * write out our new nodes, but we won't make them visible until those + * operations complete + */ + list_for_each_entry_safe(p, n, &b->write_blocked, write_blocked_list) { + BUG_ON(p->mode != BTREE_INTERIOR_UPDATING_NODE); + + p->mode = BTREE_INTERIOR_UPDATING_AS; + list_del(&p->write_blocked_list); + p->b = NULL; + p->parent_as = as; + closure_get(&as->cl); + } + + /* Add this node to the list of nodes being freed: */ + BUG_ON(as->nr_pending >= ARRAY_SIZE(as->pending)); + + d = &as->pending[as->nr_pending++]; + d->index_update_done = false; + d->seq = b->data->keys.seq; + d->btree_id = b->btree_id; + d->level = b->level; + bkey_copy(&d->key, &b->key); + + mutex_unlock(&c->btree_interior_update_lock); +} + +static void btree_node_interior_verify(struct btree *b) +{ + struct btree_node_iter iter; + struct bkey_packed *k; + + BUG_ON(!b->level); + + bch_btree_node_iter_init(&iter, b, b->key.k.p, false, false); +#if 1 + BUG_ON(!(k = bch_btree_node_iter_peek(&iter, b)) || + bkey_cmp_left_packed(b, k, &b->key.k.p)); + + BUG_ON((bch_btree_node_iter_advance(&iter, b), + !bch_btree_node_iter_end(&iter))); +#else + const char *msg; + + msg = "not found"; + k = bch_btree_node_iter_peek(&iter, b); + if (!k) + goto err; + + msg = "isn't what it should be"; + if (bkey_cmp_left_packed(b, k, &b->key.k.p)) + goto err; + + bch_btree_node_iter_advance(&iter, b); + + msg = "isn't last key"; + if (!bch_btree_node_iter_end(&iter)) + goto err; + return; +err: + bch_dump_btree_node(b); + printk(KERN_ERR "last key %llu:%llu %s\n", b->key.k.p.inode, + b->key.k.p.offset, msg); + BUG(); +#endif +} + +static enum btree_insert_ret +bch_btree_insert_keys_interior(struct btree *b, + struct btree_iter *iter, + struct keylist *insert_keys, + struct btree_interior_update *as, + struct btree_reserve *res) +{ + struct cache_set *c = iter->c; + struct btree_iter *linked; + struct btree_node_iter node_iter; + struct bkey_i *insert = bch_keylist_front(insert_keys); + struct bkey_packed *k; + + BUG_ON(!btree_node_intent_locked(iter, btree_node_root(c, b)->level)); + BUG_ON(!b->level); + BUG_ON(!as || as->b); + verify_keys_sorted(insert_keys); + + btree_node_lock_for_insert(b, iter); + + if (bch_keylist_u64s(insert_keys) > + bch_btree_keys_u64s_remaining(c, b)) { + btree_node_unlock_write(b, iter); + return BTREE_INSERT_BTREE_NODE_FULL; + } + + /* Don't screw up @iter's position: */ + node_iter = iter->node_iters[b->level]; + + /* + * btree_split(), btree_gc_coalesce() will insert keys before + * the iterator's current position - they know the keys go in + * the node the iterator points to: + */ + while ((k = bch_btree_node_iter_prev_all(&node_iter, b)) && + (bkey_cmp_packed(b, k, &insert->k) >= 0)) + ; + + while (!bch_keylist_empty(insert_keys)) { + insert = bch_keylist_front(insert_keys); + + bch_insert_fixup_btree_ptr(iter, b, insert, + &node_iter, &res->disk_res); + bch_keylist_pop_front(insert_keys); + } + + btree_interior_update_updated_btree(c, as, b); + + for_each_linked_btree_node(iter, b, linked) + bch_btree_node_iter_peek(&linked->node_iters[b->level], + b); + bch_btree_node_iter_peek(&iter->node_iters[b->level], b); + + bch_btree_iter_verify(iter, b); + + if (bch_maybe_compact_whiteouts(c, b)) + bch_btree_iter_reinit_node(iter, b); + + btree_node_unlock_write(b, iter); + + btree_node_interior_verify(b); + return BTREE_INSERT_OK; +} + +/* + * Move keys from n1 (original replacement node, now lower node) to n2 (higher + * node) + */ +static struct btree *__btree_split_node(struct btree_iter *iter, struct btree *n1, + struct btree_reserve *reserve) +{ + size_t nr_packed = 0, nr_unpacked = 0; + struct btree *n2; + struct bset *set1, *set2; + struct bkey_packed *k, *prev = NULL; + + n2 = bch_btree_node_alloc(iter->c, n1->level, iter->btree_id, reserve); + n2->data->max_key = n1->data->max_key; + n2->data->format = n1->format; + n2->key.k.p = n1->key.k.p; + + btree_node_set_format(n2, n2->data->format); + + set1 = btree_bset_first(n1); + set2 = btree_bset_first(n2); + + /* + * Has to be a linear search because we don't have an auxiliary + * search tree yet + */ + k = set1->start; + while (1) { + if (bkey_next(k) == bset_bkey_last(set1)) + break; + if (k->_data - set1->_data >= (le16_to_cpu(set1->u64s) * 3) / 5) + break; + + if (bkey_packed(k)) + nr_packed++; + else + nr_unpacked++; + + prev = k; + k = bkey_next(k); + } + + BUG_ON(!prev); + + n1->key.k.p = bkey_unpack_pos(n1, prev); + n1->data->max_key = n1->key.k.p; + n2->data->min_key = + btree_type_successor(n1->btree_id, n1->key.k.p); + + set2->u64s = cpu_to_le16((u64 *) bset_bkey_last(set1) - (u64 *) k); + set1->u64s = cpu_to_le16(le16_to_cpu(set1->u64s) - le16_to_cpu(set2->u64s)); + + set_btree_bset_end(n1, n1->set); + set_btree_bset_end(n2, n2->set); + + n2->nr.live_u64s = le16_to_cpu(set2->u64s); + n2->nr.bset_u64s[0] = le16_to_cpu(set2->u64s); + n2->nr.packed_keys = n1->nr.packed_keys - nr_packed; + n2->nr.unpacked_keys = n1->nr.unpacked_keys - nr_unpacked; + + n1->nr.live_u64s = le16_to_cpu(set1->u64s); + n1->nr.bset_u64s[0] = le16_to_cpu(set1->u64s); + n1->nr.packed_keys = nr_packed; + n1->nr.unpacked_keys = nr_unpacked; + + BUG_ON(!set1->u64s); + BUG_ON(!set2->u64s); + + memcpy_u64s(set2->start, + bset_bkey_last(set1), + le16_to_cpu(set2->u64s)); + + btree_node_reset_sib_u64s(n1); + btree_node_reset_sib_u64s(n2); + + bch_verify_btree_nr_keys(n1); + bch_verify_btree_nr_keys(n2); + + if (n1->level) { + btree_node_interior_verify(n1); + btree_node_interior_verify(n2); + } + + return n2; +} + +/* + * For updates to interior nodes, we've got to do the insert before we split + * because the stuff we're inserting has to be inserted atomically. Post split, + * the keys might have to go in different nodes and the split would no longer be + * atomic. + * + * Worse, if the insert is from btree node coalescing, if we do the insert after + * we do the split (and pick the pivot) - the pivot we pick might be between + * nodes that were coalesced, and thus in the middle of a child node post + * coalescing: + */ +static void btree_split_insert_keys(struct btree_iter *iter, struct btree *b, + struct keylist *keys, + struct btree_reserve *res) +{ + struct btree_node_iter node_iter; + struct bkey_i *k = bch_keylist_front(keys); + struct bkey_packed *p; + struct bset *i; + + BUG_ON(btree_node_type(b) != BKEY_TYPE_BTREE); + + bch_btree_node_iter_init(&node_iter, b, k->k.p, false, false); + + while (!bch_keylist_empty(keys)) { + k = bch_keylist_front(keys); + + BUG_ON(bch_keylist_u64s(keys) > + bch_btree_keys_u64s_remaining(iter->c, b)); + BUG_ON(bkey_cmp(k->k.p, b->data->min_key) < 0); + BUG_ON(bkey_cmp(k->k.p, b->data->max_key) > 0); + + bch_insert_fixup_btree_ptr(iter, b, k, &node_iter, &res->disk_res); + bch_keylist_pop_front(keys); + } + + /* + * We can't tolerate whiteouts here - with whiteouts there can be + * duplicate keys, and it would be rather bad if we picked a duplicate + * for the pivot: + */ + i = btree_bset_first(b); + p = i->start; + while (p != bset_bkey_last(i)) + if (bkey_deleted(p)) { + le16_add_cpu(&i->u64s, -p->u64s); + set_btree_bset_end(b, b->set); + memmove_u64s_down(p, bkey_next(p), + (u64 *) bset_bkey_last(i) - + (u64 *) p); + } else + p = bkey_next(p); + + BUG_ON(b->nsets != 1 || + b->nr.live_u64s != le16_to_cpu(btree_bset_first(b)->u64s)); + + btree_node_interior_verify(b); +} + +static void btree_split(struct btree *b, struct btree_iter *iter, + struct keylist *insert_keys, + struct btree_reserve *reserve, + struct btree_interior_update *as) +{ + struct cache_set *c = iter->c; + struct btree *parent = iter->nodes[b->level + 1]; + struct btree *n1, *n2 = NULL, *n3 = NULL; + u64 start_time = local_clock(); + + BUG_ON(!parent && (b != btree_node_root(c, b))); + BUG_ON(!btree_node_intent_locked(iter, btree_node_root(c, b)->level)); + + bch_btree_interior_update_will_free_node(c, as, b); + + n1 = btree_node_alloc_replacement(c, b, reserve); + if (b->level) + btree_split_insert_keys(iter, n1, insert_keys, reserve); + + if (__set_blocks(n1->data, + le16_to_cpu(n1->data->keys.u64s), + block_bytes(c)) > BTREE_SPLIT_THRESHOLD(c)) { + trace_bcache_btree_node_split(c, b, b->nr.live_u64s); + + n2 = __btree_split_node(iter, n1, reserve); + + bch_btree_build_aux_trees(n2); + bch_btree_build_aux_trees(n1); + six_unlock_write(&n2->lock); + six_unlock_write(&n1->lock); + + bch_btree_node_write(c, n2, &as->cl, SIX_LOCK_intent, -1); + + /* + * Note that on recursive parent_keys == insert_keys, so we + * can't start adding new keys to parent_keys before emptying it + * out (which we did with btree_split_insert_keys() above) + */ + bch_keylist_add(&as->parent_keys, &n1->key); + bch_keylist_add(&as->parent_keys, &n2->key); + + if (!parent) { + /* Depth increases, make a new root */ + n3 = __btree_root_alloc(c, b->level + 1, + iter->btree_id, + reserve); + n3->sib_u64s[0] = U16_MAX; + n3->sib_u64s[1] = U16_MAX; + + btree_split_insert_keys(iter, n3, &as->parent_keys, + reserve); + bch_btree_node_write(c, n3, &as->cl, SIX_LOCK_intent, -1); + } + } else { + trace_bcache_btree_node_compact(c, b, b->nr.live_u64s); + + bch_btree_build_aux_trees(n1); + six_unlock_write(&n1->lock); + + bch_keylist_add(&as->parent_keys, &n1->key); + } + + bch_btree_node_write(c, n1, &as->cl, SIX_LOCK_intent, -1); + + /* New nodes all written, now make them visible: */ + + if (parent) { + /* Split a non root node */ + bch_btree_insert_node(parent, iter, &as->parent_keys, + reserve, as); + } else if (n3) { + bch_btree_set_root(iter, n3, as, reserve); + } else { + /* Root filled up but didn't need to be split */ + bch_btree_set_root(iter, n1, as, reserve); + } + + btree_open_bucket_put(c, n1); + if (n2) + btree_open_bucket_put(c, n2); + if (n3) + btree_open_bucket_put(c, n3); + + /* + * Note - at this point other linked iterators could still have @b read + * locked; we're depending on the bch_btree_iter_node_replace() calls + * below removing all references to @b so we don't return with other + * iterators pointing to a node they have locked that's been freed. + * + * We have to free the node first because the bch_iter_node_replace() + * calls will drop _our_ iterator's reference - and intent lock - to @b. + */ + bch_btree_node_free_inmem(iter, b); + + /* Successful split, update the iterator to point to the new nodes: */ + + if (n3) + bch_btree_iter_node_replace(iter, n3); + if (n2) + bch_btree_iter_node_replace(iter, n2); + bch_btree_iter_node_replace(iter, n1); + + bch_time_stats_update(&c->btree_split_time, start_time); +} + +/** + * bch_btree_insert_node - insert bkeys into a given btree node + * + * @iter: btree iterator + * @insert_keys: list of keys to insert + * @hook: insert callback + * @persistent: if not null, @persistent will wait on journal write + * + * Inserts as many keys as it can into a given btree node, splitting it if full. + * If a split occurred, this function will return early. This can only happen + * for leaf nodes -- inserts into interior nodes have to be atomic. + */ +void bch_btree_insert_node(struct btree *b, + struct btree_iter *iter, + struct keylist *insert_keys, + struct btree_reserve *reserve, + struct btree_interior_update *as) +{ + BUG_ON(!b->level); + BUG_ON(!reserve || !as); + + switch (bch_btree_insert_keys_interior(b, iter, insert_keys, + as, reserve)) { + case BTREE_INSERT_OK: + break; + case BTREE_INSERT_BTREE_NODE_FULL: + btree_split(b, iter, insert_keys, reserve, as); + break; + default: + BUG(); + } +} + +static int bch_btree_split_leaf(struct btree_iter *iter, unsigned flags) +{ + struct cache_set *c = iter->c; + struct btree *b = iter->nodes[0]; + struct btree_reserve *reserve; + struct btree_interior_update *as; + struct closure cl; + int ret = 0; + + closure_init_stack(&cl); + + /* Hack, because gc and splitting nodes doesn't mix yet: */ + if (!down_read_trylock(&c->gc_lock)) { + bch_btree_iter_unlock(iter); + down_read(&c->gc_lock); + } + + /* + * XXX: figure out how far we might need to split, + * instead of locking/reserving all the way to the root: + */ + if (!bch_btree_iter_set_locks_want(iter, U8_MAX)) { + ret = -EINTR; + goto out; + } + + reserve = bch_btree_reserve_get(c, b, 0, flags, &cl); + if (IS_ERR(reserve)) { + ret = PTR_ERR(reserve); + if (ret == -EAGAIN) { + bch_btree_iter_unlock(iter); + up_read(&c->gc_lock); + closure_sync(&cl); + return -EINTR; + } + goto out; + } + + as = bch_btree_interior_update_alloc(c); + + btree_split(b, iter, NULL, reserve, as); + bch_btree_reserve_put(c, reserve); + + bch_btree_iter_set_locks_want(iter, 1); +out: + up_read(&c->gc_lock); + return ret; +} + +enum btree_node_sibling { + btree_prev_sib, + btree_next_sib, +}; + +static struct btree *btree_node_get_sibling(struct btree_iter *iter, + struct btree *b, + enum btree_node_sibling sib) +{ + struct btree *parent; + struct btree_node_iter node_iter; + struct bkey_packed *k; + BKEY_PADDED(k) tmp; + struct btree *ret; + unsigned level = b->level; + + parent = iter->nodes[level + 1]; + if (!parent) + return NULL; + + if (!btree_node_relock(iter, level + 1)) { + bch_btree_iter_set_locks_want(iter, level + 2); + return ERR_PTR(-EINTR); + } + + node_iter = iter->node_iters[parent->level]; + + k = bch_btree_node_iter_peek_all(&node_iter, parent); + BUG_ON(bkey_cmp_left_packed(parent, k, &b->key.k.p)); + + do { + k = sib == btree_prev_sib + ? bch_btree_node_iter_prev_all(&node_iter, parent) + : (bch_btree_node_iter_advance(&node_iter, parent), + bch_btree_node_iter_peek_all(&node_iter, parent)); + if (!k) + return NULL; + } while (bkey_deleted(k)); + + bkey_unpack(parent, &tmp.k, k); + + ret = bch_btree_node_get(iter, &tmp.k, level, SIX_LOCK_intent); + + if (IS_ERR(ret) && PTR_ERR(ret) == -EINTR) { + btree_node_unlock(iter, level); + ret = bch_btree_node_get(iter, &tmp.k, level, SIX_LOCK_intent); + } + + if (!IS_ERR(ret) && !btree_node_relock(iter, level)) { + six_unlock_intent(&ret->lock); + ret = ERR_PTR(-EINTR); + } + + return ret; +} + +static int __foreground_maybe_merge(struct btree_iter *iter, + enum btree_node_sibling sib) +{ + struct cache_set *c = iter->c; + struct btree_reserve *reserve; + struct btree_interior_update *as; + struct bkey_format_state new_s; + struct bkey_format new_f; + struct bkey_i delete; + struct btree *b, *m, *n, *prev, *next, *parent; + struct closure cl; + size_t sib_u64s; + int ret = 0; + + closure_init_stack(&cl); +retry: + if (!btree_node_relock(iter, iter->level)) + return 0; + + b = iter->nodes[iter->level]; + + parent = iter->nodes[b->level + 1]; + if (!parent) + return 0; + + if (b->sib_u64s[sib] > BTREE_FOREGROUND_MERGE_THRESHOLD(c)) + return 0; + + /* XXX: can't be holding read locks */ + m = btree_node_get_sibling(iter, b, sib); + if (IS_ERR(m)) { + ret = PTR_ERR(m); + goto out; + } + + /* NULL means no sibling: */ + if (!m) { + b->sib_u64s[sib] = U16_MAX; + return 0; + } + + if (sib == btree_prev_sib) { + prev = m; + next = b; + } else { + prev = b; + next = m; + } + + bch_bkey_format_init(&new_s); + __bch_btree_calc_format(&new_s, b); + __bch_btree_calc_format(&new_s, m); + new_f = bch_bkey_format_done(&new_s); + + sib_u64s = btree_node_u64s_with_format(b, &new_f) + + btree_node_u64s_with_format(m, &new_f); + + if (sib_u64s > BTREE_FOREGROUND_MERGE_HYSTERESIS(c)) { + sib_u64s -= BTREE_FOREGROUND_MERGE_HYSTERESIS(c); + sib_u64s /= 2; + sib_u64s += BTREE_FOREGROUND_MERGE_HYSTERESIS(c); + } + + sib_u64s = min(sib_u64s, btree_max_u64s(c)); + b->sib_u64s[sib] = sib_u64s; + + if (b->sib_u64s[sib] > BTREE_FOREGROUND_MERGE_THRESHOLD(c)) { + six_unlock_intent(&m->lock); + return 0; + } + + /* We're changing btree topology, doesn't mix with gc: */ + if (!down_read_trylock(&c->gc_lock)) { + six_unlock_intent(&m->lock); + bch_btree_iter_unlock(iter); + + down_read(&c->gc_lock); + up_read(&c->gc_lock); + ret = -EINTR; + goto out; + } + + if (!bch_btree_iter_set_locks_want(iter, U8_MAX)) { + ret = -EINTR; + goto out_unlock; + } + + reserve = bch_btree_reserve_get(c, b, 0, + BTREE_INSERT_NOFAIL| + BTREE_INSERT_USE_RESERVE, + &cl); + if (IS_ERR(reserve)) { + ret = PTR_ERR(reserve); + goto out_unlock; + } + + as = bch_btree_interior_update_alloc(c); + + bch_btree_interior_update_will_free_node(c, as, b); + bch_btree_interior_update_will_free_node(c, as, m); + + n = bch_btree_node_alloc(c, b->level, b->btree_id, reserve); + n->data->min_key = prev->data->min_key; + n->data->max_key = next->data->max_key; + n->data->format = new_f; + n->key.k.p = next->key.k.p; + + btree_node_set_format(n, new_f); + + bch_btree_sort_into(c, n, prev); + bch_btree_sort_into(c, n, next); + + bch_btree_build_aux_trees(n); + six_unlock_write(&n->lock); + + bkey_init(&delete.k); + delete.k.p = prev->key.k.p; + bch_keylist_add(&as->parent_keys, &delete); + bch_keylist_add(&as->parent_keys, &n->key); + + bch_btree_node_write(c, n, &as->cl, SIX_LOCK_intent, -1); + + bch_btree_insert_node(parent, iter, &as->parent_keys, reserve, as); + + btree_open_bucket_put(c, n); + bch_btree_node_free_inmem(iter, b); + bch_btree_node_free_inmem(iter, m); + bch_btree_iter_node_replace(iter, n); + + bch_btree_iter_verify(iter, n); + + bch_btree_reserve_put(c, reserve); +out_unlock: + if (ret != -EINTR && ret != -EAGAIN) + bch_btree_iter_set_locks_want(iter, 1); + six_unlock_intent(&m->lock); + up_read(&c->gc_lock); +out: + if (ret == -EAGAIN || ret == -EINTR) { + bch_btree_iter_unlock(iter); + ret = -EINTR; + } + + closure_sync(&cl); + + if (ret == -EINTR) { + ret = bch_btree_iter_traverse(iter); + if (!ret) + goto retry; + } + + return ret; +} + +static int inline foreground_maybe_merge(struct btree_iter *iter, + enum btree_node_sibling sib) +{ + struct cache_set *c = iter->c; + struct btree *b; + + if (!btree_node_locked(iter, iter->level)) + return 0; + + b = iter->nodes[iter->level]; + if (b->sib_u64s[sib] > BTREE_FOREGROUND_MERGE_THRESHOLD(c)) + return 0; + + return __foreground_maybe_merge(iter, sib); +} + +/** + * btree_insert_key - insert a key one key into a leaf node + */ +static enum btree_insert_ret +btree_insert_key(struct btree_insert *trans, + struct btree_insert_entry *insert) +{ + struct cache_set *c = trans->c; + struct btree_iter *iter = insert->iter; + struct btree *b = iter->nodes[0]; + enum btree_insert_ret ret; + int old_u64s = le16_to_cpu(btree_bset_last(b)->u64s); + int old_live_u64s = b->nr.live_u64s; + int live_u64s_added, u64s_added; + + ret = !btree_node_is_extents(b) + ? bch_insert_fixup_key(trans, insert) + : bch_insert_fixup_extent(trans, insert); + + live_u64s_added = (int) b->nr.live_u64s - old_live_u64s; + u64s_added = (int) le16_to_cpu(btree_bset_last(b)->u64s) - old_u64s; + + if (b->sib_u64s[0] != U16_MAX && live_u64s_added < 0) + b->sib_u64s[0] = max(0, (int) b->sib_u64s[0] + live_u64s_added); + if (b->sib_u64s[1] != U16_MAX && live_u64s_added < 0) + b->sib_u64s[1] = max(0, (int) b->sib_u64s[1] + live_u64s_added); + + if (u64s_added > live_u64s_added && + bch_maybe_compact_whiteouts(iter->c, b)) + bch_btree_iter_reinit_node(iter, b); + + trace_bcache_btree_insert_key(c, b, insert->k); + return ret; +} + +static bool same_leaf_as_prev(struct btree_insert *trans, + struct btree_insert_entry *i) +{ + /* + * Because we sorted the transaction entries, if multiple iterators + * point to the same leaf node they'll always be adjacent now: + */ + return i != trans->entries && + i[0].iter->nodes[0] == i[-1].iter->nodes[0]; +} + +#define trans_for_each_entry(trans, i) \ + for ((i) = (trans)->entries; (i) < (trans)->entries + (trans)->nr; (i)++) + +static void multi_lock_write(struct btree_insert *trans) +{ + struct btree_insert_entry *i; + + trans_for_each_entry(trans, i) + if (!same_leaf_as_prev(trans, i)) + btree_node_lock_for_insert(i->iter->nodes[0], i->iter); +} + +static void multi_unlock_write(struct btree_insert *trans) +{ + struct btree_insert_entry *i; + + trans_for_each_entry(trans, i) + if (!same_leaf_as_prev(trans, i)) + btree_node_unlock_write(i->iter->nodes[0], i->iter); +} + +static int btree_trans_entry_cmp(const void *_l, const void *_r) +{ + const struct btree_insert_entry *l = _l; + const struct btree_insert_entry *r = _r; + + return btree_iter_cmp(l->iter, r->iter); +} + +/* Normal update interface: */ + +/** + * __bch_btree_insert_at - insert keys at given iterator positions + * + * This is main entry point for btree updates. + * + * Return values: + * -EINTR: locking changed, this function should be called again. Only returned + * if passed BTREE_INSERT_ATOMIC. + * -EROFS: cache set read only + * -EIO: journal or btree node IO error + */ +int __bch_btree_insert_at(struct btree_insert *trans) +{ + struct cache_set *c = trans->c; + struct btree_insert_entry *i; + struct btree_iter *split = NULL; + bool cycle_gc_lock = false; + unsigned u64s; + int ret; + + trans_for_each_entry(trans, i) { + EBUG_ON(i->iter->level); + EBUG_ON(bkey_cmp(bkey_start_pos(&i->k->k), i->iter->pos)); + } + + sort(trans->entries, trans->nr, sizeof(trans->entries[0]), + btree_trans_entry_cmp, NULL); + + if (unlikely(!percpu_ref_tryget(&c->writes))) + return -EROFS; +retry_locks: + ret = -EINTR; + trans_for_each_entry(trans, i) + if (!bch_btree_iter_set_locks_want(i->iter, 1)) + goto err; +retry: + trans->did_work = false; + u64s = 0; + trans_for_each_entry(trans, i) + if (!i->done) + u64s += jset_u64s(i->k->k.u64s); + + memset(&trans->journal_res, 0, sizeof(trans->journal_res)); + + ret = !(trans->flags & BTREE_INSERT_JOURNAL_REPLAY) + ? bch_journal_res_get(&c->journal, + &trans->journal_res, + u64s, u64s) + : 0; + if (ret) + goto err; + + multi_lock_write(trans); + + u64s = 0; + trans_for_each_entry(trans, i) { + /* Multiple inserts might go to same leaf: */ + if (!same_leaf_as_prev(trans, i)) + u64s = 0; + + /* + * bch_btree_node_insert_fits() must be called under write lock: + * with only an intent lock, another thread can still call + * bch_btree_node_write(), converting an unwritten bset to a + * written one + */ + if (!i->done) { + u64s += i->k->k.u64s; + if (!bch_btree_node_insert_fits(c, + i->iter->nodes[0], u64s)) { + split = i->iter; + goto unlock; + } + } + } + + ret = 0; + split = NULL; + cycle_gc_lock = false; + + trans_for_each_entry(trans, i) { + if (i->done) + continue; + + switch (btree_insert_key(trans, i)) { + case BTREE_INSERT_OK: + i->done = true; + break; + case BTREE_INSERT_JOURNAL_RES_FULL: + case BTREE_INSERT_NEED_TRAVERSE: + ret = -EINTR; + break; + case BTREE_INSERT_NEED_RESCHED: + ret = -EAGAIN; + break; + case BTREE_INSERT_BTREE_NODE_FULL: + split = i->iter; + break; + case BTREE_INSERT_ENOSPC: + ret = -ENOSPC; + break; + case BTREE_INSERT_NEED_GC_LOCK: + cycle_gc_lock = true; + ret = -EINTR; + break; + default: + BUG(); + } + + if (!trans->did_work && (ret || split)) + break; + } +unlock: + multi_unlock_write(trans); + bch_journal_res_put(&c->journal, &trans->journal_res); + + if (split) + goto split; + if (ret) + goto err; + + /* + * hack: iterators are inconsistent when they hit end of leaf, until + * traversed again + */ + trans_for_each_entry(trans, i) + if (i->iter->at_end_of_leaf) + goto out; + + trans_for_each_entry(trans, i) + if (!same_leaf_as_prev(trans, i)) { + foreground_maybe_merge(i->iter, btree_prev_sib); + foreground_maybe_merge(i->iter, btree_next_sib); + } +out: + /* make sure we didn't lose an error: */ + if (!ret && IS_ENABLED(CONFIG_BCACHE_DEBUG)) + trans_for_each_entry(trans, i) + BUG_ON(!i->done); + + percpu_ref_put(&c->writes); + return ret; +split: + /* + * have to drop journal res before splitting, because splitting means + * allocating new btree nodes, and holding a journal reservation + * potentially blocks the allocator: + */ + ret = bch_btree_split_leaf(split, trans->flags); + if (ret) + goto err; + /* + * if the split didn't have to drop locks the insert will still be + * atomic (in the BTREE_INSERT_ATOMIC sense, what the caller peeked() + * and is overwriting won't have changed) + */ + goto retry_locks; +err: + if (cycle_gc_lock) { + down_read(&c->gc_lock); + up_read(&c->gc_lock); + } + + if (ret == -EINTR) { + trans_for_each_entry(trans, i) { + int ret2 = bch_btree_iter_traverse(i->iter); + if (ret2) { + ret = ret2; + goto out; + } + } + + /* + * BTREE_ITER_ATOMIC means we have to return -EINTR if we + * dropped locks: + */ + if (!(trans->flags & BTREE_INSERT_ATOMIC)) + goto retry; + } + + goto out; +} + +int bch_btree_insert_list_at(struct btree_iter *iter, + struct keylist *keys, + struct disk_reservation *disk_res, + struct extent_insert_hook *hook, + u64 *journal_seq, unsigned flags) +{ + BUG_ON(flags & BTREE_INSERT_ATOMIC); + BUG_ON(bch_keylist_empty(keys)); + verify_keys_sorted(keys); + + while (!bch_keylist_empty(keys)) { + /* need to traverse between each insert */ + int ret = bch_btree_iter_traverse(iter); + if (ret) + return ret; + + ret = bch_btree_insert_at(iter->c, disk_res, hook, + journal_seq, flags, + BTREE_INSERT_ENTRY(iter, bch_keylist_front(keys))); + if (ret) + return ret; + + bch_keylist_pop_front(keys); + } + + return 0; +} + +/** + * bch_btree_insert_check_key - insert dummy key into btree + * + * We insert a random key on a cache miss, then compare exchange on it + * once the cache promotion or backing device read completes. This + * ensures that if this key is written to after the read, the read will + * lose and not overwrite the key with stale data. + * + * Return values: + * -EAGAIN: @iter->cl was put on a waitlist waiting for btree node allocation + * -EINTR: btree node was changed while upgrading to write lock + */ +int bch_btree_insert_check_key(struct btree_iter *iter, + struct bkey_i *check_key) +{ + struct bpos saved_pos = iter->pos; + struct bkey_i_cookie *cookie; + BKEY_PADDED(key) tmp; + int ret; + + BUG_ON(bkey_cmp(iter->pos, bkey_start_pos(&check_key->k))); + + check_key->k.type = KEY_TYPE_COOKIE; + set_bkey_val_bytes(&check_key->k, sizeof(struct bch_cookie)); + + cookie = bkey_i_to_cookie(check_key); + get_random_bytes(&cookie->v, sizeof(cookie->v)); + + bkey_copy(&tmp.key, check_key); + + ret = bch_btree_insert_at(iter->c, NULL, NULL, NULL, + BTREE_INSERT_ATOMIC, + BTREE_INSERT_ENTRY(iter, &tmp.key)); + + bch_btree_iter_rewind(iter, saved_pos); + + return ret; +} + +/** + * bch_btree_insert - insert keys into the extent btree + * @c: pointer to struct cache_set + * @id: btree to insert into + * @insert_keys: list of keys to insert + * @hook: insert callback + */ +int bch_btree_insert(struct cache_set *c, enum btree_id id, + struct bkey_i *k, + struct disk_reservation *disk_res, + struct extent_insert_hook *hook, + u64 *journal_seq, int flags) +{ + struct btree_iter iter; + int ret, ret2; + + bch_btree_iter_init_intent(&iter, c, id, bkey_start_pos(&k->k)); + + ret = bch_btree_iter_traverse(&iter); + if (unlikely(ret)) + goto out; + + ret = bch_btree_insert_at(c, disk_res, hook, journal_seq, flags, + BTREE_INSERT_ENTRY(&iter, k)); +out: ret2 = bch_btree_iter_unlock(&iter); + + return ret ?: ret2; +} + +/** + * bch_btree_update - like bch_btree_insert(), but asserts that we're + * overwriting an existing key + */ +int bch_btree_update(struct cache_set *c, enum btree_id id, + struct bkey_i *k, u64 *journal_seq) +{ + struct btree_iter iter; + struct bkey_s_c u; + int ret; + + EBUG_ON(id == BTREE_ID_EXTENTS); + + bch_btree_iter_init_intent(&iter, c, id, k->k.p); + + u = bch_btree_iter_peek_with_holes(&iter); + ret = btree_iter_err(u); + if (ret) + return ret; + + if (bkey_deleted(u.k)) { + bch_btree_iter_unlock(&iter); + return -ENOENT; + } + + ret = bch_btree_insert_at(c, NULL, NULL, journal_seq, 0, + BTREE_INSERT_ENTRY(&iter, k)); + bch_btree_iter_unlock(&iter); + return ret; +} + +/* + * bch_btree_delete_range - delete everything within a given range + * + * Range is a half open interval - [start, end) + */ +int bch_btree_delete_range(struct cache_set *c, enum btree_id id, + struct bpos start, + struct bpos end, + u64 version, + struct disk_reservation *disk_res, + struct extent_insert_hook *hook, + u64 *journal_seq) +{ + struct btree_iter iter; + struct bkey_s_c k; + int ret = 0; + + bch_btree_iter_init_intent(&iter, c, id, start); + + while ((k = bch_btree_iter_peek(&iter)).k && + !(ret = btree_iter_err(k))) { + unsigned max_sectors = KEY_SIZE_MAX & (~0 << c->block_bits); + /* really shouldn't be using a bare, unpadded bkey_i */ + struct bkey_i delete; + + if (bkey_cmp(iter.pos, end) >= 0) + break; + + bkey_init(&delete.k); + + /* + * For extents, iter.pos won't necessarily be the same as + * bkey_start_pos(k.k) (for non extents they always will be the + * same). It's important that we delete starting from iter.pos + * because the range we want to delete could start in the middle + * of k. + * + * (bch_btree_iter_peek() does guarantee that iter.pos >= + * bkey_start_pos(k.k)). + */ + delete.k.p = iter.pos; + delete.k.version = version; + + if (iter.is_extents) { + /* + * The extents btree is special - KEY_TYPE_DISCARD is + * used for deletions, not KEY_TYPE_DELETED. This is an + * internal implementation detail that probably + * shouldn't be exposed (internally, KEY_TYPE_DELETED is + * used as a proxy for k->size == 0): + */ + delete.k.type = KEY_TYPE_DISCARD; + + /* create the biggest key we can */ + bch_key_resize(&delete.k, max_sectors); + bch_cut_back(end, &delete.k); + } + + ret = bch_btree_insert_at(c, disk_res, hook, journal_seq, + BTREE_INSERT_NOFAIL, + BTREE_INSERT_ENTRY(&iter, &delete)); + if (ret) + break; + + bch_btree_iter_cond_resched(&iter); + } + + bch_btree_iter_unlock(&iter); + return ret; +} + +/** + * bch_btree_node_rewrite - Rewrite/move a btree node + * + * Returns 0 on success, -EINTR or -EAGAIN on failure (i.e. + * btree_check_reserve() has to wait) + */ +int bch_btree_node_rewrite(struct btree_iter *iter, struct btree *b, + struct closure *cl) +{ + struct cache_set *c = iter->c; + struct btree *n, *parent = iter->nodes[b->level + 1]; + struct btree_reserve *reserve; + struct btree_interior_update *as; + unsigned flags = BTREE_INSERT_NOFAIL; + + /* + * if caller is going to wait if allocating reserve fails, then this is + * a rewrite that must succeed: + */ + if (cl) + flags |= BTREE_INSERT_USE_RESERVE; + + if (!bch_btree_iter_set_locks_want(iter, U8_MAX)) + return -EINTR; + + reserve = bch_btree_reserve_get(c, b, 0, flags, cl); + if (IS_ERR(reserve)) { + trace_bcache_btree_gc_rewrite_node_fail(c, b); + return PTR_ERR(reserve); + } + + as = bch_btree_interior_update_alloc(c); + + bch_btree_interior_update_will_free_node(c, as, b); + + n = btree_node_alloc_replacement(c, b, reserve); + + bch_btree_build_aux_trees(n); + six_unlock_write(&n->lock); + + trace_bcache_btree_gc_rewrite_node(c, b); + + bch_btree_node_write(c, n, &as->cl, SIX_LOCK_intent, -1); + + if (parent) { + bch_btree_insert_node(parent, iter, + &keylist_single(&n->key), + reserve, as); + } else { + bch_btree_set_root(iter, n, as, reserve); + } + + btree_open_bucket_put(c, n); + + bch_btree_node_free_inmem(iter, b); + + BUG_ON(!bch_btree_iter_node_replace(iter, n)); + + bch_btree_reserve_put(c, reserve); + return 0; +} |