[PATCH v5 09/13] migration/multifd: Prepare to introduce DSA acceleration on the multifd path.

[Yichen Wang]

From: Hao Xiang <hao.xiang@linux.dev>

1. Refactor multifd_send_thread function.
2. Introduce the batch task structure in MultiFDSendParams.

Signed-off-by: Hao Xiang <hao.xiang@linux.dev>
Signed-off-by: Yichen Wang <yichen.wang@bytedance.com>
---
 include/qemu/dsa.h  |  41 ++++---
 migration/multifd.c |   4 +
 migration/multifd.h |   3 +
 util/dsa.c          | 270 +++++++++++++++++++++++---------------------
 4 files changed, 172 insertions(+), 146 deletions(-)

diff --git a/include/qemu/dsa.h b/include/qemu/dsa.h
index 20bb88d48c..fd0305a7c7 100644
--- a/include/qemu/dsa.h
+++ b/include/qemu/dsa.h
@@ -16,6 +16,7 @@
 #define QEMU_DSA_H
 
 #include "qemu/error-report.h"
+#include "exec/cpu-common.h"
 #include "qemu/thread.h"
 #include "qemu/queue.h"
 
@@ -70,10 +71,11 @@ typedef struct QemuDsaBatchTask {
     QemuDsaTaskStatus status;
     int batch_size;
     bool *results;
+    /* Address of each pages in pages */
+    ram_addr_t *addr;
     QSIMPLEQ_ENTRY(QemuDsaBatchTask) entry;
 } QemuDsaBatchTask;
 
-
 /**
  * @brief Initializes DSA devices.
  *
@@ -106,15 +108,13 @@ void qemu_dsa_cleanup(void);
 bool qemu_dsa_is_running(void);
 
 /**
- * @brief Initializes a buffer zero batch task.
+ * @brief Initializes a buffer zero DSA batch task.
  *
- * @param task A pointer to the batch task to initialize.
- * @param results A pointer to an array of zero page checking results.
- * @param batch_size The number of DSA tasks in the batch.
+ * @param batch_size The number of zero page checking tasks in the batch.
+ * @return A pointer to the zero page checking tasks initialized.
  */
-void
-buffer_zero_batch_task_init(QemuDsaBatchTask *task,
-                            bool *results, int batch_size);
+QemuDsaBatchTask *
+buffer_zero_batch_task_init(int batch_size);
 
 /**
  * @brief Performs the proper cleanup on a DSA batch task.
@@ -139,6 +139,8 @@ buffer_is_zero_dsa_batch_sync(QemuDsaBatchTask *batch_task,
 
 #else
 
+typedef struct QemuDsaBatchTask {} QemuDsaBatchTask;
+
 static inline bool qemu_dsa_is_running(void)
 {
     return false;
@@ -146,19 +148,28 @@ static inline bool qemu_dsa_is_running(void)
 
 static inline int qemu_dsa_init(const char *dsa_parameter, Error **errp)
 {
-    if (dsa_parameter != NULL && strlen(dsa_parameter) != 0) {
-        error_setg(errp, "DSA is not supported.");
-        return -1;
-    }
-
-    return 0;
+    error_setg(errp, "DSA accelerator is not enabled.");
+    return -1;
 }
 
 static inline void qemu_dsa_start(void) {}
 
 static inline void qemu_dsa_stop(void) {}
 
-static inline void qemu_dsa_cleanup(void) {}
+static inline QemuDsaBatchTask *buffer_zero_batch_task_init(int batch_size)
+{
+    return NULL;
+}
+
+static inline void buffer_zero_batch_task_destroy(QemuDsaBatchTask *task) {}
+
+static inline int
+buffer_is_zero_dsa_batch_sync(QemuDsaBatchTask *batch_task,
+                              const void **buf, size_t count, size_t len)
+{
+    error_setg(errp, "DSA accelerator is not enabled.");
+    return -1;
+}
 
 #endif
 
diff --git a/migration/multifd.c b/migration/multifd.c
index 0b4cbaddfe..6f8edd4b6a 100644
--- a/migration/multifd.c
+++ b/migration/multifd.c
@@ -13,6 +13,7 @@
 #include "qemu/osdep.h"
 #include "qemu/cutils.h"
 #include "qemu/rcu.h"
+#include "qemu/dsa.h"
 #include "exec/target_page.h"
 #include "sysemu/sysemu.h"
 #include "exec/ramblock.h"
@@ -792,6 +793,8 @@ static bool multifd_send_cleanup_channel(MultiFDSendParams 
*p, Error **errp)
     p->name = NULL;
     multifd_pages_clear(p->pages);
     p->pages = NULL;
+    buffer_zero_batch_task_destroy(p->dsa_batch_task);
+    p->dsa_batch_task = NULL;
     p->packet_len = 0;
     g_free(p->packet);
     p->packet = NULL;
@@ -1182,6 +1185,7 @@ bool multifd_send_setup(void)
         qemu_sem_init(&p->sem_sync, 0);
         p->id = i;
         p->pages = multifd_pages_init(page_count);
+        p->dsa_batch_task = buffer_zero_batch_task_init(page_count);
 
         if (use_packets) {
             p->packet_len = sizeof(MultiFDPacket_t)
diff --git a/migration/multifd.h b/migration/multifd.h
index 0ecd6f47d7..027f57bf4e 100644
--- a/migration/multifd.h
+++ b/migration/multifd.h
@@ -14,6 +14,7 @@
 #define QEMU_MIGRATION_MULTIFD_H
 
 #include "ram.h"
+#include "qemu/dsa.h"
 
 typedef struct MultiFDRecvData MultiFDRecvData;
 
@@ -137,6 +138,8 @@ typedef struct {
      * pending_job != 0 -> multifd_channel can use it.
      */
     MultiFDPages_t *pages;
