migration/rdma: producer/consumer send path

index 444123cbca..50f5d7b839 100644
@@ -302,6 +302,12 @@ typedef struct RDMAContext {
     /* number of outstanding writes */
     int nb_sent;
 
+    /* nb_sent statistics */
+    uint64_t nbsent_max;            /* peak outstanding writes */
+    uint64_t nbsent_posts;          /* posts (nb_sent++) */
+    uint64_t nbsent_full_hits;      /* posts that hit a full SQ */
+    uint64_t nbsent_zero_hits;      /* drains to zero */
+
     /* store info about current buffer so that we can
        merge it with future sends */
     uint64_t current_addr;

@@ -1164,6 +1170,11 @@ static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
     return result;
 }
 
+/* nb_sent statistics counters */
+static void rdma_nbsent_inc(RDMAContext *rdma, uint64_t nb_sent);
+static void rdma_nbsent_full(RDMAContext *rdma);
+static void rdma_nbsent_zero(RDMAContext *rdma);
+
 /*
  * Consult the connection manager to see a work request
  * (of any kind) has completed.

@@ -1213,6 +1224,9 @@ static int qemu_rdma_poll(RDMAContext *rdma, struct ibv_cq *cq,
 
         if (rdma->nb_sent > 0) {
             rdma->nb_sent--;
+            if (rdma->nb_sent == 0) {
+                rdma_nbsent_zero(rdma);
+            }
         }
     } else {
         trace_rdma_poll_other(wr_id, rdma->nb_sent);

@@ -1438,6 +1452,7 @@ static bool qemu_rdma_post_send(RDMAContext *rdma,
     ret = ibv_post_send(rdma->qp, send_wr, &bad_wr);
     if (ret == ENOMEM && rdma->nb_sent) {
         trace_rdma_post_send_queue_full(send_wr->wr_id, rdma->nb_sent);
+        rdma_nbsent_full(rdma);
         ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
         if (ret < 0) {
             error_setg(errp, "rdma migration: failed to make room for %s",

@@ -1953,6 +1968,7 @@ static int qemu_rdma_write_flush(RDMAContext *rdma, Error **errp)
     if (ret == 0) {
         rdma->nb_sent++;
         trace_rdma_write_flush(rdma->nb_sent);
+        rdma_nbsent_inc(rdma, rdma->nb_sent);
     }
 
     rdma->current_length = 0;

@@ -2998,6 +3014,41 @@ err:
     return -1;
 }
 
+static void rdma_nbsent_inc(RDMAContext *rdma, uint64_t nb_sent)
+{
+    rdma->nbsent_posts++;
+    if (nb_sent > rdma->nbsent_max) {
+        rdma->nbsent_max = nb_sent;
+    }
+}
+
+static void rdma_nbsent_full(RDMAContext *rdma)
+{
+    rdma->nbsent_full_hits++;
+}
+
+static void rdma_nbsent_zero(RDMAContext *rdma)
+{
+    rdma->nbsent_zero_hits++;
+}
+
+static void rdma_stats_reset(RDMAContext *rdma)
+{
+    rdma->nbsent_max = 0;
+    rdma->nbsent_posts = 0;
+    rdma->nbsent_full_hits = 0;
+    rdma->nbsent_zero_hits = 0;
+}
+
+static void rdma_nbsent_stats_dump(RDMAContext *rdma)
+{
+    if (rdma->nbsent_posts == 0) {
+        return;
+    }
+    trace_rdma_nbsent_stats(rdma->nbsent_max, rdma->nbsent_posts,
+                            rdma->nbsent_full_hits, rdma->nbsent_zero_hits);
+}
+
 int rdma_control_save_page(QEMUFile *f, ram_addr_t block_offset,
                            ram_addr_t offset, size_t size)
 {

@@ -3547,6 +3598,10 @@ int rdma_registration_start(QEMUFile *f, uint64_t flags)
         return -1;
     }
 
+    if (flags == RAM_CONTROL_SETUP) {
+        rdma_stats_reset(rdma);
+    }
+
     trace_rdma_registration_start(flags);
     qemu_put_be64(f, RAM_SAVE_FLAG_HOOK);
     return qemu_fflush(f);

@@ -3581,6 +3636,11 @@ int rdma_registration_stop(QEMUFile *f, uint64_t flags)
     }
 
     qemu_fflush(f);
+
+    if (flags == RAM_CONTROL_FINISH) {
+        rdma_nbsent_stats_dump(rdma);
+    }
+
     ret = qemu_rdma_write_flush(rdma, &err);
 
     if (ret < 0) {

index 50f5d7b839..e4ede950de 100644
@@ -29,6 +29,7 @@
 #include "qemu/main-loop.h"
 #include "qemu/module.h"
 #include "qemu/rcu.h"
+#include "qemu/thread.h"
 #include "qemu/sockets.h"
 #include "qemu/coroutine.h"
 #include "system/memory.h"

@@ -278,6 +279,33 @@ typedef struct RDMALocalBlocks {
     RDMALocalBlock *block;
 } RDMALocalBlocks;
 
+/*
+ * Producer/consumer split (source side): the migration thread (producer)
+ * scans dirty pages and enqueues chunk-write and barrier commands into an
+ * SPSC ring; a dedicated consumer thread owns the QP/CQ and does all RDMA.
+ */
+typedef enum {
+    RDMA_CMD_WRITE,     /* write one dirty span: block/chunk + offset/length */
+    RDMA_CMD_FLUSH,     /* drain nb_sent -> 0 (round boundary) */
+    RDMA_CMD_REG_STOP,  /* send REGISTER_FINISHED; flags = RAM_CONTROL_* */
+    RDMA_CMD_FILE,      /* QEMUFile control bytes: iov/niov */
+    RDMA_CMD_FINISH,    /* drain then stop the consumer */
+} RDMACmdType;
+
+typedef struct RDMACmd {
+    RDMACmdType type;
+    uint32_t block_index;
+    uint32_t chunk_index;
+    uint64_t offset;              /* WRITE: dirty span start, block-relative */
+    uint32_t length;             /* WRITE: dirty span length (within a chunk) */
+    uint64_t flags;               /* REG_STOP */
+    const struct iovec *iov;      /* FILE (producer blocks -> stays valid) */
+    size_t niov;                  /* FILE */
+    unsigned seq;                 /* barrier completion sequence (0 = none) */
+} RDMACmd;
+
+typedef struct RDMASendRing RDMASendRing;
+
 /*
  * Main data structure for RDMA state.
  * While there is only one copy of this structure being allocated right now,

@@ -317,6 +345,17 @@ typedef struct RDMAContext {
     /* index of the chunk in the current ram block */
     int current_chunk;
 
+    /* producer/consumer split (source side) */
+    RDMASendRing *send_ring;      /* SPSC ring; allocated at consumer start */
+    QemuThread consumer_thread;
+    bool consumer_started;        /* producer-only: route via ring once set */
+
+    /* producer-only: pending contiguous WRITE span, not yet enqueued */
+    int pend_block;               /* block index, -1 = none */
+    uint32_t pend_chunk;          /* chunk index of the span */
+    uint64_t pend_offset;         /* span start, block-relative bytes */
+    uint64_t pend_length;         /* span length in bytes (within one chunk) */
+
     bool pin_all;
 
     /*

@@ -2671,6 +2710,49 @@ static NotifierWithReturn rdma_ram_round_notifier = {
     .notify = rdma_ram_round_notify,
 };
 
+/*
+ * ---------------------------------------------------------------------------
+ * Producer/consumer send ring (source side).
+ *
+ * Lock-free SPSC ring: the producer (migration thread) owns head, the
+ * consumer (RDMA thread) owns tail.  A mutex+condvars are used only to
+ * sleep/wake at the empty/full/abort/barrier edges (guarded by the *_waiting
+ * flags so the steady state stays lock-free).  Barriers ride the ring as
+ * ordered descriptors and publish completion via barrier_done_seq.
+ *
+ * head/tail are free-running unsigned counters, never reset.  Wrap is
+ * implicit: occupancy is the modular difference head - tail (always in
+ * [0, N] since the producer stops at full), and the slot index is
+ * counter & MASK.  Because N is a power of two and 2^32 is an exact multiple
+ * of N, masking stays continuous across the 2^32 counter wrap.  Barrier seqs
+ * use the same modular scheme with signed distance; seq 0 is reserved as the
+ * "no barrier" sentinel and skipped on wrap.
+ * ---------------------------------------------------------------------------
+ */
+#define RDMA_SEND_RING_SIZE 2048            /* power of two; > SQ depth (512) */
+#define RDMA_SEND_RING_MASK (RDMA_SEND_RING_SIZE - 1)
+
+struct RDMASendRing {
+    RDMACmd  ring[RDMA_SEND_RING_SIZE];
+    unsigned head;                          /* producer owns */
+    unsigned tail;                          /* consumer owns */
+    unsigned barrier_done_seq;              /* consumer owns; producer reads */
+    unsigned barrier_next_seq;              /* producer owns */
+    bool     abort;                         /* sticky */
+    QemuMutex wait_lock;
+    QemuCond  not_full;                     /* producer sleeps */
+    QemuCond  not_empty;                    /* consumer sleeps */
+    QemuCond  barrier_cv;                   /* producer sleeps for ack */
+    bool     producer_waiting;
+    bool     consumer_waiting;
+
+    /* statistics, dumped at FINISH */
+    uint64_t pushwait_max_ns;    /* longest producer wait on a full ring */
+    uint64_t pushwait_total_ns;  /* total producer wait on a full ring */
+    uint64_t occupancy_max;      /* peak queued commands */
+    uint64_t idle_max_ns;        /* longest consumer wait on an empty ring */
+};
+
 static int qio_channel_rdma_set_blocking(QIOChannel *ioc,
                                          bool blocking,
                                          Error **errp)

index e4ede950de..2d4807c6b8 100644
@@ -1214,6 +1214,9 @@ static void rdma_nbsent_inc(RDMAContext *rdma, uint64_t nb_sent);
 static void rdma_nbsent_full(RDMAContext *rdma);
 static void rdma_nbsent_zero(RDMAContext *rdma);
 
