migration/multifd: Support incoming mapped-ram stream format

For the incoming mapped-ram migration we need to read the ramblock
headers, get the pages bitmap and send the host address of each
non-zero page to the multifd channel thread for writing.

Usage on HMP is:

(qemu) migrate_set_capability multifd on
(qemu) migrate_set_capability mapped-ram on
(qemu) migrate_incoming file:migfile

(the ram.h include needs to move because we've been previously relying
on it being included from migration.c. Now file.h will start including
multifd.h before migration.o is processed)

Reviewed-by: Peter Xu <peterx@redhat.com>
Signed-off-by: Fabiano Rosas <farosas@suse.de>
Link: https://lore.kernel.org/r/20240229153017.2221-22-farosas@suse.de
Signed-off-by: Peter Xu <peterx@redhat.com>
This commit is contained in:
Fabiano Rosas 2024-02-29 12:30:15 -03:00 committed by Peter Xu
parent f427d90b98
commit a49d15a38d
5 changed files with 73 additions and 6 deletions

View file

@ -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;
}

View file

@ -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

View file

@ -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.
*/

View file

@ -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);

View file

@ -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;
}