+    /* Zero page checking batch task */
+    QemuDsaBatchTask *dsa_batch_task;
 
     /* thread local variables. No locking required */
 
diff --git a/util/dsa.c b/util/dsa.c
index 74b9aa1331..5aba1ae23a 100644
--- a/util/dsa.c
+++ b/util/dsa.c
@@ -696,93 +696,81 @@ static void dsa_completion_thread_stop(void *opaque)
 }
 
 /**
- * @brief Initializes a buffer zero comparison DSA task.
+ * @brief Check if DSA is running.
  *
- * @param descriptor A pointer to the DSA task descriptor.
- * @param completion A pointer to the DSA task completion record.
+ * @return True if DSA is running, otherwise false.
  */
+bool qemu_dsa_is_running(void)
+{
+    return completion_thread.running;
+}
+
 static void
-buffer_zero_task_init_int(struct dsa_hw_desc *descriptor,
-                          struct dsa_completion_record *completion)
+dsa_globals_init(void)
 {
-    descriptor->opcode = DSA_OPCODE_COMPVAL;
-    descriptor->flags = IDXD_OP_FLAG_RCR | IDXD_OP_FLAG_CRAV;
-    descriptor->comp_pattern = (uint64_t)0;
-    descriptor->completion_addr = (uint64_t)completion;
+    max_retry_count = DSA_WQ_DEPTH;
 }
 
 /**
- * @brief Initializes a buffer zero batch task.
+ * @brief Initializes DSA devices.
+ *
+ * @param dsa_parameter A list of DSA device path from migration parameter.
  *
- * @param task A pointer to the batch task to initialize.
- * @param results A pointer to an array of zero page checking results.
- * @param batch_size The number of DSA tasks in the batch.
+ * @return int Zero if successful, otherwise non zero.
  */
-void
-buffer_zero_batch_task_init(QemuDsaBatchTask *task,
-                            bool *results, int batch_size)
+int qemu_dsa_init(const char *dsa_parameter, Error **errp)
 {
-    int descriptors_size = sizeof(*task->descriptors) * batch_size;
-    memset(task, 0, sizeof(*task));
-
-    task->descriptors =
-        (struct dsa_hw_desc *)qemu_memalign(64, descriptors_size);
-    memset(task->descriptors, 0, descriptors_size);
-    task->completions = (struct dsa_completion_record *)qemu_memalign(
-        32, sizeof(*task->completions) * batch_size);
-    task->results = results;
-    task->batch_size = batch_size;
+    dsa_globals_init();
 
-    task->batch_completion.status = DSA_COMP_NONE;
-    task->batch_descriptor.completion_addr = (uint64_t)&task->batch_completion;
-    /* TODO: Ensure that we never send a batch with count <= 1 */
-    task->batch_descriptor.desc_count = 0;
-    task->batch_descriptor.opcode = DSA_OPCODE_BATCH;
-    task->batch_descriptor.flags = IDXD_OP_FLAG_RCR | IDXD_OP_FLAG_CRAV;
-    task->batch_descriptor.desc_list_addr = (uintptr_t)task->descriptors;
-    task->status = QEMU_DSA_TASK_READY;
-    task->group = &dsa_group;
-    task->device = dsa_device_group_get_next_device(&dsa_group);
+    return dsa_device_group_init(&dsa_group, dsa_parameter, errp);
+}
 