+static void rdma_consumer_start(RDMAContext *rdma);
+static void rdma_consumer_stop(RDMAContext *rdma);
+
 /*
  * Consult the connection manager to see a work request
  * (of any kind) has completed.

@@ -2108,6 +2111,9 @@ static void qemu_rdma_cleanup(RDMAContext *rdma)
 {
     Error *err = NULL;
 
+    /* Free the send ring before touching the QP. */
+    rdma_consumer_stop(rdma);
+
     if (rdma->cm_id && rdma->connected) {
         if ((rdma->errored ||
              migrate_get_current()->state == MIGRATION_STATUS_CANCELLING) &&

@@ -2753,6 +2759,91 @@ struct RDMASendRing {
     uint64_t idle_max_ns;        /* longest consumer wait on an empty ring */
 };
 
+static bool ring_full(RDMASendRing *r)
+{
+    return (r->head - qatomic_load_acquire(&r->tail)) >= RDMA_SEND_RING_SIZE;
+}
+
+/* Producer -> wake the consumer if it is idle on an empty ring. */
+static void ring_wake_consumer(RDMASendRing *r)
+{
+    smp_mb();   /* order the *_waiting load below vs the sleeper's store */
+    if (qatomic_read(&r->consumer_waiting)) {
+        qemu_mutex_lock(&r->wait_lock);
+        qemu_cond_signal(&r->not_empty);
+        qemu_mutex_unlock(&r->wait_lock);
+    }
+}
+
+static int ring_push(RDMAContext *rdma, const RDMACmd *cmd)
+{
+    RDMASendRing *r = rdma->send_ring;
+
+    assert(r);
+
+    if (ring_full(r)) {
+        qemu_mutex_lock(&r->wait_lock);
+        r->producer_waiting = true;
+        smp_mb();                       /* pair with ring_wake_producer() */
+        while (ring_full(r) && !r->abort) {
+            qemu_cond_wait(&r->not_full, &r->wait_lock);
+        }
+        r->producer_waiting = false;
+        qemu_mutex_unlock(&r->wait_lock);
+    }
+    if (qatomic_read(&r->abort)) {
+        return -1;
+    }
+    r->ring[r->head & RDMA_SEND_RING_MASK] = *cmd;
+    qatomic_store_release(&r->head, r->head + 1);   /* publish payload+slot */
+    ring_wake_consumer(r);
+    return 0;
+}
+
+/*
+ * Producer: allocate the send ring once (idempotent).  The consumer
+ * thread that drains it is created in a later change.
+ */
+static void rdma_consumer_start(RDMAContext *rdma)
+{
+    RDMASendRing *r;
+
+    if (rdma->consumer_started) {
+        return;
+    }
+    r = g_new0(RDMASendRing, 1);
+    qemu_mutex_init(&r->wait_lock);
+    qemu_cond_init(&r->not_full);
+    qemu_cond_init(&r->not_empty);
+    qemu_cond_init(&r->barrier_cv);
+    rdma->send_ring = r;
+    rdma->pend_block = -1;        /* no pending WRITE span yet */
+    /* Producer-only flag (migration thread + cleanup, never concurrent). */
+    rdma->consumer_started = true;
+}
+
+/* Producer: tear down the send ring. */
+static void rdma_consumer_stop(RDMAContext *rdma)
+{
+    RDMASendRing *r = rdma->send_ring;
+    RDMACmd cmd = { .type = RDMA_CMD_FINISH };
+
+    if (!rdma->consumer_started) {
+        return;
+    }
+    /* Ask the consumer to drain and exit; if it already aborted, just join. */
+    if (!qatomic_read(&r->abort)) {
+        ring_push(rdma, &cmd);
+    }
+    qemu_mutex_destroy(&r->wait_lock);
+    qemu_cond_destroy(&r->not_full);
+    qemu_cond_destroy(&r->not_empty);
+    qemu_cond_destroy(&r->barrier_cv);
+    g_free(r);
+    rdma->send_ring = NULL;
+    rdma->consumer_started = false;
+}
+
 static int qio_channel_rdma_set_blocking(QIOChannel *ioc,
                                          bool blocking,
                                          Error **errp)

@@ -3810,6 +3901,11 @@ int rdma_registration_stop(QEMUFile *f, uint64_t flags)
         goto err;
     }
 
+    if (flags == RAM_CONTROL_SETUP) {
+        /* Setup complete: allocate the send ring (consumer added later). */
+        rdma_consumer_start(rdma);
+    }
+
     return 0;
 err:
     rdma->errored = true;

index 2d4807c6b8..c3070b7a92 100644
@@ -2111,7 +2111,7 @@ static void qemu_rdma_cleanup(RDMAContext *rdma)
 {
     Error *err = NULL;
 
-    /* Free the send ring before touching the QP. */
+    /* Stop and join the consumer before touching the QP it owns. */
     rdma_consumer_stop(rdma);
 
     if (rdma->cm_id && rdma->connected) {

@@ -2764,6 +2764,37 @@ static bool ring_full(RDMASendRing *r)
     return (r->head - qatomic_load_acquire(&r->tail)) >= RDMA_SEND_RING_SIZE;
 }
 
+static bool ring_empty(RDMASendRing *r)
+{
+    return (qatomic_load_acquire(&r->head) - r->tail) == 0;
+}
+
+static void ring_set_abort(RDMAContext *rdma)
+{
+    RDMASendRing *r = rdma->send_ring;
+
+    if (!r) {
+        return;
+    }
+    qemu_mutex_lock(&r->wait_lock);
+    qatomic_set(&r->abort, true);
+    qemu_cond_broadcast(&r->not_full);
+    qemu_cond_broadcast(&r->not_empty);
+    qemu_cond_broadcast(&r->barrier_cv);
+    qemu_mutex_unlock(&r->wait_lock);
+}
+
+static void ring_wake_producer(RDMASendRing *r)
+{
+    smp_mb();   /* order the *_waiting load below vs the sleeper's store */
+    if (qatomic_read(&r->producer_waiting)) {
+        qemu_mutex_lock(&r->wait_lock);
+        qemu_cond_broadcast(&r->not_full);
+        qemu_cond_broadcast(&r->barrier_cv);
+        qemu_mutex_unlock(&r->wait_lock);
+    }
+}
+
 /* Producer -> wake the consumer if it is idle on an empty ring. */
 static void ring_wake_consumer(RDMASendRing *r)
 {

@@ -2800,10 +2831,84 @@ static int ring_push(RDMAContext *rdma, const RDMACmd *cmd)
     return 0;
 }
 
-/*
- * Producer: allocate the send ring once (idempotent).  The consumer
- * thread that drains it is created in a later change.
- */
+static bool ring_pop(RDMASendRing *r, RDMACmd *cmd)
+{
+    if (ring_empty(r)) {
+        return false;
+    }
+    *cmd = r->ring[r->tail & RDMA_SEND_RING_MASK];   /* after load_acquire */
+    return true;
+}
+
+static void ring_advance_tail(RDMASendRing *r)
+{
+    qatomic_store_release(&r->tail, r->tail + 1);
+    ring_wake_producer(r);
+}
+
+/* Consumer: sleep until the ring is non-empty or aborted. */
+static void ring_idle_wait(RDMASendRing *r)
+{
+    qemu_mutex_lock(&r->wait_lock);
+    r->consumer_waiting = true;
+    smp_mb();                           /* pair with ring_wake_consumer() */
+    while (ring_empty(r) && !r->abort) {
+        qemu_cond_wait(&r->not_empty, &r->wait_lock);
+    }
+    r->consumer_waiting = false;
+    qemu_mutex_unlock(&r->wait_lock);
+}
+
+static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
+                                  bool *stop)
+{
+    Error *err = NULL;
+    int ret = 0;
+
+    switch (cmd->type) {
+    case RDMA_CMD_FINISH:
+        ret = qemu_rdma_drain_cq(rdma, &err);
+        *stop = true;
+        break;
+    default:
+        break;
+    }
+    if (ret < 0 && err) {
+        error_report_err(err);
+    }
+    return ret;
+}
+
+static void *rdma_consumer_thread(void *opaque)
+{
+    RDMAContext *rdma = opaque;
+    RDMASendRing *r = rdma->send_ring;
+    bool stop = false;
+
+    rcu_register_thread();
+
+    while (!qatomic_read(&r->abort) && !stop) {
+        RDMACmd cmd;
+
+        if (ring_pop(r, &cmd)) {
+            if (rdma_consumer_dispatch(rdma, &cmd, &stop) < 0) {
+                ring_set_abort(rdma);
+                break;
+            }
+            ring_advance_tail(r);
+            continue;
+        }
+        ring_idle_wait(r);
+    }
+
+    if (qatomic_read(&r->abort)) {
+        /* make the failure visible to the producer's rdma_errored() checks */
+        rdma->errored = true;
+    }
+    rcu_unregister_thread();
+    return NULL;
+}
+
 static void rdma_consumer_start(RDMAContext *rdma)
 {
     RDMASendRing *r;

@@ -2820,9 +2925,10 @@ static void rdma_consumer_start(RDMAContext *rdma)
     rdma->pend_block = -1;        /* no pending WRITE span yet */
     /* Producer-only flag (migration thread + cleanup, never concurrent). */
     rdma->consumer_started = true;
+    qemu_thread_create(&rdma->consumer_thread, "rdma-send",
+                       rdma_consumer_thread, rdma, QEMU_THREAD_JOINABLE);
 }
 
-/* Producer: tear down the send ring. */
 static void rdma_consumer_stop(RDMAContext *rdma)
 {
     RDMASendRing *r = rdma->send_ring;

@@ -2835,6 +2941,8 @@ static void rdma_consumer_stop(RDMAContext *rdma)
     if (!qatomic_read(&r->abort)) {
         ring_push(rdma, &cmd);
     }
+    qemu_thread_join(&rdma->consumer_thread);
+
     qemu_mutex_destroy(&r->wait_lock);
     qemu_cond_destroy(&r->not_full);
     qemu_cond_destroy(&r->not_empty);

index c3070b7a92..d11f6e0da8 100644
@@ -1216,6 +1216,7 @@ static void rdma_nbsent_zero(RDMAContext *rdma);
 
 static void rdma_consumer_start(RDMAContext *rdma);
 static void rdma_consumer_stop(RDMAContext *rdma);
+static void rdma_nbsent_stats_dump(RDMAContext *rdma);
 
 /*
  * Consult the connection manager to see a work request

@@ -1986,6 +1987,33 @@ static int qemu_rdma_write_one(RDMAContext *rdma,
     return 0;
 }
 
+/* Consumer: write one whole chunk */
+static int qemu_rdma_write_chunk(RDMAContext *rdma, uint32_t block_index,
+                                 uint32_t chunk_index, Error **errp)
+{
+    RDMALocalBlock *block = &rdma->local_ram_blocks.block[block_index];
+    uint64_t chunk_off = (uint64_t)chunk_index * migrate_rdma_chunk_size();
+    uint64_t len = block->length - chunk_off;
+    int ret;
+
+    if (len > migrate_rdma_chunk_size()) {
+        len = migrate_rdma_chunk_size();
+    }
+
+    /* write_one wants current_addr in ram_addr_t space (block->offset based) */
+    ret = qemu_rdma_write_one(rdma, block_index, block->offset + chunk_off,
+                              len, errp);
+    if (ret < 0) {
+        return -1;
+    }
+    if (ret == 0) {
+        rdma->nb_sent++;
+        trace_rdma_write_flush(rdma->nb_sent);
+        rdma_nbsent_inc(rdma, rdma->nb_sent);
+    }
+    return 0;
+}
+
 /*
  * Push out any unwritten RDMA operations.
  *

@@ -2846,6 +2874,13 @@ static void ring_advance_tail(RDMASendRing *r)
     ring_wake_producer(r);
 }
 
+/* Consumer: publish barrier completion. */
+static void ring_barrier_complete(RDMASendRing *r, unsigned seq)
+{
+    qatomic_store_release(&r->barrier_done_seq, seq);
+    ring_wake_producer(r);
+}
+
 /* Consumer: sleep until the ring is non-empty or aborted. */
 static void ring_idle_wait(RDMASendRing *r)
 {

@@ -2859,6 +2894,23 @@ static void ring_idle_wait(RDMASendRing *r)
     qemu_mutex_unlock(&r->wait_lock);
 }
 
+/* Consumer: send REGISTER_FINISHED; for FINISH drain and dump stats first. */
+static int rdma_consumer_reg_stop(RDMAContext *rdma, uint64_t flags,
+                                  Error **errp)
+{
+    RDMAControlHeader head = { .len = 0,
+                               .type = RDMA_CONTROL_REGISTER_FINISHED,
+                               .repeat = 1 };
+
+    if (flags == RAM_CONTROL_FINISH) {
+        if (qemu_rdma_drain_cq(rdma, errp) < 0) {
+            return -1;
+        }
+        rdma_nbsent_stats_dump(rdma);
+    }
+    return qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL, errp);
+}
+
 static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
                                   bool *stop)
 {

@@ -2866,6 +2918,16 @@ static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
     int ret = 0;
 
     switch (cmd->type) {
+    case RDMA_CMD_WRITE:
+        ret = qemu_rdma_write_chunk(rdma, cmd->block_index,
+                                    cmd->chunk_index, &err);
+        break;
+    case RDMA_CMD_FLUSH:
+        ret = qemu_rdma_drain_cq(rdma, &err);
+        break;
+    case RDMA_CMD_REG_STOP:
+        ret = rdma_consumer_reg_stop(rdma, cmd->flags, &err);
+        break;
     case RDMA_CMD_FINISH:
         ret = qemu_rdma_drain_cq(rdma, &err);
         *stop = true;

@@ -2895,6 +2957,9 @@ static void *rdma_consumer_thread(void *opaque)
                 ring_set_abort(rdma);
                 break;
             }
+            if (cmd.seq) {
+                ring_barrier_complete(r, cmd.seq);
+            }
             ring_advance_tail(r);
             continue;
         }

