Merge tag 'for-linus-20180929' of git://git.kernel.dk/linux-block
[muen/linux.git] / drivers / md / bcache / journal.c
1 // SPDX-License-Identifier: GPL-2.0
2 /*
3  * bcache journalling code, for btree insertions
4  *
5  * Copyright 2012 Google, Inc.
6  */
7
8 #include "bcache.h"
9 #include "btree.h"
10 #include "debug.h"
11 #include "extents.h"
12
13 #include <trace/events/bcache.h>
14
15 /*
16  * Journal replay/recovery:
17  *
18  * This code is all driven from run_cache_set(); we first read the journal
19  * entries, do some other stuff, then we mark all the keys in the journal
20  * entries (same as garbage collection would), then we replay them - reinserting
21  * them into the cache in precisely the same order as they appear in the
22  * journal.
23  *
24  * We only journal keys that go in leaf nodes, which simplifies things quite a
25  * bit.
26  */
27
28 static void journal_read_endio(struct bio *bio)
29 {
30         struct closure *cl = bio->bi_private;
31
32         closure_put(cl);
33 }
34
35 static int journal_read_bucket(struct cache *ca, struct list_head *list,
36                                unsigned int bucket_index)
37 {
38         struct journal_device *ja = &ca->journal;
39         struct bio *bio = &ja->bio;
40
41         struct journal_replay *i;
42         struct jset *j, *data = ca->set->journal.w[0].data;
43         struct closure cl;
44         unsigned int len, left, offset = 0;
45         int ret = 0;
46         sector_t bucket = bucket_to_sector(ca->set, ca->sb.d[bucket_index]);
47
48         closure_init_stack(&cl);
49
50         pr_debug("reading %u", bucket_index);
51
52         while (offset < ca->sb.bucket_size) {
53 reread:         left = ca->sb.bucket_size - offset;
54                 len = min_t(unsigned int, left, PAGE_SECTORS << JSET_BITS);
55
56                 bio_reset(bio);
57                 bio->bi_iter.bi_sector  = bucket + offset;
58                 bio_set_dev(bio, ca->bdev);
59                 bio->bi_iter.bi_size    = len << 9;
60
61                 bio->bi_end_io  = journal_read_endio;
62                 bio->bi_private = &cl;
63                 bio_set_op_attrs(bio, REQ_OP_READ, 0);
64                 bch_bio_map(bio, data);
65
66                 closure_bio_submit(ca->set, bio, &cl);
67                 closure_sync(&cl);
68
69                 /* This function could be simpler now since we no longer write
70                  * journal entries that overlap bucket boundaries; this means
71                  * the start of a bucket will always have a valid journal entry
72                  * if it has any journal entries at all.
73                  */
74
75                 j = data;
76                 while (len) {
77                         struct list_head *where;
78                         size_t blocks, bytes = set_bytes(j);
79
80                         if (j->magic != jset_magic(&ca->sb)) {
81                                 pr_debug("%u: bad magic", bucket_index);
82                                 return ret;
83                         }
84
85                         if (bytes > left << 9 ||
86                             bytes > PAGE_SIZE << JSET_BITS) {
87                                 pr_info("%u: too big, %zu bytes, offset %u",
88                                         bucket_index, bytes, offset);
89                                 return ret;
90                         }
91
92                         if (bytes > len << 9)
93                                 goto reread;
94
95                         if (j->csum != csum_set(j)) {
96                                 pr_info("%u: bad csum, %zu bytes, offset %u",
97                                         bucket_index, bytes, offset);
98                                 return ret;
99                         }
100
101                         blocks = set_blocks(j, block_bytes(ca->set));
102
103                         while (!list_empty(list)) {
104                                 i = list_first_entry(list,
105                                         struct journal_replay, list);
106                                 if (i->j.seq >= j->last_seq)
107                                         break;
108                                 list_del(&i->list);
109                                 kfree(i);
110                         }
111
112                         list_for_each_entry_reverse(i, list, list) {
113                                 if (j->seq == i->j.seq)
114                                         goto next_set;
115
116                                 if (j->seq < i->j.last_seq)
117                                         goto next_set;
118
119                                 if (j->seq > i->j.seq) {
120                                         where = &i->list;
121                                         goto add;
122                                 }
123                         }
124
125                         where = list;
126 add:
127                         i = kmalloc(offsetof(struct journal_replay, j) +
128                                     bytes, GFP_KERNEL);
129                         if (!i)
130                                 return -ENOMEM;
131                         memcpy(&i->j, j, bytes);
132                         list_add(&i->list, where);
133                         ret = 1;
134
135                         ja->seq[bucket_index] = j->seq;
136 next_set:
137                         offset  += blocks * ca->sb.block_size;
138                         len     -= blocks * ca->sb.block_size;
139                         j = ((void *) j) + blocks * block_bytes(ca);
140                 }
141         }
142
143         return ret;
144 }
145
146 int bch_journal_read(struct cache_set *c, struct list_head *list)
147 {
148 #define read_bucket(b)                                                  \
149         ({                                                              \
150                 int ret = journal_read_bucket(ca, list, b);             \
151                 __set_bit(b, bitmap);                                   \
152                 if (ret < 0)                                            \
153                         return ret;                                     \
154                 ret;                                                    \
155         })
156
157         struct cache *ca;
158         unsigned int iter;
159
160         for_each_cache(ca, c, iter) {
161                 struct journal_device *ja = &ca->journal;
162                 DECLARE_BITMAP(bitmap, SB_JOURNAL_BUCKETS);
163                 unsigned int i, l, r, m;
164                 uint64_t seq;
165
166                 bitmap_zero(bitmap, SB_JOURNAL_BUCKETS);
167                 pr_debug("%u journal buckets", ca->sb.njournal_buckets);
168
169                 /*
170                  * Read journal buckets ordered by golden ratio hash to quickly
171                  * find a sequence of buckets with valid journal entries
172                  */
173                 for (i = 0; i < ca->sb.njournal_buckets; i++) {
174                         /*
175                          * We must try the index l with ZERO first for
176                          * correctness due to the scenario that the journal
177                          * bucket is circular buffer which might have wrapped
178                          */
179                         l = (i * 2654435769U) % ca->sb.njournal_buckets;
180
181                         if (test_bit(l, bitmap))
182                                 break;
183
184                         if (read_bucket(l))
185                                 goto bsearch;
186                 }
187
188                 /*
189                  * If that fails, check all the buckets we haven't checked
190                  * already
191                  */
192                 pr_debug("falling back to linear search");
193
194                 for (l = find_first_zero_bit(bitmap, ca->sb.njournal_buckets);
195                      l < ca->sb.njournal_buckets;
196                      l = find_next_zero_bit(bitmap, ca->sb.njournal_buckets,
197                                             l + 1))
198                         if (read_bucket(l))
199                                 goto bsearch;
200
201                 /* no journal entries on this device? */
202                 if (l == ca->sb.njournal_buckets)
203                         continue;
204 bsearch:
205                 BUG_ON(list_empty(list));
206
207                 /* Binary search */
208                 m = l;
209                 r = find_next_bit(bitmap, ca->sb.njournal_buckets, l + 1);
210                 pr_debug("starting binary search, l %u r %u", l, r);
211
212                 while (l + 1 < r) {
213                         seq = list_entry(list->prev, struct journal_replay,
214                                          list)->j.seq;
215
216                         m = (l + r) >> 1;
217                         read_bucket(m);
218
219                         if (seq != list_entry(list->prev, struct journal_replay,
220                                               list)->j.seq)
221                                 l = m;
222                         else
223                                 r = m;
224                 }
225
226                 /*
227                  * Read buckets in reverse order until we stop finding more
228                  * journal entries
229                  */
230                 pr_debug("finishing up: m %u njournal_buckets %u",
231                          m, ca->sb.njournal_buckets);
232                 l = m;
233
234                 while (1) {
235                         if (!l--)
236                                 l = ca->sb.njournal_buckets - 1;
237
238                         if (l == m)
239                                 break;
240
241                         if (test_bit(l, bitmap))
242                                 continue;
243
244                         if (!read_bucket(l))
245                                 break;
246                 }
247
248                 seq = 0;
249
250                 for (i = 0; i < ca->sb.njournal_buckets; i++)
251                         if (ja->seq[i] > seq) {
252                                 seq = ja->seq[i];
253                                 /*
254                                  * When journal_reclaim() goes to allocate for
255                                  * the first time, it'll use the bucket after
256                                  * ja->cur_idx
257                                  */
258                                 ja->cur_idx = i;
259                                 ja->last_idx = ja->discard_idx = (i + 1) %
260                                         ca->sb.njournal_buckets;
261
262                         }
263         }
264
265         if (!list_empty(list))
266                 c->journal.seq = list_entry(list->prev,
267                                             struct journal_replay,
268                                             list)->j.seq;
269
270         return 0;
271 #undef read_bucket
272 }
273
274 void bch_journal_mark(struct cache_set *c, struct list_head *list)
275 {
276         atomic_t p = { 0 };
277         struct bkey *k;
278         struct journal_replay *i;
279         struct journal *j = &c->journal;
280         uint64_t last = j->seq;
281
282         /*
283          * journal.pin should never fill up - we never write a journal
284          * entry when it would fill up. But if for some reason it does, we
285          * iterate over the list in reverse order so that we can just skip that
286          * refcount instead of bugging.
287          */
288
289         list_for_each_entry_reverse(i, list, list) {
290                 BUG_ON(last < i->j.seq);
291                 i->pin = NULL;
292
293                 while (last-- != i->j.seq)
294                         if (fifo_free(&j->pin) > 1) {
295                                 fifo_push_front(&j->pin, p);
296                                 atomic_set(&fifo_front(&j->pin), 0);
297                         }
298
299                 if (fifo_free(&j->pin) > 1) {
300                         fifo_push_front(&j->pin, p);
301                         i->pin = &fifo_front(&j->pin);
302                         atomic_set(i->pin, 1);
303                 }
304
305                 for (k = i->j.start;
306                      k < bset_bkey_last(&i->j);
307                      k = bkey_next(k))
308                         if (!__bch_extent_invalid(c, k)) {
309                                 unsigned int j;
310
311                                 for (j = 0; j < KEY_PTRS(k); j++)
312                                         if (ptr_available(c, k, j))
313                                                 atomic_inc(&PTR_BUCKET(c, k, j)->pin);
314
315                                 bch_initial_mark_key(c, 0, k);
316                         }
317         }
318 }
319
320 int bch_journal_replay(struct cache_set *s, struct list_head *list)
321 {
322         int ret = 0, keys = 0, entries = 0;
323         struct bkey *k;
324         struct journal_replay *i =
325                 list_entry(list->prev, struct journal_replay, list);
326
327         uint64_t start = i->j.last_seq, end = i->j.seq, n = start;
328         struct keylist keylist;
329
330         list_for_each_entry(i, list, list) {
331                 BUG_ON(i->pin && atomic_read(i->pin) != 1);
332
333                 cache_set_err_on(n != i->j.seq, s,
334 "bcache: journal entries %llu-%llu missing! (replaying %llu-%llu)",
335                                  n, i->j.seq - 1, start, end);
336
337                 for (k = i->j.start;
338                      k < bset_bkey_last(&i->j);
339                      k = bkey_next(k)) {
340                         trace_bcache_journal_replay_key(k);
341
342                         bch_keylist_init_single(&keylist, k);
343
344                         ret = bch_btree_insert(s, &keylist, i->pin, NULL);
345                         if (ret)
346                                 goto err;
347
348                         BUG_ON(!bch_keylist_empty(&keylist));
349                         keys++;
350
351                         cond_resched();
352                 }
353
354                 if (i->pin)
355                         atomic_dec(i->pin);
356                 n = i->j.seq + 1;
357                 entries++;
358         }
359
360         pr_info("journal replay done, %i keys in %i entries, seq %llu",
361                 keys, entries, end);
362 err:
363         while (!list_empty(list)) {
364                 i = list_first_entry(list, struct journal_replay, list);
365                 list_del(&i->list);
366                 kfree(i);
367         }
368
369         return ret;
370 }
371
372 /* Journalling */
373 #define journal_max_cmp(l, r) \
374         (fifo_idx(&c->journal.pin, btree_current_write(l)->journal) < \
375          fifo_idx(&(c)->journal.pin, btree_current_write(r)->journal))
376 #define journal_min_cmp(l, r) \
377         (fifo_idx(&c->journal.pin, btree_current_write(l)->journal) > \
378          fifo_idx(&(c)->journal.pin, btree_current_write(r)->journal))
379
380 static void btree_flush_write(struct cache_set *c)
381 {
382         /*
383          * Try to find the btree node with that references the oldest journal
384          * entry, best is our current candidate and is locked if non NULL:
385          */
386         struct btree *b;
387         int i;
388
389         atomic_long_inc(&c->flush_write);
390
391 retry:
392         spin_lock(&c->journal.lock);
393         if (heap_empty(&c->flush_btree)) {
394                 for_each_cached_btree(b, c, i)
395                         if (btree_current_write(b)->journal) {
396                                 if (!heap_full(&c->flush_btree))
397                                         heap_add(&c->flush_btree, b,
398                                                  journal_max_cmp);
399                                 else if (journal_max_cmp(b,
400                                          heap_peek(&c->flush_btree))) {
401                                         c->flush_btree.data[0] = b;
402                                         heap_sift(&c->flush_btree, 0,
403                                                   journal_max_cmp);
404                                 }
405                         }
406
407                 for (i = c->flush_btree.used / 2 - 1; i >= 0; --i)
408                         heap_sift(&c->flush_btree, i, journal_min_cmp);
409         }
410
411         b = NULL;
412         heap_pop(&c->flush_btree, b, journal_min_cmp);
413         spin_unlock(&c->journal.lock);
414
415         if (b) {
416                 mutex_lock(&b->write_lock);
417                 if (!btree_current_write(b)->journal) {
418                         mutex_unlock(&b->write_lock);
419                         /* We raced */
420                         atomic_long_inc(&c->retry_flush_write);
421                         goto retry;
422                 }
423
424                 __bch_btree_node_write(b, NULL);
425                 mutex_unlock(&b->write_lock);
426         }
427 }
428
429 #define last_seq(j)     ((j)->seq - fifo_used(&(j)->pin) + 1)
430
431 static void journal_discard_endio(struct bio *bio)
432 {
433         struct journal_device *ja =
434                 container_of(bio, struct journal_device, discard_bio);
435         struct cache *ca = container_of(ja, struct cache, journal);
436
437         atomic_set(&ja->discard_in_flight, DISCARD_DONE);
438
439         closure_wake_up(&ca->set->journal.wait);
440         closure_put(&ca->set->cl);
441 }
442
443 static void journal_discard_work(struct work_struct *work)
444 {
445         struct journal_device *ja =
446                 container_of(work, struct journal_device, discard_work);
447
448         submit_bio(&ja->discard_bio);
449 }
450
451 static void do_journal_discard(struct cache *ca)
452 {
453         struct journal_device *ja = &ca->journal;
454         struct bio *bio = &ja->discard_bio;
455
456         if (!ca->discard) {
457                 ja->discard_idx = ja->last_idx;
458                 return;
459         }
460
461         switch (atomic_read(&ja->discard_in_flight)) {
462         case DISCARD_IN_FLIGHT:
463                 return;
464
465         case DISCARD_DONE:
466                 ja->discard_idx = (ja->discard_idx + 1) %
467                         ca->sb.njournal_buckets;
468
469                 atomic_set(&ja->discard_in_flight, DISCARD_READY);
470                 /* fallthrough */
471
472         case DISCARD_READY:
473                 if (ja->discard_idx == ja->last_idx)
474                         return;
475
476                 atomic_set(&ja->discard_in_flight, DISCARD_IN_FLIGHT);
477
478                 bio_init(bio, bio->bi_inline_vecs, 1);
479                 bio_set_op_attrs(bio, REQ_OP_DISCARD, 0);
480                 bio->bi_iter.bi_sector  = bucket_to_sector(ca->set,
481                                                 ca->sb.d[ja->discard_idx]);
482                 bio_set_dev(bio, ca->bdev);
483                 bio->bi_iter.bi_size    = bucket_bytes(ca);
484                 bio->bi_end_io          = journal_discard_endio;
485
486                 closure_get(&ca->set->cl);
487                 INIT_WORK(&ja->discard_work, journal_discard_work);
488                 queue_work(bch_journal_wq, &ja->discard_work);
489         }
490 }
491
492 static void journal_reclaim(struct cache_set *c)
493 {
494         struct bkey *k = &c->journal.key;
495         struct cache *ca;
496         uint64_t last_seq;
497         unsigned int iter, n = 0;
498         atomic_t p __maybe_unused;
499
500         atomic_long_inc(&c->reclaim);
501
502         while (!atomic_read(&fifo_front(&c->journal.pin)))
503                 fifo_pop(&c->journal.pin, p);
504
505         last_seq = last_seq(&c->journal);
506
507         /* Update last_idx */
508
509         for_each_cache(ca, c, iter) {
510                 struct journal_device *ja = &ca->journal;
511
512                 while (ja->last_idx != ja->cur_idx &&
513                        ja->seq[ja->last_idx] < last_seq)
514                         ja->last_idx = (ja->last_idx + 1) %
515                                 ca->sb.njournal_buckets;
516         }
517
518         for_each_cache(ca, c, iter)
519                 do_journal_discard(ca);
520
521         if (c->journal.blocks_free)
522                 goto out;
523
524         /*
525          * Allocate:
526          * XXX: Sort by free journal space
527          */
528
529         for_each_cache(ca, c, iter) {
530                 struct journal_device *ja = &ca->journal;
531                 unsigned int next = (ja->cur_idx + 1) % ca->sb.njournal_buckets;
532
533                 /* No space available on this device */
534                 if (next == ja->discard_idx)
535                         continue;
536
537                 ja->cur_idx = next;
538                 k->ptr[n++] = MAKE_PTR(0,
539                                   bucket_to_sector(c, ca->sb.d[ja->cur_idx]),
540                                   ca->sb.nr_this_dev);
541         }
542
543         bkey_init(k);
544         SET_KEY_PTRS(k, n);
545
546         if (n)
547                 c->journal.blocks_free = c->sb.bucket_size >> c->block_bits;
548 out:
549         if (!journal_full(&c->journal))
550                 __closure_wake_up(&c->journal.wait);
551 }
552
553 void bch_journal_next(struct journal *j)
554 {
555         atomic_t p = { 1 };
556
557         j->cur = (j->cur == j->w)
558                 ? &j->w[1]
559                 : &j->w[0];
560
561         /*
562          * The fifo_push() needs to happen at the same time as j->seq is
563          * incremented for last_seq() to be calculated correctly
564          */
565         BUG_ON(!fifo_push(&j->pin, p));
566         atomic_set(&fifo_back(&j->pin), 1);
567
568         j->cur->data->seq       = ++j->seq;
569         j->cur->dirty           = false;
570         j->cur->need_write      = false;
571         j->cur->data->keys      = 0;
572
573         if (fifo_full(&j->pin))
574                 pr_debug("journal_pin full (%zu)", fifo_used(&j->pin));
575 }
576
577 static void journal_write_endio(struct bio *bio)
578 {
579         struct journal_write *w = bio->bi_private;
580
581         cache_set_err_on(bio->bi_status, w->c, "journal io error");
582         closure_put(&w->c->journal.io);
583 }
584
585 static void journal_write(struct closure *cl);
586
587 static void journal_write_done(struct closure *cl)
588 {
589         struct journal *j = container_of(cl, struct journal, io);
590         struct journal_write *w = (j->cur == j->w)
591                 ? &j->w[1]
592                 : &j->w[0];
593
594         __closure_wake_up(&w->wait);
595         continue_at_nobarrier(cl, journal_write, bch_journal_wq);
596 }
597
598 static void journal_write_unlock(struct closure *cl)
599         __releases(&c->journal.lock)
600 {
601         struct cache_set *c = container_of(cl, struct cache_set, journal.io);
602
603         c->journal.io_in_flight = 0;
604         spin_unlock(&c->journal.lock);
605 }
606
607 static void journal_write_unlocked(struct closure *cl)
608         __releases(c->journal.lock)
609 {
610         struct cache_set *c = container_of(cl, struct cache_set, journal.io);
611         struct cache *ca;
612         struct journal_write *w = c->journal.cur;
613         struct bkey *k = &c->journal.key;
614         unsigned int i, sectors = set_blocks(w->data, block_bytes(c)) *
615                 c->sb.block_size;
616
617         struct bio *bio;
618         struct bio_list list;
619
620         bio_list_init(&list);
621
622         if (!w->need_write) {
623                 closure_return_with_destructor(cl, journal_write_unlock);
624                 return;
625         } else if (journal_full(&c->journal)) {
626                 journal_reclaim(c);
627                 spin_unlock(&c->journal.lock);
628
629                 btree_flush_write(c);
630                 continue_at(cl, journal_write, bch_journal_wq);
631                 return;
632         }
633
634         c->journal.blocks_free -= set_blocks(w->data, block_bytes(c));
635
636         w->data->btree_level = c->root->level;
637
638         bkey_copy(&w->data->btree_root, &c->root->key);
639         bkey_copy(&w->data->uuid_bucket, &c->uuid_bucket);
640
641         for_each_cache(ca, c, i)
642                 w->data->prio_bucket[ca->sb.nr_this_dev] = ca->prio_buckets[0];
643
644         w->data->magic          = jset_magic(&c->sb);
645         w->data->version        = BCACHE_JSET_VERSION;
646         w->data->last_seq       = last_seq(&c->journal);
647         w->data->csum           = csum_set(w->data);
648
649         for (i = 0; i < KEY_PTRS(k); i++) {
650                 ca = PTR_CACHE(c, k, i);
651                 bio = &ca->journal.bio;
652
653                 atomic_long_add(sectors, &ca->meta_sectors_written);
654
655                 bio_reset(bio);
656                 bio->bi_iter.bi_sector  = PTR_OFFSET(k, i);
657                 bio_set_dev(bio, ca->bdev);
658                 bio->bi_iter.bi_size = sectors << 9;
659
660                 bio->bi_end_io  = journal_write_endio;
661                 bio->bi_private = w;
662                 bio_set_op_attrs(bio, REQ_OP_WRITE,
663                                  REQ_SYNC|REQ_META|REQ_PREFLUSH|REQ_FUA);
664                 bch_bio_map(bio, w->data);
665
666                 trace_bcache_journal_write(bio);
667                 bio_list_add(&list, bio);
668
669                 SET_PTR_OFFSET(k, i, PTR_OFFSET(k, i) + sectors);
670
671                 ca->journal.seq[ca->journal.cur_idx] = w->data->seq;
672         }
673
674         atomic_dec_bug(&fifo_back(&c->journal.pin));
675         bch_journal_next(&c->journal);
676         journal_reclaim(c);
677
678         spin_unlock(&c->journal.lock);
679
680         while ((bio = bio_list_pop(&list)))
681                 closure_bio_submit(c, bio, cl);
682
683         continue_at(cl, journal_write_done, NULL);
684 }
685
686 static void journal_write(struct closure *cl)
687 {
688         struct cache_set *c = container_of(cl, struct cache_set, journal.io);
689
690         spin_lock(&c->journal.lock);
691         journal_write_unlocked(cl);
692 }
693
694 static void journal_try_write(struct cache_set *c)
695         __releases(c->journal.lock)
696 {
697         struct closure *cl = &c->journal.io;
698         struct journal_write *w = c->journal.cur;
699
700         w->need_write = true;
701
702         if (!c->journal.io_in_flight) {
703                 c->journal.io_in_flight = 1;
704                 closure_call(cl, journal_write_unlocked, NULL, &c->cl);
705         } else {
706                 spin_unlock(&c->journal.lock);
707         }
708 }
709
710 static struct journal_write *journal_wait_for_write(struct cache_set *c,
711                                                     unsigned int nkeys)
712         __acquires(&c->journal.lock)
713 {
714         size_t sectors;
715         struct closure cl;
716         bool wait = false;
717
718         closure_init_stack(&cl);
719
720         spin_lock(&c->journal.lock);
721
722         while (1) {
723                 struct journal_write *w = c->journal.cur;
724
725                 sectors = __set_blocks(w->data, w->data->keys + nkeys,
726                                        block_bytes(c)) * c->sb.block_size;
727
728                 if (sectors <= min_t(size_t,
729                                      c->journal.blocks_free * c->sb.block_size,
730                                      PAGE_SECTORS << JSET_BITS))
731                         return w;
732
733                 if (wait)
734                         closure_wait(&c->journal.wait, &cl);
735
736                 if (!journal_full(&c->journal)) {
737                         if (wait)
738                                 trace_bcache_journal_entry_full(c);
739
740                         /*
741                          * XXX: If we were inserting so many keys that they
742                          * won't fit in an _empty_ journal write, we'll
743                          * deadlock. For now, handle this in
744                          * bch_keylist_realloc() - but something to think about.
745                          */
746                         BUG_ON(!w->data->keys);
747
748                         journal_try_write(c); /* unlocks */
749                 } else {
750                         if (wait)
751                                 trace_bcache_journal_full(c);
752
753                         journal_reclaim(c);
754                         spin_unlock(&c->journal.lock);
755
756                         btree_flush_write(c);
757                 }
758
759                 closure_sync(&cl);
760                 spin_lock(&c->journal.lock);
761                 wait = true;
762         }
763 }
764
765 static void journal_write_work(struct work_struct *work)
766 {
767         struct cache_set *c = container_of(to_delayed_work(work),
768                                            struct cache_set,
769                                            journal.work);
770         spin_lock(&c->journal.lock);
771         if (c->journal.cur->dirty)
772                 journal_try_write(c);
773         else
774                 spin_unlock(&c->journal.lock);
775 }
776
777 /*
778  * Entry point to the journalling code - bio_insert() and btree_invalidate()
779  * pass bch_journal() a list of keys to be journalled, and then
780  * bch_journal() hands those same keys off to btree_insert_async()
781  */
782
783 atomic_t *bch_journal(struct cache_set *c,
784                       struct keylist *keys,
785                       struct closure *parent)
786 {
787         struct journal_write *w;
788         atomic_t *ret;
789
790         if (!CACHE_SYNC(&c->sb))
791                 return NULL;
792
793         w = journal_wait_for_write(c, bch_keylist_nkeys(keys));
794
795         memcpy(bset_bkey_last(w->data), keys->keys, bch_keylist_bytes(keys));
796         w->data->keys += bch_keylist_nkeys(keys);
797
798         ret = &fifo_back(&c->journal.pin);
799         atomic_inc(ret);
800
801         if (parent) {
802                 closure_wait(&w->wait, parent);
803                 journal_try_write(c);
804         } else if (!w->dirty) {
805                 w->dirty = true;
806                 schedule_delayed_work(&c->journal.work,
807                                       msecs_to_jiffies(c->journal_delay_ms));
808                 spin_unlock(&c->journal.lock);
809         } else {
810                 spin_unlock(&c->journal.lock);
811         }
812
813
814         return ret;
815 }
816
817 void bch_journal_meta(struct cache_set *c, struct closure *cl)
818 {
819         struct keylist keys;
820         atomic_t *ref;
821
822         bch_keylist_init(&keys);
823
824         ref = bch_journal(c, &keys, cl);
825         if (ref)
826                 atomic_dec_bug(ref);
827 }
828
829 void bch_journal_free(struct cache_set *c)
830 {
831         free_pages((unsigned long) c->journal.w[1].data, JSET_BITS);
832         free_pages((unsigned long) c->journal.w[0].data, JSET_BITS);
833         free_fifo(&c->journal.pin);
834         free_heap(&c->flush_btree);
835 }
836
837 int bch_journal_alloc(struct cache_set *c)
838 {
839         struct journal *j = &c->journal;
840
841         spin_lock_init(&j->lock);
842         INIT_DELAYED_WORK(&j->work, journal_write_work);
843
844         c->journal_delay_ms = 100;
845
846         j->w[0].c = c;
847         j->w[1].c = c;
848
849         if (!(init_heap(&c->flush_btree, 128, GFP_KERNEL)) ||
850             !(init_fifo(&j->pin, JOURNAL_PIN, GFP_KERNEL)) ||
851             !(j->w[0].data = (void *) __get_free_pages(GFP_KERNEL, JSET_BITS)) ||
852             !(j->w[1].data = (void *) __get_free_pages(GFP_KERNEL, JSET_BITS)))
853                 return -ENOMEM;
854
855         return 0;
856 }