git.sesse.net Git - bcachefs-tools-debian/blob - libbcachefs/alloc_background.c

   1 // SPDX-License-Identifier: GPL-2.0
   2 #include "bcachefs.h"
   3 #include "alloc_background.h"
   4 #include "alloc_foreground.h"
   5 #include "btree_cache.h"
   6 #include "btree_io.h"
   7 #include "btree_key_cache.h"
   8 #include "btree_update.h"
   9 #include "btree_update_interior.h"
  10 #include "btree_gc.h"
  11 #include "buckets.h"
  12 #include "clock.h"
  13 #include "debug.h"
  14 #include "ec.h"
  15 #include "error.h"
  16 #include "recovery.h"
  17
  18 #include <linux/kthread.h>
  19 #include <linux/math64.h>
  20 #include <linux/random.h>
  21 #include <linux/rculist.h>
  22 #include <linux/rcupdate.h>
  23 #include <linux/sched/task.h>
  24 #include <linux/sort.h>
  25 #include <trace/events/bcachefs.h>
  26
  27 static const char * const bch2_alloc_field_names[] = {
  28 #define x(name, bytes) #name,
  29         BCH_ALLOC_FIELDS()
  30 #undef x
  31         NULL
  32 };
  33
  34 static void bch2_recalc_oldest_io(struct bch_fs *, struct bch_dev *, int);
  35
  36 /* Ratelimiting/PD controllers */
  37
  38 static void pd_controllers_update(struct work_struct *work)
  39 {
  40         struct bch_fs *c = container_of(to_delayed_work(work),
  41                                            struct bch_fs,
  42                                            pd_controllers_update);
  43         struct bch_dev *ca;
  44         unsigned i;
  45
  46         for_each_member_device(ca, c, i) {
  47                 struct bch_dev_usage stats = bch2_dev_usage_read(c, ca);
  48
  49                 u64 free = bucket_to_sector(ca,
  50                                 __dev_buckets_free(ca, stats)) << 9;
  51                 /*
  52                  * Bytes of internal fragmentation, which can be
  53                  * reclaimed by copy GC
  54                  */
  55                 s64 fragmented = (bucket_to_sector(ca,
  56                                         stats.buckets[BCH_DATA_USER] +
  57                                         stats.buckets[BCH_DATA_CACHED]) -
  58                                   (stats.sectors[BCH_DATA_USER] +
  59                                    stats.sectors[BCH_DATA_CACHED])) << 9;
  60
  61                 fragmented = max(0LL, fragmented);
  62
  63                 bch2_pd_controller_update(&ca->copygc_pd,
  64                                          free, fragmented, -1);
  65         }
  66
  67         schedule_delayed_work(&c->pd_controllers_update,
  68                               c->pd_controllers_update_seconds * HZ);
  69 }
  70
  71 /* Persistent alloc info: */
  72
  73 static inline u64 get_alloc_field(const struct bch_alloc *a,
  74                                   const void **p, unsigned field)
  75 {
  76         unsigned bytes = BCH_ALLOC_FIELD_BYTES[field];
  77         u64 v;
  78
  79         if (!(a->fields & (1 << field)))
  80                 return 0;
  81
  82         switch (bytes) {
  83         case 1:
  84                 v = *((const u8 *) *p);
  85                 break;
  86         case 2:
  87                 v = le16_to_cpup(*p);
  88                 break;
  89         case 4:
  90                 v = le32_to_cpup(*p);
  91                 break;
  92         case 8:
  93                 v = le64_to_cpup(*p);
  94                 break;
  95         default:
  96                 BUG();
  97         }
  98
  99         *p += bytes;
 100         return v;
 101 }
 102
 103 static inline void put_alloc_field(struct bkey_i_alloc *a, void **p,
 104                                    unsigned field, u64 v)
 105 {
 106         unsigned bytes = BCH_ALLOC_FIELD_BYTES[field];
 107
 108         if (!v)
 109                 return;
 110
 111         a->v.fields |= 1 << field;
 112
 113         switch (bytes) {
 114         case 1:
 115                 *((u8 *) *p) = v;
 116                 break;
 117         case 2:
 118                 *((__le16 *) *p) = cpu_to_le16(v);
 119                 break;
 120         case 4:
 121                 *((__le32 *) *p) = cpu_to_le32(v);
 122                 break;
 123         case 8:
 124                 *((__le64 *) *p) = cpu_to_le64(v);
 125                 break;
 126         default:
 127                 BUG();
 128         }
 129
 130         *p += bytes;
 131 }
 132
 133 struct bkey_alloc_unpacked bch2_alloc_unpack(struct bkey_s_c k)
 134 {
 135         struct bkey_alloc_unpacked ret = { .gen = 0 };
 136
 137         if (k.k->type == KEY_TYPE_alloc) {
 138                 const struct bch_alloc *a = bkey_s_c_to_alloc(k).v;
 139                 const void *d = a->data;
 140                 unsigned idx = 0;
 141
 142                 ret.gen = a->gen;
 143
 144 #define x(_name, _bits) ret._name = get_alloc_field(a, &d, idx++);
 145                 BCH_ALLOC_FIELDS()
 146 #undef  x
 147         }
 148         return ret;
 149 }
 150
 151 void bch2_alloc_pack(struct bkey_i_alloc *dst,
 152                      const struct bkey_alloc_unpacked src)
 153 {
 154         unsigned idx = 0;
 155         void *d = dst->v.data;
 156         unsigned bytes;
 157
 158         dst->v.fields   = 0;
 159         dst->v.gen      = src.gen;
 160
 161 #define x(_name, _bits) put_alloc_field(dst, &d, idx++, src._name);
 162         BCH_ALLOC_FIELDS()
 163 #undef  x
 164
 165         bytes = (void *) d - (void *) &dst->v;
 166         set_bkey_val_bytes(&dst->k, bytes);
 167         memset_u64s_tail(&dst->v, 0, bytes);
 168 }
 169
 170 static unsigned bch_alloc_val_u64s(const struct bch_alloc *a)
 171 {
 172         unsigned i, bytes = offsetof(struct bch_alloc, data);
 173
 174         for (i = 0; i < ARRAY_SIZE(BCH_ALLOC_FIELD_BYTES); i++)
 175                 if (a->fields & (1 << i))
 176                         bytes += BCH_ALLOC_FIELD_BYTES[i];
 177
 178         return DIV_ROUND_UP(bytes, sizeof(u64));
 179 }
 180
 181 const char *bch2_alloc_invalid(const struct bch_fs *c, struct bkey_s_c k)
 182 {
 183         struct bkey_s_c_alloc a = bkey_s_c_to_alloc(k);
 184
 185         if (k.k->p.inode >= c->sb.nr_devices ||
 186             !c->devs[k.k->p.inode])
 187                 return "invalid device";
 188
 189         /* allow for unknown fields */
 190         if (bkey_val_u64s(a.k) < bch_alloc_val_u64s(a.v))
 191                 return "incorrect value size";
 192
 193         return NULL;
 194 }
 195
 196 void bch2_alloc_to_text(struct printbuf *out, struct bch_fs *c,
 197                         struct bkey_s_c k)
 198 {
 199         struct bkey_s_c_alloc a = bkey_s_c_to_alloc(k);
 200         const void *d = a.v->data;
 201         unsigned i;
 202
 203         pr_buf(out, "gen %u", a.v->gen);
 204
 205         for (i = 0; i < BCH_ALLOC_FIELD_NR; i++)
 206                 if (a.v->fields & (1 << i))
 207                         pr_buf(out, " %s %llu",
 208                                bch2_alloc_field_names[i],
 209                                get_alloc_field(a.v, &d, i));
 210 }
 211
 212 static int bch2_alloc_read_fn(struct bch_fs *c, enum btree_id id,
 213                               unsigned level, struct bkey_s_c k)
 214 {
 215         if (!level)
 216                 bch2_mark_key(c, k, 0, 0, NULL, 0,
 217                               BTREE_TRIGGER_ALLOC_READ|
 218                               BTREE_TRIGGER_NOATOMIC);
 219
 220         return 0;
 221 }
 222
 223 int bch2_alloc_read(struct bch_fs *c, struct journal_keys *journal_keys)
 224 {
 225         struct bch_dev *ca;
 226         unsigned i;
 227         int ret = 0;
 228
 229         ret = bch2_btree_and_journal_walk(c, journal_keys, BTREE_ID_ALLOC,
 230                                           NULL, bch2_alloc_read_fn);
 231         if (ret) {
 232                 bch_err(c, "error reading alloc info: %i", ret);
 233                 return ret;
 234         }
 235
 236         percpu_down_write(&c->mark_lock);
 237         bch2_dev_usage_from_buckets(c);
 238         percpu_up_write(&c->mark_lock);
 239
 240         mutex_lock(&c->bucket_clock[READ].lock);
 241         for_each_member_device(ca, c, i) {
 242                 down_read(&ca->bucket_lock);
 243                 bch2_recalc_oldest_io(c, ca, READ);
 244                 up_read(&ca->bucket_lock);
 245         }
 246         mutex_unlock(&c->bucket_clock[READ].lock);
 247
 248         mutex_lock(&c->bucket_clock[WRITE].lock);
 249         for_each_member_device(ca, c, i) {
 250                 down_read(&ca->bucket_lock);
 251                 bch2_recalc_oldest_io(c, ca, WRITE);
 252                 up_read(&ca->bucket_lock);
 253         }
 254         mutex_unlock(&c->bucket_clock[WRITE].lock);
 255
 256         return 0;
 257 }
 258
 259 enum alloc_write_ret {
 260         ALLOC_WROTE,
 261         ALLOC_NOWROTE,
 262         ALLOC_END,
 263 };
 264
 265 static int bch2_alloc_write_key(struct btree_trans *trans,
 266                                 struct btree_iter *iter,
 267                                 unsigned flags)
 268 {
 269         struct bch_fs *c = trans->c;
 270         struct bkey_s_c k;
 271         struct bch_dev *ca;
 272         struct bucket_array *ba;
 273         struct bucket *g;
 274         struct bucket_mark m;
 275         struct bkey_alloc_unpacked old_u, new_u;
 276         __BKEY_PADDED(k, 8) alloc_key; /* hack: */
 277         struct bkey_i_alloc *a;
 278         int ret;
 279 retry:
 280         bch2_trans_begin(trans);
 281
 282         ret = bch2_btree_key_cache_flush(trans,
 283                         BTREE_ID_ALLOC, iter->pos);
 284         if (ret)
 285                 goto err;
 286
 287         k = bch2_btree_iter_peek_slot(iter);
 288         ret = bkey_err(k);
 289         if (ret)
 290                 goto err;
 291
 292         old_u = bch2_alloc_unpack(k);
 293
 294         if (iter->pos.inode >= c->sb.nr_devices ||
 295             !c->devs[iter->pos.inode])
 296                 return ALLOC_END;
 297
 298         percpu_down_read(&c->mark_lock);
 299         ca      = bch_dev_bkey_exists(c, iter->pos.inode);
 300         ba      = bucket_array(ca);
 301
 302         if (iter->pos.offset >= ba->nbuckets) {
 303                 percpu_up_read(&c->mark_lock);
 304                 return ALLOC_END;
 305         }
 306
 307         g       = &ba->b[iter->pos.offset];
 308         m       = READ_ONCE(g->mark);
 309         new_u   = alloc_mem_to_key(g, m);
 310         percpu_up_read(&c->mark_lock);
 311
 312         if (!bkey_alloc_unpacked_cmp(old_u, new_u))
 313                 return ALLOC_NOWROTE;
 314
 315         a = bkey_alloc_init(&alloc_key.k);
 316         a->k.p = iter->pos;
 317         bch2_alloc_pack(a, new_u);
 318
 319         bch2_trans_update(trans, iter, &a->k_i,
 320                           BTREE_TRIGGER_NORUN);
 321         ret = bch2_trans_commit(trans, NULL, NULL,
 322                                 BTREE_INSERT_NOFAIL|
 323                                 BTREE_INSERT_USE_RESERVE|
 324                                 flags);
 325 err:
 326         if (ret == -EINTR)
 327                 goto retry;
 328         return ret;
 329 }
 330
 331 int bch2_alloc_write(struct bch_fs *c, unsigned flags, bool *wrote)
 332 {
 333         struct btree_trans trans;
 334         struct btree_iter *iter;
 335         struct bch_dev *ca;
 336         unsigned i;
 337         int ret = 0;
 338
 339         BUG_ON(BKEY_ALLOC_VAL_U64s_MAX > 8);
 340
 341         bch2_trans_init(&trans, c, BTREE_ITER_MAX, 0);
 342
 343         iter = bch2_trans_get_iter(&trans, BTREE_ID_ALLOC, POS_MIN,
 344                                    BTREE_ITER_SLOTS|BTREE_ITER_INTENT);
 345
 346         for_each_rw_member(ca, c, i) {
 347                 unsigned first_bucket;
 348
 349                 percpu_down_read(&c->mark_lock);
 350                 first_bucket = bucket_array(ca)->first_bucket;
 351                 percpu_up_read(&c->mark_lock);
 352
 353                 bch2_btree_iter_set_pos(iter, POS(i, first_bucket));
 354
 355                 while (1) {
 356                         ret = bch2_alloc_write_key(&trans, iter, flags);
 357                         if (ret < 0 || ret == ALLOC_END)
 358                                 break;
 359                         if (ret == ALLOC_WROTE)
 360                                 *wrote = true;
 361                         bch2_btree_iter_next_slot(iter);
 362                 }
 363
 364                 if (ret < 0) {
 365                         percpu_ref_put(&ca->io_ref);
 366                         break;
 367                 }
 368         }
 369
 370         bch2_trans_exit(&trans);
 371
 372         return ret < 0 ? ret : 0;
 373 }
 374
 375 /* Bucket IO clocks: */
 376
 377 static void bch2_recalc_oldest_io(struct bch_fs *c, struct bch_dev *ca, int rw)
 378 {
 379         struct bucket_clock *clock = &c->bucket_clock[rw];
 380         struct bucket_array *buckets = bucket_array(ca);
 381         struct bucket *g;
 382         u16 max_last_io = 0;
 383         unsigned i;
 384
 385         lockdep_assert_held(&c->bucket_clock[rw].lock);
 386
 387         /* Recalculate max_last_io for this device: */
 388         for_each_bucket(g, buckets)
 389                 max_last_io = max(max_last_io, bucket_last_io(c, g, rw));
 390
 391         ca->max_last_bucket_io[rw] = max_last_io;
 392
 393         /* Recalculate global max_last_io: */
 394         max_last_io = 0;
 395
 396         for_each_member_device(ca, c, i)
 397                 max_last_io = max(max_last_io, ca->max_last_bucket_io[rw]);
 398
 399         clock->max_last_io = max_last_io;
 400 }
 401
 402 static void bch2_rescale_bucket_io_times(struct bch_fs *c, int rw)
 403 {
 404         struct bucket_clock *clock = &c->bucket_clock[rw];
 405         struct bucket_array *buckets;
 406         struct bch_dev *ca;
 407         struct bucket *g;
 408         unsigned i;
 409
 410         trace_rescale_prios(c);
 411
 412         for_each_member_device(ca, c, i) {
 413                 down_read(&ca->bucket_lock);
 414                 buckets = bucket_array(ca);
 415
 416                 for_each_bucket(g, buckets)
 417                         g->io_time[rw] = clock->hand -
 418                         bucket_last_io(c, g, rw) / 2;
 419
 420                 bch2_recalc_oldest_io(c, ca, rw);
 421
 422                 up_read(&ca->bucket_lock);
 423         }
 424 }
 425
 426 static inline u64 bucket_clock_freq(u64 capacity)
 427 {
 428         return max(capacity >> 10, 2028ULL);
 429 }
 430
 431 static void bch2_inc_clock_hand(struct io_timer *timer)
 432 {
 433         struct bucket_clock *clock = container_of(timer,
 434                                                 struct bucket_clock, rescale);
 435         struct bch_fs *c = container_of(clock,
 436                                         struct bch_fs, bucket_clock[clock->rw]);
 437         struct bch_dev *ca;
 438         u64 capacity;
 439         unsigned i;
 440
 441         mutex_lock(&clock->lock);
 442
 443         /* if clock cannot be advanced more, rescale prio */
 444         if (clock->max_last_io >= U16_MAX - 2)
 445                 bch2_rescale_bucket_io_times(c, clock->rw);
 446
 447         BUG_ON(clock->max_last_io >= U16_MAX - 2);
 448
 449         for_each_member_device(ca, c, i)
 450                 ca->max_last_bucket_io[clock->rw]++;
 451         clock->max_last_io++;
 452         clock->hand++;
 453
 454         mutex_unlock(&clock->lock);
 455
 456         capacity = READ_ONCE(c->capacity);
 457
 458         if (!capacity)
 459                 return;
 460
 461         /*
 462          * we only increment when 0.1% of the filesystem capacity has been read
 463          * or written too, this determines if it's time
 464          *
 465          * XXX: we shouldn't really be going off of the capacity of devices in
 466          * RW mode (that will be 0 when we're RO, yet we can still service
 467          * reads)
 468          */
 469         timer->expire += bucket_clock_freq(capacity);
 470
 471         bch2_io_timer_add(&c->io_clock[clock->rw], timer);
 472 }
 473
 474 static void bch2_bucket_clock_init(struct bch_fs *c, int rw)
 475 {
 476         struct bucket_clock *clock = &c->bucket_clock[rw];
 477
 478         clock->hand             = 1;
 479         clock->rw               = rw;
 480         clock->rescale.fn       = bch2_inc_clock_hand;
 481         clock->rescale.expire   = bucket_clock_freq(c->capacity);
 482         mutex_init(&clock->lock);
 483 }
 484
 485 /* Background allocator thread: */
 486
 487 /*
 488  * Scans for buckets to be invalidated, invalidates them, rewrites prios/gens
 489  * (marking them as invalidated on disk), then optionally issues discard
 490  * commands to the newly free buckets, then puts them on the various freelists.
 491  */
 492
 493 #define BUCKET_GC_GEN_MAX       96U
 494
 495 /**
 496  * wait_buckets_available - wait on reclaimable buckets
 497  *
 498  * If there aren't enough available buckets to fill up free_inc, wait until
 499  * there are.
 500  */
 501 static int wait_buckets_available(struct bch_fs *c, struct bch_dev *ca)
 502 {
 503         unsigned long gc_count = c->gc_count;
 504         int ret = 0;
 505
 506         ca->allocator_state = ALLOCATOR_BLOCKED;
 507         closure_wake_up(&c->freelist_wait);
 508
 509         while (1) {
 510                 set_current_state(TASK_INTERRUPTIBLE);
 511                 if (kthread_should_stop()) {
 512                         ret = 1;
 513                         break;
 514                 }
 515
 516                 if (gc_count != c->gc_count)
 517                         ca->inc_gen_really_needs_gc = 0;
 518
 519                 if ((ssize_t) (dev_buckets_available(c, ca) -
 520                                ca->inc_gen_really_needs_gc) >=
 521                     (ssize_t) fifo_free(&ca->free_inc))
 522                         break;
 523
 524                 up_read(&c->gc_lock);
 525                 schedule();
 526                 try_to_freeze();
 527                 down_read(&c->gc_lock);
 528         }
 529
 530         __set_current_state(TASK_RUNNING);
 531         ca->allocator_state = ALLOCATOR_RUNNING;
 532         closure_wake_up(&c->freelist_wait);
 533
 534         return ret;
 535 }
 536
 537 static bool bch2_can_invalidate_bucket(struct bch_dev *ca,
 538                                        size_t bucket,
 539                                        struct bucket_mark mark)
 540 {
 541         u8 gc_gen;
 542
 543         if (!is_available_bucket(mark))
 544                 return false;
 545
 546         if (ca->buckets_nouse &&
 547             test_bit(bucket, ca->buckets_nouse))
 548                 return false;
 549
 550         gc_gen = bucket_gc_gen(ca, bucket);
 551
 552         if (gc_gen >= BUCKET_GC_GEN_MAX / 2)
 553                 ca->inc_gen_needs_gc++;
 554
 555         if (gc_gen >= BUCKET_GC_GEN_MAX)
 556                 ca->inc_gen_really_needs_gc++;
 557
 558         return gc_gen < BUCKET_GC_GEN_MAX;
 559 }
 560
 561 /*
 562  * Determines what order we're going to reuse buckets, smallest bucket_key()
 563  * first.
 564  *
 565  *
 566  * - We take into account the read prio of the bucket, which gives us an
 567  *   indication of how hot the data is -- we scale the prio so that the prio
 568  *   farthest from the clock is worth 1/8th of the closest.
 569  *
 570  * - The number of sectors of cached data in the bucket, which gives us an
 571  *   indication of the cost in cache misses this eviction will cause.
 572  *
 573  * - If hotness * sectors used compares equal, we pick the bucket with the
 574  *   smallest bucket_gc_gen() - since incrementing the same bucket's generation
 575  *   number repeatedly forces us to run mark and sweep gc to avoid generation
 576  *   number wraparound.
 577  */
 578
 579 static unsigned long bucket_sort_key(struct bch_fs *c, struct bch_dev *ca,
 580                                      size_t b, struct bucket_mark m)
 581 {
 582         unsigned last_io = bucket_last_io(c, bucket(ca, b), READ);
 583         unsigned max_last_io = ca->max_last_bucket_io[READ];
 584
 585         /*
 586          * Time since last read, scaled to [0, 8) where larger value indicates
 587          * more recently read data:
 588          */
 589         unsigned long hotness = (max_last_io - last_io) * 7 / max_last_io;
 590
 591         /* How much we want to keep the data in this bucket: */
 592         unsigned long data_wantness =
 593                 (hotness + 1) * bucket_sectors_used(m);
 594
 595         unsigned long needs_journal_commit =
 596                 bucket_needs_journal_commit(m, c->journal.last_seq_ondisk);
 597
 598         return  (data_wantness << 9) |
 599                 (needs_journal_commit << 8) |
 600                 (bucket_gc_gen(ca, b) / 16);
 601 }
 602
 603 static inline int bucket_alloc_cmp(alloc_heap *h,
 604                                    struct alloc_heap_entry l,
 605                                    struct alloc_heap_entry r)
 606 {
 607         return  cmp_int(l.key, r.key) ?:
 608                 cmp_int(r.nr, l.nr) ?:
 609                 cmp_int(l.bucket, r.bucket);
 610 }
 611
 612 static inline int bucket_idx_cmp(const void *_l, const void *_r)
 613 {
 614         const struct alloc_heap_entry *l = _l, *r = _r;
 615
 616         return cmp_int(l->bucket, r->bucket);
 617 }
 618
 619 static void find_reclaimable_buckets_lru(struct bch_fs *c, struct bch_dev *ca)
 620 {
 621         struct bucket_array *buckets;
 622         struct alloc_heap_entry e = { 0 };
 623         size_t b, i, nr = 0;
 624
 625         ca->alloc_heap.used = 0;
 626
 627         mutex_lock(&c->bucket_clock[READ].lock);
 628         down_read(&ca->bucket_lock);
 629
 630         buckets = bucket_array(ca);
 631
 632         bch2_recalc_oldest_io(c, ca, READ);
 633
 634         /*
 635          * Find buckets with lowest read priority, by building a maxheap sorted
 636          * by read priority and repeatedly replacing the maximum element until
 637          * all buckets have been visited.
 638          */
 639         for (b = ca->mi.first_bucket; b < ca->mi.nbuckets; b++) {
 640                 struct bucket_mark m = READ_ONCE(buckets->b[b].mark);
 641                 unsigned long key = bucket_sort_key(c, ca, b, m);
 642
 643                 if (!bch2_can_invalidate_bucket(ca, b, m))
 644                         continue;
 645
 646                 if (e.nr && e.bucket + e.nr == b && e.key == key) {
 647                         e.nr++;
 648                 } else {
 649                         if (e.nr)
 650                                 heap_add_or_replace(&ca->alloc_heap, e,
 651                                         -bucket_alloc_cmp, NULL);
 652
 653                         e = (struct alloc_heap_entry) {
 654                                 .bucket = b,
 655                                 .nr     = 1,
 656                                 .key    = key,
 657                         };
 658                 }
 659
 660                 cond_resched();
 661         }
 662
 663         if (e.nr)
 664                 heap_add_or_replace(&ca->alloc_heap, e,
 665                                 -bucket_alloc_cmp, NULL);
 666
 667         for (i = 0; i < ca->alloc_heap.used; i++)
 668                 nr += ca->alloc_heap.data[i].nr;
 669
 670         while (nr - ca->alloc_heap.data[0].nr >= ALLOC_SCAN_BATCH(ca)) {
 671                 nr -= ca->alloc_heap.data[0].nr;
 672                 heap_pop(&ca->alloc_heap, e, -bucket_alloc_cmp, NULL);
 673         }
 674
 675         up_read(&ca->bucket_lock);
 676         mutex_unlock(&c->bucket_clock[READ].lock);
 677 }
 678
 679 static void find_reclaimable_buckets_fifo(struct bch_fs *c, struct bch_dev *ca)
 680 {
 681         struct bucket_array *buckets = bucket_array(ca);
 682         struct bucket_mark m;
 683         size_t b, start;
 684
 685         if (ca->fifo_last_bucket <  ca->mi.first_bucket ||
 686             ca->fifo_last_bucket >= ca->mi.nbuckets)
 687                 ca->fifo_last_bucket = ca->mi.first_bucket;
 688
 689         start = ca->fifo_last_bucket;
 690
 691         do {
 692                 ca->fifo_last_bucket++;
 693                 if (ca->fifo_last_bucket == ca->mi.nbuckets)
 694                         ca->fifo_last_bucket = ca->mi.first_bucket;
 695
 696                 b = ca->fifo_last_bucket;
 697                 m = READ_ONCE(buckets->b[b].mark);
 698
 699                 if (bch2_can_invalidate_bucket(ca, b, m)) {
 700                         struct alloc_heap_entry e = { .bucket = b, .nr = 1, };
 701
 702                         heap_add(&ca->alloc_heap, e, bucket_alloc_cmp, NULL);
 703                         if (heap_full(&ca->alloc_heap))
 704                                 break;
 705                 }
 706
 707                 cond_resched();
 708         } while (ca->fifo_last_bucket != start);
 709 }
 710
 711 static void find_reclaimable_buckets_random(struct bch_fs *c, struct bch_dev *ca)
 712 {
 713         struct bucket_array *buckets = bucket_array(ca);
 714         struct bucket_mark m;
 715         size_t checked, i;
 716
 717         for (checked = 0;
 718              checked < ca->mi.nbuckets / 2;
 719              checked++) {
 720                 size_t b = bch2_rand_range(ca->mi.nbuckets -
 721                                            ca->mi.first_bucket) +
 722                         ca->mi.first_bucket;
 723
 724                 m = READ_ONCE(buckets->b[b].mark);
 725
 726                 if (bch2_can_invalidate_bucket(ca, b, m)) {
 727                         struct alloc_heap_entry e = { .bucket = b, .nr = 1, };
 728
 729                         heap_add(&ca->alloc_heap, e, bucket_alloc_cmp, NULL);
 730                         if (heap_full(&ca->alloc_heap))
 731                                 break;
 732                 }
 733
 734                 cond_resched();
 735         }
 736
 737         sort(ca->alloc_heap.data,
 738              ca->alloc_heap.used,
 739              sizeof(ca->alloc_heap.data[0]),
 740              bucket_idx_cmp, NULL);
 741
 742         /* remove duplicates: */
 743         for (i = 0; i + 1 < ca->alloc_heap.used; i++)
 744                 if (ca->alloc_heap.data[i].bucket ==
 745                     ca->alloc_heap.data[i + 1].bucket)
 746                         ca->alloc_heap.data[i].nr = 0;
 747 }
 748
 749 static size_t find_reclaimable_buckets(struct bch_fs *c, struct bch_dev *ca)
 750 {
 751         size_t i, nr = 0;
 752
 753         ca->inc_gen_needs_gc                    = 0;
 754
 755         switch (ca->mi.replacement) {
 756         case CACHE_REPLACEMENT_LRU:
 757                 find_reclaimable_buckets_lru(c, ca);
 758                 break;
 759         case CACHE_REPLACEMENT_FIFO:
 760                 find_reclaimable_buckets_fifo(c, ca);
 761                 break;
 762         case CACHE_REPLACEMENT_RANDOM:
 763                 find_reclaimable_buckets_random(c, ca);
 764                 break;
 765         }
 766
 767         heap_resort(&ca->alloc_heap, bucket_alloc_cmp, NULL);
 768
 769         for (i = 0; i < ca->alloc_heap.used; i++)
 770                 nr += ca->alloc_heap.data[i].nr;
 771
 772         return nr;
 773 }
 774
 775 static inline long next_alloc_bucket(struct bch_dev *ca)
 776 {
 777         struct alloc_heap_entry e, *top = ca->alloc_heap.data;
 778
 779         while (ca->alloc_heap.used) {
 780                 if (top->nr) {
 781                         size_t b = top->bucket;
 782
 783                         top->bucket++;
 784                         top->nr--;
 785                         return b;
 786                 }
 787
 788                 heap_pop(&ca->alloc_heap, e, bucket_alloc_cmp, NULL);
 789         }
 790
 791         return -1;
 792 }
 793
 794 /*
 795  * returns sequence number of most recent journal entry that updated this
 796  * bucket:
 797  */
 798 static u64 bucket_journal_seq(struct bch_fs *c, struct bucket_mark m)
 799 {
 800         if (m.journal_seq_valid) {
 801                 u64 journal_seq = atomic64_read(&c->journal.seq);
 802                 u64 bucket_seq  = journal_seq;
 803
 804                 bucket_seq &= ~((u64) U16_MAX);
 805                 bucket_seq |= m.journal_seq;
 806
 807                 if (bucket_seq > journal_seq)
 808                         bucket_seq -= 1 << 16;
 809
 810                 return bucket_seq;
 811         } else {
 812                 return 0;
 813         }
 814 }
 815
 816 static int bch2_invalidate_one_bucket2(struct btree_trans *trans,
 817                                        struct bch_dev *ca,
 818                                        struct btree_iter *iter,
 819                                        u64 *journal_seq, unsigned flags)
 820 {
 821 #if 0
 822         __BKEY_PADDED(k, BKEY_ALLOC_VAL_U64s_MAX) alloc_key;
 823 #else
 824         /* hack: */
 825         __BKEY_PADDED(k, 8) alloc_key;
 826 #endif
 827         struct bch_fs *c = trans->c;
 828         struct bkey_i_alloc *a;
 829         struct bkey_alloc_unpacked u;
 830         struct bucket *g;
 831         struct bucket_mark m;
 832         bool invalidating_cached_data;
 833         size_t b;
 834         int ret = 0;
 835
 836         BUG_ON(!ca->alloc_heap.used ||
 837                !ca->alloc_heap.data[0].nr);
 838         b = ca->alloc_heap.data[0].bucket;
 839
 840         /* first, put on free_inc and mark as owned by allocator: */
 841         percpu_down_read(&c->mark_lock);
 842         spin_lock(&c->freelist_lock);
 843
 844         verify_not_on_freelist(c, ca, b);
 845
 846         BUG_ON(!fifo_push(&ca->free_inc, b));
 847
 848         g = bucket(ca, b);
 849         m = READ_ONCE(g->mark);
 850
 851         invalidating_cached_data = m.cached_sectors != 0;
 852
 853         /*
 854          * If we're not invalidating cached data, we only increment the bucket
 855          * gen in memory here, the incremented gen will be updated in the btree
 856          * by bch2_trans_mark_pointer():
 857          */
 858
 859         if (!invalidating_cached_data)
 860                 bch2_invalidate_bucket(c, ca, b, &m);
 861         else
 862                 bch2_mark_alloc_bucket(c, ca, b, true, gc_pos_alloc(c, NULL), 0);
 863
 864         spin_unlock(&c->freelist_lock);
 865         percpu_up_read(&c->mark_lock);
 866
 867         if (!invalidating_cached_data)
 868                 goto out;
 869
 870         /*
 871          * If the read-only path is trying to shut down, we can't be generating
 872          * new btree updates:
 873          */
 874         if (test_bit(BCH_FS_ALLOCATOR_STOPPING, &c->flags)) {
 875                 ret = 1;
 876                 goto out;
 877         }
 878
 879         BUG_ON(BKEY_ALLOC_VAL_U64s_MAX > 8);
 880
 881         bch2_btree_iter_set_pos(iter, POS(ca->dev_idx, b));
 882 retry:
 883         ret = bch2_btree_iter_traverse(iter);
 884         if (ret)
 885                 return ret;
 886
 887         percpu_down_read(&c->mark_lock);
 888         g = bucket(ca, iter->pos.offset);
 889         m = READ_ONCE(g->mark);
 890         u = alloc_mem_to_key(g, m);
 891
 892         percpu_up_read(&c->mark_lock);
 893
 894         invalidating_cached_data = u.cached_sectors != 0;
 895
 896         u.gen++;
 897         u.data_type     = 0;
 898         u.dirty_sectors = 0;
 899         u.cached_sectors = 0;
 900         u.read_time     = c->bucket_clock[READ].hand;
 901         u.write_time    = c->bucket_clock[WRITE].hand;
 902
 903         a = bkey_alloc_init(&alloc_key.k);
 904         a->k.p = iter->pos;
 905         bch2_alloc_pack(a, u);
 906
 907         bch2_trans_update(trans, iter, &a->k_i,
 908                           BTREE_TRIGGER_BUCKET_INVALIDATE);
 909
 910         /*
 911          * XXX:
 912          * when using deferred btree updates, we have journal reclaim doing
 913          * btree updates and thus requiring the allocator to make forward
 914          * progress, and here the allocator is requiring space in the journal -
 915          * so we need a journal pre-reservation:
 916          */
 917         ret = bch2_trans_commit(trans, NULL,
 918                                 invalidating_cached_data ? journal_seq : NULL,
 919                                 BTREE_INSERT_NOUNLOCK|
 920                                 BTREE_INSERT_NOCHECK_RW|
 921                                 BTREE_INSERT_NOFAIL|
 922                                 BTREE_INSERT_USE_RESERVE|
 923                                 BTREE_INSERT_USE_ALLOC_RESERVE|
 924                                 flags);
 925         if (ret == -EINTR)
 926                 goto retry;
 927 out:
 928         if (!ret) {
 929                 /* remove from alloc_heap: */
 930                 struct alloc_heap_entry e, *top = ca->alloc_heap.data;
 931
 932                 top->bucket++;
 933                 top->nr--;
 934
 935                 if (!top->nr)
 936                         heap_pop(&ca->alloc_heap, e, bucket_alloc_cmp, NULL);
 937
 938                 /*
 939                  * Make sure we flush the last journal entry that updated this
 940                  * bucket (i.e. deleting the last reference) before writing to
 941                  * this bucket again:
 942                  */
 943                 *journal_seq = max(*journal_seq, bucket_journal_seq(c, m));
 944         } else {
 945                 size_t b2;
 946
 947                 /* remove from free_inc: */
 948                 percpu_down_read(&c->mark_lock);
 949                 spin_lock(&c->freelist_lock);
 950
 951                 bch2_mark_alloc_bucket(c, ca, b, false,
 952                                        gc_pos_alloc(c, NULL), 0);
 953
 954                 BUG_ON(!fifo_pop_back(&ca->free_inc, b2));
 955                 BUG_ON(b != b2);
 956
 957                 spin_unlock(&c->freelist_lock);
 958                 percpu_up_read(&c->mark_lock);
 959         }
 960
 961         return ret < 0 ? ret : 0;
 962 }
 963
 964 /*
 965  * Pull buckets off ca->alloc_heap, invalidate them, move them to ca->free_inc:
 966  */
 967 static int bch2_invalidate_buckets(struct bch_fs *c, struct bch_dev *ca)
 968 {
 969         struct btree_trans trans;
 970         struct btree_iter *iter;
 971         u64 journal_seq = 0;
 972         int ret = 0;
 973
 974         bch2_trans_init(&trans, c, 0, 0);
 975
 976         iter = bch2_trans_get_iter(&trans, BTREE_ID_ALLOC,
 977                                    POS(ca->dev_idx, 0),
 978                                    BTREE_ITER_CACHED|
 979                                    BTREE_ITER_CACHED_NOFILL|
 980                                    BTREE_ITER_INTENT);
 981
 982         /* Only use nowait if we've already invalidated at least one bucket: */
 983         while (!ret &&
 984                !fifo_full(&ca->free_inc) &&
 985                ca->alloc_heap.used)
 986                 ret = bch2_invalidate_one_bucket2(&trans, ca, iter, &journal_seq,
 987                                 BTREE_INSERT_GC_LOCK_HELD|
 988                                 (!fifo_empty(&ca->free_inc)
 989                                  ? BTREE_INSERT_NOWAIT : 0));
 990
 991         bch2_trans_exit(&trans);
 992
 993         /* If we used NOWAIT, don't return the error: */
 994         if (!fifo_empty(&ca->free_inc))
 995                 ret = 0;
 996         if (ret) {
 997                 bch_err(ca, "error invalidating buckets: %i", ret);
 998                 return ret;
 999         }
1000
1001         if (journal_seq)
1002                 ret = bch2_journal_flush_seq(&c->journal, journal_seq);
1003         if (ret) {
1004                 bch_err(ca, "journal error: %i", ret);
1005                 return ret;
1006         }
1007
1008         return 0;
1009 }
1010
1011 static int push_invalidated_bucket(struct bch_fs *c, struct bch_dev *ca, size_t bucket)
1012 {
1013         unsigned i;
1014         int ret = 0;
1015
1016         while (1) {
1017                 set_current_state(TASK_INTERRUPTIBLE);
1018
1019                 spin_lock(&c->freelist_lock);
1020                 for (i = 0; i < RESERVE_NR; i++) {
1021
1022                         /*
1023                          * Don't strand buckets on the copygc freelist until
1024                          * after recovery is finished:
1025                          */
1026                         if (!test_bit(BCH_FS_STARTED, &c->flags) &&
1027                             i == RESERVE_MOVINGGC)
1028                                 continue;
1029
1030                         if (fifo_push(&ca->free[i], bucket)) {
1031                                 fifo_pop(&ca->free_inc, bucket);
1032
1033                                 closure_wake_up(&c->freelist_wait);
1034                                 ca->allocator_state = ALLOCATOR_RUNNING;
1035
1036                                 spin_unlock(&c->freelist_lock);
1037                                 goto out;
1038                         }
1039                 }
1040
1041                 if (ca->allocator_state != ALLOCATOR_BLOCKED_FULL) {
1042                         ca->allocator_state = ALLOCATOR_BLOCKED_FULL;
1043                         closure_wake_up(&c->freelist_wait);
1044                 }
1045
1046                 spin_unlock(&c->freelist_lock);
1047
1048                 if ((current->flags & PF_KTHREAD) &&
1049                     kthread_should_stop()) {
1050                         ret = 1;
1051                         break;
1052                 }
1053
1054                 schedule();
1055                 try_to_freeze();
1056         }
1057 out:
1058         __set_current_state(TASK_RUNNING);
1059         return ret;
1060 }
1061
1062 /*
1063  * Pulls buckets off free_inc, discards them (if enabled), then adds them to
1064  * freelists, waiting until there's room if necessary:
1065  */
1066 static int discard_invalidated_buckets(struct bch_fs *c, struct bch_dev *ca)
1067 {
1068         while (!fifo_empty(&ca->free_inc)) {
1069                 size_t bucket = fifo_peek(&ca->free_inc);
1070
1071                 if (ca->mi.discard &&
1072                     blk_queue_discard(bdev_get_queue(ca->disk_sb.bdev)))
1073                         blkdev_issue_discard(ca->disk_sb.bdev,
1074                                              bucket_to_sector(ca, bucket),
1075                                              ca->mi.bucket_size, GFP_NOIO, 0);
1076
1077                 if (push_invalidated_bucket(c, ca, bucket))
1078                         return 1;
1079         }
1080
1081         return 0;
1082 }
1083
1084 /**
1085  * bch_allocator_thread - move buckets from free_inc to reserves
1086  *
1087  * The free_inc FIFO is populated by find_reclaimable_buckets(), and
1088  * the reserves are depleted by bucket allocation. When we run out
1089  * of free_inc, try to invalidate some buckets and write out
1090  * prios and gens.
1091  */
1092 static int bch2_allocator_thread(void *arg)
1093 {
1094         struct bch_dev *ca = arg;
1095         struct bch_fs *c = ca->fs;
1096         size_t nr;
1097         int ret;
1098
1099         set_freezable();
1100         ca->allocator_state = ALLOCATOR_RUNNING;
1101
1102         while (1) {
1103                 cond_resched();
1104
1105                 pr_debug("discarding %zu invalidated buckets",
1106                          fifo_used(&ca->free_inc));
1107
1108                 ret = discard_invalidated_buckets(c, ca);
1109                 if (ret)
1110                         goto stop;
1111
1112                 down_read(&c->gc_lock);
1113
1114                 ret = bch2_invalidate_buckets(c, ca);
1115                 if (ret) {
1116                         up_read(&c->gc_lock);
1117                         goto stop;
1118                 }
1119
1120                 if (!fifo_empty(&ca->free_inc)) {
1121                         up_read(&c->gc_lock);
1122                         continue;
1123                 }
1124
1125                 pr_debug("free_inc now empty");
1126
1127                 do {
1128                         /*
1129                          * Find some buckets that we can invalidate, either
1130                          * they're completely unused, or only contain clean data
1131                          * that's been written back to the backing device or
1132                          * another cache tier
1133                          */
1134
1135                         pr_debug("scanning for reclaimable buckets");
1136
1137                         nr = find_reclaimable_buckets(c, ca);
1138
1139                         pr_debug("found %zu buckets", nr);
1140
1141                         trace_alloc_batch(ca, nr, ca->alloc_heap.size);
1142
1143                         if ((ca->inc_gen_needs_gc >= ALLOC_SCAN_BATCH(ca) ||
1144                              ca->inc_gen_really_needs_gc) &&
1145                             c->gc_thread) {
1146                                 atomic_inc(&c->kick_gc);
1147                                 wake_up_process(c->gc_thread);
1148                         }
1149
1150                         /*
1151                          * If we found any buckets, we have to invalidate them
1152                          * before we scan for more - but if we didn't find very
1153                          * many we may want to wait on more buckets being
1154                          * available so we don't spin:
1155                          */
1156                         if (!nr ||
1157                             (nr < ALLOC_SCAN_BATCH(ca) &&
1158                              !fifo_empty(&ca->free[RESERVE_NONE]))) {
1159                                 ret = wait_buckets_available(c, ca);
1160                                 if (ret) {
1161                                         up_read(&c->gc_lock);
1162                                         goto stop;
1163                                 }
1164                         }
1165                 } while (!nr);
1166
1167                 up_read(&c->gc_lock);
1168
1169                 pr_debug("%zu buckets to invalidate", nr);
1170
1171                 /*
1172                  * alloc_heap is now full of newly-invalidated buckets: next,
1173                  * write out the new bucket gens:
1174                  */
1175         }
1176
1177 stop:
1178         pr_debug("alloc thread stopping (ret %i)", ret);
1179         ca->allocator_state = ALLOCATOR_STOPPED;
1180         closure_wake_up(&c->freelist_wait);
1181         return 0;
1182 }
1183
1184 /* Startup/shutdown (ro/rw): */
1185
1186 void bch2_recalc_capacity(struct bch_fs *c)
1187 {
1188         struct bch_dev *ca;
1189         u64 capacity = 0, reserved_sectors = 0, gc_reserve;
1190         unsigned bucket_size_max = 0;
1191         unsigned long ra_pages = 0;
1192         unsigned i, j;
1193
1194         lockdep_assert_held(&c->state_lock);
1195
1196         for_each_online_member(ca, c, i) {
1197                 struct backing_dev_info *bdi = ca->disk_sb.bdev->bd_bdi;
1198
1199                 ra_pages += bdi->ra_pages;
1200         }
1201
1202         bch2_set_ra_pages(c, ra_pages);
1203
1204         for_each_rw_member(ca, c, i) {
1205                 u64 dev_reserve = 0;
1206
1207                 /*
1208                  * We need to reserve buckets (from the number
1209                  * of currently available buckets) against
1210                  * foreground writes so that mainly copygc can
1211                  * make forward progress.
1212                  *
1213                  * We need enough to refill the various reserves
1214                  * from scratch - copygc will use its entire
1215                  * reserve all at once, then run against when
1216                  * its reserve is refilled (from the formerly
1217                  * available buckets).
1218                  *
1219                  * This reserve is just used when considering if
1220                  * allocations for foreground writes must wait -
1221                  * not -ENOSPC calculations.
1222                  */
1223                 for (j = 0; j < RESERVE_NONE; j++)
1224                         dev_reserve += ca->free[j].size;
1225
1226                 dev_reserve += 1;       /* btree write point */
1227                 dev_reserve += 1;       /* copygc write point */
1228                 dev_reserve += 1;       /* rebalance write point */
1229
1230                 dev_reserve *= ca->mi.bucket_size;
1231
1232                 ca->copygc_threshold = dev_reserve;
1233
1234                 capacity += bucket_to_sector(ca, ca->mi.nbuckets -
1235                                              ca->mi.first_bucket);
1236
1237                 reserved_sectors += dev_reserve * 2;
1238
1239                 bucket_size_max = max_t(unsigned, bucket_size_max,
1240                                         ca->mi.bucket_size);
1241         }
1242
1243         gc_reserve = c->opts.gc_reserve_bytes
1244                 ? c->opts.gc_reserve_bytes >> 9
1245                 : div64_u64(capacity * c->opts.gc_reserve_percent, 100);
1246
1247         reserved_sectors = max(gc_reserve, reserved_sectors);
1248
1249         reserved_sectors = min(reserved_sectors, capacity);
1250
1251         c->capacity = capacity - reserved_sectors;
1252
1253         c->bucket_size_max = bucket_size_max;
1254
1255         if (c->capacity) {
1256                 bch2_io_timer_add(&c->io_clock[READ],
1257                                  &c->bucket_clock[READ].rescale);
1258                 bch2_io_timer_add(&c->io_clock[WRITE],
1259                                  &c->bucket_clock[WRITE].rescale);
1260         } else {
1261                 bch2_io_timer_del(&c->io_clock[READ],
1262                                  &c->bucket_clock[READ].rescale);
1263                 bch2_io_timer_del(&c->io_clock[WRITE],
1264                                  &c->bucket_clock[WRITE].rescale);
1265         }
1266
1267         /* Wake up case someone was waiting for buckets */
1268         closure_wake_up(&c->freelist_wait);
1269 }
1270
1271 static bool bch2_dev_has_open_write_point(struct bch_fs *c, struct bch_dev *ca)
1272 {
1273         struct open_bucket *ob;
1274         bool ret = false;
1275
1276         for (ob = c->open_buckets;
1277              ob < c->open_buckets + ARRAY_SIZE(c->open_buckets);
1278              ob++) {
1279                 spin_lock(&ob->lock);
1280                 if (ob->valid && !ob->on_partial_list &&
1281                     ob->ptr.dev == ca->dev_idx)
1282                         ret = true;
1283                 spin_unlock(&ob->lock);
1284         }
1285
1286         return ret;
1287 }
1288
1289 /* device goes ro: */
1290 void bch2_dev_allocator_remove(struct bch_fs *c, struct bch_dev *ca)
1291 {
1292         unsigned i;
1293
1294         BUG_ON(ca->alloc_thread);
1295
1296         /* First, remove device from allocation groups: */
1297
1298         for (i = 0; i < ARRAY_SIZE(c->rw_devs); i++)
1299                 clear_bit(ca->dev_idx, c->rw_devs[i].d);
1300
1301         /*
1302          * Capacity is calculated based off of devices in allocation groups:
1303          */
1304         bch2_recalc_capacity(c);
1305
1306         /* Next, close write points that point to this device... */
1307         for (i = 0; i < ARRAY_SIZE(c->write_points); i++)
1308                 bch2_writepoint_stop(c, ca, &c->write_points[i]);
1309
1310         bch2_writepoint_stop(c, ca, &ca->copygc_write_point);
1311         bch2_writepoint_stop(c, ca, &c->rebalance_write_point);
1312         bch2_writepoint_stop(c, ca, &c->btree_write_point);
1313
1314         mutex_lock(&c->btree_reserve_cache_lock);
1315         while (c->btree_reserve_cache_nr) {
1316                 struct btree_alloc *a =
1317                         &c->btree_reserve_cache[--c->btree_reserve_cache_nr];
1318
1319                 bch2_open_buckets_put(c, &a->ob);
1320         }
1321         mutex_unlock(&c->btree_reserve_cache_lock);
1322
1323         while (1) {
1324                 struct open_bucket *ob;
1325
1326                 spin_lock(&c->freelist_lock);
1327                 if (!ca->open_buckets_partial_nr) {
1328                         spin_unlock(&c->freelist_lock);
1329                         break;
1330                 }
1331                 ob = c->open_buckets +
1332                         ca->open_buckets_partial[--ca->open_buckets_partial_nr];
1333                 ob->on_partial_list = false;
1334                 spin_unlock(&c->freelist_lock);
1335
1336                 bch2_open_bucket_put(c, ob);
1337         }
1338
1339         bch2_ec_stop_dev(c, ca);
1340
1341         /*
1342          * Wake up threads that were blocked on allocation, so they can notice
1343          * the device can no longer be removed and the capacity has changed:
1344          */
1345         closure_wake_up(&c->freelist_wait);
1346
1347         /*
1348          * journal_res_get() can block waiting for free space in the journal -
1349          * it needs to notice there may not be devices to allocate from anymore:
1350          */
1351         wake_up(&c->journal.wait);
1352
1353         /* Now wait for any in flight writes: */
1354
1355         closure_wait_event(&c->open_buckets_wait,
1356                            !bch2_dev_has_open_write_point(c, ca));
1357 }
1358
1359 /* device goes rw: */
1360 void bch2_dev_allocator_add(struct bch_fs *c, struct bch_dev *ca)
1361 {
1362         unsigned i;
1363
1364         for (i = 0; i < ARRAY_SIZE(c->rw_devs); i++)
1365                 if (ca->mi.data_allowed & (1 << i))
1366                         set_bit(ca->dev_idx, c->rw_devs[i].d);
1367 }
1368
1369 void bch2_dev_allocator_quiesce(struct bch_fs *c, struct bch_dev *ca)
1370 {
1371         if (ca->alloc_thread)
1372                 closure_wait_event(&c->freelist_wait,
1373                                    ca->allocator_state != ALLOCATOR_RUNNING);
1374 }
1375
1376 /* stop allocator thread: */
1377 void bch2_dev_allocator_stop(struct bch_dev *ca)
1378 {
1379         struct task_struct *p;
1380
1381         p = rcu_dereference_protected(ca->alloc_thread, 1);
1382         ca->alloc_thread = NULL;
1383
1384         /*
1385          * We need an rcu barrier between setting ca->alloc_thread = NULL and
1386          * the thread shutting down to avoid bch2_wake_allocator() racing:
1387          *
1388          * XXX: it would be better to have the rcu barrier be asynchronous
1389          * instead of blocking us here
1390          */
1391         synchronize_rcu();
1392
1393         if (p) {
1394                 kthread_stop(p);
1395                 put_task_struct(p);
1396         }
1397 }
1398
1399 /* start allocator thread: */
1400 int bch2_dev_allocator_start(struct bch_dev *ca)
1401 {
1402         struct task_struct *p;
1403
1404         /*
1405          * allocator thread already started?
1406          */
1407         if (ca->alloc_thread)
1408                 return 0;
1409
1410         p = kthread_create(bch2_allocator_thread, ca,
1411                            "bch_alloc[%s]", ca->name);
1412         if (IS_ERR(p))
1413                 return PTR_ERR(p);
1414
1415         get_task_struct(p);
1416         rcu_assign_pointer(ca->alloc_thread, p);
1417         wake_up_process(p);
1418         return 0;
1419 }
1420
1421 void bch2_fs_allocator_background_init(struct bch_fs *c)
1422 {
1423         spin_lock_init(&c->freelist_lock);
1424         bch2_bucket_clock_init(c, READ);
1425         bch2_bucket_clock_init(c, WRITE);
1426
1427         c->pd_controllers_update_seconds = 5;
1428         INIT_DELAYED_WORK(&c->pd_controllers_update, pd_controllers_update);
1429 }