index d11f6e0da8..5ec6a87819 100644
@@ -2523,6 +2523,34 @@ static RDMAContext *qemu_rdma_data_init(InetSocketAddress *saddr, Error **errp)
  * SEND messages for control only.
  * VM's ram is handled with regular RDMA messages.
  */
+static ssize_t qemu_rdma_write_file(RDMAContext *rdma, const struct iovec *iov,
+                                    size_t niov, Error **errp)
+{
+    ssize_t done = 0;
+
+    for (size_t i = 0; i < niov; i++) {
+        size_t remaining = iov[i].iov_len;
+        uint8_t *data = iov[i].iov_base;
+
+        while (remaining) {
+            RDMAControlHeader head = {};
+            size_t len = MIN(remaining, RDMA_SEND_INCREMENT);
+
+            remaining -= len;
+            head.len = len;
+            head.type = RDMA_CONTROL_QEMU_FILE;
+
+            if (qemu_rdma_exchange_send(rdma, &head, data,
+                                        NULL, NULL, NULL, errp) < 0) {
+                return -1;
+            }
+            data += len;
+            done += len;
+        }
+    }
+    return done;
+}
+
 static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
                                        const struct iovec *iov,
                                        size_t niov,

@@ -2535,7 +2563,6 @@ static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
     RDMAContext *rdma;
     int ret;
     ssize_t done = 0;
-    size_t len;
 
     RCU_READ_LOCK_GUARD();
     rdma = qatomic_rcu_read(&rioc->rdmaout);

@@ -2561,29 +2588,10 @@ static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
         return -1;
     }
 
