]> git.sesse.net Git - bcachefs-tools-debian/blob - libbcachefs/journal_reclaim.c
Update bcachefs sources to da8056a215 bcachefs: Store logical location of journal...
[bcachefs-tools-debian] / libbcachefs / journal_reclaim.c
1 // SPDX-License-Identifier: GPL-2.0
2
3 #include "bcachefs.h"
4 #include "btree_key_cache.h"
5 #include "error.h"
6 #include "journal.h"
7 #include "journal_io.h"
8 #include "journal_reclaim.h"
9 #include "replicas.h"
10 #include "super.h"
11
12 #include <linux/kthread.h>
13 #include <linux/sched/mm.h>
14 #include <trace/events/bcachefs.h>
15
16 /* Free space calculations: */
17
18 static unsigned journal_space_from(struct journal_device *ja,
19                                    enum journal_space_from from)
20 {
21         switch (from) {
22         case journal_space_discarded:
23                 return ja->discard_idx;
24         case journal_space_clean_ondisk:
25                 return ja->dirty_idx_ondisk;
26         case journal_space_clean:
27                 return ja->dirty_idx;
28         default:
29                 BUG();
30         }
31 }
32
33 unsigned bch2_journal_dev_buckets_available(struct journal *j,
34                                             struct journal_device *ja,
35                                             enum journal_space_from from)
36 {
37         unsigned available = !test_bit(JOURNAL_NOCHANGES, &j->flags)
38                 ? ((journal_space_from(ja, from) -
39                     ja->cur_idx - 1 + ja->nr) % ja->nr)
40                 : ja->nr;
41
42         /*
43          * Don't use the last bucket unless writing the new last_seq
44          * will make another bucket available:
45          */
46         if (available && ja->dirty_idx_ondisk == ja->dirty_idx)
47                 --available;
48
49         return available;
50 }
51
52 static void journal_set_remaining(struct journal *j, unsigned u64s_remaining)
53 {
54         union journal_preres_state old, new;
55         u64 v = atomic64_read(&j->prereserved.counter);
56
57         do {
58                 old.v = new.v = v;
59                 new.remaining = u64s_remaining;
60         } while ((v = atomic64_cmpxchg(&j->prereserved.counter,
61                                        old.v, new.v)) != old.v);
62 }
63
64 static inline unsigned get_unwritten_sectors(struct journal *j, unsigned *idx)
65 {
66         unsigned sectors = 0;
67
68         while (!sectors && *idx != j->reservations.idx) {
69                 sectors = j->buf[*idx].sectors;
70
71                 *idx = (*idx + 1) & JOURNAL_BUF_MASK;
72         }
73
74         return sectors;
75 }
76
77 static struct journal_space
78 journal_dev_space_available(struct journal *j, struct bch_dev *ca,
79                             enum journal_space_from from)
80 {
81         struct journal_device *ja = &ca->journal;
82         unsigned sectors, buckets, unwritten, idx = j->reservations.unwritten_idx;
83
84         if (from == journal_space_total)
85                 return (struct journal_space) {
86                         .next_entry     = ca->mi.bucket_size,
87                         .total          = ca->mi.bucket_size * ja->nr,
88                 };
89
90         buckets = bch2_journal_dev_buckets_available(j, ja, from);
91         sectors = ja->sectors_free;
92
93         /*
94          * We that we don't allocate the space for a journal entry
95          * until we write it out - thus, account for it here:
96          */
97         while ((unwritten = get_unwritten_sectors(j, &idx))) {
98                 /* entry won't fit on this device, skip: */
99                 if (unwritten > ca->mi.bucket_size)
100                         continue;
101
102                 if (unwritten >= sectors) {
103                         if (!buckets) {
104                                 sectors = 0;
105                                 break;
106                         }
107
108                         buckets--;
109                         sectors = ca->mi.bucket_size;
110                 }
111
112                 sectors -= unwritten;
113         }
114
115         if (sectors < ca->mi.bucket_size && buckets) {
116                 buckets--;
117                 sectors = ca->mi.bucket_size;
118         }
119
120         return (struct journal_space) {
121                 .next_entry     = sectors,
122                 .total          = sectors + buckets * ca->mi.bucket_size,
123         };
124 }
125
126 static struct journal_space __journal_space_available(struct journal *j, unsigned nr_devs_want,
127                             enum journal_space_from from)
128 {
129         struct bch_fs *c = container_of(j, struct bch_fs, journal);
130         struct bch_dev *ca;
131         unsigned i, pos, nr_devs = 0;
132         struct journal_space space, dev_space[BCH_SB_MEMBERS_MAX];
133
134         BUG_ON(nr_devs_want > ARRAY_SIZE(dev_space));
135
136         rcu_read_lock();
137         for_each_member_device_rcu(ca, c, i,
138                                    &c->rw_devs[BCH_DATA_journal]) {
139                 if (!ca->journal.nr)
140                         continue;
141
142                 space = journal_dev_space_available(j, ca, from);
143                 if (!space.next_entry)
144                         continue;
145
146                 for (pos = 0; pos < nr_devs; pos++)
147                         if (space.total > dev_space[pos].total)
148                                 break;
149
150                 array_insert_item(dev_space, nr_devs, pos, space);
151         }
152         rcu_read_unlock();
153
154         if (nr_devs < nr_devs_want)
155                 return (struct journal_space) { 0, 0 };
156
157         /*
158          * We sorted largest to smallest, and we want the smallest out of the
159          * @nr_devs_want largest devices:
160          */
161         return dev_space[nr_devs_want - 1];
162 }
163
164 void bch2_journal_space_available(struct journal *j)
165 {
166         struct bch_fs *c = container_of(j, struct bch_fs, journal);
167         struct bch_dev *ca;
168         unsigned clean, clean_ondisk, total;
169         s64 u64s_remaining = 0;
170         unsigned max_entry_size  = min(j->buf[0].buf_size >> 9,
171                                        j->buf[1].buf_size >> 9);
172         unsigned i, nr_online = 0, nr_devs_want;
173         bool can_discard = false;
174         int ret = 0;
175
176         lockdep_assert_held(&j->lock);
177
178         rcu_read_lock();
179         for_each_member_device_rcu(ca, c, i,
180                                    &c->rw_devs[BCH_DATA_journal]) {
181                 struct journal_device *ja = &ca->journal;
182
183                 if (!ja->nr)
184                         continue;
185
186                 while (ja->dirty_idx != ja->cur_idx &&
187                        ja->bucket_seq[ja->dirty_idx] < journal_last_seq(j))
188                         ja->dirty_idx = (ja->dirty_idx + 1) % ja->nr;
189
190                 while (ja->dirty_idx_ondisk != ja->dirty_idx &&
191                        ja->bucket_seq[ja->dirty_idx_ondisk] < j->last_seq_ondisk)
192                         ja->dirty_idx_ondisk = (ja->dirty_idx_ondisk + 1) % ja->nr;
193
194                 if (ja->discard_idx != ja->dirty_idx_ondisk)
195                         can_discard = true;
196
197                 max_entry_size = min_t(unsigned, max_entry_size, ca->mi.bucket_size);
198                 nr_online++;
199         }
200         rcu_read_unlock();
201
202         j->can_discard = can_discard;
203
204         if (nr_online < c->opts.metadata_replicas_required) {
205                 ret = cur_entry_insufficient_devices;
206                 goto out;
207         }
208
209         nr_devs_want = min_t(unsigned, nr_online, c->opts.metadata_replicas);
210
211         for (i = 0; i < journal_space_nr; i++)
212                 j->space[i] = __journal_space_available(j, nr_devs_want, i);
213
214         clean_ondisk    = j->space[journal_space_clean_ondisk].total;
215         clean           = j->space[journal_space_clean].total;
216         total           = j->space[journal_space_total].total;
217
218         if (!clean_ondisk &&
219             j->reservations.idx ==
220             j->reservations.unwritten_idx) {
221                 char *buf = kmalloc(4096, GFP_ATOMIC);
222
223                 bch_err(c, "journal stuck");
224                 if (buf) {
225                         __bch2_journal_debug_to_text(&_PBUF(buf, 4096), j);
226                         pr_err("\n%s", buf);
227                         kfree(buf);
228                 }
229
230                 bch2_fatal_error(c);
231                 ret = cur_entry_journal_stuck;
232         } else if (!j->space[journal_space_discarded].next_entry)
233                 ret = cur_entry_journal_full;
234         else if (!fifo_free(&j->pin))
235                 ret = cur_entry_journal_pin_full;
236
237         if ((j->space[journal_space_clean_ondisk].next_entry <
238              j->space[journal_space_clean_ondisk].total) &&
239             (clean - clean_ondisk <= total / 8) &&
240             (clean_ondisk * 2 > clean ))
241                 set_bit(JOURNAL_MAY_SKIP_FLUSH, &j->flags);
242         else
243                 clear_bit(JOURNAL_MAY_SKIP_FLUSH, &j->flags);
244
245         u64s_remaining  = (u64) clean << 6;
246         u64s_remaining -= (u64) total << 3;
247         u64s_remaining = max(0LL, u64s_remaining);
248         u64s_remaining /= 4;
249         u64s_remaining = min_t(u64, u64s_remaining, U32_MAX);
250 out:
251         j->cur_entry_sectors    = !ret ? j->space[journal_space_discarded].next_entry : 0;
252         j->cur_entry_error      = ret;
253         journal_set_remaining(j, u64s_remaining);
254         journal_check_may_get_unreserved(j);
255
256         if (!ret)
257                 journal_wake(j);
258 }
259
260 /* Discards - last part of journal reclaim: */
261
262 static bool should_discard_bucket(struct journal *j, struct journal_device *ja)
263 {
264         bool ret;
265
266         spin_lock(&j->lock);
267         ret = ja->discard_idx != ja->dirty_idx_ondisk;
268         spin_unlock(&j->lock);
269
270         return ret;
271 }
272
273 /*
274  * Advance ja->discard_idx as long as it points to buckets that are no longer
275  * dirty, issuing discards if necessary:
276  */
277 void bch2_journal_do_discards(struct journal *j)
278 {
279         struct bch_fs *c = container_of(j, struct bch_fs, journal);
280         struct bch_dev *ca;
281         unsigned iter;
282
283         mutex_lock(&j->discard_lock);
284
285         for_each_rw_member(ca, c, iter) {
286                 struct journal_device *ja = &ca->journal;
287
288                 while (should_discard_bucket(j, ja)) {
289                         if (ca->mi.discard &&
290                             blk_queue_discard(bdev_get_queue(ca->disk_sb.bdev)))
291                                 blkdev_issue_discard(ca->disk_sb.bdev,
292                                         bucket_to_sector(ca,
293                                                 ja->buckets[ja->discard_idx]),
294                                         ca->mi.bucket_size, GFP_NOIO, 0);
295
296                         spin_lock(&j->lock);
297                         ja->discard_idx = (ja->discard_idx + 1) % ja->nr;
298
299                         bch2_journal_space_available(j);
300                         spin_unlock(&j->lock);
301                 }
302         }
303
304         mutex_unlock(&j->discard_lock);
305 }
306
307 /*
308  * Journal entry pinning - machinery for holding a reference on a given journal
309  * entry, holding it open to ensure it gets replayed during recovery:
310  */
311
312 static void bch2_journal_reclaim_fast(struct journal *j)
313 {
314         struct journal_entry_pin_list temp;
315         bool popped = false;
316
317         lockdep_assert_held(&j->lock);
318
319         /*
320          * Unpin journal entries whose reference counts reached zero, meaning
321          * all btree nodes got written out
322          */
323         while (!fifo_empty(&j->pin) &&
324                !atomic_read(&fifo_peek_front(&j->pin).count)) {
325                 BUG_ON(!list_empty(&fifo_peek_front(&j->pin).list));
326                 BUG_ON(!list_empty(&fifo_peek_front(&j->pin).flushed));
327                 BUG_ON(!fifo_pop(&j->pin, temp));
328                 popped = true;
329         }
330
331         if (popped)
332                 bch2_journal_space_available(j);
333 }
334
335 void __bch2_journal_pin_put(struct journal *j, u64 seq)
336 {
337         struct journal_entry_pin_list *pin_list = journal_seq_pin(j, seq);
338
339         if (atomic_dec_and_test(&pin_list->count))
340                 bch2_journal_reclaim_fast(j);
341 }
342
343 void bch2_journal_pin_put(struct journal *j, u64 seq)
344 {
345         struct journal_entry_pin_list *pin_list = journal_seq_pin(j, seq);
346
347         if (atomic_dec_and_test(&pin_list->count)) {
348                 spin_lock(&j->lock);
349                 bch2_journal_reclaim_fast(j);
350                 spin_unlock(&j->lock);
351         }
352 }
353
354 static inline void __journal_pin_drop(struct journal *j,
355                                       struct journal_entry_pin *pin)
356 {
357         struct journal_entry_pin_list *pin_list;
358
359         if (!journal_pin_active(pin))
360                 return;
361
362         if (j->flush_in_progress == pin)
363                 j->flush_in_progress_dropped = true;
364
365         pin_list = journal_seq_pin(j, pin->seq);
366         pin->seq = 0;
367         list_del_init(&pin->list);
368
369         /*
370          * Unpinning a journal entry make make journal_next_bucket() succeed, if
371          * writing a new last_seq will now make another bucket available:
372          */
373         if (atomic_dec_and_test(&pin_list->count) &&
374             pin_list == &fifo_peek_front(&j->pin))
375                 bch2_journal_reclaim_fast(j);
376         else if (fifo_used(&j->pin) == 1 &&
377                  atomic_read(&pin_list->count) == 1)
378                 journal_wake(j);
379 }
380
381 void bch2_journal_pin_drop(struct journal *j,
382                            struct journal_entry_pin *pin)
383 {
384         spin_lock(&j->lock);
385         __journal_pin_drop(j, pin);
386         spin_unlock(&j->lock);
387 }
388
389 void bch2_journal_pin_set(struct journal *j, u64 seq,
390                           struct journal_entry_pin *pin,
391                           journal_pin_flush_fn flush_fn)
392 {
393         struct journal_entry_pin_list *pin_list;
394
395         spin_lock(&j->lock);
396
397         if (seq < journal_last_seq(j)) {
398                 /*
399                  * bch2_journal_pin_copy() raced with bch2_journal_pin_drop() on
400                  * the src pin - with the pin dropped, the entry to pin might no
401                  * longer to exist, but that means there's no longer anything to
402                  * copy and we can bail out here:
403                  */
404                 spin_unlock(&j->lock);
405                 return;
406         }
407
408         pin_list = journal_seq_pin(j, seq);
409
410         __journal_pin_drop(j, pin);
411
412         atomic_inc(&pin_list->count);
413         pin->seq        = seq;
414         pin->flush      = flush_fn;
415
416         if (flush_fn == bch2_btree_key_cache_journal_flush)
417                 list_add(&pin->list, &pin_list->key_cache_list);
418         else if (flush_fn)
419                 list_add(&pin->list, &pin_list->list);
420         else
421                 list_add(&pin->list, &pin_list->flushed);
422         spin_unlock(&j->lock);
423
424         /*
425          * If the journal is currently full,  we might want to call flush_fn
426          * immediately:
427          */
428         journal_wake(j);
429 }
430
431 /**
432  * bch2_journal_pin_flush: ensure journal pin callback is no longer running
433  */
434 void bch2_journal_pin_flush(struct journal *j, struct journal_entry_pin *pin)
435 {
436         BUG_ON(journal_pin_active(pin));
437
438         wait_event(j->pin_flush_wait, j->flush_in_progress != pin);
439 }
440
441 /*
442  * Journal reclaim: flush references to open journal entries to reclaim space in
443  * the journal
444  *
445  * May be done by the journal code in the background as needed to free up space
446  * for more journal entries, or as part of doing a clean shutdown, or to migrate
447  * data off of a specific device:
448  */
449
450 static struct journal_entry_pin *
451 journal_get_next_pin(struct journal *j,
452                      bool get_any,
453                      bool get_key_cache,
454                      u64 max_seq, u64 *seq)
455 {
456         struct journal_entry_pin_list *pin_list;
457         struct journal_entry_pin *ret = NULL;
458
459         fifo_for_each_entry_ptr(pin_list, &j->pin, *seq) {
460                 if (*seq > max_seq && !get_any && !get_key_cache)
461                         break;
462
463                 if (*seq <= max_seq || get_any) {
464                         ret = list_first_entry_or_null(&pin_list->list,
465                                 struct journal_entry_pin, list);
466                         if (ret)
467                                 return ret;
468                 }
469
470                 if (*seq <= max_seq || get_any || get_key_cache) {
471                         ret = list_first_entry_or_null(&pin_list->key_cache_list,
472                                 struct journal_entry_pin, list);
473                         if (ret)
474                                 return ret;
475                 }
476         }
477
478         return NULL;
479 }
480
481 /* returns true if we did work */
482 static size_t journal_flush_pins(struct journal *j, u64 seq_to_flush,
483                                  unsigned min_any,
484                                  unsigned min_key_cache)
485 {
486         struct journal_entry_pin *pin;
487         size_t nr_flushed = 0;
488         journal_pin_flush_fn flush_fn;
489         u64 seq;
490         int err;
491
492         lockdep_assert_held(&j->reclaim_lock);
493
494         while (1) {
495                 cond_resched();
496
497                 j->last_flushed = jiffies;
498
499                 spin_lock(&j->lock);
500                 pin = journal_get_next_pin(j,
501                                            min_any != 0,
502                                            min_key_cache != 0,
503                                            seq_to_flush, &seq);
504                 if (pin) {
505                         BUG_ON(j->flush_in_progress);
506                         j->flush_in_progress = pin;
507                         j->flush_in_progress_dropped = false;
508                         flush_fn = pin->flush;
509                 }
510                 spin_unlock(&j->lock);
511
512                 if (!pin)
513                         break;
514
515                 if (min_key_cache && pin->flush == bch2_btree_key_cache_journal_flush)
516                         min_key_cache--;
517
518                 if (min_any)
519                         min_any--;
520
521                 err = flush_fn(j, pin, seq);
522
523                 spin_lock(&j->lock);
524                 /* Pin might have been dropped or rearmed: */
525                 if (likely(!err && !j->flush_in_progress_dropped))
526                         list_move(&pin->list, &journal_seq_pin(j, seq)->flushed);
527                 j->flush_in_progress = NULL;
528                 j->flush_in_progress_dropped = false;
529                 spin_unlock(&j->lock);
530
531                 wake_up(&j->pin_flush_wait);
532
533                 if (err)
534                         break;
535
536                 nr_flushed++;
537         }
538
539         return nr_flushed;
540 }
541
542 static u64 journal_seq_to_flush(struct journal *j)
543 {
544         struct bch_fs *c = container_of(j, struct bch_fs, journal);
545         struct bch_dev *ca;
546         u64 seq_to_flush = 0;
547         unsigned iter;
548
549         spin_lock(&j->lock);
550
551         for_each_rw_member(ca, c, iter) {
552                 struct journal_device *ja = &ca->journal;
553                 unsigned nr_buckets, bucket_to_flush;
554
555                 if (!ja->nr)
556                         continue;
557
558                 /* Try to keep the journal at most half full: */
559                 nr_buckets = ja->nr / 2;
560
561                 /* And include pre-reservations: */
562                 nr_buckets += DIV_ROUND_UP(j->prereserved.reserved,
563                                            (ca->mi.bucket_size << 6) -
564                                            journal_entry_overhead(j));
565
566                 nr_buckets = min(nr_buckets, ja->nr);
567
568                 bucket_to_flush = (ja->cur_idx + nr_buckets) % ja->nr;
569                 seq_to_flush = max(seq_to_flush,
570                                    ja->bucket_seq[bucket_to_flush]);
571         }
572
573         /* Also flush if the pin fifo is more than half full */
574         seq_to_flush = max_t(s64, seq_to_flush,
575                              (s64) journal_cur_seq(j) -
576                              (j->pin.size >> 1));
577         spin_unlock(&j->lock);
578
579         return seq_to_flush;
580 }
581
582 /**
583  * bch2_journal_reclaim - free up journal buckets
584  *
585  * Background journal reclaim writes out btree nodes. It should be run
586  * early enough so that we never completely run out of journal buckets.
587  *
588  * High watermarks for triggering background reclaim:
589  * - FIFO has fewer than 512 entries left
590  * - fewer than 25% journal buckets free
591  *
592  * Background reclaim runs until low watermarks are reached:
593  * - FIFO has more than 1024 entries left
594  * - more than 50% journal buckets free
595  *
596  * As long as a reclaim can complete in the time it takes to fill up
597  * 512 journal entries or 25% of all journal buckets, then
598  * journal_next_bucket() should not stall.
599  */
600 static int __bch2_journal_reclaim(struct journal *j, bool direct)
601 {
602         struct bch_fs *c = container_of(j, struct bch_fs, journal);
603         bool kthread = (current->flags & PF_KTHREAD) != 0;
604         u64 seq_to_flush;
605         size_t min_nr, min_key_cache, nr_flushed;
606         unsigned flags;
607         int ret = 0;
608
609         /*
610          * We can't invoke memory reclaim while holding the reclaim_lock -
611          * journal reclaim is required to make progress for memory reclaim
612          * (cleaning the caches), so we can't get stuck in memory reclaim while
613          * we're holding the reclaim lock:
614          */
615         lockdep_assert_held(&j->reclaim_lock);
616         flags = memalloc_noreclaim_save();
617
618         do {
619                 if (kthread && kthread_should_stop())
620                         break;
621
622                 if (bch2_journal_error(j)) {
623                         ret = -EIO;
624                         break;
625                 }
626
627                 bch2_journal_do_discards(j);
628
629                 seq_to_flush = journal_seq_to_flush(j);
630                 min_nr = 0;
631
632                 /*
633                  * If it's been longer than j->reclaim_delay_ms since we last flushed,
634                  * make sure to flush at least one journal pin:
635                  */
636                 if (time_after(jiffies, j->last_flushed +
637                                msecs_to_jiffies(c->opts.journal_reclaim_delay)))
638                         min_nr = 1;
639
640                 if (j->prereserved.reserved * 4 > j->prereserved.remaining)
641                         min_nr = 1;
642
643                 if (fifo_free(&j->pin) <= 32)
644                         min_nr = 1;
645
646                 if (atomic_read(&c->btree_cache.dirty) * 2 > c->btree_cache.used)
647                         min_nr = 1;
648
649                 trace_journal_reclaim_start(c,
650                                 min_nr,
651                                 j->prereserved.reserved,
652                                 j->prereserved.remaining,
653                                 atomic_read(&c->btree_cache.dirty),
654                                 c->btree_cache.used,
655                                 atomic_long_read(&c->btree_key_cache.nr_dirty),
656                                 atomic_long_read(&c->btree_key_cache.nr_keys));
657
658                 min_key_cache = min(bch2_nr_btree_keys_need_flush(c), (size_t) 128);
659
660                 nr_flushed = journal_flush_pins(j, seq_to_flush,
661                                                 min_nr, min_key_cache);
662
663                 if (direct)
664                         j->nr_direct_reclaim += nr_flushed;
665                 else
666                         j->nr_background_reclaim += nr_flushed;
667                 trace_journal_reclaim_finish(c, nr_flushed);
668
669                 if (nr_flushed)
670                         wake_up(&j->reclaim_wait);
671         } while ((min_nr || min_key_cache) && !direct);
672
673         memalloc_noreclaim_restore(flags);
674
675         return ret;
676 }
677
678 int bch2_journal_reclaim(struct journal *j)
679 {
680         return __bch2_journal_reclaim(j, true);
681 }
682
683 static int bch2_journal_reclaim_thread(void *arg)
684 {
685         struct journal *j = arg;
686         struct bch_fs *c = container_of(j, struct bch_fs, journal);
687         unsigned long delay, now;
688         int ret = 0;
689
690         set_freezable();
691
692         j->last_flushed = jiffies;
693
694         while (!ret && !kthread_should_stop()) {
695                 j->reclaim_kicked = false;
696
697                 mutex_lock(&j->reclaim_lock);
698                 ret = __bch2_journal_reclaim(j, false);
699                 mutex_unlock(&j->reclaim_lock);
700
701                 now = jiffies;
702                 delay = msecs_to_jiffies(c->opts.journal_reclaim_delay);
703                 j->next_reclaim = j->last_flushed + delay;
704
705                 if (!time_in_range(j->next_reclaim, now, now + delay))
706                         j->next_reclaim = now + delay;
707
708                 while (1) {
709                         set_current_state(TASK_INTERRUPTIBLE);
710                         if (kthread_should_stop())
711                                 break;
712                         if (j->reclaim_kicked)
713                                 break;
714                         if (time_after_eq(jiffies, j->next_reclaim))
715                                 break;
716                         freezable_schedule_timeout(j->next_reclaim - jiffies);
717
718                 }
719                 __set_current_state(TASK_RUNNING);
720         }
721
722         return 0;
723 }
724
725 void bch2_journal_reclaim_stop(struct journal *j)
726 {
727         struct task_struct *p = j->reclaim_thread;
728
729         j->reclaim_thread = NULL;
730
731         if (p) {
732                 kthread_stop(p);
733                 put_task_struct(p);
734         }
735 }
736
737 int bch2_journal_reclaim_start(struct journal *j)
738 {
739         struct bch_fs *c = container_of(j, struct bch_fs, journal);
740         struct task_struct *p;
741
742         if (j->reclaim_thread)
743                 return 0;
744
745         p = kthread_create(bch2_journal_reclaim_thread, j,
746                            "bch-reclaim/%s", c->name);
747         if (IS_ERR(p)) {
748                 bch_err(c, "error creating journal reclaim thread: %li", PTR_ERR(p));
749                 return PTR_ERR(p);
750         }
751
752         get_task_struct(p);
753         j->reclaim_thread = p;
754         wake_up_process(p);
755         return 0;
756 }
757
758 static int journal_flush_done(struct journal *j, u64 seq_to_flush,
759                               bool *did_work)
760 {
761         int ret;
762
763         ret = bch2_journal_error(j);
764         if (ret)
765                 return ret;
766
767         mutex_lock(&j->reclaim_lock);
768
769         *did_work = journal_flush_pins(j, seq_to_flush, 0, 0) != 0;
770
771         spin_lock(&j->lock);
772         /*
773          * If journal replay hasn't completed, the unreplayed journal entries
774          * hold refs on their corresponding sequence numbers
775          */
776         ret = !test_bit(JOURNAL_REPLAY_DONE, &j->flags) ||
777                 journal_last_seq(j) > seq_to_flush ||
778                 (fifo_used(&j->pin) == 1 &&
779                  atomic_read(&fifo_peek_front(&j->pin).count) == 1);
780
781         spin_unlock(&j->lock);
782         mutex_unlock(&j->reclaim_lock);
783
784         return ret;
785 }
786
787 bool bch2_journal_flush_pins(struct journal *j, u64 seq_to_flush)
788 {
789         bool did_work = false;
790
791         if (!test_bit(JOURNAL_STARTED, &j->flags))
792                 return false;
793
794         closure_wait_event(&j->async_wait,
795                 journal_flush_done(j, seq_to_flush, &did_work));
796
797         return did_work;
798 }
799
800 int bch2_journal_flush_device_pins(struct journal *j, int dev_idx)
801 {
802         struct bch_fs *c = container_of(j, struct bch_fs, journal);
803         struct journal_entry_pin_list *p;
804         u64 iter, seq = 0;
805         int ret = 0;
806
807         spin_lock(&j->lock);
808         fifo_for_each_entry_ptr(p, &j->pin, iter)
809                 if (dev_idx >= 0
810                     ? bch2_dev_list_has_dev(p->devs, dev_idx)
811                     : p->devs.nr < c->opts.metadata_replicas)
812                         seq = iter;
813         spin_unlock(&j->lock);
814
815         bch2_journal_flush_pins(j, seq);
816
817         ret = bch2_journal_error(j);
818         if (ret)
819                 return ret;
820
821         mutex_lock(&c->replicas_gc_lock);
822         bch2_replicas_gc_start(c, 1 << BCH_DATA_journal);
823
824         seq = 0;
825
826         spin_lock(&j->lock);
827         while (!ret && seq < j->pin.back) {
828                 struct bch_replicas_padded replicas;
829
830                 seq = max(seq, journal_last_seq(j));
831                 bch2_devlist_to_replicas(&replicas.e, BCH_DATA_journal,
832                                          journal_seq_pin(j, seq)->devs);
833                 seq++;
834
835                 spin_unlock(&j->lock);
836                 ret = bch2_mark_replicas(c, &replicas.e);
837                 spin_lock(&j->lock);
838         }
839         spin_unlock(&j->lock);
840
841         ret = bch2_replicas_gc_end(c, ret);
842         mutex_unlock(&c->replicas_gc_lock);
843
844         return ret;
845 }