]> git.sesse.net Git - bcachefs-tools-debian/blobdiff - libbcachefs/journal_io.c
Update bcachefs sources to 39a84c99af2d bcachefs: Clamp replicas_required to replicas
[bcachefs-tools-debian] / libbcachefs / journal_io.c
index 65878542940d924e9d6656ff5590a14fe8a70b14..057e7c6113157775366b187aad54c613cd91404b 100644 (file)
@@ -4,6 +4,7 @@
 #include "alloc_foreground.h"
 #include "btree_io.h"
 #include "btree_update_interior.h"
+#include "btree_write_buffer.h"
 #include "buckets.h"
 #include "checksum.h"
 #include "disk_groups.h"
 #include "sb-clean.h"
 #include "trace.h"
 
+void bch2_journal_ptrs_to_text(struct printbuf *out, struct bch_fs *c,
+                              struct journal_replay *j)
+{
+       darray_for_each(j->ptrs, i) {
+               struct bch_dev *ca = bch_dev_bkey_exists(c, i->dev);
+               u64 offset;
+
+               div64_u64_rem(i->sector, ca->mi.bucket_size, &offset);
+
+               if (i != j->ptrs.data)
+                       prt_printf(out, " ");
+               prt_printf(out, "%u:%u:%u (sector %llu)",
+                          i->dev, i->bucket, i->bucket_offset, i->sector);
+       }
+}
+
+static void bch2_journal_replay_to_text(struct printbuf *out, struct bch_fs *c,
+                                       struct journal_replay *j)
+{
+       prt_printf(out, "seq %llu ", le64_to_cpu(j->j.seq));
+
+       bch2_journal_ptrs_to_text(out, c, j);
+
+       struct jset_entry *entry;
+       for_each_jset_entry_type(entry, &j->j, BCH_JSET_ENTRY_datetime) {
+               struct jset_entry_datetime *datetime =
+                       container_of(entry, struct jset_entry_datetime, entry);
+               bch2_prt_datetime(out, le64_to_cpu(datetime->seconds));
+               break;
+       }
+}
+
 static struct nonce journal_nonce(const struct jset *jset)
 {
        return (struct nonce) {{
@@ -26,11 +59,15 @@ static struct nonce journal_nonce(const struct jset *jset)
        }};
 }
 
-static bool jset_csum_good(struct bch_fs *c, struct jset *j)
+static bool jset_csum_good(struct bch_fs *c, struct jset *j, struct bch_csum *csum)
 {
-       return bch2_checksum_type_valid(c, JSET_CSUM_TYPE(j)) &&
-               !bch2_crc_cmp(j->csum,
-                             csum_vstruct(c, JSET_CSUM_TYPE(j), journal_nonce(j), j));
+       if (!bch2_checksum_type_valid(c, JSET_CSUM_TYPE(j))) {
+               *csum = (struct bch_csum) {};
+               return false;
+       }
+
+       *csum = csum_vstruct(c, JSET_CSUM_TYPE(j), journal_nonce(j), j);
+       return !bch2_crc_cmp(j->csum, *csum);
 }
 
 static inline u32 journal_entry_radix_idx(struct bch_fs *c, u64 seq)
@@ -47,8 +84,7 @@ static void __journal_replay_free(struct bch_fs *c,
 
        BUG_ON(*p != i);
        *p = NULL;
-       kvpfree(i, offsetof(struct journal_replay, j) +
-               vstruct_bytes(&i->j));
+       kvfree(i);
 }
 
 static void journal_replay_free(struct bch_fs *c, struct journal_replay *i)
