[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[Qemu-devel] [PATCH v8 16/20] migration: Create thread infrastructure fo
From: |
Juan Quintela |
Subject: |
[Qemu-devel] [PATCH v8 16/20] migration: Create thread infrastructure for multifd recv side |
Date: |
Wed, 13 Sep 2017 12:59:49 +0200 |
We make the locking and the transfer of information specific, even if we
are still receiving things through the main thread.
Signed-off-by: Juan Quintela <address@hidden>
--
We split when we create the main channel and where we start the main
migration thread, so we wait for the creation of the other threads.
Use multifd_clear_pages().
Don't remove object_unref()
---
migration/migration.c | 7 ++++---
migration/migration.h | 1 +
migration/ram.c | 55 +++++++++++++++++++++++++++++++++++++++++++++++----
migration/socket.c | 2 +-
4 files changed, 57 insertions(+), 8 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index 085177ca26..674c254c3f 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -388,7 +388,7 @@ static void migration_incoming_setup(QEMUFile *f)
qemu_file_set_blocking(f, false);
}
-static void migration_incoming_process(void)
+void migration_incoming_process(void)
{
Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
qemu_coroutine_enter(co);
@@ -406,9 +406,10 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
if (!mis->from_src_file) {
QEMUFile *f = qemu_fopen_channel_input(ioc);
- migration_fd_process_incoming(f);
+ migration_incoming_setup(f);
+ return;
}
- /* We still only have a single channel. Nothing to do here yet */
+ multifd_new_channel(ioc);
}
/**
diff --git a/migration/migration.h b/migration/migration.h
index da39d4c711..382a0449d5 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -158,6 +158,7 @@ void migrate_set_state(int *state, int old_state, int
new_state);
void migration_fd_process_incoming(QEMUFile *f);
void migration_ioc_process_incoming(QIOChannel *ioc);
+void migration_incoming_process(void);
bool migration_has_all_channels(void);
diff --git a/migration/ram.c b/migration/ram.c
index 3f60f7e40c..d6aa516f6d 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -601,13 +601,18 @@ static uint16_t multifd_send_page(uint8_t *address, bool
last_page)
}
struct MultiFDRecvParams {
+ /* not changed */
uint8_t id;
char *name;
QemuThread thread;
QIOChannel *c;
+ QemuSemaphore ready;
QemuSemaphore sem;
QemuMutex mutex;
+ /* proteced by param mutex */
bool quit;
+ multifd_pages_t pages;
+ bool done;
};
typedef struct MultiFDRecvParams MultiFDRecvParams;
@@ -617,6 +622,7 @@ struct {
int count;
/* Should we finish */
bool quit;
+ multifd_pages_t pages;
} *multifd_recv_state;
static void terminate_multifd_recv_threads(Error *errp)
@@ -638,6 +644,7 @@ static void terminate_multifd_recv_threads(Error *errp)
p->quit = true;
qemu_sem_post(&p->sem);
qemu_mutex_unlock(&p->mutex);
+ multifd_clear_pages(&p->pages);
}
}
@@ -662,6 +669,7 @@ int multifd_load_cleanup(Error **errp)
}
g_free(multifd_recv_state->params);
multifd_recv_state->params = NULL;
+ multifd_clear_pages(&multifd_recv_state->pages);
g_free(multifd_recv_state);
multifd_recv_state = NULL;
@@ -672,12 +680,20 @@ static void *multifd_recv_thread(void *opaque)
{
MultiFDRecvParams *p = opaque;
+ qemu_sem_post(&p->ready);
while (true) {
qemu_mutex_lock(&p->mutex);
if (p->quit) {
qemu_mutex_unlock(&p->mutex);
break;
}
+ if (p->pages.num) {
+ p->pages.num = 0;
+ p->done = true;
+ qemu_mutex_unlock(&p->mutex);
+ qemu_sem_post(&p->ready);
+ continue;
+ }
qemu_mutex_unlock(&p->mutex);
qemu_sem_wait(&p->sem);
}
@@ -723,8 +739,11 @@ void multifd_new_channel(QIOChannel *ioc)
}
qemu_mutex_init(&p->mutex);
qemu_sem_init(&p->sem, 0);
+ qemu_sem_init(&p->ready, 0);
p->quit = false;
p->id = id;
+ p->done = false;
+ multifd_init_pages(&p->pages);
p->c = ioc;
multifd_recv_state->count++;
p->name = g_strdup_printf("multifdrecv_%d", id);
@@ -744,6 +763,7 @@ int multifd_load_setup(void)
multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
multifd_recv_state->count = 0;
multifd_recv_state->quit = false;
+ multifd_init_pages(&multifd_recv_state->pages);
return 0;
}
@@ -752,6 +772,36 @@ int multifd_created_channels(void)
return multifd_recv_state->count;
}
+static void multifd_recv_page(uint8_t *address, uint16_t fd_num)
+{
+ int thread_count;
+ MultiFDRecvParams *p;
+ multifd_pages_t *pages = &multifd_recv_state->pages;
+
+ pages->iov[pages->num].iov_base = address;
+ pages->iov[pages->num].iov_len = TARGET_PAGE_SIZE;
+ pages->num++;
+
+ if (fd_num == MULTIFD_CONTINUE) {
+ return;
+ }
+
+ thread_count = migrate_multifd_channels();
+ assert(fd_num < thread_count);
+ p = &multifd_recv_state->params[fd_num];
+
+ qemu_sem_wait(&p->ready);
+
+ qemu_mutex_lock(&p->mutex);
+ p->done = false;
+ iov_copy(p->pages.iov, pages->num, pages->iov, pages->num, 0,
+ iov_size(pages->iov, pages->num));
+ p->pages.num = pages->num;
+ pages->num = 0;
+ qemu_mutex_unlock(&p->mutex);
+ qemu_sem_post(&p->sem);
+}
+
/**
* save_page_header: write page header to wire
*
@@ -3063,10 +3113,7 @@ static int ram_load(QEMUFile *f, void *opaque, int
version_id)
case RAM_SAVE_FLAG_MULTIFD_PAGE:
fd_num = qemu_get_be16(f);
- if (fd_num != 0) {
- /* this is yet an unused variable, changed later */
- fd_num = fd_num;
- }
+ multifd_recv_page(host, fd_num);
qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
break;
diff --git a/migration/socket.c b/migration/socket.c
index 22fb05edc8..63dedb22e8 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -180,12 +180,12 @@ static gboolean
socket_accept_incoming_migration(QIOChannel *ioc,
qio_channel_set_name(QIO_CHANNEL(sioc), "migration-socket-incoming");
migration_channel_process_incoming(QIO_CHANNEL(sioc));
- object_unref(OBJECT(sioc));
out:
if (migration_has_all_channels()) {
/* Close listening socket as its no longer needed */
qio_channel_close(ioc, NULL);
+ migration_incoming_process();
return G_SOURCE_REMOVE;
} else {
return G_SOURCE_CONTINUE;
--
2.13.5
- [Qemu-devel] [PATCH v8 11/20] migration: Split migration_fd_process_incoming, (continued)
- [Qemu-devel] [PATCH v8 11/20] migration: Split migration_fd_process_incoming, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 10/20] migration: Create multifd migration threads, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 12/20] migration: Start of multiple fd work, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 13/20] migration: Create ram_multifd_page, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 09/20] migration: Create x-multifd-page-count parameter, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 14/20] migration: Really use multiple pages at a time, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 15/20] migration: Send the fd number which we are going to use for this page, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 17/20] migration: Test new fd infrastructure, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 16/20] migration: Create thread infrastructure for multifd recv side,
Juan Quintela <=
- [Qemu-devel] [PATCH v8 19/20] migration: Transfer pages over new channels, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 20/20] migration: Flush receive queue, Juan Quintela, 2017/09/13
- [Qemu-devel] [PATCH v8 18/20] migration: Rename initial_bytes, Juan Quintela, 2017/09/13