btrfs: raid56: fix use-after-free of rbio in bio end_io wakeups

The read/write submission rounds in raid56 count in-flight bios in
rbio->stripes_pending and wait with a bare
wait_event(rbio->io_wait, atomic_read(&rbio->stripes_pending) == 0),
while each bio's end_io does
"if (atomic_dec_and_test(&stripes_pending)) wake_up(&rbio->io_wait)".

Once the final decrement makes the counter visible as zero, the waiter
can pass its condition check without consuming the wakeup -- via
wait_event()'s fast path when all bios complete before the waiter
arrives, or a condition recheck after an earlier wakeup -- and proceed
to free the rbio through rbio_orig_end_io().  The end_io context is
then still inside wake_up() operating on the freed rbio's embedded
waitqueue lock.

This shows up under sustained raid56 RMW load in KVM guests as
recurring "pvqspinlock: lock ... has corrupted value 0x0!" warnings
from __pv_queued_spin_unlock_slowpath with a
__wake_up <- raid_wait_write_end_io <- bio_endio call trace: paravirt
spinlocks detect the unlock of the recycled lock word.  On bare metal
the use-after-free is silent and almost always harmless, which is how
it has survived; it is a plausible match for long-standing sporadic
crash reports on busy raid5 filesystems that never reproduce under
sanitizer kernels (the instrumentation widens the dec-to-wake window
so the race is always lost).

Convert the pair to a completion.  wait_for_completion()'s fast path
takes the completion's own lock, so it cannot return before complete()
has released it: the completing context is provably finished with the
rbio before the waiter can free it.  A submitter-held bias count keeps
one completion per submission round even when a round submits zero
bios or every bio finishes before the submitter starts waiting; the
scrub path only waits when finish_parity_scrub() actually began a
round.

Assisted-by: Claude:claude-fable-5
Signed-off-by: Zygo Blaxell <ce3g8jdj@umail.furryterror.org>

diff --git a/fs/btrfs/raid56.c b/fs/btrfs/raid56.c
index 8ec24dbb180f9266b04625c9f11dd9b649d7a3f4..4e1bede0d3d9126a0153ef08d7dc56ae6194ad07 100644 (file)
--- a/fs/btrfs/raid56.c
+++ b/fs/btrfs/raid56.c
@@ -1095,7 +1095,7 @@ static struct btrfs_raid_bio *alloc_rbio(struct btrfs_fs_info *fs_info,
        }
 
        bio_list_init(&rbio->bio_list);
-       init_waitqueue_head(&rbio->io_wait);
+       init_completion(&rbio->io_done);
        INIT_LIST_HEAD(&rbio->plug_list);
        spin_lock_init(&rbio->bio_list_lock);
        INIT_LIST_HEAD(&rbio->stripe_cache);
@@ -1676,6 +1676,40 @@ static void verify_bio_data_sectors(struct btrfs_raid_bio *rbio,
        }
 }
 
+/*
+ * Begin a submission round of @nr bios whose end_io functions report through
+ * rbio_dec_io().  stripes_pending gets one extra count on behalf of the
+ * submitter, dropped by rbio_wait_io(): this guarantees exactly one
+ * completion per round even when @nr is 0 or every bio finishes before the
+ * submitter starts waiting.
+ *
+ * The completion is what makes the round safe against a use-after-free the
+ * old bare waitqueue had: once the final bio's decrement made
+ * stripes_pending visible as zero, the waiter could pass its wait_event()
+ * condition check and free the rbio while the bio's end_io context was still
+ * inside wake_up() on the rbio's embedded waitqueue lock.  With a
+ * completion, the waiter cannot return before complete() has released the
+ * completion's lock, so the completing context is finished with the rbio.
+ */
+static void rbio_begin_io(struct btrfs_raid_bio *rbio, unsigned int nr)
+{
+       reinit_completion(&rbio->io_done);
+       atomic_set(&rbio->stripes_pending, nr + 1);
+}
+
+static void rbio_dec_io(struct btrfs_raid_bio *rbio)
+{
+       if (atomic_dec_and_test(&rbio->stripes_pending))
+               complete(&rbio->io_done);
+}
+
+static void rbio_wait_io(struct btrfs_raid_bio *rbio)
+{
+       /* Drop the submitter's count from rbio_begin_io(). */
+       rbio_dec_io(rbio);
+       wait_for_completion(&rbio->io_done);
+}
+
 static void raid_wait_read_end_io(struct bio *bio)
 {
        struct btrfs_raid_bio *rbio = bio->bi_private;
@@ -1688,8 +1722,7 @@ static void raid_wait_read_end_io(struct bio *bio)
        }
 
        bio_put(bio);