@@ -79,9 +115,9 @@ static int journal_entry_add(struct bch_fs *c, struct bch_dev *ca,
 {
        struct genradix_iter iter;
        struct journal_replay **_i, *i, *dup;
-       struct journal_ptr *ptr;
        size_t bytes = vstruct_bytes(j);
        u64 last_seq = !JSET_NO_FLUSH(j) ? le64_to_cpu(j->last_seq) : 0;
+       struct printbuf buf = PRINTBUF;
        int ret = JOURNAL_ENTRY_ADD_OK;
 
        /* Is this entry older than the range we need? */
@@ -126,72 +162,61 @@ static int journal_entry_add(struct bch_fs *c, struct bch_dev *ca,
         */
        dup = *_i;
        if (dup) {
-               if (bytes == vstruct_bytes(&dup->j) &&
-                   !memcmp(j, &dup->j, bytes)) {
-                       i = dup;
-                       goto found;
-               }
+               bool identical = bytes == vstruct_bytes(&dup->j) &&
+                       !memcmp(j, &dup->j, bytes);
+               bool not_identical = !identical &&
+                       entry_ptr.csum_good &&
+                       dup->csum_good;
+
+               bool same_device = false;
+               darray_for_each(dup->ptrs, ptr)
+                       if (ptr->dev == ca->dev_idx)
+                               same_device = true;
+
+               ret = darray_push(&dup->ptrs, entry_ptr);
+               if (ret)
+                       goto out;
 
-               if (!entry_ptr.csum_good) {
-                       i = dup;
-                       goto found;
-               }
+               bch2_journal_replay_to_text(&buf, c, dup);
+
+               fsck_err_on(same_device,
+                           c, journal_entry_dup_same_device,
+                           "duplicate journal entry on same device\n  %s",
+                           buf.buf);
+
+               fsck_err_on(not_identical,
+                           c, journal_entry_replicas_data_mismatch,
+                           "found duplicate but non identical journal entries\n  %s",
+                           buf.buf);
 
-               if (!dup->csum_good)
+               if (entry_ptr.csum_good && !identical)
                        goto replace;
 
-               fsck_err(c, journal_entry_replicas_data_mismatch,
-                        "found duplicate but non identical journal entries (seq %llu)",
-                        le64_to_cpu(j->seq));
-               i = dup;
-               goto found;
+               goto out;
        }
 replace:
-       i = kvpmalloc(offsetof(struct journal_replay, j) + bytes, GFP_KERNEL);
+       i = kvmalloc(offsetof(struct journal_replay, j) + bytes, GFP_KERNEL);
        if (!i)
                return -BCH_ERR_ENOMEM_journal_entry_add;
 
-       i->nr_ptrs      = 0;
+       darray_init(&i->ptrs);
        i->csum_good    = entry_ptr.csum_good;
        i->ignore       = false;
        unsafe_memcpy(&i->j, j, bytes, "embedded variable length struct");
-       i->ptrs[i->nr_ptrs++] = entry_ptr;
 
        if (dup) {
-               if (dup->nr_ptrs >= ARRAY_SIZE(dup->ptrs)) {
-                       bch_err(c, "found too many copies of journal entry %llu",
-                               le64_to_cpu(i->j.seq));
-                       dup->nr_ptrs = ARRAY_SIZE(dup->ptrs) - 1;
-               }
-
                /* The first ptr should represent the jset we kept: */
-               memcpy(i->ptrs + i->nr_ptrs,
-                      dup->ptrs,
-                      sizeof(dup->ptrs[0]) * dup->nr_ptrs);
-               i->nr_ptrs += dup->nr_ptrs;
+               darray_for_each(dup->ptrs, ptr)
+                       darray_push(&i->ptrs, *ptr);
                __journal_replay_free(c, dup);
+       } else {
+               darray_push(&i->ptrs, entry_ptr);
        }
 
        *_i = i;
-       return 0;
-found:
-       for (ptr = i->ptrs; ptr < i->ptrs + i->nr_ptrs; ptr++) {
-               if (ptr->dev == ca->dev_idx) {
-                       bch_err(c, "duplicate journal entry %llu on same device",
-                               le64_to_cpu(i->j.seq));
-                       goto out;
-               }
-       }
-
-       if (i->nr_ptrs >= ARRAY_SIZE(i->ptrs)) {
-               bch_err(c, "found too many copies of journal entry %llu",
-                       le64_to_cpu(i->j.seq));
-               goto out;
-       }
-
-       i->ptrs[i->nr_ptrs++] = entry_ptr;
 out:
 fsck_err:
+       printbuf_exit(&buf);
        return ret;
 }
 
@@ -408,8 +433,10 @@ static int journal_entry_btree_root_validate(struct bch_fs *c,
                return 0;
        }
 
-       return journal_validate_key(c, jset, entry, 1, entry->btree_id, k,
-                                   version, big_endian, flags);
+       ret = journal_validate_key(c, jset, entry, 1, entry->btree_id, k,
+                                  version, big_endian, flags);
+       if (ret == FSCK_DELETED_KEY)
+               ret = 0;
 fsck_err:
        return ret;
 }
@@ -547,6 +574,7 @@ static int journal_entry_data_usage_validate(struct bch_fs *c,
        struct jset_entry_data_usage *u =
                container_of(entry, struct jset_entry_data_usage, entry);
        unsigned bytes = jset_u64s(le16_to_cpu(entry->u64s)) * sizeof(u64);
+       struct printbuf err = PRINTBUF;
        int ret = 0;
 
        if (journal_entry_err_on(bytes < sizeof(*u) ||
@@ -555,10 +583,19 @@ static int journal_entry_data_usage_validate(struct bch_fs *c,
                                 journal_entry_data_usage_bad_size,
                                 "invalid journal entry usage: bad size")) {
                journal_entry_null_range(entry, vstruct_next(entry));
-               return ret;
+               goto out;
        }
 
+       if (journal_entry_err_on(bch2_replicas_entry_validate(&u->r, c->disk_sb.sb, &err),
+                                c, version, jset, entry,
+                                journal_entry_data_usage_bad_size,
+                                "invalid journal entry usage: %s", err.buf)) {
+               journal_entry_null_range(entry, vstruct_next(entry));
+               goto out;
+       }
+out:
 fsck_err:
+       printbuf_exit(&err);
        return ret;
 }
 
@@ -666,17 +703,12 @@ static void journal_entry_dev_usage_to_text(struct printbuf *out, struct bch_fs
        prt_printf(out, "dev=%u", le32_to_cpu(u->dev));
 
        for (i = 0; i < nr_types; i++) {
-               if (i < BCH_DATA_NR)
-                       prt_printf(out, " %s", bch2_data_types[i]);
-               else
-                       prt_printf(out, " (unknown data type %u)", i);
+               bch2_prt_data_type(out, i);
                prt_printf(out, ": buckets=%llu sectors=%llu fragmented=%llu",
                       le64_to_cpu(u->d[i].buckets),
                       le64_to_cpu(u->d[i].sectors),
                       le64_to_cpu(u->d[i].fragmented));
        }
-
-       prt_printf(out, " buckets_ec: %llu", le64_to_cpu(u->buckets_ec));
 }
 
 static int journal_entry_log_validate(struct bch_fs *c,
@@ -713,6 +745,53 @@ static void journal_entry_overwrite_to_text(struct printbuf *out, struct bch_fs
        journal_entry_btree_keys_to_text(out, c, entry);
 }
 
+static int journal_entry_write_buffer_keys_validate(struct bch_fs *c,
+                               struct jset *jset,
+                               struct jset_entry *entry,
+                               unsigned version, int big_endian,
+                               enum bkey_invalid_flags flags)
+{
+       return journal_entry_btree_keys_validate(c, jset, entry,
+                               version, big_endian, READ);
+}
+
+static void journal_entry_write_buffer_keys_to_text(struct printbuf *out, struct bch_fs *c,
+                                           struct jset_entry *entry)
+{
+       journal_entry_btree_keys_to_text(out, c, entry);
+}
+
+static int journal_entry_datetime_validate(struct bch_fs *c,
+                               struct jset *jset,
+                               struct jset_entry *entry,
+                               unsigned version, int big_endian,
+                               enum bkey_invalid_flags flags)
+{
+       unsigned bytes = vstruct_bytes(entry);
+       unsigned expected = 16;
+       int ret = 0;
+
+       if (journal_entry_err_on(vstruct_bytes(entry) < expected,
+                                c, version, jset, entry,
+                                journal_entry_dev_usage_bad_size,
+                                "bad size (%u < %u)",
+                                bytes, expected)) {
+               journal_entry_null_range(entry, vstruct_next(entry));
+               return ret;
+       }
+fsck_err:
+       return ret;
+}
+
+static void journal_entry_datetime_to_text(struct printbuf *out, struct bch_fs *c,
+                                           struct jset_entry *entry)
+{
+       struct jset_entry_datetime *datetime =
+               container_of(entry, struct jset_entry_datetime, entry);
+
+       bch2_prt_datetime(out, le64_to_cpu(datetime->seconds));
+}
+
 struct jset_entry_ops {
        int (*validate)(struct bch_fs *, struct jset *,
                        struct jset_entry *, unsigned, int,
@@ -756,7 +835,6 @@ void bch2_journal_entry_to_text(struct printbuf *out, struct bch_fs *c,
 static int jset_validate_entries(struct bch_fs *c, struct jset *jset,
                                 enum bkey_invalid_flags flags)
 {
-       struct jset_entry *entry;
        unsigned version = le32_to_cpu(jset->version);
        int ret = 0;
 
@@ -886,11 +964,11 @@ static int journal_read_buf_realloc(struct journal_read_buf *b,
                return -BCH_ERR_ENOMEM_journal_read_buf_realloc;
 
        new_size = roundup_pow_of_two(new_size);
-       n = kvpmalloc(new_size, GFP_KERNEL);
+       n = kvmalloc(new_size, GFP_KERNEL);
        if (!n)
                return -BCH_ERR_ENOMEM_journal_read_buf_realloc;
 
-       kvpfree(b->data, b->size);
+       kvfree(b->data);
        b->data = n;
        b->size = new_size;
        return 0;
@@ -908,6 +986,7 @@ static int journal_read_bucket(struct bch_dev *ca,
        u64 offset = bucket_to_sector(ca, ja->buckets[bucket]),
            end = offset + ca->mi.bucket_size;
        bool saw_bad = false, csum_good;
+       struct printbuf err = PRINTBUF;
        int ret = 0;
 
        pr_debug("reading %u", bucket);
@@ -940,7 +1019,7 @@ reread:
                                 * found on a different device, and missing or
                                 * no journal entries will be handled later
                                 */
-                               return 0;
+                               goto out;
                        }
 
                        j = buf->data;
@@ -957,12 +1036,12 @@ reread:
                                ret = journal_read_buf_realloc(buf,
                                                        vstruct_bytes(j));
                                if (ret)
-                                       return ret;
+                                       goto err;
                        }
                        goto reread;
                case JOURNAL_ENTRY_NONE:
                        if (!saw_bad)
-                               return 0;
+                               goto out;
                        /*
                         * On checksum error we don't really trust the size
                         * field of the journal entry we read, so try reading
@@ -971,7 +1050,7 @@ reread:
                        sectors = block_sectors(c);
                        goto next_block;
                default:
-                       return ret;
+                       goto err;
                }
 
                /*
@@ -981,20 +1060,28 @@ reread:
                 * bucket:
                 */
                if (le64_to_cpu(j->seq) < ja->bucket_seq[bucket])
-                       return 0;
+                       goto out;
 
                ja->bucket_seq[bucket] = le64_to_cpu(j->seq);
 
-               csum_good = jset_csum_good(c, j);
+               enum bch_csum_type csum_type = JSET_CSUM_TYPE(j);
+               struct bch_csum csum;
+               csum_good = jset_csum_good(c, j, &csum);
+
                if (bch2_dev_io_err_on(!csum_good, ca, BCH_MEMBER_ERROR_checksum,
-                                      "journal checksum error"))
+                                      "%s",
+                                      (printbuf_reset(&err),
+                                       prt_str(&err, "journal "),
+                                       bch2_csum_err_msg(&err, csum_type, j->csum, csum),
+                                       err.buf)))
                        saw_bad = true;
 
                ret = bch2_encrypt(c, JSET_CSUM_TYPE(j), journal_nonce(j),
                             j->encrypted_start,
                             vstruct_end(j) - (void *) j->encrypted_start);
                bch2_fs_fatal_err_on(ret, c,
-                               "error decrypting journal entry: %i", ret);
+                               "error decrypting journal entry: %s",
+                               bch2_err_str(ret));
 
                mutex_lock(&jlist->lock);
                ret = journal_entry_add(c, ca, (struct journal_ptr) {
@@ -1013,7 +1100,7 @@ reread:
                case JOURNAL_ENTRY_ADD_OUT_OF_RANGE:
                        break;
                default:
-                       return ret;
+                       goto err;
                }
 next_block:
                pr_debug("next");
@@ -1022,13 +1109,16 @@ next_block:
                j = ((void *) j) + (sectors << 9);
        }
 
-       return 0;
+out:
+       ret = 0;
+err:
+       printbuf_exit(&err);
+       return ret;
 }
 
-static void bch2_journal_read_device(struct closure *cl)
+static CLOSURE_CALLBACK(bch2_journal_read_device)
 {
-       struct journal_device *ja =
-               container_of(cl, struct journal_device, read);
+       closure_type(ja, struct journal_device, read);
        struct bch_dev *ca = container_of(ja, struct bch_dev, journal);
        struct bch_fs *c = ca->fs;
        struct journal_list *jlist =
@@ -1063,22 +1153,27 @@ static void bch2_journal_read_device(struct closure *cl)
                if (!r)
                        continue;
 
-               for (i = 0; i < r->nr_ptrs; i++) {
-                       if (r->ptrs[i].dev == ca->dev_idx) {
-                               unsigned wrote = bucket_remainder(ca, r->ptrs[i].sector) +
+               darray_for_each(r->ptrs, i)
+                       if (i->dev == ca->dev_idx) {
+                               unsigned wrote = bucket_remainder(ca, i->sector) +
                                        vstruct_sectors(&r->j, c->block_bits);
 
-                               ja->cur_idx = r->ptrs[i].bucket;
+                               ja->cur_idx = i->bucket;
                                ja->sectors_free = ca->mi.bucket_size - wrote;
                                goto found;
                        }
-               }
        }
 found:
        mutex_unlock(&jlist->lock);
 
        if (ja->bucket_seq[ja->cur_idx] &&
            ja->sectors_free == ca->mi.bucket_size) {
+#if 0
+               /*
+                * Debug code for ZNS support, where we (probably) want to be
+                * correlated where we stopped in the journal to the zone write
+                * points:
+                */
                bch_err(c, "ja->sectors_free == ca->mi.bucket_size");
                bch_err(c, "cur_idx %u/%u", ja->cur_idx, ja->nr);
                for (i = 0; i < 3; i++) {
@@ -1086,6 +1181,7 @@ found:
 
                        bch_err(c, "bucket_seq[%u] = %llu", idx, ja->bucket_seq[idx]);
                }
+#endif
                ja->sectors_free = 0;
        }
 
@@ -1098,7 +1194,7 @@ found:
                ja->dirty_idx = (ja->cur_idx + 1) % ja->nr;
 out:
        bch_verbose(c, "journal read done on device %s, ret %i", ca->name, ret);
-       kvpfree(buf.data, buf.size);
+       kvfree(buf.data);
        percpu_ref_put(&ca->io_ref);
        closure_return(cl);
        return;
@@ -1109,27 +1205,6 @@ err:
        goto out;
 }
 
-void bch2_journal_ptrs_to_text(struct printbuf *out, struct bch_fs *c,
-                              struct journal_replay *j)
-{
-       unsigned i;
-
-       for (i = 0; i < j->nr_ptrs; i++) {
-               struct bch_dev *ca = bch_dev_bkey_exists(c, j->ptrs[i].dev);
-               u64 offset;
-
-               div64_u64_rem(j->ptrs[i].sector, ca->mi.bucket_size, &offset);
-
-               if (i)
-                       prt_printf(out, " ");
-               prt_printf(out, "%u:%u:%u (sector %llu)",
-                      j->ptrs[i].dev,
-                      j->ptrs[i].bucket,
-                      j->ptrs[i].bucket_offset,
-                      j->ptrs[i].sector);
-       }
-}
-
 int bch2_journal_read(struct bch_fs *c,
                      u64 *last_seq,
                      u64 *blacklist_seq,
@@ -1138,8 +1213,6 @@ int bch2_journal_read(struct bch_fs *c,
        struct journal_list jlist;
        struct journal_replay *i, **_i, *prev = NULL;
        struct genradix_iter radix_iter;
-       struct bch_dev *ca;
-       unsigned iter;
        struct printbuf buf = PRINTBUF;
        bool degraded = false, last_write_torn = false;
        u64 seq;
@@ -1150,7 +1223,7 @@ int bch2_journal_read(struct bch_fs *c,
        jlist.last_seq = 0;
        jlist.ret = 0;
 
-       for_each_member_device(ca, c, iter) {
+       for_each_member_device(c, ca) {
                if (!c->opts.fsck &&
                    !(bch2_dev_has_data(c, ca) & (1 << BCH_DATA_journal)))
                        continue;
@@ -1309,32 +1382,31 @@ int bch2_journal_read(struct bch_fs *c,
                        .e.data_type = BCH_DATA_journal,
                        .e.nr_required = 1,
                };
-               unsigned ptr;
 
                i = *_i;
                if (!i || i->ignore)
                        continue;
 
-               for (ptr = 0; ptr < i->nr_ptrs; ptr++) {
-                       ca = bch_dev_bkey_exists(c, i->ptrs[ptr].dev);
+               darray_for_each(i->ptrs, ptr) {
+                       struct bch_dev *ca = bch_dev_bkey_exists(c, ptr->dev);
 
-                       if (!i->ptrs[ptr].csum_good)
-                               bch_err_dev_offset(ca, i->ptrs[ptr].sector,
+                       if (!ptr->csum_good)
+                               bch_err_dev_offset(ca, ptr->sector,
                                                   "invalid journal checksum, seq %llu%s",
                                                   le64_to_cpu(i->j.seq),
                                                   i->csum_good ? " (had good copy on another device)" : "");
                }
 
                ret = jset_validate(c,
-                                   bch_dev_bkey_exists(c, i->ptrs[0].dev),
+                                   bch_dev_bkey_exists(c, i->ptrs.data[0].dev),
                                    &i->j,
-                                   i->ptrs[0].sector,
+                                   i->ptrs.data[0].sector,
                                    READ);
                if (ret)
                        goto err;
 
-               for (ptr = 0; ptr < i->nr_ptrs; ptr++)
-                       replicas.e.devs[replicas.e.nr_devs++] = i->ptrs[ptr].dev;
+               darray_for_each(i->ptrs, ptr)
+                       replicas.e.devs[replicas.e.nr_devs++] = ptr->dev;
 
                bch2_replicas_entry_sort(&replicas.e);
 
@@ -1434,6 +1506,8 @@ static int journal_write_alloc(struct journal *j, struct journal_buf *w)
                c->opts.foreground_target;
        unsigned i, replicas = 0, replicas_want =
                READ_ONCE(c->opts.metadata_replicas);
+       unsigned replicas_need = min(replicas_want,
+                                    READ_ONCE(c->opts.metadata_replicas_required));
 
        rcu_read_lock();
 retry:
@@ -1482,11 +1556,13 @@ done:
 
        BUG_ON(bkey_val_u64s(&w->key.k) > BCH_REPLICAS_MAX);
 
-       return replicas >= c->opts.metadata_replicas_required ? 0 : -EROFS;
+       return replicas >= replicas_need ? 0 : -EROFS;
 }
 
 static void journal_buf_realloc(struct journal *j, struct journal_buf *buf)
 {
+       struct bch_fs *c = container_of(j, struct bch_fs, journal);
+
        /* we aren't holding j->lock: */
        unsigned new_size = READ_ONCE(j->buf_size_want);
        void *new_buf;
@@ -1494,7 +1570,12 @@ static void journal_buf_realloc(struct journal *j, struct journal_buf *buf)
        if (buf->buf_size >= new_size)
                return;
 
-       new_buf = kvpmalloc(new_size, GFP_NOFS|__GFP_NOWARN);
+       size_t btree_write_buffer_size = new_size / 64;
+
+       if (bch2_btree_write_buffer_resize(c, btree_write_buffer_size))
+               return;
+
+       new_buf = kvmalloc(new_size, GFP_NOFS|__GFP_NOWARN);
        if (!new_buf)
                return;
 
@@ -1505,7 +1586,7 @@ static void journal_buf_realloc(struct journal *j, struct journal_buf *buf)
        swap(buf->buf_size,     new_size);
        spin_unlock(&j->lock);
 
-       kvpfree(new_buf, new_size);
+       kvfree(new_buf);
 }
 
 static inline struct journal_buf *journal_last_unwritten_buf(struct journal *j)
@@ -1513,19 +1594,19 @@ static inline struct journal_buf *journal_last_unwritten_buf(struct journal *j)
        return j->buf + (journal_last_unwritten_seq(j) & JOURNAL_BUF_MASK);
 }
 
-static void journal_write_done(struct closure *cl)
+static CLOSURE_CALLBACK(journal_write_done)
 {
-       struct journal *j = container_of(cl, struct journal, io);
+       closure_type(w, struct journal_buf, io);
+       struct journal *j = container_of(w, struct journal, buf[w->idx]);
        struct bch_fs *c = container_of(j, struct bch_fs, journal);
-       struct journal_buf *w = journal_last_unwritten_buf(j);
        struct bch_replicas_padded replicas;
        union journal_res_state old, new;
-       u64 v, seq;
+       u64 v, seq = le64_to_cpu(w->data->seq);
        int err = 0;
 
-       bch2_time_stats_update(!JSET_NO_FLUSH(w->data)
-                              ? j->flush_write_time
-                              : j->noflush_write_time, j->write_start_time);
+       time_stats_update(!JSET_NO_FLUSH(w->data)
+                         ? j->flush_write_time
+                         : j->noflush_write_time, j->write_start_time);
 
        if (!w->devs_written.nr) {
                bch_err(c, "unable to write journal to sufficient devices");
@@ -1540,59 +1621,68 @@ static void journal_write_done(struct closure *cl)
        if (err)
                bch2_fatal_error(c);
 
-       spin_lock(&j->lock);
-       seq = le64_to_cpu(w->data->seq);
+       closure_debug_destroy(cl);
 
+       spin_lock(&j->lock);
        if (seq >= j->pin.front)
                journal_seq_pin(j, seq)->devs = w->devs_written;
+       if (err && (!j->err_seq || seq < j->err_seq))
+               j->err_seq      = seq;
+       w->write_done = true;
 
-       if (!err) {
-               if (!JSET_NO_FLUSH(w->data)) {
+       bool completed = false;
+
+       for (seq = journal_last_unwritten_seq(j);
+            seq <= journal_cur_seq(j);
+            seq++) {
+               w = j->buf + (seq & JOURNAL_BUF_MASK);
+               if (!w->write_done)
+                       break;
+
+               if (!j->err_seq && !JSET_NO_FLUSH(w->data)) {
                        j->flushed_seq_ondisk = seq;
                        j->last_seq_ondisk = w->last_seq;
 
                        bch2_do_discards(c);
                        closure_wake_up(&c->freelist_wait);
-
                        bch2_reset_alloc_cursors(c);
                }
-       } else if (!j->err_seq || seq < j->err_seq)
-               j->err_seq      = seq;
 
-       j->seq_ondisk           = seq;
+               j->seq_ondisk = seq;
 
-       /*
-        * Updating last_seq_ondisk may let bch2_journal_reclaim_work() discard
-        * more buckets:
-        *
-        * Must come before signaling write completion, for
-        * bch2_fs_journal_stop():
-        */
-       if (j->watermark != BCH_WATERMARK_stripe)
-               journal_reclaim_kick(&c->journal);
+               /*
+                * Updating last_seq_ondisk may let bch2_journal_reclaim_work() discard
+                * more buckets:
+                *
+                * Must come before signaling write completion, for
+                * bch2_fs_journal_stop():
+                */
+               if (j->watermark != BCH_WATERMARK_stripe)
+                       journal_reclaim_kick(&c->journal);
 
-       /* also must come before signalling write completion: */
-       closure_debug_destroy(cl);
+               v = atomic64_read(&j->reservations.counter);
+               do {
+                       old.v = new.v = v;
+                       BUG_ON(journal_state_count(new, new.unwritten_idx));
+                       BUG_ON(new.unwritten_idx != (seq & JOURNAL_BUF_MASK));
 
-       v = atomic64_read(&j->reservations.counter);
-       do {
-               old.v = new.v = v;
-               BUG_ON(journal_state_count(new, new.unwritten_idx));
+                       new.unwritten_idx++;
+               } while ((v = atomic64_cmpxchg(&j->reservations.counter, old.v, new.v)) != old.v);
 
-               new.unwritten_idx++;
-       } while ((v = atomic64_cmpxchg(&j->reservations.counter,
-                                      old.v, new.v)) != old.v);
+               completed = true;
+       }
 
-       bch2_journal_space_available(j);
+       if (completed) {
+               bch2_journal_reclaim_fast(j);
+               bch2_journal_space_available(j);
 
-       closure_wake_up(&w->wait);
-       journal_wake(j);
+               track_event_change(&c->times[BCH_TIME_blocked_journal_max_in_flight], false);
 
-       if (!journal_state_count(new, new.unwritten_idx) &&
-           journal_last_unwritten_seq(j) <= journal_cur_seq(j)) {
-               spin_unlock(&j->lock);
-               closure_call(&j->io, bch2_journal_write, c->io_complete_wq, NULL);
-       } else if (journal_last_unwritten_seq(j) == journal_cur_seq(j) &&
+               closure_wake_up(&w->wait);
+               journal_wake(j);
+       }
+
+       if (journal_last_unwritten_seq(j) == journal_cur_seq(j) &&
                   new.cur_entry_offset < JOURNAL_ENTRY_CLOSED_VAL) {
                struct journal_buf *buf = journal_cur_buf(j);
                long delta = buf->expires - jiffies;
@@ -1602,47 +1692,46 @@ static void journal_write_done(struct closure *cl)
                 * previous entries still in flight - the current journal entry
                 * might want to be written now:
                 */
-
-               spin_unlock(&j->lock);
-               mod_delayed_work(c->io_complete_wq, &j->write_work, max(0L, delta));
-       } else {
-               spin_unlock(&j->lock);
+               mod_delayed_work(j->wq, &j->write_work, max(0L, delta));
        }
+
+       spin_unlock(&j->lock);
 }
 
 static void journal_write_endio(struct bio *bio)
 {
-       struct bch_dev *ca = bio->bi_private;
+       struct journal_bio *jbio = container_of(bio, struct journal_bio, bio);
+       struct bch_dev *ca = jbio->ca;
        struct journal *j = &ca->fs->journal;
-       struct journal_buf *w = journal_last_unwritten_buf(j);
-       unsigned long flags;
+       struct journal_buf *w = j->buf + jbio->buf_idx;
 
        if (bch2_dev_io_err_on(bio->bi_status, ca, BCH_MEMBER_ERROR_write,
                               "error writing journal entry %llu: %s",
                               le64_to_cpu(w->data->seq),
                               bch2_blk_status_to_str(bio->bi_status)) ||
            bch2_meta_write_fault("journal")) {
+               unsigned long flags;
+
                spin_lock_irqsave(&j->err_lock, flags);
                bch2_dev_list_drop_dev(&w->devs_written, ca->dev_idx);
                spin_unlock_irqrestore(&j->err_lock, flags);
        }
 
-       closure_put(&j->io);
+       closure_put(&w->io);
        percpu_ref_put(&ca->io_ref);
 }
 
-static void do_journal_write(struct closure *cl)
+static CLOSURE_CALLBACK(do_journal_write)
 {
-       struct journal *j = container_of(cl, struct journal, io);
+       closure_type(w, struct journal_buf, io);
+       struct journal *j = container_of(w, struct journal, buf[w->idx]);
        struct bch_fs *c = container_of(j, struct bch_fs, journal);
-       struct bch_dev *ca;
-       struct journal_buf *w = journal_last_unwritten_buf(j);
-       struct bch_extent_ptr *ptr;
-       struct bio *bio;
        unsigned sectors = vstruct_sectors(w->data, c->block_bits);
 
        extent_for_each_ptr(bkey_i_to_s_extent(&w->key), ptr) {
-               ca = bch_dev_bkey_exists(c, ptr->dev);
+               struct bch_dev *ca = bch_dev_bkey_exists(c, ptr->dev);
+               struct journal_device *ja = &ca->journal;
+
                if (!percpu_ref_tryget(&ca->io_ref)) {
                        /* XXX: fix this */
                        bch_err(c, "missing device for journal write\n");
@@ -1652,7 +1741,7 @@ static void do_journal_write(struct closure *cl)
                this_cpu_add(ca->io_done->sectors[WRITE][BCH_DATA_journal],
                             sectors);
 
-               bio = ca->journal.bio;
+               struct bio *bio = &ja->bio[w->idx]->bio;
                bio_reset(bio, ca->disk_sb.bdev, REQ_OP_WRITE|REQ_SYNC|REQ_META);
                bio->bi_iter.bi_sector  = ptr->offset;
                bio->bi_end_io          = journal_write_endio;
@@ -1671,16 +1760,23 @@ static void do_journal_write(struct closure *cl)
                trace_and_count(c, journal_write, bio);
                closure_bio_submit(bio, cl);
 
-               ca->journal.bucket_seq[ca->journal.cur_idx] =
-                       le64_to_cpu(w->data->seq);
+               ja->bucket_seq[ja->cur_idx] = le64_to_cpu(w->data->seq);
        }
 
-       continue_at(cl, journal_write_done, c->io_complete_wq);
+       continue_at(cl, journal_write_done, j->wq);
 }
 
-static void bch2_journal_entries_postprocess(struct bch_fs *c, struct jset *jset)
+static int bch2_journal_write_prep(struct journal *j, struct journal_buf *w)
 {
-       struct jset_entry *i, *next, *prev = NULL;
+       struct bch_fs *c = container_of(j, struct bch_fs, journal);
+       struct jset_entry *start, *end;
+       struct jset *jset = w->data;
+       struct journal_keys_to_wb wb = { NULL };
+       unsigned sectors, bytes, u64s;
+       unsigned long btree_roots_have = 0;
+       bool validate_before_checksum = false;
+       u64 seq = le64_to_cpu(jset->seq);
+       int ret;
 
        /*
         * Simple compaction, dropping empty jset_entries (from journal
@@ -1690,120 +1786,62 @@ static void bch2_journal_entries_postprocess(struct bch_fs *c, struct jset *jset
         * If we wanted to be really fancy here, we could sort all the keys in
         * the jset and drop keys that were overwritten - probably not worth it:
         */
-       vstruct_for_each_safe(jset, i, next) {
+       vstruct_for_each(jset, i) {
                unsigned u64s = le16_to_cpu(i->u64s);
 
                /* Empty entry: */
                if (!u64s)
                        continue;
 
-               if (i->type == BCH_JSET_ENTRY_btree_root)
+               /*
+                * New btree roots are set by journalling them; when the journal
+                * entry gets written we have to propagate them to
+                * c->btree_roots
+                *
+                * But, every journal entry we write has to contain all the
+                * btree roots (at least for now); so after we copy btree roots
+                * to c->btree_roots we have to get any missing btree roots and
+                * add them to this journal entry:
+                */
+               switch (i->type) {
+               case BCH_JSET_ENTRY_btree_root:
                        bch2_journal_entry_to_btree_root(c, i);
-
-               /* Can we merge with previous entry? */
-               if (prev &&
-                   i->btree_id == prev->btree_id &&
-                   i->level    == prev->level &&
-                   i->type     == prev->type &&
-                   i->type     == BCH_JSET_ENTRY_btree_keys &&
-                   le16_to_cpu(prev->u64s) + u64s <= U16_MAX) {
-                       memmove_u64s_down(vstruct_next(prev),
-                                         i->_data,
-                                         u64s);
-                       le16_add_cpu(&prev->u64s, u64s);
-                       continue;
+                       __set_bit(i->btree_id, &btree_roots_have);
+                       break;
+               case BCH_JSET_ENTRY_write_buffer_keys:
+                       EBUG_ON(!w->need_flush_to_write_buffer);
+
+                       if (!wb.wb)
+                               bch2_journal_keys_to_write_buffer_start(c, &wb, seq);
+
+                       struct bkey_i *k;
+                       jset_entry_for_each_key(i, k) {
+                               ret = bch2_journal_key_to_wb(c, &wb, i->btree_id, k);
+                               if (ret) {
+                                       bch2_fs_fatal_error(c, "-ENOMEM flushing journal keys to btree write buffer");
+                                       bch2_journal_keys_to_write_buffer_end(c, &wb);
+                                       return ret;
+                               }
+                       }
+                       i->type = BCH_JSET_ENTRY_btree_keys;
+                       break;
                }
-
-               /* Couldn't merge, move i into new position (after prev): */
-               prev = prev ? vstruct_next(prev) : jset->start;
-               if (i != prev)
-                       memmove_u64s_down(prev, i, jset_u64s(u64s));
-       }
-
-       prev = prev ? vstruct_next(prev) : jset->start;
-       jset->u64s = cpu_to_le32((u64 *) prev - jset->_data);
-}
-
-void bch2_journal_write(struct closure *cl)
-{
-       struct journal *j = container_of(cl, struct journal, io);
-       struct bch_fs *c = container_of(j, struct bch_fs, journal);
-       struct bch_dev *ca;
-       struct journal_buf *w = journal_last_unwritten_buf(j);
-       struct bch_replicas_padded replicas;
-       struct jset_entry *start, *end;
-       struct jset *jset;
-       struct bio *bio;
-       struct printbuf journal_debug_buf = PRINTBUF;
-       bool validate_before_checksum = false;
-       unsigned i, sectors, bytes, u64s, nr_rw_members = 0;
-       int ret;
-
-       BUG_ON(BCH_SB_CLEAN(c->disk_sb.sb));
-
-       journal_buf_realloc(j, w);
-       jset = w->data;
-
-       j->write_start_time = local_clock();
-
-       spin_lock(&j->lock);
-
-       /*
-        * If the journal is in an error state - we did an emergency shutdown -
-        * we prefer to continue doing journal writes. We just mark them as
-        * noflush so they'll never be used, but they'll still be visible by the
-        * list_journal tool - this helps in debugging.
-        *
-        * There's a caveat: the first journal write after marking the
-        * superblock dirty must always be a flush write, because on startup
-        * from a clean shutdown we didn't necessarily read the journal and the
-        * new journal write might overwrite whatever was in the journal
-        * previously - we can't leave the journal without any flush writes in
-        * it.
-        *
-        * So if we're in an error state, and we're still starting up, we don't
-        * write anything at all.
-        */
-       if (!test_bit(JOURNAL_NEED_FLUSH_WRITE, &j->flags) &&
-           (bch2_journal_error(j) ||
-            w->noflush ||
-            (!w->must_flush &&
-             (jiffies - j->last_flush_write) < msecs_to_jiffies(c->opts.journal_flush_delay) &&
-             test_bit(JOURNAL_MAY_SKIP_FLUSH, &j->flags)))) {
-               w->noflush = true;
-               SET_JSET_NO_FLUSH(jset, true);
-               jset->last_seq  = 0;
-               w->last_seq     = 0;
-
-               j->nr_noflush_writes++;
-       } else if (!bch2_journal_error(j)) {
-               j->last_flush_write = jiffies;
-               j->nr_flush_writes++;
-               clear_bit(JOURNAL_NEED_FLUSH_WRITE, &j->flags);
-       } else {
-               spin_unlock(&j->lock);
-               goto err;
        }
-       spin_unlock(&j->lock);
-
-       /*
-        * New btree roots are set by journalling them; when the journal entry
-        * gets written we have to propagate them to c->btree_roots
-        *
-        * But, every journal entry we write has to contain all the btree roots
-        * (at least for now); so after we copy btree roots to c->btree_roots we
-        * have to get any missing btree roots and add them to this journal
-        * entry:
-        */
 
-       bch2_journal_entries_postprocess(c, jset);
+       if (wb.wb)
+               bch2_journal_keys_to_write_buffer_end(c, &wb);
+       w->need_flush_to_write_buffer = false;
 
        start = end = vstruct_last(jset);
 
-       end     = bch2_btree_roots_to_journal_entries(c, jset->start, end);
+       end     = bch2_btree_roots_to_journal_entries(c, end, btree_roots_have);
 
-       bch2_journal_super_entries_add_common(c, &end,
-                               le64_to_cpu(jset->seq));
+       struct jset_entry_datetime *d =
+               container_of(jset_entry_init(&end, sizeof(*d)), struct jset_entry_datetime, entry);
+       d->entry.type   = BCH_JSET_ENTRY_datetime;
+       d->seconds      = cpu_to_le64(ktime_get_real_seconds());
+
+       bch2_journal_super_entries_add_common(c, &end, seq);
        u64s    = (u64 *) end - (u64 *) start;
        BUG_ON(u64s > j->entry_u64s_reserved);
 
@@ -1816,7 +1854,7 @@ void bch2_journal_write(struct closure *cl)
                bch2_fs_fatal_error(c, "aieeee! journal write overran available space, %zu > %u (extra %u reserved %u/%u)",
                                    vstruct_bytes(jset), w->sectors << 9,
                                    u64s, w->u64s_reserved, j->entry_u64s_reserved);
-               goto err;
+               return -EINVAL;
        }
 
        jset->magic             = cpu_to_le64(jset_magic(c));
@@ -1826,7 +1864,7 @@ void bch2_journal_write(struct closure *cl)
        SET_JSET_CSUM_TYPE(jset, bch2_meta_checksum_type(c));
 
        if (!JSET_NO_FLUSH(jset) && journal_entry_empty(jset))
-               j->last_empty_seq = le64_to_cpu(jset->seq);
+               j->last_empty_seq = seq;
 
        if (bch2_csum_type_is_encryption(JSET_CSUM_TYPE(jset)))
                validate_before_checksum = true;
@@ -1835,64 +1873,142 @@ void bch2_journal_write(struct closure *cl)
                validate_before_checksum = true;
 
        if (validate_before_checksum &&
-           jset_validate(c, NULL, jset, 0, WRITE))
-               goto err;
+           (ret = jset_validate(c, NULL, jset, 0, WRITE)))
+               return ret;
 
        ret = bch2_encrypt(c, JSET_CSUM_TYPE(jset), journal_nonce(jset),
                    jset->encrypted_start,
                    vstruct_end(jset) - (void *) jset->encrypted_start);
        if (bch2_fs_fatal_err_on(ret, c,
                        "error decrypting journal entry: %i", ret))
-               goto err;
+               return ret;
 
        jset->csum = csum_vstruct(c, JSET_CSUM_TYPE(jset),
                                  journal_nonce(jset), jset);
 
        if (!validate_before_checksum &&
-           jset_validate(c, NULL, jset, 0, WRITE))
-               goto err;
+           (ret = jset_validate(c, NULL, jset, 0, WRITE)))
+               return ret;
 
        memset((void *) jset + bytes, 0, (sectors << 9) - bytes);
+       return 0;
+}
+
+static int bch2_journal_write_pick_flush(struct journal *j, struct journal_buf *w)
+{
+       struct bch_fs *c = container_of(j, struct bch_fs, journal);
+       int error = bch2_journal_error(j);
+
+       /*
+        * If the journal is in an error state - we did an emergency shutdown -
+        * we prefer to continue doing journal writes. We just mark them as
+        * noflush so they'll never be used, but they'll still be visible by the
+        * list_journal tool - this helps in debugging.
+        *
+        * There's a caveat: the first journal write after marking the
+        * superblock dirty must always be a flush write, because on startup
+        * from a clean shutdown we didn't necessarily read the journal and the
+        * new journal write might overwrite whatever was in the journal
+        * previously - we can't leave the journal without any flush writes in
+        * it.
+        *
+        * So if we're in an error state, and we're still starting up, we don't
+        * write anything at all.
+        */
+       if (error && test_bit(JOURNAL_NEED_FLUSH_WRITE, &j->flags))
+               return -EIO;
+
+       if (error ||
+           w->noflush ||
+           (!w->must_flush &&
+            (jiffies - j->last_flush_write) < msecs_to_jiffies(c->opts.journal_flush_delay) &&
+            test_bit(JOURNAL_MAY_SKIP_FLUSH, &j->flags))) {
+               w->noflush = true;
+               SET_JSET_NO_FLUSH(w->data, true);
+               w->data->last_seq       = 0;
+               w->last_seq             = 0;
+
+               j->nr_noflush_writes++;
+       } else {
+               j->last_flush_write = jiffies;
+               j->nr_flush_writes++;
+               clear_bit(JOURNAL_NEED_FLUSH_WRITE, &j->flags);
+       }
+
+       return 0;
+}
+
+CLOSURE_CALLBACK(bch2_journal_write)
+{
+       closure_type(w, struct journal_buf, io);
+       struct journal *j = container_of(w, struct journal, buf[w->idx]);
+       struct bch_fs *c = container_of(j, struct bch_fs, journal);
+       struct bch_replicas_padded replicas;
+       struct printbuf journal_debug_buf = PRINTBUF;
+       unsigned nr_rw_members = 0;
+       int ret;
+
+       BUG_ON(BCH_SB_CLEAN(c->disk_sb.sb));
+       BUG_ON(w->write_allocated);
+
+       j->write_start_time = local_clock();
 
-retry_alloc:
        spin_lock(&j->lock);
-       ret = journal_write_alloc(j, w);
+       ret = bch2_journal_write_pick_flush(j, w);
+       spin_unlock(&j->lock);
+       if (ret)
+               goto err;
+
+       mutex_lock(&j->buf_lock);
+       journal_buf_realloc(j, w);
+
+       ret = bch2_journal_write_prep(j, w);
+       mutex_unlock(&j->buf_lock);
+       if (ret)
+               goto err;
+
+       j->entry_bytes_written += vstruct_bytes(w->data);
+
+       while (1) {
+               spin_lock(&j->lock);
+               ret = journal_write_alloc(j, w);
+               if (!ret || !j->can_discard)
+                       break;
 
-       if (ret && j->can_discard) {
                spin_unlock(&j->lock);
                bch2_journal_do_discards(j);
-               goto retry_alloc;
        }
 
-       if (ret)
+       if (ret) {
                __bch2_journal_debug_to_text(&journal_debug_buf, j);
+               spin_unlock(&j->lock);
+               bch_err(c, "Unable to allocate journal write:\n%s",
+                       journal_debug_buf.buf);
+               printbuf_exit(&journal_debug_buf);
+               goto err;
+       }
 
        /*
         * write is allocated, no longer need to account for it in
         * bch2_journal_space_available():
         */
        w->sectors = 0;
+       w->write_allocated = true;
 
        /*
         * journal entry has been compacted and allocated, recalculate space
         * available:
         */
        bch2_journal_space_available(j);
+       bch2_journal_do_writes(j);
        spin_unlock(&j->lock);
 
-       if (ret) {
-               bch_err(c, "Unable to allocate journal write:\n%s",
-                       journal_debug_buf.buf);
-               printbuf_exit(&journal_debug_buf);
-               goto err;
-       }
-
        w->devs_written = bch2_bkey_devs(bkey_i_to_s_c(&w->key));
 
        if (c->opts.nochanges)
                goto no_io;
 
-       for_each_rw_member(ca, c, i)
+       for_each_rw_member(c, ca)
                nr_rw_members++;
 
        if (nr_rw_members > 1)
@@ -1908,24 +2024,29 @@ retry_alloc:
        if (ret)
                goto err;
 
-       if (!JSET_NO_FLUSH(jset) && w->separate_flush) {
-               for_each_rw_member(ca, c, i) {
+       if (!JSET_NO_FLUSH(w->data))
+               closure_wait_event(&j->async_wait, j->seq_ondisk + 1 == le64_to_cpu(w->data->seq));
+
+       if (!JSET_NO_FLUSH(w->data) && w->separate_flush) {
+               for_each_rw_member(c, ca) {
                        percpu_ref_get(&ca->io_ref);
 
-                       bio = ca->journal.bio;
-                       bio_reset(bio, ca->disk_sb.bdev, REQ_OP_FLUSH);
+                       struct journal_device *ja = &ca->journal;
+                       struct bio *bio = &ja->bio[w->idx]->bio;
+                       bio_reset(bio, ca->disk_sb.bdev,
+                                 REQ_OP_WRITE|REQ_SYNC|REQ_META|REQ_PREFLUSH);
                        bio->bi_end_io          = journal_write_endio;
                        bio->bi_private         = ca;
                        closure_bio_submit(bio, cl);
                }
        }
 
-       continue_at(cl, do_journal_write, c->io_complete_wq);
+       continue_at(cl, do_journal_write, j->wq);
        return;
 no_io:
-       continue_at(cl, journal_write_done, c->io_complete_wq);
+       continue_at(cl, journal_write_done, j->wq);
        return;
 err:
        bch2_fatal_error(c);
-       continue_at(cl, journal_write_done, c->io_complete_wq);
+       continue_at(cl, journal_write_done, j->wq);
 }