@@ -13,7 +13,6 @@
#include "channel.h"
#include "file.h"
#include "migration.h"
-#include "multifd.h"
#include "io/channel-file.h"
#include "io/channel-util.h"
#include "options.h"
@@ -204,3 +203,20 @@ int file_write_ramblock_iov(QIOChannel *ioc, const struct iovec *iov,
return (ret < 0) ? ret : 0;
}
+
+int multifd_file_recv_data(MultiFDRecvParams *p, Error **errp)
+{
+ MultiFDRecvData *data = p->data;
+ size_t ret;
+
+ ret = qio_channel_pread(p->c, (char *) data->opaque,
+ data->size, data->file_offset, errp);
+ if (ret != data->size) {
+ error_prepend(errp,
+ "multifd recv (%u): read 0x%zx, expected 0x%zx",
+ p->id, ret, data->size);
+ return -1;
+ }
+
+ return 0;
+}
@@ -11,6 +11,7 @@
#include "qapi/qapi-types-migration.h"
#include "io/task.h"
#include "channel.h"
+#include "multifd.h"
void file_start_incoming_migration(FileMigrationArgs *file_args, Error **errp);
@@ -21,4 +22,5 @@ void file_cleanup_outgoing_migration(void);
bool file_send_channel_create(gpointer opaque, Error **errp);
int file_write_ramblock_iov(QIOChannel *ioc, const struct iovec *iov,
int niov, RAMBlock *block, Error **errp);
+int multifd_file_recv_data(MultiFDRecvParams *p, Error **errp);
#endif
@@ -18,7 +18,6 @@
#include "qemu/error-report.h"
#include "qapi/error.h"
#include "file.h"
-#include "ram.h"
#include "migration.h"
#include "migration-stats.h"
#include "socket.h"
@@ -251,7 +250,7 @@ static int nocomp_recv(MultiFDRecvParams *p, Error **errp)
uint32_t flags;
if (!multifd_use_packets()) {
- return 0;
+ return multifd_file_recv_data(p, errp);
}
flags = p->flags & MULTIFD_FLAG_COMPRESSION_MASK;
@@ -1331,22 +1330,48 @@ void multifd_recv_cleanup(void)
void multifd_recv_sync_main(void)
{
int thread_count = migrate_multifd_channels();
+ bool file_based = !multifd_use_packets();
int i;
- if (!migrate_multifd() || !multifd_use_packets()) {
+ if (!migrate_multifd()) {
return;
}
+ /*
+ * File-based channels don't use packets and therefore need to
+ * wait for more work. Release them to start the sync.
+ */
+ if (file_based) {
+ for (i = 0; i < thread_count; i++) {
+ MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+ trace_multifd_recv_sync_main_signal(p->id);
+ qemu_sem_post(&p->sem);
+ }
+ }
+
/*
* Initiate the synchronization by waiting for all channels.
+ *
* For socket-based migration this means each channel has received
* the SYNC packet on the stream.
+ *
+ * For file-based migration this means each channel is done with
+ * the work (pending_job=false).
*/
for (i = 0; i < thread_count; i++) {
trace_multifd_recv_sync_main_wait(i);
qemu_sem_wait(&multifd_recv_state->sem_sync);
}
+ if (file_based) {
+ /*
+ * For file-based loading is done in one iteration. We're
+ * done.
+ */
+ return;
+ }
+
/*
* Sync done. Release the channels for the next iteration.
*/
@@ -13,6 +13,8 @@
#ifndef QEMU_MIGRATION_MULTIFD_H
#define QEMU_MIGRATION_MULTIFD_H
+#include "ram.h"
+
typedef struct MultiFDRecvData MultiFDRecvData;
bool multifd_send_setup(void);
@@ -3952,6 +3952,22 @@ void colo_flush_ram_cache(void)
trace_colo_flush_ram_cache_end();
}
+static size_t ram_load_multifd_pages(void *host_addr, size_t size,
+ uint64_t offset)
+{
+ MultiFDRecvData *data = multifd_get_recv_data();
+
+ data->opaque = host_addr;
+ data->file_offset = offset;
+ data->size = size;
+
+ if (!multifd_recv()) {
+ return 0;
+ }
+
+ return size;
+}
+
static bool read_ramblock_mapped_ram(QEMUFile *f, RAMBlock *block,
long num_pages, unsigned long *bitmap,
Error **errp)
@@ -3981,8 +3997,14 @@ static bool read_ramblock_mapped_ram(QEMUFile *f, RAMBlock *block,
size = MIN(unread, MAPPED_RAM_LOAD_BUF_SIZE);
- read = qemu_get_buffer_at(f, host, size,
- block->pages_offset + offset);
+ if (migrate_multifd()) {
+ read = ram_load_multifd_pages(host, size,
+ block->pages_offset + offset);
+ } else {
+ read = qemu_get_buffer_at(f, host, size,
+ block->pages_offset + offset);
+ }
+
if (!read) {
goto err;
}