-       if (atomic_dec_and_test(&rbio->stripes_pending))
-               wake_up(&rbio->io_wait);
+       rbio_dec_io(rbio);
 }
 
 static void submit_read_wait_bio_list(struct btrfs_raid_bio *rbio,
@@ -1697,7 +1730,7 @@ static void submit_read_wait_bio_list(struct btrfs_raid_bio *rbio,
 {
        struct bio *bio;
 
-       atomic_set(&rbio->stripes_pending, bio_list_size(bio_list));
+       rbio_begin_io(rbio, bio_list_size(bio_list));
        while ((bio = bio_list_pop(bio_list))) {
                bio->bi_end_io = raid_wait_read_end_io;
 
@@ -1710,7 +1743,7 @@ static void submit_read_wait_bio_list(struct btrfs_raid_bio *rbio,
                submit_bio(bio);
        }
 
-       wait_event(rbio->io_wait, atomic_read(&rbio->stripes_pending) == 0);
+       rbio_wait_io(rbio);
 }
 
 static int alloc_rbio_data_pages(struct btrfs_raid_bio *rbio)
@@ -2394,8 +2427,7 @@ static void raid_wait_write_end_io(struct bio *bio)
        if (bio->bi_status)
                rbio_update_error_bitmap(rbio, bio);
        bio_put(bio);
-       if (atomic_dec_and_test(&rbio->stripes_pending))
-               wake_up(&rbio->io_wait);
+       rbio_dec_io(rbio);
 }
 
 static void submit_write_bios(struct btrfs_raid_bio *rbio,
@@ -2403,7 +2435,7 @@ static void submit_write_bios(struct btrfs_raid_bio *rbio,
 {
        struct bio *bio;
 
-       atomic_set(&rbio->stripes_pending, bio_list_size(bio_list));
+       rbio_begin_io(rbio, bio_list_size(bio_list));
        while ((bio = bio_list_pop(bio_list))) {
                bio->bi_end_io = raid_wait_write_end_io;
 
@@ -2509,7 +2541,7 @@ static void rmw_rbio(struct btrfs_raid_bio *rbio)
        /* We should have at least one bio assembled. */
        ASSERT(bio_list_size(&bio_list));
        submit_write_bios(rbio, &bio_list);
-       wait_event(rbio->io_wait, atomic_read(&rbio->stripes_pending) == 0);
+       rbio_wait_io(rbio);
 
        /* We may have more errors than our tolerance during the read. */
        for (sectornr = 0; sectornr < rbio->stripe_nsectors; sectornr++) {
@@ -2969,7 +3001,9 @@ static void scrub_rbio(struct btrfs_raid_bio *rbio)
         * and writeback the good content.
         */
        ret = finish_parity_scrub(rbio);
-       wait_event(rbio->io_wait, atomic_read(&rbio->stripes_pending) == 0);
+       /* A submission round only began if finish_parity_scrub() succeeded. */
+       if (ret == 0)
+               rbio_wait_io(rbio);
        for (sector_nr = 0; sector_nr < rbio->stripe_nsectors; sector_nr++) {
                int found_errors;
 

diff --git a/fs/btrfs/raid56.h b/fs/btrfs/raid56.h
index 8542648199f1a485371e3622c74aff028293b0a0..47f8e85db2949e73e2007876d1114d8b4278ae30 100644 (file)
--- a/fs/btrfs/raid56.h
+++ b/fs/btrfs/raid56.h
@@ -13,6 +13,7 @@
 #include <linux/bio.h>
 #include <linux/refcount.h>
 #include <linux/workqueue.h>
+#include <linux/completion.h>
 #include "volumes.h"
 
 struct page;
@@ -178,9 +179,16 @@ struct btrfs_raid_bio {
 
        refcount_t refs;
 
+       /*
+        * In-flight bios of the current submission round, plus one bias
+        * count held by the submitter and dropped when it starts waiting.
+        * The final decrement completes io_done; a completion (unlike a
+        * bare waitqueue) guarantees the completing context is finished
+        * with the rbio before the waiter can return and free it.
+        */
        atomic_t stripes_pending;
 
-       wait_queue_head_t io_wait;
+       struct completion io_done;
 
        /* Bitmap to record which horizontal stripe has data */
        unsigned long dbitmap;