btree.c

   1// SPDX-License-Identifier: GPL-2.0
   2/*
   3 * Copyright (C) 2010 Kent Overstreet <kent.overstreet@gmail.com>
   4 *
   5 * Uses a block device as cache for other block devices; optimized for SSDs.
   6 * All allocation is done in buckets, which should match the erase block size
   7 * of the device.
   8 *
   9 * Buckets containing cached data are kept on a heap sorted by priority;
  10 * bucket priority is increased on cache hit, and periodically all the buckets
  11 * on the heap have their priority scaled down. This currently is just used as
  12 * an LRU but in the future should allow for more intelligent heuristics.
  13 *
  14 * Buckets have an 8 bit counter; freeing is accomplished by incrementing the
  15 * counter. Garbage collection is used to remove stale pointers.
  16 *
  17 * Indexing is done via a btree; nodes are not necessarily fully sorted, rather
  18 * as keys are inserted we only sort the pages that have not yet been written.
  19 * When garbage collection is run, we resort the entire node.
  20 *
  21 * All configuration is done via sysfs; see Documentation/admin-guide/bcache.rst.
  22 */
  23
  24#include "bcache.h"
  25#include "btree.h"
  26#include "debug.h"
  27#include "extents.h"
  28
  29#include <linux/slab.h>
  30#include <linux/bitops.h>
  31#include <linux/hash.h>
  32#include <linux/kthread.h>
  33#include <linux/prefetch.h>
  34#include <linux/random.h>
  35#include <linux/rcupdate.h>
  36#include <linux/sched/clock.h>
  37#include <linux/rculist.h>
  38#include <linux/delay.h>
  39#include <linux/sort.h>
  40#include <trace/events/bcache.h>
  41
  42/*
  43 * Todo:
  44 * register_bcache: Return errors out to userspace correctly
  45 *
  46 * Writeback: don't undirty key until after a cache flush
  47 *
  48 * Create an iterator for key pointers
  49 *
  50 * On btree write error, mark bucket such that it won't be freed from the cache
  51 *
  52 * Journalling:
  53 *   Check for bad keys in replay
  54 *   Propagate barriers
  55 *   Refcount journal entries in journal_replay
  56 *
  57 * Garbage collection:
  58 *   Finish incremental gc
  59 *   Gc should free old UUIDs, data for invalid UUIDs
  60 *
  61 * Provide a way to list backing device UUIDs we have data cached for, and
  62 * probably how long it's been since we've seen them, and a way to invalidate
  63 * dirty data for devices that will never be attached again
  64 *
  65 * Keep 1 min/5 min/15 min statistics of how busy a block device has been, so
  66 * that based on that and how much dirty data we have we can keep writeback
  67 * from being starved
  68 *
  69 * Add a tracepoint or somesuch to watch for writeback starvation
  70 *
  71 * When btree depth > 1 and splitting an interior node, we have to make sure
  72 * alloc_bucket() cannot fail. This should be true but is not completely
  73 * obvious.
  74 *
  75 * Plugging?
  76 *
  77 * If data write is less than hard sector size of ssd, round up offset in open
  78 * bucket to the next whole sector
  79 *
  80 * Superblock needs to be fleshed out for multiple cache devices
  81 *
  82 * Add a sysfs tunable for the number of writeback IOs in flight
  83 *
  84 * Add a sysfs tunable for the number of open data buckets
  85 *
  86 * IO tracking: Can we track when one process is doing io on behalf of another?
  87 * IO tracking: Don't use just an average, weigh more recent stuff higher
  88 *
  89 * Test module load/unload
  90 */
  91
  92#define MAX_GC_TIMES_SHIFT	7  /* 128 loops */
  93#define GC_NODES_MIN		10
  94#define GC_SLEEP_MS_MIN		10
  95#define GC_SLEEP_MS		100
  96
  97#define PTR_DIRTY_BIT		(((uint64_t) 1 << 36))
  98
  99#define PTR_HASH(c, k)							\
 100	(((k)->ptr[0] >> c->bucket_bits) | PTR_GEN(k, 0))
 101
 102static struct workqueue_struct *btree_io_wq;
 103
 104#define insert_lock(s, b)	((b)->level <= (s)->lock)
 105
 106
 107static inline struct bset *write_block(struct btree *b)
 108{
 109	return ((void *) btree_bset_first(b)) + b->written * block_bytes(b->c->cache);
 110}
 111
 112static void bch_btree_init_next(struct btree *b)
 113{
 114	/* If not a leaf node, always sort */
 115	if (b->level && b->keys.nsets)
 116		bch_btree_sort(&b->keys, &b->c->sort);
 117	else
 118		bch_btree_sort_lazy(&b->keys, &b->c->sort);
 119
 120	if (b->written < btree_blocks(b))
 121		bch_bset_init_next(&b->keys, write_block(b),
 122				   bset_magic(&b->c->cache->sb));
 123
 124}
 125
 126/* Btree key manipulation */
 127
 128void bkey_put(struct cache_set *c, struct bkey *k)
 129{
 130	unsigned int i;
 131
 132	for (i = 0; i < KEY_PTRS(k); i++)
 133		if (ptr_available(c, k, i))
 134			atomic_dec_bug(&PTR_BUCKET(c, k, i)->pin);
 135}
 136
 137/* Btree IO */
 138
 139static uint64_t btree_csum_set(struct btree *b, struct bset *i)
 140{
 141	uint64_t crc = b->key.ptr[0];
 142	void *data = (void *) i + 8, *end = bset_bkey_last(i);
 143
 144	crc = crc64_be(crc, data, end - data);
 145	return crc ^ 0xffffffffffffffffULL;
 146}
 147
 148void bch_btree_node_read_done(struct btree *b)
 149{
 150	const char *err = "bad btree header";
 151	struct bset *i = btree_bset_first(b);
 152	struct btree_iter *iter;
 153
 154	/*
 155	 * c->fill_iter can allocate an iterator with more memory space
 156	 * than static MAX_BSETS.
 157	 * See the comment arount cache_set->fill_iter.
 158	 */
 159	iter = mempool_alloc(&b->c->fill_iter, GFP_NOIO);
 160	iter->size = b->c->cache->sb.bucket_size / b->c->cache->sb.block_size;
 161	iter->used = 0;
 162
 163#ifdef CONFIG_BCACHE_DEBUG
 164	iter->b = &b->keys;
 165#endif
 166
 167	if (!i->seq)
 168		goto err;
 169
 170	for (;
 171	     b->written < btree_blocks(b) && i->seq == b->keys.set[0].data->seq;
 172	     i = write_block(b)) {
 173		err = "unsupported bset version";
 174		if (i->version > BCACHE_BSET_VERSION)
 175			goto err;
 176
 177		err = "bad btree header";
 178		if (b->written + set_blocks(i, block_bytes(b->c->cache)) >
 179		    btree_blocks(b))
 180			goto err;
 181
 182		err = "bad magic";
 183		if (i->magic != bset_magic(&b->c->cache->sb))
 184			goto err;
 185
 186		err = "bad checksum";
 187		switch (i->version) {
 188		case 0:
 189			if (i->csum != csum_set(i))
 190				goto err;
 191			break;
 192		case BCACHE_BSET_VERSION:
 193			if (i->csum != btree_csum_set(b, i))
 194				goto err;
 195			break;
 196		}
 197
 198		err = "empty set";
 199		if (i != b->keys.set[0].data && !i->keys)
 200			goto err;
 201
 202		bch_btree_iter_push(iter, i->start, bset_bkey_last(i));
 203
 204		b->written += set_blocks(i, block_bytes(b->c->cache));
 205	}
 206
 207	err = "corrupted btree";
 208	for (i = write_block(b);
 209	     bset_sector_offset(&b->keys, i) < KEY_SIZE(&b->key);
 210	     i = ((void *) i) + block_bytes(b->c->cache))
 211		if (i->seq == b->keys.set[0].data->seq)
 212			goto err;
 213
 214	bch_btree_sort_and_fix_extents(&b->keys, iter, &b->c->sort);
 215
 216	i = b->keys.set[0].data;
 217	err = "short btree key";
 218	if (b->keys.set[0].size &&
 219	    bkey_cmp(&b->key, &b->keys.set[0].end) < 0)
 220		goto err;
 221
 222	if (b->written < btree_blocks(b))
 223		bch_bset_init_next(&b->keys, write_block(b),
 224				   bset_magic(&b->c->cache->sb));
 225out:
 226	mempool_free(iter, &b->c->fill_iter);
 227	return;
 228err:
 229	set_btree_node_io_error(b);
 230	bch_cache_set_error(b->c, "%s at bucket %zu, block %u, %u keys",
 231			    err, PTR_BUCKET_NR(b->c, &b->key, 0),
 232			    bset_block_offset(b, i), i->keys);
 233	goto out;
 234}
 235
 236static void btree_node_read_endio(struct bio *bio)
 237{
 238	struct closure *cl = bio->bi_private;
 239
 240	closure_put(cl);
 241}
 242
 243static void bch_btree_node_read(struct btree *b)
 244{
 245	uint64_t start_time = local_clock();
 246	struct closure cl;
 247	struct bio *bio;
 248
 249	trace_bcache_btree_read(b);
 250
 251	closure_init_stack(&cl);
 252
 253	bio = bch_bbio_alloc(b->c);
 254	bio->bi_iter.bi_size = KEY_SIZE(&b->key) << 9;
 255	bio->bi_end_io	= btree_node_read_endio;
 256	bio->bi_private	= &cl;
 257	bio->bi_opf = REQ_OP_READ | REQ_META;
 258
 259	bch_bio_map(bio, b->keys.set[0].data);
 260
 261	bch_submit_bbio(bio, b->c, &b->key, 0);
 262	closure_sync(&cl);
 263
 264	if (bio->bi_status)
 265		set_btree_node_io_error(b);
 266
 267	bch_bbio_free(bio, b->c);
 268
 269	if (btree_node_io_error(b))
 270		goto err;
 271
 272	bch_btree_node_read_done(b);
 273	bch_time_stats_update(&b->c->btree_read_time, start_time);
 274
 275	return;
 276err:
 277	bch_cache_set_error(b->c, "io error reading bucket %zu",
 278			    PTR_BUCKET_NR(b->c, &b->key, 0));
 279}
 280
 281static void btree_complete_write(struct btree *b, struct btree_write *w)
 282{
 283	if (w->prio_blocked &&
 284	    !atomic_sub_return(w->prio_blocked, &b->c->prio_blocked))
 285		wake_up_allocators(b->c);
 286
 287	if (w->journal) {
 288		atomic_dec_bug(w->journal);
 289		__closure_wake_up(&b->c->journal.wait);
 290	}
 291
 292	w->prio_blocked	= 0;
 293	w->journal	= NULL;
 294}
 295
 296static CLOSURE_CALLBACK(btree_node_write_unlock)
 297{
 298	closure_type(b, struct btree, io);
 299
 300	up(&b->io_mutex);
 301}
 302
 303static CLOSURE_CALLBACK(__btree_node_write_done)
 304{
 305	closure_type(b, struct btree, io);
 306	struct btree_write *w = btree_prev_write(b);
 307
 308	bch_bbio_free(b->bio, b->c);
 309	b->bio = NULL;
 310	btree_complete_write(b, w);
 311
 312	if (btree_node_dirty(b))
 313		queue_delayed_work(btree_io_wq, &b->work, 30 * HZ);
 314
 315	closure_return_with_destructor(cl, btree_node_write_unlock);
 316}
 317
 318static CLOSURE_CALLBACK(btree_node_write_done)
 319{
 320	closure_type(b, struct btree, io);
 321
 322	bio_free_pages(b->bio);
 323	__btree_node_write_done(&cl->work);
 324}
 325
 326static void btree_node_write_endio(struct bio *bio)
 327{
 328	struct closure *cl = bio->bi_private;
 329	struct btree *b = container_of(cl, struct btree, io);
 330
 331	if (bio->bi_status)
 332		set_btree_node_io_error(b);
 333
 334	bch_bbio_count_io_errors(b->c, bio, bio->bi_status, "writing btree");
 335	closure_put(cl);
 336}
 337
 338static void do_btree_node_write(struct btree *b)
 339{
 340	struct closure *cl = &b->io;
 341	struct bset *i = btree_bset_last(b);
 342	BKEY_PADDED(key) k;
 343
 344	i->version	= BCACHE_BSET_VERSION;
 345	i->csum		= btree_csum_set(b, i);
 346
 347	BUG_ON(b->bio);
 348	b->bio = bch_bbio_alloc(b->c);
 349
 350	b->bio->bi_end_io	= btree_node_write_endio;
 351	b->bio->bi_private	= cl;
 352	b->bio->bi_iter.bi_size	= roundup(set_bytes(i), block_bytes(b->c->cache));
 353	b->bio->bi_opf		= REQ_OP_WRITE | REQ_META | REQ_FUA;
 354	bch_bio_map(b->bio, i);
 355
 356	/*
 357	 * If we're appending to a leaf node, we don't technically need FUA -
 358	 * this write just needs to be persisted before the next journal write,
 359	 * which will be marked FLUSH|FUA.
 360	 *
 361	 * Similarly if we're writing a new btree root - the pointer is going to
 362	 * be in the next journal entry.
 363	 *
 364	 * But if we're writing a new btree node (that isn't a root) or
 365	 * appending to a non leaf btree node, we need either FUA or a flush
 366	 * when we write the parent with the new pointer. FUA is cheaper than a
 367	 * flush, and writes appending to leaf nodes aren't blocking anything so
 368	 * just make all btree node writes FUA to keep things sane.
 369	 */
 370
 371	bkey_copy(&k.key, &b->key);
 372	SET_PTR_OFFSET(&k.key, 0, PTR_OFFSET(&k.key, 0) +
 373		       bset_sector_offset(&b->keys, i));
 374
 375	if (!bch_bio_alloc_pages(b->bio, GFP_NOWAIT)) {
 376		struct bio_vec *bv;
 377		void *addr = (void *) ((unsigned long) i & ~(PAGE_SIZE - 1));
 378		struct bvec_iter_all iter_all;
 379
 380		bio_for_each_segment_all(bv, b->bio, iter_all) {
 381			memcpy(page_address(bv->bv_page), addr, PAGE_SIZE);
 382			addr += PAGE_SIZE;
 383		}
 384
 385		bch_submit_bbio(b->bio, b->c, &k.key, 0);
 386
 387		continue_at(cl, btree_node_write_done, NULL);
 388	} else {
 389		/*
 390		 * No problem for multipage bvec since the bio is
 391		 * just allocated
 392		 */
 393		b->bio->bi_vcnt = 0;
 394		bch_bio_map(b->bio, i);
 395
 396		bch_submit_bbio(b->bio, b->c, &k.key, 0);
 397
 398		closure_sync(cl);
 399		continue_at_nobarrier(cl, __btree_node_write_done, NULL);
 400	}
 401}
 402
 403void __bch_btree_node_write(struct btree *b, struct closure *parent)
 404{
 405	struct bset *i = btree_bset_last(b);
 406
 407	lockdep_assert_held(&b->write_lock);
 408
 409	trace_bcache_btree_write(b);
 410
 411	BUG_ON(current->bio_list);
 412	BUG_ON(b->written >= btree_blocks(b));
 413	BUG_ON(b->written && !i->keys);
 414	BUG_ON(btree_bset_first(b)->seq != i->seq);
 415	bch_check_keys(&b->keys, "writing");
 416
 417	cancel_delayed_work(&b->work);
 418
 419	/* If caller isn't waiting for write, parent refcount is cache set */
 420	down(&b->io_mutex);
 421	closure_init(&b->io, parent ?: &b->c->cl);
 422
 423	clear_bit(BTREE_NODE_dirty,	 &b->flags);
 424	change_bit(BTREE_NODE_write_idx, &b->flags);
 425
 426	do_btree_node_write(b);
 427
 428	atomic_long_add(set_blocks(i, block_bytes(b->c->cache)) * b->c->cache->sb.block_size,
 429			&b->c->cache->btree_sectors_written);
 430
 431	b->written += set_blocks(i, block_bytes(b->c->cache));
 432}
 433
 434void bch_btree_node_write(struct btree *b, struct closure *parent)
 435{
 436	unsigned int nsets = b->keys.nsets;
 437
 438	lockdep_assert_held(&b->lock);
 439
 440	__bch_btree_node_write(b, parent);
 441
 442	/*
 443	 * do verify if there was more than one set initially (i.e. we did a
 444	 * sort) and we sorted down to a single set:
 445	 */
 446	if (nsets && !b->keys.nsets)
 447		bch_btree_verify(b);
 448
 449	bch_btree_init_next(b);
 450}
 451
 452static void bch_btree_node_write_sync(struct btree *b)
 453{
 454	struct closure cl;
 455
 456	closure_init_stack(&cl);
 457
 458	mutex_lock(&b->write_lock);
 459	bch_btree_node_write(b, &cl);
 460	mutex_unlock(&b->write_lock);
 461
 462	closure_sync(&cl);
 463}
 464
 465static void btree_node_write_work(struct work_struct *w)
 466{
 467	struct btree *b = container_of(to_delayed_work(w), struct btree, work);
 468
 469	mutex_lock(&b->write_lock);
 470	if (btree_node_dirty(b))
 471		__bch_btree_node_write(b, NULL);
 472	mutex_unlock(&b->write_lock);
 473}
 474
 475static void bch_btree_leaf_dirty(struct btree *b, atomic_t *journal_ref)
 476{
 477	struct bset *i = btree_bset_last(b);
 478	struct btree_write *w = btree_current_write(b);
 479
 480	lockdep_assert_held(&b->write_lock);
 481
 482	BUG_ON(!b->written);
 483	BUG_ON(!i->keys);
 484
 485	if (!btree_node_dirty(b))
 486		queue_delayed_work(btree_io_wq, &b->work, 30 * HZ);
 487
 488	set_btree_node_dirty(b);
 489
 490	/*
 491	 * w->journal is always the oldest journal pin of all bkeys
 492	 * in the leaf node, to make sure the oldest jset seq won't
 493	 * be increased before this btree node is flushed.
 494	 */
 495	if (journal_ref) {
 496		if (w->journal &&
 497		    journal_pin_cmp(b->c, w->journal, journal_ref)) {
 498			atomic_dec_bug(w->journal);
 499			w->journal = NULL;
 500		}
 501
 502		if (!w->journal) {
 503			w->journal = journal_ref;
 504			atomic_inc(w->journal);
 505		}
 506	}
 507
 508	/* Force write if set is too big */
 509	if (set_bytes(i) > PAGE_SIZE - 48 &&
 510	    !current->bio_list)
 511		bch_btree_node_write(b, NULL);
 512}
 513
 514/*
 515 * Btree in memory cache - allocation/freeing
 516 * mca -> memory cache
 517 */
 518
 519#define mca_reserve(c)	(((!IS_ERR_OR_NULL(c->root) && c->root->level) \
 520			  ? c->root->level : 1) * 8 + 16)
 521#define mca_can_free(c)						\
 522	max_t(int, 0, c->btree_cache_used - mca_reserve(c))
 523
 524static void mca_data_free(struct btree *b)
 525{
 526	BUG_ON(b->io_mutex.count != 1);
 527
 528	bch_btree_keys_free(&b->keys);
 529
 530	b->c->btree_cache_used--;
 531	list_move(&b->list, &b->c->btree_cache_freed);
 532}
 533
 534static void mca_bucket_free(struct btree *b)
 535{
 536	BUG_ON(btree_node_dirty(b));
 537
 538	b->key.ptr[0] = 0;
 539	hlist_del_init_rcu(&b->hash);
 540	list_move(&b->list, &b->c->btree_cache_freeable);
 541}
 542
 543static unsigned int btree_order(struct bkey *k)
 544{
 545	return ilog2(KEY_SIZE(k) / PAGE_SECTORS ?: 1);
 546}
 547
 548static void mca_data_alloc(struct btree *b, struct bkey *k, gfp_t gfp)
 549{
 550	if (!bch_btree_keys_alloc(&b->keys,
 551				  max_t(unsigned int,
 552					ilog2(b->c->btree_pages),
 553					btree_order(k)),
 554				  gfp)) {
 555		b->c->btree_cache_used++;
 556		list_move(&b->list, &b->c->btree_cache);
 557	} else {
 558		list_move(&b->list, &b->c->btree_cache_freed);
 559	}
 560}
 561
 562#ifdef CONFIG_PROVE_LOCKING
 563static int btree_lock_cmp_fn(const struct lockdep_map *_a,
 564			     const struct lockdep_map *_b)
 565{
 566	const struct btree *a = container_of(_a, struct btree, lock.dep_map);
 567	const struct btree *b = container_of(_b, struct btree, lock.dep_map);
 568
 569	return -cmp_int(a->level, b->level) ?: bkey_cmp(&a->key, &b->key);
 570}
 571
 572static void btree_lock_print_fn(const struct lockdep_map *map)
 573{
 574	const struct btree *b = container_of(map, struct btree, lock.dep_map);
 575
 576	printk(KERN_CONT " l=%u %llu:%llu", b->level,
 577	       KEY_INODE(&b->key), KEY_OFFSET(&b->key));
 578}
 579#endif
 580
 581static struct btree *mca_bucket_alloc(struct cache_set *c,
 582				      struct bkey *k, gfp_t gfp)
 583{
 584	/*
 585	 * kzalloc() is necessary here for initialization,
 586	 * see code comments in bch_btree_keys_init().
 587	 */
 588	struct btree *b = kzalloc(sizeof(struct btree), gfp);
 589
 590	if (!b)
 591		return NULL;
 592
 593	init_rwsem(&b->lock);
 594	lock_set_cmp_fn(&b->lock, btree_lock_cmp_fn, btree_lock_print_fn);
 595	mutex_init(&b->write_lock);
 596	lockdep_set_novalidate_class(&b->write_lock);
 597	INIT_LIST_HEAD(&b->list);
 598	INIT_DELAYED_WORK(&b->work, btree_node_write_work);
 599	b->c = c;
 600	sema_init(&b->io_mutex, 1);
 601
 602	mca_data_alloc(b, k, gfp);
 603	return b;
 604}
 605
 606static int mca_reap(struct btree *b, unsigned int min_order, bool flush)
 607{
 608	struct closure cl;
 609
 610	closure_init_stack(&cl);
 611	lockdep_assert_held(&b->c->bucket_lock);
 612
 613	if (!down_write_trylock(&b->lock))
 614		return -ENOMEM;
 615
 616	BUG_ON(btree_node_dirty(b) && !b->keys.set[0].data);
 617
 618	if (b->keys.page_order < min_order)
 619		goto out_unlock;
 620
 621	if (!flush) {
 622		if (btree_node_dirty(b))
 623			goto out_unlock;
 624
 625		if (down_trylock(&b->io_mutex))
 626			goto out_unlock;
 627		up(&b->io_mutex);
 628	}
 629
 630retry:
 631	/*
 632	 * BTREE_NODE_dirty might be cleared in btree_flush_btree() by
 633	 * __bch_btree_node_write(). To avoid an extra flush, acquire
 634	 * b->write_lock before checking BTREE_NODE_dirty bit.
 635	 */
 636	mutex_lock(&b->write_lock);
 637	/*
 638	 * If this btree node is selected in btree_flush_write() by journal
 639	 * code, delay and retry until the node is flushed by journal code
 640	 * and BTREE_NODE_journal_flush bit cleared by btree_flush_write().
 641	 */
 642	if (btree_node_journal_flush(b)) {
 643		pr_debug("bnode %p is flushing by journal, retry\n", b);
 644		mutex_unlock(&b->write_lock);
 645		udelay(1);
 646		goto retry;
 647	}
 648
 649	if (btree_node_dirty(b))
 650		__bch_btree_node_write(b, &cl);
 651	mutex_unlock(&b->write_lock);
 652
 653	closure_sync(&cl);
 654
 655	/* wait for any in flight btree write */
 656	down(&b->io_mutex);
 657	up(&b->io_mutex);
 658
 659	return 0;
 660out_unlock:
 661	rw_unlock(true, b);
 662	return -ENOMEM;
 663}
 664
 665static unsigned long bch_mca_scan(struct shrinker *shrink,
 666				  struct shrink_control *sc)
 667{
 668	struct cache_set *c = shrink->private_data;
 669	struct btree *b, *t;
 670	unsigned long i, nr = sc->nr_to_scan;
 671	unsigned long freed = 0;
 672	unsigned int btree_cache_used;
 673
 674	if (c->shrinker_disabled)
 675		return SHRINK_STOP;
 676
 677	if (c->btree_cache_alloc_lock)
 678		return SHRINK_STOP;
 679
 680	/* Return -1 if we can't do anything right now */
 681	if (sc->gfp_mask & __GFP_IO)
 682		mutex_lock(&c->bucket_lock);
 683	else if (!mutex_trylock(&c->bucket_lock))
 684		return -1;
 685
 686	/*
 687	 * It's _really_ critical that we don't free too many btree nodes - we
 688	 * have to always leave ourselves a reserve. The reserve is how we
 689	 * guarantee that allocating memory for a new btree node can always
 690	 * succeed, so that inserting keys into the btree can always succeed and
 691	 * IO can always make forward progress:
 692	 */
 693	nr /= c->btree_pages;
 694	if (nr == 0)
 695		nr = 1;
 696	nr = min_t(unsigned long, nr, mca_can_free(c));
 697
 698	i = 0;
 699	btree_cache_used = c->btree_cache_used;
 700	list_for_each_entry_safe_reverse(b, t, &c->btree_cache_freeable, list) {
 701		if (nr <= 0)
 702			goto out;
 703
 704		if (!mca_reap(b, 0, false)) {
 705			mca_data_free(b);
 706			rw_unlock(true, b);
 707			freed++;
 708		}
 709		nr--;
 710		i++;
 711	}
 712
 713	list_for_each_entry_safe_reverse(b, t, &c->btree_cache, list) {
 714		if (nr <= 0 || i >= btree_cache_used)
 715			goto out;
 716
 717		if (!mca_reap(b, 0, false)) {
 718			mca_bucket_free(b);
 719			mca_data_free(b);
 720			rw_unlock(true, b);
 721			freed++;
 722		}
 723
 724		nr--;
 725		i++;
 726	}
 727out:
 728	mutex_unlock(&c->bucket_lock);
 729	return freed * c->btree_pages;
 730}
 731
 732static unsigned long bch_mca_count(struct shrinker *shrink,
 733				   struct shrink_control *sc)
 734{
 735	struct cache_set *c = shrink->private_data;
 736
 737	if (c->shrinker_disabled)
 738		return 0;
 739
 740	if (c->btree_cache_alloc_lock)
 741		return 0;
 742
 743	return mca_can_free(c) * c->btree_pages;
 744}
 745
 746void bch_btree_cache_free(struct cache_set *c)
 747{
 748	struct btree *b;
 749	struct closure cl;
 750
 751	closure_init_stack(&cl);
 752
 753	if (c->shrink)
 754		shrinker_free(c->shrink);
 755
 756	mutex_lock(&c->bucket_lock);
 757
 758#ifdef CONFIG_BCACHE_DEBUG
 759	if (c->verify_data)
 760		list_move(&c->verify_data->list, &c->btree_cache);
 761
 762	free_pages((unsigned long) c->verify_ondisk, ilog2(meta_bucket_pages(&c->cache->sb)));
 763#endif
 764
 765	list_splice(&c->btree_cache_freeable,
 766		    &c->btree_cache);
 767
 768	while (!list_empty(&c->btree_cache)) {
 769		b = list_first_entry(&c->btree_cache, struct btree, list);
 770
 771		/*
 772		 * This function is called by cache_set_free(), no I/O
 773		 * request on cache now, it is unnecessary to acquire
 774		 * b->write_lock before clearing BTREE_NODE_dirty anymore.
 775		 */
 776		if (btree_node_dirty(b)) {
 777			btree_complete_write(b, btree_current_write(b));
 778			clear_bit(BTREE_NODE_dirty, &b->flags);
 779		}
 780		mca_data_free(b);
 781	}
 782
 783	while (!list_empty(&c->btree_cache_freed)) {
 784		b = list_first_entry(&c->btree_cache_freed,
 785				     struct btree, list);
 786		list_del(&b->list);
 787		cancel_delayed_work_sync(&b->work);
 788		kfree(b);
 789	}
 790
 791	mutex_unlock(&c->bucket_lock);
 792}
 793
 794int bch_btree_cache_alloc(struct cache_set *c)
 795{
 796	unsigned int i;
 797
 798	for (i = 0; i < mca_reserve(c); i++)
 799		if (!mca_bucket_alloc(c, &ZERO_KEY, GFP_KERNEL))
 800			return -ENOMEM;
 801
 802	list_splice_init(&c->btree_cache,
 803			 &c->btree_cache_freeable);
 804
 805#ifdef CONFIG_BCACHE_DEBUG
 806	mutex_init(&c->verify_lock);
 807
 808	c->verify_ondisk = (void *)
 809		__get_free_pages(GFP_KERNEL|__GFP_COMP,
 810				 ilog2(meta_bucket_pages(&c->cache->sb)));
 811	if (!c->verify_ondisk) {
 812		/*
 813		 * Don't worry about the mca_rereserve buckets
 814		 * allocated in previous for-loop, they will be
 815		 * handled properly in bch_cache_set_unregister().
 816		 */
 817		return -ENOMEM;
 818	}
 819
 820	c->verify_data = mca_bucket_alloc(c, &ZERO_KEY, GFP_KERNEL);
 821
 822	if (c->verify_data &&
 823	    c->verify_data->keys.set->data)
 824		list_del_init(&c->verify_data->list);
 825	else
 826		c->verify_data = NULL;
 827#endif
 828
 829	c->shrink = shrinker_alloc(0, "md-bcache:%pU", c->set_uuid);
 830	if (!c->shrink) {
 831		pr_warn("bcache: %s: could not allocate shrinker\n", __func__);
 832		return 0;
 833	}
 834
 835	c->shrink->count_objects = bch_mca_count;
 836	c->shrink->scan_objects = bch_mca_scan;
 837	c->shrink->seeks = 4;
 838	c->shrink->batch = c->btree_pages * 2;
 839	c->shrink->private_data = c;
 840
 841	shrinker_register(c->shrink);
 842
 843	return 0;
 844}
 845
 846/* Btree in memory cache - hash table */
 847
 848static struct hlist_head *mca_hash(struct cache_set *c, struct bkey *k)
 849{
 850	return &c->bucket_hash[hash_32(PTR_HASH(c, k), BUCKET_HASH_BITS)];
 851}
 852
 853static struct btree *mca_find(struct cache_set *c, struct bkey *k)
 854{
 855	struct btree *b;
 856
 857	rcu_read_lock();
 858	hlist_for_each_entry_rcu(b, mca_hash(c, k), hash)
 859		if (PTR_HASH(c, &b->key) == PTR_HASH(c, k))
 860			goto out;
 861	b = NULL;
 862out:
 863	rcu_read_unlock();
 864	return b;
 865}
 866
 867static int mca_cannibalize_lock(struct cache_set *c, struct btree_op *op)
 868{
 869	spin_lock(&c->btree_cannibalize_lock);
 870	if (likely(c->btree_cache_alloc_lock == NULL)) {
 871		c->btree_cache_alloc_lock = current;
 872	} else if (c->btree_cache_alloc_lock != current) {
 873		if (op)
 874			prepare_to_wait(&c->btree_cache_wait, &op->wait,
 875					TASK_UNINTERRUPTIBLE);
 876		spin_unlock(&c->btree_cannibalize_lock);
 877		return -EINTR;
 878	}
 879	spin_unlock(&c->btree_cannibalize_lock);
 880
 881	return 0;
 882}
 883
 884static struct btree *mca_cannibalize(struct cache_set *c, struct btree_op *op,
 885				     struct bkey *k)
 886{
 887	struct btree *b;
 888
 889	trace_bcache_btree_cache_cannibalize(c);
 890
 891	if (mca_cannibalize_lock(c, op))
 892		return ERR_PTR(-EINTR);
 893
 894	list_for_each_entry_reverse(b, &c->btree_cache, list)
 895		if (!mca_reap(b, btree_order(k), false))
 896			return b;
 897
 898	list_for_each_entry_reverse(b, &c->btree_cache, list)
 899		if (!mca_reap(b, btree_order(k), true))
 900			return b;
 901
 902	WARN(1, "btree cache cannibalize failed\n");
 903	return ERR_PTR(-ENOMEM);
 904}
 905
 906/*
 907 * We can only have one thread cannibalizing other cached btree nodes at a time,
 908 * or we'll deadlock. We use an open coded mutex to ensure that, which a
 909 * cannibalize_bucket() will take. This means every time we unlock the root of
 910 * the btree, we need to release this lock if we have it held.
 911 */
 912void bch_cannibalize_unlock(struct cache_set *c)
 913{
 914	spin_lock(&c->btree_cannibalize_lock);
 915	if (c->btree_cache_alloc_lock == current) {
 916		c->btree_cache_alloc_lock = NULL;
 917		wake_up(&c->btree_cache_wait);
 918	}
 919	spin_unlock(&c->btree_cannibalize_lock);
 920}
 921
 922static struct btree *mca_alloc(struct cache_set *c, struct btree_op *op,
 923			       struct bkey *k, int level)
 924{
 925	struct btree *b;
 926
 927	BUG_ON(current->bio_list);
 928
 929	lockdep_assert_held(&c->bucket_lock);
 930
 931	if (mca_find(c, k))
 932		return NULL;
 933
 934	/* btree_free() doesn't free memory; it sticks the node on the end of
 935	 * the list. Check if there's any freed nodes there:
 936	 */
 937	list_for_each_entry(b, &c->btree_cache_freeable, list)
 938		if (!mca_reap(b, btree_order(k), false))
 939			goto out;
 940
 941	/* We never free struct btree itself, just the memory that holds the on
 942	 * disk node. Check the freed list before allocating a new one:
 943	 */
 944	list_for_each_entry(b, &c->btree_cache_freed, list)
 945		if (!mca_reap(b, 0, false)) {
 946			mca_data_alloc(b, k, __GFP_NOWARN|GFP_NOIO);
 947			if (!b->keys.set[0].data)
 948				goto err;
 949			else
 950				goto out;
 951		}
 952
 953	b = mca_bucket_alloc(c, k, __GFP_NOWARN|GFP_NOIO);
 954	if (!b)
 955		goto err;
 956
 957	BUG_ON(!down_write_trylock(&b->lock));
 958	if (!b->keys.set->data)
 959		goto err;
 960out:
 961	BUG_ON(b->io_mutex.count != 1);
 962
 963	bkey_copy(&b->key, k);
 964	list_move(&b->list, &c->btree_cache);
 965	hlist_del_init_rcu(&b->hash);
 966	hlist_add_head_rcu(&b->hash, mca_hash(c, k));
 967
 968	lock_set_subclass(&b->lock.dep_map, level + 1, _THIS_IP_);
 969	b->parent	= (void *) ~0UL;
 970	b->flags	= 0;
 971	b->written	= 0;
 972	b->level	= level;
 973
 974	if (!b->level)
 975		bch_btree_keys_init(&b->keys, &bch_extent_keys_ops,
 976				    &b->c->expensive_debug_checks);
 977	else
 978		bch_btree_keys_init(&b->keys, &bch_btree_keys_ops,
 979				    &b->c->expensive_debug_checks);
 980
 981	return b;
 982err:
 983	if (b)
 984		rw_unlock(true, b);
 985
 986	b = mca_cannibalize(c, op, k);
 987	if (!IS_ERR(b))
 988		goto out;
 989
 990	return b;
 991}
 992
 993/*
 994 * bch_btree_node_get - find a btree node in the cache and lock it, reading it
 995 * in from disk if necessary.
 996 *
 997 * If IO is necessary and running under submit_bio_noacct, returns -EAGAIN.
 998 *
 999 * The btree node will have either a read or a write lock held, depending on
1000 * level and op->lock.
1001 *
1002 * Note: Only error code or btree pointer will be returned, it is unncessary
1003 *       for callers to check NULL pointer.
1004 */
1005struct btree *bch_btree_node_get(struct cache_set *c, struct btree_op *op,
1006				 struct bkey *k, int level, bool write,
1007				 struct btree *parent)
1008{
1009	int i = 0;
1010	struct btree *b;
1011
1012	BUG_ON(level < 0);
1013retry:
1014	b = mca_find(c, k);
1015
1016	if (!b) {
1017		if (current->bio_list)
1018			return ERR_PTR(-EAGAIN);
1019
1020		mutex_lock(&c->bucket_lock);
1021		b = mca_alloc(c, op, k, level);
1022		mutex_unlock(&c->bucket_lock);
1023
1024		if (!b)
1025			goto retry;
1026		if (IS_ERR(b))
1027			return b;
1028
1029		bch_btree_node_read(b);
1030
1031		if (!write)
1032			downgrade_write(&b->lock);
1033	} else {
1034		rw_lock(write, b, level);
1035		if (PTR_HASH(c, &b->key) != PTR_HASH(c, k)) {
1036			rw_unlock(write, b);
1037			goto retry;
1038		}
1039		BUG_ON(b->level != level);
1040	}
1041
1042	if (btree_node_io_error(b)) {
1043		rw_unlock(write, b);
1044		return ERR_PTR(-EIO);
1045	}
1046
1047	BUG_ON(!b->written);
1048
1049	b->parent = parent;
1050
1051	for (; i <= b->keys.nsets && b->keys.set[i].size; i++) {
1052		prefetch(b->keys.set[i].tree);
1053		prefetch(b->keys.set[i].data);
1054	}
1055
1056	for (; i <= b->keys.nsets; i++)
1057		prefetch(b->keys.set[i].data);
1058
1059	return b;
1060}
1061
1062static void btree_node_prefetch(struct btree *parent, struct bkey *k)
1063{
1064	struct btree *b;
1065
1066	mutex_lock(&parent->c->bucket_lock);
1067	b = mca_alloc(parent->c, NULL, k, parent->level - 1);
1068	mutex_unlock(&parent->c->bucket_lock);
1069
1070	if (!IS_ERR_OR_NULL(b)) {
1071		b->parent = parent;
1072		bch_btree_node_read(b);
1073		rw_unlock(true, b);
1074	}
1075}
1076
1077/* Btree alloc */
1078
1079static void btree_node_free(struct btree *b)
1080{
1081	trace_bcache_btree_node_free(b);
1082
1083	BUG_ON(b == b->c->root);
1084
1085retry:
1086	mutex_lock(&b->write_lock);
1087	/*
1088	 * If the btree node is selected and flushing in btree_flush_write(),
1089	 * delay and retry until the BTREE_NODE_journal_flush bit cleared,
1090	 * then it is safe to free the btree node here. Otherwise this btree
1091	 * node will be in race condition.
1092	 */
1093	if (btree_node_journal_flush(b)) {
1094		mutex_unlock(&b->write_lock);
1095		pr_debug("bnode %p journal_flush set, retry\n", b);
1096		udelay(1);
1097		goto retry;
1098	}
1099
1100	if (btree_node_dirty(b)) {
1101		btree_complete_write(b, btree_current_write(b));
1102		clear_bit(BTREE_NODE_dirty, &b->flags);
1103	}
1104
1105	mutex_unlock(&b->write_lock);
1106
1107	cancel_delayed_work(&b->work);
1108
1109	mutex_lock(&b->c->bucket_lock);
1110	bch_bucket_free(b->c, &b->key);
1111	mca_bucket_free(b);
1112	mutex_unlock(&b->c->bucket_lock);
1113}
1114
1115/*
1116 * Only error code or btree pointer will be returned, it is unncessary for
1117 * callers to check NULL pointer.
1118 */
1119struct btree *__bch_btree_node_alloc(struct cache_set *c, struct btree_op *op,
1120				     int level, bool wait,
1121				     struct btree *parent)
1122{
1123	BKEY_PADDED(key) k;
1124	struct btree *b;
1125
1126	mutex_lock(&c->bucket_lock);
1127retry:
1128	/* return ERR_PTR(-EAGAIN) when it fails */
1129	b = ERR_PTR(-EAGAIN);
1130	if (__bch_bucket_alloc_set(c, RESERVE_BTREE, &k.key, wait))
1131		goto err;
1132
1133	bkey_put(c, &k.key);
1134	SET_KEY_SIZE(&k.key, c->btree_pages * PAGE_SECTORS);
1135
1136	b = mca_alloc(c, op, &k.key, level);
1137	if (IS_ERR(b))
1138		goto err_free;
1139
1140	if (!b) {
1141		cache_bug(c,
1142			"Tried to allocate bucket that was in btree cache");
1143		goto retry;
1144	}
1145
1146	b->parent = parent;
1147	bch_bset_init_next(&b->keys, b->keys.set->data, bset_magic(&b->c->cache->sb));
1148
1149	mutex_unlock(&c->bucket_lock);
1150
1151	trace_bcache_btree_node_alloc(b);
1152	return b;
1153err_free:
1154	bch_bucket_free(c, &k.key);
1155err:
1156	mutex_unlock(&c->bucket_lock);
1157
1158	trace_bcache_btree_node_alloc_fail(c);
1159	return b;
1160}
1161
1162static struct btree *bch_btree_node_alloc(struct cache_set *c,
1163					  struct btree_op *op, int level,
1164					  struct btree *parent)
1165{
1166	return __bch_btree_node_alloc(c, op, level, op != NULL, parent);
1167}
1168
1169static struct btree *btree_node_alloc_replacement(struct btree *b,
1170						  struct btree_op *op)
1171{
1172	struct btree *n = bch_btree_node_alloc(b->c, op, b->level, b->parent);
1173
1174	if (!IS_ERR(n)) {
1175		mutex_lock(&n->write_lock);
1176		bch_btree_sort_into(&b->keys, &n->keys, &b->c->sort);
1177		bkey_copy_key(&n->key, &b->key);
1178		mutex_unlock(&n->write_lock);
1179	}
1180
1181	return n;
1182}
1183
1184static void make_btree_freeing_key(struct btree *b, struct bkey *k)
1185{
1186	unsigned int i;
1187
1188	mutex_lock(&b->c->bucket_lock);
1189
1190	atomic_inc(&b->c->prio_blocked);
1191
1192	bkey_copy(k, &b->key);
1193	bkey_copy_key(k, &ZERO_KEY);
1194
1195	for (i = 0; i < KEY_PTRS(k); i++)
1196		SET_PTR_GEN(k, i,
1197			    bch_inc_gen(b->c->cache,
1198					PTR_BUCKET(b->c, &b->key, i)));
1199
1200	mutex_unlock(&b->c->bucket_lock);
1201}
1202
1203static int btree_check_reserve(struct btree *b, struct btree_op *op)
1204{
1205	struct cache_set *c = b->c;
1206	struct cache *ca = c->cache;
1207	unsigned int reserve = (c->root->level - b->level) * 2 + 1;
1208
1209	mutex_lock(&c->bucket_lock);
1210
1211	if (fifo_used(&ca->free[RESERVE_BTREE]) < reserve) {
1212		if (op)
1213			prepare_to_wait(&c->btree_cache_wait, &op->wait,
1214					TASK_UNINTERRUPTIBLE);
1215		mutex_unlock(&c->bucket_lock);
1216		return -EINTR;
1217	}
1218
1219	mutex_unlock(&c->bucket_lock);
1220
1221	return mca_cannibalize_lock(b->c, op);
1222}
1223
1224/* Garbage collection */
1225
1226static uint8_t __bch_btree_mark_key(struct cache_set *c, int level,
1227				    struct bkey *k)
1228{
1229	uint8_t stale = 0;
1230	unsigned int i;
1231	struct bucket *g;
1232
1233	/*
1234	 * ptr_invalid() can't return true for the keys that mark btree nodes as
1235	 * freed, but since ptr_bad() returns true we'll never actually use them
1236	 * for anything and thus we don't want mark their pointers here
1237	 */
1238	if (!bkey_cmp(k, &ZERO_KEY))
1239		return stale;
1240
1241	for (i = 0; i < KEY_PTRS(k); i++) {
1242		if (!ptr_available(c, k, i))
1243			continue;
1244
1245		g = PTR_BUCKET(c, k, i);
1246
1247		if (gen_after(g->last_gc, PTR_GEN(k, i)))
1248			g->last_gc = PTR_GEN(k, i);
1249
1250		if (ptr_stale(c, k, i)) {
1251			stale = max(stale, ptr_stale(c, k, i));
1252			continue;
1253		}
1254
1255		cache_bug_on(GC_MARK(g) &&
1256			     (GC_MARK(g) == GC_MARK_METADATA) != (level != 0),
1257			     c, "inconsistent ptrs: mark = %llu, level = %i",
1258			     GC_MARK(g), level);
1259
1260		if (level)
1261			SET_GC_MARK(g, GC_MARK_METADATA);
1262		else if (KEY_DIRTY(k))
1263			SET_GC_MARK(g, GC_MARK_DIRTY);
1264		else if (!GC_MARK(g))
1265			SET_GC_MARK(g, GC_MARK_RECLAIMABLE);
1266
1267		/* guard against overflow */
1268		SET_GC_SECTORS_USED(g, min_t(unsigned int,
1269					     GC_SECTORS_USED(g) + KEY_SIZE(k),
1270					     MAX_GC_SECTORS_USED));
1271
1272		BUG_ON(!GC_SECTORS_USED(g));
1273	}
1274
1275	return stale;
1276}
1277
1278#define btree_mark_key(b, k)	__bch_btree_mark_key(b->c, b->level, k)
1279
1280void bch_initial_mark_key(struct cache_set *c, int level, struct bkey *k)
1281{
1282	unsigned int i;
1283
1284	for (i = 0; i < KEY_PTRS(k); i++)
1285		if (ptr_available(c, k, i) &&
1286		    !ptr_stale(c, k, i)) {
1287			struct bucket *b = PTR_BUCKET(c, k, i);
1288
1289			b->gen = PTR_GEN(k, i);
1290
1291			if (level && bkey_cmp(k, &ZERO_KEY))
1292				b->prio = BTREE_PRIO;
1293			else if (!level && b->prio == BTREE_PRIO)
1294				b->prio = INITIAL_PRIO;
1295		}
1296
1297	__bch_btree_mark_key(c, level, k);
1298}
1299
1300void bch_update_bucket_in_use(struct cache_set *c, struct gc_stat *stats)
1301{
1302	stats->in_use = (c->nbuckets - c->avail_nbuckets) * 100 / c->nbuckets;
1303}
1304
1305static bool btree_gc_mark_node(struct btree *b, struct gc_stat *gc)
1306{
1307	uint8_t stale = 0;
1308	unsigned int keys = 0, good_keys = 0;
1309	struct bkey *k;
1310	struct btree_iter_stack iter;
1311	struct bset_tree *t;
1312
1313	gc->nodes++;
1314
1315	for_each_key_filter(&b->keys, k, &iter, bch_ptr_invalid) {
1316		stale = max(stale, btree_mark_key(b, k));
1317		keys++;
1318
1319		if (bch_ptr_bad(&b->keys, k))
1320			continue;
1321
1322		gc->key_bytes += bkey_u64s(k);
1323		gc->nkeys++;
1324		good_keys++;
1325
1326		gc->data += KEY_SIZE(k);
1327	}
1328
1329	for (t = b->keys.set; t <= &b->keys.set[b->keys.nsets]; t++)
1330		btree_bug_on(t->size &&
1331			     bset_written(&b->keys, t) &&
1332			     bkey_cmp(&b->key, &t->end) < 0,
1333			     b, "found short btree key in gc");
1334
1335	if (b->c->gc_always_rewrite)
1336		return true;
1337
1338	if (stale > 10)
1339		return true;
1340
1341	if ((keys - good_keys) * 2 > keys)
1342		return true;
1343
1344	return false;
1345}
1346
1347#define GC_MERGE_NODES	4U
1348
1349struct gc_merge_info {
1350	struct btree	*b;
1351	unsigned int	keys;
1352};
1353
1354static int bch_btree_insert_node(struct btree *b, struct btree_op *op,
1355				 struct keylist *insert_keys,
1356				 atomic_t *journal_ref,
1357				 struct bkey *replace_key);
1358
1359static int btree_gc_coalesce(struct btree *b, struct btree_op *op,
1360			     struct gc_stat *gc, struct gc_merge_info *r)
1361{
1362	unsigned int i, nodes = 0, keys = 0, blocks;
1363	struct btree *new_nodes[GC_MERGE_NODES];
1364	struct keylist keylist;
1365	struct closure cl;
1366	struct bkey *k;
1367
1368	bch_keylist_init(&keylist);
1369
1370	if (btree_check_reserve(b, NULL))
1371		return 0;
1372
1373	memset(new_nodes, 0, sizeof(new_nodes));
1374	closure_init_stack(&cl);
1375
1376	while (nodes < GC_MERGE_NODES && !IS_ERR_OR_NULL(r[nodes].b))
1377		keys += r[nodes++].keys;
1378
1379	blocks = btree_default_blocks(b->c) * 2 / 3;
1380
1381	if (nodes < 2 ||
1382	    __set_blocks(b->keys.set[0].data, keys,
1383			 block_bytes(b->c->cache)) > blocks * (nodes - 1))
1384		return 0;
1385
1386	for (i = 0; i < nodes; i++) {
1387		new_nodes[i] = btree_node_alloc_replacement(r[i].b, NULL);
1388		if (IS_ERR(new_nodes[i]))
1389			goto out_nocoalesce;
1390	}
1391
1392	/*
1393	 * We have to check the reserve here, after we've allocated our new
1394	 * nodes, to make sure the insert below will succeed - we also check
1395	 * before as an optimization to potentially avoid a bunch of expensive
1396	 * allocs/sorts
1397	 */
1398	if (btree_check_reserve(b, NULL))
1399		goto out_nocoalesce;
1400
1401	for (i = 0; i < nodes; i++)
1402		mutex_lock(&new_nodes[i]->write_lock);
1403
1404	for (i = nodes - 1; i > 0; --i) {
1405		struct bset *n1 = btree_bset_first(new_nodes[i]);
1406		struct bset *n2 = btree_bset_first(new_nodes[i - 1]);
1407		struct bkey *k, *last = NULL;
1408
1409		keys = 0;
1410
1411		if (i > 1) {
1412			for (k = n2->start;
1413			     k < bset_bkey_last(n2);
1414			     k = bkey_next(k)) {
1415				if (__set_blocks(n1, n1->keys + keys +
1416						 bkey_u64s(k),
1417						 block_bytes(b->c->cache)) > blocks)
1418					break;
1419
1420				last = k;
1421				keys += bkey_u64s(k);
1422			}
1423		} else {
1424			/*
1425			 * Last node we're not getting rid of - we're getting
1426			 * rid of the node at r[0]. Have to try and fit all of
1427			 * the remaining keys into this node; we can't ensure
1428			 * they will always fit due to rounding and variable
1429			 * length keys (shouldn't be possible in practice,
1430			 * though)
1431			 */
1432			if (__set_blocks(n1, n1->keys + n2->keys,
1433					 block_bytes(b->c->cache)) >
1434			    btree_blocks(new_nodes[i]))
1435				goto out_unlock_nocoalesce;
1436
1437			keys = n2->keys;
1438			/* Take the key of the node we're getting rid of */
1439			last = &r->b->key;
1440		}
1441
1442		BUG_ON(__set_blocks(n1, n1->keys + keys, block_bytes(b->c->cache)) >
1443		       btree_blocks(new_nodes[i]));
1444
1445		if (last)
1446			bkey_copy_key(&new_nodes[i]->key, last);
1447
1448		memcpy(bset_bkey_last(n1),
1449		       n2->start,
1450		       (void *) bset_bkey_idx(n2, keys) - (void *) n2->start);
1451
1452		n1->keys += keys;
1453		r[i].keys = n1->keys;
1454
1455		memmove(n2->start,
1456			bset_bkey_idx(n2, keys),
1457			(void *) bset_bkey_last(n2) -
1458			(void *) bset_bkey_idx(n2, keys));
1459
1460		n2->keys -= keys;
1461
1462		if (__bch_keylist_realloc(&keylist,
1463					  bkey_u64s(&new_nodes[i]->key)))
1464			goto out_unlock_nocoalesce;
1465
1466		bch_btree_node_write(new_nodes[i], &cl);
1467		bch_keylist_add(&keylist, &new_nodes[i]->key);
1468	}
1469
1470	for (i = 0; i < nodes; i++)
1471		mutex_unlock(&new_nodes[i]->write_lock);
1472
1473	closure_sync(&cl);
1474
1475	/* We emptied out this node */
1476	BUG_ON(btree_bset_first(new_nodes[0])->keys);
1477	btree_node_free(new_nodes[0]);
1478	rw_unlock(true, new_nodes[0]);
1479	new_nodes[0] = NULL;
1480
1481	for (i = 0; i < nodes; i++) {
1482		if (__bch_keylist_realloc(&keylist, bkey_u64s(&r[i].b->key)))
1483			goto out_nocoalesce;
1484
1485		make_btree_freeing_key(r[i].b, keylist.top);
1486		bch_keylist_push(&keylist);
1487	}
1488
1489	bch_btree_insert_node(b, op, &keylist, NULL, NULL);
1490	BUG_ON(!bch_keylist_empty(&keylist));
1491
1492	for (i = 0; i < nodes; i++) {
1493		btree_node_free(r[i].b);
1494		rw_unlock(true, r[i].b);
1495
1496		r[i].b = new_nodes[i];
1497	}
1498
1499	memmove(r, r + 1, sizeof(r[0]) * (nodes - 1));
1500	r[nodes - 1].b = ERR_PTR(-EINTR);
1501
1502	trace_bcache_btree_gc_coalesce(nodes);
1503	gc->nodes--;
1504
1505	bch_keylist_free(&keylist);
1506
1507	/* Invalidated our iterator */
1508	return -EINTR;
1509
1510out_unlock_nocoalesce:
1511	for (i = 0; i < nodes; i++)
1512		mutex_unlock(&new_nodes[i]->write_lock);
1513
1514out_nocoalesce:
1515	closure_sync(&cl);
1516
1517	while ((k = bch_keylist_pop(&keylist)))
1518		if (!bkey_cmp(k, &ZERO_KEY))
1519			atomic_dec(&b->c->prio_blocked);
1520	bch_keylist_free(&keylist);
1521
1522	for (i = 0; i < nodes; i++)
1523		if (!IS_ERR_OR_NULL(new_nodes[i])) {
1524			btree_node_free(new_nodes[i]);
1525			rw_unlock(true, new_nodes[i]);
1526		}
1527	return 0;
1528}
1529
1530static int btree_gc_rewrite_node(struct btree *b, struct btree_op *op,
1531				 struct btree *replace)
1532{
1533	struct keylist keys;
1534	struct btree *n;
1535
1536	if (btree_check_reserve(b, NULL))
1537		return 0;
1538
1539	n = btree_node_alloc_replacement(replace, NULL);
1540	if (IS_ERR(n))
1541		return 0;
1542
1543	/* recheck reserve after allocating replacement node */
1544	if (btree_check_reserve(b, NULL)) {
1545		btree_node_free(n);
1546		rw_unlock(true, n);
1547		return 0;
1548	}
1549
1550	bch_btree_node_write_sync(n);
1551
1552	bch_keylist_init(&keys);
1553	bch_keylist_add(&keys, &n->key);
1554
1555	make_btree_freeing_key(replace, keys.top);
1556	bch_keylist_push(&keys);
1557
1558	bch_btree_insert_node(b, op, &keys, NULL, NULL);
1559	BUG_ON(!bch_keylist_empty(&keys));
1560
1561	btree_node_free(replace);
1562	rw_unlock(true, n);
1563
1564	/* Invalidated our iterator */
1565	return -EINTR;
1566}
1567
1568static unsigned int btree_gc_count_keys(struct btree *b)
1569{
1570	struct bkey *k;
1571	struct btree_iter_stack iter;
1572	unsigned int ret = 0;
1573
1574	for_each_key_filter(&b->keys, k, &iter, bch_ptr_bad)
1575		ret += bkey_u64s(k);
1576
1577	return ret;
1578}
1579
1580static size_t btree_gc_min_nodes(struct cache_set *c)
1581{
1582	size_t min_nodes = GC_NODES_MIN;
1583
1584	if (atomic_read(&c->search_inflight) == 0) {
1585		size_t n = c->gc_stats.nodes >> MAX_GC_TIMES_SHIFT;
1586
1587		if (min_nodes < n)
1588			min_nodes = n;
1589	}
1590
1591	return min_nodes;
1592}
1593
1594static uint64_t btree_gc_sleep_ms(struct cache_set *c)
1595{
1596	uint64_t sleep_ms;
1597
1598	if (atomic_read(&c->bucket_wait_cnt) > 0)
1599		sleep_ms = GC_SLEEP_MS_MIN;
1600	else
1601		sleep_ms = GC_SLEEP_MS;
1602
1603	return sleep_ms;
1604}
1605
1606static int btree_gc_recurse(struct btree *b, struct btree_op *op,
1607			    struct closure *writes, struct gc_stat *gc)
1608{
1609	int ret = 0;
1610	bool should_rewrite;
1611	struct bkey *k;
1612	struct btree_iter_stack iter;
1613	struct gc_merge_info r[GC_MERGE_NODES];
1614	struct gc_merge_info *i, *last = r + ARRAY_SIZE(r) - 1;
1615
1616	bch_btree_iter_stack_init(&b->keys, &iter, &b->c->gc_done);
1617
1618	for (i = r; i < r + ARRAY_SIZE(r); i++)
1619		i->b = ERR_PTR(-EINTR);
1620
1621	while (1) {
1622		k = bch_btree_iter_next_filter(&iter.iter, &b->keys,
1623					       bch_ptr_bad);
1624		if (k) {
1625			r->b = bch_btree_node_get(b->c, op, k, b->level - 1,
1626						  true, b);
1627			if (IS_ERR(r->b)) {
1628				ret = PTR_ERR(r->b);
1629				break;
1630			}
1631
1632			r->keys = btree_gc_count_keys(r->b);
1633
1634			ret = btree_gc_coalesce(b, op, gc, r);
1635			if (ret)
1636				break;
1637		}
1638
1639		if (!last->b)
1640			break;
1641
1642		if (!IS_ERR(last->b)) {
1643			should_rewrite = btree_gc_mark_node(last->b, gc);
1644			if (should_rewrite) {
1645				ret = btree_gc_rewrite_node(b, op, last->b);
1646				if (ret)
1647					break;
1648			}
1649
1650			if (last->b->level) {
1651				ret = btree_gc_recurse(last->b, op, writes, gc);
1652				if (ret)
1653					break;
1654			}
1655
1656			bkey_copy_key(&b->c->gc_done, &last->b->key);
1657
1658			/*
1659			 * Must flush leaf nodes before gc ends, since replace
1660			 * operations aren't journalled
1661			 */
1662			mutex_lock(&last->b->write_lock);
1663			if (btree_node_dirty(last->b))
1664				bch_btree_node_write(last->b, writes);
1665			mutex_unlock(&last->b->write_lock);
1666			rw_unlock(true, last->b);
1667		}
1668
1669		memmove(r + 1, r, sizeof(r[0]) * (GC_MERGE_NODES - 1));
1670		r->b = NULL;
1671
1672		if (gc->nodes >= (gc->nodes_pre + btree_gc_min_nodes(b->c))) {
1673			gc->nodes_pre =  gc->nodes;
1674			ret = -EAGAIN;
1675			break;
1676		}
1677
1678		if (need_resched()) {
1679			ret = -EAGAIN;
1680			break;
1681		}
1682	}
1683
1684	for (i = r; i < r + ARRAY_SIZE(r); i++)
1685		if (!IS_ERR_OR_NULL(i->b)) {
1686			mutex_lock(&i->b->write_lock);
1687			if (btree_node_dirty(i->b))
1688				bch_btree_node_write(i->b, writes);
1689			mutex_unlock(&i->b->write_lock);
1690			rw_unlock(true, i->b);
1691		}
1692
1693	return ret;
1694}
1695
1696static int bch_btree_gc_root(struct btree *b, struct btree_op *op,
1697			     struct closure *writes, struct gc_stat *gc)
1698{
1699	struct btree *n = NULL;
1700	int ret = 0;
1701	bool should_rewrite;
1702
1703	should_rewrite = btree_gc_mark_node(b, gc);
1704	if (should_rewrite) {
1705		n = btree_node_alloc_replacement(b, NULL);
1706
1707		if (!IS_ERR(n)) {
1708			bch_btree_node_write_sync(n);
1709
1710			bch_btree_set_root(n);
1711			btree_node_free(b);
1712			rw_unlock(true, n);
1713
1714			return -EINTR;
1715		}
1716	}
1717
1718	__bch_btree_mark_key(b->c, b->level + 1, &b->key);
1719
1720	if (b->level) {
1721		ret = btree_gc_recurse(b, op, writes, gc);
1722		if (ret)
1723			return ret;
1724	}
1725
1726	bkey_copy_key(&b->c->gc_done, &b->key);
1727
1728	return ret;
1729}
1730
1731static void btree_gc_start(struct cache_set *c)
1732{
1733	struct cache *ca;
1734	struct bucket *b;
1735
1736	if (!c->gc_mark_valid)
1737		return;
1738
1739	mutex_lock(&c->bucket_lock);
1740
1741	c->gc_done = ZERO_KEY;
1742
1743	ca = c->cache;
1744	for_each_bucket(b, ca) {
1745		b->last_gc = b->gen;
1746		if (bch_can_invalidate_bucket(ca, b))
1747			b->reclaimable_in_gc = 1;
1748		if (!atomic_read(&b->pin)) {
1749			SET_GC_MARK(b, 0);
1750			SET_GC_SECTORS_USED(b, 0);
1751		}
1752	}
1753
1754	c->gc_mark_valid = 0;
1755	mutex_unlock(&c->bucket_lock);
1756}
1757
1758static void bch_btree_gc_finish(struct cache_set *c)
1759{
1760	struct bucket *b;
1761	struct cache *ca;
1762	unsigned int i, j;
1763	uint64_t *k;
1764
1765	mutex_lock(&c->bucket_lock);
1766
1767	set_gc_sectors(c);
1768	c->gc_mark_valid = 1;
1769	c->need_gc	= 0;
1770
1771	for (i = 0; i < KEY_PTRS(&c->uuid_bucket); i++)
1772		SET_GC_MARK(PTR_BUCKET(c, &c->uuid_bucket, i),
1773			    GC_MARK_METADATA);
1774
1775	/* don't reclaim buckets to which writeback keys point */
1776	rcu_read_lock();
1777	for (i = 0; i < c->devices_max_used; i++) {
1778		struct bcache_device *d = c->devices[i];
1779		struct cached_dev *dc;
1780		struct keybuf_key *w, *n;
1781
1782		if (!d || UUID_FLASH_ONLY(&c->uuids[i]))
1783			continue;
1784		dc = container_of(d, struct cached_dev, disk);
1785
1786		spin_lock(&dc->writeback_keys.lock);
1787		rbtree_postorder_for_each_entry_safe(w, n,
1788					&dc->writeback_keys.keys, node)
1789			for (j = 0; j < KEY_PTRS(&w->key); j++)
1790				SET_GC_MARK(PTR_BUCKET(c, &w->key, j),
1791					    GC_MARK_DIRTY);
1792		spin_unlock(&dc->writeback_keys.lock);
1793	}
1794	rcu_read_unlock();
1795
1796	c->avail_nbuckets = 0;
1797
1798	ca = c->cache;
1799	ca->invalidate_needs_gc = 0;
1800
1801	for (k = ca->sb.d; k < ca->sb.d + ca->sb.keys; k++)
1802		SET_GC_MARK(ca->buckets + *k, GC_MARK_METADATA);
1803
1804	for (k = ca->prio_buckets;
1805	     k < ca->prio_buckets + prio_buckets(ca) * 2; k++)
1806		SET_GC_MARK(ca->buckets + *k, GC_MARK_METADATA);
1807
1808	for_each_bucket(b, ca) {
1809		c->need_gc	= max(c->need_gc, bucket_gc_gen(b));
1810
1811		if (b->reclaimable_in_gc)
1812			b->reclaimable_in_gc = 0;
1813
1814		if (atomic_read(&b->pin))
1815			continue;
1816
1817		BUG_ON(!GC_MARK(b) && GC_SECTORS_USED(b));
1818
1819		if (!GC_MARK(b) || GC_MARK(b) == GC_MARK_RECLAIMABLE)
1820			c->avail_nbuckets++;
1821	}
1822
1823	mutex_unlock(&c->bucket_lock);
1824}
1825
1826static void bch_btree_gc(struct cache_set *c)
1827{
1828	int ret;
1829	struct gc_stat stats;
1830	struct closure writes;
1831	struct btree_op op;
1832	uint64_t start_time = local_clock();
1833
1834	trace_bcache_gc_start(c);
1835
1836	memset(&stats, 0, sizeof(struct gc_stat));
1837	closure_init_stack(&writes);
1838	bch_btree_op_init(&op, SHRT_MAX);
1839
1840	btree_gc_start(c);
1841
1842	/* if CACHE_SET_IO_DISABLE set, gc thread should stop too */
1843	do {
1844		ret = bcache_btree_root(gc_root, c, &op, &writes, &stats);
1845		closure_sync(&writes);
1846		cond_resched();
1847
1848		if (ret == -EAGAIN)
1849			schedule_timeout_interruptible(
1850				msecs_to_jiffies(btree_gc_sleep_ms(c)));
1851		else if (ret)
1852			pr_warn("gc failed!\n");
1853	} while (ret && !test_bit(CACHE_SET_IO_DISABLE, &c->flags));
1854
1855	bch_btree_gc_finish(c);
1856	wake_up_allocators(c);
1857
1858	bch_time_stats_update(&c->btree_gc_time, start_time);
1859
1860	stats.key_bytes *= sizeof(uint64_t);
1861	stats.data	<<= 9;
1862	bch_update_bucket_in_use(c, &stats);
1863	memcpy(&c->gc_stats, &stats, sizeof(struct gc_stat));
1864
1865	trace_bcache_gc_end(c);
1866
1867	bch_moving_gc(c);
1868}
1869
1870static bool gc_should_run(struct cache_set *c)
1871{
1872	struct cache *ca = c->cache;
1873
1874	if (ca->invalidate_needs_gc)
1875		return true;
1876
1877	if (atomic_read(&c->sectors_to_gc) < 0)
1878		return true;
1879
1880	return false;
1881}
1882
1883static int bch_gc_thread(void *arg)
1884{
1885	struct cache_set *c = arg;
1886
1887	while (1) {
1888		wait_event_interruptible(c->gc_wait,
1889			   kthread_should_stop() ||
1890			   test_bit(CACHE_SET_IO_DISABLE, &c->flags) ||
1891			   gc_should_run(c));
1892
1893		if (kthread_should_stop() ||
1894		    test_bit(CACHE_SET_IO_DISABLE, &c->flags))
1895			break;
1896
1897		set_gc_sectors(c);
1898		bch_btree_gc(c);
1899	}
1900
1901	wait_for_kthread_stop();
1902	return 0;
1903}
1904
1905int bch_gc_thread_start(struct cache_set *c)
1906{
1907	c->gc_thread = kthread_run(bch_gc_thread, c, "bcache_gc");
1908	return PTR_ERR_OR_ZERO(c->gc_thread);
1909}
1910
1911/* Initial partial gc */
1912
1913static int bch_btree_check_recurse(struct btree *b, struct btree_op *op)
1914{
1915	int ret = 0;
1916	struct bkey *k, *p = NULL;
1917	struct btree_iter_stack iter;
1918
1919	for_each_key_filter(&b->keys, k, &iter, bch_ptr_invalid)
1920		bch_initial_mark_key(b->c, b->level, k);
1921
1922	bch_initial_mark_key(b->c, b->level + 1, &b->key);
1923
1924	if (b->level) {
1925		bch_btree_iter_stack_init(&b->keys, &iter, NULL);
1926
1927		do {
1928			k = bch_btree_iter_next_filter(&iter.iter, &b->keys,
1929						       bch_ptr_bad);
1930			if (k) {
1931				btree_node_prefetch(b, k);
1932				/*
1933				 * initiallize c->gc_stats.nodes
1934				 * for incremental GC
1935				 */
1936				b->c->gc_stats.nodes++;
1937			}
1938
1939			if (p)
1940				ret = bcache_btree(check_recurse, p, b, op);
1941
1942			p = k;
1943		} while (p && !ret);
1944	}
1945
1946	return ret;
1947}
1948
1949
1950static int bch_btree_check_thread(void *arg)
1951{
1952	int ret;
1953	struct btree_check_info *info = arg;
1954	struct btree_check_state *check_state = info->state;
1955	struct cache_set *c = check_state->c;
1956	struct btree_iter_stack iter;
1957	struct bkey *k, *p;
1958	int cur_idx, prev_idx, skip_nr;
1959
1960	k = p = NULL;
1961	cur_idx = prev_idx = 0;
1962	ret = 0;
1963
1964	/* root node keys are checked before thread created */
1965	bch_btree_iter_stack_init(&c->root->keys, &iter, NULL);
1966	k = bch_btree_iter_next_filter(&iter.iter, &c->root->keys, bch_ptr_bad);
1967	BUG_ON(!k);
1968
1969	p = k;
1970	while (k) {
1971		/*
1972		 * Fetch a root node key index, skip the keys which
1973		 * should be fetched by other threads, then check the
1974		 * sub-tree indexed by the fetched key.
1975		 */
1976		spin_lock(&check_state->idx_lock);
1977		cur_idx = check_state->key_idx;
1978		check_state->key_idx++;
1979		spin_unlock(&check_state->idx_lock);
1980
1981		skip_nr = cur_idx - prev_idx;
1982
1983		while (skip_nr) {
1984			k = bch_btree_iter_next_filter(&iter.iter,
1985						       &c->root->keys,
1986						       bch_ptr_bad);
1987			if (k)
1988				p = k;
1989			else {
1990				/*
1991				 * No more keys to check in root node,
1992				 * current checking threads are enough,
1993				 * stop creating more.
1994				 */
1995				atomic_set(&check_state->enough, 1);
1996				/* Update check_state->enough earlier */
1997				smp_mb__after_atomic();
1998				goto out;
1999			}
2000			skip_nr--;
2001			cond_resched();
2002		}
2003
2004		if (p) {
2005			struct btree_op op;
2006
2007			btree_node_prefetch(c->root, p);
2008			c->gc_stats.nodes++;
2009			bch_btree_op_init(&op, 0);
2010			ret = bcache_btree(check_recurse, p, c->root, &op);
2011			/*
2012			 * The op may be added to cache_set's btree_cache_wait
2013			 * in mca_cannibalize(), must ensure it is removed from
2014			 * the list and release btree_cache_alloc_lock before
2015			 * free op memory.
2016			 * Otherwise, the btree_cache_wait will be damaged.
2017			 */
2018			bch_cannibalize_unlock(c);
2019			finish_wait(&c->btree_cache_wait, &(&op)->wait);
2020			if (ret)
2021				goto out;
2022		}
2023		p = NULL;
2024		prev_idx = cur_idx;
2025		cond_resched();
2026	}
2027
2028out:
2029	info->result = ret;
2030	/* update check_state->started among all CPUs */
2031	smp_mb__before_atomic();
2032	if (atomic_dec_and_test(&check_state->started))
2033		wake_up(&check_state->wait);
2034
2035	return ret;
2036}
2037
2038
2039
2040static int bch_btree_chkthread_nr(void)
2041{
2042	int n = num_online_cpus()/2;
2043
2044	if (n == 0)
2045		n = 1;
2046	else if (n > BCH_BTR_CHKTHREAD_MAX)
2047		n = BCH_BTR_CHKTHREAD_MAX;
2048
2049	return n;
2050}
2051
2052int bch_btree_check(struct cache_set *c)
2053{
2054	int ret = 0;
2055	int i;
2056	struct bkey *k = NULL;
2057	struct btree_iter_stack iter;
2058	struct btree_check_state check_state;
2059
2060	/* check and mark root node keys */
2061	for_each_key_filter(&c->root->keys, k, &iter, bch_ptr_invalid)
2062		bch_initial_mark_key(c, c->root->level, k);
2063
2064	bch_initial_mark_key(c, c->root->level + 1, &c->root->key);
2065
2066	if (c->root->level == 0)
2067		return 0;
2068
2069	memset(&check_state, 0, sizeof(struct btree_check_state));
2070	check_state.c = c;
2071	check_state.total_threads = bch_btree_chkthread_nr();
2072	check_state.key_idx = 0;
2073	spin_lock_init(&check_state.idx_lock);
2074	atomic_set(&check_state.started, 0);
2075	atomic_set(&check_state.enough, 0);
2076	init_waitqueue_head(&check_state.wait);
2077
2078	rw_lock(0, c->root, c->root->level);
2079	/*
2080	 * Run multiple threads to check btree nodes in parallel,
2081	 * if check_state.enough is non-zero, it means current
2082	 * running check threads are enough, unncessary to create
2083	 * more.
2084	 */
2085	for (i = 0; i < check_state.total_threads; i++) {
2086		/* fetch latest check_state.enough earlier */
2087		smp_mb__before_atomic();
2088		if (atomic_read(&check_state.enough))
2089			break;
2090
2091		check_state.infos[i].result = 0;
2092		check_state.infos[i].state = &check_state;
2093
2094		check_state.infos[i].thread =
2095			kthread_run(bch_btree_check_thread,
2096				    &check_state.infos[i],
2097				    "bch_btrchk[%d]", i);
2098		if (IS_ERR(check_state.infos[i].thread)) {
2099			pr_err("fails to run thread bch_btrchk[%d]\n", i);
2100			for (--i; i >= 0; i--)
2101				kthread_stop(check_state.infos[i].thread);
2102			ret = -ENOMEM;
2103			goto out;
2104		}
2105		atomic_inc(&check_state.started);
2106	}
2107
2108	/*
2109	 * Must wait for all threads to stop.
2110	 */
2111	wait_event(check_state.wait, atomic_read(&check_state.started) == 0);
2112
2113	for (i = 0; i < check_state.total_threads; i++) {
2114		if (check_state.infos[i].result) {
2115			ret = check_state.infos[i].result;
2116			goto out;
2117		}
2118	}
2119
2120out:
2121	rw_unlock(0, c->root);
2122	return ret;
2123}
2124
2125void bch_initial_gc_finish(struct cache_set *c)
2126{
2127	struct cache *ca = c->cache;
2128	struct bucket *b;
2129
2130	bch_btree_gc_finish(c);
2131
2132	mutex_lock(&c->bucket_lock);
2133
2134	/*
2135	 * We need to put some unused buckets directly on the prio freelist in
2136	 * order to get the allocator thread started - it needs freed buckets in
2137	 * order to rewrite the prios and gens, and it needs to rewrite prios
2138	 * and gens in order to free buckets.
2139	 *
2140	 * This is only safe for buckets that have no live data in them, which
2141	 * there should always be some of.
2142	 */
2143	for_each_bucket(b, ca) {
2144		if (fifo_full(&ca->free[RESERVE_PRIO]) &&
2145		    fifo_full(&ca->free[RESERVE_BTREE]))
2146			break;
2147
2148		if (bch_can_invalidate_bucket(ca, b) &&
2149		    !GC_MARK(b)) {
2150			__bch_invalidate_one_bucket(ca, b);
2151			if (!fifo_push(&ca->free[RESERVE_PRIO],
2152			   b - ca->buckets))
2153				fifo_push(&ca->free[RESERVE_BTREE],
2154					  b - ca->buckets);
2155		}
2156	}
2157
2158	mutex_unlock(&c->bucket_lock);
2159}
2160
2161/* Btree insertion */
2162
2163static bool btree_insert_key(struct btree *b, struct bkey *k,
2164			     struct bkey *replace_key)
2165{
2166	unsigned int status;
2167
2168	BUG_ON(bkey_cmp(k, &b->key) > 0);
2169
2170	status = bch_btree_insert_key(&b->keys, k, replace_key);
2171	if (status != BTREE_INSERT_STATUS_NO_INSERT) {
2172		bch_check_keys(&b->keys, "%u for %s", status,
2173			       replace_key ? "replace" : "insert");
2174
2175		trace_bcache_btree_insert_key(b, k, replace_key != NULL,
2176					      status);
2177		return true;
2178	} else
2179		return false;
2180}
2181
2182static size_t insert_u64s_remaining(struct btree *b)
2183{
2184	long ret = bch_btree_keys_u64s_remaining(&b->keys);
2185
2186	/*
2187	 * Might land in the middle of an existing extent and have to split it
2188	 */
2189	if (b->keys.ops->is_extents)
2190		ret -= KEY_MAX_U64S;
2191
2192	return max(ret, 0L);
2193}
2194
2195static bool bch_btree_insert_keys(struct btree *b, struct btree_op *op,
2196				  struct keylist *insert_keys,
2197				  struct bkey *replace_key)
2198{
2199	bool ret = false;
2200	int oldsize = bch_count_data(&b->keys);
2201
2202	while (!bch_keylist_empty(insert_keys)) {
2203		struct bkey *k = insert_keys->keys;
2204
2205		if (bkey_u64s(k) > insert_u64s_remaining(b))
2206			break;
2207
2208		if (bkey_cmp(k, &b->key) <= 0) {
2209			if (!b->level)
2210				bkey_put(b->c, k);
2211
2212			ret |= btree_insert_key(b, k, replace_key);
2213			bch_keylist_pop_front(insert_keys);
2214		} else if (bkey_cmp(&START_KEY(k), &b->key) < 0) {
2215			BKEY_PADDED(key) temp;
2216			bkey_copy(&temp.key, insert_keys->keys);
2217
2218			bch_cut_back(&b->key, &temp.key);
2219			bch_cut_front(&b->key, insert_keys->keys);
2220
2221			ret |= btree_insert_key(b, &temp.key, replace_key);
2222			break;
2223		} else {
2224			break;
2225		}
2226	}
2227
2228	if (!ret)
2229		op->insert_collision = true;
2230
2231	BUG_ON(!bch_keylist_empty(insert_keys) && b->level);
2232
2233	BUG_ON(bch_count_data(&b->keys) < oldsize);
2234	return ret;
2235}
2236
2237static int btree_split(struct btree *b, struct btree_op *op,
2238		       struct keylist *insert_keys,
2239		       struct bkey *replace_key)
2240{
2241	bool split;
2242	struct btree *n1, *n2 = NULL, *n3 = NULL;
2243	uint64_t start_time = local_clock();
2244	struct closure cl;
2245	struct keylist parent_keys;
2246
2247	closure_init_stack(&cl);
2248	bch_keylist_init(&parent_keys);
2249
2250	if (btree_check_reserve(b, op)) {
2251		if (!b->level)
2252			return -EINTR;
2253		else
2254			WARN(1, "insufficient reserve for split\n");
2255	}
2256
2257	n1 = btree_node_alloc_replacement(b, op);
2258	if (IS_ERR(n1))
2259		goto err;
2260
2261	split = set_blocks(btree_bset_first(n1),
2262			   block_bytes(n1->c->cache)) > (btree_blocks(b) * 4) / 5;
2263
2264	if (split) {
2265		unsigned int keys = 0;
2266
2267		trace_bcache_btree_node_split(b, btree_bset_first(n1)->keys);
2268
2269		n2 = bch_btree_node_alloc(b->c, op, b->level, b->parent);
2270		if (IS_ERR(n2))
2271			goto err_free1;
2272
2273		if (!b->parent) {
2274			n3 = bch_btree_node_alloc(b->c, op, b->level + 1, NULL);
2275			if (IS_ERR(n3))
2276				goto err_free2;
2277		}
2278
2279		mutex_lock(&n1->write_lock);
2280		mutex_lock(&n2->write_lock);
2281
2282		bch_btree_insert_keys(n1, op, insert_keys, replace_key);
2283
2284		/*
2285		 * Has to be a linear search because we don't have an auxiliary
2286		 * search tree yet
2287		 */
2288
2289		while (keys < (btree_bset_first(n1)->keys * 3) / 5)
2290			keys += bkey_u64s(bset_bkey_idx(btree_bset_first(n1),
2291							keys));
2292
2293		bkey_copy_key(&n1->key,
2294			      bset_bkey_idx(btree_bset_first(n1), keys));
2295		keys += bkey_u64s(bset_bkey_idx(btree_bset_first(n1), keys));
2296
2297		btree_bset_first(n2)->keys = btree_bset_first(n1)->keys - keys;
2298		btree_bset_first(n1)->keys = keys;
2299
2300		memcpy(btree_bset_first(n2)->start,
2301		       bset_bkey_last(btree_bset_first(n1)),
2302		       btree_bset_first(n2)->keys * sizeof(uint64_t));
2303
2304		bkey_copy_key(&n2->key, &b->key);
2305
2306		bch_keylist_add(&parent_keys, &n2->key);
2307		bch_btree_node_write(n2, &cl);
2308		mutex_unlock(&n2->write_lock);
2309		rw_unlock(true, n2);
2310	} else {
2311		trace_bcache_btree_node_compact(b, btree_bset_first(n1)->keys);
2312
2313		mutex_lock(&n1->write_lock);
2314		bch_btree_insert_keys(n1, op, insert_keys, replace_key);
2315	}
2316
2317	bch_keylist_add(&parent_keys, &n1->key);
2318	bch_btree_node_write(n1, &cl);
2319	mutex_unlock(&n1->write_lock);
2320
2321	if (n3) {
2322		/* Depth increases, make a new root */
2323		mutex_lock(&n3->write_lock);
2324		bkey_copy_key(&n3->key, &MAX_KEY);
2325		bch_btree_insert_keys(n3, op, &parent_keys, NULL);
2326		bch_btree_node_write(n3, &cl);
2327		mutex_unlock(&n3->write_lock);
2328
2329		closure_sync(&cl);
2330		bch_btree_set_root(n3);
2331		rw_unlock(true, n3);
2332	} else if (!b->parent) {
2333		/* Root filled up but didn't need to be split */
2334		closure_sync(&cl);
2335		bch_btree_set_root(n1);
2336	} else {
2337		/* Split a non root node */
2338		closure_sync(&cl);
2339		make_btree_freeing_key(b, parent_keys.top);
2340		bch_keylist_push(&parent_keys);
2341
2342		bch_btree_insert_node(b->parent, op, &parent_keys, NULL, NULL);
2343		BUG_ON(!bch_keylist_empty(&parent_keys));
2344	}
2345
2346	btree_node_free(b);
2347	rw_unlock(true, n1);
2348
2349	bch_time_stats_update(&b->c->btree_split_time, start_time);
2350
2351	return 0;
2352err_free2:
2353	bkey_put(b->c, &n2->key);
2354	btree_node_free(n2);
2355	rw_unlock(true, n2);
2356err_free1:
2357	bkey_put(b->c, &n1->key);
2358	btree_node_free(n1);
2359	rw_unlock(true, n1);
2360err:
2361	WARN(1, "bcache: btree split failed (level %u)", b->level);
2362
2363	if (n3 == ERR_PTR(-EAGAIN) ||
2364	    n2 == ERR_PTR(-EAGAIN) ||
2365	    n1 == ERR_PTR(-EAGAIN))
2366		return -EAGAIN;
2367
2368	return -ENOMEM;
2369}
2370
2371static int bch_btree_insert_node(struct btree *b, struct btree_op *op,
2372				 struct keylist *insert_keys,
2373				 atomic_t *journal_ref,
2374				 struct bkey *replace_key)
2375{
2376	struct closure cl;
2377
2378	BUG_ON(b->level && replace_key);
2379
2380	closure_init_stack(&cl);
2381
2382	mutex_lock(&b->write_lock);
2383
2384	if (write_block(b) != btree_bset_last(b) &&
2385	    b->keys.last_set_unwritten)
2386		bch_btree_init_next(b); /* just wrote a set */
2387
2388	if (bch_keylist_nkeys(insert_keys) > insert_u64s_remaining(b)) {
2389		mutex_unlock(&b->write_lock);
2390		goto split;
2391	}
2392
2393	BUG_ON(write_block(b) != btree_bset_last(b));
2394
2395	if (bch_btree_insert_keys(b, op, insert_keys, replace_key)) {
2396		if (!b->level)
2397			bch_btree_leaf_dirty(b, journal_ref);
2398		else
2399			bch_btree_node_write(b, &cl);
2400	}
2401
2402	mutex_unlock(&b->write_lock);
2403
2404	/* wait for btree node write if necessary, after unlock */
2405	closure_sync(&cl);
2406
2407	return 0;
2408split:
2409	if (current->bio_list) {
2410		op->lock = b->c->root->level + 1;
2411		return -EAGAIN;
2412	} else if (op->lock <= b->c->root->level) {
2413		op->lock = b->c->root->level + 1;
2414		return -EINTR;
2415	} else {
2416		/* Invalidated all iterators */
2417		int ret = btree_split(b, op, insert_keys, replace_key);
2418
2419		if (bch_keylist_empty(insert_keys))
2420			return 0;
2421		else if (!ret)
2422			return -EINTR;
2423		return ret;
2424	}
2425}
2426
2427int bch_btree_insert_check_key(struct btree *b, struct btree_op *op,
2428			       struct bkey *check_key)
2429{
2430	int ret = -EINTR;
2431	uint64_t btree_ptr = b->key.ptr[0];
2432	unsigned long seq = b->seq;
2433	struct keylist insert;
2434	bool upgrade = op->lock == -1;
2435
2436	bch_keylist_init(&insert);
2437
2438	if (upgrade) {
2439		rw_unlock(false, b);
2440		rw_lock(true, b, b->level);
2441
2442		if (b->key.ptr[0] != btree_ptr ||
2443		    b->seq != seq + 1) {
2444			op->lock = b->level;
2445			goto out;
2446		}
2447	}
2448
2449	SET_KEY_PTRS(check_key, 1);
2450	get_random_bytes(&check_key->ptr[0], sizeof(uint64_t));
2451
2452	SET_PTR_DEV(check_key, 0, PTR_CHECK_DEV);
2453
2454	bch_keylist_add(&insert, check_key);
2455
2456	ret = bch_btree_insert_node(b, op, &insert, NULL, NULL);
2457
2458	BUG_ON(!ret && !bch_keylist_empty(&insert));
2459out:
2460	if (upgrade)
2461		downgrade_write(&b->lock);
2462	return ret;
2463}
2464
2465struct btree_insert_op {
2466	struct btree_op	op;
2467	struct keylist	*keys;
2468	atomic_t	*journal_ref;
2469	struct bkey	*replace_key;
2470};
2471
2472static int btree_insert_fn(struct btree_op *b_op, struct btree *b)
2473{
2474	struct btree_insert_op *op = container_of(b_op,
2475					struct btree_insert_op, op);
2476
2477	int ret = bch_btree_insert_node(b, &op->op, op->keys,
2478					op->journal_ref, op->replace_key);
2479	if (ret && !bch_keylist_empty(op->keys))
2480		return ret;
2481	else
2482		return MAP_DONE;
2483}
2484
2485int bch_btree_insert(struct cache_set *c, struct keylist *keys,
2486		     atomic_t *journal_ref, struct bkey *replace_key)
2487{
2488	struct btree_insert_op op;
2489	int ret = 0;
2490
2491	BUG_ON(current->bio_list);
2492	BUG_ON(bch_keylist_empty(keys));
2493
2494	bch_btree_op_init(&op.op, 0);
2495	op.keys		= keys;
2496	op.journal_ref	= journal_ref;
2497	op.replace_key	= replace_key;
2498
2499	while (!ret && !bch_keylist_empty(keys)) {
2500		op.op.lock = 0;
2501		ret = bch_btree_map_leaf_nodes(&op.op, c,
2502					       &START_KEY(keys->keys),
2503					       btree_insert_fn);
2504	}
2505
2506	if (ret) {
2507		struct bkey *k;
2508
2509		pr_err("error %i\n", ret);
2510
2511		while ((k = bch_keylist_pop(keys)))
2512			bkey_put(c, k);
2513	} else if (op.op.insert_collision)
2514		ret = -ESRCH;
2515
2516	return ret;
2517}
2518
2519void bch_btree_set_root(struct btree *b)
2520{
2521	unsigned int i;
2522	struct closure cl;
2523
2524	closure_init_stack(&cl);
2525
2526	trace_bcache_btree_set_root(b);
2527
2528	BUG_ON(!b->written);
2529
2530	for (i = 0; i < KEY_PTRS(&b->key); i++)
2531		BUG_ON(PTR_BUCKET(b->c, &b->key, i)->prio != BTREE_PRIO);
2532
2533	mutex_lock(&b->c->bucket_lock);
2534	list_del_init(&b->list);
2535	mutex_unlock(&b->c->bucket_lock);
2536
2537	b->c->root = b;
2538
2539	bch_journal_meta(b->c, &cl);
2540	closure_sync(&cl);
2541}
2542
2543/* Map across nodes or keys */
2544
2545static int bch_btree_map_nodes_recurse(struct btree *b, struct btree_op *op,
2546				       struct bkey *from,
2547				       btree_map_nodes_fn *fn, int flags)
2548{
2549	int ret = MAP_CONTINUE;
2550
2551	if (b->level) {
2552		struct bkey *k;
2553		struct btree_iter_stack iter;
2554
2555		bch_btree_iter_stack_init(&b->keys, &iter, from);
2556
2557		while ((k = bch_btree_iter_next_filter(&iter.iter, &b->keys,
2558						       bch_ptr_bad))) {
2559			ret = bcache_btree(map_nodes_recurse, k, b,
2560				    op, from, fn, flags);
2561			from = NULL;
2562
2563			if (ret != MAP_CONTINUE)
2564				return ret;
2565		}
2566	}
2567
2568	if (!b->level || flags == MAP_ALL_NODES)
2569		ret = fn(op, b);
2570
2571	return ret;
2572}
2573
2574int __bch_btree_map_nodes(struct btree_op *op, struct cache_set *c,
2575			  struct bkey *from, btree_map_nodes_fn *fn, int flags)
2576{
2577	return bcache_btree_root(map_nodes_recurse, c, op, from, fn, flags);
2578}
2579
2580int bch_btree_map_keys_recurse(struct btree *b, struct btree_op *op,
2581				      struct bkey *from, btree_map_keys_fn *fn,
2582				      int flags)
2583{
2584	int ret = MAP_CONTINUE;
2585	struct bkey *k;
2586	struct btree_iter_stack iter;
2587
2588	bch_btree_iter_stack_init(&b->keys, &iter, from);
2589
2590	while ((k = bch_btree_iter_next_filter(&iter.iter, &b->keys,
2591					       bch_ptr_bad))) {
2592		ret = !b->level
2593			? fn(op, b, k)
2594			: bcache_btree(map_keys_recurse, k,
2595				       b, op, from, fn, flags);
2596		from = NULL;
2597
2598		if (ret != MAP_CONTINUE)
2599			return ret;
2600	}
2601
2602	if (!b->level && (flags & MAP_END_KEY))
2603		ret = fn(op, b, &KEY(KEY_INODE(&b->key),
2604				     KEY_OFFSET(&b->key), 0));
2605
2606	return ret;
2607}
2608
2609int bch_btree_map_keys(struct btree_op *op, struct cache_set *c,
2610		       struct bkey *from, btree_map_keys_fn *fn, int flags)
2611{
2612	return bcache_btree_root(map_keys_recurse, c, op, from, fn, flags);
2613}
2614
2615/* Keybuf code */
2616
2617static inline int keybuf_cmp(struct keybuf_key *l, struct keybuf_key *r)
2618{
2619	/* Overlapping keys compare equal */
2620	if (bkey_cmp(&l->key, &START_KEY(&r->key)) <= 0)
2621		return -1;
2622	if (bkey_cmp(&START_KEY(&l->key), &r->key) >= 0)
2623		return 1;
2624	return 0;
2625}
2626
2627static inline int keybuf_nonoverlapping_cmp(struct keybuf_key *l,
2628					    struct keybuf_key *r)
2629{
2630	return clamp_t(int64_t, bkey_cmp(&l->key, &r->key), -1, 1);
2631}
2632
2633struct refill {
2634	struct btree_op	op;
2635	unsigned int	nr_found;
2636	struct keybuf	*buf;
2637	struct bkey	*end;
2638	keybuf_pred_fn	*pred;
2639};
2640
2641static int refill_keybuf_fn(struct btree_op *op, struct btree *b,
2642			    struct bkey *k)
2643{
2644	struct refill *refill = container_of(op, struct refill, op);
2645	struct keybuf *buf = refill->buf;
2646	int ret = MAP_CONTINUE;
2647
2648	if (bkey_cmp(k, refill->end) > 0) {
2649		ret = MAP_DONE;
2650		goto out;
2651	}
2652
2653	if (!KEY_SIZE(k)) /* end key */
2654		goto out;
2655
2656	if (refill->pred(buf, k)) {
2657		struct keybuf_key *w;
2658
2659		spin_lock(&buf->lock);
2660
2661		w = array_alloc(&buf->freelist);
2662		if (!w) {
2663			spin_unlock(&buf->lock);
2664			return MAP_DONE;
2665		}
2666
2667		w->private = NULL;
2668		bkey_copy(&w->key, k);
2669
2670		if (RB_INSERT(&buf->keys, w, node, keybuf_cmp))
2671			array_free(&buf->freelist, w);
2672		else
2673			refill->nr_found++;
2674
2675		if (array_freelist_empty(&buf->freelist))
2676			ret = MAP_DONE;
2677
2678		spin_unlock(&buf->lock);
2679	}
2680out:
2681	buf->last_scanned = *k;
2682	return ret;
2683}
2684
2685void bch_refill_keybuf(struct cache_set *c, struct keybuf *buf,
2686		       struct bkey *end, keybuf_pred_fn *pred)
2687{
2688	struct bkey start = buf->last_scanned;
2689	struct refill refill;
2690
2691	cond_resched();
2692
2693	bch_btree_op_init(&refill.op, -1);
2694	refill.nr_found	= 0;
2695	refill.buf	= buf;
2696	refill.end	= end;
2697	refill.pred	= pred;
2698
2699	bch_btree_map_keys(&refill.op, c, &buf->last_scanned,
2700			   refill_keybuf_fn, MAP_END_KEY);
2701
2702	trace_bcache_keyscan(refill.nr_found,
2703			     KEY_INODE(&start), KEY_OFFSET(&start),
2704			     KEY_INODE(&buf->last_scanned),
2705			     KEY_OFFSET(&buf->last_scanned));
2706
2707	spin_lock(&buf->lock);
2708
2709	if (!RB_EMPTY_ROOT(&buf->keys)) {
2710		struct keybuf_key *w;
2711
2712		w = RB_FIRST(&buf->keys, struct keybuf_key, node);
2713		buf->start	= START_KEY(&w->key);
2714
2715		w = RB_LAST(&buf->keys, struct keybuf_key, node);
2716		buf->end	= w->key;
2717	} else {
2718		buf->start	= MAX_KEY;
2719		buf->end	= MAX_KEY;
2720	}
2721
2722	spin_unlock(&buf->lock);
2723}
2724
2725static void __bch_keybuf_del(struct keybuf *buf, struct keybuf_key *w)
2726{
2727	rb_erase(&w->node, &buf->keys);
2728	array_free(&buf->freelist, w);
2729}
2730
2731void bch_keybuf_del(struct keybuf *buf, struct keybuf_key *w)
2732{
2733	spin_lock(&buf->lock);
2734	__bch_keybuf_del(buf, w);
2735	spin_unlock(&buf->lock);
2736}
2737
2738bool bch_keybuf_check_overlapping(struct keybuf *buf, struct bkey *start,
2739				  struct bkey *end)
2740{
2741	bool ret = false;
2742	struct keybuf_key *p, *w, s;
2743
2744	s.key = *start;
2745
2746	if (bkey_cmp(end, &buf->start) <= 0 ||
2747	    bkey_cmp(start, &buf->end) >= 0)
2748		return false;
2749
2750	spin_lock(&buf->lock);
2751	w = RB_GREATER(&buf->keys, s, node, keybuf_nonoverlapping_cmp);
2752
2753	while (w && bkey_cmp(&START_KEY(&w->key), end) < 0) {
2754		p = w;
2755		w = RB_NEXT(w, node);
2756
2757		if (p->private)
2758			ret = true;
2759		else
2760			__bch_keybuf_del(buf, p);
2761	}
2762
2763	spin_unlock(&buf->lock);
2764	return ret;
2765}
2766
2767struct keybuf_key *bch_keybuf_next(struct keybuf *buf)
2768{
2769	struct keybuf_key *w;
2770
2771	spin_lock(&buf->lock);
2772
2773	w = RB_FIRST(&buf->keys, struct keybuf_key, node);
2774
2775	while (w && w->private)
2776		w = RB_NEXT(w, node);
2777
2778	if (w)
2779		w->private = ERR_PTR(-EINTR);
2780
2781	spin_unlock(&buf->lock);
2782	return w;
2783}
2784
2785struct keybuf_key *bch_keybuf_next_rescan(struct cache_set *c,
2786					  struct keybuf *buf,
2787					  struct bkey *end,
2788					  keybuf_pred_fn *pred)
2789{
2790	struct keybuf_key *ret;
2791
2792	while (1) {
2793		ret = bch_keybuf_next(buf);
2794		if (ret)
2795			break;
2796
2797		if (bkey_cmp(&buf->last_scanned, end) >= 0) {
2798			pr_debug("scan finished\n");
2799			break;
2800		}
2801
2802		bch_refill_keybuf(c, buf, end, pred);
2803	}
2804
2805	return ret;
2806}
2807
2808void bch_keybuf_init(struct keybuf *buf)
2809{
2810	buf->last_scanned	= MAX_KEY;
2811	buf->keys		= RB_ROOT;
2812
2813	spin_lock_init(&buf->lock);
2814	array_allocator_init(&buf->freelist);
2815}
2816
2817void bch_btree_exit(void)
2818{
2819	if (btree_io_wq)
2820		destroy_workqueue(btree_io_wq);
2821}
2822
2823int __init bch_btree_init(void)
2824{
2825	btree_io_wq = alloc_workqueue("bch_btree_io",
2826				      WQ_MEM_RECLAIM | WQ_PERCPU, 0);
2827	if (!btree_io_wq)
2828		return -ENOMEM;
2829
2830	return 0;
2831}