migration/multifd: Remove p->quit from recv side

Like we did on the sending side, replace the p->quit per-channel flag
with a global atomic 'exiting' flag.

Signed-off-by: Fabiano Rosas <farosas@suse.de>
Reviewed-by: Peter Xu <peterx@redhat.com>
Link: https://lore.kernel.org/r/20240220224138.24759-5-farosas@suse.de
Signed-off-by: Peter Xu <peterx@redhat.com>
This commit is contained in:
Fabiano Rosas 2024-02-20 19:41:08 -03:00 committed by Peter Xu
parent 6d79bd6818
commit 11dd7be575

View file

@ -79,6 +79,19 @@ struct {
MultiFDMethods *ops; MultiFDMethods *ops;
} *multifd_send_state; } *multifd_send_state;
struct {
MultiFDRecvParams *params;
/* number of created threads */
int count;
/* syncs main thread and channels */
QemuSemaphore sem_sync;
/* global number of generated multifd packets */
uint64_t packet_num;
int exiting;
/* multifd ops */
MultiFDMethods *ops;
} *multifd_recv_state;
/* Multifd without compression */ /* Multifd without compression */
/** /**
@ -440,6 +453,11 @@ static bool multifd_send_should_exit(void)
return qatomic_read(&multifd_send_state->exiting); 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 * 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 * function can be used to kick the main thread out of waiting on either of
@ -1063,24 +1081,16 @@ bool multifd_send_setup(void)
return true; return true;
} }
struct {
MultiFDRecvParams *params;
/* number of created threads */
int count;
/* syncs main thread and channels */
QemuSemaphore sem_sync;
/* global number of generated multifd packets */
uint64_t packet_num;
/* multifd ops */
MultiFDMethods *ops;
} *multifd_recv_state;
static void multifd_recv_terminate_threads(Error *err) static void multifd_recv_terminate_threads(Error *err)
{ {
int i; int i;
trace_multifd_recv_terminate_threads(err != NULL); trace_multifd_recv_terminate_threads(err != NULL);
if (qatomic_xchg(&multifd_recv_state->exiting, 1)) {
return;
}
if (err) { if (err) {
MigrationState *s = migrate_get_current(); MigrationState *s = migrate_get_current();
migrate_set_error(s, err); migrate_set_error(s, err);
@ -1094,8 +1104,6 @@ static void multifd_recv_terminate_threads(Error *err)
for (i = 0; i < migrate_multifd_channels(); i++) { for (i = 0; i < migrate_multifd_channels(); i++) {
MultiFDRecvParams *p = &multifd_recv_state->params[i]; MultiFDRecvParams *p = &multifd_recv_state->params[i];
qemu_mutex_lock(&p->mutex);
p->quit = true;
/* /*
* We could arrive here for two reasons: * We could arrive here for two reasons:
* - normal quit, i.e. everything went fine, just finished * - normal quit, i.e. everything went fine, just finished
@ -1105,7 +1113,6 @@ static void multifd_recv_terminate_threads(Error *err)
if (p->c) { if (p->c) {
qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL); qio_channel_shutdown(p->c, QIO_CHANNEL_SHUTDOWN_BOTH, NULL);
} }
qemu_mutex_unlock(&p->mutex);
} }
} }
@ -1210,7 +1217,7 @@ static void *multifd_recv_thread(void *opaque)
while (true) { while (true) {
uint32_t flags; uint32_t flags;
if (p->quit) { if (multifd_recv_should_exit()) {
break; break;
} }
@ -1274,6 +1281,7 @@ int multifd_recv_setup(Error **errp)
multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state)); multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count); multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
qatomic_set(&multifd_recv_state->count, 0); qatomic_set(&multifd_recv_state->count, 0);
qatomic_set(&multifd_recv_state->exiting, 0);
qemu_sem_init(&multifd_recv_state->sem_sync, 0); qemu_sem_init(&multifd_recv_state->sem_sync, 0);
multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()]; multifd_recv_state->ops = multifd_ops[migrate_multifd_compression()];
@ -1282,7 +1290,6 @@ int multifd_recv_setup(Error **errp)
qemu_mutex_init(&p->mutex); qemu_mutex_init(&p->mutex);
qemu_sem_init(&p->sem_sync, 0); qemu_sem_init(&p->sem_sync, 0);
p->quit = false;
p->id = i; p->id = i;
p->packet_len = sizeof(MultiFDPacket_t) p->packet_len = sizeof(MultiFDPacket_t)
+ sizeof(uint64_t) * page_count; + sizeof(uint64_t) * page_count;