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

   1 // SPDX-License-Identifier: GPL-2.0
   2 /*
   3  * bcachefs journalling code, for btree insertions
   4  *
   5  * Copyright 2012 Google, Inc.
   6  */
   7
   8 #include "bcachefs.h"
   9 #include "alloc_foreground.h"
  10 #include "bkey_methods.h"
  11 #include "btree_gc.h"
  12 #include "buckets.h"
  13 #include "journal.h"
  14 #include "journal_io.h"
  15 #include "journal_reclaim.h"
  16 #include "journal_seq_blacklist.h"
  17 #include "super-io.h"
  18
  19 #include <trace/events/bcachefs.h>
  20
  21 static bool __journal_entry_is_open(union journal_res_state state)
  22 {
  23         return state.cur_entry_offset < JOURNAL_ENTRY_CLOSED_VAL;
  24 }
  25
  26 static bool journal_entry_is_open(struct journal *j)
  27 {
  28         return __journal_entry_is_open(j->reservations);
  29 }
  30
  31 static void journal_pin_new_entry(struct journal *j, int count)
  32 {
  33         struct journal_entry_pin_list *p;
  34
  35         /*
  36          * The fifo_push() needs to happen at the same time as j->seq is
  37          * incremented for journal_last_seq() to be calculated correctly
  38          */
  39         atomic64_inc(&j->seq);
  40         p = fifo_push_ref(&j->pin);
  41
  42         INIT_LIST_HEAD(&p->list);
  43         INIT_LIST_HEAD(&p->flushed);
  44         atomic_set(&p->count, count);
  45         p->devs.nr = 0;
  46 }
  47
  48 static void bch2_journal_buf_init(struct journal *j)
  49 {
  50         struct journal_buf *buf = journal_cur_buf(j);
  51
  52         memset(buf->has_inode, 0, sizeof(buf->has_inode));
  53
  54         memset(buf->data, 0, sizeof(*buf->data));
  55         buf->data->seq  = cpu_to_le64(journal_cur_seq(j));
  56         buf->data->u64s = 0;
  57 }
  58
  59 void bch2_journal_halt(struct journal *j)
  60 {
  61         union journal_res_state old, new;
  62         u64 v = atomic64_read(&j->reservations.counter);
  63
  64         do {
  65                 old.v = new.v = v;
  66                 if (old.cur_entry_offset == JOURNAL_ENTRY_ERROR_VAL)
  67                         return;
  68
  69                 new.cur_entry_offset = JOURNAL_ENTRY_ERROR_VAL;
  70         } while ((v = atomic64_cmpxchg(&j->reservations.counter,
  71                                        old.v, new.v)) != old.v);
  72
  73         journal_wake(j);
  74         closure_wake_up(&journal_cur_buf(j)->wait);
  75 }
  76
  77 /* journal entry close/open: */
  78
  79 void __bch2_journal_buf_put(struct journal *j, bool need_write_just_set)
  80 {
  81         if (!need_write_just_set &&
  82             test_bit(JOURNAL_NEED_WRITE, &j->flags))
  83                 bch2_time_stats_update(j->delay_time,
  84                                        j->need_write_time);
  85
  86         clear_bit(JOURNAL_NEED_WRITE, &j->flags);
  87
  88         closure_call(&j->io, bch2_journal_write, system_highpri_wq, NULL);
  89 }
  90
  91 /*
  92  * Returns true if journal entry is now closed:
  93  */
  94 static bool __journal_entry_close(struct journal *j)
  95 {
  96         struct bch_fs *c = container_of(j, struct bch_fs, journal);
  97         struct journal_buf *buf = journal_cur_buf(j);
  98         union journal_res_state old, new;
  99         u64 v = atomic64_read(&j->reservations.counter);
 100         bool set_need_write = false;
 101         unsigned sectors;
 102
 103         lockdep_assert_held(&j->lock);
 104
 105         do {
 106                 old.v = new.v = v;
 107                 if (old.cur_entry_offset == JOURNAL_ENTRY_CLOSED_VAL)
 108                         return true;
 109
 110                 if (old.cur_entry_offset == JOURNAL_ENTRY_ERROR_VAL) {
 111                         /* this entry will never be written: */
 112                         closure_wake_up(&buf->wait);
 113                         return true;
 114                 }
 115
 116                 if (!test_bit(JOURNAL_NEED_WRITE, &j->flags)) {
 117                         set_bit(JOURNAL_NEED_WRITE, &j->flags);
 118                         j->need_write_time = local_clock();
 119                         set_need_write = true;
 120                 }
 121
 122                 if (new.prev_buf_unwritten)
 123                         return false;
 124
 125                 new.cur_entry_offset = JOURNAL_ENTRY_CLOSED_VAL;
 126                 new.idx++;
 127                 new.prev_buf_unwritten = 1;
 128
 129                 BUG_ON(journal_state_count(new, new.idx));
 130         } while ((v = atomic64_cmpxchg(&j->reservations.counter,
 131                                        old.v, new.v)) != old.v);
 132
 133         buf->data->u64s         = cpu_to_le32(old.cur_entry_offset);
 134
 135         sectors = vstruct_blocks_plus(buf->data, c->block_bits,
 136                                       buf->u64s_reserved) << c->block_bits;
 137         BUG_ON(sectors > buf->sectors);
 138         buf->sectors = sectors;
 139
 140         bkey_extent_init(&buf->key);
 141
 142         /*
 143          * We have to set last_seq here, _before_ opening a new journal entry:
 144          *
 145          * A threads may replace an old pin with a new pin on their current
 146          * journal reservation - the expectation being that the journal will
 147          * contain either what the old pin protected or what the new pin
 148          * protects.
 149          *
 150          * After the old pin is dropped journal_last_seq() won't include the old
 151          * pin, so we can only write the updated last_seq on the entry that
 152          * contains whatever the new pin protects.
 153          *
 154          * Restated, we can _not_ update last_seq for a given entry if there
 155          * could be a newer entry open with reservations/pins that have been
 156          * taken against it.
 157          *
 158          * Hence, we want update/set last_seq on the current journal entry right
 159          * before we open a new one:
 160          */
 161         buf->data->last_seq     = cpu_to_le64(journal_last_seq(j));
 162
 163         if (journal_entry_empty(buf->data))
 164                 clear_bit(JOURNAL_NOT_EMPTY, &j->flags);
 165         else
 166                 set_bit(JOURNAL_NOT_EMPTY, &j->flags);
 167
 168         journal_pin_new_entry(j, 1);
 169
 170         bch2_journal_buf_init(j);
 171
 172         cancel_delayed_work(&j->write_work);
 173
 174         bch2_journal_space_available(j);
 175
 176         bch2_journal_buf_put(j, old.idx, set_need_write);
 177         return true;
 178 }
 179
 180 static bool journal_entry_close(struct journal *j)
 181 {
 182         bool ret;
 183
 184         spin_lock(&j->lock);
 185         ret = __journal_entry_close(j);
 186         spin_unlock(&j->lock);
 187
 188         return ret;
 189 }
 190
 191 /*
 192  * should _only_ called from journal_res_get() - when we actually want a
 193  * journal reservation - journal entry is open means journal is dirty:
 194  *
 195  * returns:
 196  * 0:           success
 197  * -ENOSPC:     journal currently full, must invoke reclaim
 198  * -EAGAIN:     journal blocked, must wait
 199  * -EROFS:      insufficient rw devices or journal error
 200  */
 201 static int journal_entry_open(struct journal *j)
 202 {
 203         struct journal_buf *buf = journal_cur_buf(j);
 204         union journal_res_state old, new;
 205         int u64s;
 206         u64 v;
 207
 208         lockdep_assert_held(&j->lock);
 209         BUG_ON(journal_entry_is_open(j));
 210
 211         if (j->blocked)
 212                 return -EAGAIN;
 213
 214         if (j->cur_entry_error)
 215                 return j->cur_entry_error;
 216
 217         BUG_ON(!j->cur_entry_sectors);
 218
 219         buf->u64s_reserved      = j->entry_u64s_reserved;
 220         buf->disk_sectors       = j->cur_entry_sectors;
 221         buf->sectors            = min(buf->disk_sectors, buf->buf_size >> 9);
 222
 223         u64s = (int) (buf->sectors << 9) / sizeof(u64) -
 224                 journal_entry_overhead(j);
 225         u64s  = clamp_t(int, u64s, 0, JOURNAL_ENTRY_CLOSED_VAL - 1);
 226
 227         if (u64s <= le32_to_cpu(buf->data->u64s))
 228                 return -ENOSPC;
 229
 230         /*
 231          * Must be set before marking the journal entry as open:
 232          */
 233         j->cur_entry_u64s = u64s;
 234
 235         v = atomic64_read(&j->reservations.counter);
 236         do {
 237                 old.v = new.v = v;
 238
 239                 if (old.cur_entry_offset == JOURNAL_ENTRY_ERROR_VAL)
 240                         return -EROFS;
 241
 242                 /* Handle any already added entries */
 243                 new.cur_entry_offset = le32_to_cpu(buf->data->u64s);
 244
 245                 EBUG_ON(journal_state_count(new, new.idx));
 246                 journal_state_inc(&new);
 247         } while ((v = atomic64_cmpxchg(&j->reservations.counter,
 248                                        old.v, new.v)) != old.v);
 249
 250         if (j->res_get_blocked_start)
 251                 bch2_time_stats_update(j->blocked_time,
 252                                        j->res_get_blocked_start);
 253         j->res_get_blocked_start = 0;
 254
 255         mod_delayed_work(system_freezable_wq,
 256                          &j->write_work,
 257                          msecs_to_jiffies(j->write_delay_ms));
 258         journal_wake(j);
 259         return 0;
 260 }
 261
 262 static bool journal_quiesced(struct journal *j)
 263 {
 264         union journal_res_state state = READ_ONCE(j->reservations);
 265         bool ret = !state.prev_buf_unwritten && !__journal_entry_is_open(state);
 266
 267         if (!ret)
 268                 journal_entry_close(j);
 269         return ret;
 270 }
 271
 272 static void journal_quiesce(struct journal *j)
 273 {
 274         wait_event(j->wait, journal_quiesced(j));
 275 }
 276
 277 static void journal_write_work(struct work_struct *work)
 278 {
 279         struct journal *j = container_of(work, struct journal, write_work.work);
 280
 281         journal_entry_close(j);
 282 }
 283
 284 /*
 285  * Given an inode number, if that inode number has data in the journal that
 286  * hasn't yet been flushed, return the journal sequence number that needs to be
 287  * flushed:
 288  */
 289 u64 bch2_inode_journal_seq(struct journal *j, u64 inode)
 290 {
 291         size_t h = hash_64(inode, ilog2(sizeof(j->buf[0].has_inode) * 8));
 292         u64 seq = 0;
 293
 294         if (!test_bit(h, j->buf[0].has_inode) &&
 295             !test_bit(h, j->buf[1].has_inode))
 296                 return 0;
 297
 298         spin_lock(&j->lock);
 299         if (test_bit(h, journal_cur_buf(j)->has_inode))
 300                 seq = journal_cur_seq(j);
 301         else if (test_bit(h, journal_prev_buf(j)->has_inode))
 302                 seq = journal_cur_seq(j) - 1;
 303         spin_unlock(&j->lock);
 304
 305         return seq;
 306 }
 307
 308 static int __journal_res_get(struct journal *j, struct journal_res *res,
 309                              unsigned flags)
 310 {
 311         struct bch_fs *c = container_of(j, struct bch_fs, journal);
 312         struct journal_buf *buf;
 313         bool can_discard;
 314         int ret;
 315 retry:
 316         if (journal_res_get_fast(j, res, flags))
 317                 return 0;
 318
 319         if (bch2_journal_error(j))
 320                 return -EROFS;
 321
 322         spin_lock(&j->lock);
 323
 324         /*
 325          * Recheck after taking the lock, so we don't race with another thread
 326          * that just did journal_entry_open() and call journal_entry_close()
 327          * unnecessarily
 328          */
 329         if (journal_res_get_fast(j, res, flags)) {
 330                 spin_unlock(&j->lock);
 331                 return 0;
 332         }
 333
 334         if (!(flags & JOURNAL_RES_GET_RESERVED) &&
 335             !test_bit(JOURNAL_MAY_GET_UNRESERVED, &j->flags)) {
 336                 /*
 337                  * Don't want to close current journal entry, just need to
 338                  * invoke reclaim:
 339                  */
 340                 ret = -ENOSPC;
 341                 goto unlock;
 342         }
 343
 344         /*
 345          * If we couldn't get a reservation because the current buf filled up,
 346          * and we had room for a bigger entry on disk, signal that we want to
 347          * realloc the journal bufs:
 348          */
 349         buf = journal_cur_buf(j);
 350         if (journal_entry_is_open(j) &&
 351             buf->buf_size >> 9 < buf->disk_sectors &&
 352             buf->buf_size < JOURNAL_ENTRY_SIZE_MAX)
 353                 j->buf_size_want = max(j->buf_size_want, buf->buf_size << 1);
 354
 355         if (journal_entry_is_open(j) &&
 356             !__journal_entry_close(j)) {
 357                 /*
 358                  * We failed to get a reservation on the current open journal
 359                  * entry because it's full, and we can't close it because
 360                  * there's still a previous one in flight:
 361                  */
 362                 trace_journal_entry_full(c);
 363                 ret = -EAGAIN;
 364         } else {
 365                 ret = journal_entry_open(j);
 366         }
 367 unlock:
 368         if ((ret == -EAGAIN || ret == -ENOSPC) &&
 369             !j->res_get_blocked_start)
 370                 j->res_get_blocked_start = local_clock() ?: 1;
 371
 372         can_discard = j->can_discard;
 373         spin_unlock(&j->lock);
 374
 375         if (!ret)
 376                 goto retry;
 377
 378         if (ret == -ENOSPC) {
 379                 WARN_ONCE(!can_discard && (flags & JOURNAL_RES_GET_RESERVED),
 380                           "JOURNAL_RES_GET_RESERVED set but journal full");
 381
 382                 /*
 383                  * Journal is full - can't rely on reclaim from work item due to
 384                  * freezing:
 385                  */
 386                 trace_journal_full(c);
 387
 388                 if (!(flags & JOURNAL_RES_GET_NONBLOCK)) {
 389                         if (can_discard) {
 390                                 bch2_journal_do_discards(j);
 391                                 goto retry;
 392                         }
 393
 394                         if (mutex_trylock(&j->reclaim_lock)) {
 395                                 bch2_journal_reclaim(j);
 396                                 mutex_unlock(&j->reclaim_lock);
 397                         }
 398                 }
 399
 400                 ret = -EAGAIN;
 401         }
 402
 403         return ret;
 404 }
 405
 406 /*
 407  * Essentially the entry function to the journaling code. When bcachefs is doing
 408  * a btree insert, it calls this function to get the current journal write.
 409  * Journal write is the structure used set up journal writes. The calling
 410  * function will then add its keys to the structure, queuing them for the next
 411  * write.
 412  *
 413  * To ensure forward progress, the current task must not be holding any
 414  * btree node write locks.
 415  */
 416 int bch2_journal_res_get_slowpath(struct journal *j, struct journal_res *res,
 417                                   unsigned flags)
 418 {
 419         int ret;
 420
 421         closure_wait_event(&j->async_wait,
 422                    (ret = __journal_res_get(j, res, flags)) != -EAGAIN ||
 423                    (flags & JOURNAL_RES_GET_NONBLOCK));
 424         return ret;
 425 }
 426
 427 /* journal_preres: */
 428
 429 static bool journal_preres_available(struct journal *j,
 430                                      struct journal_preres *res,
 431                                      unsigned new_u64s,
 432                                      unsigned flags)
 433 {
 434         bool ret = bch2_journal_preres_get_fast(j, res, new_u64s, flags);
 435
 436         if (!ret)
 437                 bch2_journal_reclaim_work(&j->reclaim_work.work);
 438
 439         return ret;
 440 }
 441
 442 int __bch2_journal_preres_get(struct journal *j,
 443                               struct journal_preres *res,
 444                               unsigned new_u64s,
 445                               unsigned flags)
 446 {
 447         int ret;
 448
 449         closure_wait_event(&j->preres_wait,
 450                    (ret = bch2_journal_error(j)) ||
 451                    journal_preres_available(j, res, new_u64s, flags));
 452         return ret;
 453 }
 454
 455 /* journal_entry_res: */
 456
 457 void bch2_journal_entry_res_resize(struct journal *j,
 458                                    struct journal_entry_res *res,
 459                                    unsigned new_u64s)
 460 {
 461         union journal_res_state state;
 462         int d = new_u64s - res->u64s;
 463
 464         spin_lock(&j->lock);
 465
 466         j->entry_u64s_reserved += d;
 467         if (d <= 0)
 468                 goto out;
 469
 470         j->cur_entry_u64s = max_t(int, 0, j->cur_entry_u64s - d);
 471         smp_mb();
 472         state = READ_ONCE(j->reservations);
 473
 474         if (state.cur_entry_offset < JOURNAL_ENTRY_CLOSED_VAL &&
 475             state.cur_entry_offset > j->cur_entry_u64s) {
 476                 j->cur_entry_u64s += d;
 477                 /*
 478                  * Not enough room in current journal entry, have to flush it:
 479                  */
 480                 __journal_entry_close(j);
 481         } else {
 482                 journal_cur_buf(j)->u64s_reserved += d;
 483         }
 484 out:
 485         spin_unlock(&j->lock);
 486         res->u64s += d;
 487 }
 488
 489 /* journal flushing: */
 490
 491 u64 bch2_journal_last_unwritten_seq(struct journal *j)
 492 {
 493         u64 seq;
 494
 495         spin_lock(&j->lock);
 496         seq = journal_cur_seq(j);
 497         if (j->reservations.prev_buf_unwritten)
 498                 seq--;
 499         spin_unlock(&j->lock);
 500
 501         return seq;
 502 }
 503
 504 /**
 505  * bch2_journal_open_seq_async - try to open a new journal entry if @seq isn't
 506  * open yet, or wait if we cannot
 507  *
 508  * used by the btree interior update machinery, when it needs to write a new
 509  * btree root - every journal entry contains the roots of all the btrees, so it
 510  * doesn't need to bother with getting a journal reservation
 511  */
 512 int bch2_journal_open_seq_async(struct journal *j, u64 seq, struct closure *cl)
 513 {
 514         struct bch_fs *c = container_of(j, struct bch_fs, journal);
 515         int ret;
 516
 517         spin_lock(&j->lock);
 518
 519         /*
 520          * Can't try to open more than one sequence number ahead:
 521          */
 522         BUG_ON(journal_cur_seq(j) < seq && !journal_entry_is_open(j));
 523
 524         if (journal_cur_seq(j) > seq ||
 525             journal_entry_is_open(j)) {
 526                 spin_unlock(&j->lock);
 527                 return 0;
 528         }
 529
 530         if (journal_cur_seq(j) < seq &&
 531             !__journal_entry_close(j)) {
 532                 /* haven't finished writing out the previous one: */
 533                 trace_journal_entry_full(c);
 534                 ret = -EAGAIN;
 535         } else {
 536                 BUG_ON(journal_cur_seq(j) != seq);
 537
 538                 ret = journal_entry_open(j);
 539         }
 540
 541         if ((ret == -EAGAIN || ret == -ENOSPC) &&
 542             !j->res_get_blocked_start)
 543                 j->res_get_blocked_start = local_clock() ?: 1;
 544
 545         if (ret == -EAGAIN || ret == -ENOSPC)
 546                 closure_wait(&j->async_wait, cl);
 547
 548         spin_unlock(&j->lock);
 549
 550         if (ret == -ENOSPC) {
 551                 trace_journal_full(c);
 552                 bch2_journal_reclaim_work(&j->reclaim_work.work);
 553                 ret = -EAGAIN;
 554         }
 555
 556         return ret;
 557 }
 558
 559 static int journal_seq_error(struct journal *j, u64 seq)
 560 {
 561         union journal_res_state state = READ_ONCE(j->reservations);
 562
 563         if (seq == journal_cur_seq(j))
 564                 return bch2_journal_error(j);
 565
 566         if (seq + 1 == journal_cur_seq(j) &&
 567             !state.prev_buf_unwritten &&
 568             seq > j->seq_ondisk)
 569                 return -EIO;
 570
 571         return 0;
 572 }
 573
 574 static inline struct journal_buf *
 575 journal_seq_to_buf(struct journal *j, u64 seq)
 576 {
 577         /* seq should be for a journal entry that has been opened: */
 578         BUG_ON(seq > journal_cur_seq(j));
 579         BUG_ON(seq == journal_cur_seq(j) &&
 580                j->reservations.cur_entry_offset == JOURNAL_ENTRY_CLOSED_VAL);
 581
 582         if (seq == journal_cur_seq(j))
 583                 return journal_cur_buf(j);
 584         if (seq + 1 == journal_cur_seq(j) &&
 585             j->reservations.prev_buf_unwritten)
 586                 return journal_prev_buf(j);
 587         return NULL;
 588 }
 589
 590 /**
 591  * bch2_journal_wait_on_seq - wait for a journal entry to be written
 592  *
 593  * does _not_ cause @seq to be written immediately - if there is no other
 594  * activity to cause the relevant journal entry to be filled up or flushed it
 595  * can wait for an arbitrary amount of time (up to @j->write_delay_ms, which is
 596  * configurable).
 597  */
 598 void bch2_journal_wait_on_seq(struct journal *j, u64 seq,
 599                               struct closure *parent)
 600 {
 601         struct journal_buf *buf;
 602
 603         spin_lock(&j->lock);
 604
 605         if ((buf = journal_seq_to_buf(j, seq))) {
 606                 if (!closure_wait(&buf->wait, parent))
 607                         BUG();
 608
 609                 if (seq == journal_cur_seq(j)) {
 610                         smp_mb();
 611                         if (bch2_journal_error(j))
 612                                 closure_wake_up(&buf->wait);
 613                 }
 614         }
 615
 616         spin_unlock(&j->lock);
 617 }
 618
 619 /**
 620  * bch2_journal_flush_seq_async - wait for a journal entry to be written
 621  *
 622  * like bch2_journal_wait_on_seq, except that it triggers a write immediately if
 623  * necessary
 624  */
 625 void bch2_journal_flush_seq_async(struct journal *j, u64 seq,
 626                                   struct closure *parent)
 627 {
 628         struct journal_buf *buf;
 629
 630         spin_lock(&j->lock);
 631
 632         if (parent &&
 633             (buf = journal_seq_to_buf(j, seq)))
 634                 if (!closure_wait(&buf->wait, parent))
 635                         BUG();
 636
 637         if (seq == journal_cur_seq(j))
 638                 __journal_entry_close(j);
 639         spin_unlock(&j->lock);
 640 }
 641
 642 static int journal_seq_flushed(struct journal *j, u64 seq)
 643 {
 644         int ret;
 645
 646         spin_lock(&j->lock);
 647         ret = seq <= j->seq_ondisk ? 1 : journal_seq_error(j, seq);
 648
 649         if (seq == journal_cur_seq(j))
 650                 __journal_entry_close(j);
 651         spin_unlock(&j->lock);
 652
 653         return ret;
 654 }
 655
 656 int bch2_journal_flush_seq(struct journal *j, u64 seq)
 657 {
 658         u64 start_time = local_clock();
 659         int ret, ret2;
 660
 661         ret = wait_event_killable(j->wait, (ret2 = journal_seq_flushed(j, seq)));
 662
 663         bch2_time_stats_update(j->flush_seq_time, start_time);
 664
 665         return ret ?: ret2 < 0 ? ret2 : 0;
 666 }
 667
 668 /**
 669  * bch2_journal_meta_async - force a journal entry to be written
 670  */
 671 void bch2_journal_meta_async(struct journal *j, struct closure *parent)
 672 {
 673         struct journal_res res;
 674
 675         memset(&res, 0, sizeof(res));
 676
 677         bch2_journal_res_get(j, &res, jset_u64s(0), 0);
 678         bch2_journal_res_put(j, &res);
 679
 680         bch2_journal_flush_seq_async(j, res.seq, parent);
 681 }
 682
 683 int bch2_journal_meta(struct journal *j)
 684 {
 685         struct journal_res res;
 686         int ret;
 687
 688         memset(&res, 0, sizeof(res));
 689
 690         ret = bch2_journal_res_get(j, &res, jset_u64s(0), 0);
 691         if (ret)
 692                 return ret;
 693
 694         bch2_journal_res_put(j, &res);
 695
 696         return bch2_journal_flush_seq(j, res.seq);
 697 }
 698
 699 /*
 700  * bch2_journal_flush_async - if there is an open journal entry, or a journal
 701  * still being written, write it and wait for the write to complete
 702  */
 703 void bch2_journal_flush_async(struct journal *j, struct closure *parent)
 704 {
 705         u64 seq, journal_seq;
 706
 707         spin_lock(&j->lock);
 708         journal_seq = journal_cur_seq(j);
 709
 710         if (journal_entry_is_open(j)) {
 711                 seq = journal_seq;
 712         } else if (journal_seq) {
 713                 seq = journal_seq - 1;
 714         } else {
 715                 spin_unlock(&j->lock);
 716                 return;
 717         }
 718         spin_unlock(&j->lock);
 719
 720         bch2_journal_flush_seq_async(j, seq, parent);
 721 }
 722
 723 int bch2_journal_flush(struct journal *j)
 724 {
 725         u64 seq, journal_seq;
 726
 727         spin_lock(&j->lock);
 728         journal_seq = journal_cur_seq(j);
 729
 730         if (journal_entry_is_open(j)) {
 731                 seq = journal_seq;
 732         } else if (journal_seq) {
 733                 seq = journal_seq - 1;
 734         } else {
 735                 spin_unlock(&j->lock);
 736                 return 0;
 737         }
 738         spin_unlock(&j->lock);
 739
 740         return bch2_journal_flush_seq(j, seq);
 741 }
 742
 743 /* block/unlock the journal: */
 744
 745 void bch2_journal_unblock(struct journal *j)
 746 {
 747         spin_lock(&j->lock);
 748         j->blocked--;
 749         spin_unlock(&j->lock);
 750
 751         journal_wake(j);
 752 }
 753
 754 void bch2_journal_block(struct journal *j)
 755 {
 756         spin_lock(&j->lock);
 757         j->blocked++;
 758         spin_unlock(&j->lock);
 759
 760         journal_quiesce(j);
 761 }
 762
 763 /* allocate journal on a device: */
 764
 765 static int __bch2_set_nr_journal_buckets(struct bch_dev *ca, unsigned nr,
 766                                          bool new_fs, struct closure *cl)
 767 {
 768         struct bch_fs *c = ca->fs;
 769         struct journal_device *ja = &ca->journal;
 770         struct bch_sb_field_journal *journal_buckets;
 771         u64 *new_bucket_seq = NULL, *new_buckets = NULL;
 772         int ret = 0;
 773
 774         /* don't handle reducing nr of buckets yet: */
 775         if (nr <= ja->nr)
 776                 return 0;
 777
 778         ret = -ENOMEM;
 779         new_buckets     = kzalloc(nr * sizeof(u64), GFP_KERNEL);
 780         new_bucket_seq  = kzalloc(nr * sizeof(u64), GFP_KERNEL);
 781         if (!new_buckets || !new_bucket_seq)
 782                 goto err;
 783
 784         journal_buckets = bch2_sb_resize_journal(&ca->disk_sb,
 785                                                  nr + sizeof(*journal_buckets) / sizeof(u64));
 786         if (!journal_buckets)
 787                 goto err;
 788
 789         /*
 790          * We may be called from the device add path, before the new device has
 791          * actually been added to the running filesystem:
 792          */
 793         if (c)
 794                 spin_lock(&c->journal.lock);
 795
 796         memcpy(new_buckets,     ja->buckets,    ja->nr * sizeof(u64));
 797         memcpy(new_bucket_seq,  ja->bucket_seq, ja->nr * sizeof(u64));
 798         swap(new_buckets,       ja->buckets);
 799         swap(new_bucket_seq,    ja->bucket_seq);
 800
 801         if (c)
 802                 spin_unlock(&c->journal.lock);
 803
 804         while (ja->nr < nr) {
 805                 struct open_bucket *ob = NULL;
 806                 unsigned pos;
 807                 long bucket;
 808
 809                 if (new_fs) {
 810                         bucket = bch2_bucket_alloc_new_fs(ca);
 811                         if (bucket < 0) {
 812                                 ret = -ENOSPC;
 813                                 goto err;
 814                         }
 815                 } else {
 816                         ob = bch2_bucket_alloc(c, ca, RESERVE_ALLOC,
 817                                                false, cl);
 818                         if (IS_ERR(ob)) {
 819                                 ret = cl ? -EAGAIN : -ENOSPC;
 820                                 goto err;
 821                         }
 822
 823                         bucket = sector_to_bucket(ca, ob->ptr.offset);
 824                 }
 825
 826                 if (c) {
 827                         percpu_down_read(&c->mark_lock);
 828                         spin_lock(&c->journal.lock);
 829                 }
 830
 831                 pos = ja->nr ? (ja->cur_idx + 1) % ja->nr : 0;
 832                 __array_insert_item(ja->buckets,                ja->nr, pos);
 833                 __array_insert_item(ja->bucket_seq,             ja->nr, pos);
 834                 __array_insert_item(journal_buckets->buckets,   ja->nr, pos);
 835                 ja->nr++;
 836
 837                 ja->buckets[pos] = bucket;
 838                 ja->bucket_seq[pos] = 0;
 839                 journal_buckets->buckets[pos] = cpu_to_le64(bucket);
 840
 841                 if (pos <= ja->discard_idx)
 842                         ja->discard_idx = (ja->discard_idx + 1) % ja->nr;
 843                 if (pos <= ja->dirty_idx_ondisk)
 844                         ja->dirty_idx_ondisk = (ja->dirty_idx_ondisk + 1) % ja->nr;
 845                 if (pos <= ja->dirty_idx)
 846                         ja->dirty_idx = (ja->dirty_idx + 1) % ja->nr;
 847                 if (pos <= ja->cur_idx)
 848                         ja->cur_idx = (ja->cur_idx + 1) % ja->nr;
 849
 850                 bch2_mark_metadata_bucket(c, ca, bucket, BCH_DATA_JOURNAL,
 851                                           ca->mi.bucket_size,
 852                                           gc_phase(GC_PHASE_SB),
 853                                           0);
 854
 855                 if (c) {
 856                         spin_unlock(&c->journal.lock);
 857                         percpu_up_read(&c->mark_lock);
 858                 }
 859
 860                 if (!new_fs)
 861                         bch2_open_bucket_put(c, ob);
 862         }
 863
 864         ret = 0;
 865 err:
 866         kfree(new_bucket_seq);
 867         kfree(new_buckets);
 868
 869         return ret;
 870 }
 871
 872 /*
 873  * Allocate more journal space at runtime - not currently making use if it, but
 874  * the code works:
 875  */
 876 int bch2_set_nr_journal_buckets(struct bch_fs *c, struct bch_dev *ca,
 877                                 unsigned nr)
 878 {
 879         struct journal_device *ja = &ca->journal;
 880         struct closure cl;
 881         unsigned current_nr;
 882         int ret;
 883
 884         closure_init_stack(&cl);
 885
 886         do {
 887                 struct disk_reservation disk_res = { 0, 0 };
 888
 889                 closure_sync(&cl);
 890
 891                 mutex_lock(&c->sb_lock);
 892                 current_nr = ja->nr;
 893
 894                 /*
 895                  * note: journal buckets aren't really counted as _sectors_ used yet, so
 896                  * we don't need the disk reservation to avoid the BUG_ON() in buckets.c
 897                  * when space used goes up without a reservation - but we do need the
 898                  * reservation to ensure we'll actually be able to allocate:
 899                  */
 900
 901                 if (bch2_disk_reservation_get(c, &disk_res,
 902                                               bucket_to_sector(ca, nr - ja->nr), 1, 0)) {
 903                         mutex_unlock(&c->sb_lock);
 904                         return -ENOSPC;
 905                 }
 906
 907                 ret = __bch2_set_nr_journal_buckets(ca, nr, false, &cl);
 908
 909                 bch2_disk_reservation_put(c, &disk_res);
 910
 911                 if (ja->nr != current_nr)
 912                         bch2_write_super(c);
 913                 mutex_unlock(&c->sb_lock);
 914         } while (ret == -EAGAIN);
 915
 916         return ret;
 917 }
 918
 919 int bch2_dev_journal_alloc(struct bch_dev *ca)
 920 {
 921         unsigned nr;
 922
 923         if (dynamic_fault("bcachefs:add:journal_alloc"))
 924                 return -ENOMEM;
 925
 926         /*
 927          * clamp journal size to 1024 buckets or 512MB (in sectors), whichever
 928          * is smaller:
 929          */
 930         nr = clamp_t(unsigned, ca->mi.nbuckets >> 8,
 931                      BCH_JOURNAL_BUCKETS_MIN,
 932                      min(1 << 10,
 933                          (1 << 20) / ca->mi.bucket_size));
 934
 935         return __bch2_set_nr_journal_buckets(ca, nr, true, NULL);
 936 }
 937
 938 /* startup/shutdown: */
 939
 940 static bool bch2_journal_writing_to_device(struct journal *j, unsigned dev_idx)
 941 {
 942         union journal_res_state state;
 943         struct journal_buf *w;
 944         bool ret;
 945
 946         spin_lock(&j->lock);
 947         state = READ_ONCE(j->reservations);
 948         w = j->buf + !state.idx;
 949
 950         ret = state.prev_buf_unwritten &&
 951                 bch2_bkey_has_device(bkey_i_to_s_c(&w->key), dev_idx);
 952         spin_unlock(&j->lock);
 953
 954         return ret;
 955 }
 956
 957 void bch2_dev_journal_stop(struct journal *j, struct bch_dev *ca)
 958 {
 959         wait_event(j->wait, !bch2_journal_writing_to_device(j, ca->dev_idx));
 960 }
 961
 962 void bch2_fs_journal_stop(struct journal *j)
 963 {
 964         bch2_journal_flush_all_pins(j);
 965
 966         wait_event(j->wait, journal_entry_close(j));
 967
 968         /* do we need to write another journal entry? */
 969         if (test_bit(JOURNAL_NOT_EMPTY, &j->flags))
 970                 bch2_journal_meta(j);
 971
 972         journal_quiesce(j);
 973
 974         BUG_ON(!bch2_journal_error(j) &&
 975                test_bit(JOURNAL_NOT_EMPTY, &j->flags));
 976
 977         cancel_delayed_work_sync(&j->write_work);
 978         cancel_delayed_work_sync(&j->reclaim_work);
 979 }
 980
 981 int bch2_fs_journal_start(struct journal *j, u64 cur_seq,
 982                           struct list_head *journal_entries)
 983 {
 984         struct bch_fs *c = container_of(j, struct bch_fs, journal);
 985         struct journal_entry_pin_list *p;
 986         struct journal_replay *i;
 987         u64 last_seq = cur_seq, nr, seq;
 988
 989         if (!list_empty(journal_entries))
 990                 last_seq = le64_to_cpu(list_last_entry(journal_entries,
 991                                 struct journal_replay, list)->j.last_seq);
 992
 993         nr = cur_seq - last_seq;
 994
 995         if (nr + 1 > j->pin.size) {
 996                 free_fifo(&j->pin);
 997                 init_fifo(&j->pin, roundup_pow_of_two(nr + 1), GFP_KERNEL);
 998                 if (!j->pin.data) {
 999                         bch_err(c, "error reallocating journal fifo (%llu open entries)", nr);
1000                         return -ENOMEM;
1001                 }
1002         }
1003
1004         j->replay_journal_seq   = last_seq;
1005         j->replay_journal_seq_end = cur_seq;
1006         j->last_seq_ondisk      = last_seq;
1007         j->pin.front            = last_seq;
1008         j->pin.back             = cur_seq;
1009         atomic64_set(&j->seq, cur_seq - 1);
1010
1011         fifo_for_each_entry_ptr(p, &j->pin, seq) {
1012                 INIT_LIST_HEAD(&p->list);
1013                 INIT_LIST_HEAD(&p->flushed);
1014                 atomic_set(&p->count, 1);
1015                 p->devs.nr = 0;
1016         }
1017
1018         list_for_each_entry(i, journal_entries, list) {
1019                 seq = le64_to_cpu(i->j.seq);
1020                 BUG_ON(seq >= cur_seq);
1021
1022                 if (seq < last_seq)
1023                         continue;
1024
1025                 journal_seq_pin(j, seq)->devs = i->devs;
1026         }
1027
1028         spin_lock(&j->lock);
1029
1030         set_bit(JOURNAL_STARTED, &j->flags);
1031
1032         journal_pin_new_entry(j, 1);
1033         bch2_journal_buf_init(j);
1034
1035         c->last_bucket_seq_cleanup = journal_cur_seq(j);
1036
1037         bch2_journal_space_available(j);
1038         spin_unlock(&j->lock);
1039
1040         return 0;
1041 }
1042
1043 /* init/exit: */
1044
1045 void bch2_dev_journal_exit(struct bch_dev *ca)
1046 {
1047         kfree(ca->journal.bio);
1048         kfree(ca->journal.buckets);
1049         kfree(ca->journal.bucket_seq);
1050
1051         ca->journal.bio         = NULL;
1052         ca->journal.buckets     = NULL;
1053         ca->journal.bucket_seq  = NULL;
1054 }
1055
1056 int bch2_dev_journal_init(struct bch_dev *ca, struct bch_sb *sb)
1057 {
1058         struct journal_device *ja = &ca->journal;
1059         struct bch_sb_field_journal *journal_buckets =
1060                 bch2_sb_get_journal(sb);
1061         unsigned i;
1062
1063         ja->nr = bch2_nr_journal_buckets(journal_buckets);
1064
1065         ja->bucket_seq = kcalloc(ja->nr, sizeof(u64), GFP_KERNEL);
1066         if (!ja->bucket_seq)
1067                 return -ENOMEM;
1068
1069         ca->journal.bio = bio_kmalloc(GFP_KERNEL,
1070                         DIV_ROUND_UP(JOURNAL_ENTRY_SIZE_MAX, PAGE_SIZE));
1071         if (!ca->journal.bio)
1072                 return -ENOMEM;
1073
1074         ja->buckets = kcalloc(ja->nr, sizeof(u64), GFP_KERNEL);
1075         if (!ja->buckets)
1076                 return -ENOMEM;
1077
1078         for (i = 0; i < ja->nr; i++)
1079                 ja->buckets[i] = le64_to_cpu(journal_buckets->buckets[i]);
1080
1081         return 0;
1082 }
1083
1084 void bch2_fs_journal_exit(struct journal *j)
1085 {
1086         kvpfree(j->buf[1].data, j->buf[1].buf_size);
1087         kvpfree(j->buf[0].data, j->buf[0].buf_size);
1088         free_fifo(&j->pin);
1089 }
1090
1091 int bch2_fs_journal_init(struct journal *j)
1092 {
1093         struct bch_fs *c = container_of(j, struct bch_fs, journal);
1094         static struct lock_class_key res_key;
1095         int ret = 0;
1096
1097         pr_verbose_init(c->opts, "");
1098
1099         spin_lock_init(&j->lock);
1100         spin_lock_init(&j->err_lock);
1101         init_waitqueue_head(&j->wait);
1102         INIT_DELAYED_WORK(&j->write_work, journal_write_work);
1103         INIT_DELAYED_WORK(&j->reclaim_work, bch2_journal_reclaim_work);
1104         init_waitqueue_head(&j->pin_flush_wait);
1105         mutex_init(&j->reclaim_lock);
1106         mutex_init(&j->discard_lock);
1107
1108         lockdep_init_map(&j->res_map, "journal res", &res_key, 0);
1109
1110         j->buf[0].buf_size      = JOURNAL_ENTRY_SIZE_MIN;
1111         j->buf[1].buf_size      = JOURNAL_ENTRY_SIZE_MIN;
1112         j->write_delay_ms       = 1000;
1113         j->reclaim_delay_ms     = 100;
1114
1115         /* Btree roots: */
1116         j->entry_u64s_reserved +=
1117                 BTREE_ID_NR * (JSET_KEYS_U64s + BKEY_EXTENT_U64s_MAX);
1118
1119         atomic64_set(&j->reservations.counter,
1120                 ((union journal_res_state)
1121                  { .cur_entry_offset = JOURNAL_ENTRY_CLOSED_VAL }).v);
1122
1123         if (!(init_fifo(&j->pin, JOURNAL_PIN, GFP_KERNEL)) ||
1124             !(j->buf[0].data = kvpmalloc(j->buf[0].buf_size, GFP_KERNEL)) ||
1125             !(j->buf[1].data = kvpmalloc(j->buf[1].buf_size, GFP_KERNEL))) {
1126                 ret = -ENOMEM;
1127                 goto out;
1128         }
1129
1130         j->pin.front = j->pin.back = 1;
1131 out:
1132         pr_verbose_init(c->opts, "ret %i", ret);
1133         return ret;
1134 }
1135
1136 /* debug: */
1137
1138 ssize_t bch2_journal_print_debug(struct journal *j, char *buf)
1139 {
1140         struct printbuf out = _PBUF(buf, PAGE_SIZE);
1141         struct bch_fs *c = container_of(j, struct bch_fs, journal);
1142         union journal_res_state s;
1143         struct bch_dev *ca;
1144         unsigned iter;
1145
1146         rcu_read_lock();
1147         spin_lock(&j->lock);
1148         s = READ_ONCE(j->reservations);
1149
1150         pr_buf(&out,
1151                "active journal entries:\t%llu\n"
1152                "seq:\t\t\t%llu\n"
1153                "last_seq:\t\t%llu\n"
1154                "last_seq_ondisk:\t%llu\n"
1155                "prereserved:\t\t%u/%u\n"
1156                "current entry sectors:\t%u\n"
1157                "current entry:\t\t",
1158                fifo_used(&j->pin),
1159                journal_cur_seq(j),
1160                journal_last_seq(j),
1161                j->last_seq_ondisk,
1162                j->prereserved.reserved,
1163                j->prereserved.remaining,
1164                j->cur_entry_sectors);
1165
1166         switch (s.cur_entry_offset) {
1167         case JOURNAL_ENTRY_ERROR_VAL:
1168                 pr_buf(&out, "error\n");
1169                 break;
1170         case JOURNAL_ENTRY_CLOSED_VAL:
1171                 pr_buf(&out, "closed\n");
1172                 break;
1173         default:
1174                 pr_buf(&out, "%u/%u\n",
1175                        s.cur_entry_offset,
1176                        j->cur_entry_u64s);
1177                 break;
1178         }
1179
1180         pr_buf(&out,
1181                "current entry refs:\t%u\n"
1182                "prev entry unwritten:\t",
1183                journal_state_count(s, s.idx));
1184
1185         if (s.prev_buf_unwritten)
1186                 pr_buf(&out, "yes, ref %u sectors %u\n",
1187                        journal_state_count(s, !s.idx),
1188                        journal_prev_buf(j)->sectors);
1189         else
1190                 pr_buf(&out, "no\n");
1191
1192         pr_buf(&out,
1193                "need write:\t\t%i\n"
1194                "replay done:\t\t%i\n",
1195                test_bit(JOURNAL_NEED_WRITE,     &j->flags),
1196                test_bit(JOURNAL_REPLAY_DONE,    &j->flags));
1197
1198         for_each_member_device_rcu(ca, c, iter,
1199                                    &c->rw_devs[BCH_DATA_JOURNAL]) {
1200                 struct journal_device *ja = &ca->journal;
1201
1202                 if (!ja->nr)
1203                         continue;
1204
1205                 pr_buf(&out,
1206                        "dev %u:\n"
1207                        "\tnr\t\t%u\n"
1208                        "\tavailable\t%u:%u\n"
1209                        "\tdiscard_idx\t\t%u\n"
1210                        "\tdirty_idx_ondisk\t%u (seq %llu)\n"
1211                        "\tdirty_idx\t\t%u (seq %llu)\n"
1212                        "\tcur_idx\t\t%u (seq %llu)\n",
1213                        iter, ja->nr,
1214                        bch2_journal_dev_buckets_available(j, ja, journal_space_discarded),
1215                        ja->sectors_free,
1216                        ja->discard_idx,
1217                        ja->dirty_idx_ondisk,    ja->bucket_seq[ja->dirty_idx_ondisk],
1218                        ja->dirty_idx,           ja->bucket_seq[ja->dirty_idx],
1219                        ja->cur_idx,             ja->bucket_seq[ja->cur_idx]);
1220         }
1221
1222         spin_unlock(&j->lock);
1223         rcu_read_unlock();
1224
1225         return out.pos - buf;
1226 }
1227
1228 ssize_t bch2_journal_print_pins(struct journal *j, char *buf)
1229 {
1230         struct printbuf out = _PBUF(buf, PAGE_SIZE);
1231         struct journal_entry_pin_list *pin_list;
1232         struct journal_entry_pin *pin;
1233         u64 i;
1234
1235         spin_lock(&j->lock);
1236         fifo_for_each_entry_ptr(pin_list, &j->pin, i) {
1237                 pr_buf(&out, "%llu: count %u\n",
1238                        i, atomic_read(&pin_list->count));
1239
1240                 list_for_each_entry(pin, &pin_list->list, list)
1241                         pr_buf(&out, "\t%px %ps\n",
1242                                pin, pin->flush);
1243
1244                 if (!list_empty(&pin_list->flushed))
1245                         pr_buf(&out, "flushed:\n");
1246
1247                 list_for_each_entry(pin, &pin_list->flushed, list)
1248                         pr_buf(&out, "\t%px %ps\n",
1249                                pin, pin->flush);
1250         }
1251         spin_unlock(&j->lock);
1252
1253         return out.pos - buf;
1254 }