-    for (int i = 0; i < task->batch_size; i++) {
-        buffer_zero_task_init_int(&task->descriptors[i],
-                                  &task->completions[i]);
+/**
+ * @brief Start logic to enable using DSA.
+ *
+ */
+void qemu_dsa_start(void)
+{
+    if (dsa_group.num_dsa_devices == 0) {
+        return;
     }
-
-    qemu_sem_init(&task->sem_task_complete, 0);
-    task->completion_callback = buffer_zero_dsa_completion;
+    if (dsa_group.running) {
+        return;
+    }
+    dsa_device_group_start(&dsa_group);
+    dsa_completion_thread_init(&completion_thread, &dsa_group);
 }
 
 /**
- * @brief Performs the proper cleanup on a DSA batch task.
+ * @brief Stop the device group and the completion thread.
  *
- * @param task A pointer to the batch task to cleanup.
  */
-void
-buffer_zero_batch_task_destroy(QemuDsaBatchTask *task)
+void qemu_dsa_stop(void)
 {
-    qemu_vfree(task->descriptors);
-    qemu_vfree(task->completions);
-    task->results = NULL;
+    QemuDsaDeviceGroup *group = &dsa_group;
 
-    qemu_sem_destroy(&task->sem_task_complete);
+    if (!group->running) {
+        return;
+    }
+
+    dsa_completion_thread_stop(&completion_thread);
+    dsa_empty_task_queue(group);
 }
 
 /**
- * @brief Resets a buffer zero comparison DSA batch task.
+ * @brief Clean up system resources created for DSA offloading.
  *
- * @param task A pointer to the batch task.
- * @param count The number of DSA tasks this batch task will contain.
  */
-static void
-buffer_zero_batch_task_reset(QemuDsaBatchTask *task, size_t count)
+void qemu_dsa_cleanup(void)
 {
-    task->batch_completion.status = DSA_COMP_NONE;
-    task->batch_descriptor.desc_count = count;
-    task->task_type = QEMU_DSA_BATCH_TASK;
-    task->status = QEMU_DSA_TASK_READY;
+    qemu_dsa_stop();
+    dsa_device_group_cleanup(&dsa_group);
 }
 
+
+/* Buffer zero comparison DSA task implementations */
+/* =============================================== */
+
 /**
  * @brief Sets a buffer zero comparison DSA task.
  *
@@ -817,6 +805,21 @@ buffer_zero_task_reset(QemuDsaBatchTask *task)
     task->status = QEMU_DSA_TASK_READY;
 }
 
+/**
+ * @brief Resets a buffer zero comparison DSA batch task.
+ *
+ * @param task A pointer to the batch task.
+ * @param count The number of DSA tasks this batch task will contain.
+ */
+static void
+buffer_zero_batch_task_reset(QemuDsaBatchTask *task, size_t count)
+{
+    task->batch_completion.status = DSA_COMP_NONE;
+    task->batch_descriptor.desc_count = count;
+    task->task_type = QEMU_DSA_BATCH_TASK;
+    task->status = QEMU_DSA_TASK_READY;
+}
+
 /**
  * @brief Sets a buffer zero comparison DSA task.
  *
@@ -923,6 +926,7 @@ buffer_zero_dsa_wait(QemuDsaBatchTask *batch_task)
  *
  * @param batch_task A pointer to the batch task.
  */
+
 static void
 buffer_zero_cpu_fallback(QemuDsaBatchTask *batch_task)
 {
@@ -966,78 +970,6 @@ buffer_zero_cpu_fallback(QemuDsaBatchTask *batch_task)
     }
 }
 