-    for (int i = 0; i < niov; i++) {
-        size_t remaining = iov[i].iov_len;
-        uint8_t * data = (void *)iov[i].iov_base;
-        while (remaining) {
-            RDMAControlHeader head = {};
-
-            len = MIN(remaining, RDMA_SEND_INCREMENT);
-            remaining -= len;
-
-            head.len = len;
-            head.type = RDMA_CONTROL_QEMU_FILE;
-
-            ret = qemu_rdma_exchange_send(rdma, &head,
-                                          data, NULL, NULL, NULL, errp);
-
-            if (ret < 0) {
-                rdma->errored = true;
-                return -1;
-            }
-
-            data += len;
-            done += len;
-        }
+    done = qemu_rdma_write_file(rdma, iov, niov, errp);
+    if (done < 0) {
+        rdma->errored = true;
+        return -1;
     }
 
     return done;

@@ -2928,12 +2936,14 @@ static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
     case RDMA_CMD_REG_STOP:
         ret = rdma_consumer_reg_stop(rdma, cmd->flags, &err);
         break;
+    case RDMA_CMD_FILE:
+        ret = qemu_rdma_write_file(rdma, cmd->iov, cmd->niov, &err) < 0 ?
+              -1 : 0;
+        break;
     case RDMA_CMD_FINISH:
         ret = qemu_rdma_drain_cq(rdma, &err);
         *stop = true;
         break;
-    default:
-        break;
     }
     if (ret < 0 && err) {
         error_report_err(err);

index 5ec6a87819..a0ff73e313 100644
@@ -327,7 +327,7 @@ typedef struct RDMAContext {
      */
     int control_ready_expected;
 
-    /* number of outstanding writes */
+    /* number of outstanding writes (consumer thread only) */
     int nb_sent;
 
     /* nb_sent statistics */

@@ -336,15 +336,6 @@ typedef struct RDMAContext {
     uint64_t nbsent_full_hits;      /* posts that hit a full SQ */
     uint64_t nbsent_zero_hits;      /* drains to zero */
 
-    /* store info about current buffer so that we can
-       merge it with future sends */
-    uint64_t current_addr;
-    uint64_t current_length;
-    /* index of ram block the current buffer belongs to */
-    int current_index;
-    /* index of the chunk in the current ram block */
-    int current_chunk;
-
     /* producer/consumer split (source side) */
     RDMASendRing *send_ring;      /* SPSC ring; allocated at consumer start */
     QemuThread consumer_thread;

@@ -1216,6 +1207,12 @@ static void rdma_nbsent_zero(RDMAContext *rdma);
 
 static void rdma_consumer_start(RDMAContext *rdma);
 static void rdma_consumer_stop(RDMAContext *rdma);
+static int ring_push_write(RDMAContext *rdma, uint32_t bidx, uint32_t cidx,
+                           uint64_t offset, uint64_t length);
+static int producer_flush_pending(RDMAContext *rdma);
+static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
+                        const struct iovec *iov, size_t niov);
+static void ring_set_abort(RDMAContext *rdma);
 static void rdma_nbsent_stats_dump(RDMAContext *rdma);
 
 /*

@@ -1987,151 +1984,24 @@ static int qemu_rdma_write_one(RDMAContext *rdma,
     return 0;
 }
 
-/* Consumer: write one whole chunk */
+/* Consumer: write one dirty span (offset/length are block-relative bytes) */
 static int qemu_rdma_write_chunk(RDMAContext *rdma, uint32_t block_index,
-                                 uint32_t chunk_index, Error **errp)
+                                 uint64_t offset, uint64_t length, Error **errp)
 {
     RDMALocalBlock *block = &rdma->local_ram_blocks.block[block_index];
-    uint64_t chunk_off = (uint64_t)chunk_index * migrate_rdma_chunk_size();
-    uint64_t len = block->length - chunk_off;
     int ret;
 
-    if (len > migrate_rdma_chunk_size()) {
-        len = migrate_rdma_chunk_size();
-    }
-
     /* write_one wants current_addr in ram_addr_t space (block->offset based) */
-    ret = qemu_rdma_write_one(rdma, block_index, block->offset + chunk_off,
-                              len, errp);
-    if (ret < 0) {
-        return -1;
-    }
-    if (ret == 0) {
-        rdma->nb_sent++;
-        trace_rdma_write_flush(rdma->nb_sent);
-        rdma_nbsent_inc(rdma, rdma->nb_sent);
-    }
-    return 0;
-}
-
-/*
- * Push out any unwritten RDMA operations.
- *
- * We support sending out multiple chunks at the same time.
- * Not all of them need to get signaled in the completion queue.
- */
-static int qemu_rdma_write_flush(RDMAContext *rdma, Error **errp)
-{
-    int ret;
-
-    if (!rdma->current_length) {
-        return 0;
-    }
-
-    ret = qemu_rdma_write_one(rdma, rdma->current_index, rdma->current_addr,
-                              rdma->current_length, errp);
-
+    ret = qemu_rdma_write_one(rdma, block_index, block->offset + offset,
+                              length, errp);
     if (ret < 0) {
         return -1;
     }
-
     if (ret == 0) {
         rdma->nb_sent++;
         trace_rdma_write_flush(rdma->nb_sent);
         rdma_nbsent_inc(rdma, rdma->nb_sent);
     }
-
-    rdma->current_length = 0;
-    rdma->current_addr = 0;
-
-    return 0;
-}
-
-static inline bool qemu_rdma_buffer_mergeable(RDMAContext *rdma,
-                    uint64_t offset, uint64_t len)
-{
-    RDMALocalBlock *block;
-    uint8_t *host_addr;
-    uint8_t *chunk_end;
-
-    if (rdma->current_index < 0) {
-        return false;
-    }
-
-    if (rdma->current_chunk < 0) {
-        return false;
-    }
-
-    block = &(rdma->local_ram_blocks.block[rdma->current_index]);
-    host_addr = block->local_host_addr + (offset - block->offset);
-    chunk_end = ram_chunk_end(block, rdma->current_chunk);
-
-    if (rdma->current_length == 0) {
-        return false;
-    }
-
-    /*
-     * Only merge into chunk sequentially.
-     */
-    if (offset != (rdma->current_addr + rdma->current_length)) {
-        return false;
-    }
-
-    if (offset < block->offset) {
-        return false;
-    }
-
-    if ((offset + len) > (block->offset + block->length)) {
-        return false;
-    }
-
-    if ((host_addr + len) > chunk_end) {
-        return false;
-    }
-
-    return true;
-}
-
-/*
- * We're not actually writing here, but doing three things:
- *
- * 1. Identify the chunk the buffer belongs to.
- * 2. If the chunk is full or the buffer doesn't belong to the current
- *    chunk, then start a new chunk and flush() the old chunk.
- * 3. To keep the hardware busy, we also group chunks into batches
- *    and only require that a batch gets acknowledged in the completion
- *    queue instead of each individual chunk.
- */
-static int qemu_rdma_write(RDMAContext *rdma,
-                           uint64_t block_offset, uint64_t offset,
-                           uint64_t len, Error **errp)
-{
-    uint64_t current_addr = block_offset + offset;
-    uint64_t index = rdma->current_index;
-    uint64_t chunk = rdma->current_chunk;
-
-    /* If we cannot merge it, we flush the current buffer first. */
-    if (!qemu_rdma_buffer_mergeable(rdma, current_addr, len)) {
-        if (qemu_rdma_write_flush(rdma, errp) < 0) {
-            return -1;
-        }
-        rdma->current_length = 0;
-        rdma->current_addr = current_addr;
-
-        qemu_rdma_search_ram_block(rdma, block_offset,
-                                   offset, len, &index, &chunk);
-        rdma->current_index = index;
-        rdma->current_chunk = chunk;
-    }
-
-    /* merge it */
-    rdma->current_length += len;
-
-    /* flush it if buffer is too large */
-    if (rdma->current_length >= rdma_merge_max()) {
-        return qemu_rdma_write_flush(rdma, errp);
-    }
-
     return 0;
 }
 

@@ -2510,8 +2380,7 @@ static RDMAContext *qemu_rdma_data_init(InetSocketAddress *saddr, Error **errp)
     RDMAContext *rdma = NULL;
 
     rdma = g_new0(RDMAContext, 1);
-    rdma->current_index = -1;
-    rdma->current_chunk = -1;
+    rdma->pend_block = -1;        /* no pending WRITE span yet */
 
     rdma->host = g_strdup(saddr->host);
     rdma->port = atoi(saddr->port);

@@ -2561,7 +2430,6 @@ static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
 {
     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(ioc);
     RDMAContext *rdma;
-    int ret;
     ssize_t done = 0;
 
     RCU_READ_LOCK_GUARD();

@@ -2578,22 +2446,28 @@ static ssize_t qio_channel_rdma_writev(QIOChannel *ioc,
         return -1;
     }
 
+    if (!rdma->consumer_started) {
+        /* Setup phase: no consumer yet, no queued RAM writes -> send inline. */
+        done = qemu_rdma_write_file(rdma, iov, niov, errp);
+        if (done < 0) {
+            rdma->errored = true;
+            return -1;
+        }
+        return done;
+    }
+
     /*
-     * Push out any writes that
-     * we're queued up for VM's ram.
+     * Consumer owns the QP.  Route through the ring so this control message
+     * is ordered after all chunk writes queued so far; block until it has
+     * been sent (iov stays valid for the duration).
      */
-    ret = qemu_rdma_write_flush(rdma, errp);
-    if (ret < 0) {
-        rdma->errored = true;
+    if (ring_barrier(rdma, RDMA_CMD_FILE, 0, iov, niov) < 0) {
+        error_setg(errp, "RDMA consumer aborted during control write");
         return -1;
     }
-
-    done = qemu_rdma_write_file(rdma, iov, niov, errp);
-    if (done < 0) {
-        rdma->errored = true;
-        return -1;
+    for (size_t i = 0; i < niov; i++) {
+        done += iov[i].iov_len;
     }
-
     return done;
 }
 

@@ -2705,10 +2579,6 @@ static ssize_t qio_channel_rdma_readv(QIOChannel *ioc,
 /* Block until all outstanding writes have been delivered by the hardware. */
 static int qemu_rdma_drain_cq(RDMAContext *rdma, Error **errp)
 {
-    if (qemu_rdma_write_flush(rdma, errp) < 0) {
-        return -1;
-    }
-
     while (rdma->nb_sent) {
         if (qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL) < 0) {
             error_setg(errp, "rdma migration: completion polling error");

@@ -2745,7 +2615,16 @@ static int rdma_ram_round_notify(NotifierWithReturn *n G_GNUC_UNUSED,
         return -1;
     }
 
-    return qemu_rdma_drain_cq(rdma, errp);
+    /* Complete all chunks in the ring before the next round. */
+    if (ring_barrier(rdma, RDMA_CMD_FLUSH, 0, NULL, 0) < 0) {
+        error_setg(errp, "RDMA consumer aborted during round drain");
+        return -1;
+    }
+    /*
+     * The FLUSH barrier already flushed any pending span; start the next
+     * round with a clean accumulator (no cross-round merging).
+     */
+    return 0;
 }
 
 static NotifierWithReturn rdma_ram_round_notifier = {

@@ -2846,6 +2725,7 @@ static int ring_push(RDMAContext *rdma, const RDMACmd *cmd)
 {
     RDMASendRing *r = rdma->send_ring;
 
+    /* SETUP starts the consumer before any WRITE/FLUSH/REG_STOP/FILE. */
     assert(r);
 
     if (ring_full(r)) {

@@ -2867,6 +2747,87 @@ static int ring_push(RDMAContext *rdma, const RDMACmd *cmd)
     return 0;
 }
 
+static int ring_push_write(RDMAContext *rdma, uint32_t bidx, uint32_t cidx,
+                           uint64_t offset, uint64_t length)
+{
+    RDMACmd cmd = {
+        .type = RDMA_CMD_WRITE,
+        .block_index = bidx,
+        .chunk_index = cidx,
+        .offset = offset,
+        .length = length,
+    };
+    return ring_push(rdma, &cmd);
+}
+
+/* Producer: enqueue the accumulated contiguous span (if any) as one WRITE. */
+static int producer_flush_pending(RDMAContext *rdma)
+{
+    int ret;
+
+    if (rdma->pend_block < 0) {
+        return 0;
+    }
+    ret = ring_push_write(rdma, rdma->pend_block, rdma->pend_chunk,
+                          rdma->pend_offset, rdma->pend_length);
+    rdma->pend_block = -1;
+    return ret;
+}
+
+/* Producer: block until the consumer has completed barrier seq (or abort). */
+static int ring_wait_barrier(RDMASendRing *r, unsigned seq)
+{
+    if ((int)(qatomic_load_acquire(&r->barrier_done_seq) - seq) < 0) {
+        qemu_mutex_lock(&r->wait_lock);
+        r->producer_waiting = true;
+        smp_mb();                       /* pair with ring_wake_producer() */
+        while ((int)(qatomic_load_acquire(&r->barrier_done_seq) - seq) < 0 &&
+               !r->abort) {
+            qemu_cond_wait(&r->barrier_cv, &r->wait_lock);
+        }
+        r->producer_waiting = false;
+        qemu_mutex_unlock(&r->wait_lock);
+    }
+    /* A completed barrier is success even after a late abort. */
+    return (int)(qatomic_load_acquire(&r->barrier_done_seq) - seq) >= 0 ?
+           0 : -1;
+}
+
+/* Producer: enqueue a barrier command and wait for its completion. */
+static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
+                        const struct iovec *iov, size_t niov)
+{
+    RDMASendRing *r = rdma->send_ring;
+    unsigned seq;
+    RDMACmd cmd;
+
+    assert(r);          /* consumer started at SETUP; see ring_push() */
+
+    /* Order any accumulated writes ahead of this barrier. */
+    if (producer_flush_pending(rdma) < 0) {
+        return -1;
+    }
+
+    seq = ++r->barrier_next_seq;
+
+    if (seq == 0) {                 /* seq 0 is the "no barrier" sentinel */
+        seq = ++r->barrier_next_seq;
+    }
+    cmd = (RDMACmd){
+        .type = type,
+        .flags = flags,
+        .iov = iov,
+        .niov = niov,
+        .seq = seq,
+    };
+
+    if (ring_push(rdma, &cmd) < 0) {
+        return -1;
+    }
+
+    return ring_wait_barrier(r, seq);
+}
+
 static bool ring_pop(RDMASendRing *r, RDMACmd *cmd)
 {
     if (ring_empty(r)) {

@@ -2928,7 +2889,7 @@ static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
     switch (cmd->type) {
     case RDMA_CMD_WRITE:
         ret = qemu_rdma_write_chunk(rdma, cmd->block_index,
-                                    cmd->chunk_index, &err);
+                                    cmd->offset, cmd->length, &err);
         break;
     case RDMA_CMD_FLUSH:
         ret = qemu_rdma_drain_cq(rdma, &err);

@@ -2973,6 +2934,15 @@ static void *rdma_consumer_thread(void *opaque)
             ring_advance_tail(r);
             continue;
         }
+        if (rdma->nb_sent > 0) {
+            /* Ring empty but writes in flight: reap (bounded, error-aware). */
+            if (qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE,
+                                         NULL) < 0) {
+                ring_set_abort(rdma);
+                break;
+            }
+            continue;
+        }
         ring_idle_wait(r);
     }
 

@@ -3014,6 +2984,7 @@ static void rdma_consumer_stop(RDMAContext *rdma)
     }
     /* Ask the consumer to drain and exit; if it already aborted, just join. */
     if (!qatomic_read(&r->abort)) {
+        producer_flush_pending(rdma);   /* order the last span before FINISH */
         ring_push(rdma, &cmd);
     }
     qemu_thread_join(&rdma->consumer_thread);

@@ -3298,7 +3269,8 @@ static int qemu_rdma_save_page(QEMUFile *f, ram_addr_t block_offset,
     QIOChannelRDMA *rioc = QIO_CHANNEL_RDMA(qemu_file_get_ioc(f));
     Error *err = NULL;
     RDMAContext *rdma;
-    int ret;
+    RDMALocalBlock *block;
+    uint64_t bidx, first_chunk, chunk_size, pos, end;
 
     RCU_READ_LOCK_GUARD();
     rdma = qatomic_rcu_read(&rioc->rdmaout);

@@ -3313,60 +3285,46 @@ static int qemu_rdma_save_page(QEMUFile *f, ram_addr_t block_offset,
 
     qemu_fflush(f);
 
-    /*
-     * Add this page to the current 'chunk'. If the chunk
-     * is full, or the page doesn't belong to the current chunk,
-     * an actual RDMA write will occur and a new chunk will be formed.
-     */
-    ret = qemu_rdma_write(rdma, block_offset, offset, size, &err);
-    if (ret < 0) {
-        error_report_err(err);
-        goto err;
-    }
+    qemu_rdma_search_ram_block(rdma, block_offset, offset, size,
+                               &bidx, &first_chunk);
+    block = &rdma->local_ram_blocks.block[bidx];
+    chunk_size = migrate_rdma_chunk_size();
 
     /*
-     * Drain the Completion Queue if possible, but do not block,
-     * just poll.
-     *
-     * If nothing to poll, the end of the iteration will do this
-     * again to make sure we don't overflow the request queue.
+     * Split the dirty range into per-chunk segments (a WRITE never crosses a
+     * chunk boundary) and coalesce with the pending span: pages are scanned
+     * in increasing order, so contiguous dirty pages extend one span, and we
+     * enqueue only the actually-dirty bytes instead of whole chunks.
      */
-    while (1) {
-        uint64_t wr_id, wr_id_in;
-        ret = qemu_rdma_poll(rdma, rdma->recv_cq, &wr_id_in, NULL);
-
-        if (ret < 0) {
-            error_report("rdma migration: polling error");
-            goto err;
-        }
-
-        wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
-
-        if (wr_id == RDMA_WRID_NONE) {
-            break;
-        }
-    }
-
-    while (1) {
-        uint64_t wr_id, wr_id_in;
-        ret = qemu_rdma_poll(rdma, rdma->send_cq, &wr_id_in, NULL);
-
-        if (ret < 0) {
-            error_report("rdma migration: polling error");
-            goto err;
-        }
-
-        wr_id = wr_id_in & RDMA_WRID_TYPE_MASK;
-
-        if (wr_id == RDMA_WRID_NONE) {
-            break;
+    pos = (block_offset + offset) - block->offset;
+    end = pos + size;
+    while (pos < end) {
+        uint64_t c = pos / chunk_size;
+        uint64_t seg_end = MIN(end, (c + 1) * chunk_size);
+        uint64_t seg_len = seg_end - pos;
+
+        if (rdma->pend_block == (int)bidx && rdma->pend_chunk == (uint32_t)c &&
+            rdma->pend_offset + rdma->pend_length == pos) {
+            rdma->pend_length += seg_len;       /* extend the current span */
+        } else {
+            if (producer_flush_pending(rdma) < 0) {
+                error_setg(&err, "rdma migration: consumer aborted");
+                goto err;
+            }
+            rdma->pend_block = bidx;
+            rdma->pend_chunk = c;
+            rdma->pend_offset = pos;
+            rdma->pend_length = seg_len;
         }
+        pos = seg_end;
     }
 
     return RAM_SAVE_CONTROL_DELAYED;
 
 err:
+    error_report_err(err);
     rdma->errored = true;
+    ring_set_abort(rdma);
     return -1;
 }
 

@@ -3991,20 +3949,25 @@ int rdma_registration_stop(QEMUFile *f, uint64_t flags)
         return -1;
     }
 
-    qemu_fflush(f);
-
-    if (flags == RAM_CONTROL_FINISH) {
-        rdma_nbsent_stats_dump(rdma);
-    }
+    trace_rdma_registration_stop(flags);
 
-    ret = qemu_rdma_write_flush(rdma, &err);
+    qemu_fflush(f);
 
-    if (ret < 0) {
-        error_report_err(err);
-        goto err;
-    }
+    if (flags != RAM_CONTROL_SETUP) {
+        /*
+         * ROUND / FINISH: the consumer sends REGISTER_FINISHED, ordered
+         * after all chunk writes queued so far.  For FINISH it also drains
+         * outstanding writes and dumps the stats first.
+         */
+        if (ring_barrier(rdma, RDMA_CMD_REG_STOP, flags, NULL, 0) < 0) {
+            error_report("rdma migration: consumer aborted at registration"
+                         " stop");
+            goto err;
+        }
+        return 0;
 
-    if (flags == RAM_CONTROL_SETUP) {
+    } else {
+        /* SETUP runs on the producer, before the consumer exists. */
         RDMAControlHeader resp = {.type = RDMA_CONTROL_RAM_BLOCKS_RESULT };
         RDMALocalBlocks *local = &rdma->local_ram_blocks;
         int reg_result_idx, nb_dest_blocks;

@@ -4072,26 +4035,24 @@ int rdma_registration_stop(QEMUFile *f, uint64_t flags)
                     rdma->dest_blocks[i].remote_host_addr;
             local->block[i].remote_rkey = rdma->dest_blocks[i].remote_rkey;
         }
-    }
-
-    trace_rdma_registration_stop(flags);
 
-    head.type = RDMA_CONTROL_REGISTER_FINISHED;
-    ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL, &err);
+        head.type = RDMA_CONTROL_REGISTER_FINISHED;
+        ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL,
+                                      NULL, &err);
 
-    if (ret < 0) {
-        error_report_err(err);
-        goto err;
-    }
+        if (ret < 0) {
+            error_report_err(err);
+            goto err;
+        }
 
-    if (flags == RAM_CONTROL_SETUP) {
-        /* Setup complete: allocate the send ring (consumer added later). */
+        /* Setup complete: start the consumer; it now owns all QP work. */
         rdma_consumer_start(rdma);
     }
 
     return 0;
 err:
     rdma->errored = true;
+    ring_set_abort(rdma);
     return -1;
 }
 

index a0ff73e313..628afaccaf 100644
@@ -32,6 +32,7 @@
 #include "qemu/thread.h"
 #include "qemu/sockets.h"
 #include "qemu/coroutine.h"
+#include "qemu/timer.h"
 #include "system/memory.h"
 #include <sys/socket.h>
 #include <netdb.h>

@@ -1213,7 +1214,9 @@ static int producer_flush_pending(RDMAContext *rdma);
 static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
                         const struct iovec *iov, size_t niov);
 static void ring_set_abort(RDMAContext *rdma);
+static unsigned ring_occupancy(RDMAContext *rdma);
 static void rdma_nbsent_stats_dump(RDMAContext *rdma);
+static void rdma_prodcons_stats_dump(RDMASendRing *r);
 
 /*
  * Consult the connection manager to see a work request

@@ -2616,6 +2619,7 @@ static int rdma_ram_round_notify(NotifierWithReturn *n G_GNUC_UNUSED,
     }
 
     /* Complete all chunks in the ring before the next round. */
+    trace_rdma_ram_round_notify(ring_occupancy(rdma));
     if (ring_barrier(rdma, RDMA_CMD_FLUSH, 0, NULL, 0) < 0) {
         error_setg(errp, "RDMA consumer aborted during round drain");
         return -1;

@@ -2684,6 +2688,14 @@ static bool ring_empty(RDMASendRing *r)
     return (qatomic_load_acquire(&r->head) - r->tail) == 0;
 }
 
+/* Producer-side snapshot of queued-but-unprocessed commands in the ring. */
+static unsigned ring_occupancy(RDMAContext *rdma)
+{
+    RDMASendRing *r = rdma->send_ring;
+
+    return r->head - qatomic_read(&r->tail);
+}
+
 static void ring_set_abort(RDMAContext *rdma)
 {
     RDMASendRing *r = rdma->send_ring;

@@ -2703,6 +2715,7 @@ static void ring_wake_producer(RDMASendRing *r)
 {
     smp_mb();   /* order the *_waiting load below vs the sleeper's store */
     if (qatomic_read(&r->producer_waiting)) {
+        trace_rdma_ring_wake_producer();
         qemu_mutex_lock(&r->wait_lock);
         qemu_cond_broadcast(&r->not_full);
         qemu_cond_broadcast(&r->barrier_cv);

@@ -2715,20 +2728,38 @@ static void ring_wake_consumer(RDMASendRing *r)
 {
     smp_mb();   /* order the *_waiting load below vs the sleeper's store */
     if (qatomic_read(&r->consumer_waiting)) {
+        trace_rdma_ring_wake_consumer();
         qemu_mutex_lock(&r->wait_lock);
         qemu_cond_signal(&r->not_empty);
         qemu_mutex_unlock(&r->wait_lock);
     }
 }
 
+static const char *rdma_cmd_type_str(RDMACmdType type)
+{
+    switch (type) {
+    case RDMA_CMD_WRITE:    return "WRITE";
+    case RDMA_CMD_FLUSH:    return "FLUSH";
+    case RDMA_CMD_REG_STOP: return "REG_STOP";
+    case RDMA_CMD_FILE:     return "FILE";
+    case RDMA_CMD_FINISH:   return "FINISH";
+    default:                return "UNKNOWN";
+    }
+}
+
 static int ring_push(RDMAContext *rdma, const RDMACmd *cmd)
 {
     RDMASendRing *r = rdma->send_ring;
+    unsigned occupancy;
 
     /* SETUP starts the consumer before any WRITE/FLUSH/REG_STOP/FILE. */
     assert(r);
+    trace_rdma_ring_push_begin(rdma_cmd_type_str(cmd->type),
+                               cmd->chunk_index);
 
     if (ring_full(r)) {
+        int64_t wait_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+
         qemu_mutex_lock(&r->wait_lock);
         r->producer_waiting = true;
         smp_mb();                       /* pair with ring_wake_producer() */

@@ -2737,12 +2768,23 @@ static int ring_push(RDMAContext *rdma, const RDMACmd *cmd)
         }
         r->producer_waiting = false;
         qemu_mutex_unlock(&r->wait_lock);
+
+        wait_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - wait_ns;
+        r->pushwait_total_ns += wait_ns;
+        if ((uint64_t)wait_ns > r->pushwait_max_ns) {
+            r->pushwait_max_ns = wait_ns;
+        }
+        trace_rdma_ring_push_full_wait(cmd->type, wait_ns);
     }
     if (qatomic_read(&r->abort)) {
         return -1;
     }
     r->ring[r->head & RDMA_SEND_RING_MASK] = *cmd;
     qatomic_store_release(&r->head, r->head + 1);   /* publish payload+slot */
+    occupancy = r->head - qatomic_read(&r->tail);
+    if (occupancy > r->occupancy_max) {
+        r->occupancy_max = occupancy;
+    }
     ring_wake_consumer(r);
     return 0;
 }

@@ -2798,8 +2840,10 @@ static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
                         const struct iovec *iov, size_t niov)
 {
     RDMASendRing *r = rdma->send_ring;
+    int64_t wait_ns;
     unsigned seq;
     RDMACmd cmd;
+    int ret;
 
     assert(r);          /* consumer started at SETUP; see ring_push() */
 

@@ -2813,6 +2857,7 @@ static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
     if (seq == 0) {                 /* seq 0 is the "no barrier" sentinel */
         seq = ++r->barrier_next_seq;
     }
+    trace_rdma_ring_barrier_begin(type, seq);
     cmd = (RDMACmd){
         .type = type,
         .flags = flags,

@@ -2822,10 +2867,16 @@ static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
     };
 
     if (ring_push(rdma, &cmd) < 0) {
+        trace_rdma_ring_barrier_end(type, seq, 0, -1);
         return -1;
     }
 
-    return ring_wait_barrier(r, seq);
+    /* time spent blocked waiting for the consumer to complete the barrier */
+    wait_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+    ret = ring_wait_barrier(r, seq);
+    wait_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - wait_ns;
+    trace_rdma_ring_barrier_end(type, seq, wait_ns, ret);
+    return ret;
 }
 
 static bool ring_pop(RDMASendRing *r, RDMACmd *cmd)

@@ -2853,12 +2904,21 @@ static void ring_barrier_complete(RDMASendRing *r, unsigned seq)
 /* Consumer: sleep until the ring is non-empty or aborted. */
 static void ring_idle_wait(RDMASendRing *r)
 {
+    int64_t start_ns, idle_ns;
+
     qemu_mutex_lock(&r->wait_lock);
     r->consumer_waiting = true;
     smp_mb();                           /* pair with ring_wake_consumer() */
+    start_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME);
+    trace_rdma_consumer_idle_begin();
     while (ring_empty(r) && !r->abort) {
         qemu_cond_wait(&r->not_empty, &r->wait_lock);
     }
+    idle_ns = qemu_clock_get_ns(QEMU_CLOCK_REALTIME) - start_ns;
+    trace_rdma_consumer_idle_end(idle_ns);
+    if ((uint64_t)idle_ns > r->idle_max_ns) {
+        r->idle_max_ns = idle_ns;
+    }
     r->consumer_waiting = false;
     qemu_mutex_unlock(&r->wait_lock);
 }

@@ -2876,6 +2936,7 @@ static int rdma_consumer_reg_stop(RDMAContext *rdma, uint64_t flags,
             return -1;
         }
         rdma_nbsent_stats_dump(rdma);
+        rdma_prodcons_stats_dump(rdma->send_ring);
     }
     return qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL, errp);
 }

