motorhead/qemu
1
0
Fork 0
mirror of https://github.com/Motorhead1991/qemu.git synced 2025-08-03 15:53:54 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions
qemu/migration/multifd.c
Peter Xu 10801e08ac migration/multifd: Allow to sync with sender threads only 
Teach multifd_send_sync_main() to sync with threads only.

We already have such requests, which is when mapped-ram is enabled with
multifd.  In that case, no SYNC messages will be pushed to the stream when
multifd syncs the sender threads because there's no destination threads
waiting for that.  The whole point of the sync is to make sure all threads
finished their jobs.

So fundamentally we have a request to do the sync in different ways:

  - Either to sync the threads only,
  - Or to sync the threads but also with the destination side.

Mapped-ram did it already because of the use_packet check in the sync
handler of the sender thread.  It works.

However it may stop working when e.g. VFIO may start to reuse multifd
channels to push device states.  In that case VFIO has similar request on
"thread-only sync" however we can't check a flag because such sync request
can still come from RAM which needs the on-wire notifications.

Paving way for that by allowing the multifd_send_sync_main() to specify
what kind of sync the caller needs.  We can use it for mapped-ram already.

No functional change intended.

Signed-off-by: Peter Xu <peterx@redhat.com>
Reviewed-by: Fabiano Rosas <farosas@suse.de>
Message-Id: <20241206224755.1108686-3-peterx@redhat.com>
Signed-off-by: Fabiano Rosas <farosas@suse.de>
2025-01-09 17:38:12 -03:00

1341 lines
38 KiB
C
Raw Blame History

