Creation of the threads, nothing inside yet.
Signed-off-by: Juan Quintela <quint...@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilb...@redhat.com>
--
Use pointers instead of long array names
Move to use semaphores instead of conditions as paolo suggestion
---
include/migration/migration.h | 4 +
migration/migration.c | 6 ++
migration/ram.c | 168 ++++++++++++++++++++++++++++++++++++++++++
3 files changed, 178 insertions(+)
diff --git a/include/migration/migration.h b/include/migration/migration.h
index 3c7f165..13fac75 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -256,6 +256,10 @@ MigrationState *migrate_get_current(void);
int migrate_multifd_threads(void);
int migrate_multifd_group(void);
+void migrate_multifd_send_threads_create(void);
+void migrate_multifd_send_threads_join(void);
+void migrate_multifd_recv_threads_create(void);
+void migrate_multifd_recv_threads_join(void);
void migrate_compress_threads_create(void);
void migrate_compress_threads_join(void);
diff --git a/migration/migration.c b/migration/migration.c
index 2b2d0a8..d2705d7 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -344,6 +344,7 @@ static void process_incoming_migration_bh(void *opaque)
MIGRATION_STATUS_FAILED);
error_report_err(local_err);
migrate_decompress_threads_join();
+ migrate_multifd_recv_threads_join();
exit(EXIT_FAILURE);
}
@@ -368,6 +369,7 @@ static void process_incoming_migration_bh(void *opaque)
runstate_set(global_state_get_runstate());
}
migrate_decompress_threads_join();
+ migrate_multifd_recv_threads_join();
/*
* This must happen after any state changes since as soon as an external
* observer sees this event they might start to prod at the VM assuming
@@ -433,6 +435,7 @@ static void process_incoming_migration_co(void *opaque)
MIGRATION_STATUS_FAILED);
error_report("load of migration failed: %s", strerror(-ret));
migrate_decompress_threads_join();
+ migrate_multifd_recv_threads_join();
exit(EXIT_FAILURE);
}
@@ -445,6 +448,7 @@ void migration_fd_process_incoming(QEMUFile *f)
Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
migrate_decompress_threads_create();
+ migrate_multifd_recv_threads_create();
qemu_file_set_blocking(f, false);
qemu_coroutine_enter(co);
}
@@ -974,6 +978,7 @@ static void migrate_fd_cleanup(void *opaque)
qemu_mutex_lock_iothread();
migrate_compress_threads_join();
+ migrate_multifd_send_threads_join();
qemu_fclose(s->to_dst_file);
s->to_dst_file = NULL;
}
@@ -2100,6 +2105,7 @@ void migrate_fd_connect(MigrationState *s)
}
migrate_compress_threads_create();
+ migrate_multifd_send_threads_create();
qemu_thread_create(&s->thread, "live_migration", migration_thread, s,
QEMU_THREAD_JOINABLE);
s->migration_thread_running = true;
diff --git a/migration/ram.c b/migration/ram.c
index 4422ee8..0cb19cf 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -382,6 +382,174 @@ void migrate_compress_threads_create(void)
}
}
+/* Multiple fd's */
+
+struct MultiFDSendParams {
+ QemuThread thread;
+ QemuSemaphore sem;
+ QemuMutex mutex;
+ bool quit;
+};
+typedef struct MultiFDSendParams MultiFDSendParams;
+
+static MultiFDSendParams *multifd_send;
+
+static void *multifd_send_thread(void *opaque)
+{
+ MultiFDSendParams *params = opaque;
+
+ while (true) {
+ qemu_mutex_lock(¶ms->mutex);
+ if (params->quit) {
+ qemu_mutex_unlock(¶ms->mutex);
+ break;
+ }
+ qemu_mutex_unlock(¶ms->mutex);
+ qemu_sem_wait(¶ms->sem);
+ }
+
+ return NULL;
+}
+
+static void terminate_multifd_send_threads(void)
+{
+ int i, thread_count;
+
+ thread_count = migrate_multifd_threads();
+ for (i = 0; i < thread_count; i++) {
+ MultiFDSendParams *p = &multifd_send[i];
+
+ qemu_mutex_lock(&p->mutex);
+ p->quit = true;
+ qemu_sem_post(&p->sem);
+ qemu_mutex_unlock(&p->mutex);
+ }
+}
+
+void migrate_multifd_send_threads_join(void)
+{
+ int i, thread_count;
+
+ if (!migrate_use_multifd()) {
+ return;
+ }
+ terminate_multifd_send_threads();
+ thread_count = migrate_multifd_threads();
+ for (i = 0; i < thread_count; i++) {
+ MultiFDSendParams *p = &multifd_send[i];
+
+ qemu_thread_join(&p->thread);
+ qemu_mutex_destroy(&p->mutex);
+ qemu_sem_destroy(&p->sem);
+ }
+ g_free(multifd_send);
+ multifd_send = NULL;
+}
+
+void migrate_multifd_send_threads_create(void)
+{
+ int i, thread_count;
+
+ if (!migrate_use_multifd()) {
+ return;
+ }
+ thread_count = migrate_multifd_threads();
+ multifd_send = g_new0(MultiFDSendParams, thread_count);
+ for (i = 0; i < thread_count; i++) {
+ char thread_name[15];
+ MultiFDSendParams *p = &multifd_send[i];
+
+ qemu_mutex_init(&p->mutex);
+ qemu_sem_init(&p->sem, 0);
+ p->quit = false;
+ snprintf(thread_name, 15, "multifd_send_%d", i);
+ qemu_thread_create(&p->thread, thread_name, multifd_send_thread, p,
+ QEMU_THREAD_JOINABLE);
+ }
+}
+
+struct MultiFDRecvParams {
+ QemuThread thread;
+ QemuSemaphore sem;
+ QemuMutex mutex;
+ bool quit;
+};
+typedef struct MultiFDRecvParams MultiFDRecvParams;
+
+static MultiFDRecvParams *multifd_recv;
+
+static void *multifd_recv_thread(void *opaque)
+{
+ MultiFDRecvParams *params = opaque;
+
+ while (true) {
+ qemu_mutex_lock(¶ms->mutex);
+ if (params->quit) {
+ qemu_mutex_unlock(¶ms->mutex);
+ break;
+ }
+ qemu_mutex_unlock(¶ms->mutex);
+ qemu_sem_wait(¶ms->sem);
+ }
+
+ return NULL;
+}
+
+static void terminate_multifd_recv_threads(void)
+{
+ int i, thread_count;
+
+ thread_count = migrate_multifd_threads();
+ for (i = 0; i < thread_count; i++) {
+ MultiFDRecvParams *p = &multifd_recv[i];
+
+ qemu_mutex_lock(&p->mutex);
+ p->quit = true;
+ qemu_sem_post(&p->sem);
+ qemu_mutex_unlock(&p->mutex);
+ }
+}
+
+void migrate_multifd_recv_threads_join(void)
+{
+ int i, thread_count;
+
+ if (!migrate_use_multifd()) {
+ return;
+ }
+ terminate_multifd_recv_threads();
+ thread_count = migrate_multifd_threads();
+ for (i = 0; i < thread_count; i++) {
+ MultiFDRecvParams *p = &multifd_recv[i];
+
+ qemu_thread_join(&p->thread);
+ qemu_mutex_destroy(&p->mutex);
+ qemu_sem_destroy(&p->sem);
+ }
+ g_free(multifd_recv);
+ multifd_recv = NULL;
+}
+
+void migrate_multifd_recv_threads_create(void)
+{
+ int i, thread_count;
+
+ if (!migrate_use_multifd()) {
+ return;
+ }
+ thread_count = migrate_multifd_threads();
+ multifd_recv = g_new0(MultiFDRecvParams, thread_count);
+ for (i = 0; i < thread_count; i++) {
+ MultiFDRecvParams *p = &multifd_recv[i];
+
+ qemu_mutex_init(&p->mutex);
+ qemu_sem_init(&p->sem, 0);
+ p->quit = false;
+ qemu_thread_create(&p->thread, "multifd_recv", multifd_recv_thread, p,
+ QEMU_THREAD_JOINABLE);
+ }
+}
+
/**
* save_page_header: Write page header to wire
*
--
2.7.4