@@ -2886,6 +2947,9 @@ static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
     Error *err = NULL;
     int ret = 0;
 
+    trace_rdma_consumer_dispatch(cmd->type, cmd->block_index,
+                                 cmd->chunk_index, cmd->seq);
+
     switch (cmd->type) {
     case RDMA_CMD_WRITE:
         ret = qemu_rdma_write_chunk(rdma, cmd->block_index,

@@ -2919,6 +2983,7 @@ static void *rdma_consumer_thread(void *opaque)
     bool stop = false;
 
     rcu_register_thread();
+    trace_rdma_consumer_thread_start();
 
     while (!qatomic_read(&r->abort) && !stop) {
         RDMACmd cmd;

@@ -2950,6 +3015,7 @@ static void *rdma_consumer_thread(void *opaque)
         /* make the failure visible to the producer's rdma_errored() checks */
         rdma->errored = true;
     }
+    trace_rdma_consumer_thread_stop(qatomic_read(&r->abort));
     rcu_unregister_thread();
     return NULL;
 }

@@ -3363,6 +3429,13 @@ static void rdma_nbsent_stats_dump(RDMAContext *rdma)
                             rdma->nbsent_full_hits, rdma->nbsent_zero_hits);
 }
 
+/* Producer/consumer ring stats for the whole migration (consumer at FINISH). */
+static void rdma_prodcons_stats_dump(RDMASendRing *r)
+{
+    trace_rdma_prodcons_stats(r->idle_max_ns, r->pushwait_max_ns,
+                              r->pushwait_total_ns, r->occupancy_max);
+}
+
 int rdma_control_save_page(QEMUFile *f, ram_addr_t block_offset,
                            ram_addr_t offset, size_t size)
 {

index 628afaccaf..db1f1e3fc6 100644
@@ -67,6 +67,9 @@ static inline uint64_t rdma_merge_max(void)
 #define RDMA_CONTROL_MAX_BUFFER (512 * 1024)
 #define RDMA_CONTROL_MAX_COMMANDS_PER_MESSAGE 4096
 
+/* Max WRITE commands the consumer drains per pass. */
+#define RDMA_CONSUMER_BATCH_MAX 128
+
 #define RDMA_CONTROL_VERSION_CURRENT 1
 /*
  * Capabilities for negotiation.

@@ -2688,6 +2691,12 @@ static bool ring_empty(RDMASendRing *r)
     return (qatomic_load_acquire(&r->head) - r->tail) == 0;
 }
 
+/* Consumer-side count of published, unconsumed commands. */
+static unsigned ring_avail(RDMASendRing *r)
+{
+    return qatomic_load_acquire(&r->head) - r->tail;
+}
+
 /* Producer-side snapshot of queued-but-unprocessed commands in the ring. */
 static unsigned ring_occupancy(RDMAContext *rdma)
 {

@@ -2879,18 +2888,10 @@ static int ring_barrier(RDMAContext *rdma, RDMACmdType type, uint64_t flags,
     return ret;
 }
 
-static bool ring_pop(RDMASendRing *r, RDMACmd *cmd)
+/* Consumer: release n drained slots at once, waking a blocked producer once. */
+static void ring_advance_tail_by(RDMASendRing *r, unsigned n)
 {
-    if (ring_empty(r)) {
-        return false;
-    }
-    *cmd = r->ring[r->tail & RDMA_SEND_RING_MASK];   /* after load_acquire */
-    return true;
-}
-
-static void ring_advance_tail(RDMASendRing *r)
-{
-    qatomic_store_release(&r->tail, r->tail + 1);
+    qatomic_store_release(&r->tail, r->tail + n);
     ring_wake_producer(r);
 }
 

@@ -2986,29 +2987,63 @@ static void *rdma_consumer_thread(void *opaque)
     trace_rdma_consumer_thread_start();
 
     while (!qatomic_read(&r->abort) && !stop) {
-        RDMACmd cmd;
+        unsigned avail = ring_avail(r);
+        unsigned batch, processed = 0;
+
+        if (avail == 0) {
+            /*
+             * Ring empty: reap ready completions without blocking, then
+             * sleep until the next push so a new command wakes us at once.
+             * Round draining is handled by RDMA_CMD_FLUSH.
+             */
+            while (rdma->nb_sent > 0) {
+                uint64_t wr_id;
+
+                if (qemu_rdma_poll(rdma, rdma->send_cq, &wr_id, NULL) < 0) {
+                    ring_set_abort(rdma);
+                    break;
+                }
+                if (wr_id == RDMA_WRID_NONE) {
+                    break;              /* nothing more ready right now */
+                }
+            }
+            if (qatomic_read(&r->abort)) {
+                break;
+            }
+            ring_idle_wait(r);
+            continue;
+        }
+
+        /*
+         * Drain up to RDMA_CONSUMER_BATCH_MAX commands, or fewer if that is
+         * all that is queued, then publish tail once for the whole batch.
+         */
+        batch = avail < RDMA_CONSUMER_BATCH_MAX ? avail
+                                                : RDMA_CONSUMER_BATCH_MAX;
+
+        while (processed < batch) {
+            unsigned idx = r->tail + processed;
+            RDMACmd cmd = r->ring[idx & RDMA_SEND_RING_MASK];
 
-        if (ring_pop(r, &cmd)) {
             if (rdma_consumer_dispatch(rdma, &cmd, &stop) < 0) {
                 ring_set_abort(rdma);
                 break;
             }
+            processed++;
             if (cmd.seq) {
                 ring_barrier_complete(r, cmd.seq);
             }
-            ring_advance_tail(r);
-            continue;
-        }
-        if (rdma->nb_sent > 0) {
-            /* Ring empty but writes in flight: reap (bounded, error-aware). */
-            if (qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE,
-                                         NULL) < 0) {
-                ring_set_abort(rdma);
+            if (stop) {
                 break;
             }
-            continue;
         }
-        ring_idle_wait(r);
+
+        if (processed) {
+            ring_advance_tail_by(r, processed);
+        }
+        if (qatomic_read(&r->abort)) {
+            break;
+        }
     }
 
     if (qatomic_read(&r->abort)) {

index db1f1e3fc6..bffc7d33ef 100644
@@ -3870,6 +3870,8 @@ int rdma_registration_handle(QEMUFile *f)
             trace_rdma_registration_handle_register(head.repeat);
 
             reg_resp.repeat = head.repeat;
+            /* one result per requested chunk (source may batch registers) */
+            reg_resp.len = head.repeat * sizeof(RDMARegisterResult);
             registers = (RDMARegister *) rdma->wr_data[idx].control_curr;
 
             /* Making sure the register buffers to read are valid */

index bffc7d33ef..aa1a81619f 100644
@@ -1208,6 +1208,7 @@ static uint64_t qemu_rdma_make_wrid(uint64_t wr_id, uint64_t index,
 static void rdma_nbsent_inc(RDMAContext *rdma, uint64_t nb_sent);
 static void rdma_nbsent_full(RDMAContext *rdma);
 static void rdma_nbsent_zero(RDMAContext *rdma);
+static void rdma_reg_batch_record(RDMASendRing *r, unsigned count);
 
 static void rdma_consumer_start(RDMAContext *rdma);
 static void rdma_consumer_stop(RDMAContext *rdma);

@@ -1806,149 +1807,41 @@ static int qemu_rdma_exchange_recv(RDMAContext *rdma, RDMAControlHeader *head,
     return 0;
 }
 
-/*
- * Write an actual chunk of memory using RDMA.
- *
- * If we're using dynamic registration on the dest-side, we have to
- * send a registration command first.
- */
-static int qemu_rdma_write_one(RDMAContext *rdma,
-                               int current_index, uint64_t current_addr,
-                               uint64_t length, Error **errp)
+/* Post one dirty span of an already-registered chunk (both pin modes). */
+static int qemu_rdma_write_one(RDMAContext *rdma, int current_index,
+                               uint64_t current_addr, uint64_t length,
+                               Error **errp)
 {
     struct ibv_sge sge;
     struct ibv_send_wr send_wr = { 0 };
-    int reg_result_idx, ret;
-    uint64_t chunk, chunks;
-    uint64_t chunk_size = migrate_rdma_chunk_size();
+    RDMALocalBlock *block = &rdma->local_ram_blocks.block[current_index];
+    uint64_t chunk;
     uint8_t *chunk_start, *chunk_end;
-    RDMALocalBlock *block = &(rdma->local_ram_blocks.block[current_index]);
-    RDMARegister reg;
-    RDMARegisterResult *reg_result;
-    RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
-    RDMAControlHeader head = { .len = sizeof(RDMARegister),
-                               .type = RDMA_CONTROL_REGISTER_REQUEST,
-                               .repeat = 1,
-                             };
 
     sge.addr = (uintptr_t)(block->local_host_addr +
-                            (current_addr - block->offset));
+                           (current_addr - block->offset));
     sge.length = length;
 
-    chunk = ram_chunk_index(block->local_host_addr,
-                            (uint8_t *)(uintptr_t)sge.addr);
+    chunk = ram_chunk_index(block->local_host_addr, (uint8_t *)sge.addr);
     chunk_start = ram_chunk_start(block, chunk);
-    chunks = length / chunk_size;
-
-    if (chunks && ((length % chunk_size) == 0)) {
-        chunks--;
-    }
-
-    trace_rdma_write_one_top(chunks + 1,
-                                  (chunks + 1) * chunk_size / 1024 / 1024);
+    chunk_end = ram_chunk_end(block, chunk);
 
-    chunk_end = ram_chunk_end(block, chunk + chunks);
-
-    if (!rdma->pin_all) {
+    if (rdma->pin_all) {
+        send_wr.wr.rdma.rkey = block->remote_rkey;
+    } else {
         if (!block->remote_keys[chunk]) {
-            /*
-             * This chunk has not yet been registered, so first check to see
-             * if the entire chunk is zero. If so, tell the other size to
-             * memset() + madvise() the entire chunk without RDMA.
-             */
-
-            if (buffer_is_zero((void *)(uintptr_t)sge.addr, length)) {
-                RDMACompress comp = {
-                                        .offset = current_addr,
-                                        .value = 0,
-                                        .block_idx = current_index,
-                                        .length = length,
-                                    };
-
-                head.len = sizeof(comp);
-                head.type = RDMA_CONTROL_COMPRESS;
-
-                trace_rdma_write_one_zero(chunk, sge.length,
-                                               current_index, current_addr);
-
-                compress_to_network(rdma, &comp);
-                ret = qemu_rdma_exchange_send(rdma, &head,
-                                (uint8_t *) &comp, NULL, NULL, NULL, errp);
-
-                if (ret < 0) {
-                    return -1;
-                }
-
-                /*
-                 * TODO: Here we are sending something, but we are not
-                 * accounting for anything transferred.  The following is wrong:
-                 *
-                 * stat64_add(&mig_stats.rdma_bytes, sge.length);
-                 *
-                 * because we are using some kind of compression.  I
-                 * would think that head.len would be the more similar
-                 * thing to a correct value.
-                 */
-                qatomic_add(&mig_stats.zero_pages,
-                            sge.length / qemu_target_page_size());
-                return 1;
-            }
-
-            /*
-             * Otherwise, tell other side to register.
-             */
-            reg.current_index = current_index;
-            reg.current_addr = current_addr;
-            reg.chunks = chunks;
-
-            trace_rdma_write_one_sendreg(chunk, sge.length, current_index,
-                                              current_addr);
-
-            register_to_network(rdma, &reg);
-            ret = qemu_rdma_exchange_send(rdma, &head, (uint8_t *) &reg,
-                                    &resp, &reg_result_idx, NULL, errp);
-            if (ret < 0) {
-                return -1;
-            }
-
-            /* try to overlap this single registration with the one we sent. */
-            if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
-                                                &sge.lkey, NULL, chunk,
-                                                chunk_start, chunk_end)) {
-                error_setg(errp, "cannot get lkey");
-                return -1;
-            }
-
-            reg_result = (RDMARegisterResult *)
-                    rdma->wr_data[reg_result_idx].control_curr;
-
-            network_to_result(reg_result);
-
-            trace_rdma_write_one_recvregres(block->remote_keys[chunk],
-                                                 reg_result->rkey, chunk);
-
-            block->remote_keys[chunk] = reg_result->rkey;
-            block->remote_host_addr = reg_result->host_addr;
-        } else {
-            /* already registered before */
-            if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
-                                                &sge.lkey, NULL, chunk,
-                                                chunk_start, chunk_end)) {
-                error_setg(errp, "cannot get lkey!");
-                return -1;
-            }
+            error_setg(errp, "rdma: write_one on unregistered chunk %" PRIu64,
+                       chunk);
+            return -1;
         }
-
         send_wr.wr.rdma.rkey = block->remote_keys[chunk];
-    } else {
-        send_wr.wr.rdma.rkey = block->remote_rkey;
+    }
 
-        if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr,
-                                                     &sge.lkey, NULL, chunk,
-                                                     chunk_start, chunk_end)) {
-            error_setg(errp, "cannot get lkey!");
-            return -1;
-        }
+    /* Local MR (idempotent): pins our own memory so the NIC can read it. */
+    if (qemu_rdma_register_and_get_keys(rdma, block, sge.addr, &sge.lkey,
+                                        NULL, chunk, chunk_start, chunk_end)) {
+        error_setg(errp, "cannot get lkey");
+        return -1;
     }
 
     /*

@@ -1956,18 +1849,17 @@ static int qemu_rdma_write_one(RDMAContext *rdma,
      * We will use this information at the time of completion
      * to identify the completed write in trace output.
      */