-/**
- * @brief Check if DSA is running.
- *
- * @return True if DSA is running, otherwise false.
- */
-bool qemu_dsa_is_running(void)
-{
-    return completion_thread.running;
-}
-
-static void
-dsa_globals_init(void)
-{
-    max_retry_count = DSA_WQ_DEPTH;
-}
-
-/**
- * @brief Initializes DSA devices.
- *
- * @param dsa_parameter A list of DSA device path from migration parameter.
- *
- * @return int Zero if successful, otherwise non zero.
- */
-int qemu_dsa_init(const char *dsa_parameter, Error **errp)
-{
-    dsa_globals_init();
-
-    return dsa_device_group_init(&dsa_group, dsa_parameter, errp);
-}
-
-/**
- * @brief Start logic to enable using DSA.
- *
- */
-void qemu_dsa_start(void)
-{
-    if (dsa_group.num_dsa_devices == 0) {
-        return;
-    }
-    if (dsa_group.running) {
-        return;
-    }
-    dsa_device_group_start(&dsa_group);
-    dsa_completion_thread_init(&completion_thread, &dsa_group);
-}
-
-/**
- * @brief Stop the device group and the completion thread.
- *
- */
-void qemu_dsa_stop(void)
-{
-    QemuDsaDeviceGroup *group = &dsa_group;
-
-    if (!group->running) {
-        return;
-    }
-
-    dsa_completion_thread_stop(&completion_thread);
-    dsa_empty_task_queue(group);
-}
-
-/**
- * @brief Clean up system resources created for DSA offloading.
- *
- */
-void qemu_dsa_cleanup(void)
-{
-    qemu_dsa_stop();
-    dsa_device_group_cleanup(&dsa_group);
-}
-
 /**
  * @brief Performs buffer zero comparison on a DSA batch task synchronously.
  *
@@ -1073,3 +1005,79 @@ buffer_is_zero_dsa_batch_sync(QemuDsaBatchTask 
*batch_task,
     return 0;
 }
 
+/**
+ * @brief Initializes a buffer zero comparison DSA task.
+ *
+ * @param descriptor A pointer to the DSA task descriptor.
+ * @param completion A pointer to the DSA task completion record.
+ */
+static void
+buffer_zero_task_init_int(struct dsa_hw_desc *descriptor,
+                          struct dsa_completion_record *completion)
+{
+    descriptor->opcode = DSA_OPCODE_COMPVAL;
+    descriptor->flags = IDXD_OP_FLAG_RCR | IDXD_OP_FLAG_CRAV;
+    descriptor->comp_pattern = (uint64_t)0;
+    descriptor->completion_addr = (uint64_t)completion;
+}
+
+/**
+ * @brief Initializes a buffer zero DSA batch task.
+ *
+ * @param batch_size The number of zero page checking tasks in the batch.
+ * @return A pointer to the zero page checking tasks initialized.
+ */
+QemuDsaBatchTask *
+buffer_zero_batch_task_init(int batch_size)
+{
+    QemuDsaBatchTask *task = qemu_memalign(64, sizeof(QemuDsaBatchTask));
+    int descriptors_size = sizeof(*task->descriptors) * batch_size;
+
+    memset(task, 0, sizeof(*task));
+    task->addr = g_new0(ram_addr_t, batch_size);
+    task->results = g_new0(bool, batch_size);
+    task->batch_size = batch_size;
+    task->descriptors =
+        (struct dsa_hw_desc *)qemu_memalign(64, descriptors_size);
+    memset(task->descriptors, 0, descriptors_size);
+    task->completions = (struct dsa_completion_record *)qemu_memalign(
+        32, sizeof(*task->completions) * batch_size);
+
+    task->batch_completion.status = DSA_COMP_NONE;
+    task->batch_descriptor.completion_addr = (uint64_t)&task->batch_completion;
+    /* TODO: Ensure that we never send a batch with count <= 1 */
+    task->batch_descriptor.desc_count = 0;
+    task->batch_descriptor.opcode = DSA_OPCODE_BATCH;
+    task->batch_descriptor.flags = IDXD_OP_FLAG_RCR | IDXD_OP_FLAG_CRAV;
+    task->batch_descriptor.desc_list_addr = (uintptr_t)task->descriptors;
+    task->status = QEMU_DSA_TASK_READY;
+    task->group = &dsa_group;
+    task->device = dsa_device_group_get_next_device(&dsa_group);
+
+    for (int i = 0; i < task->batch_size; i++) {
+        buffer_zero_task_init_int(&task->descriptors[i],
+                                  &task->completions[i]);
+    }
+
+    qemu_sem_init(&task->sem_task_complete, 0);
+    task->completion_callback = buffer_zero_dsa_completion;
+
+    return task;
+}
+
+/**
+ * @brief Performs the proper cleanup on a DSA batch task.
+ *
+ * @param task A pointer to the batch task to cleanup.
+ */
+void
+buffer_zero_batch_task_destroy(QemuDsaBatchTask *task)
+{
+    g_free(task->addr);
+    g_free(task->results);
+    qemu_vfree(task->descriptors);
+    qemu_vfree(task->completions);
+    task->results = NULL;
+    qemu_sem_destroy(&task->sem_task_complete);
+    qemu_vfree(task);
+}
-- 
Yichen Wang