[PATCH RFC 10/12] migration/rdma: use multiRDMA to send RAM block for rdma-pin-all mode

[Zhimin Feng]

From: fengzhimin <address@hidden>

Send the RAM block through MultiRDMA channels for using rdma-pin-all option,
and we choose the channel to send data through polling the MultiRDMA thread.

Signed-off-by: fengzhimin <address@hidden>
---
 migration/rdma.c | 66 +++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 63 insertions(+), 3 deletions(-)

diff --git a/migration/rdma.c b/migration/rdma.c
index 425dfa709d..36261f1fc8 100644
--- a/migration/rdma.c
+++ b/migration/rdma.c
@@ -427,6 +427,8 @@ typedef struct {
     QEMUFile *file;
     /* sem where to wait for more work */
     QemuSemaphore sem;
+    /* syncs main thread and channels */
+    QemuSemaphore sem_sync;
     /* this mutex protects the following parameters */
     QemuMutex mutex;
     /* is this channel thread running */
@@ -439,6 +441,8 @@ struct {
     MultiRDMASendParams *params;
     /* number of created threads */
     int count;
+    /* index of current RDMA channels */
+    int current_RDMA_index;
     /* this mutex protects the following parameters */
     QemuMutex mutex_sync;
     /* number of registered multiRDMA channels */
@@ -2043,6 +2047,18 @@ static int qemu_rdma_exchange_recv(RDMAContext *rdma, 
RDMAControlHeader *head,
     return 0;
 }
 
+/*
+ * Get the multiRDMA channel used to send data.
+ */
+static int get_multiRDMA_channel(void)
+{
+    int thread_count = migrate_multiRDMA_channels();
+    multiRDMA_send_state->current_RDMA_index++;
+    multiRDMA_send_state->current_RDMA_index %= thread_count;
+
+    return multiRDMA_send_state->current_RDMA_index;
+}
+
 /*
  * Write an actual chunk of memory using RDMA.
  *
@@ -2068,6 +2084,16 @@ static int qemu_rdma_write_one(QEMUFile *f, RDMAContext 
*rdma,
                                .repeat = 1,
                              };
 
+    if (migrate_use_multiRDMA() &&
+        migrate_use_rdma_pin_all()) {
+        /* The multiRDMA threads only send ram block */
+        if (strcmp(block->block_name, "mach-virt.ram") == 0) {
+            int channel = get_multiRDMA_channel();
+            rdma = multiRDMA_send_state->params[channel].rdma;
+            block = &(rdma->local_ram_blocks.block[current_index]);
+        }
+    }
+
 retry:
     sge.addr = (uintptr_t)(block->local_host_addr +
                             (current_addr - block->offset));
@@ -2285,8 +2311,22 @@ static int qemu_rdma_write_flush(QEMUFile *f, 
RDMAContext *rdma)
     }
 
     if (ret == 0) {
-        rdma->nb_sent++;
-        trace_qemu_rdma_write_flush(rdma->nb_sent);
+        if (migrate_use_multiRDMA() &&
+            migrate_use_rdma_pin_all()) {
+            /* The multiRDMA threads only send ram block */
+            RDMALocalBlock *block = NULL;
+            block = &(rdma->local_ram_blocks.block[rdma->current_index]);
+            if (strcmp(block->block_name, "mach-virt.ram") == 0) {
+                int current_RDMA = multiRDMA_send_state->current_RDMA_index;
+                multiRDMA_send_state->params[current_RDMA].rdma->nb_sent++;
+            } else {
+                rdma->nb_sent++;
+                trace_qemu_rdma_write_flush(rdma->nb_sent);
+            }
+        } else {
+            rdma->nb_sent++;
+            trace_qemu_rdma_write_flush(rdma->nb_sent);
+        }
     }
 
     rdma->current_length = 0;
@@ -4015,11 +4055,15 @@ static int qemu_rdma_registration_stop(QEMUFile *f, 
void *opaque,
     ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
 
     if (migrate_use_multiRDMA()) {
-        /* Inform src send_thread to send FINISHED signal */
+        /*
+         * Inform src send_thread to send FINISHED signal.
+         * Wait for multiRDMA send threads to poll the CQE.
+         */
         int i;
         int thread_count = migrate_multiRDMA_channels();
         for (i = 0; i < thread_count; i++) {
             qemu_sem_post(&multiRDMA_send_state->params[i].sem);
+            qemu_sem_wait(&multiRDMA_send_state->params[i].sem_sync);
         }
     }
 
@@ -4592,12 +4636,25 @@ static void *multiRDMA_send_thread(void *opaque)
         }
         qemu_mutex_unlock(&p->mutex);
 
+        /* To complete polling(CQE) */
+        while (rdma->nb_sent) {
+            ret = qemu_rdma_block_for_wrid(rdma, RDMA_WRID_RDMA_WRITE, NULL);
+            if (ret < 0) {
+                error_report("multiRDMA migration: "
+                             "complete polling error!");
+                return NULL;
+            }
+        }
+
         /* Send FINISHED to the destination */
         head.type = RDMA_CONTROL_REGISTER_FINISHED;
         ret = qemu_rdma_exchange_send(rdma, &head, NULL, NULL, NULL, NULL);
         if (ret < 0) {
             return NULL;
         }
+
+        /* sync main thread */
+        qemu_sem_post(&p->sem_sync);
     }
 
     qemu_mutex_lock(&p->mutex);
@@ -4637,6 +4694,7 @@ static int multiRDMA_save_setup(const char *host_port, 
Error **errp)
                                           thread_count);
     atomic_set(&multiRDMA_send_state->count, 0);
     atomic_set(&multiRDMA_send_state->reg_mr_channels, 0);
+    atomic_set(&multiRDMA_send_state->current_RDMA_index, 0);
     qemu_mutex_init(&multiRDMA_send_state->mutex_sync);
     qemu_sem_init(&multiRDMA_send_state->sem_sync, 0);
 
@@ -4665,6 +4723,7 @@ static int multiRDMA_save_setup(const char *host_port, 
Error **errp)
         f = qemu_fopen_rdma(multiRDMA_send_state->params[i].rdma, "wb");
         qemu_mutex_init(&p->mutex);
         qemu_sem_init(&p->sem, 0);
+        qemu_sem_init(&p->sem_sync, 0);
         p->quit = false;
         p->id = i;
         p->running = true;
@@ -4730,6 +4789,7 @@ int multiRDMA_save_cleanup(void)
         MultiRDMASendParams *p = &multiRDMA_send_state->params[i];
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
+        qemu_sem_destroy(&p->sem_sync);
         g_free(p->name);
         p->name = NULL;
         qemu_rdma_cleanup(multiRDMA_send_state->params[i].rdma);
-- 
2.19.1