/*
* Multifd common code
*
* Copyright (c) 2019-2020 Red Hat Inc
*
* Authors:
* Juan Quintela <quintela@redhat.com>
*
* This work is licensed under the terms of the GNU GPL, version 2 or later.
* See the COPYING file in the top-level directory.
*/
#include "qemu/osdep.h"
#include "qemu/cutils.h"
#include "qemu/rcu.h"
#include "exec/target_page.h"
#include "system/system.h"
#include "exec/ramblock.h"
#include "qemu/error-report.h"
#include "qapi/error.h"
#include "file.h"
#include "migration.h"
#include "migration-stats.h"
#include "socket.h"
#include "tls.h"
#include "qemu-file.h"
#include "trace.h"
#include "multifd.h"
#include "threadinfo.h"
#include "options.h"
#include "qemu/yank.h"
#include "io/channel-file.h"
#include "io/channel-socket.h"
#include "yank_functions.h"
/* Multiple fd's */
#define MULTIFD_MAGIC 0x11223344U
#define MULTIFD_VERSION 1
typedef struct {
uint32_t magic;
uint32_t version;
unsigned char uuid[16]; /* QemuUUID */
uint8_t id;
uint8_t unused1[7]; /* Reserved for future use */
uint64_t unused2[4]; /* Reserved for future use */
} __attribute__((packed)) MultiFDInit_t;
struct {
MultiFDSendParams *params;
/*
* Global number of generated multifd packets.
*
* Note that we used 'uintptr_t' because it'll naturally support atomic
* operations on both 32bit / 64 bits hosts. It means on 32bit systems
* multifd will overflow the packet_num easier, but that should be
* fine.
*
* Another option is to use QEMU's Stat64 then it'll be 64 bits on all
* hosts, however so far it does not support atomic fetch_add() yet.
* Make it easy for now.
*/
uintptr_t packet_num;
/*
* Synchronization point past which no more channels will be
* created.
*/
QemuSemaphore channels_created;
/* send channels ready */
QemuSemaphore channels_ready;
/*
* Have we already run terminate threads. There is a race when it
* happens that we got one error while we are exiting.
* We will use atomic operations. Only valid values are 0 and 1.
*/
int exiting;
/* multifd ops */
const MultiFDMethods *ops;
} *multifd_send_state;
struct {
MultiFDRecvParams *params;
MultiFDRecvData *data;
/* number of created threads */
int count;
/*
* This is always posted by the recv threads, the migration thread
* uses it to wait for recv threads to finish assigned tasks.
*/
QemuSemaphore sem_sync;
/* global number of generated multifd packets */
uint64_t packet_num;
int exiting;
/* multifd ops */
const MultiFDMethods *ops;
} *multifd_recv_state;
MultiFDSendData *multifd_send_data_alloc(void)
{
size_t max_payload_size, size_minus_payload;
/*
* MultiFDPages_t has a flexible array at the end, account for it
* when allocating MultiFDSendData. Use max() in case other types
* added to the union in the future are larger than
* (MultiFDPages_t + flex array).
*/
max_payload_size = MAX(multifd_ram_payload_size(), sizeof(MultiFDPayload));
/*
* Account for any holes the compiler might insert. We can't pack
* the structure because that misaligns the members and triggers
* Waddress-of-packed-member.
*/
size_minus_payload = sizeof(MultiFDSendData) - sizeof(MultiFDPayload);
return g_malloc0(size_minus_payload + max_payload_size);
}
static bool multifd_use_packets(void)
{
return !migrate_mapped_ram();
}
void multifd_send_channel_created(void)
{
qemu_sem_post(&multifd_send_state->channels_created);
}
static const MultiFDMethods *multifd_ops[MULTIFD_COMPRESSION__MAX] = {};
void multifd_register_ops(int method, const MultiFDMethods *ops)
{
assert(0 <= method && method < MULTIFD_COMPRESSION__MAX);
assert(!multifd_ops[method]);
multifd_ops[method] = ops;
}
static int multifd_send_initial_packet(MultiFDSendParams *p, Error **errp)
{
MultiFDInit_t msg = {};
size_t size = sizeof(msg);
int ret;
msg.magic = cpu_to_be32(MULTIFD_MAGIC);
msg.version = cpu_to_be32(MULTIFD_VERSION);
msg.id = p->id;
memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
ret = qio_channel_write_all(p->c, (char *)&msg, size, errp);
if (ret != 0) {
return -1;
}
stat64_add(&mig_stats.multifd_bytes, size);
return 0;
}
static int multifd_recv_initial_packet(QIOChannel *c, Error **errp)
{
MultiFDInit_t msg;
int ret;
ret = qio_channel_read_all(c, (char *)&msg, sizeof(msg), errp);
if (ret != 0) {
return -1;
}
msg.magic = be32_to_cpu(msg.magic);
msg.version = be32_to_cpu(msg.version);
if (msg.magic != MULTIFD_MAGIC) {
error_setg(errp, "multifd: received packet magic %x "
"expected %x", msg.magic, MULTIFD_MAGIC);
return -1;
}
if (msg.version != MULTIFD_VERSION) {
error_setg(errp, "multifd: received packet version %u "
"expected %u", msg.version, MULTIFD_VERSION);
return -1;
}
if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
char *msg_uuid = qemu_uuid_unparse_strdup((const QemuUUID *)msg.uuid);
error_setg(errp, "multifd: received uuid '%s' and expected "
"uuid '%s' for channel %hhd", msg_uuid, uuid, msg.id);
g_free(uuid);
g_free(msg_uuid);
return -1;
}
if (msg.id > migrate_multifd_channels()) {
error_setg(errp, "multifd: received channel id %u is greater than "
"number of channels %u", msg.id, migrate_multifd_channels());
return -1;
}
return msg.id;
}
void multifd_send_fill_packet(MultiFDSendParams *p)
{
MultiFDPacket_t *packet = p->packet;
uint64_t packet_num;
bool sync_packet = p->flags & MULTIFD_FLAG_SYNC;
memset(packet, 0, p->packet_len);
packet->magic = cpu_to_be32(MULTIFD_MAGIC);
packet->version = cpu_to_be32(MULTIFD_VERSION);
packet->flags = cpu_to_be32(p->flags);
packet->next_packet_size = cpu_to_be32(p->next_packet_size);
packet_num = qatomic_fetch_inc(&multifd_send_state->packet_num);
packet->packet_num = cpu_to_be64(packet_num);
p->packets_sent++;
if (!sync_packet) {
multifd_ram_fill_packet(p);
}
trace_multifd_send_fill(p->id, packet_num,
p->flags, p->next_packet_size);
}
static int multifd_recv_unfill_packet(MultiFDRecvParams *p, Error **errp)
{
const MultiFDPacket_t *packet = p->packet;
uint32_t magic = be32_to_cpu(packet->magic);
uint32_t version = be32_to_cpu(packet->version);
int ret = 0;
if (magic != MULTIFD_MAGIC) {
error_setg(errp, "multifd: received packet magic %x, expected %x",
magic, MULTIFD_MAGIC);
return -1;
}
if (version != MULTIFD_VERSION) {
error_setg(errp, "multifd: received packet version %u, expected %u",
version, MULTIFD_VERSION);
return -1;
}
p->flags = be32_to_cpu(packet->flags);
p->next_packet_size = be32_to_cpu(packet->next_packet_size);
p->packet_num = be64_to_cpu(packet->packet_num);
p->packets_recved++;
if (!(p->flags & MULTIFD_FLAG_SYNC)) {
ret = multifd_ram_unfill_packet(p, errp);
}
trace_multifd_recv_unfill(p->id, p->packet_num, p->flags,
p->next_packet_size);
return ret;
}
static bool multifd_send_should_exit(void)
{
return qatomic_read(&multifd_send_state->exiting);
}
static bool multifd_recv_should_exit(void)
{
return qatomic_read(&multifd_recv_state->exiting);
}
/*
* The migration thread can wait on either of the two semaphores. This
* function can be used to kick the main thread out of waiting on either of
* them. Should mostly only be called when something wrong happened with
* the current multifd send thread.
*/
static void multifd_send_kick_main(MultiFDSendParams *p)
{
qemu_sem_post(&p->sem_sync);
qemu_sem_post(&multifd_send_state->channels_ready);
}
/*
* multifd_send() works by exchanging the MultiFDSendData object
* provided by the caller with an unused MultiFDSendData object from
* the next channel that is found to be idle.
*
* The channel owns the data until it finishes transmitting and the
* caller owns the empty object until it fills it with data and calls
* this function again. No locking necessary.
*
* Switching is safe because both the migration thread and the channel
* thread have barriers in place to serialize access.
*
* Returns true if succeed, false otherwise.
*/
bool multifd_send(MultiFDSendData **send_data)
{
int i;
static int next_channel;
MultiFDSendParams *p = NULL; /* make happy gcc */
MultiFDSendData *tmp;
if (multifd_send_should_exit()) {
return false;
}
/* We wait here, until at least one channel is ready */
qemu_sem_wait(&multifd_send_state->channels_ready);
/*
* next_channel can remain from a previous migration that was
* using more channels, so ensure it doesn't overflow if the
* limit is lower now.
*/
next_channel %= migrate_multifd_channels();
for (i = next_channel;; i = (i + 1) % migrate_multifd_channels()) {
if (multifd_send_should_exit()) {
return false;
}
p = &multifd_send_state->params[i];
/*
* Lockless read to p->pending_job is safe, because only multifd
* sender thread can clear it.
*/
if (qatomic_read(&p->pending_job) == false) {
next_channel = (i + 1) % migrate_multifd_channels();
break;
}
}
/*
* Make sure we read p->pending_job before all the rest. Pairs with
* qatomic_store_release() in multifd_send_thread().
*/
smp_mb_acquire();
assert(multifd_payload_empty(p->data));
/*
* Swap the pointers. The channel gets the client data for
* transferring and the client gets back an unused data slot.
*/
tmp = *send_data;
*send_data = p->data;
p->data = tmp;
/*
* Making sure p->data is setup before marking pending_job=true. Pairs
* with the qatomic_load_acquire() in multifd_send_thread().
*/
qatomic_store_release(&p->pending_job, true);
qemu_sem_post(&p->sem);
return true;
}
/* Multifd send side hit an error; remember it and prepare to quit */
static void multifd_send_set_error(Error *err)
{
/*
* We don't want to exit each threads twice. Depending on where
* we get the error, or if there are two independent errors in two
* threads at the same time, we can end calling this function
* twice.
*/
if (qatomic_xchg(&multifd_send_state->exiting, 1)) {
return;
}
if (err) {
MigrationState *s = migrate_get_current();
migrate_set_error(s, err);
if (s->state == MIGRATION_STATUS_SETUP ||
s->state == MIGRATION_STATUS_PRE_SWITCHOVER ||
s->state == MIGRATION_STATUS_DEVICE ||
s->state == MIGRATION_STATUS_ACTIVE) {
migrate_set_state(&s->state, s->state,
MIGRATION_STATUS_FAILED);
}
}
}
static void multifd_send_terminate_threads(void)
{
int i;
trace_multifd_send_terminate_threads();
/*
* Tell everyone we're quitting. No xchg() needed here; we simply
* always set it.
*/
qatomic_set(&multifd_send_state->exiting, 1);
/*
* Firstly, kick all threads out; no matter whether they are just idle,
* or blocked in an IO system call.
*/
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
qemu_sem_post(&p->sem);
if (p->c) {
qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
}
}
/*
* Finally recycle all the threads.
*/
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
if (p->tls_thread_created) {
qemu_thread_join(&p->tls_thread);
}
if (p->thread_created) {
qemu_thread_join(&p->thread);
}
}
}
static bool multifd_send_cleanup_channel(MultiFDSendParams *p, Error **errp)
{
if (p->c) {
migration_ioc_unregister_yank(p->c);
/*
* The object_unref() cannot guarantee the fd will always be
* released because finalize() of the iochannel is only
* triggered on the last reference and it's not guaranteed
* that we always hold the last refcount when reaching here.
*
* Closing the fd explicitly has the benefit that if there is any
* registered I/O handler callbacks on such fd, that will get a
* POLLNVAL event and will further trigger the cleanup to finally
* release the IOC.
*
* FIXME: It should logically be guaranteed that all multifd
* channels have no I/O handler callback registered when reaching
* here, because migration thread will wait for all multifd channel
* establishments to complete during setup. Since
* migrate_fd_cleanup() will be scheduled in main thread too, all
* previous callbacks should guarantee to be completed when
* reaching here. See multifd_send_state.channels_created and its
* usage. In the future, we could replace this with an assert
* making sure we're the last reference, or simply drop it if above
* is more clear to be justified.
*/
qio_channel_close(p->c, &error_abort);
object_unref(OBJECT(p->c));
p->c = NULL;
}
qemu_sem_destroy(&p->sem);
qemu_sem_destroy(&p->sem_sync);
g_free(p->name);
p->name = NULL;
g_free(p->data);
p->data = NULL;
p->packet_len = 0;
g_free(p->packet);
p->packet = NULL;
multifd_send_state->ops->send_cleanup(p, errp);
assert(!p->iov);
return *errp == NULL;
}
static void multifd_send_cleanup_state(void)
{
file_cleanup_outgoing_migration();
socket_cleanup_outgoing_migration();
qemu_sem_destroy(&multifd_send_state->channels_created);
qemu_sem_destroy(&multifd_send_state->channels_ready);
g_free(multifd_send_state->params);
multifd_send_state->params = NULL;
g_free(multifd_send_state);
multifd_send_state = NULL;
}
void multifd_send_shutdown(void)
{
int i;
if (!migrate_multifd()) {
return;
}
multifd_send_terminate_threads();
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
Error *local_err = NULL;
if (!multifd_send_cleanup_channel(p, &local_err)) {
migrate_set_error(migrate_get_current(), local_err);
error_free(local_err);
}
}
multifd_send_cleanup_state();
}
static int multifd_zero_copy_flush(QIOChannel *c)
{
int ret;
Error *err = NULL;
ret = qio_channel_flush(c, &err);
if (ret < 0) {
error_report_err(err);
return -1;
}
if (ret == 1) {
stat64_add(&mig_stats.dirty_sync_missed_zero_copy, 1);
}
return ret;
}
int multifd_send_sync_main(MultiFDSyncReq req)
{
int i;
bool flush_zero_copy;
assert(req != MULTIFD_SYNC_NONE);
flush_zero_copy = migrate_zero_copy_send();
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
if (multifd_send_should_exit()) {
return -1;
}
trace_multifd_send_sync_main_signal(p->id);
/*
* We should be the only user so far, so not possible to be set by
* others concurrently.
*/
assert(qatomic_read(&p->pending_sync) == MULTIFD_SYNC_NONE);
qatomic_set(&p->pending_sync, req);
qemu_sem_post(&p->sem);
}
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
if (multifd_send_should_exit()) {
return -1;
}
qemu_sem_wait(&multifd_send_state->channels_ready);
trace_multifd_send_sync_main_wait(p->id);
qemu_sem_wait(&p->sem_sync);
if (flush_zero_copy && p->c && (multifd_zero_copy_flush(p->c) < 0)) {
return -1;
}
}
trace_multifd_send_sync_main(multifd_send_state->packet_num);
return 0;
}
static void *multifd_send_thread(void *opaque)
{
MultiFDSendParams *p = opaque;
MigrationThread *thread = NULL;
Error *local_err = NULL;
int ret = 0;
bool use_packets = multifd_use_packets();
thread = migration_threads_add(p->name, qemu_get_thread_id());
trace_multifd_send_thread_start(p->id);
rcu_register_thread();
if (use_packets) {
if (multifd_send_initial_packet(p, &local_err) < 0) {
ret = -1;
goto out;
}
}
while (true) {
qemu_sem_post(&multifd_send_state->channels_ready);
qemu_sem_wait(&p->sem);
if (multifd_send_should_exit()) {
break;
}
/*
* Read pending_job flag before p->data. Pairs with the
* qatomic_store_release() in multifd_send().
*/
if (qatomic_load_acquire(&p->pending_job)) {
p->flags = 0;
p->iovs_num = 0;
assert(!multifd_payload_empty(p->data));
ret = multifd_send_state->ops->send_prepare(p, &local_err);
if (ret != 0) {
break;
}
if (migrate_mapped_ram()) {
ret = file_write_ramblock_iov(p->c, p->iov, p->iovs_num,
&p->data->u.ram, &local_err);
} else {
ret = qio_channel_writev_full_all(p->c, p->iov, p->iovs_num,
NULL, 0, p->write_flags,
&local_err);
}
if (ret != 0) {
break;
}
stat64_add(&mig_stats.multifd_bytes,
(uint64_t)p->next_packet_size + p->packet_len);
p->next_packet_size = 0;
multifd_set_payload_type(p->data, MULTIFD_PAYLOAD_NONE);
/*
* Making sure p->data is published before saying "we're
* free". Pairs with the smp_mb_acquire() in
* multifd_send().
*/
qatomic_store_release(&p->pending_job, false);
} else {
MultiFDSyncReq req = qatomic_read(&p->pending_sync);
/*
* If not a normal job, must be a sync request. Note that
* pending_sync is a standalone flag (unlike pending_job), so
* it doesn't require explicit memory barriers.
*/
assert(req != MULTIFD_SYNC_NONE);
/* Only push the SYNC message if it involves a remote sync */
if (req == MULTIFD_SYNC_ALL) {
p->flags = MULTIFD_FLAG_SYNC;
multifd_send_fill_packet(p);
ret = qio_channel_write_all(p->c, (void *)p->packet,
p->packet_len, &local_err);
if (ret != 0) {
break;
}
/* p->next_packet_size will always be zero for a SYNC packet */
stat64_add(&mig_stats.multifd_bytes, p->packet_len);
}
qatomic_set(&p->pending_sync, MULTIFD_SYNC_NONE);
qemu_sem_post(&p->sem_sync);
}
}
out:
if (ret) {
assert(local_err);
trace_multifd_send_error(p->id);
multifd_send_set_error(local_err);
multifd_send_kick_main(p);
error_free(local_err);
}
rcu_unregister_thread();
migration_threads_remove(thread);
trace_multifd_send_thread_end(p->id, p->packets_sent);
return NULL;
}
static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque);
typedef struct {
MultiFDSendParams *p;
QIOChannelTLS *tioc;
} MultiFDTLSThreadArgs;
static void *multifd_tls_handshake_thread(void *opaque)
{
MultiFDTLSThreadArgs *args = opaque;
qio_channel_tls_handshake(args->tioc,
multifd_new_send_channel_async,
args->p,
NULL,
NULL);
g_free(args);
return NULL;
}
static bool multifd_tls_channel_connect(MultiFDSendParams *p,
QIOChannel *ioc,
Error **errp)
{
MigrationState *s = migrate_get_current();
const char *hostname = s->hostname;
MultiFDTLSThreadArgs *args;
QIOChannelTLS *tioc;
tioc = migration_tls_client_create(ioc, hostname, errp);
if (!tioc) {
return false;
}
/*
* Ownership of the socket channel now transfers to the newly
* created TLS channel, which has already taken a reference.
*/
object_unref(OBJECT(ioc));
trace_multifd_tls_outgoing_handshake_start(ioc, tioc, hostname);
qio_channel_set_name(QIO_CHANNEL(tioc), "multifd-tls-outgoing");
args = g_new0(MultiFDTLSThreadArgs, 1);
args->tioc = tioc;
args->p = p;
p->tls_thread_created = true;
qemu_thread_create(&p->tls_thread, MIGRATION_THREAD_SRC_TLS,
multifd_tls_handshake_thread, args,
QEMU_THREAD_JOINABLE);
return true;
}
void multifd_channel_connect(MultiFDSendParams *p, QIOChannel *ioc)
{
qio_channel_set_delay(ioc, false);
migration_ioc_register_yank(ioc);
/* Setup p->c only if the channel is completely setup */
p->c = ioc;
p->thread_created = true;
qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
QEMU_THREAD_JOINABLE);
}
/*
* When TLS is enabled this function is called once to establish the
* TLS connection and a second time after the TLS handshake to create
* the multifd channel. Without TLS it goes straight into the channel
* creation.
*/
static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
{
MultiFDSendParams *p = opaque;
QIOChannel *ioc = QIO_CHANNEL(qio_task_get_source(task));
Error *local_err = NULL;
bool ret;
trace_multifd_new_send_channel_async(p->id);
if (qio_task_propagate_error(task, &local_err)) {
ret = false;
goto out;
}
trace_multifd_set_outgoing_channel(ioc, object_get_typename(OBJECT(ioc)),
migrate_get_current()->hostname);
if (migrate_channel_requires_tls_upgrade(ioc)) {
ret = multifd_tls_channel_connect(p, ioc, &local_err);
if (ret) {
return;
}
} else {
multifd_channel_connect(p, ioc);
ret = true;
}
out:
/*
* Here we're not interested whether creation succeeded, only that
* it happened at all.
*/
multifd_send_channel_created();
if (ret) {
return;
}
trace_multifd_new_send_channel_async_error(p->id, local_err);
multifd_send_set_error(local_err);
/*
* For error cases (TLS or non-TLS), IO channel is always freed here
* rather than when cleanup multifd: since p->c is not set, multifd
* cleanup code doesn't even know its existence.
*/
object_unref(OBJECT(ioc));
error_free(local_err);
}
static bool multifd_new_send_channel_create(gpointer opaque, Error **errp)
{
if (!multifd_use_packets()) {
return file_send_channel_create(opaque, errp);
}
socket_send_channel_create(multifd_new_send_channel_async, opaque);
return true;
}
bool multifd_send_setup(void)
{
MigrationState *s = migrate_get_current();
int thread_count, ret = 0;
uint32_t page_count = multifd_ram_page_count();
bool use_packets = multifd_use_packets();
uint8_t i;
if (!migrate_multifd()) {
return true;
}
thread_count = migrate_multifd_channels();
multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
qemu_sem_init(&multifd_send_state->channels_created, 0);
qemu_sem_init(&multifd_send_state->channels_ready, 0);
qatomic_set(&multifd_send_state->exiting, 0);
multifd_send_state->ops = multifd_ops[migrate_multifd_compression()];
for (i = 0; i < thread_count; i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
Error *local_err = NULL;
qemu_sem_init(&p->sem, 0);
qemu_sem_init(&p->sem_sync, 0);
p->id = i;
p->data = multifd_send_data_alloc();
if (use_packets) {
p->packet_len = sizeof(MultiFDPacket_t)
+ sizeof(uint64_t) * page_count;
p->packet = g_malloc0(p->packet_len);
}
p->name = g_strdup_printf(MIGRATION_THREAD_SRC_MULTIFD, i);
p->write_flags = 0;
if (!multifd_new_send_channel_create(p, &local_err)) {
migrate_set_error(s, local_err);
ret = -1;
}
}
/*
* Wait until channel creation has started for all channels. The
* creation can still fail, but no more channels will be created
* past this point.
*/
for (i = 0; i < thread_count; i++) {
qemu_sem_wait(&multifd_send_state->channels_created);
}
if (ret) {
goto err;
}
for (i = 0; i < thread_count; i++) {
MultiFDSendParams *p = &multifd_send_state->params[i];
Error *local_err = NULL;
ret = multifd_send_state->ops->send_setup(p, &local_err);
if (ret) {
migrate_set_error(s, local_err);
goto err;
}
assert(p->iov);
}
return true;
err:
migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
MIGRATION_STATUS_FAILED);
return false;
}
bool multifd_recv(void)
{
int i;
static int next_recv_channel;
MultiFDRecvParams *p = NULL;
MultiFDRecvData *data = multifd_recv_state->data;
/*
* next_channel can remain from a previous migration that was
* using more channels, so ensure it doesn't overflow if the
* limit is lower now.
*/
next_recv_channel %= migrate_multifd_channels();
for (i = next_recv_channel;; i = (i + 1) % migrate_multifd_channels()) {
if (multifd_recv_should_exit()) {
return false;
}
p = &multifd_recv_state->params[i];
if (qatomic_read(&p->pending_job) == false) {
next_recv_channel = (i + 1) % migrate_multifd_channels();
break;
}
}
/*
* Order pending_job read before manipulating p->data below. Pairs
* with qatomic_store_release() at multifd_recv_thread().
*/
smp_mb_acquire();
assert(!p->data->size);
multifd_recv_state->data = p->data;
p->data = data;
/*
* Order p->data update before setting pending_job. Pairs with
* qatomic_load_acquire() at multifd_recv_thread().
*/
qatomic_store_release(&p->pending_job, true);
qemu_sem_post(&p->sem);
return true;
}
MultiFDRecvData *multifd_get_recv_data(void)
{
return multifd_recv_state->data;
}
static void multifd_recv_terminate_threads(Error *err)
{
int i;
trace_multifd_recv_terminate_threads(err != NULL);
if (qatomic_xchg(&multifd_recv_state->exiting, 1)) {
return;
}
if (err) {
MigrationState *s = migrate_get_current();
migrate_set_error(s, err);
if (s->state == MIGRATION_STATUS_SETUP ||
s->state == MIGRATION_STATUS_ACTIVE) {
migrate_set_state(&s->state, s->state,
MIGRATION_STATUS_FAILED);
}
}
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDRecvParams *p = &multifd_recv_state->params[i];
/*
* The migration thread and channels interact differently
* depending on the presence of packets.
*/
if (multifd_use_packets()) {
/*
* The channel receives as long as there are packets. When
* packets end (i.e. MULTIFD_FLAG_SYNC is reached), the
* channel waits for the migration thread to sync. If the
* sync never happens, do it here.
*/
qemu_sem_post(&p->sem_sync);
} else {
/*
* The channel waits for the migration thread to give it
* work. When the migration thread runs out of work, it
* releases the channel and waits for any pending work to
* finish. If we reach here (e.g. due to error) before the
* work runs out, release the channel.
*/
qemu_sem_post(&p->sem);
}
/*
* We could arrive here for two reasons:
* - normal quit, i.e. everything went fine, just finished
* - error quit: We close the channels so the channel threads
* finish the qio_channel_read_all_eof()
*/
if (p->c) {
qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
}
}
}
void multifd_recv_shutdown(void)
{
if (migrate_multifd()) {
multifd_recv_terminate_threads(NULL);
}
}
static void multifd_recv_cleanup_channel(MultiFDRecvParams *p)
{
migration_ioc_unregister_yank(p->c);
object_unref(OBJECT(p->c));
p->c = NULL;
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem_sync);
qemu_sem_destroy(&p->sem);
g_free(p->data);
p->data = NULL;
g_free(p->name);
p->name = NULL;
p->packet_len = 0;
g_free(p->packet);
p->packet = NULL;
g_free(p->normal);
p->normal = NULL;
g_free(p->zero);
p->zero = NULL;
multifd_recv_state->ops->recv_cleanup(p);
}
static void multifd_recv_cleanup_state(void)
{
qemu_sem_destroy(&multifd_recv_state->sem_sync);
g_free(multifd_recv_state->params);
multifd_recv_state->params = NULL;
g_free(multifd_recv_state->data);
multifd_recv_state->data = NULL;
g_free(multifd_recv_state);
multifd_recv_state = NULL;
}
void multifd_recv_cleanup(void)
{
int i;
if (!migrate_multifd()) {
return;
}
multifd_recv_terminate_threads(NULL);
for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDRecvParams *p = &multifd_recv_state->params[i];
if (p->thread_created) {
qemu_thread_join(&p->thread);
}
}
for (i = 0; i < migrate_multifd_channels(); i++) {
multifd_recv_cleanup_channel(&multifd_recv_state->params[i]);
}
multifd_recv_cleanup_state();
}
void multifd_recv_sync_main(void)
{
int thread_count = migrate_multifd_channels();
bool file_based = !multifd_use_packets();
int i;
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.
*/
for (i = 0; i < thread_count; i++) {
MultiFDRecvParams *p = &multifd_recv_state->params[i];
WITH_QEMU_LOCK_GUARD(&p->mutex) {
if (multifd_recv_state->packet_num < p->packet_num) {
multifd_recv_state->packet_num = p->packet_num;
}
}
trace_multifd_recv_sync_main_signal(p->id);
qemu_sem_post(&p->sem_sync);
}
trace_multifd_recv_sync_main(multifd_recv_state->packet_num);
}
static void *multifd_recv_thread(void *opaque)
{
MultiFDRecvParams *p = opaque;
Error *local_err = NULL;
bool use_packets = multifd_use_packets();
int ret;
trace_multifd_recv_thread_start(p->id);
rcu_register_thread();
while (true) {
uint32_t flags = 0;
bool has_data = false;
p->normal_num = 0;
if (use_packets) {
if (multifd_recv_should_exit()) {
break;
}
ret = qio_channel_read_all_eof(p->c, (void *)p->packet,
p->packet_len, &local_err);
if (ret == 0 || ret == -1) { /* 0: EOF -1: Error */
break;
}
qemu_mutex_lock(&p->mutex);
ret = multifd_recv_unfill_packet(p, &local_err);
if (ret) {
qemu_mutex_unlock(&p->mutex);
break;
}
flags = p->flags;
/* recv methods don't know how to handle the SYNC flag */
p->flags &= ~MULTIFD_FLAG_SYNC;
if (!(flags & MULTIFD_FLAG_SYNC)) {
has_data = p->normal_num || p->zero_num;
}
qemu_mutex_unlock(&p->mutex);
} else {
/*
* No packets, so we need to wait for the vmstate code to
* give us work.
*/
qemu_sem_wait(&p->sem);
if (multifd_recv_should_exit()) {
break;
}
/* pairs with qatomic_store_release() at multifd_recv() */
if (!qatomic_load_acquire(&p->pending_job)) {
/*
* Migration thread did not send work, this is
* equivalent to pending_sync on the sending
* side. Post sem_sync to notify we reached this
* point.
*/
qemu_sem_post(&multifd_recv_state->sem_sync);
continue;
}
has_data = !!p->data->size;
}
if (has_data) {
ret = multifd_recv_state->ops->recv(p, &local_err);
if (ret != 0) {
break;
}
}
if (use_packets) {
if (flags & MULTIFD_FLAG_SYNC) {
qemu_sem_post(&multifd_recv_state->sem_sync);
qemu_sem_wait(&p->sem_sync);
}
} else {
p->data->size = 0;
/*
* Order data->size update before clearing
* pending_job. Pairs with smp_mb_acquire() at
* multifd_recv().
*/
qatomic_store_release(&p->pending_job, false);
}
}
if (local_err) {
multifd_recv_terminate_threads(local_err);
error_free(local_err);
}
rcu_unregister_thread();
trace_multifd_recv_thread_end(p->id, p->packets_recved);
return NULL;
}
int multifd_recv_setup(Error **errp)
{
int thread_count;
uint32_t page_count = multifd_ram_page_count();
bool use_packets = multifd_use_packets();
uint8_t i;
/*
* Return successfully if multiFD recv state is already initialised
* or multiFD is not enabled.
*/
if (multifd_recv_state || !migrate_multifd()) {
return 0;
}
thread_count = migrate_multifd_channels();
multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
multifd_recv_state->data = g_new0(MultiFDRecvData, 1);
multifd_recv_state->data->size = 0;
qatomic_set(&multifd_recv_state->count, 0);
qatomic_set(&multifd_recv_state->exiting, 0);
qemu_sem_init(&multifd_recv_state->sem_sync, 0);
multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
for (i = 0; i < thread_count; i++) {
MultiFDRecvParams *p = &multifd_recv_state->params[i];
qemu_mutex_init(&p->mutex);
qemu_sem_init(&p->sem_sync, 0);
qemu_sem_init(&p->sem, 0);
p->pending_job = false;
p->id = i;
p->data = g_new0(MultiFDRecvData, 1);
p->data->size = 0;
if (use_packets) {
p->packet_len = sizeof(MultiFDPacket_t)
+ sizeof(uint64_t) * page_count;
p->packet = g_malloc0(p->packet_len);
}
p->name = g_strdup_printf(MIGRATION_THREAD_DST_MULTIFD, i);
p->normal = g_new0(ram_addr_t, page_count);
p->zero = g_new0(ram_addr_t, page_count);
}
for (i = 0; i < thread_count; i++) {
MultiFDRecvParams *p = &multifd_recv_state->params[i];
int ret;
ret = multifd_recv_state->ops->recv_setup(p, errp);
if (ret) {
return ret;
}
}
return 0;
}
bool multifd_recv_all_channels_created(void)
{
int thread_count = migrate_multifd_channels();
if (!migrate_multifd()) {
return true;
}
if (!multifd_recv_state) {
/* Called before any connections created */
return false;
}
return thread_count == qatomic_read(&multifd_recv_state->count);
}
/*
* Try to receive all multifd channels to get ready for the migration.
* Sets @errp when failing to receive the current channel.
*/
void multifd_recv_new_channel(QIOChannel *ioc, Error **errp)
{
MultiFDRecvParams *p;
Error *local_err = NULL;
bool use_packets = multifd_use_packets();
int id;
if (use_packets) {
id = multifd_recv_initial_packet(ioc, &local_err);
if (id < 0) {
multifd_recv_terminate_threads(local_err);
error_propagate_prepend(errp, local_err,
"failed to receive packet"
" via multifd channel %d: ",
qatomic_read(&multifd_recv_state->count));
return;
}
trace_multifd_recv_new_channel(id);
} else {
id = qatomic_read(&multifd_recv_state->count);
}
p = &multifd_recv_state->params[id];
if (p->c != NULL) {
error_setg(&local_err, "multifd: received id '%d' already setup'",
id);
multifd_recv_terminate_threads(local_err);
error_propagate(errp, local_err);
return;
}
p->c = ioc;
object_ref(OBJECT(ioc));
p->thread_created = true;
qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
QEMU_THREAD_JOINABLE);
qatomic_inc(&multifd_recv_state->count);
}
Reference in a new issue View git blame Copy permalink