-    send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE,
-                                        current_index, chunk);
-
+    send_wr.wr_id = qemu_rdma_make_wrid(RDMA_WRID_RDMA_WRITE, current_index,
+                                        chunk);
     send_wr.opcode = IBV_WR_RDMA_WRITE;
     send_wr.send_flags = IBV_SEND_SIGNALED;
     send_wr.sg_list = &sge;
     send_wr.num_sge = 1;
     send_wr.wr.rdma.remote_addr = block->remote_host_addr +
-                                (current_addr - block->offset);
+                                  (current_addr - block->offset);
 
     trace_rdma_write_one_post(chunk, sge.addr, send_wr.wr.rdma.remote_addr,
-                                   sge.length);
+                              sge.length);
 
     if (!qemu_rdma_post_send(rdma, &send_wr, errp)) {
         return -1;

@@ -1986,27 +1878,86 @@ static int qemu_rdma_write_one(RDMAContext *rdma,
     qatomic_add(&mig_stats.rdma_bytes, sge.length);
     ram_transferred_add(sge.length);
     rdma->total_writes++;
+    rdma->nb_sent++;
+    trace_rdma_write_flush(rdma->nb_sent);
+    rdma_nbsent_inc(rdma, rdma->nb_sent);
+    return 0;
+}
 
+/* Zero an all-zero span on the destination without RDMA (pin_all=off). */
+static int qemu_rdma_write_compress(RDMAContext *rdma, int current_index,
+                                    uint64_t current_addr, uint64_t length,
+                                    Error **errp)
+{
+    RDMALocalBlock *block = &rdma->local_ram_blocks.block[current_index];
+    uint64_t chunk = (current_addr - block->offset) / migrate_rdma_chunk_size();
+    RDMACompress comp = {
+        .offset = current_addr,
+        .value = 0,
+        .block_idx = current_index,
+        .length = length,
+    };
+    RDMAControlHeader head = { .len = sizeof(comp),
+                               .type = RDMA_CONTROL_COMPRESS,
+                               .repeat = 1 };
+
+    trace_rdma_write_one_zero(chunk, length, current_index, current_addr);
+
+    compress_to_network(rdma, &comp);
+    if (qemu_rdma_exchange_send(rdma, &head, (uint8_t *)&comp,
+                                NULL, NULL, NULL, errp) < 0) {
+        return -1;
+    }
+    qatomic_add(&mig_stats.zero_pages, length / qemu_target_page_size());
     return 0;
 }
 
-/* Consumer: write one dirty span (offset/length are block-relative bytes) */
-static int qemu_rdma_write_chunk(RDMAContext *rdma, uint32_t block_index,
-                                 uint64_t offset, uint64_t length, Error **errp)
+/*
+ * Register a batch of never-seen non-zero chunks in one REGISTER_REQUEST
+ * (pin_all=off); regs[] carry current_index/current_addr in host order.  One
+ * control round-trip, then cache each chunk's remote rkey/host_addr so the
+ * following write_one() calls can post.
+ */
+static int qemu_rdma_write_register(RDMAContext *rdma, RDMARegister *regs,
+                                    unsigned count, Error **errp)
 {
-    RDMALocalBlock *block = &rdma->local_ram_blocks.block[block_index];
-    int ret;
+    RDMARegister net[RDMA_CONSUMER_BATCH_MAX];
+    RDMARegisterResult *results;
+    RDMAControlHeader resp = { .type = RDMA_CONTROL_REGISTER_RESULT };
+    RDMAControlHeader head = { .type = RDMA_CONTROL_REGISTER_REQUEST };
+    int reg_result_idx;
+    unsigned i;
 
-    /* write_one wants current_addr in ram_addr_t space (block->offset based) */
-    ret = qemu_rdma_write_one(rdma, block_index, block->offset + offset,
-                              length, errp);
-    if (ret < 0) {
+    assert(count >= 1 && count <= RDMA_CONSUMER_BATCH_MAX);
+
+    /* network-order copy to send; regs[] stays host-order for caching below */
+    for (i = 0; i < count; i++) {
+        net[i] = regs[i];
+        net[i].chunks = 0;              /* exactly one chunk per entry */
+        register_to_network(rdma, &net[i]);
+    }
+    head.len = count * sizeof(RDMARegister);
+    head.repeat = count;
+
+    rdma_reg_batch_record(rdma->send_ring, count);
+
+    if (qemu_rdma_exchange_send(rdma, &head, (uint8_t *)net,
+                                &resp, &reg_result_idx, NULL, errp) < 0) {
         return -1;
     }
-    if (ret == 0) {
-        rdma->nb_sent++;
-        trace_rdma_write_flush(rdma->nb_sent);
-        rdma_nbsent_inc(rdma, rdma->nb_sent);
+
+    results = (RDMARegisterResult *)rdma->wr_data[reg_result_idx].control_curr;
+    for (i = 0; i < count; i++) {
+        RDMALocalBlock *block =
+            &rdma->local_ram_blocks.block[regs[i].current_index];
+        uint64_t chunk =
+            (regs[i].current_addr - block->offset) / migrate_rdma_chunk_size();
+
+        network_to_result(&results[i]);
+        trace_rdma_write_one_recvregres(block->remote_keys[chunk],
+                                        results[i].rkey, chunk);
+        block->remote_keys[chunk] = results[i].rkey;
+        block->remote_host_addr = results[i].host_addr;
     }
     return 0;
 }

@@ -2679,6 +2630,7 @@ struct RDMASendRing {
     uint64_t pushwait_total_ns;  /* total producer wait on a full ring */
     uint64_t occupancy_max;      /* peak queued commands */
     uint64_t idle_max_ns;        /* longest consumer wait on an empty ring */
+    uint64_t reg_batch_max;      /* largest batched REGISTER_REQUEST */
 };
 
 static bool ring_full(RDMASendRing *r)

@@ -2942,6 +2894,105 @@ static int rdma_consumer_reg_stop(RDMAContext *rdma, uint64_t flags,
     return qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL, errp);
 }
 
+/*
+ * Dirty-span parameters for a ring WRITE command: block, block-relative
+ * offset, migration address and length.  The span is the actually-dirtied
+ * byte range (within one chunk), not the whole chunk.
+ */
+static void rdma_get_chunk_args(RDMAContext *rdma, const RDMACmd *cmd,
+                                RDMALocalBlock **block, uint64_t *off,
+                                uint64_t *addr, uint64_t *len)
+{
+    RDMALocalBlock *b = &rdma->local_ram_blocks.block[cmd->block_index];
+
+    *block = b;
+    *off = cmd->offset;
+    *addr = b->offset + cmd->offset;
+    *len = cmd->length;
+}
+
+/*
+ * Consumer: process the maximal run of consecutive RDMA_CMD_WRITE commands at
+ * ring position 'pos' (at most 'max'), stopping at the first barrier.  Each
+ * command carries one dirty span.  Post ready (already registered / pin_all)
+ * and zero spans as they are met; defer spans whose chunk needs registering,
+ * register those chunks in one batched REGISTER_REQUEST, then post the spans.
+ * Returns the WRITE commands consumed, or -1.
+ */
+static int qemu_rdma_write_run(RDMAContext *rdma, RDMASendRing *r,
+                               unsigned pos, unsigned max, Error **errp)
+{
+    unsigned reg_ring_idx[RDMA_CONSUMER_BATCH_MAX]; /* deferred span writes */
+    RDMARegister regs[RDMA_CONSUMER_BATCH_MAX];     /* distinct chunks to reg */
+    unsigned n = 0, regn = 0, nreg = 0, i;
+    int last_bidx = -1;             /* last chunk queued into regs[] */
+    int64_t last_cidx = -1;
+
+    /* Phase 1: scan the run; post ready/zero now, defer never-registered. */
+    while (n < max) {
+        RDMACmd *cmd = &r->ring[(pos + n) & RDMA_SEND_RING_MASK];
+        RDMALocalBlock *block;
+        uint64_t off, addr, len;
+
+        if (cmd->type != RDMA_CMD_WRITE) {
+            break;
+        }
+        rdma_get_chunk_args(rdma, cmd, &block, &off, &addr, &len);
+
+        if (rdma->pin_all || block->remote_keys[cmd->chunk_index]) {
+            if (qemu_rdma_write_one(rdma, cmd->block_index, addr, len,
+                                    errp) < 0) {
+                return -1;
+            }
+        } else if (buffer_is_zero((void *)(uintptr_t)
+                       (block->local_host_addr + off), len)) {
+            if (qemu_rdma_write_compress(rdma, cmd->block_index, addr, len,
+                                         errp) < 0) {
+                return -1;
+            }
+        } else {
+            /* Defer: record every span for a post-registration write. */
+            reg_ring_idx[regn++] = pos + n;
+            /*
+             * Registration is per whole chunk.  Chunk indices arrive
+             * non-decreasing within a run, so a repeat chunk is the last one
+             * queued; add each chunk to the REGISTER batch only once.
+             */
+            if ((int)cmd->block_index != last_bidx ||
+                (int64_t)cmd->chunk_index != last_cidx) {
+                regs[nreg].current_index = cmd->block_index;
+                regs[nreg].current_addr = addr;
+                nreg++;
+                last_bidx = cmd->block_index;
+                last_cidx = cmd->chunk_index;
+            }
+        }
+        n++;
+    }
+
+    if (regn == 0) {
+        return (int)n;
+    }
+
+    /* Phase 2: one REGISTER_REQUEST for the deduped deferred chunks. */
+    if (qemu_rdma_write_register(rdma, regs, nreg, errp) < 0) {
+        return -1;
+    }
+
+    /* Phase 3: post every deferred span (chunks are now registered). */
+    for (i = 0; i < regn; i++) {
+        RDMACmd *cmd = &r->ring[reg_ring_idx[i] & RDMA_SEND_RING_MASK];
+        RDMALocalBlock *block;
+        uint64_t off, addr, len;
+
+        rdma_get_chunk_args(rdma, cmd, &block, &off, &addr, &len);
+        if (qemu_rdma_write_one(rdma, cmd->block_index, addr, len, errp) < 0) {
+            return -1;
+        }
+    }
+    return (int)n;
+}
+
 static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
                                   bool *stop)
 {

@@ -2953,8 +3004,8 @@ static int rdma_consumer_dispatch(RDMAContext *rdma, const RDMACmd *cmd,
 
     switch (cmd->type) {
     case RDMA_CMD_WRITE:
-        ret = qemu_rdma_write_chunk(rdma, cmd->block_index,
-                                    cmd->offset, cmd->length, &err);
+        /* WRITE runs are handled in bulk by qemu_rdma_write_run(). */
+        g_assert_not_reached();
         break;
     case RDMA_CMD_FLUSH:
         ret = qemu_rdma_drain_cq(rdma, &err);

@@ -2981,6 +3032,7 @@ static void *rdma_consumer_thread(void *opaque)
 {
     RDMAContext *rdma = opaque;
     RDMASendRing *r = rdma->send_ring;
+    Error *err = NULL;
     bool stop = false;
 
     rcu_register_thread();

@@ -3025,6 +3077,21 @@ static void *rdma_consumer_thread(void *opaque)
             unsigned idx = r->tail + processed;
             RDMACmd cmd = r->ring[idx & RDMA_SEND_RING_MASK];
 
+            /* WRITE runs are drained in bulk with batched registration. */
+            if (cmd.type == RDMA_CMD_WRITE) {
+                int n = qemu_rdma_write_run(rdma, r, idx,
+                                            batch - processed, &err);
+                if (n < 0) {
+                    if (err) {
+                        error_report_err(err);
+                    }
+                    ring_set_abort(rdma);
+                    break;
+                }
+                processed += (unsigned)n;
+                continue;
+            }
+
             if (rdma_consumer_dispatch(rdma, &cmd, &stop) < 0) {
                 ring_set_abort(rdma);
                 break;

@@ -3447,6 +3514,13 @@ static void rdma_nbsent_zero(RDMAContext *rdma)
     rdma->nbsent_zero_hits++;
 }
 
+static void rdma_reg_batch_record(RDMASendRing *r, unsigned count)
+{
+    if (count > r->reg_batch_max) {
+        r->reg_batch_max = count;
+    }
+}
+
 static void rdma_stats_reset(RDMAContext *rdma)
 {
     rdma->nbsent_max = 0;

index aa1a81619f..863165de26 100644
@@ -1940,6 +1940,7 @@ static int qemu_rdma_write_register(RDMAContext *rdma, RDMARegister *regs,
     head.repeat = count;
 
     rdma_reg_batch_record(rdma->send_ring, count);
+    trace_rdma_write_register_batch(count);
 
     if (qemu_rdma_exchange_send(rdma, &head, (uint8_t *)net,
                                 &resp, &reg_result_idx, NULL, errp) < 0) {

@@ -2970,6 +2971,7 @@ static int qemu_rdma_write_run(RDMAContext *rdma, RDMASendRing *r,
         n++;
     }
 
+    trace_rdma_write_run(n);
     if (regn == 0) {
         return (int)n;
     }

@@ -3072,6 +3074,7 @@ static void *rdma_consumer_thread(void *opaque)
          */
         batch = avail < RDMA_CONSUMER_BATCH_MAX ? avail
                                                 : RDMA_CONSUMER_BATCH_MAX;
+        trace_rdma_consumer_batch(batch, avail);
 
         while (processed < batch) {
             unsigned idx = r->tail + processed;

@@ -3542,7 +3545,8 @@ static void rdma_nbsent_stats_dump(RDMAContext *rdma)
 static void rdma_prodcons_stats_dump(RDMASendRing *r)
 {
     trace_rdma_prodcons_stats(r->idle_max_ns, r->pushwait_max_ns,
-                              r->pushwait_total_ns, r->occupancy_max);
+                              r->pushwait_total_ns, r->occupancy_max,
+                              r->reg_batch_max);
 }
 
 int rdma_control_save_page(QEMUFile *f, ram_addr_t block_offset,

index 863165de26..29a2480448 100644
@@ -75,12 +75,15 @@ static inline uint64_t rdma_merge_max(void)
  * Capabilities for negotiation.
  */
 #define RDMA_CAPABILITY_PIN_ALL 0x01
+/* Peer replies to a batched REGISTER_REQUEST with one result per chunk. */
+#define RDMA_CAPABILITY_REG_BATCH 0x02
 
 /*
  * Add the other flags above to this list of known capabilities
  * as they are introduced.
  */
-static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL;
+static uint32_t known_capabilities = RDMA_CAPABILITY_PIN_ALL |
+                                     RDMA_CAPABILITY_REG_BATCH;
 
 /*
  * A work request ID is 64-bits and we split up these bits

@@ -352,6 +355,7 @@ typedef struct RDMAContext {
     uint64_t pend_length;         /* span length in bytes (within one chunk) */
 
     bool pin_all;
+    bool reg_batch;   /* peer negotiated RDMA_CAPABILITY_REG_BATCH */
 
     /*
      * infiniband-specific variables for opening the device

@@ -1947,6 +1951,14 @@ static int qemu_rdma_write_register(RDMAContext *rdma, RDMARegister *regs,
         return -1;
     }
 
+    /* resp.repeat is echoed back unchanged by old peers; resp.len is not. */
+    if (resp.len != count * sizeof(RDMARegisterResult)) {
+        error_setg(errp, "rdma: REGISTER_RESULT size mismatch: "
+                   "requested %u chunks (%zu bytes), peer returned %u bytes",
+                   count, count * sizeof(RDMARegisterResult), resp.len);
+        return -1;
+    }
+
     results = (RDMARegisterResult *)rdma->wr_data[reg_result_idx].control_curr;
     for (i = 0; i < count; i++) {
         RDMALocalBlock *block =

@@ -2172,6 +2184,8 @@ static int qemu_rdma_connect(RDMAContext *rdma, bool return_path,
         cap.flags |= RDMA_CAPABILITY_PIN_ALL;
     }
 
+    cap.flags |= RDMA_CAPABILITY_REG_BATCH;
+
     caps_to_network(&cap);
 
     ret = qemu_rdma_post_recv_control(rdma, RDMA_WRID_READY, errp);

@@ -2219,6 +2233,13 @@ static int qemu_rdma_connect(RDMAContext *rdma, bool return_path,
         rdma->pin_all = false;
     }
 
+    rdma->reg_batch = !!(cap.flags & RDMA_CAPABILITY_REG_BATCH);
+    if (!rdma->reg_batch) {
+        warn_report("RDMA: Server does not support batched chunk "
+                    "registration. Will register one chunk per request.");
+    }
+    trace_rdma_connect_reg_batch_outcome(rdma->reg_batch);
+
     trace_rdma_connect_pin_all_outcome(rdma->pin_all);
 
     rdma_ack_cm_event(cm_event);

@@ -2952,6 +2973,14 @@ static int qemu_rdma_write_run(RDMAContext *rdma, RDMASendRing *r,
                 return -1;
             }
         } else {
+            bool new_chunk = (int)cmd->block_index != last_bidx ||
+                             (int64_t)cmd->chunk_index != last_cidx;
+
+            /* Cap at one never-seen chunk per request without REG_BATCH. */
+            if (new_chunk && nreg == 1 && !rdma->reg_batch) {
+                break;
+            }
+
             /* Defer: record every span for a post-registration write. */
             reg_ring_idx[regn++] = pos + n;
             /*

@@ -2959,8 +2988,7 @@ static int qemu_rdma_write_run(RDMAContext *rdma, RDMASendRing *r,
              * non-decreasing within a run, so a repeat chunk is the last one
              * queued; add each chunk to the REGISTER batch only once.
              */
-            if ((int)cmd->block_index != last_bidx ||
-                (int64_t)cmd->chunk_index != last_cidx) {
+            if (new_chunk) {
                 regs[nreg].current_index = cmd->block_index;
                 regs[nreg].current_addr = addr;
                 nreg++;

@@ -3663,6 +3691,9 @@ static int qemu_rdma_accept(RDMAContext *rdma)
     if (cap.flags & RDMA_CAPABILITY_PIN_ALL) {
         rdma->pin_all = true;
     }
+    if (cap.flags & RDMA_CAPABILITY_REG_BATCH) {
+        rdma->reg_batch = true;
+    }
 
     rdma->cm_id = cm_event->id;
     verbs = cm_event->id->verbs;