* [Qemu-devel] [PATCH v6 00/19] Multifd
@ 2017-08-08 16:26 Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming() Juan Quintela
` (19 more replies)
0 siblings, 20 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Hi
Things NOT done yet:
- still uses error_abort/fail. I need to redo all the error pages to
be able to return with errors from other threads.
- still connects synchronusly. I need to redo the other changes to fix this.
- have to change the tests as danp wanted, I still don't undertand them.
Done:
- Improve migration_ioc_porcess_incoming
- teach about G_SOURCE_REMOVE/CONTINUE
- Add test for migration_has_all_channels
- use DEFIN_PROP*
- change recv_state to use pointers to parameters
make easier to receive channels out of order
- use g_strdup_printf()
- improve count of threads to know when we have to finish
- report channel id's on errors
- Use last_page parameter for multifd_send_page() sooner
- Improve commets for address
- use g_new0() instead of g_malloc()
- create MULTIFD_CONTINUE instead of using UINT16_MAX
- clear memory used by group of pages
once there, pass everything to the global state variables instead of being
local to the function. This way it works if we cancel migration and start
a new one
- Really wait to create the migration_thread until all channels are created
- split initial_bytes setup to make clearer following patches.
- createRAM_SAVE_FLAG_MULTIFD_SYNC macro, to make clear what we are doing
- move setting of need_flush to inside bitmap_sync
- Lots of other small changes & reorderings
Please, comment.
[v5]
- tests from qio functions (a.k.a. make danp happy)
- 1st message from one channel to the other contains:
<uuid> multifd <channel number>
This would allow us to create more channels as we want them.
a.k.a. Making dave happy
- Waiting in reception for new channels using qio listeners
Getting threads, qio and reference counters working at the same time
was interesing.
Another make danp happy.
- Lots and lots of small changes and fixes. Notice that the last 70 patches
that I merged or so what to make this series easier/smaller.
- NOT DONE: I haven't been woring on measuring performance
differences, this was about getting the creation of the
threads/channels right.
So, what I want:
- Are people happy with how I have (ab)used qio channels? (yes danp,
that is you).
- My understanding is th
ToDo:
- Make paolo happy: He wanted to test using control information
through each channel, not only pages. This requires yet more
cleanups to be able to have more than one QEMUFile/RAMState open at
the same time.
- How I create multiple channels. Things I know:
* with current changes, it should work with fd/channels (the multifd bits),
but we don;t have a way to pass multiple fd;s or exec files.
Danp, any idea about how to create an UI for it?
* My idea is that we would split current code to be:
+ channel creation at migration.c
+ rest of bits at ram.c
+ change format to:
<uuid> main <rest of migration capabilities/paramentes> so we can check
<uuid> postcopy <no clue what parameters are needed>
Dave wanted a way to create a new fd for postcopy for some time
+ Adding new channels is easy
- Performance data/numbers: Yes, I wanted to get this out at once, I
would continue with this.
Please, review.
[v4]
This is the 4th version of multifd. Changes:
- XBZRLE don't need to be checked for
- Documentation and defaults are consistent
- split socketArgs
- use iovec instead of creating something similar.
- We use now the exported size of target page (another HACK removal)
- created qio_chanel_{wirtev,readv}_all functions. the _full() name
was already taken.
What they do is the same that the without _all() function, but if it
returns due to blocking it redo the call.
- it is checkpatch.pl clean now.
Please comment, Juan.
Juan Quintela (19):
migration: Create migration_ioc_process_incoming()
migration: Teach it about G_SOURCE_REMOVE
migration: Add comments to channel functions
migration: Create migration_has_all_channels
qio: Create new qio_channel_{readv,writev}_all
migration: Add multifd capability
migration: Create x-multifd-threads parameter
migration: Create x-multifd-group parameter
migration: Create multifd migration threads
migration: Split migration_fd_process_incoming
migration: Start of multiple fd work
migration: Create ram_multifd_page
migration: Really use multiple pages at a time
migration: Send the fd number which we are going to use for this page
migration: Create thread infrastructure for multifd recv side
migration: Test new fd infrastructure
migration: Rename initial_bytes
migration: Transfer pages over new channels
migration: Flush receive queue
hmp.c | 16 ++
include/glib-compat.h | 2 +
include/io/channel.h | 46 ++++
io/channel.c | 77 ++++++
migration/channel.c | 18 +-
migration/exec.c | 2 +-
migration/fd.c | 2 +-
migration/migration.c | 143 ++++++++++-
migration/migration.h | 8 +
migration/qemu-file-channel.c | 29 +--
migration/ram.c | 537 ++++++++++++++++++++++++++++++++++++++++-
migration/ram.h | 8 +
migration/socket.c | 48 +++-
migration/socket.h | 10 +
qapi-schema.json | 44 +++-
tests/io-channel-helpers.c | 55 +++++
tests/io-channel-helpers.h | 4 +
tests/test-io-channel-buffer.c | 55 ++++-
18 files changed, 1049 insertions(+), 55 deletions(-)
--
2.9.4
^ permalink raw reply [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming()
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 3:47 ` Peter Xu
2017-08-11 14:50 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE Juan Quintela
` (18 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We pass the ioc instead of the fd. This will allow us to have more
than one channel open. We also make sure that we set the
from_src_file sooner, so we don't need to pass it as a parameter.
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
migration/channel.c | 3 +--
migration/migration.c | 25 ++++++++++++++++++++-----
migration/migration.h | 2 ++
3 files changed, 23 insertions(+), 7 deletions(-)
diff --git a/migration/channel.c b/migration/channel.c
index 3b7252f..edceebd 100644
--- a/migration/channel.c
+++ b/migration/channel.c
@@ -36,8 +36,7 @@ void migration_channel_process_incoming(QIOChannel *ioc)
error_report_err(local_err);
}
} else {
- QEMUFile *f = qemu_fopen_channel_input(ioc);
- migration_fd_process_incoming(f);
+ migration_ioc_process_incoming(ioc);
}
}
diff --git a/migration/migration.c b/migration/migration.c
index c3fe0ed..86e41e8 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -307,17 +307,16 @@ static void process_incoming_migration_bh(void *opaque)
static void process_incoming_migration_co(void *opaque)
{
- QEMUFile *f = opaque;
MigrationIncomingState *mis = migration_incoming_get_current();
PostcopyState ps;
int ret;
- mis->from_src_file = f;
+ assert(mis->from_src_file);
mis->largest_page_size = qemu_ram_pagesize_largest();
postcopy_state_set(POSTCOPY_INCOMING_NONE);
migrate_set_state(&mis->state, MIGRATION_STATUS_NONE,
MIGRATION_STATUS_ACTIVE);
- ret = qemu_loadvm_state(f);
+ ret = qemu_loadvm_state(mis->from_src_file);
ps = postcopy_state_get();
trace_process_incoming_migration_co_end(ret, ps);
@@ -365,12 +364,28 @@ static void process_incoming_migration_co(void *opaque)
void migration_fd_process_incoming(QEMUFile *f)
{
- Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
-
+ Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
+ MigrationIncomingState *mis = migration_incoming_get_current();
+
+ if (!mis->from_src_file) {
+ mis->from_src_file = f;
+ }
qemu_file_set_blocking(f, false);
qemu_coroutine_enter(co);
}
+void migration_ioc_process_incoming(QIOChannel *ioc)
+{
+ MigrationIncomingState *mis = migration_incoming_get_current();
+
+ if (!mis->from_src_file) {
+ QEMUFile *f = qemu_fopen_channel_input(ioc);
+ mis->from_src_file = f;
+ migration_fd_process_incoming(f);
+ }
+ /* We still only have a single channel. Nothing to do here yet */
+}
+
/*
* Send a 'SHUT' message on the return channel with the given value
* to indicate that we've finished with the RP. Non-0 value indicates
diff --git a/migration/migration.h b/migration/migration.h
index 148c9fa..99c398d 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -20,6 +20,7 @@
#include "exec/cpu-common.h"
#include "qemu/coroutine_int.h"
#include "hw/qdev.h"
+#include "io/channel.h"
/* State for the incoming migration */
struct MigrationIncomingState {
@@ -152,6 +153,7 @@ struct MigrationState
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);
uint64_t migrate_max_downtime(void);
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming() Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-10 14:56 ` Daniel P. Berrange
2017-08-11 3:49 ` Peter Xu
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 03/19] migration: Add comments to channel functions Juan Quintela
` (17 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
As this is defined on glib 2.32, add compatibility macros for older glibs.
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
include/glib-compat.h | 2 ++
migration/exec.c | 2 +-
migration/fd.c | 2 +-
migration/socket.c | 2 +-
4 files changed, 5 insertions(+), 3 deletions(-)
diff --git a/include/glib-compat.h b/include/glib-compat.h
index fcffcd3..e15aca2 100644
--- a/include/glib-compat.h
+++ b/include/glib-compat.h
@@ -223,6 +223,8 @@ static inline gboolean g_hash_table_contains(GHashTable *hash_table,
{
return g_hash_table_lookup_extended(hash_table, key, NULL, NULL);
}
+#define G_SOURCE_CONTINUE TRUE
+#define G_SOURCE_REMOVE FALSE
#endif
#ifndef g_assert_true
diff --git a/migration/exec.c b/migration/exec.c
index 08b599e..f3be1ba 100644
--- a/migration/exec.c
+++ b/migration/exec.c
@@ -49,7 +49,7 @@ static gboolean exec_accept_incoming_migration(QIOChannel *ioc,
{
migration_channel_process_incoming(ioc);
object_unref(OBJECT(ioc));
- return FALSE; /* unregister */
+ return G_SOURCE_REMOVE;
}
void exec_start_incoming_migration(const char *command, Error **errp)
diff --git a/migration/fd.c b/migration/fd.c
index 30f5258..30de4b9 100644
--- a/migration/fd.c
+++ b/migration/fd.c
@@ -49,7 +49,7 @@ static gboolean fd_accept_incoming_migration(QIOChannel *ioc,
{
migration_channel_process_incoming(ioc);
object_unref(OBJECT(ioc));
- return FALSE; /* unregister */
+ return G_SOURCE_REMOVE;
}
void fd_start_incoming_migration(const char *infd, Error **errp)
diff --git a/migration/socket.c b/migration/socket.c
index 757d382..b02d37d 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -154,7 +154,7 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
out:
/* Close listening socket as its no longer needed */
qio_channel_close(ioc, NULL);
- return FALSE; /* unregister */
+ return G_SOURCE_REMOVE;
}
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 03/19] migration: Add comments to channel functions
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming() Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-10 14:57 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels Juan Quintela
` (16 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
migration/channel.c | 15 +++++++++++++++
1 file changed, 15 insertions(+)
diff --git a/migration/channel.c b/migration/channel.c
index edceebd..79b0f8b 100644
--- a/migration/channel.c
+++ b/migration/channel.c
@@ -19,6 +19,14 @@
#include "qapi/error.h"
#include "io/channel-tls.h"
+/**
+ * @migration_channel_process_incoming - Create new incoming migration channel
+ *
+ * Notice that TLS is special. For it we listen in a listener socket,
+ * and then create a new client socket from the TLS library.
+ *
+ * @ioc: Channel to wich we are connecting
+ */
void migration_channel_process_incoming(QIOChannel *ioc)
{
MigrationState *s = migrate_get_current();
@@ -41,6 +49,13 @@ void migration_channel_process_incoming(QIOChannel *ioc)
}
+/**
+ * @migration_channel_connect - Create new outgoing migration channel
+ *
+ * @s: Current migration state
+ * @ioc: Channel to wich we are connecting
+ * @hostname: Where we want to connect
+ */
void migration_channel_connect(MigrationState *s,
QIOChannel *ioc,
const char *hostname)
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (2 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 03/19] migration: Add comments to channel functions Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-10 14:58 ` Daniel P. Berrange
2017-08-10 15:17 ` Eric Blake
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 05/19] qio: Create new qio_channel_{readv, writev}_all Juan Quintela
` (15 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
This functions allows us to decide when to close the listener socket.
For now, we only need one connection.
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
migration/migration.c | 11 +++++++++++
migration/migration.h | 2 ++
migration/socket.c | 10 +++++++---
3 files changed, 20 insertions(+), 3 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index 86e41e8..2c42834 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -386,6 +386,17 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
/* We still only have a single channel. Nothing to do here yet */
}
+/**
+ * @migration_has_all_channels: We have received all channels that we need
+ *
+ * Returns true when we have got connections to all the channels that
+ * we need for migration.
+ */
+bool migration_has_all_channels(void)
+{
+ return true;
+}
+
/*
* Send a 'SHUT' message on the return channel with the given value
* to indicate that we've finished with the RP. Non-0 value indicates
diff --git a/migration/migration.h b/migration/migration.h
index 99c398d..1881e4a 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -155,6 +155,8 @@ 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);
+bool migration_has_all_channels(void);
+
uint64_t migrate_max_downtime(void);
void migrate_fd_error(MigrationState *s, const Error *error);
diff --git a/migration/socket.c b/migration/socket.c
index b02d37d..dee8690 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -152,9 +152,13 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
object_unref(OBJECT(sioc));
out:
- /* Close listening socket as its no longer needed */
- qio_channel_close(ioc, NULL);
- return G_SOURCE_REMOVE;
+ if (migration_has_all_channels()) {
+ /* Close listening socket as its no longer needed */
+ qio_channel_close(ioc, NULL);
+ return G_SOURCE_REMOVE;
+ } else {
+ return G_SOURCE_CONTINUE;
+ }
}
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 05/19] qio: Create new qio_channel_{readv, writev}_all
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (3 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-10 15:04 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability Juan Quintela
` (14 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
The functions waits until it is able to write the full iov.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Add tests.
fix reader to check for len == 0.
---
include/io/channel.h | 46 +++++++++++++++++++++++++
io/channel.c | 77 ++++++++++++++++++++++++++++++++++++++++++
migration/qemu-file-channel.c | 29 +---------------
tests/io-channel-helpers.c | 55 ++++++++++++++++++++++++++++++
tests/io-channel-helpers.h | 4 +++
tests/test-io-channel-buffer.c | 55 ++++++++++++++++++++++++++++--
6 files changed, 235 insertions(+), 31 deletions(-)
diff --git a/include/io/channel.h b/include/io/channel.h
index db9bb02..bfc97e2 100644
--- a/include/io/channel.h
+++ b/include/io/channel.h
@@ -269,6 +269,52 @@ ssize_t qio_channel_writev_full(QIOChannel *ioc,
Error **errp);
/**
+ * qio_channel_readv_all:
+ * @ioc: the channel object
+ * @iov: the array of memory regions to read data into
+ * @niov: the length of the @iov array
+ * @errp: pointer to a NULL-initialized error object
+ *
+ * Read data from the IO channel, storing it in the
+ * memory regions referenced by @iov. Each element
+ * in the @iov will be fully populated with data
+ * before the next one is used. The @niov parameter
+ * specifies the total number of elements in @iov.
+ *
+ * Returns: the number of bytes read, or -1 on error,
+ * or QIO_CHANNEL_ERR_BLOCK if no data is available
+ * and the channel is non-blocking
+ */
+ssize_t qio_channel_readv_all(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ Error **errp);
+
+
+/**
+ * qio_channel_writev_all:
+ * @ioc: the channel object
+ * @iov: the array of memory regions to write data from
+ * @niov: the length of the @iov array
+ * @errp: pointer to a NULL-initialized error object
+ *
+ * Write data to the IO channel, reading it from the
+ * memory regions referenced by @iov. Each element
+ * in the @iov will be fully sent, before the next
+ * one is used. The @niov parameter specifies the
+ * total number of elements in @iov.
+ *
+ * It is required for all @iov data to be fully
+ * sent.
+ *
+ * Returns: the number of bytes sent, or -1 on error,
+ */
+ssize_t qio_channel_writev_all(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ Error **erp);
+
+/**
* qio_channel_readv:
* @ioc: the channel object
* @iov: the array of memory regions to read data into
diff --git a/io/channel.c b/io/channel.c
index 1cfb8b3..0b521f9 100644
--- a/io/channel.c
+++ b/io/channel.c
@@ -22,6 +22,7 @@
#include "io/channel.h"
#include "qapi/error.h"
#include "qemu/main-loop.h"
+#include "qemu/iov.h"
bool qio_channel_has_feature(QIOChannel *ioc,
QIOChannelFeature feature)
@@ -85,6 +86,82 @@ ssize_t qio_channel_writev_full(QIOChannel *ioc,
}
+
+ssize_t qio_channel_readv_all(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ Error **errp)
+{
+ ssize_t done = 0;
+ struct iovec *local_iov = g_new(struct iovec, niov);
+ struct iovec *local_iov_head = local_iov;
+ unsigned int nlocal_iov = niov;
+
+ nlocal_iov = iov_copy(local_iov, nlocal_iov,
+ iov, niov,
+ 0, iov_size(iov, niov));
+
+ while (nlocal_iov > 0) {
+ ssize_t len;
+ len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
+ if (len == QIO_CHANNEL_ERR_BLOCK) {
+ qio_channel_wait(ioc, G_IO_OUT);
+ continue;
+ } else if (len < 0) {
+ error_setg_errno(errp, EIO,
+ "Channel was not able to read full iov");
+ done = -1;
+ goto cleanup;
+ } else if (len == 0) {
+ goto cleanup;
+ }
+
+ iov_discard_front(&local_iov, &nlocal_iov, len);
+ done += len;
+ }
+
+ cleanup:
+ g_free(local_iov_head);
+ return done;
+}
+
+ssize_t qio_channel_writev_all(QIOChannel *ioc,
+ const struct iovec *iov,
+ size_t niov,
+ Error **errp)
+{
+ ssize_t done = 0;
+ struct iovec *local_iov = g_new(struct iovec, niov);
+ struct iovec *local_iov_head = local_iov;
+ unsigned int nlocal_iov = niov;
+
+ nlocal_iov = iov_copy(local_iov, nlocal_iov,
+ iov, niov,
+ 0, iov_size(iov, niov));
+
+ while (nlocal_iov > 0) {
+ ssize_t len;
+ len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);
+ if (len == QIO_CHANNEL_ERR_BLOCK) {
+ qio_channel_wait(ioc, G_IO_OUT);
+ continue;
+ }
+ if (len < 0) {
+ error_setg_errno(errp, EIO,
+ "Channel was not able to write full iov");
+ done = -1;
+ goto cleanup;
+ }
+
+ iov_discard_front(&local_iov, &nlocal_iov, len);
+ done += len;
+ }
+
+ cleanup:
+ g_free(local_iov_head);
+ return done;
+}
+
ssize_t qio_channel_readv(QIOChannel *ioc,
const struct iovec *iov,
size_t niov,
diff --git a/migration/qemu-file-channel.c b/migration/qemu-file-channel.c
index e202d73..457ea6c 100644
--- a/migration/qemu-file-channel.c
+++ b/migration/qemu-file-channel.c
@@ -36,35 +36,8 @@ static ssize_t channel_writev_buffer(void *opaque,
int64_t pos)
{
QIOChannel *ioc = QIO_CHANNEL(opaque);
- ssize_t done = 0;
- struct iovec *local_iov = g_new(struct iovec, iovcnt);
- struct iovec *local_iov_head = local_iov;
- unsigned int nlocal_iov = iovcnt;
- nlocal_iov = iov_copy(local_iov, nlocal_iov,
- iov, iovcnt,
- 0, iov_size(iov, iovcnt));
-
- while (nlocal_iov > 0) {
- ssize_t len;
- len = qio_channel_writev(ioc, local_iov, nlocal_iov, NULL);
- if (len == QIO_CHANNEL_ERR_BLOCK) {
- qio_channel_wait(ioc, G_IO_OUT);
- continue;
- }
- if (len < 0) {
- /* XXX handle Error objects */
- done = -EIO;
- goto cleanup;
- }
-
- iov_discard_front(&local_iov, &nlocal_iov, len);
- done += len;
- }
-
- cleanup:
- g_free(local_iov_head);
- return done;
+ return qio_channel_writev_all(ioc, iov, iovcnt, NULL);
}
diff --git a/tests/io-channel-helpers.c b/tests/io-channel-helpers.c
index 05e5579..3d76d95 100644
--- a/tests/io-channel-helpers.c
+++ b/tests/io-channel-helpers.c
@@ -21,6 +21,7 @@
#include "qemu/osdep.h"
#include "io-channel-helpers.h"
#include "qapi/error.h"
+#include "qemu/iov.h"
struct QIOChannelTest {
QIOChannel *src;
@@ -153,6 +154,45 @@ static gpointer test_io_thread_reader(gpointer opaque)
return NULL;
}
+static gpointer test_io_thread_writer_all(gpointer opaque)
+{
+ QIOChannelTest *data = opaque;
+ size_t niov = data->niov;
+ ssize_t ret;
+
+ qio_channel_set_blocking(data->src, data->blocking, NULL);
+
+ ret = qio_channel_writev_all(data->src,
+ data->inputv,
+ niov,
+ &data->writeerr);
+ if (ret != iov_size(data->inputv, data->niov)) {
+ error_setg(&data->writeerr, "Unexpected I/O error");
+ }
+
+ return NULL;
+}
+
+/* This thread receives all data using iovecs */
+static gpointer test_io_thread_reader_all(gpointer opaque)
+{
+ QIOChannelTest *data = opaque;
+ size_t niov = data->niov;
+ ssize_t ret;
+
+ qio_channel_set_blocking(data->dst, data->blocking, NULL);
+
+ ret = qio_channel_readv_all(data->dst,
+ data->outputv,
+ niov,
+ &data->readerr);
+
+ if (ret != iov_size(data->inputv, data->niov)) {
+ error_setg(&data->readerr, "Unexpected I/O error");
+ }
+
+ return NULL;
+}
QIOChannelTest *qio_channel_test_new(void)
{
@@ -231,6 +271,21 @@ void qio_channel_test_run_reader(QIOChannelTest *test,
test->dst = NULL;
}
+void qio_channel_test_run_writer_all(QIOChannelTest *test,
+ QIOChannel *src)
+{
+ test->src = src;
+ test_io_thread_writer_all(test);
+ test->src = NULL;
+}
+
+void qio_channel_test_run_reader_all(QIOChannelTest *test,
+ QIOChannel *dst)
+{
+ test->dst = dst;
+ test_io_thread_reader_all(test);
+ test->dst = NULL;
+}
void qio_channel_test_validate(QIOChannelTest *test)
{
diff --git a/tests/io-channel-helpers.h b/tests/io-channel-helpers.h
index fedc64f..17b9647 100644
--- a/tests/io-channel-helpers.h
+++ b/tests/io-channel-helpers.h
@@ -36,6 +36,10 @@ void qio_channel_test_run_writer(QIOChannelTest *test,
QIOChannel *src);
void qio_channel_test_run_reader(QIOChannelTest *test,
QIOChannel *dst);
+void qio_channel_test_run_writer_all(QIOChannelTest *test,
+ QIOChannel *src);
+void qio_channel_test_run_reader_all(QIOChannelTest *test,
+ QIOChannel *dst);
void qio_channel_test_validate(QIOChannelTest *test);
diff --git a/tests/test-io-channel-buffer.c b/tests/test-io-channel-buffer.c
index 64722a2..4bf64ae 100644
--- a/tests/test-io-channel-buffer.c
+++ b/tests/test-io-channel-buffer.c
@@ -22,8 +22,7 @@
#include "io/channel-buffer.h"
#include "io-channel-helpers.h"
-
-static void test_io_channel_buf(void)
+static void test_io_channel_buf1(void)
{
QIOChannelBuffer *buf;
QIOChannelTest *test;
@@ -39,6 +38,53 @@ static void test_io_channel_buf(void)
object_unref(OBJECT(buf));
}
+static void test_io_channel_buf2(void)
+{
+ QIOChannelBuffer *buf;
+ QIOChannelTest *test;
+
+ buf = qio_channel_buffer_new(0);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_writer_all(test, QIO_CHANNEL(buf));
+ buf->offset = 0;
+ qio_channel_test_run_reader(test, QIO_CHANNEL(buf));
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(buf));
+}
+
+static void test_io_channel_buf3(void)
+{
+ QIOChannelBuffer *buf;
+ QIOChannelTest *test;
+
+ buf = qio_channel_buffer_new(0);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_writer(test, QIO_CHANNEL(buf));
+ buf->offset = 0;
+ qio_channel_test_run_reader_all(test, QIO_CHANNEL(buf));
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(buf));
+}
+
+static void test_io_channel_buf4(void)
+{
+ QIOChannelBuffer *buf;
+ QIOChannelTest *test;
+
+ buf = qio_channel_buffer_new(0);
+
+ test = qio_channel_test_new();
+ qio_channel_test_run_writer_all(test, QIO_CHANNEL(buf));
+ buf->offset = 0;
+ qio_channel_test_run_reader_all(test, QIO_CHANNEL(buf));
+ qio_channel_test_validate(test);
+
+ object_unref(OBJECT(buf));
+}
int main(int argc, char **argv)
{
@@ -46,6 +92,9 @@ int main(int argc, char **argv)
g_test_init(&argc, &argv, NULL);
- g_test_add_func("/io/channel/buf", test_io_channel_buf);
+ g_test_add_func("/io/channel/buf1", test_io_channel_buf1);
+ g_test_add_func("/io/channel/buf2", test_io_channel_buf2);
+ g_test_add_func("/io/channel/buf3", test_io_channel_buf3);
+ g_test_add_func("/io/channel/buf4", test_io_channel_buf4);
return g_test_run();
}
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (4 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 05/19] qio: Create new qio_channel_{readv, writev}_all Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 7:14 ` Peter Xu
2017-08-11 14:52 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter Juan Quintela
` (13 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
--
Use new DEFINE_PROP
---
migration/migration.c | 10 ++++++++++
migration/migration.h | 1 +
qapi-schema.json | 3 ++-
3 files changed, 13 insertions(+), 1 deletion(-)
diff --git a/migration/migration.c b/migration/migration.c
index 2c42834..4e6ad87 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -1450,6 +1450,15 @@ bool migrate_use_events(void)
return s->enabled_capabilities[MIGRATION_CAPABILITY_EVENTS];
}
+bool migrate_use_multifd(void)
+{
+ MigrationState *s;
+
+ s = migrate_get_current();
+
+ return s->enabled_capabilities[MIGRATION_CAPABILITY_X_MULTIFD];
+}
+
int migrate_use_xbzrle(void)
{
MigrationState *s;
@@ -2228,6 +2237,7 @@ static Property migration_properties[] = {
DEFINE_PROP_MIG_CAP("x-release-ram", MIGRATION_CAPABILITY_RELEASE_RAM),
DEFINE_PROP_MIG_CAP("x-block", MIGRATION_CAPABILITY_BLOCK),
DEFINE_PROP_MIG_CAP("x-return-path", MIGRATION_CAPABILITY_RETURN_PATH),
+ DEFINE_PROP_MIG_CAP("x-multifd", MIGRATION_CAPABILITY_X_MULTIFD),
DEFINE_PROP_END_OF_LIST(),
};
diff --git a/migration/migration.h b/migration/migration.h
index 1881e4a..b7437f1 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -174,6 +174,7 @@ bool migrate_postcopy_ram(void);
bool migrate_zero_blocks(void);
bool migrate_auto_converge(void);
+bool migrate_use_multifd(void);
int migrate_use_xbzrle(void);
int64_t migrate_xbzrle_cache_size(void);
diff --git a/qapi-schema.json b/qapi-schema.json
index 802ea53..521e15c 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -918,13 +918,14 @@
#
# @return-path: If enabled, migration will use the return path even
# for precopy. (since 2.10)
+# @x-multifd: Use more than one fd for migration (since 2.11)
#
# Since: 1.2
##
{ 'enum': 'MigrationCapability',
'data': ['xbzrle', 'rdma-pin-all', 'auto-converge', 'zero-blocks',
'compress', 'events', 'postcopy-ram', 'x-colo', 'release-ram',
- 'block', 'return-path' ] }
+ 'block', 'return-path', 'x-multifd'] }
##
# @MigrationCapabilityStatus:
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (5 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 7:15 ` Peter Xu
2017-08-11 14:56 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter Juan Quintela
` (12 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Indicates the number of threads that we would create. By default we
create 2 threads.
Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
--
Catch inconsistent defaults (eric).
Improve comment stating that number of threads is the same than number
of sockets
Use new DEFIN_PROP_*
---
hmp.c | 7 +++++++
migration/migration.c | 26 ++++++++++++++++++++++++++
migration/migration.h | 1 +
qapi-schema.json | 25 ++++++++++++++++++++++---
4 files changed, 56 insertions(+), 3 deletions(-)
diff --git a/hmp.c b/hmp.c
index fd80dce..7899813 100644
--- a/hmp.c
+++ b/hmp.c
@@ -337,6 +337,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
monitor_printf(mon, "%s: %s\n",
MigrationParameter_lookup[MIGRATION_PARAMETER_BLOCK_INCREMENTAL],
params->block_incremental ? "on" : "off");
+ monitor_printf(mon, "%s: %" PRId64 "\n",
+ MigrationParameter_lookup[MIGRATION_PARAMETER_X_MULTIFD_THREADS],
+ params->x_multifd_threads);
}
qapi_free_MigrationParameters(params);
@@ -1622,6 +1625,10 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
p->has_block_incremental = true;
visit_type_bool(v, param, &p->block_incremental, &err);
break;
+ case MIGRATION_PARAMETER_X_MULTIFD_THREADS:
+ p->has_x_multifd_threads = true;
+ visit_type_int(v, param, &p->x_multifd_threads, &err);
+ break;
}
if (err) {
diff --git a/migration/migration.c b/migration/migration.c
index 4e6ad87..6ed9600 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -78,6 +78,7 @@
* Note: Please change this default value to 10000 when we support hybrid mode.
*/
#define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY 200
+#define DEFAULT_MIGRATE_MULTIFD_THREADS 2
static NotifierList migration_state_notifiers =
NOTIFIER_LIST_INITIALIZER(migration_state_notifiers);
@@ -483,6 +484,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
params->x_checkpoint_delay = s->parameters.x_checkpoint_delay;
params->has_block_incremental = true;
params->block_incremental = s->parameters.block_incremental;
+ params->has_x_multifd_threads = true;
+ params->x_multifd_threads = s->parameters.x_multifd_threads;
return params;
}
@@ -764,6 +767,13 @@ static bool migrate_params_check(MigrationParameters *params, Error **errp)
"is invalid, it should be positive");
return false;
}
+ if (params->has_x_multifd_threads &&
+ (params->x_multifd_threads < 1 || params->x_multifd_threads > 255)) {
+ error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
+ "multifd_threads",
+ "is invalid, it should be in the range of 1 to 255");
+ return false;
+ }
return true;
}
@@ -882,6 +892,9 @@ static void migrate_params_apply(MigrateSetParameters *params)
if (params->has_block_incremental) {
s->parameters.block_incremental = params->block_incremental;
}
+ if (params->has_x_multifd_threads) {
+ s->parameters.x_multifd_threads = params->x_multifd_threads;
+ }
}
void qmp_migrate_set_parameters(MigrateSetParameters *params, Error **errp)
@@ -1459,6 +1472,15 @@ bool migrate_use_multifd(void)
return s->enabled_capabilities[MIGRATION_CAPABILITY_X_MULTIFD];
}
+int migrate_multifd_threads(void)
+{
+ MigrationState *s;
+
+ s = migrate_get_current();
+
+ return s->parameters.x_multifd_threads;
+}
+
int migrate_use_xbzrle(void)
{
MigrationState *s;
@@ -2224,6 +2246,9 @@ static Property migration_properties[] = {
DEFINE_PROP_INT64("x-checkpoint-delay", MigrationState,
parameters.x_checkpoint_delay,
DEFAULT_MIGRATE_X_CHECKPOINT_DELAY),
+ DEFINE_PROP_INT64("x-multifd-threads", MigrationState,
+ parameters.x_multifd_threads,
+ DEFAULT_MIGRATE_MULTIFD_THREADS),
/* Migration capabilities */
DEFINE_PROP_MIG_CAP("x-xbzrle", MIGRATION_CAPABILITY_XBZRLE),
@@ -2281,6 +2306,7 @@ static void migration_instance_init(Object *obj)
params->has_downtime_limit = true;
params->has_x_checkpoint_delay = true;
params->has_block_incremental = true;
+ params->has_x_multifd_threads = true;
}
/*
diff --git a/migration/migration.h b/migration/migration.h
index b7437f1..8e4803d 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -175,6 +175,7 @@ bool migrate_zero_blocks(void);
bool migrate_auto_converge(void);
bool migrate_use_multifd(void);
+int migrate_multifd_threads(void);
int migrate_use_xbzrle(void);
int64_t migrate_xbzrle_cache_size(void);
diff --git a/qapi-schema.json b/qapi-schema.json
index 521e15c..3fe1a64 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -918,6 +918,7 @@
#
# @return-path: If enabled, migration will use the return path even
# for precopy. (since 2.10)
+#
# @x-multifd: Use more than one fd for migration (since 2.11)
#
# Since: 1.2
@@ -1043,13 +1044,19 @@
# migrated and the destination must already have access to the
# same backing chain as was used on the source. (since 2.10)
#
+# @x-multifd-threads: Number of threads used to migrate data in
+# parallel. This is the same number that the
+# number of sockets used for migration.
+# The default value is 2 (since 2.11)
+#
# Since: 2.4
##
{ 'enum': 'MigrationParameter',
'data': ['compress-level', 'compress-threads', 'decompress-threads',
'cpu-throttle-initial', 'cpu-throttle-increment',
'tls-creds', 'tls-hostname', 'max-bandwidth',
- 'downtime-limit', 'x-checkpoint-delay', 'block-incremental' ] }
+ 'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
+ 'x-multifd-threads'] }
##
# @MigrateSetParameters:
@@ -1105,6 +1112,11 @@
# migrated and the destination must already have access to the
# same backing chain as was used on the source. (since 2.10)
#
+# @x-multifd-threads: Number of threads used to migrate data in
+# parallel. This is the same number that the
+# number of sockets used for migration.
+# The default value is 2 (since 2.11)
+#
# Since: 2.4
##
# TODO either fuse back into MigrationParameters, or make
@@ -1120,7 +1132,8 @@
'*max-bandwidth': 'int',
'*downtime-limit': 'int',
'*x-checkpoint-delay': 'int',
- '*block-incremental': 'bool' } }
+ '*block-incremental': 'bool',
+ '*x-multifd-threads': 'int' } }
##
# @migrate-set-parameters:
@@ -1191,6 +1204,11 @@
# migrated and the destination must already have access to the
# same backing chain as was used on the source. (since 2.10)
#
+# @x-multifd-threads: Number of threads used to migrate data in
+# parallel. This is the same number that the
+# number of sockets used for migration.
+# The default value is 2 (since 2.11)
+#
# Since: 2.4
##
{ 'struct': 'MigrationParameters',
@@ -1204,7 +1222,8 @@
'*max-bandwidth': 'int',
'*downtime-limit': 'int',
'*x-checkpoint-delay': 'int',
- '*block-incremental': 'bool' } }
+ '*block-incremental': 'bool',
+ '*x-multifd-threads': 'int'} }
##
# @query-migrate-parameters:
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (6 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 7:16 ` Peter Xu
2017-08-11 15:03 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 09/19] migration: Create multifd migration threads Juan Quintela
` (11 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Indicates how many pages we are going to send in each batch to a multifd
thread.
Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
--
Be consistent with defaults and documentation
Use new DEFINE_PROP_*
---
hmp.c | 7 +++++++
migration/migration.c | 26 ++++++++++++++++++++++++++
migration/migration.h | 1 +
qapi-schema.json | 17 ++++++++++++++---
4 files changed, 48 insertions(+), 3 deletions(-)
diff --git a/hmp.c b/hmp.c
index 7899813..a52035a 100644
--- a/hmp.c
+++ b/hmp.c
@@ -340,6 +340,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
monitor_printf(mon, "%s: %" PRId64 "\n",
MigrationParameter_lookup[MIGRATION_PARAMETER_X_MULTIFD_THREADS],
params->x_multifd_threads);
+ monitor_printf(mon, "%s: %" PRId64 "\n",
+ MigrationParameter_lookup[MIGRATION_PARAMETER_X_MULTIFD_GROUP],
+ params->x_multifd_group);
}
qapi_free_MigrationParameters(params);
@@ -1629,6 +1632,10 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
p->has_x_multifd_threads = true;
visit_type_int(v, param, &p->x_multifd_threads, &err);
break;
+ case MIGRATION_PARAMETER_X_MULTIFD_GROUP:
+ p->has_x_multifd_group = true;
+ visit_type_int(v, param, &p->x_multifd_group, &err);
+ break;
}
if (err) {
diff --git a/migration/migration.c b/migration/migration.c
index 6ed9600..6c4eb4b 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -79,6 +79,7 @@
*/
#define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY 200
#define DEFAULT_MIGRATE_MULTIFD_THREADS 2
+#define DEFAULT_MIGRATE_MULTIFD_GROUP 16
static NotifierList migration_state_notifiers =
NOTIFIER_LIST_INITIALIZER(migration_state_notifiers);
@@ -486,6 +487,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
params->block_incremental = s->parameters.block_incremental;
params->has_x_multifd_threads = true;
params->x_multifd_threads = s->parameters.x_multifd_threads;
+ params->has_x_multifd_group = true;
+ params->x_multifd_group = s->parameters.x_multifd_group;
return params;
}
@@ -774,6 +777,13 @@ static bool migrate_params_check(MigrationParameters *params, Error **errp)
"is invalid, it should be in the range of 1 to 255");
return false;
}
+ if (params->has_x_multifd_group &&
+ (params->x_multifd_group < 1 || params->x_multifd_group > 10000)) {
+ error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
+ "multifd_group",
+ "is invalid, it should be in the range of 1 to 10000");
+ return false;
+ }
return true;
}
@@ -895,6 +905,9 @@ static void migrate_params_apply(MigrateSetParameters *params)
if (params->has_x_multifd_threads) {
s->parameters.x_multifd_threads = params->x_multifd_threads;
}
+ if (params->has_x_multifd_group) {
+ s->parameters.x_multifd_group = params->x_multifd_group;
+ }
}
void qmp_migrate_set_parameters(MigrateSetParameters *params, Error **errp)
@@ -1481,6 +1494,15 @@ int migrate_multifd_threads(void)
return s->parameters.x_multifd_threads;
}
+int migrate_multifd_group(void)
+{
+ MigrationState *s;
+
+ s = migrate_get_current();
+
+ return s->parameters.x_multifd_group;
+}
+
int migrate_use_xbzrle(void)
{
MigrationState *s;
@@ -2249,6 +2271,9 @@ static Property migration_properties[] = {
DEFINE_PROP_INT64("x-multifd-threads", MigrationState,
parameters.x_multifd_threads,
DEFAULT_MIGRATE_MULTIFD_THREADS),
+ DEFINE_PROP_INT64("x-multifd-group", MigrationState,
+ parameters.x_multifd_group,
+ DEFAULT_MIGRATE_MULTIFD_GROUP),
/* Migration capabilities */
DEFINE_PROP_MIG_CAP("x-xbzrle", MIGRATION_CAPABILITY_XBZRLE),
@@ -2307,6 +2332,7 @@ static void migration_instance_init(Object *obj)
params->has_x_checkpoint_delay = true;
params->has_block_incremental = true;
params->has_x_multifd_threads = true;
+ params->has_x_multifd_group = true;
}
/*
diff --git a/migration/migration.h b/migration/migration.h
index 8e4803d..ddf8c89 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -176,6 +176,7 @@ bool migrate_zero_blocks(void);
bool migrate_auto_converge(void);
bool migrate_use_multifd(void);
int migrate_multifd_threads(void);
+int migrate_multifd_group(void);
int migrate_use_xbzrle(void);
int64_t migrate_xbzrle_cache_size(void);
diff --git a/qapi-schema.json b/qapi-schema.json
index 3fe1a64..2d4dbd3 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -1049,6 +1049,9 @@
# number of sockets used for migration.
# The default value is 2 (since 2.11)
#
+# @x-multifd-group: Number of pages sent together to a thread
+# The default value is 16 (since 2.11)
+#
# Since: 2.4
##
{ 'enum': 'MigrationParameter',
@@ -1056,7 +1059,7 @@
'cpu-throttle-initial', 'cpu-throttle-increment',
'tls-creds', 'tls-hostname', 'max-bandwidth',
'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
- 'x-multifd-threads'] }
+ 'x-multifd-threads', 'x-multifd-group'] }
##
# @MigrateSetParameters:
@@ -1117,6 +1120,9 @@
# number of sockets used for migration.
# The default value is 2 (since 2.11)
#
+# @x-multifd-group: Number of pages sent together to a thread
+# The default value is 16 (since 2.11)
+#
# Since: 2.4
##
# TODO either fuse back into MigrationParameters, or make
@@ -1133,7 +1139,8 @@
'*downtime-limit': 'int',
'*x-checkpoint-delay': 'int',
'*block-incremental': 'bool',
- '*x-multifd-threads': 'int' } }
+ '*x-multifd-threads': 'int',
+ '*x-multifd-group': 'int' } }
##
# @migrate-set-parameters:
@@ -1209,6 +1216,9 @@
# number of sockets used for migration.
# The default value is 2 (since 2.11)
#
+# @x-multifd-group: Number of pages sent together to a thread
+# The default value is 16 (since 2.11)
+#
# Since: 2.4
##
{ 'struct': 'MigrationParameters',
@@ -1223,7 +1233,8 @@
'*downtime-limit': 'int',
'*x-checkpoint-delay': 'int',
'*block-incremental': 'bool',
- '*x-multifd-threads': 'int'} }
+ '*x-multifd-threads': 'int',
+ '*x-multifd-group': 'int'} }
##
# @query-migrate-parameters:
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 09/19] migration: Create multifd migration threads
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (7 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-15 6:42 ` Peter Xu
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming Juan Quintela
` (10 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Creation of the threads, nothing inside yet.
Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
--
Use pointers instead of long array names
Move to use semaphores instead of conditions as paolo suggestion
Put all the state inside one struct.
Use a counter for the number of threads created. Needed during cancellation.
Add error return to thread creation
Add id field
Rename functions to multifd_save/load_setup/cleanup
Change recv parameters to a pointer to struct
---
migration/migration.c | 16 ++++-
migration/ram.c | 194 ++++++++++++++++++++++++++++++++++++++++++++++++++
migration/ram.h | 5 ++
3 files changed, 214 insertions(+), 1 deletion(-)
diff --git a/migration/migration.c b/migration/migration.c
index 6c4eb4b..d031c93 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -296,6 +296,7 @@ static void process_incoming_migration_bh(void *opaque)
} else {
runstate_set(global_state_get_runstate());
}
+ multifd_load_cleanup();
/*
* 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
@@ -358,6 +359,7 @@ static void process_incoming_migration_co(void *opaque)
MIGRATION_STATUS_FAILED);
error_report("load of migration failed: %s", strerror(-ret));
qemu_fclose(mis->from_src_file);
+ multifd_load_cleanup();
exit(EXIT_FAILURE);
}
mis->bh = qemu_bh_new(process_incoming_migration_bh, mis);
@@ -368,7 +370,12 @@ void migration_fd_process_incoming(QEMUFile *f)
{
Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
MigrationIncomingState *mis = migration_incoming_get_current();
-
+
+ if (multifd_load_setup() != 0) {
+ /* We haven't been able to create multifd threads
+ nothing better to do */
+ exit(EXIT_FAILURE);
+ }
if (!mis->from_src_file) {
mis->from_src_file = f;
}
@@ -1028,6 +1035,7 @@ static void migrate_fd_cleanup(void *opaque)
}
qemu_mutex_lock_iothread();
+ multifd_save_cleanup();
qemu_fclose(s->to_dst_file);
s->to_dst_file = NULL;
}
@@ -2217,6 +2225,12 @@ void migrate_fd_connect(MigrationState *s)
}
}
+ if (multifd_save_setup() != 0) {
+ migrate_set_state(&s->state, MIGRATION_STATUS_SETUP,
+ MIGRATION_STATUS_FAILED);
+ migrate_fd_cleanup(s);
+ return;
+ }
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 e18b3e2..9fb3496 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -356,6 +356,200 @@ static void compress_threads_save_setup(void)
}
}
+/* Multiple fd's */
+
+struct MultiFDSendParams {
+ uint8_t id;
+ QemuThread thread;
+ QemuSemaphore sem;
+ QemuMutex mutex;
+ bool quit;
+};
+typedef struct MultiFDSendParams MultiFDSendParams;
+
+struct {
+ MultiFDSendParams *params;
+ /* number of created threads */
+ int count;
+} *multifd_send_state;
+
+static void terminate_multifd_send_threads(void)
+{
+ int i;
+
+ for (i = 0; i < multifd_send_state->count; i++) {
+ MultiFDSendParams *p = &multifd_send_state->params[i];
+
+ qemu_mutex_lock(&p->mutex);
+ p->quit = true;
+ qemu_sem_post(&p->sem);
+ qemu_mutex_unlock(&p->mutex);
+ }
+}
+
+void multifd_save_cleanup(void)
+{
+ int i;
+
+ if (!migrate_use_multifd()) {
+ return;
+ }
+ terminate_multifd_send_threads();
+ for (i = 0; i < multifd_send_state->count; i++) {
+ MultiFDSendParams *p = &multifd_send_state->params[i];
+
+ qemu_thread_join(&p->thread);
+ qemu_mutex_destroy(&p->mutex);
+ qemu_sem_destroy(&p->sem);
+ }
+ g_free(multifd_send_state->params);
+ multifd_send_state->params = NULL;
+ g_free(multifd_send_state);
+ multifd_send_state = NULL;
+}
+
+static void *multifd_send_thread(void *opaque)
+{
+ MultiFDSendParams *p = opaque;
+
+ while (true) {
+ qemu_mutex_lock(&p->mutex);
+ if (p->quit) {
+ qemu_mutex_unlock(&p->mutex);
+ break;
+ }
+ qemu_mutex_unlock(&p->mutex);
+ qemu_sem_wait(&p->sem);
+ }
+
+ return NULL;
+}
+
+int multifd_save_setup(void)
+{
+ int thread_count;
+ uint8_t i;
+
+ if (!migrate_use_multifd()) {
+ return 0;
+ }
+ thread_count = migrate_multifd_threads();
+ multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
+ multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
+ multifd_send_state->count = 0;
+ for (i = 0; i < thread_count; i++) {
+ char thread_name[16];
+ MultiFDSendParams *p = &multifd_send_state->params[i];
+
+ qemu_mutex_init(&p->mutex);
+ qemu_sem_init(&p->sem, 0);
+ p->quit = false;
+ p->id = i;
+ snprintf(thread_name, sizeof(thread_name), "multifdsend_%d", i);
+ qemu_thread_create(&p->thread, thread_name, multifd_send_thread, p,
+ QEMU_THREAD_JOINABLE);
+ multifd_send_state->count++;
+ }
+ return 0;
+}
+
+struct MultiFDRecvParams {
+ uint8_t id;
+ QemuThread thread;
+ QemuSemaphore sem;
+ QemuMutex mutex;
+ bool quit;
+};
+typedef struct MultiFDRecvParams MultiFDRecvParams;
+
+struct {
+ MultiFDRecvParams **params;
+ /* number of created threads */
+ int count;
+} *multifd_recv_state;
+
+static void terminate_multifd_recv_threads(void)
+{
+ int i;
+
+ for (i = 0; i < multifd_recv_state->count; i++) {
+ MultiFDRecvParams *p = multifd_recv_state->params[i];
+
+ qemu_mutex_lock(&p->mutex);
+ p->quit = true;
+ qemu_sem_post(&p->sem);
+ qemu_mutex_unlock(&p->mutex);
+ }
+}
+
+void multifd_load_cleanup(void)
+{
+ int i;
+
+ if (!migrate_use_multifd()) {
+ return;
+ }
+ terminate_multifd_recv_threads();
+ for (i = 0; i < multifd_recv_state->count; i++) {
+ MultiFDRecvParams *p = multifd_recv_state->params[i];
+
+ qemu_thread_join(&p->thread);
+ qemu_mutex_destroy(&p->mutex);
+ qemu_sem_destroy(&p->sem);
+ g_free(p);
+ multifd_recv_state->params[i] = NULL;
+ }
+ g_free(multifd_recv_state->params);
+ multifd_recv_state->params = NULL;
+ g_free(multifd_recv_state);
+ multifd_recv_state = NULL;
+}
+
+static void *multifd_recv_thread(void *opaque)
+{
+ MultiFDRecvParams *p = opaque;
+
+ while (true) {
+ qemu_mutex_lock(&p->mutex);
+ if (p->quit) {
+ qemu_mutex_unlock(&p->mutex);
+ break;
+ }
+ qemu_mutex_unlock(&p->mutex);
+ qemu_sem_wait(&p->sem);
+ }
+
+ return NULL;
+}
+
+int multifd_load_setup(void)
+{
+ int thread_count;
+ uint8_t i;
+
+ if (!migrate_use_multifd()) {
+ return 0;
+ }
+ thread_count = migrate_multifd_threads();
+ multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
+ multifd_recv_state->params = g_new0(MultiFDRecvParams *, thread_count);
+ multifd_recv_state->count = 0;
+ for (i = 0; i < thread_count; i++) {
+ char thread_name[16];
+ MultiFDRecvParams *p = multifd_recv_state->params[i];
+
+ qemu_mutex_init(&p->mutex);
+ qemu_sem_init(&p->sem, 0);
+ p->quit = false;
+ p->id = i;
+ snprintf(thread_name, sizeof(thread_name), "multifdrecv_%d", i);
+ qemu_thread_create(&p->thread, thread_name, multifd_recv_thread, p,
+ QEMU_THREAD_JOINABLE);
+ multifd_recv_state->count++;
+ }
+ return 0;
+}
+
/**
* save_page_header: write page header to wire
*
diff --git a/migration/ram.h b/migration/ram.h
index c081fde..93c2bb4 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -39,6 +39,11 @@ int64_t xbzrle_cache_resize(int64_t new_size);
uint64_t ram_bytes_remaining(void);
uint64_t ram_bytes_total(void);
+int multifd_save_setup(void);
+void multifd_save_cleanup(void);
+int multifd_load_setup(void);
+void multifd_load_cleanup(void);
+
uint64_t ram_pagesize_summary(void);
int ram_save_queue_pages(const char *rbname, ram_addr_t start, ram_addr_t len);
void acct_update_position(QEMUFile *f, size_t size, bool zero);
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (8 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 09/19] migration: Create multifd migration threads Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 15:10 ` Daniel P. Berrange
2017-08-15 6:43 ` Peter Xu
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work Juan Quintela
` (9 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We need that on later patches.
Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
---
migration/migration.c | 14 ++++++++++++--
1 file changed, 12 insertions(+), 2 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index d031c93..e36e880 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -366,9 +366,8 @@ static void process_incoming_migration_co(void *opaque)
qemu_bh_schedule(mis->bh);
}
-void migration_fd_process_incoming(QEMUFile *f)
+static void migration_incoming_setup(QEMUFile *f)
{
- Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
MigrationIncomingState *mis = migration_incoming_get_current();
if (multifd_load_setup() != 0) {
@@ -380,9 +379,20 @@ void migration_fd_process_incoming(QEMUFile *f)
mis->from_src_file = f;
}
qemu_file_set_blocking(f, false);
+}
+
+static void migration_incoming_process(void)
+{
+ Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
qemu_coroutine_enter(co);
}
+void migration_fd_process_incoming(QEMUFile *f)
+{
+ migration_incoming_setup(f);
+ migration_incoming_process();
+}
+
void migration_ioc_process_incoming(QIOChannel *ioc)
{
MigrationIncomingState *mis = migration_incoming_get_current();
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (9 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 15:22 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 12/19] migration: Create ram_multifd_page Juan Quintela
` (8 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We create new channels for each new thread created. We send through
them a string containing <uuid> multifd <channel number> so we are
sure that we connect the right channels in both sides.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Split SocketArgs into incoming and outgoing args
Use UUID's on the initial message, so we are sure we are connecting to
the right channel.
Remove init semaphore. Now that we use uuids on the init message, we
know that this is our channel.
Fix recv socket destwroy, we were destroying send channels.
This was very interesting, because we were using an unreferred object
without problems.
Move to struct of pointers
init channel sooner.
split recv thread creation.
listen on main thread
We count the number of created threads to know when we need to stop listening
Use g_strdup_printf
report channel id on errors
---
migration/migration.c | 5 +++
migration/ram.c | 99 +++++++++++++++++++++++++++++++++++++++++++--------
migration/ram.h | 3 ++
migration/socket.c | 36 ++++++++++++++++++-
migration/socket.h | 10 ++++++
5 files changed, 138 insertions(+), 15 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index e36e880..944d6e2 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -413,6 +413,11 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
*/
bool migration_has_all_channels(void)
{
+ if (migrate_use_multifd()) {
+ int thread_count = migrate_multifd_threads();
+
+ return thread_count == multifd_created_threads();
+ }
return true;
}
diff --git a/migration/ram.c b/migration/ram.c
index 9fb3496..e9fa556 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -36,6 +36,7 @@
#include "xbzrle.h"
#include "ram.h"
#include "migration.h"
+#include "socket.h"
#include "migration/register.h"
#include "migration/misc.h"
#include "qemu-file.h"
@@ -46,6 +47,8 @@
#include "exec/ram_addr.h"
#include "qemu/rcu_queue.h"
#include "migration/colo.h"
+#include "sysemu/sysemu.h"
+#include "qemu/uuid.h"
/***********************************************************/
/* ram save/restore */
@@ -361,6 +364,7 @@ static void compress_threads_save_setup(void)
struct MultiFDSendParams {
uint8_t id;
QemuThread thread;
+ QIOChannel *c;
QemuSemaphore sem;
QemuMutex mutex;
bool quit;
@@ -401,6 +405,7 @@ void multifd_save_cleanup(void)
qemu_thread_join(&p->thread);
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem);
+ socket_send_channel_destroy(p->c);
}
g_free(multifd_send_state->params);
multifd_send_state->params = NULL;
@@ -408,9 +413,26 @@ void multifd_save_cleanup(void)
multifd_send_state = NULL;
}
+/* Default uuid for multifd when qemu is not started with uuid */
+static char multifd_uuid[] = "5c49fd7e-af88-4a07-b6e8-091fd696ad40";
+/* strlen(multifd) + '-' + <channel id> + '-' + UUID_FMT + '\0' */
+#define MULTIFD_UUID_MSG (7 + 1 + 3 + 1 + UUID_FMT_LEN + 1)
+
static void *multifd_send_thread(void *opaque)
{
MultiFDSendParams *p = opaque;
+ char *string;
+ char *string_uuid;
+
+ if (qemu_uuid_set) {
+ string_uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
+ } else {
+ string_uuid = g_strdup(multifd_uuid);
+ }
+ string = g_strdup_printf("%s multifd %03d", string_uuid, p->id);
+ g_free(string_uuid);
+ qio_channel_write(p->c, string, MULTIFD_UUID_MSG, &error_abort);
+ g_free(string);
while (true) {
qemu_mutex_lock(&p->mutex);
@@ -445,6 +467,12 @@ int multifd_save_setup(void)
qemu_sem_init(&p->sem, 0);
p->quit = false;
p->id = i;
+ p->c = socket_send_channel_create();
+ if (!p->c) {
+ error_report("Error creating a send channel");
+ multifd_save_cleanup();
+ return -1;
+ }
snprintf(thread_name, sizeof(thread_name), "multifdsend_%d", i);
qemu_thread_create(&p->thread, thread_name, multifd_send_thread, p,
QEMU_THREAD_JOINABLE);
@@ -456,6 +484,7 @@ int multifd_save_setup(void)
struct MultiFDRecvParams {
uint8_t id;
QemuThread thread;
+ QIOChannel *c;
QemuSemaphore sem;
QemuMutex mutex;
bool quit;
@@ -466,12 +495,16 @@ struct {
MultiFDRecvParams **params;
/* number of created threads */
int count;
+ /* Should we finish */
+ bool quit;
} *multifd_recv_state;
static void terminate_multifd_recv_threads(void)
{
int i;
+ multifd_recv_state->quit = true;
+
for (i = 0; i < multifd_recv_state->count; i++) {
MultiFDRecvParams *p = multifd_recv_state->params[i];
@@ -496,6 +529,7 @@ void multifd_load_cleanup(void)
qemu_thread_join(&p->thread);
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem);
+ socket_recv_channel_destroy(p->c);
g_free(p);
multifd_recv_state->params[i] = NULL;
}
@@ -522,10 +556,54 @@ static void *multifd_recv_thread(void *opaque)
return NULL;
}
+void multifd_new_channel(QIOChannel *ioc)
+{
+ MultiFDRecvParams *p = g_new0(MultiFDRecvParams, 1);
+ MigrationState *s = migrate_get_current();
+ char string[MULTIFD_UUID_MSG];
+ char string_uuid[UUID_FMT_LEN];
+ char *uuid;
+ int id;
+
+ qio_channel_read(ioc, string, sizeof(string), &error_abort);
+ sscanf(string, "%s multifd %03d", string_uuid, &id);
+
+ if (qemu_uuid_set) {
+ uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
+ } else {
+ uuid = g_strdup(multifd_uuid);
+ }
+ if (strcmp(string_uuid, uuid)) {
+ error_report("multifd: received uuid '%s' and expected uuid '%s'"
+ " for channel %d", string_uuid, uuid, id);
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return;
+ }
+ g_free(uuid);
+
+ if (multifd_recv_state->params[id] != NULL) {
+ error_report("multifd: received id '%d' is already setup'", id);
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return;
+ }
+ qemu_mutex_init(&p->mutex);
+ qemu_sem_init(&p->sem, 0);
+ p->quit = false;
+ p->id = id;
+ p->c = ioc;
+ atomic_set(&multifd_recv_state->params[id], p);
+ multifd_recv_state->count++;
+ qemu_thread_create(&p->thread, "multifd_recv", multifd_recv_thread, p,
+ QEMU_THREAD_JOINABLE);
+}
+
int multifd_load_setup(void)
{
int thread_count;
- uint8_t i;
if (!migrate_use_multifd()) {
return 0;
@@ -534,22 +612,15 @@ int multifd_load_setup(void)
multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
multifd_recv_state->params = g_new0(MultiFDRecvParams *, thread_count);
multifd_recv_state->count = 0;
- for (i = 0; i < thread_count; i++) {
- char thread_name[16];
- MultiFDRecvParams *p = multifd_recv_state->params[i];
-
- qemu_mutex_init(&p->mutex);
- qemu_sem_init(&p->sem, 0);
- p->quit = false;
- p->id = i;
- snprintf(thread_name, sizeof(thread_name), "multifdrecv_%d", i);
- qemu_thread_create(&p->thread, thread_name, multifd_recv_thread, p,
- QEMU_THREAD_JOINABLE);
- multifd_recv_state->count++;
- }
+ multifd_recv_state->quit = false;
return 0;
}
+int multifd_created_threads(void)
+{
+ return multifd_recv_state->count;
+}
+
/**
* save_page_header: write page header to wire
*
diff --git a/migration/ram.h b/migration/ram.h
index 93c2bb4..b3dabf2 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -31,6 +31,7 @@
#include "qemu-common.h"
#include "exec/cpu-common.h"
+#include "io/channel.h"
extern MigrationStats ram_counters;
extern XBZRLECacheStats xbzrle_counters;
@@ -43,6 +44,8 @@ int multifd_save_setup(void);
void multifd_save_cleanup(void);
int multifd_load_setup(void);
void multifd_load_cleanup(void);
+void multifd_new_channel(QIOChannel *ioc);
+int multifd_created_threads(void);
uint64_t ram_pagesize_summary(void);
int ram_save_queue_pages(const char *rbname, ram_addr_t start, ram_addr_t len);
diff --git a/migration/socket.c b/migration/socket.c
index dee8690..5dd6f42 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -26,6 +26,38 @@
#include "io/channel-socket.h"
#include "trace.h"
+int socket_recv_channel_destroy(QIOChannel *recv)
+{
+ /* Remove channel */
+ object_unref(OBJECT(recv));
+ return 0;
+}
+
+struct SocketOutgoingArgs {
+ SocketAddress *saddr;
+ Error **errp;
+} outgoing_args;
+
+QIOChannel *socket_send_channel_create(void)
+{
+ QIOChannelSocket *sioc = qio_channel_socket_new();
+
+ qio_channel_socket_connect_sync(sioc, outgoing_args.saddr,
+ outgoing_args.errp);
+ qio_channel_set_delay(QIO_CHANNEL(sioc), false);
+ return QIO_CHANNEL(sioc);
+}
+
+int socket_send_channel_destroy(QIOChannel *send)
+{
+ /* Remove channel */
+ object_unref(OBJECT(send));
+ if (outgoing_args.saddr) {
+ qapi_free_SocketAddress(outgoing_args.saddr);
+ outgoing_args.saddr = NULL;
+ }
+ return 0;
+}
static SocketAddress *tcp_build_address(const char *host_port, Error **errp)
{
@@ -96,6 +128,9 @@ static void socket_start_outgoing_migration(MigrationState *s,
struct SocketConnectData *data = g_new0(struct SocketConnectData, 1);
data->s = s;
+ outgoing_args.saddr = saddr;
+ outgoing_args.errp = errp;
+
if (saddr->type == SOCKET_ADDRESS_TYPE_INET) {
data->hostname = g_strdup(saddr->u.inet.host);
}
@@ -106,7 +141,6 @@ static void socket_start_outgoing_migration(MigrationState *s,
socket_outgoing_migration,
data,
socket_connect_data_free);
- qapi_free_SocketAddress(saddr);
}
void tcp_start_outgoing_migration(MigrationState *s,
diff --git a/migration/socket.h b/migration/socket.h
index 6b91e9d..dabce0e 100644
--- a/migration/socket.h
+++ b/migration/socket.h
@@ -16,6 +16,16 @@
#ifndef QEMU_MIGRATION_SOCKET_H
#define QEMU_MIGRATION_SOCKET_H
+
+#include "io/channel.h"
+
+QIOChannel *socket_recv_channel_create(void);
+int socket_recv_channel_destroy(QIOChannel *recv);
+
+QIOChannel *socket_send_channel_create(void);
+
+int socket_send_channel_destroy(QIOChannel *send);
+
void tcp_start_incoming_migration(const char *host_port, Error **errp);
void tcp_start_outgoing_migration(MigrationState *s, const char *host_port,
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 12/19] migration: Create ram_multifd_page
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (10 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 15:25 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 13/19] migration: Really use multiple pages at a time Juan Quintela
` (7 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
The function still don't use multifd, but we have simplified
ram_save_page, xbzrle and RDMA stuff is gone. We have added a new
counter and a new flag for this type of pages.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Add last_page parameter
Add commets for done and address
---
hmp.c | 2 ++
migration/migration.c | 1 +
migration/ram.c | 94 ++++++++++++++++++++++++++++++++++++++++++++++++++-
qapi-schema.json | 5 ++-
4 files changed, 100 insertions(+), 2 deletions(-)
diff --git a/hmp.c b/hmp.c
index a52035a..bc2b071 100644
--- a/hmp.c
+++ b/hmp.c
@@ -234,6 +234,8 @@ void hmp_info_migrate(Monitor *mon, const QDict *qdict)
monitor_printf(mon, "postcopy request count: %" PRIu64 "\n",
info->ram->postcopy_requests);
}
+ monitor_printf(mon, "multifd: %" PRIu64 " pages\n",
+ info->ram->multifd);
}
if (info->has_disk) {
diff --git a/migration/migration.c b/migration/migration.c
index 944d6e2..8e9505a 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -549,6 +549,7 @@ static void populate_ram_info(MigrationInfo *info, MigrationState *s)
info->ram->dirty_sync_count = ram_counters.dirty_sync_count;
info->ram->postcopy_requests = ram_counters.postcopy_requests;
info->ram->page_size = qemu_target_page_size();
+ info->ram->multifd = ram_counters.multifd;
if (migrate_use_xbzrle()) {
info->has_xbzrle_cache = true;
diff --git a/migration/ram.c b/migration/ram.c
index e9fa556..03f3427 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -68,6 +68,7 @@
#define RAM_SAVE_FLAG_XBZRLE 0x40
/* 0x80 is reserved in migration.h start with 0x100 next */
#define RAM_SAVE_FLAG_COMPRESS_PAGE 0x100
+#define RAM_SAVE_FLAG_MULTIFD_PAGE 0x200
static inline bool is_zero_range(uint8_t *p, uint64_t size)
{
@@ -362,12 +363,22 @@ static void compress_threads_save_setup(void)
/* Multiple fd's */
struct MultiFDSendParams {
+ /* not changed */
uint8_t id;
QemuThread thread;
QIOChannel *c;
QemuSemaphore sem;
QemuMutex mutex;
+ /* protected by param mutex */
bool quit;
+ /* This is a temp field. We are using it now to transmit
+ something the address of the page. Later in the series, we
+ change it for the real page.
+ */
+ uint8_t *address;
+ /* protected by multifd mutex */
+ /* has the thread finish the last submitted job */
+ bool done;
};
typedef struct MultiFDSendParams MultiFDSendParams;
@@ -375,6 +386,8 @@ struct {
MultiFDSendParams *params;
/* number of created threads */
int count;
+ QemuMutex mutex;
+ QemuSemaphore sem;
} *multifd_send_state;
static void terminate_multifd_send_threads(void)
@@ -433,6 +446,7 @@ static void *multifd_send_thread(void *opaque)
g_free(string_uuid);
qio_channel_write(p->c, string, MULTIFD_UUID_MSG, &error_abort);
g_free(string);
+ qemu_sem_post(&multifd_send_state->sem);
while (true) {
qemu_mutex_lock(&p->mutex);
@@ -440,6 +454,15 @@ static void *multifd_send_thread(void *opaque)
qemu_mutex_unlock(&p->mutex);
break;
}
+ if (p->address) {
+ p->address = 0;
+ qemu_mutex_unlock(&p->mutex);
+ qemu_mutex_lock(&multifd_send_state->mutex);
+ p->done = true;
+ qemu_mutex_unlock(&multifd_send_state->mutex);
+ qemu_sem_post(&multifd_send_state->sem);
+ continue;
+ }
qemu_mutex_unlock(&p->mutex);
qemu_sem_wait(&p->sem);
}
@@ -459,6 +482,8 @@ int multifd_save_setup(void)
multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
multifd_send_state->count = 0;
+ qemu_mutex_init(&multifd_send_state->mutex);
+ qemu_sem_init(&multifd_send_state->sem, 0);
for (i = 0; i < thread_count; i++) {
char thread_name[16];
MultiFDSendParams *p = &multifd_send_state->params[i];
@@ -467,6 +492,8 @@ int multifd_save_setup(void)
qemu_sem_init(&p->sem, 0);
p->quit = false;
p->id = i;
+ p->done = true;
+ p->address = 0;
p->c = socket_send_channel_create();
if (!p->c) {
error_report("Error creating a send channel");
@@ -481,6 +508,30 @@ int multifd_save_setup(void)
return 0;
}
+static uint16_t multifd_send_page(uint8_t *address, bool last_page)
+{
+ int i;
+ MultiFDSendParams *p = NULL; /* make happy gcc */
+
+ qemu_sem_wait(&multifd_send_state->sem);
+ qemu_mutex_lock(&multifd_send_state->mutex);
+ for (i = 0; i < multifd_send_state->count; i++) {
+ p = &multifd_send_state->params[i];
+
+ if (p->done) {
+ p->done = false;
+ break;
+ }
+ }
+ qemu_mutex_unlock(&multifd_send_state->mutex);
+ qemu_mutex_lock(&p->mutex);
+ p->address = address;
+ qemu_mutex_unlock(&p->mutex);
+ qemu_sem_post(&p->sem);
+
+ return 0;
+}
+
struct MultiFDRecvParams {
uint8_t id;
QemuThread thread;
@@ -1051,6 +1102,32 @@ static int ram_save_page(RAMState *rs, PageSearchStatus *pss, bool last_stage)
return pages;
}
+static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
+ bool last_stage)
+{
+ int pages;
+ uint8_t *p;
+ RAMBlock *block = pss->block;
+ ram_addr_t offset = pss->page << TARGET_PAGE_BITS;
+
+ p = block->host + offset;
+
+ pages = save_zero_page(rs, block, offset, p);
+ if (pages == -1) {
+ ram_counters.transferred +=
+ save_page_header(rs, rs->f, block,
+ offset | RAM_SAVE_FLAG_MULTIFD_PAGE);
+ qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
+ multifd_send_page(p, rs->migration_dirty_pages == 1);
+ ram_counters.transferred += TARGET_PAGE_SIZE;
+ pages = 1;
+ ram_counters.normal++;
+ ram_counters.multifd++;
+ }
+
+ return pages;
+}
+
static int do_compress_ram_page(QEMUFile *f, RAMBlock *block,
ram_addr_t offset)
{
@@ -1479,6 +1556,8 @@ static int ram_save_target_page(RAMState *rs, PageSearchStatus *pss,
if (migrate_use_compression() &&
(rs->ram_bulk_stage || !migrate_use_xbzrle())) {
res = ram_save_compressed_page(rs, pss, last_stage);
+ } else if (migrate_use_multifd()) {
+ res = ram_multifd_page(rs, pss, last_stage);
} else {
res = ram_save_page(rs, pss, last_stage);
}
@@ -2771,6 +2850,10 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
if (!migrate_use_compression()) {
invalid_flags |= RAM_SAVE_FLAG_COMPRESS_PAGE;
}
+
+ if (!migrate_use_multifd()) {
+ invalid_flags |= RAM_SAVE_FLAG_MULTIFD_PAGE;
+ }
/* This RCU critical section can be very long running.
* When RCU reclaims in the code start to become numerous,
* it will be necessary to reduce the granularity of this
@@ -2795,13 +2878,17 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
if (flags & invalid_flags & RAM_SAVE_FLAG_COMPRESS_PAGE) {
error_report("Received an unexpected compressed page");
}
+ if (flags & invalid_flags & RAM_SAVE_FLAG_MULTIFD_PAGE) {
+ error_report("Received an unexpected multifd page");
+ }
ret = -EINVAL;
break;
}
if (flags & (RAM_SAVE_FLAG_ZERO | RAM_SAVE_FLAG_PAGE |
- RAM_SAVE_FLAG_COMPRESS_PAGE | RAM_SAVE_FLAG_XBZRLE)) {
+ RAM_SAVE_FLAG_COMPRESS_PAGE | RAM_SAVE_FLAG_XBZRLE |
+ RAM_SAVE_FLAG_MULTIFD_PAGE)) {
RAMBlock *block = ram_block_from_stream(f, flags);
host = host_from_ram_block_offset(block, addr);
@@ -2889,6 +2976,11 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
break;
}
break;
+
+ case RAM_SAVE_FLAG_MULTIFD_PAGE:
+ qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
+ break;
+
case RAM_SAVE_FLAG_EOS:
/* normal exit */
break;
diff --git a/qapi-schema.json b/qapi-schema.json
index 2d4dbd3..bb1fe0a 100644
--- a/qapi-schema.json
+++ b/qapi-schema.json
@@ -617,6 +617,8 @@
# @page-size: The number of bytes per page for the various page-based
# statistics (since 2.10)
#
+# @multifd: number of pages sent with multifd (since 2.10)
+#
# Since: 0.14.0
##
{ 'struct': 'MigrationStats',
@@ -624,7 +626,8 @@
'duplicate': 'int', 'skipped': 'int', 'normal': 'int',
'normal-bytes': 'int', 'dirty-pages-rate' : 'int',
'mbps' : 'number', 'dirty-sync-count' : 'int',
- 'postcopy-requests' : 'int', 'page-size' : 'int' } }
+ 'postcopy-requests' : 'int', 'page-size' : 'int',
+ 'multifd' : 'int'} }
##
# @XBZRLECacheStats:
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 13/19] migration: Really use multiple pages at a time
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (11 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 12/19] migration: Create ram_multifd_page Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 14/19] migration: Send the fd number which we are going to use for this page Juan Quintela
` (6 subsequent siblings)
19 siblings, 0 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We now send several pages at a time each time that we wakeup a thread.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Use iovec's instead of creating the equivalent.
Clear memory used by pages (dave)
Use g_new0(danp)
define MULTIFD_CONTINUE
---
migration/ram.c | 57 ++++++++++++++++++++++++++++++++++++++++++++++++---------
1 file changed, 48 insertions(+), 9 deletions(-)
diff --git a/migration/ram.c b/migration/ram.c
index 03f3427..7310da9 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -49,6 +49,7 @@
#include "migration/colo.h"
#include "sysemu/sysemu.h"
#include "qemu/uuid.h"
+#include "qemu/iov.h"
/***********************************************************/
/* ram save/restore */
@@ -362,6 +363,15 @@ static void compress_threads_save_setup(void)
/* Multiple fd's */
+/* used to continue on the same multifd group */
+#define MULTIFD_CONTINUE UINT16_MAX
+
+typedef struct {
+ int num;
+ size_t size;
+ struct iovec *iov;
+} multifd_pages_t;
+
struct MultiFDSendParams {
/* not changed */
uint8_t id;
@@ -371,11 +381,7 @@ struct MultiFDSendParams {
QemuMutex mutex;
/* protected by param mutex */
bool quit;
- /* This is a temp field. We are using it now to transmit
- something the address of the page. Later in the series, we
- change it for the real page.
- */
- uint8_t *address;
+ multifd_pages_t pages;
/* protected by multifd mutex */
/* has the thread finish the last submitted job */
bool done;
@@ -388,8 +394,24 @@ struct {
int count;
QemuMutex mutex;
QemuSemaphore sem;
+ multifd_pages_t pages;
} *multifd_send_state;
+static void multifd_init_group(multifd_pages_t *pages)
+{
+ pages->num = 0;
+ pages->size = migrate_multifd_group();
+ pages->iov = g_new0(struct iovec, pages->size);
+}
+
+static void multifd_clear_group(multifd_pages_t *pages)
+{
+ pages->num = 0;
+ pages->size = 0;
+ g_free(pages->iov);
+ pages->iov = NULL;
+}
+
static void terminate_multifd_send_threads(void)
{
int i;
@@ -419,9 +441,11 @@ void multifd_save_cleanup(void)
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem);
socket_send_channel_destroy(p->c);
+ multifd_clear_group(&p->pages);
}
g_free(multifd_send_state->params);
multifd_send_state->params = NULL;
+ multifd_clear_group(&multifd_send_state->pages);
g_free(multifd_send_state);
multifd_send_state = NULL;
}
@@ -454,8 +478,8 @@ static void *multifd_send_thread(void *opaque)
qemu_mutex_unlock(&p->mutex);
break;
}
- if (p->address) {
- p->address = 0;
+ if (p->pages.num) {
+ p->pages.num = 0;
qemu_mutex_unlock(&p->mutex);
qemu_mutex_lock(&multifd_send_state->mutex);
p->done = true;
@@ -484,6 +508,7 @@ int multifd_save_setup(void)
multifd_send_state->count = 0;
qemu_mutex_init(&multifd_send_state->mutex);
qemu_sem_init(&multifd_send_state->sem, 0);
+ multifd_init_group(&multifd_send_state->pages);
for (i = 0; i < thread_count; i++) {
char thread_name[16];
MultiFDSendParams *p = &multifd_send_state->params[i];
@@ -493,7 +518,7 @@ int multifd_save_setup(void)
p->quit = false;
p->id = i;
p->done = true;
- p->address = 0;
+ multifd_init_group(&p->pages);
p->c = socket_send_channel_create();
if (!p->c) {
error_report("Error creating a send channel");
@@ -512,6 +537,17 @@ static uint16_t multifd_send_page(uint8_t *address, bool last_page)
{
int i;
MultiFDSendParams *p = NULL; /* make happy gcc */
+ multifd_pages_t *pages = &multifd_send_state->pages;
+
+ pages->iov[pages->num].iov_base = address;
+ pages->iov[pages->num].iov_len = TARGET_PAGE_SIZE;
+ pages->num++;
+
+ if (!last_page) {
+ if (pages->num < (pages->size - 1)) {
+ return MULTIFD_CONTINUE;
+ }
+ }
qemu_sem_wait(&multifd_send_state->sem);
qemu_mutex_lock(&multifd_send_state->mutex);
@@ -525,7 +561,10 @@ static uint16_t multifd_send_page(uint8_t *address, bool last_page)
}
qemu_mutex_unlock(&multifd_send_state->mutex);
qemu_mutex_lock(&p->mutex);
- p->address = address;
+ p->pages.num = pages->num;
+ iov_copy(p->pages.iov, pages->num, pages->iov, pages->num, 0,
+ iov_size(pages->iov, pages->num));
+ pages->num = 0;
qemu_mutex_unlock(&p->mutex);
qemu_sem_post(&p->sem);
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 14/19] migration: Send the fd number which we are going to use for this page
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (12 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 13/19] migration: Really use multiple pages at a time Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-14 15:47 ` Dr. David Alan Gilbert
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side Juan Quintela
` (5 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We are still sending the page through the main channel, that would
change later in the series
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
migration/ram.c | 13 +++++++++++--
1 file changed, 11 insertions(+), 2 deletions(-)
diff --git a/migration/ram.c b/migration/ram.c
index 7310da9..169cfc9 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -568,7 +568,7 @@ static uint16_t multifd_send_page(uint8_t *address, bool last_page)
qemu_mutex_unlock(&p->mutex);
qemu_sem_post(&p->sem);
- return 0;
+ return i;
}
struct MultiFDRecvParams {
@@ -1145,6 +1145,7 @@ static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
bool last_stage)
{
int pages;
+ uint16_t fd_num;
uint8_t *p;
RAMBlock *block = pss->block;
ram_addr_t offset = pss->page << TARGET_PAGE_BITS;
@@ -1156,8 +1157,10 @@ static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
ram_counters.transferred +=
save_page_header(rs, rs->f, block,
offset | RAM_SAVE_FLAG_MULTIFD_PAGE);
+ fd_num = multifd_send_page(p, rs->migration_dirty_pages == 1);
+ qemu_put_be16(rs->f, fd_num);
+ ram_counters.transferred += 2; /* size of fd_num */
qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
- multifd_send_page(p, rs->migration_dirty_pages == 1);
ram_counters.transferred += TARGET_PAGE_SIZE;
pages = 1;
ram_counters.normal++;
@@ -2907,6 +2910,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
while (!postcopy_running && !ret && !(flags & RAM_SAVE_FLAG_EOS)) {
ram_addr_t addr, total_ram_bytes;
void *host = NULL;
+ uint16_t fd_num;
uint8_t ch;
addr = qemu_get_be64(f);
@@ -3017,6 +3021,11 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
break;
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;
+ }
qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
break;
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (13 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 14/19] migration: Send the fd number which we are going to use for this page Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 4:35 ` Peter Xu
2017-08-11 15:29 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure Juan Quintela
` (4 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
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 <quintela@redhat.com>
--
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_group().
---
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 8e9505a..b78dffc 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -381,7 +381,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);
@@ -400,9 +400,10 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
if (!mis->from_src_file) {
QEMUFile *f = qemu_fopen_channel_input(ioc);
mis->from_src_file = f;
- 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 ddf8c89..1442b33 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -154,6 +154,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 169cfc9..eb0015e 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -572,12 +572,17 @@ static uint16_t multifd_send_page(uint8_t *address, bool last_page)
}
struct MultiFDRecvParams {
+ /* not changed */
uint8_t id;
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;
@@ -587,6 +592,7 @@ struct {
int count;
/* Should we finish */
bool quit;
+ multifd_pages_t pages;
} *multifd_recv_state;
static void terminate_multifd_recv_threads(void)
@@ -602,6 +608,7 @@ static void terminate_multifd_recv_threads(void)
p->quit = true;
qemu_sem_post(&p->sem);
qemu_mutex_unlock(&p->mutex);
+ multifd_clear_group(&p->pages);
}
}
@@ -625,6 +632,7 @@ void multifd_load_cleanup(void)
}
g_free(multifd_recv_state->params);
multifd_recv_state->params = NULL;
+ multifd_clear_group(&multifd_recv_state->pages);
g_free(multifd_recv_state);
multifd_recv_state = NULL;
}
@@ -633,12 +641,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);
}
@@ -682,8 +698,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_group(&p->pages);
p->c = ioc;
atomic_set(&multifd_recv_state->params[id], p);
multifd_recv_state->count++;
@@ -703,6 +722,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_group(&multifd_recv_state->pages);
return 0;
}
@@ -711,6 +731,36 @@ int multifd_created_threads(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_threads();
+ 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
*
@@ -3022,10 +3072,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 5dd6f42..3af9f7c 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -183,12 +183,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.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (14 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 15:32 ` Daniel P. Berrange
2017-08-11 15:35 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 17/19] migration: Rename initial_bytes Juan Quintela
` (3 subsequent siblings)
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We just send the address through the alternate channels and test that it
is ok.
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
migration/ram.c | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 50 insertions(+)
diff --git a/migration/ram.c b/migration/ram.c
index eb0015e..42ad126 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -479,8 +479,26 @@ static void *multifd_send_thread(void *opaque)
break;
}
if (p->pages.num) {
+ int i;
+ int num;
+
+ num = p->pages.num;
p->pages.num = 0;
qemu_mutex_unlock(&p->mutex);
+
+ for (i = 0; i < num; i++) {
+ if (qio_channel_write(p->c,
+ (const char *)&p->pages.iov[i].iov_base,
+ sizeof(uint8_t *), &error_abort)
+ != sizeof(uint8_t *)) {
+ MigrationState *s = migrate_get_current();
+
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_send_threads();
+ return NULL;
+ }
+ }
qemu_mutex_lock(&multifd_send_state->mutex);
p->done = true;
qemu_mutex_unlock(&multifd_send_state->mutex);
@@ -640,6 +658,7 @@ void multifd_load_cleanup(void)
static void *multifd_recv_thread(void *opaque)
{
MultiFDRecvParams *p = opaque;
+ uint8_t *recv_address;
qemu_sem_post(&p->ready);
while (true) {
@@ -649,7 +668,38 @@ static void *multifd_recv_thread(void *opaque)
break;
}
if (p->pages.num) {
+ int i;
+ int num;
+
+ num = p->pages.num;
p->pages.num = 0;
+
+ for (i = 0; i < num; i++) {
+ if (qio_channel_read(p->c,
+ (char *)&recv_address,
+ sizeof(uint8_t *), &error_abort)
+ != sizeof(uint8_t *)) {
+ MigrationState *s = migrate_get_current();
+
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return NULL;
+ }
+ if (recv_address != p->pages.iov[i].iov_base) {
+ MigrationState *s = migrate_get_current();
+
+ printf("We received %p what we were expecting %p (%d)\n",
+ recv_address,
+ p->pages.iov[i].iov_base, i);
+
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return NULL;
+ }
+ }
+
p->done = true;
qemu_mutex_unlock(&p->mutex);
qemu_sem_post(&p->ready);
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 17/19] migration: Rename initial_bytes
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (15 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 19:01 ` Dr. David Alan Gilbert
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels Juan Quintela
` (2 subsequent siblings)
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Now it is called qemu_file_bytes that reflects better what it does,
and we create qemu_file_bytes_now to not have to call qemu_ftell() twice.
Signed-off-by: Juan Quintela <quintela@redhat.com>
---
migration/migration.c | 9 +++++----
1 file changed, 5 insertions(+), 4 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index b78dffc..974ff92 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -2044,13 +2044,13 @@ static void *migration_thread(void *opaque)
/* Used by the bandwidth calcs, updated later */
int64_t initial_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
int64_t setup_start = qemu_clock_get_ms(QEMU_CLOCK_HOST);
- int64_t initial_bytes = 0;
/*
* The final stage happens when the remaining data is smaller than
* this threshold; it's calculated from the requested downtime and
* measured bandwidth
*/
int64_t threshold_size = 0;
+ int64_t qemu_file_bytes = 0;
int64_t start_time = initial_time;
int64_t end_time;
bool old_vm_running = false;
@@ -2138,8 +2138,9 @@ static void *migration_thread(void *opaque)
}
current_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
if (current_time >= initial_time + BUFFER_DELAY) {
- uint64_t transferred_bytes = qemu_ftell(s->to_dst_file) -
- initial_bytes;
+ uint64_t qemu_file_bytes_now = qemu_ftell(s->to_dst_file);
+ uint64_t transferred_bytes =
+ qemu_file_bytes_now - qemu_file_bytes;
uint64_t time_spent = current_time - initial_time;
double bandwidth = (double)transferred_bytes / time_spent;
threshold_size = bandwidth * s->parameters.downtime_limit;
@@ -2158,7 +2159,7 @@ static void *migration_thread(void *opaque)
qemu_file_reset_rate_limit(s->to_dst_file);
initial_time = current_time;
- initial_bytes = qemu_ftell(s->to_dst_file);
+ qemu_file_bytes = qemu_file_bytes_now;
}
if (qemu_file_rate_limit(s->to_dst_file)) {
/* usleep expects microseconds */
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (16 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 17/19] migration: Rename initial_bytes Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 15:34 ` Daniel P. Berrange
2017-08-16 16:38 ` Dr. David Alan Gilbert
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 19/19] migration: Flush receive queue Juan Quintela
2017-08-08 23:37 ` [Qemu-devel] [PATCH v6 00/19] Multifd no-reply
19 siblings, 2 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We switch for sending the page number to send real pages.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Remove the HACK bit, now we have the function that calculates the size
of a page exported.
---
migration/migration.c | 7 +++++-
migration/ram.c | 59 +++++++++++++++++----------------------------------
2 files changed, 25 insertions(+), 41 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index 974ff92..aac3cdc 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -2051,6 +2051,7 @@ static void *migration_thread(void *opaque)
*/
int64_t threshold_size = 0;
int64_t qemu_file_bytes = 0;
+ int64_t multifd_pages = 0;
int64_t start_time = initial_time;
int64_t end_time;
bool old_vm_running = false;
@@ -2139,8 +2140,11 @@ static void *migration_thread(void *opaque)
current_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
if (current_time >= initial_time + BUFFER_DELAY) {
uint64_t qemu_file_bytes_now = qemu_ftell(s->to_dst_file);
+ uint64_t multifd_pages_now = ram_counters.multifd;
uint64_t transferred_bytes =
- qemu_file_bytes_now - qemu_file_bytes;
+ (qemu_file_bytes_now - qemu_file_bytes) +
+ (multifd_pages_now - multifd_pages) *
+ qemu_target_page_size();
uint64_t time_spent = current_time - initial_time;
double bandwidth = (double)transferred_bytes / time_spent;
threshold_size = bandwidth * s->parameters.downtime_limit;
@@ -2160,6 +2164,7 @@ static void *migration_thread(void *opaque)
qemu_file_reset_rate_limit(s->to_dst_file);
initial_time = current_time;
qemu_file_bytes = qemu_file_bytes_now;
+ multifd_pages = multifd_pages_now;
}
if (qemu_file_rate_limit(s->to_dst_file)) {
/* usleep expects microseconds */
diff --git a/migration/ram.c b/migration/ram.c
index 42ad126..f337360 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -479,25 +479,21 @@ static void *multifd_send_thread(void *opaque)
break;
}
if (p->pages.num) {
- int i;
int num;
num = p->pages.num;
p->pages.num = 0;
qemu_mutex_unlock(&p->mutex);
- for (i = 0; i < num; i++) {
- if (qio_channel_write(p->c,
- (const char *)&p->pages.iov[i].iov_base,
- sizeof(uint8_t *), &error_abort)
- != sizeof(uint8_t *)) {
- MigrationState *s = migrate_get_current();
+ if (qio_channel_writev_all(p->c, p->pages.iov,
+ num, &error_abort)
+ != num * TARGET_PAGE_SIZE) {
+ MigrationState *s = migrate_get_current();
- migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
- MIGRATION_STATUS_FAILED);
- terminate_multifd_send_threads();
- return NULL;
- }
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_send_threads();
+ return NULL;
}
qemu_mutex_lock(&multifd_send_state->mutex);
p->done = true;
@@ -658,7 +654,6 @@ void multifd_load_cleanup(void)
static void *multifd_recv_thread(void *opaque)
{
MultiFDRecvParams *p = opaque;
- uint8_t *recv_address;
qemu_sem_post(&p->ready);
while (true) {
@@ -668,38 +663,21 @@ static void *multifd_recv_thread(void *opaque)
break;
}
if (p->pages.num) {
- int i;
int num;
num = p->pages.num;
p->pages.num = 0;
- for (i = 0; i < num; i++) {
- if (qio_channel_read(p->c,
- (char *)&recv_address,
- sizeof(uint8_t *), &error_abort)
- != sizeof(uint8_t *)) {
- MigrationState *s = migrate_get_current();
+ if (qio_channel_readv_all(p->c, p->pages.iov,
+ num, &error_abort)
+ != num * TARGET_PAGE_SIZE) {
+ MigrationState *s = migrate_get_current();
- migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
- MIGRATION_STATUS_FAILED);
- terminate_multifd_recv_threads();
- return NULL;
- }
- if (recv_address != p->pages.iov[i].iov_base) {
- MigrationState *s = migrate_get_current();
-
- printf("We received %p what we were expecting %p (%d)\n",
- recv_address,
- p->pages.iov[i].iov_base, i);
-
- migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
- MIGRATION_STATUS_FAILED);
- terminate_multifd_recv_threads();
- return NULL;
- }
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return NULL;
}
-
p->done = true;
qemu_mutex_unlock(&p->mutex);
qemu_sem_post(&p->ready);
@@ -1259,8 +1237,10 @@ static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
offset | RAM_SAVE_FLAG_MULTIFD_PAGE);
fd_num = multifd_send_page(p, rs->migration_dirty_pages == 1);
qemu_put_be16(rs->f, fd_num);
+ if (fd_num != MULTIFD_CONTINUE) {
+ qemu_fflush(rs->f);
+ }
ram_counters.transferred += 2; /* size of fd_num */
- qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
ram_counters.transferred += TARGET_PAGE_SIZE;
pages = 1;
ram_counters.normal++;
@@ -3123,7 +3103,6 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
case RAM_SAVE_FLAG_MULTIFD_PAGE:
fd_num = qemu_get_be16(f);
multifd_recv_page(host, fd_num);
- qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
break;
case RAM_SAVE_FLAG_EOS:
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 19/19] migration: Flush receive queue
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (17 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels Juan Quintela
@ 2017-08-08 16:26 ` Juan Quintela
2017-08-11 4:16 ` Peter Xu
2017-08-08 23:37 ` [Qemu-devel] [PATCH v6 00/19] Multifd no-reply
19 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:26 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
Each time that we sync the bitmap, it is a possiblity that we receive
a page that is being processed by a different thread. We fix this
problem just making sure that we wait for all receiving threads to
finish its work before we procedeed with the next stage.
We are low on page flags, so we use a combination that is not valid to
emit that message: MULTIFD_PAGE and COMPRESSED.
I tried to make a migration command for it, but it don't work because
we sync the bitmap sometimes when we have already sent the beggining
of the section, so I just added a new page flag.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Create RAM_SAVE_FLAG_MULTIFD_SYNC (dave suggestion)
Move the set of need_flush to inside the bitmap_sync code (peter suggestion)
---
migration/ram.c | 54 ++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 54 insertions(+)
diff --git a/migration/ram.c b/migration/ram.c
index f337360..ee08fd2 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -71,6 +71,13 @@
#define RAM_SAVE_FLAG_COMPRESS_PAGE 0x100
#define RAM_SAVE_FLAG_MULTIFD_PAGE 0x200
+/* We are getting low on pages flags, so we start using combinations
+ When we need to flush a page, we sent it as
+ RAM_SAVE_FLAG_MULTIFD_PAGE | RAM_SAVE_FLAG_COMPRESS_PAGE
+ We don't allow that combination
+*/
+#define RAM_SAVE_FLAG_MULTIFD_SYNC (RAM_SAVE_FLAG_MULTIFD_PAGE | RAM_SAVE_FLAG_ZERO)
+
static inline bool is_zero_range(uint8_t *p, uint64_t size)
{
return buffer_is_zero(p, size);
@@ -193,6 +200,9 @@ struct RAMState {
uint64_t iterations_prev;
/* Iterations since start */
uint64_t iterations;
+ /* Indicates if we have synced the bitmap and we need to assure that
+ target has processeed all previous pages */
+ bool multifd_needs_flush;
/* number of dirty bits in the bitmap */
uint64_t migration_dirty_pages;
/* protects modification of the bitmap */
@@ -592,9 +602,11 @@ struct MultiFDRecvParams {
QIOChannel *c;
QemuSemaphore ready;
QemuSemaphore sem;
+ QemuCond cond_sync;
QemuMutex mutex;
/* proteced by param mutex */
bool quit;
+ bool sync;
multifd_pages_t pages;
bool done;
};
@@ -640,6 +652,7 @@ void multifd_load_cleanup(void)
qemu_thread_join(&p->thread);
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem);
+ qemu_cond_destroy(&p->cond_sync);
socket_recv_channel_destroy(p->c);
g_free(p);
multifd_recv_state->params[i] = NULL;
@@ -679,6 +692,10 @@ static void *multifd_recv_thread(void *opaque)
return NULL;
}
p->done = true;
+ if (p->sync) {
+ qemu_cond_signal(&p->cond_sync);
+ p->sync = false;
+ }
qemu_mutex_unlock(&p->mutex);
qemu_sem_post(&p->ready);
continue;
@@ -727,9 +744,11 @@ void multifd_new_channel(QIOChannel *ioc)
qemu_mutex_init(&p->mutex);
qemu_sem_init(&p->sem, 0);
qemu_sem_init(&p->ready, 0);
+ qemu_cond_init(&p->cond_sync);
p->quit = false;
p->id = id;
p->done = false;
+ p->sync = false;
multifd_init_group(&p->pages);
p->c = ioc;
atomic_set(&multifd_recv_state->params[id], p);
@@ -789,6 +808,27 @@ static void multifd_recv_page(uint8_t *address, uint16_t fd_num)
qemu_sem_post(&p->sem);
}
+static int multifd_flush(void)
+{
+ int i, thread_count;
+
+ if (!migrate_use_multifd()) {
+ return 0;
+ }
+ thread_count = migrate_multifd_threads();
+ for (i = 0; i < thread_count; i++) {
+ MultiFDRecvParams *p = multifd_recv_state->params[i];
+
+ qemu_mutex_lock(&p->mutex);
+ while (!p->done) {
+ p->sync = true;
+ qemu_cond_wait(&p->cond_sync, &p->mutex);
+ }
+ qemu_mutex_unlock(&p->mutex);
+ }
+ return 0;
+}
+
/**
* save_page_header: write page header to wire
*
@@ -806,6 +846,12 @@ static size_t save_page_header(RAMState *rs, QEMUFile *f, RAMBlock *block,
{
size_t size, len;
+ if (rs->multifd_needs_flush &&
+ (offset & RAM_SAVE_FLAG_MULTIFD_PAGE)) {
+ offset |= RAM_SAVE_FLAG_ZERO;
+ rs->multifd_needs_flush = false;
+ }
+
if (block == rs->last_sent_block) {
offset |= RAM_SAVE_FLAG_CONTINUE;
}
@@ -1091,6 +1137,9 @@ static void migration_bitmap_sync(RAMState *rs)
if (migrate_use_events()) {
qapi_event_send_migration_pass(ram_counters.dirty_sync_count, NULL);
}
+ if (!rs->ram_bulk_stage && migrate_use_multifd()) {
+ rs->multifd_needs_flush = true;
+ }
}
/**
@@ -3009,6 +3058,11 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
break;
}
+ if ((flags & RAM_SAVE_FLAG_MULTIFD_SYNC)
+ == RAM_SAVE_FLAG_MULTIFD_SYNC) {
+ multifd_flush();
+ flags = flags & ~RAM_SAVE_FLAG_ZERO;
+ }
if (flags & (RAM_SAVE_FLAG_ZERO | RAM_SAVE_FLAG_PAGE |
RAM_SAVE_FLAG_COMPRESS_PAGE | RAM_SAVE_FLAG_XBZRLE |
RAM_SAVE_FLAG_MULTIFD_PAGE)) {
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 00/19] Multifd
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
` (18 preceding siblings ...)
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 19/19] migration: Flush receive queue Juan Quintela
@ 2017-08-08 23:37 ` no-reply
19 siblings, 0 replies; 61+ messages in thread
From: no-reply @ 2017-08-08 23:37 UTC (permalink / raw)
To: quintela; +Cc: famz, qemu-devel, lvivier, dgilbert, peterx
Hi,
This series seems to have some coding style problems. See output below for
more information:
Message-id: 20170808162629.32493-1-quintela@redhat.com
Subject: [Qemu-devel] [PATCH v6 00/19] Multifd
Type: series
=== TEST SCRIPT BEGIN ===
#!/bin/bash
BASE=base
n=1
total=$(git log --oneline $BASE.. | wc -l)
failed=0
git config --local diff.renamelimit 0
git config --local diff.renames True
commits="$(git log --format=%H --reverse $BASE..)"
for c in $commits; do
echo "Checking PATCH $n/$total: $(git log -n 1 --format=%s $c)..."
if ! git show $c --format=email | ./scripts/checkpatch.pl --mailback -; then
failed=1
echo
fi
n=$((n+1))
done
exit $failed
=== TEST SCRIPT END ===
Updating 3c8cf5a9c21ff8782164d1def7f44bd888713384
Switched to a new branch 'test'
00a9b5e8ae migration: Flush receive queue
e916ec0901 migration: Transfer pages over new channels
399ab56582 migration: Rename initial_bytes
f7873b901c migration: Test new fd infrastructure
881e5e18a6 migration: Create thread infrastructure for multifd recv side
6a65acd553 migration: Send the fd number which we are going to use for this page
da613ef42c migration: Really use multiple pages at a time
f4bedd6cc4 migration: Create ram_multifd_page
7eefb24552 migration: Start of multiple fd work
c1cfe5f5ca migration: Split migration_fd_process_incoming
8a3e50c51c migration: Create multifd migration threads
59cc6bad61 migration: Create x-multifd-group parameter
53f8404fa4 migration: Create x-multifd-threads parameter
acf538cb8e migration: Add multifd capability
4cbb7c4374 qio: Create new qio_channel_{readv, writev}_all
eec3514870 migration: Create migration_has_all_channels
0ccdee3bbc migration: Add comments to channel functions
e53ddce966 migration: Teach it about G_SOURCE_REMOVE
e64d5e3c8a migration: Create migration_ioc_process_incoming()
=== OUTPUT BEGIN ===
Checking PATCH 1/19: migration: Create migration_ioc_process_incoming()...
ERROR: trailing whitespace
#59: FILE: migration/migration.c:369:
+ $
total: 1 errors, 0 warnings, 72 lines checked
Your patch has style problems, please review. If any of these errors
are false positives report them to the maintainer, see
CHECKPATCH in MAINTAINERS.
Checking PATCH 2/19: migration: Teach it about G_SOURCE_REMOVE...
Checking PATCH 3/19: migration: Add comments to channel functions...
Checking PATCH 4/19: migration: Create migration_has_all_channels...
Checking PATCH 5/19: qio: Create new qio_channel_{readv, writev}_all...
Checking PATCH 6/19: migration: Add multifd capability...
Checking PATCH 7/19: migration: Create x-multifd-threads parameter...
Checking PATCH 8/19: migration: Create x-multifd-group parameter...
Checking PATCH 9/19: migration: Create multifd migration threads...
Checking PATCH 10/19: migration: Split migration_fd_process_incoming...
Checking PATCH 11/19: migration: Start of multiple fd work...
Checking PATCH 12/19: migration: Create ram_multifd_page...
Checking PATCH 13/19: migration: Really use multiple pages at a time...
Checking PATCH 14/19: migration: Send the fd number which we are going to use for this page...
Checking PATCH 15/19: migration: Create thread infrastructure for multifd recv side...
Checking PATCH 16/19: migration: Test new fd infrastructure...
Checking PATCH 17/19: migration: Rename initial_bytes...
Checking PATCH 18/19: migration: Transfer pages over new channels...
Checking PATCH 19/19: migration: Flush receive queue...
WARNING: line over 80 characters
#38: FILE: migration/ram.c:79:
+#define RAM_SAVE_FLAG_MULTIFD_SYNC (RAM_SAVE_FLAG_MULTIFD_PAGE | RAM_SAVE_FLAG_ZERO)
total: 0 errors, 1 warnings, 120 lines checked
Your patch has style problems, please review. If any of these errors
are false positives report them to the maintainer, see
CHECKPATCH in MAINTAINERS.
=== OUTPUT END ===
Test command exited with code: 1
---
Email generated automatically by Patchew [http://patchew.org/].
Please send your feedback to patchew-devel@freelists.org
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE Juan Quintela
@ 2017-08-10 14:56 ` Daniel P. Berrange
2017-08-11 3:49 ` Peter Xu
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-10 14:56 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:12PM +0200, Juan Quintela wrote:
> As this is defined on glib 2.32, add compatibility macros for older glibs.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> include/glib-compat.h | 2 ++
> migration/exec.c | 2 +-
> migration/fd.c | 2 +-
> migration/socket.c | 2 +-
> 4 files changed, 5 insertions(+), 3 deletions(-)
Reviewed-by: Daniel P. Berrange <berrange@redhat.com>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 03/19] migration: Add comments to channel functions
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 03/19] migration: Add comments to channel functions Juan Quintela
@ 2017-08-10 14:57 ` Daniel P. Berrange
0 siblings, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-10 14:57 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:13PM +0200, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/channel.c | 15 +++++++++++++++
> 1 file changed, 15 insertions(+)
>
> diff --git a/migration/channel.c b/migration/channel.c
> index edceebd..79b0f8b 100644
> --- a/migration/channel.c
> +++ b/migration/channel.c
> @@ -19,6 +19,14 @@
> #include "qapi/error.h"
> #include "io/channel-tls.h"
>
> +/**
> + * @migration_channel_process_incoming - Create new incoming migration channel
> + *
> + * Notice that TLS is special. For it we listen in a listener socket,
> + * and then create a new client socket from the TLS library.
> + *
> + * @ioc: Channel to wich we are connecting
> + */
s/wich/which/
> void migration_channel_process_incoming(QIOChannel *ioc)
> {
> MigrationState *s = migrate_get_current();
> @@ -41,6 +49,13 @@ void migration_channel_process_incoming(QIOChannel *ioc)
> }
>
>
> +/**
> + * @migration_channel_connect - Create new outgoing migration channel
> + *
> + * @s: Current migration state
> + * @ioc: Channel to wich we are connecting
s/wich/which/
> + * @hostname: Where we want to connect
> + */
> void migration_channel_connect(MigrationState *s,
> QIOChannel *ioc,
> const char *hostname)
> --
> 2.9.4
With those typos fixed, you can add
Reviewed-by: Daniel P. Berrange <berrange@redhat.com>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels Juan Quintela
@ 2017-08-10 14:58 ` Daniel P. Berrange
2017-08-10 15:17 ` Eric Blake
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-10 14:58 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:14PM +0200, Juan Quintela wrote:
> This functions allows us to decide when to close the listener socket.
> For now, we only need one connection.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/migration.c | 11 +++++++++++
> migration/migration.h | 2 ++
> migration/socket.c | 10 +++++++---
> 3 files changed, 20 insertions(+), 3 deletions(-)
Reviewed-by: Daniel P. Berrange <berrange@redhat.com>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 05/19] qio: Create new qio_channel_{readv, writev}_all
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 05/19] qio: Create new qio_channel_{readv, writev}_all Juan Quintela
@ 2017-08-10 15:04 ` Daniel P. Berrange
0 siblings, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-10 15:04 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:15PM +0200, Juan Quintela wrote:
> The functions waits until it is able to write the full iov.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
> --
>
> Add tests.
>
> fix reader to check for len == 0.
> ---
> include/io/channel.h | 46 +++++++++++++++++++++++++
> io/channel.c | 77 ++++++++++++++++++++++++++++++++++++++++++
> migration/qemu-file-channel.c | 29 +---------------
> tests/io-channel-helpers.c | 55 ++++++++++++++++++++++++++++++
> tests/io-channel-helpers.h | 4 +++
> tests/test-io-channel-buffer.c | 55 ++++++++++++++++++++++++++++--
> 6 files changed, 235 insertions(+), 31 deletions(-)
>
> diff --git a/include/io/channel.h b/include/io/channel.h
> index db9bb02..bfc97e2 100644
> --- a/include/io/channel.h
> +++ b/include/io/channel.h
> @@ -269,6 +269,52 @@ ssize_t qio_channel_writev_full(QIOChannel *ioc,
> Error **errp);
>
> /**
> + * qio_channel_readv_all:
> + * @ioc: the channel object
> + * @iov: the array of memory regions to read data into
> + * @niov: the length of the @iov array
> + * @errp: pointer to a NULL-initialized error object
> + *
> + * Read data from the IO channel, storing it in the
> + * memory regions referenced by @iov. Each element
> + * in the @iov will be fully populated with data
> + * before the next one is used. The @niov parameter
> + * specifies the total number of elements in @iov.
> + *
> + * Returns: the number of bytes read, or -1 on error,
> + * or QIO_CHANNEL_ERR_BLOCK if no data is available
> + * and the channel is non-blocking
This is incorrect - it'll never return QIO_CHANNEL_ERR_BLOCK.
If it seems that, it'll go into a wait until data arrives.
> + */
> +ssize_t qio_channel_readv_all(QIOChannel *ioc,
> + const struct iovec *iov,
> + size_t niov,
> + Error **errp);
> +
> +
> +/**
> + * qio_channel_writev_all:
> + * @ioc: the channel object
> + * @iov: the array of memory regions to write data from
> + * @niov: the length of the @iov array
> + * @errp: pointer to a NULL-initialized error object
> + *
> + * Write data to the IO channel, reading it from the
> + * memory regions referenced by @iov. Each element
> + * in the @iov will be fully sent, before the next
> + * one is used. The @niov parameter specifies the
> + * total number of elements in @iov.
> + *
> + * It is required for all @iov data to be fully
> + * sent.
> + *
> + * Returns: the number of bytes sent, or -1 on error,
> + */
> +ssize_t qio_channel_writev_all(QIOChannel *ioc,
> + const struct iovec *iov,
> + size_t niov,
> + Error **erp);
> +
> +/**
> * qio_channel_readv:
> * @ioc: the channel object
> * @iov: the array of memory regions to read data into
> diff --git a/io/channel.c b/io/channel.c
> index 1cfb8b3..0b521f9 100644
> --- a/io/channel.c
> +++ b/io/channel.c
> @@ -22,6 +22,7 @@
> #include "io/channel.h"
> #include "qapi/error.h"
> #include "qemu/main-loop.h"
> +#include "qemu/iov.h"
>
> bool qio_channel_has_feature(QIOChannel *ioc,
> QIOChannelFeature feature)
> @@ -85,6 +86,82 @@ ssize_t qio_channel_writev_full(QIOChannel *ioc,
> }
>
>
> +
> +ssize_t qio_channel_readv_all(QIOChannel *ioc,
> + const struct iovec *iov,
> + size_t niov,
> + Error **errp)
> +{
> + ssize_t done = 0;
> + struct iovec *local_iov = g_new(struct iovec, niov);
> + struct iovec *local_iov_head = local_iov;
> + unsigned int nlocal_iov = niov;
Should be size_t
> +
> + nlocal_iov = iov_copy(local_iov, nlocal_iov,
> + iov, niov,
> + 0, iov_size(iov, niov));
> +
> + while (nlocal_iov > 0) {
> + ssize_t len;
> + len = qio_channel_readv(ioc, local_iov, nlocal_iov, errp);
> + if (len == QIO_CHANNEL_ERR_BLOCK) {
> + qio_channel_wait(ioc, G_IO_OUT);
This should be waiting for G_IO_IN
> + continue;
> + } else if (len < 0) {
> + error_setg_errno(errp, EIO,
> + "Channel was not able to read full iov");
> + done = -1;
> + goto cleanup;
> + } else if (len == 0) {
> + goto cleanup;
> + }
> +
> + iov_discard_front(&local_iov, &nlocal_iov, len);
> + done += len;
> + }
> +
> + cleanup:
> + g_free(local_iov_head);
> + return done;
> +}
> +
> +ssize_t qio_channel_writev_all(QIOChannel *ioc,
> + const struct iovec *iov,
> + size_t niov,
> + Error **errp)
> +{
> + ssize_t done = 0;
> + struct iovec *local_iov = g_new(struct iovec, niov);
> + struct iovec *local_iov_head = local_iov;
> + unsigned int nlocal_iov = niov;
Should be size_t
> +
> + nlocal_iov = iov_copy(local_iov, nlocal_iov,
> + iov, niov,
> + 0, iov_size(iov, niov));
> +
> + while (nlocal_iov > 0) {
> + ssize_t len;
> + len = qio_channel_writev(ioc, local_iov, nlocal_iov, errp);
> + if (len == QIO_CHANNEL_ERR_BLOCK) {
> + qio_channel_wait(ioc, G_IO_OUT);
> + continue;
> + }
> + if (len < 0) {
> + error_setg_errno(errp, EIO,
> + "Channel was not able to write full iov");
> + done = -1;
> + goto cleanup;
> + }
> +
> + iov_discard_front(&local_iov, &nlocal_iov, len);
> + done += len;
> + }
> +
> + cleanup:
> + g_free(local_iov_head);
> + return done;
> +}
> +
> ssize_t qio_channel_readv(QIOChannel *ioc,
> const struct iovec *iov,
> size_t niov,
> diff --git a/migration/qemu-file-channel.c b/migration/qemu-file-channel.c
> index e202d73..457ea6c 100644
> --- a/migration/qemu-file-channel.c
> +++ b/migration/qemu-file-channel.c
> @@ -36,35 +36,8 @@ static ssize_t channel_writev_buffer(void *opaque,
> int64_t pos)
> {
> QIOChannel *ioc = QIO_CHANNEL(opaque);
> - ssize_t done = 0;
> - struct iovec *local_iov = g_new(struct iovec, iovcnt);
> - struct iovec *local_iov_head = local_iov;
> - unsigned int nlocal_iov = iovcnt;
>
> - nlocal_iov = iov_copy(local_iov, nlocal_iov,
> - iov, iovcnt,
> - 0, iov_size(iov, iovcnt));
> -
> - while (nlocal_iov > 0) {
> - ssize_t len;
> - len = qio_channel_writev(ioc, local_iov, nlocal_iov, NULL);
> - if (len == QIO_CHANNEL_ERR_BLOCK) {
> - qio_channel_wait(ioc, G_IO_OUT);
> - continue;
> - }
> - if (len < 0) {
> - /* XXX handle Error objects */
> - done = -EIO;
> - goto cleanup;
> - }
> -
> - iov_discard_front(&local_iov, &nlocal_iov, len);
> - done += len;
> - }
> -
> - cleanup:
> - g_free(local_iov_head);
> - return done;
> + return qio_channel_writev_all(ioc, iov, iovcnt, NULL);
> }
>
>
> diff --git a/tests/io-channel-helpers.c b/tests/io-channel-helpers.c
> index 05e5579..3d76d95 100644
> --- a/tests/io-channel-helpers.c
> +++ b/tests/io-channel-helpers.c
> @@ -21,6 +21,7 @@
> #include "qemu/osdep.h"
> #include "io-channel-helpers.h"
> #include "qapi/error.h"
> +#include "qemu/iov.h"
>
> struct QIOChannelTest {
> QIOChannel *src;
> @@ -153,6 +154,45 @@ static gpointer test_io_thread_reader(gpointer opaque)
> return NULL;
> }
>
> +static gpointer test_io_thread_writer_all(gpointer opaque)
> +{
> + QIOChannelTest *data = opaque;
> + size_t niov = data->niov;
> + ssize_t ret;
> +
> + qio_channel_set_blocking(data->src, data->blocking, NULL);
> +
> + ret = qio_channel_writev_all(data->src,
> + data->inputv,
> + niov,
> + &data->writeerr);
> + if (ret != iov_size(data->inputv, data->niov)) {
> + error_setg(&data->writeerr, "Unexpected I/O error");
> + }
> +
> + return NULL;
> +}
> +
> +/* This thread receives all data using iovecs */
> +static gpointer test_io_thread_reader_all(gpointer opaque)
> +{
> + QIOChannelTest *data = opaque;
> + size_t niov = data->niov;
> + ssize_t ret;
> +
> + qio_channel_set_blocking(data->dst, data->blocking, NULL);
> +
> + ret = qio_channel_readv_all(data->dst,
> + data->outputv,
> + niov,
> + &data->readerr);
> +
> + if (ret != iov_size(data->inputv, data->niov)) {
> + error_setg(&data->readerr, "Unexpected I/O error");
> + }
> +
> + return NULL;
> +}
>
> QIOChannelTest *qio_channel_test_new(void)
> {
> @@ -231,6 +271,21 @@ void qio_channel_test_run_reader(QIOChannelTest *test,
> test->dst = NULL;
> }
>
> +void qio_channel_test_run_writer_all(QIOChannelTest *test,
> + QIOChannel *src)
> +{
> + test->src = src;
> + test_io_thread_writer_all(test);
> + test->src = NULL;
> +}
> +
> +void qio_channel_test_run_reader_all(QIOChannelTest *test,
> + QIOChannel *dst)
> +{
> + test->dst = dst;
> + test_io_thread_reader_all(test);
> + test->dst = NULL;
> +}
>
> void qio_channel_test_validate(QIOChannelTest *test)
> {
> diff --git a/tests/io-channel-helpers.h b/tests/io-channel-helpers.h
> index fedc64f..17b9647 100644
> --- a/tests/io-channel-helpers.h
> +++ b/tests/io-channel-helpers.h
> @@ -36,6 +36,10 @@ void qio_channel_test_run_writer(QIOChannelTest *test,
> QIOChannel *src);
> void qio_channel_test_run_reader(QIOChannelTest *test,
> QIOChannel *dst);
> +void qio_channel_test_run_writer_all(QIOChannelTest *test,
> + QIOChannel *src);
> +void qio_channel_test_run_reader_all(QIOChannelTest *test,
> + QIOChannel *dst);
>
> void qio_channel_test_validate(QIOChannelTest *test);
>
> diff --git a/tests/test-io-channel-buffer.c b/tests/test-io-channel-buffer.c
> index 64722a2..4bf64ae 100644
> --- a/tests/test-io-channel-buffer.c
> +++ b/tests/test-io-channel-buffer.c
> @@ -22,8 +22,7 @@
> #include "io/channel-buffer.h"
> #include "io-channel-helpers.h"
>
> -
> -static void test_io_channel_buf(void)
> +static void test_io_channel_buf1(void)
> {
> QIOChannelBuffer *buf;
> QIOChannelTest *test;
> @@ -39,6 +38,53 @@ static void test_io_channel_buf(void)
> object_unref(OBJECT(buf));
> }
>
> +static void test_io_channel_buf2(void)
> +{
> + QIOChannelBuffer *buf;
> + QIOChannelTest *test;
> +
> + buf = qio_channel_buffer_new(0);
> +
> + test = qio_channel_test_new();
> + qio_channel_test_run_writer_all(test, QIO_CHANNEL(buf));
> + buf->offset = 0;
> + qio_channel_test_run_reader(test, QIO_CHANNEL(buf));
> + qio_channel_test_validate(test);
> +
> + object_unref(OBJECT(buf));
> +}
> +
> +static void test_io_channel_buf3(void)
> +{
> + QIOChannelBuffer *buf;
> + QIOChannelTest *test;
> +
> + buf = qio_channel_buffer_new(0);
> +
> + test = qio_channel_test_new();
> + qio_channel_test_run_writer(test, QIO_CHANNEL(buf));
> + buf->offset = 0;
> + qio_channel_test_run_reader_all(test, QIO_CHANNEL(buf));
> + qio_channel_test_validate(test);
> +
> + object_unref(OBJECT(buf));
> +}
> +
> +static void test_io_channel_buf4(void)
> +{
> + QIOChannelBuffer *buf;
> + QIOChannelTest *test;
> +
> + buf = qio_channel_buffer_new(0);
> +
> + test = qio_channel_test_new();
> + qio_channel_test_run_writer_all(test, QIO_CHANNEL(buf));
> + buf->offset = 0;
> + qio_channel_test_run_reader_all(test, QIO_CHANNEL(buf));
> + qio_channel_test_validate(test);
> +
> + object_unref(OBJECT(buf));
> +}
xs>
> int main(int argc, char **argv)
> {
> @@ -46,6 +92,9 @@ int main(int argc, char **argv)
>
> g_test_init(&argc, &argv, NULL);
>
> - g_test_add_func("/io/channel/buf", test_io_channel_buf);
> + g_test_add_func("/io/channel/buf1", test_io_channel_buf1);
> + g_test_add_func("/io/channel/buf2", test_io_channel_buf2);
> + g_test_add_func("/io/channel/buf3", test_io_channel_buf3);
> + g_test_add_func("/io/channel/buf4", test_io_channel_buf4);
> return g_test_run();
> }
These test changes are all still wrong. The only test changes
needed should be changing 'test_io_thread_writer' and
'test_io_thread_reader' to call qio_channel_writev_all and
qio_channel_readv_all, respectively. This will acheive test
coverage across all the existnig I/O channel test cases,
which is preferrable to special casing the buffer test case
only.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels Juan Quintela
2017-08-10 14:58 ` Daniel P. Berrange
@ 2017-08-10 15:17 ` Eric Blake
1 sibling, 0 replies; 61+ messages in thread
From: Eric Blake @ 2017-08-10 15:17 UTC (permalink / raw)
To: Juan Quintela, qemu-devel; +Cc: lvivier, dgilbert, peterx
[-- Attachment #1: Type: text/plain, Size: 382 bytes --]
On 08/08/2017 11:26 AM, Juan Quintela wrote:
> This functions allows us to decide when to close the listener socket.
s/functions/function/
> For now, we only need one connection.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
--
Eric Blake, Principal Software Engineer
Red Hat, Inc. +1-919-301-3266
Virtualization: qemu.org | libvirt.org
[-- Attachment #2: OpenPGP digital signature --]
[-- Type: application/pgp-signature, Size: 619 bytes --]
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming()
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming() Juan Quintela
@ 2017-08-11 3:47 ` Peter Xu
2017-09-06 11:07 ` Juan Quintela
2017-09-13 9:41 ` Juan Quintela
2017-08-11 14:50 ` Daniel P. Berrange
1 sibling, 2 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 3:47 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:11PM +0200, Juan Quintela wrote:
[...]
> void migration_fd_process_incoming(QEMUFile *f)
> {
> - Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
> -
> + Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
> + MigrationIncomingState *mis = migration_incoming_get_current();
> +
> + if (!mis->from_src_file) {
> + mis->from_src_file = f;
[1]
> + }
> qemu_file_set_blocking(f, false);
> qemu_coroutine_enter(co);
> }
>
> +void migration_ioc_process_incoming(QIOChannel *ioc)
> +{
> + MigrationIncomingState *mis = migration_incoming_get_current();
> +
> + if (!mis->from_src_file) {
> + QEMUFile *f = qemu_fopen_channel_input(ioc);
> + mis->from_src_file = f;
Remove this line? Since migration_fd_process_incoming() will set it up
as well below at [1].
Then we can make sure there will be only one place to setup
from_src_file.
> + migration_fd_process_incoming(f);
> + }
> + /* We still only have a single channel. Nothing to do here yet */
> +}
Thanks,
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE Juan Quintela
2017-08-10 14:56 ` Daniel P. Berrange
@ 2017-08-11 3:49 ` Peter Xu
1 sibling, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 3:49 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:12PM +0200, Juan Quintela wrote:
> As this is defined on glib 2.32, add compatibility macros for older glibs.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 19/19] migration: Flush receive queue
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 19/19] migration: Flush receive queue Juan Quintela
@ 2017-08-11 4:16 ` Peter Xu
0 siblings, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 4:16 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:29PM +0200, Juan Quintela wrote:
> Each time that we sync the bitmap, it is a possiblity that we receive
> a page that is being processed by a different thread. We fix this
> problem just making sure that we wait for all receiving threads to
> finish its work before we procedeed with the next stage.
>
> We are low on page flags, so we use a combination that is not valid to
> emit that message: MULTIFD_PAGE and COMPRESSED.
>
> I tried to make a migration command for it, but it don't work because
> we sync the bitmap sometimes when we have already sent the beggining
> of the section, so I just added a new page flag.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
Oh, I just pop up an idea on whether we can just avoid introduing the
new flush command...
Can we just hash the page address into the channel number we send?
Then if we are sending the same page, it will always be on the same
channel. Since it's on the same channel, the order of arrival is also
deterministic.
It may affect performance, at least in two ways:
- it can be slower, considering we may need to wait on specific
channel when send. But it may not be a big problem since we have a
local IOV array buffer, then when the array buffer is full, the old
send() should mostly be completed I guess.
- it can be faster, considering that we don't need the flush any more,
so receiving is faster (no need to wait for all the channels to
complete on flush).
Makes any sense?
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side Juan Quintela
@ 2017-08-11 4:35 ` Peter Xu
2017-08-11 15:29 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 4:35 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:25PM +0200, Juan Quintela wrote:
> 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 <quintela@redhat.com>
>
> --
>
> 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_group().
> ---
> 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 8e9505a..b78dffc 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -381,7 +381,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);
> @@ -400,9 +400,10 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
> if (!mis->from_src_file) {
> QEMUFile *f = qemu_fopen_channel_input(ioc);
> mis->from_src_file = f;
> - migration_fd_process_incoming(f);
> + migration_incoming_setup(f);
Here now we only setup the incoming channels, but not processing it
any more. Then would it be good we rename the function name as well?
The old "migration_ioc_process_incoming" has hints that it processed
something... And...
> + return;
> }
> - /* We still only have a single channel. Nothing to do here yet */
> + multifd_new_channel(ioc);
> }
[...]
> @@ -183,12 +183,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();
... here we only added migration_incoming_process() for sockets. Would
that break fd/exec migration?
Thanks,
> return G_SOURCE_REMOVE;
> } else {
> return G_SOURCE_CONTINUE;
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability Juan Quintela
@ 2017-08-11 7:14 ` Peter Xu
2017-08-11 14:52 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 7:14 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:16PM +0200, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter Juan Quintela
@ 2017-08-11 7:15 ` Peter Xu
2017-08-11 14:56 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 7:15 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:17PM +0200, Juan Quintela wrote:
[...]
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 521e15c..3fe1a64 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -918,6 +918,7 @@
> #
> # @return-path: If enabled, migration will use the return path even
> # for precopy. (since 2.10)
> +#
(maybe this empty line belongs to previous patch?)
> # @x-multifd: Use more than one fd for migration (since 2.11)
> #
> # Since: 1.2
> @@ -1043,13 +1044,19 @@
> # migrated and the destination must already have access to the
> # same backing chain as was used on the source. (since 2.10)
> #
> +# @x-multifd-threads: Number of threads used to migrate data in
> +# parallel. This is the same number that the
> +# number of sockets used for migration.
> +# The default value is 2 (since 2.11)
> +#
> # Since: 2.4
> ##
> { 'enum': 'MigrationParameter',
> 'data': ['compress-level', 'compress-threads', 'decompress-threads',
> 'cpu-throttle-initial', 'cpu-throttle-increment',
> 'tls-creds', 'tls-hostname', 'max-bandwidth',
> - 'downtime-limit', 'x-checkpoint-delay', 'block-incremental' ] }
> + 'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
> + 'x-multifd-threads'] }
Reviewed-by: Peter Xu <peterx@redhat.com>
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter Juan Quintela
@ 2017-08-11 7:16 ` Peter Xu
2017-08-11 15:03 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-11 7:16 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:18PM +0200, Juan Quintela wrote:
> Indicates how many pages we are going to send in each batch to a multifd
> thread.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming()
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming() Juan Quintela
2017-08-11 3:47 ` Peter Xu
@ 2017-08-11 14:50 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 14:50 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:11PM +0200, Juan Quintela wrote:
> We pass the ioc instead of the fd. This will allow us to have more
> than one channel open. We also make sure that we set the
> from_src_file sooner, so we don't need to pass it as a parameter.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/channel.c | 3 +--
> migration/migration.c | 25 ++++++++++++++++++++-----
> migration/migration.h | 2 ++
> 3 files changed, 23 insertions(+), 7 deletions(-)
Reviewed-by: Daniel P. Berrange <berrange@redhat.com>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability Juan Quintela
2017-08-11 7:14 ` Peter Xu
@ 2017-08-11 14:52 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 14:52 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:16PM +0200, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>
> --
>
> Use new DEFINE_PROP
> ---
> migration/migration.c | 10 ++++++++++
> migration/migration.h | 1 +
> qapi-schema.json | 3 ++-
> 3 files changed, 13 insertions(+), 1 deletion(-)
Reviewed-by: Daniel P. Berrange <berrange@redhat.com>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter Juan Quintela
2017-08-11 7:15 ` Peter Xu
@ 2017-08-11 14:56 ` Daniel P. Berrange
2017-09-13 10:12 ` Juan Quintela
1 sibling, 1 reply; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 14:56 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:17PM +0200, Juan Quintela wrote:
> Indicates the number of threads that we would create. By default we
> create 2 threads.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>
> --
>
> Catch inconsistent defaults (eric).
> Improve comment stating that number of threads is the same than number
> of sockets
> Use new DEFIN_PROP_*
> ---
> hmp.c | 7 +++++++
> migration/migration.c | 26 ++++++++++++++++++++++++++
> migration/migration.h | 1 +
> qapi-schema.json | 25 ++++++++++++++++++++++---
> 4 files changed, 56 insertions(+), 3 deletions(-)
>
> diff --git a/hmp.c b/hmp.c
> index fd80dce..7899813 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -337,6 +337,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
> monitor_printf(mon, "%s: %s\n",
> MigrationParameter_lookup[MIGRATION_PARAMETER_BLOCK_INCREMENTAL],
> params->block_incremental ? "on" : "off");
> + monitor_printf(mon, "%s: %" PRId64 "\n",
> + MigrationParameter_lookup[MIGRATION_PARAMETER_X_MULTIFD_THREADS],
> + params->x_multifd_threads);
> }
>
> qapi_free_MigrationParameters(params);
> @@ -1622,6 +1625,10 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
> p->has_block_incremental = true;
> visit_type_bool(v, param, &p->block_incremental, &err);
> break;
> + case MIGRATION_PARAMETER_X_MULTIFD_THREADS:
> + p->has_x_multifd_threads = true;
> + visit_type_int(v, param, &p->x_multifd_threads, &err);
> + break;
> }
>
> if (err) {
> diff --git a/migration/migration.c b/migration/migration.c
> index 4e6ad87..6ed9600 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -78,6 +78,7 @@
> * Note: Please change this default value to 10000 when we support hybrid mode.
> */
> #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY 200
> +#define DEFAULT_MIGRATE_MULTIFD_THREADS 2
>
> static NotifierList migration_state_notifiers =
> NOTIFIER_LIST_INITIALIZER(migration_state_notifiers);
> @@ -483,6 +484,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
> params->x_checkpoint_delay = s->parameters.x_checkpoint_delay;
> params->has_block_incremental = true;
> params->block_incremental = s->parameters.block_incremental;
> + params->has_x_multifd_threads = true;
> + params->x_multifd_threads = s->parameters.x_multifd_threads;
>
> return params;
> }
> @@ -764,6 +767,13 @@ static bool migrate_params_check(MigrationParameters *params, Error **errp)
> "is invalid, it should be positive");
> return false;
> }
> + if (params->has_x_multifd_threads &&
> + (params->x_multifd_threads < 1 || params->x_multifd_threads > 255)) {
> + error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
> + "multifd_threads",
> + "is invalid, it should be in the range of 1 to 255");
> + return false;
> + }
>
> return true;
> }
> @@ -882,6 +892,9 @@ static void migrate_params_apply(MigrateSetParameters *params)
> if (params->has_block_incremental) {
> s->parameters.block_incremental = params->block_incremental;
> }
> + if (params->has_x_multifd_threads) {
> + s->parameters.x_multifd_threads = params->x_multifd_threads;
> + }
> }
>
> void qmp_migrate_set_parameters(MigrateSetParameters *params, Error **errp)
> @@ -1459,6 +1472,15 @@ bool migrate_use_multifd(void)
> return s->enabled_capabilities[MIGRATION_CAPABILITY_X_MULTIFD];
> }
>
> +int migrate_multifd_threads(void)
> +{
> + MigrationState *s;
> +
> + s = migrate_get_current();
> +
> + return s->parameters.x_multifd_threads;
> +}
> +
> int migrate_use_xbzrle(void)
> {
> MigrationState *s;
> @@ -2224,6 +2246,9 @@ static Property migration_properties[] = {
> DEFINE_PROP_INT64("x-checkpoint-delay", MigrationState,
> parameters.x_checkpoint_delay,
> DEFAULT_MIGRATE_X_CHECKPOINT_DELAY),
> + DEFINE_PROP_INT64("x-multifd-threads", MigrationState,
> + parameters.x_multifd_threads,
> + DEFAULT_MIGRATE_MULTIFD_THREADS),
>
> /* Migration capabilities */
> DEFINE_PROP_MIG_CAP("x-xbzrle", MIGRATION_CAPABILITY_XBZRLE),
> @@ -2281,6 +2306,7 @@ static void migration_instance_init(Object *obj)
> params->has_downtime_limit = true;
> params->has_x_checkpoint_delay = true;
> params->has_block_incremental = true;
> + params->has_x_multifd_threads = true;
> }
>
> /*
> diff --git a/migration/migration.h b/migration/migration.h
> index b7437f1..8e4803d 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -175,6 +175,7 @@ bool migrate_zero_blocks(void);
>
> bool migrate_auto_converge(void);
> bool migrate_use_multifd(void);
> +int migrate_multifd_threads(void);
>
> int migrate_use_xbzrle(void);
> int64_t migrate_xbzrle_cache_size(void);
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 521e15c..3fe1a64 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -918,6 +918,7 @@
> #
> # @return-path: If enabled, migration will use the return path even
> # for precopy. (since 2.10)
> +#
> # @x-multifd: Use more than one fd for migration (since 2.11)
> #
> # Since: 1.2
> @@ -1043,13 +1044,19 @@
> # migrated and the destination must already have access to the
> # same backing chain as was used on the source. (since 2.10)
> #
> +# @x-multifd-threads: Number of threads used to migrate data in
> +# parallel. This is the same number that the
> +# number of sockets used for migration.
> +# The default value is 2 (since 2.11)
The feature is called "multifd" but we're configuring number of threads,
with the implicit suggestion that number of threads == number of sockets.
I wonder if its better to call this "x-multifd-connections" so that it
is explicit that we're determining the number of socket connections.
This leaves room for adapting code to use a different number of threads,
but same number of sockets. eg with post-copy, there's potential to have
2 threads for each socket (1 reading, 1 writing).
> +#
> # Since: 2.4
> ##
> { 'enum': 'MigrationParameter',
> 'data': ['compress-level', 'compress-threads', 'decompress-threads',
> 'cpu-throttle-initial', 'cpu-throttle-increment',
> 'tls-creds', 'tls-hostname', 'max-bandwidth',
> - 'downtime-limit', 'x-checkpoint-delay', 'block-incremental' ] }
> + 'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
> + 'x-multifd-threads'] }
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter Juan Quintela
2017-08-11 7:16 ` Peter Xu
@ 2017-08-11 15:03 ` Daniel P. Berrange
2017-09-13 10:25 ` Juan Quintela
1 sibling, 1 reply; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:03 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:18PM +0200, Juan Quintela wrote:
> Indicates how many pages we are going to send in each batch to a multifd
> thread.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>
> --
>
> Be consistent with defaults and documentation
> Use new DEFINE_PROP_*
> ---
> hmp.c | 7 +++++++
> migration/migration.c | 26 ++++++++++++++++++++++++++
> migration/migration.h | 1 +
> qapi-schema.json | 17 ++++++++++++++---
> 4 files changed, 48 insertions(+), 3 deletions(-)
>
> diff --git a/hmp.c b/hmp.c
> index 7899813..a52035a 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -340,6 +340,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
> monitor_printf(mon, "%s: %" PRId64 "\n",
> MigrationParameter_lookup[MIGRATION_PARAMETER_X_MULTIFD_THREADS],
> params->x_multifd_threads);
> + monitor_printf(mon, "%s: %" PRId64 "\n",
> + MigrationParameter_lookup[MIGRATION_PARAMETER_X_MULTIFD_GROUP],
> + params->x_multifd_group);
> }
>
> qapi_free_MigrationParameters(params);
> @@ -1629,6 +1632,10 @@ void hmp_migrate_set_parameter(Monitor *mon, const QDict *qdict)
> p->has_x_multifd_threads = true;
> visit_type_int(v, param, &p->x_multifd_threads, &err);
> break;
> + case MIGRATION_PARAMETER_X_MULTIFD_GROUP:
> + p->has_x_multifd_group = true;
> + visit_type_int(v, param, &p->x_multifd_group, &err);
> + break;
> }
>
> if (err) {
> diff --git a/migration/migration.c b/migration/migration.c
> index 6ed9600..6c4eb4b 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -79,6 +79,7 @@
> */
> #define DEFAULT_MIGRATE_X_CHECKPOINT_DELAY 200
> #define DEFAULT_MIGRATE_MULTIFD_THREADS 2
> +#define DEFAULT_MIGRATE_MULTIFD_GROUP 16
>
> static NotifierList migration_state_notifiers =
> NOTIFIER_LIST_INITIALIZER(migration_state_notifiers);
> @@ -486,6 +487,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
> params->block_incremental = s->parameters.block_incremental;
> params->has_x_multifd_threads = true;
> params->x_multifd_threads = s->parameters.x_multifd_threads;
> + params->has_x_multifd_group = true;
> + params->x_multifd_group = s->parameters.x_multifd_group;
>
> return params;
> }
> @@ -774,6 +777,13 @@ static bool migrate_params_check(MigrationParameters *params, Error **errp)
> "is invalid, it should be in the range of 1 to 255");
> return false;
> }
> + if (params->has_x_multifd_group &&
> + (params->x_multifd_group < 1 || params->x_multifd_group > 10000)) {
> + error_setg(errp, QERR_INVALID_PARAMETER_VALUE,
> + "multifd_group",
> + "is invalid, it should be in the range of 1 to 10000");
> + return false;
> + }
>
> return true;
> }
> @@ -895,6 +905,9 @@ static void migrate_params_apply(MigrateSetParameters *params)
> if (params->has_x_multifd_threads) {
> s->parameters.x_multifd_threads = params->x_multifd_threads;
> }
> + if (params->has_x_multifd_group) {
> + s->parameters.x_multifd_group = params->x_multifd_group;
> + }
> }
>
> void qmp_migrate_set_parameters(MigrateSetParameters *params, Error **errp)
> @@ -1481,6 +1494,15 @@ int migrate_multifd_threads(void)
> return s->parameters.x_multifd_threads;
> }
>
> +int migrate_multifd_group(void)
> +{
> + MigrationState *s;
> +
> + s = migrate_get_current();
> +
> + return s->parameters.x_multifd_group;
> +}
> +
> int migrate_use_xbzrle(void)
> {
> MigrationState *s;
> @@ -2249,6 +2271,9 @@ static Property migration_properties[] = {
> DEFINE_PROP_INT64("x-multifd-threads", MigrationState,
> parameters.x_multifd_threads,
> DEFAULT_MIGRATE_MULTIFD_THREADS),
> + DEFINE_PROP_INT64("x-multifd-group", MigrationState,
> + parameters.x_multifd_group,
> + DEFAULT_MIGRATE_MULTIFD_GROUP),
>
> /* Migration capabilities */
> DEFINE_PROP_MIG_CAP("x-xbzrle", MIGRATION_CAPABILITY_XBZRLE),
> @@ -2307,6 +2332,7 @@ static void migration_instance_init(Object *obj)
> params->has_x_checkpoint_delay = true;
> params->has_block_incremental = true;
> params->has_x_multifd_threads = true;
> + params->has_x_multifd_group = true;
> }
>
> /*
> diff --git a/migration/migration.h b/migration/migration.h
> index 8e4803d..ddf8c89 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -176,6 +176,7 @@ bool migrate_zero_blocks(void);
> bool migrate_auto_converge(void);
> bool migrate_use_multifd(void);
> int migrate_multifd_threads(void);
> +int migrate_multifd_group(void);
>
> int migrate_use_xbzrle(void);
> int64_t migrate_xbzrle_cache_size(void);
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 3fe1a64..2d4dbd3 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -1049,6 +1049,9 @@
> # number of sockets used for migration.
> # The default value is 2 (since 2.11)
> #
> +# @x-multifd-group: Number of pages sent together to a thread
> +# The default value is 16 (since 2.11)
> +#
> # Since: 2.4
> ##
> { 'enum': 'MigrationParameter',
> @@ -1056,7 +1059,7 @@
> 'cpu-throttle-initial', 'cpu-throttle-increment',
> 'tls-creds', 'tls-hostname', 'max-bandwidth',
> 'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
> - 'x-multifd-threads'] }
> + 'x-multifd-threads', 'x-multifd-group'] }
>
> ##
> # @MigrateSetParameters:
> @@ -1117,6 +1120,9 @@
> # number of sockets used for migration.
> # The default value is 2 (since 2.11)
> #
> +# @x-multifd-group: Number of pages sent together to a thread
> +# The default value is 16 (since 2.11)
Bikesheeding - how about "x-multifd-page-count" ? 'group' didn't
give me much idea what it was controlling until reading the docs.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming Juan Quintela
@ 2017-08-11 15:10 ` Daniel P. Berrange
2017-08-15 6:43 ` Peter Xu
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:10 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:20PM +0200, Juan Quintela wrote:
> We need that on later patches.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
> ---
> migration/migration.c | 14 ++++++++++++--
> 1 file changed, 12 insertions(+), 2 deletions(-)
Reviewed-by: Daniel P. Berrange <berrange@redhat.com>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work Juan Quintela
@ 2017-08-11 15:22 ` Daniel P. Berrange
2017-08-14 13:43 ` Juan Quintela
0 siblings, 1 reply; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:22 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:21PM +0200, Juan Quintela wrote:
> We create new channels for each new thread created. We send through
> them a string containing <uuid> multifd <channel number> so we are
> sure that we connect the right channels in both sides.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
> --
> Split SocketArgs into incoming and outgoing args
>
> Use UUID's on the initial message, so we are sure we are connecting to
> the right channel.
>
> Remove init semaphore. Now that we use uuids on the init message, we
> know that this is our channel.
>
> Fix recv socket destwroy, we were destroying send channels.
> This was very interesting, because we were using an unreferred object
> without problems.
>
> Move to struct of pointers
> init channel sooner.
> split recv thread creation.
> listen on main thread
> We count the number of created threads to know when we need to stop listening
> Use g_strdup_printf
> report channel id on errors
> ---
> migration/migration.c | 5 +++
> migration/ram.c | 99 +++++++++++++++++++++++++++++++++++++++++++--------
> migration/ram.h | 3 ++
> migration/socket.c | 36 ++++++++++++++++++-
> migration/socket.h | 10 ++++++
> 5 files changed, 138 insertions(+), 15 deletions(-)
>
> diff --git a/migration/migration.c b/migration/migration.c
> index e36e880..944d6e2 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -413,6 +413,11 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
> */
> bool migration_has_all_channels(void)
> {
> + if (migrate_use_multifd()) {
> + int thread_count = migrate_multifd_threads();
> +
> + return thread_count == multifd_created_threads();
> + }
> return true;
> }
>
> diff --git a/migration/ram.c b/migration/ram.c
> index 9fb3496..e9fa556 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -36,6 +36,7 @@
> #include "xbzrle.h"
> #include "ram.h"
> #include "migration.h"
> +#include "socket.h"
> #include "migration/register.h"
> #include "migration/misc.h"
> #include "qemu-file.h"
> @@ -46,6 +47,8 @@
> #include "exec/ram_addr.h"
> #include "qemu/rcu_queue.h"
> #include "migration/colo.h"
> +#include "sysemu/sysemu.h"
> +#include "qemu/uuid.h"
>
> /***********************************************************/
> /* ram save/restore */
> @@ -361,6 +364,7 @@ static void compress_threads_save_setup(void)
> struct MultiFDSendParams {
> uint8_t id;
> QemuThread thread;
> + QIOChannel *c;
> QemuSemaphore sem;
> QemuMutex mutex;
> bool quit;
> @@ -401,6 +405,7 @@ void multifd_save_cleanup(void)
> qemu_thread_join(&p->thread);
> qemu_mutex_destroy(&p->mutex);
> qemu_sem_destroy(&p->sem);
> + socket_send_channel_destroy(p->c);
> }
> g_free(multifd_send_state->params);
> multifd_send_state->params = NULL;
> @@ -408,9 +413,26 @@ void multifd_save_cleanup(void)
> multifd_send_state = NULL;
> }
>
> +/* Default uuid for multifd when qemu is not started with uuid */
> +static char multifd_uuid[] = "5c49fd7e-af88-4a07-b6e8-091fd696ad40";
Why is this better than just using the qemu_uuid unconditionally.
UUIC, it'll just be 00000000-0000-0000-0000-000000000000.
Either way you've got a non-unique UUID if multiple QEMUs are
started, so I dont see a benefit in inventing a new uuid here.
> +/* strlen(multifd) + '-' + <channel id> + '-' + UUID_FMT + '\0' */
> +#define MULTIFD_UUID_MSG (7 + 1 + 3 + 1 + UUID_FMT_LEN + 1)
> +
> static void *multifd_send_thread(void *opaque)
> {
> MultiFDSendParams *p = opaque;
> + char *string;
> + char *string_uuid;
> +
> + if (qemu_uuid_set) {
> + string_uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
> + } else {
> + string_uuid = g_strdup(multifd_uuid);
> + }
> + string = g_strdup_printf("%s multifd %03d", string_uuid, p->id);
> + g_free(string_uuid);
> + qio_channel_write(p->c, string, MULTIFD_UUID_MSG, &error_abort);
Must check return code here as it can do a short write, which won't
trigger error_abort. Also you must not use error_abort at all. QEMU
must not abort when migration hits an I/O error as that needlessly
kills the user's VM.
I also think it is not nice to be formatting a string with printf
here, sending it and then using scanf to extract the data. If we
need to send structured data, then we should define a proper binary
format for it eg
struct MigrateUUIDMsg {
uint32_t chnanelid;
QemuUUID uuid;
} __attribute__((__packed__));
and then just send the raw struct across.
> + g_free(string);
>
> while (true) {
> qemu_mutex_lock(&p->mutex);
> @@ -445,6 +467,12 @@ int multifd_save_setup(void)
> qemu_sem_init(&p->sem, 0);
> p->quit = false;
> p->id = i;
> + p->c = socket_send_channel_create();
> + if (!p->c) {
> + error_report("Error creating a send channel");
> + multifd_save_cleanup();
> + return -1;
> + }
We should pass an 'Error *' object into socket_send_channel_create()
so that we can receive & report the actual error message, instead
of a useless generic message.
> @@ -522,10 +556,54 @@ static void *multifd_recv_thread(void *opaque)
> return NULL;
> }
>
> +void multifd_new_channel(QIOChannel *ioc)
> +{
> + MultiFDRecvParams *p = g_new0(MultiFDRecvParams, 1);
> + MigrationState *s = migrate_get_current();
> + char string[MULTIFD_UUID_MSG];
> + char string_uuid[UUID_FMT_LEN];
> + char *uuid;
> + int id;
> +
> + qio_channel_read(ioc, string, sizeof(string), &error_abort);
Must not use error_abort which kills the whole VM ungracefully
> + sscanf(string, "%s multifd %03d", string_uuid, &id);
> +
> + if (qemu_uuid_set) {
> + uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
> + } else {
> + uuid = g_strdup(multifd_uuid);
> + }
> + if (strcmp(string_uuid, uuid)) {
> + error_report("multifd: received uuid '%s' and expected uuid '%s'"
> + " for channel %d", string_uuid, uuid, id);
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return;
> + }
> + g_free(uuid);
As mentioned above, we should receive a binary struct here instead
of playing games with scanf.
> +
> + if (multifd_recv_state->params[id] != NULL) {
> + error_report("multifd: received id '%d' is already setup'", id);
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return;
> + }
> + qemu_mutex_init(&p->mutex);
> + qemu_sem_init(&p->sem, 0);
> + p->quit = false;
> + p->id = id;
> + p->c = ioc;
> + atomic_set(&multifd_recv_state->params[id], p);
> + multifd_recv_state->count++;
> + qemu_thread_create(&p->thread, "multifd_recv", multifd_recv_thread, p,
> + QEMU_THREAD_JOINABLE);
> +}
> +
> int multifd_load_setup(void)
> {
> int thread_count;
> - uint8_t i;
>
> if (!migrate_use_multifd()) {
> return 0;
> @@ -534,22 +612,15 @@ int multifd_load_setup(void)
> multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
> multifd_recv_state->params = g_new0(MultiFDRecvParams *, thread_count);
> multifd_recv_state->count = 0;
> - for (i = 0; i < thread_count; i++) {
> - char thread_name[16];
> - MultiFDRecvParams *p = multifd_recv_state->params[i];
> -
> - qemu_mutex_init(&p->mutex);
> - qemu_sem_init(&p->sem, 0);
> - p->quit = false;
> - p->id = i;
> - snprintf(thread_name, sizeof(thread_name), "multifdrecv_%d", i);
> - qemu_thread_create(&p->thread, thread_name, multifd_recv_thread, p,
> - QEMU_THREAD_JOINABLE);
> - multifd_recv_state->count++;
> - }
It us a little strange to be deleting a bunch of code you just added in the
previous patch
> + multifd_recv_state->quit = false;
> return 0;
> }
> diff --git a/migration/socket.c b/migration/socket.c
> index dee8690..5dd6f42 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -26,6 +26,38 @@
> #include "io/channel-socket.h"
> #include "trace.h"
>
> +int socket_recv_channel_destroy(QIOChannel *recv)
> +{
> + /* Remove channel */
> + object_unref(OBJECT(recv));
> + return 0;
> +}
> +
> +struct SocketOutgoingArgs {
> + SocketAddress *saddr;
> + Error **errp;
> +} outgoing_args;
> +
> +QIOChannel *socket_send_channel_create(void)
> +{
> + QIOChannelSocket *sioc = qio_channel_socket_new();
> +
> + qio_channel_socket_connect_sync(sioc, outgoing_args.saddr,
> + outgoing_args.errp);
> + qio_channel_set_delay(QIO_CHANNEL(sioc), false);
> + return QIO_CHANNEL(sioc);
> +}
> +
> +int socket_send_channel_destroy(QIOChannel *send)
> +{
> + /* Remove channel */
> + object_unref(OBJECT(send));
> + if (outgoing_args.saddr) {
> + qapi_free_SocketAddress(outgoing_args.saddr);
> + outgoing_args.saddr = NULL;
> + }
> + return 0;
> +}
>
> static SocketAddress *tcp_build_address(const char *host_port, Error **errp)
> {
> @@ -96,6 +128,9 @@ static void socket_start_outgoing_migration(MigrationState *s,
> struct SocketConnectData *data = g_new0(struct SocketConnectData, 1);
>
> data->s = s;
> + outgoing_args.saddr = saddr;
> + outgoing_args.errp = errp;
Taking a stack local 'errp' pointer and saving it for later use
in a global variable is asking for trouble. There's no guarantee
that stack frame will still be valid when 'outgoing_args.errp'
is later used.
> +
> if (saddr->type == SOCKET_ADDRESS_TYPE_INET) {
> data->hostname = g_strdup(saddr->u.inet.host);
> }
> @@ -106,7 +141,6 @@ static void socket_start_outgoing_migration(MigrationState *s,
> socket_outgoing_migration,
> data,
> socket_connect_data_free);
> - qapi_free_SocketAddress(saddr);
> }
>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 12/19] migration: Create ram_multifd_page
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 12/19] migration: Create ram_multifd_page Juan Quintela
@ 2017-08-11 15:25 ` Daniel P. Berrange
0 siblings, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:25 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:22PM +0200, Juan Quintela wrote:
> The function still don't use multifd, but we have simplified
> ram_save_page, xbzrle and RDMA stuff is gone. We have added a new
> counter and a new flag for this type of pages.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
> --
> Add last_page parameter
> Add commets for done and address
> ---
> hmp.c | 2 ++
> migration/migration.c | 1 +
> migration/ram.c | 94 ++++++++++++++++++++++++++++++++++++++++++++++++++-
> qapi-schema.json | 5 ++-
> 4 files changed, 100 insertions(+), 2 deletions(-)
>
> diff --git a/hmp.c b/hmp.c
> index a52035a..bc2b071 100644
> --- a/hmp.c
> +++ b/hmp.c
> @@ -234,6 +234,8 @@ void hmp_info_migrate(Monitor *mon, const QDict *qdict)
> monitor_printf(mon, "postcopy request count: %" PRIu64 "\n",
> info->ram->postcopy_requests);
> }
> + monitor_printf(mon, "multifd: %" PRIu64 " pages\n",
> + info->ram->multifd);
> }
>
> if (info->has_disk) {
> diff --git a/migration/migration.c b/migration/migration.c
> index 944d6e2..8e9505a 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -549,6 +549,7 @@ static void populate_ram_info(MigrationInfo *info, MigrationState *s)
> info->ram->dirty_sync_count = ram_counters.dirty_sync_count;
> info->ram->postcopy_requests = ram_counters.postcopy_requests;
> info->ram->page_size = qemu_target_page_size();
> + info->ram->multifd = ram_counters.multifd;
>
> if (migrate_use_xbzrle()) {
> info->has_xbzrle_cache = true;
> diff --git a/migration/ram.c b/migration/ram.c
> index e9fa556..03f3427 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -68,6 +68,7 @@
> #define RAM_SAVE_FLAG_XBZRLE 0x40
> /* 0x80 is reserved in migration.h start with 0x100 next */
> #define RAM_SAVE_FLAG_COMPRESS_PAGE 0x100
> +#define RAM_SAVE_FLAG_MULTIFD_PAGE 0x200
>
> static inline bool is_zero_range(uint8_t *p, uint64_t size)
> {
> @@ -362,12 +363,22 @@ static void compress_threads_save_setup(void)
> /* Multiple fd's */
>
> struct MultiFDSendParams {
> + /* not changed */
> uint8_t id;
> QemuThread thread;
> QIOChannel *c;
> QemuSemaphore sem;
> QemuMutex mutex;
> + /* protected by param mutex */
> bool quit;
> + /* This is a temp field. We are using it now to transmit
> + something the address of the page. Later in the series, we
> + change it for the real page.
> + */
> + uint8_t *address;
> + /* protected by multifd mutex */
> + /* has the thread finish the last submitted job */
> + bool done;
> };
> typedef struct MultiFDSendParams MultiFDSendParams;
>
> @@ -375,6 +386,8 @@ struct {
> MultiFDSendParams *params;
> /* number of created threads */
> int count;
> + QemuMutex mutex;
> + QemuSemaphore sem;
> } *multifd_send_state;
>
> static void terminate_multifd_send_threads(void)
> @@ -433,6 +446,7 @@ static void *multifd_send_thread(void *opaque)
> g_free(string_uuid);
> qio_channel_write(p->c, string, MULTIFD_UUID_MSG, &error_abort);
> g_free(string);
> + qemu_sem_post(&multifd_send_state->sem);
>
> while (true) {
> qemu_mutex_lock(&p->mutex);
> @@ -440,6 +454,15 @@ static void *multifd_send_thread(void *opaque)
> qemu_mutex_unlock(&p->mutex);
> break;
> }
> + if (p->address) {
> + p->address = 0;
> + qemu_mutex_unlock(&p->mutex);
> + qemu_mutex_lock(&multifd_send_state->mutex);
> + p->done = true;
> + qemu_mutex_unlock(&multifd_send_state->mutex);
> + qemu_sem_post(&multifd_send_state->sem);
> + continue;
> + }
> qemu_mutex_unlock(&p->mutex);
> qemu_sem_wait(&p->sem);
> }
> @@ -459,6 +482,8 @@ int multifd_save_setup(void)
> multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
> multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
> multifd_send_state->count = 0;
> + qemu_mutex_init(&multifd_send_state->mutex);
> + qemu_sem_init(&multifd_send_state->sem, 0);
> for (i = 0; i < thread_count; i++) {
> char thread_name[16];
> MultiFDSendParams *p = &multifd_send_state->params[i];
> @@ -467,6 +492,8 @@ int multifd_save_setup(void)
> qemu_sem_init(&p->sem, 0);
> p->quit = false;
> p->id = i;
> + p->done = true;
> + p->address = 0;
> p->c = socket_send_channel_create();
> if (!p->c) {
> error_report("Error creating a send channel");
> @@ -481,6 +508,30 @@ int multifd_save_setup(void)
> return 0;
> }
>
> +static uint16_t multifd_send_page(uint8_t *address, bool last_page)
> +{
> + int i;
> + MultiFDSendParams *p = NULL; /* make happy gcc */
> +
> + qemu_sem_wait(&multifd_send_state->sem);
> + qemu_mutex_lock(&multifd_send_state->mutex);
> + for (i = 0; i < multifd_send_state->count; i++) {
> + p = &multifd_send_state->params[i];
> +
> + if (p->done) {
> + p->done = false;
> + break;
> + }
> + }
> + qemu_mutex_unlock(&multifd_send_state->mutex);
> + qemu_mutex_lock(&p->mutex);
> + p->address = address;
> + qemu_mutex_unlock(&p->mutex);
> + qemu_sem_post(&p->sem);
> +
> + return 0;
> +}
> +
> struct MultiFDRecvParams {
> uint8_t id;
> QemuThread thread;
> @@ -1051,6 +1102,32 @@ static int ram_save_page(RAMState *rs, PageSearchStatus *pss, bool last_stage)
> return pages;
> }
>
> +static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
> + bool last_stage)
> +{
> + int pages;
> + uint8_t *p;
> + RAMBlock *block = pss->block;
> + ram_addr_t offset = pss->page << TARGET_PAGE_BITS;
> +
> + p = block->host + offset;
> +
> + pages = save_zero_page(rs, block, offset, p);
> + if (pages == -1) {
> + ram_counters.transferred +=
> + save_page_header(rs, rs->f, block,
> + offset | RAM_SAVE_FLAG_MULTIFD_PAGE);
> + qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
> + multifd_send_page(p, rs->migration_dirty_pages == 1);
> + ram_counters.transferred += TARGET_PAGE_SIZE;
> + pages = 1;
> + ram_counters.normal++;
> + ram_counters.multifd++;
> + }
> +
> + return pages;
> +}
> +
> static int do_compress_ram_page(QEMUFile *f, RAMBlock *block,
> ram_addr_t offset)
> {
> @@ -1479,6 +1556,8 @@ static int ram_save_target_page(RAMState *rs, PageSearchStatus *pss,
> if (migrate_use_compression() &&
> (rs->ram_bulk_stage || !migrate_use_xbzrle())) {
> res = ram_save_compressed_page(rs, pss, last_stage);
> + } else if (migrate_use_multifd()) {
> + res = ram_multifd_page(rs, pss, last_stage);
> } else {
> res = ram_save_page(rs, pss, last_stage);
> }
> @@ -2771,6 +2850,10 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
> if (!migrate_use_compression()) {
> invalid_flags |= RAM_SAVE_FLAG_COMPRESS_PAGE;
> }
> +
> + if (!migrate_use_multifd()) {
> + invalid_flags |= RAM_SAVE_FLAG_MULTIFD_PAGE;
> + }
> /* This RCU critical section can be very long running.
> * When RCU reclaims in the code start to become numerous,
> * it will be necessary to reduce the granularity of this
> @@ -2795,13 +2878,17 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
> if (flags & invalid_flags & RAM_SAVE_FLAG_COMPRESS_PAGE) {
> error_report("Received an unexpected compressed page");
> }
> + if (flags & invalid_flags & RAM_SAVE_FLAG_MULTIFD_PAGE) {
> + error_report("Received an unexpected multifd page");
> + }
>
> ret = -EINVAL;
> break;
> }
>
> if (flags & (RAM_SAVE_FLAG_ZERO | RAM_SAVE_FLAG_PAGE |
> - RAM_SAVE_FLAG_COMPRESS_PAGE | RAM_SAVE_FLAG_XBZRLE)) {
> + RAM_SAVE_FLAG_COMPRESS_PAGE | RAM_SAVE_FLAG_XBZRLE |
> + RAM_SAVE_FLAG_MULTIFD_PAGE)) {
> RAMBlock *block = ram_block_from_stream(f, flags);
>
> host = host_from_ram_block_offset(block, addr);
> @@ -2889,6 +2976,11 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
> break;
> }
> break;
> +
> + case RAM_SAVE_FLAG_MULTIFD_PAGE:
> + qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
> + break;
> +
> case RAM_SAVE_FLAG_EOS:
> /* normal exit */
> break;
> diff --git a/qapi-schema.json b/qapi-schema.json
> index 2d4dbd3..bb1fe0a 100644
> --- a/qapi-schema.json
> +++ b/qapi-schema.json
> @@ -617,6 +617,8 @@
> # @page-size: The number of bytes per page for the various page-based
> # statistics (since 2.10)
> #
> +# @multifd: number of pages sent with multifd (since 2.10)
> +#
> # Since: 0.14.0
> ##
> { 'struct': 'MigrationStats',
> @@ -624,7 +626,8 @@
> 'duplicate': 'int', 'skipped': 'int', 'normal': 'int',
> 'normal-bytes': 'int', 'dirty-pages-rate' : 'int',
> 'mbps' : 'number', 'dirty-sync-count' : 'int',
> - 'postcopy-requests' : 'int', 'page-size' : 'int' } }
> + 'postcopy-requests' : 'int', 'page-size' : 'int',
> + 'multifd' : 'int'} }
If we enabled multifd when starting the migration, then surely *all*
pages should be sent with multifd, meaning this is no different to
the existing "normal" statistic.
IOW, I don't see any point in adding this new value.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side Juan Quintela
2017-08-11 4:35 ` Peter Xu
@ 2017-08-11 15:29 ` Daniel P. Berrange
2017-09-13 9:53 ` Juan Quintela
2017-09-13 11:26 ` Juan Quintela
1 sibling, 2 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:29 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:25PM +0200, Juan Quintela wrote:
> 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 <quintela@redhat.com>
>
> --
>
> 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_group().
> ---
> 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/socket.c b/migration/socket.c
> index 5dd6f42..3af9f7c 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -183,12 +183,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));
AFAICT, migration_channel_process_incoming() acquires its own reference
on 'sioc', so removing this object_unref means the code is now leaking a
reference
>
> 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.9.4
>
>
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure Juan Quintela
@ 2017-08-11 15:32 ` Daniel P. Berrange
2017-08-11 15:35 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:32 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:26PM +0200, Juan Quintela wrote:
> We just send the address through the alternate channels and test that it
> is ok.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/ram.c | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
> 1 file changed, 50 insertions(+)
>
> diff --git a/migration/ram.c b/migration/ram.c
> index eb0015e..42ad126 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -479,8 +479,26 @@ static void *multifd_send_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> + int i;
> + int num;
> +
> + num = p->pages.num;
> p->pages.num = 0;
> qemu_mutex_unlock(&p->mutex);
> +
> + for (i = 0; i < num; i++) {
> + if (qio_channel_write(p->c,
> + (const char *)&p->pages.iov[i].iov_base,
> + sizeof(uint8_t *), &error_abort)
> + != sizeof(uint8_t *)) {
Must not use error_abort - this kils the entire VM if we hit an I/O error.
Treating short-writes a fatal is also really not desirable.
> + MigrationState *s = migrate_get_current();
> +
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_send_threads();
> + return NULL;
> + }
> + }
> qemu_mutex_lock(&multifd_send_state->mutex);
> p->done = true;
> qemu_mutex_unlock(&multifd_send_state->mutex);
> @@ -640,6 +658,7 @@ void multifd_load_cleanup(void)
> static void *multifd_recv_thread(void *opaque)
> {
> MultiFDRecvParams *p = opaque;
> + uint8_t *recv_address;
>
> qemu_sem_post(&p->ready);
> while (true) {
> @@ -649,7 +668,38 @@ static void *multifd_recv_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> + int i;
> + int num;
> +
> + num = p->pages.num;
> p->pages.num = 0;
> +
> + for (i = 0; i < num; i++) {
> + if (qio_channel_read(p->c,
> + (char *)&recv_address,
> + sizeof(uint8_t *), &error_abort)
> + != sizeof(uint8_t *)) {
Again, don't use error_abort, and you should handle short reads.
> + MigrationState *s = migrate_get_current();
> +
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return NULL;
> + }
> + if (recv_address != p->pages.iov[i].iov_base) {
> + MigrationState *s = migrate_get_current();
> +
> + printf("We received %p what we were expecting %p (%d)\n",
> + recv_address,
> + p->pages.iov[i].iov_base, i);
Should use a trace event probe for this.
> +
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return NULL;
> + }
> + }
> +
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels Juan Quintela
@ 2017-08-11 15:34 ` Daniel P. Berrange
2017-08-16 16:38 ` Dr. David Alan Gilbert
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:34 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:28PM +0200, Juan Quintela wrote:
> We switch for sending the page number to send real pages.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
> --
>
> Remove the HACK bit, now we have the function that calculates the size
> of a page exported.
> ---
> migration/migration.c | 7 +++++-
> migration/ram.c | 59 +++++++++++++++++----------------------------------
> 2 files changed, 25 insertions(+), 41 deletions(-)
>
> diff --git a/migration/migration.c b/migration/migration.c
> index 974ff92..aac3cdc 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -2051,6 +2051,7 @@ static void *migration_thread(void *opaque)
> */
> int64_t threshold_size = 0;
> int64_t qemu_file_bytes = 0;
> + int64_t multifd_pages = 0;
> int64_t start_time = initial_time;
> int64_t end_time;
> bool old_vm_running = false;
> @@ -2139,8 +2140,11 @@ static void *migration_thread(void *opaque)
> current_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
> if (current_time >= initial_time + BUFFER_DELAY) {
> uint64_t qemu_file_bytes_now = qemu_ftell(s->to_dst_file);
> + uint64_t multifd_pages_now = ram_counters.multifd;
> uint64_t transferred_bytes =
> - qemu_file_bytes_now - qemu_file_bytes;
> + (qemu_file_bytes_now - qemu_file_bytes) +
> + (multifd_pages_now - multifd_pages) *
> + qemu_target_page_size();
> uint64_t time_spent = current_time - initial_time;
> double bandwidth = (double)transferred_bytes / time_spent;
> threshold_size = bandwidth * s->parameters.downtime_limit;
> @@ -2160,6 +2164,7 @@ static void *migration_thread(void *opaque)
> qemu_file_reset_rate_limit(s->to_dst_file);
> initial_time = current_time;
> qemu_file_bytes = qemu_file_bytes_now;
> + multifd_pages = multifd_pages_now;
> }
> if (qemu_file_rate_limit(s->to_dst_file)) {
> /* usleep expects microseconds */
> diff --git a/migration/ram.c b/migration/ram.c
> index 42ad126..f337360 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -479,25 +479,21 @@ static void *multifd_send_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> - int i;
> int num;
>
> num = p->pages.num;
> p->pages.num = 0;
> qemu_mutex_unlock(&p->mutex);
>
> - for (i = 0; i < num; i++) {
> - if (qio_channel_write(p->c,
> - (const char *)&p->pages.iov[i].iov_base,
> - sizeof(uint8_t *), &error_abort)
> - != sizeof(uint8_t *)) {
> - MigrationState *s = migrate_get_current();
> + if (qio_channel_writev_all(p->c, p->pages.iov,
> + num, &error_abort)
> + != num * TARGET_PAGE_SIZE) {
Again, should not be using error_abort - the error should be captured
so it cna be reported in any query-migrate QMP call.
> + MigrationState *s = migrate_get_current();
>
> - migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> - MIGRATION_STATUS_FAILED);
> - terminate_multifd_send_threads();
> - return NULL;
> - }
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_send_threads();
> + return NULL;
> }
> qemu_mutex_lock(&multifd_send_state->mutex);
> p->done = true;
> @@ -658,7 +654,6 @@ void multifd_load_cleanup(void)
> static void *multifd_recv_thread(void *opaque)
> {
> MultiFDRecvParams *p = opaque;
> - uint8_t *recv_address;
>
> qemu_sem_post(&p->ready);
> while (true) {
> @@ -668,38 +663,21 @@ static void *multifd_recv_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> - int i;
> int num;
>
> num = p->pages.num;
> p->pages.num = 0;
>
> - for (i = 0; i < num; i++) {
> - if (qio_channel_read(p->c,
> - (char *)&recv_address,
> - sizeof(uint8_t *), &error_abort)
> - != sizeof(uint8_t *)) {
> - MigrationState *s = migrate_get_current();
> + if (qio_channel_readv_all(p->c, p->pages.iov,
> + num, &error_abort)
> + != num * TARGET_PAGE_SIZE) {
Same note about error_abort usage.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure Juan Quintela
2017-08-11 15:32 ` Daniel P. Berrange
@ 2017-08-11 15:35 ` Daniel P. Berrange
1 sibling, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-11 15:35 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Tue, Aug 08, 2017 at 06:26:26PM +0200, Juan Quintela wrote:
> We just send the address through the alternate channels and test that it
> is ok.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/ram.c | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
> 1 file changed, 50 insertions(+)
>
> diff --git a/migration/ram.c b/migration/ram.c
> index eb0015e..42ad126 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -479,8 +479,26 @@ static void *multifd_send_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> + int i;
> + int num;
> +
> + num = p->pages.num;
> p->pages.num = 0;
> qemu_mutex_unlock(&p->mutex);
> +
> + for (i = 0; i < num; i++) {
> + if (qio_channel_write(p->c,
> + (const char *)&p->pages.iov[i].iov_base,
> + sizeof(uint8_t *), &error_abort)
> + != sizeof(uint8_t *)) {
> + MigrationState *s = migrate_get_current();
> +
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_send_threads();
> + return NULL;
> + }
> + }
> qemu_mutex_lock(&multifd_send_state->mutex);
> p->done = true;
> qemu_mutex_unlock(&multifd_send_state->mutex);
> @@ -640,6 +658,7 @@ void multifd_load_cleanup(void)
> static void *multifd_recv_thread(void *opaque)
> {
> MultiFDRecvParams *p = opaque;
> + uint8_t *recv_address;
>
> qemu_sem_post(&p->ready);
> while (true) {
> @@ -649,7 +668,38 @@ static void *multifd_recv_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> + int i;
> + int num;
> +
> + num = p->pages.num;
> p->pages.num = 0;
> +
> + for (i = 0; i < num; i++) {
> + if (qio_channel_read(p->c,
> + (char *)&recv_address,
> + sizeof(uint8_t *), &error_abort)
> + != sizeof(uint8_t *)) {
> + MigrationState *s = migrate_get_current();
> +
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return NULL;
> + }
> + if (recv_address != p->pages.iov[i].iov_base) {
> + MigrationState *s = migrate_get_current();
> +
> + printf("We received %p what we were expecting %p (%d)\n",
> + recv_address,
> + p->pages.iov[i].iov_base, i);
> +
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return NULL;
> + }
> + }
> +
> p->done = true;
> qemu_mutex_unlock(&p->mutex);
> qemu_sem_post(&p->ready);
Looks like all this code is deleting in patch 18. So IMHO just drop this
patch entirely.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 17/19] migration: Rename initial_bytes
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 17/19] migration: Rename initial_bytes Juan Quintela
@ 2017-08-11 19:01 ` Dr. David Alan Gilbert
0 siblings, 0 replies; 61+ messages in thread
From: Dr. David Alan Gilbert @ 2017-08-11 19:01 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, peterx
* Juan Quintela (quintela@redhat.com) wrote:
> Now it is called qemu_file_bytes that reflects better what it does,
> and we create qemu_file_bytes_now to not have to call qemu_ftell() twice.
Yes OK, although you could lose the 'qemu_' off the start; it's local
anyway.
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/migration.c | 9 +++++----
> 1 file changed, 5 insertions(+), 4 deletions(-)
>
> diff --git a/migration/migration.c b/migration/migration.c
> index b78dffc..974ff92 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -2044,13 +2044,13 @@ static void *migration_thread(void *opaque)
> /* Used by the bandwidth calcs, updated later */
> int64_t initial_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
> int64_t setup_start = qemu_clock_get_ms(QEMU_CLOCK_HOST);
> - int64_t initial_bytes = 0;
> /*
> * The final stage happens when the remaining data is smaller than
> * this threshold; it's calculated from the requested downtime and
> * measured bandwidth
> */
> int64_t threshold_size = 0;
> + int64_t qemu_file_bytes = 0;
> int64_t start_time = initial_time;
> int64_t end_time;
> bool old_vm_running = false;
> @@ -2138,8 +2138,9 @@ static void *migration_thread(void *opaque)
> }
> current_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
> if (current_time >= initial_time + BUFFER_DELAY) {
> - uint64_t transferred_bytes = qemu_ftell(s->to_dst_file) -
> - initial_bytes;
> + uint64_t qemu_file_bytes_now = qemu_ftell(s->to_dst_file);
> + uint64_t transferred_bytes =
> + qemu_file_bytes_now - qemu_file_bytes;
> uint64_t time_spent = current_time - initial_time;
> double bandwidth = (double)transferred_bytes / time_spent;
> threshold_size = bandwidth * s->parameters.downtime_limit;
> @@ -2158,7 +2159,7 @@ static void *migration_thread(void *opaque)
>
> qemu_file_reset_rate_limit(s->to_dst_file);
> initial_time = current_time;
> - initial_bytes = qemu_ftell(s->to_dst_file);
> + qemu_file_bytes = qemu_file_bytes_now;
> }
> if (qemu_file_rate_limit(s->to_dst_file)) {
> /* usleep expects microseconds */
> --
> 2.9.4
>
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work
2017-08-11 15:22 ` Daniel P. Berrange
@ 2017-08-14 13:43 ` Juan Quintela
2017-08-14 13:50 ` Daniel P. Berrange
0 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-08-14 13:43 UTC (permalink / raw)
To: Daniel P. Berrange; +Cc: qemu-devel, lvivier, dgilbert, peterx
"Daniel P. Berrange" <berrange@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:21PM +0200, Juan Quintela wrote:
>> We create new channels for each new thread created. We send through
>> them a string containing <uuid> multifd <channel number> so we are
>> sure that we connect the right channels in both sides.
>>
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>>
>> +/* Default uuid for multifd when qemu is not started with uuid */
>> +static char multifd_uuid[] = "5c49fd7e-af88-4a07-b6e8-091fd696ad40";
>
> Why is this better than just using the qemu_uuid unconditionally.
> UUIC, it'll just be 00000000-0000-0000-0000-000000000000.
>
> Either way you've got a non-unique UUID if multiple QEMUs are
> started, so I dont see a benefit in inventing a new uuid here.
I hate a message full of zeros, it is the default value. If you have
more than one qemu and you don't set uuid, you are asking for trouble.
But if people preffer the 00000 uuid, it is also ok with me.
>
>> +/* strlen(multifd) + '-' + <channel id> + '-' + UUID_FMT + '\0' */
>> +#define MULTIFD_UUID_MSG (7 + 1 + 3 + 1 + UUID_FMT_LEN + 1)
>> +
>> static void *multifd_send_thread(void *opaque)
>> {
>> MultiFDSendParams *p = opaque;
>> + char *string;
>> + char *string_uuid;
>> +
>> + if (qemu_uuid_set) {
>> + string_uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
>> + } else {
>> + string_uuid = g_strdup(multifd_uuid);
>> + }
>> + string = g_strdup_printf("%s multifd %03d", string_uuid, p->id);
>> + g_free(string_uuid);
>> + qio_channel_write(p->c, string, MULTIFD_UUID_MSG, &error_abort);
>
> Must check return code here as it can do a short write, which won't
> trigger error_abort. Also you must not use error_abort at all. QEMU
> must not abort when migration hits an I/O error as that needlessly
> kills the user's VM.
>
> I also think it is not nice to be formatting a string with printf
> here, sending it and then using scanf to extract the data. If we
> need to send structured data, then we should define a proper binary
> format for it eg
>
> struct MigrateUUIDMsg {
> uint32_t chnanelid;
> QemuUUID uuid;
> } __attribute__((__packed__));
>
> and then just send the raw struct across.
Not that I believe that it still works (or that it worked ever), but
qemu migration "on stream" was supposed to be Endian safe .....
>> + g_free(string);
>>
>> while (true) {
>> qemu_mutex_lock(&p->mutex);
>> @@ -445,6 +467,12 @@ int multifd_save_setup(void)
>> qemu_sem_init(&p->sem, 0);
>> p->quit = false;
>> p->id = i;
>> + p->c = socket_send_channel_create();
>> + if (!p->c) {
>> + error_report("Error creating a send channel");
>> + multifd_save_cleanup();
>> + return -1;
>> + }
>
> We should pass an 'Error *' object into socket_send_channel_create()
> so that we can receive & report the actual error message, instead
> of a useless generic message.
Yeap, that is what I told was doing next on the Cover letter.
Just that it means changing lots of functions prototypes ....
>> + qio_channel_read(ioc, string, sizeof(string), &error_abort);
>
> Must not use error_abort which kills the whole VM ungracefully
Changing also that.
>
>> + sscanf(string, "%s multifd %03d", string_uuid, &id);
>> +
>> + if (qemu_uuid_set) {
>> + uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
>> + } else {
>> + uuid = g_strdup(multifd_uuid);
>> + }
>> + if (strcmp(string_uuid, uuid)) {
>> + error_report("multifd: received uuid '%s' and expected uuid '%s'"
>> + " for channel %d", string_uuid, uuid, id);
>> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
>> + MIGRATION_STATUS_FAILED);
>> + terminate_multifd_recv_threads();
>> + return;
>> + }
>> + g_free(uuid);
>
> As mentioned above, we should receive a binary struct here instead
> of playing games with scanf.
>> @@ -534,22 +612,15 @@ int multifd_load_setup(void)
>> multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
>> multifd_recv_state->params = g_new0(MultiFDRecvParams *, thread_count);
>> multifd_recv_state->count = 0;
>> - for (i = 0; i < thread_count; i++) {
>> - char thread_name[16];
>> - MultiFDRecvParams *p = multifd_recv_state->params[i];
>> -
>> - qemu_mutex_init(&p->mutex);
>> - qemu_sem_init(&p->sem, 0);
>> - p->quit = false;
>> - p->id = i;
>> - snprintf(thread_name, sizeof(thread_name), "multifdrecv_%d", i);
>> - qemu_thread_create(&p->thread, thread_name, multifd_recv_thread, p,
>> - QEMU_THREAD_JOINABLE);
>> - multifd_recv_state->count++;
>> - }
>
> It us a little strange to be deleting a bunch of code you just added in the
> previous patch
We create the send/recv channels on previous patch. And then we switch
to create them as we receive the connections. I will try to see if
creating first the receiving threads it gets clearer.
>> static SocketAddress *tcp_build_address(const char *host_port, Error **errp)
>> {
>> @@ -96,6 +128,9 @@ static void socket_start_outgoing_migration(MigrationState *s,
>> struct SocketConnectData *data = g_new0(struct SocketConnectData, 1);
>>
>> data->s = s;
>> + outgoing_args.saddr = saddr;
>> + outgoing_args.errp = errp;
>
> Taking a stack local 'errp' pointer and saving it for later use
> in a global variable is asking for trouble. There's no guarantee
> that stack frame will still be valid when 'outgoing_args.errp'
> is later used.
Yeap, changing to an Error * pointer by thread.
Thanks, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work
2017-08-14 13:43 ` Juan Quintela
@ 2017-08-14 13:50 ` Daniel P. Berrange
0 siblings, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-08-14 13:50 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Mon, Aug 14, 2017 at 03:43:58PM +0200, Juan Quintela wrote:
> "Daniel P. Berrange" <berrange@redhat.com> wrote:
> > On Tue, Aug 08, 2017 at 06:26:21PM +0200, Juan Quintela wrote:
> >> We create new channels for each new thread created. We send through
> >> them a string containing <uuid> multifd <channel number> so we are
> >> sure that we connect the right channels in both sides.
> >>
> >> Signed-off-by: Juan Quintela <quintela@redhat.com>
> >>
>
> >> +/* Default uuid for multifd when qemu is not started with uuid */
> >> +static char multifd_uuid[] = "5c49fd7e-af88-4a07-b6e8-091fd696ad40";
> >
> > Why is this better than just using the qemu_uuid unconditionally.
> > UUIC, it'll just be 00000000-0000-0000-0000-000000000000.
> >
> > Either way you've got a non-unique UUID if multiple QEMUs are
> > started, so I dont see a benefit in inventing a new uuid here.
>
> I hate a message full of zeros, it is the default value. If you have
> more than one qemu and you don't set uuid, you are asking for trouble.
>
> But if people preffer the 00000 uuid, it is also ok with me.
I don't see a fixed UUID of all-zeros being any different from
a fixed UUID hardcoded in the source code here. Both are simply
conceptually broken in the same way if you care about distinguishing
VMs.
> > I also think it is not nice to be formatting a string with printf
> > here, sending it and then using scanf to extract the data. If we
> > need to send structured data, then we should define a proper binary
> > format for it eg
> >
> > struct MigrateUUIDMsg {
> > uint32_t chnanelid;
> > QemuUUID uuid;
> > } __attribute__((__packed__));
> >
> > and then just send the raw struct across.
>
> Not that I believe that it still works (or that it worked ever), but
> qemu migration "on stream" was supposed to be Endian safe .....
The uuid field is just a byte array, so not endian sensitive. For
the channelid, just do a hton / ntoh call before/after writing/reading
the struct.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 14/19] migration: Send the fd number which we are going to use for this page
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 14/19] migration: Send the fd number which we are going to use for this page Juan Quintela
@ 2017-08-14 15:47 ` Dr. David Alan Gilbert
0 siblings, 0 replies; 61+ messages in thread
From: Dr. David Alan Gilbert @ 2017-08-14 15:47 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, peterx
* Juan Quintela (quintela@redhat.com) wrote:
> We are still sending the page through the main channel, that would
> change later in the series
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
> migration/ram.c | 13 +++++++++++--
> 1 file changed, 11 insertions(+), 2 deletions(-)
>
> diff --git a/migration/ram.c b/migration/ram.c
> index 7310da9..169cfc9 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -568,7 +568,7 @@ static uint16_t multifd_send_page(uint8_t *address, bool last_page)
> qemu_mutex_unlock(&p->mutex);
> qemu_sem_post(&p->sem);
>
> - return 0;
> + return i;
That should probably be in the patch that defines multifd_send_page
other than that,
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
> }
>
> struct MultiFDRecvParams {
> @@ -1145,6 +1145,7 @@ static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
> bool last_stage)
> {
> int pages;
> + uint16_t fd_num;
> uint8_t *p;
> RAMBlock *block = pss->block;
> ram_addr_t offset = pss->page << TARGET_PAGE_BITS;
> @@ -1156,8 +1157,10 @@ static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
> ram_counters.transferred +=
> save_page_header(rs, rs->f, block,
> offset | RAM_SAVE_FLAG_MULTIFD_PAGE);
> + fd_num = multifd_send_page(p, rs->migration_dirty_pages == 1);
> + qemu_put_be16(rs->f, fd_num);
> + ram_counters.transferred += 2; /* size of fd_num */
> qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
> - multifd_send_page(p, rs->migration_dirty_pages == 1);
> ram_counters.transferred += TARGET_PAGE_SIZE;
> pages = 1;
> ram_counters.normal++;
> @@ -2907,6 +2910,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
> while (!postcopy_running && !ret && !(flags & RAM_SAVE_FLAG_EOS)) {
> ram_addr_t addr, total_ram_bytes;
> void *host = NULL;
> + uint16_t fd_num;
> uint8_t ch;
>
> addr = qemu_get_be64(f);
> @@ -3017,6 +3021,11 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
> break;
>
> 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;
> + }
> qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
> break;
>
> --
> 2.9.4
>
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 09/19] migration: Create multifd migration threads
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 09/19] migration: Create multifd migration threads Juan Quintela
@ 2017-08-15 6:42 ` Peter Xu
0 siblings, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-15 6:42 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:19PM +0200, Juan Quintela wrote:
[...]
> diff --git a/migration/migration.c b/migration/migration.c
> index 6c4eb4b..d031c93 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -296,6 +296,7 @@ static void process_incoming_migration_bh(void *opaque)
> } else {
> runstate_set(global_state_get_runstate());
> }
> + multifd_load_cleanup();
Maybe we can put multifd_load_cleanup() into
migration_incoming_state_destroy(), then this line can be removed,
and...
> /*
> * 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
> @@ -358,6 +359,7 @@ static void process_incoming_migration_co(void *opaque)
> MIGRATION_STATUS_FAILED);
> error_report("load of migration failed: %s", strerror(-ret));
> qemu_fclose(mis->from_src_file);
> + multifd_load_cleanup();
... here we can call migration_incoming_state_destroy() as well, and
we also don't need "qemu_fclose(mis->from_src_file)" any more.
> exit(EXIT_FAILURE);
> }
> mis->bh = qemu_bh_new(process_incoming_migration_bh, mis);
> @@ -368,7 +370,12 @@ void migration_fd_process_incoming(QEMUFile *f)
> {
> Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
> MigrationIncomingState *mis = migration_incoming_get_current();
> -
> +
> + if (multifd_load_setup() != 0) {
> + /* We haven't been able to create multifd threads
> + nothing better to do */
Nit: maybe an error_report() before quit (though I see that
multifd_load_setup() actually never fails)?
> + exit(EXIT_FAILURE);
> + }
> if (!mis->from_src_file) {
> mis->from_src_file = f;
> }
[...]
> +int multifd_load_setup(void)
> +{
> + int thread_count;
> + uint8_t i;
> +
> + if (!migrate_use_multifd()) {
> + return 0;
> + }
> + thread_count = migrate_multifd_threads();
> + multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
> + multifd_recv_state->params = g_new0(MultiFDRecvParams *, thread_count);
> + multifd_recv_state->count = 0;
> + for (i = 0; i < thread_count; i++) {
> + char thread_name[16];
> + MultiFDRecvParams *p = multifd_recv_state->params[i];
> +
> + qemu_mutex_init(&p->mutex);
multifd_recv_state->params is still an array of pointers, and we are
trying to use "p", which is still uninitialized?
> + qemu_sem_init(&p->sem, 0);
> + p->quit = false;
> + p->id = i;
> + snprintf(thread_name, sizeof(thread_name), "multifdrecv_%d", i);
> + qemu_thread_create(&p->thread, thread_name, multifd_recv_thread, p,
> + QEMU_THREAD_JOINABLE);
> + multifd_recv_state->count++;
> + }
> + return 0;
> +}
Thanks,
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming Juan Quintela
2017-08-11 15:10 ` Daniel P. Berrange
@ 2017-08-15 6:43 ` Peter Xu
1 sibling, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-08-15 6:43 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Tue, Aug 08, 2017 at 06:26:20PM +0200, Juan Quintela wrote:
> We need that on later patches.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels Juan Quintela
2017-08-11 15:34 ` Daniel P. Berrange
@ 2017-08-16 16:38 ` Dr. David Alan Gilbert
1 sibling, 0 replies; 61+ messages in thread
From: Dr. David Alan Gilbert @ 2017-08-16 16:38 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, peterx
* Juan Quintela (quintela@redhat.com) wrote:
> We switch for sending the page number to send real pages.
>
> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
> --
>
> Remove the HACK bit, now we have the function that calculates the size
> of a page exported.
> ---
> migration/migration.c | 7 +++++-
> migration/ram.c | 59 +++++++++++++++++----------------------------------
> 2 files changed, 25 insertions(+), 41 deletions(-)
>
> diff --git a/migration/migration.c b/migration/migration.c
> index 974ff92..aac3cdc 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -2051,6 +2051,7 @@ static void *migration_thread(void *opaque)
> */
> int64_t threshold_size = 0;
> int64_t qemu_file_bytes = 0;
> + int64_t multifd_pages = 0;
> int64_t start_time = initial_time;
> int64_t end_time;
> bool old_vm_running = false;
> @@ -2139,8 +2140,11 @@ static void *migration_thread(void *opaque)
> current_time = qemu_clock_get_ms(QEMU_CLOCK_REALTIME);
> if (current_time >= initial_time + BUFFER_DELAY) {
> uint64_t qemu_file_bytes_now = qemu_ftell(s->to_dst_file);
> + uint64_t multifd_pages_now = ram_counters.multifd;
> uint64_t transferred_bytes =
> - qemu_file_bytes_now - qemu_file_bytes;
> + (qemu_file_bytes_now - qemu_file_bytes) +
> + (multifd_pages_now - multifd_pages) *
> + qemu_target_page_size();
> uint64_t time_spent = current_time - initial_time;
> double bandwidth = (double)transferred_bytes / time_spent;
> threshold_size = bandwidth * s->parameters.downtime_limit;
> @@ -2160,6 +2164,7 @@ static void *migration_thread(void *opaque)
> qemu_file_reset_rate_limit(s->to_dst_file);
> initial_time = current_time;
> qemu_file_bytes = qemu_file_bytes_now;
> + multifd_pages = multifd_pages_now;
> }
> if (qemu_file_rate_limit(s->to_dst_file)) {
> /* usleep expects microseconds */
> diff --git a/migration/ram.c b/migration/ram.c
> index 42ad126..f337360 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -479,25 +479,21 @@ static void *multifd_send_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> - int i;
> int num;
>
> num = p->pages.num;
> p->pages.num = 0;
> qemu_mutex_unlock(&p->mutex);
>
> - for (i = 0; i < num; i++) {
> - if (qio_channel_write(p->c,
> - (const char *)&p->pages.iov[i].iov_base,
> - sizeof(uint8_t *), &error_abort)
> - != sizeof(uint8_t *)) {
> - MigrationState *s = migrate_get_current();
> + if (qio_channel_writev_all(p->c, p->pages.iov,
> + num, &error_abort)
One way out of the problem of errors might be to add an
Error *errp;
to MultiFD*Params, and make sure it's NULL initially;
then instead of &error_abort you can have &p->errp,
I think we're already guaranteed the lifetime of the
params is at least the length of the thread and then
you can check p->errp before you free the params.
You could also return this value as the return value of
the thread as an easy way to flag the thread failed when the
join happens.
Dave
> + != num * TARGET_PAGE_SIZE) {
> + MigrationState *s = migrate_get_current();
>
> - migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> - MIGRATION_STATUS_FAILED);
> - terminate_multifd_send_threads();
> - return NULL;
> - }
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_send_threads();
> + return NULL;
> }
> qemu_mutex_lock(&multifd_send_state->mutex);
> p->done = true;
> @@ -658,7 +654,6 @@ void multifd_load_cleanup(void)
> static void *multifd_recv_thread(void *opaque)
> {
> MultiFDRecvParams *p = opaque;
> - uint8_t *recv_address;
>
> qemu_sem_post(&p->ready);
> while (true) {
> @@ -668,38 +663,21 @@ static void *multifd_recv_thread(void *opaque)
> break;
> }
> if (p->pages.num) {
> - int i;
> int num;
>
> num = p->pages.num;
> p->pages.num = 0;
>
> - for (i = 0; i < num; i++) {
> - if (qio_channel_read(p->c,
> - (char *)&recv_address,
> - sizeof(uint8_t *), &error_abort)
> - != sizeof(uint8_t *)) {
> - MigrationState *s = migrate_get_current();
> + if (qio_channel_readv_all(p->c, p->pages.iov,
> + num, &error_abort)
> + != num * TARGET_PAGE_SIZE) {
> + MigrationState *s = migrate_get_current();
>
> - migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> - MIGRATION_STATUS_FAILED);
> - terminate_multifd_recv_threads();
> - return NULL;
> - }
> - if (recv_address != p->pages.iov[i].iov_base) {
> - MigrationState *s = migrate_get_current();
> -
> - printf("We received %p what we were expecting %p (%d)\n",
> - recv_address,
> - p->pages.iov[i].iov_base, i);
> -
> - migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> - MIGRATION_STATUS_FAILED);
> - terminate_multifd_recv_threads();
> - return NULL;
> - }
> + migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
> + MIGRATION_STATUS_FAILED);
> + terminate_multifd_recv_threads();
> + return NULL;
> }
> -
> p->done = true;
> qemu_mutex_unlock(&p->mutex);
> qemu_sem_post(&p->ready);
> @@ -1259,8 +1237,10 @@ static int ram_multifd_page(RAMState *rs, PageSearchStatus *pss,
> offset | RAM_SAVE_FLAG_MULTIFD_PAGE);
> fd_num = multifd_send_page(p, rs->migration_dirty_pages == 1);
> qemu_put_be16(rs->f, fd_num);
> + if (fd_num != MULTIFD_CONTINUE) {
> + qemu_fflush(rs->f);
> + }
> ram_counters.transferred += 2; /* size of fd_num */
> - qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
> ram_counters.transferred += TARGET_PAGE_SIZE;
> pages = 1;
> ram_counters.normal++;
> @@ -3123,7 +3103,6 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
> case RAM_SAVE_FLAG_MULTIFD_PAGE:
> fd_num = qemu_get_be16(f);
> multifd_recv_page(host, fd_num);
> - qemu_get_buffer(f, host, TARGET_PAGE_SIZE);
> break;
>
> case RAM_SAVE_FLAG_EOS:
> --
> 2.9.4
>
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming()
2017-08-11 3:47 ` Peter Xu
@ 2017-09-06 11:07 ` Juan Quintela
2017-09-08 4:16 ` Peter Xu
2017-09-13 9:41 ` Juan Quintela
1 sibling, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-09-06 11:07 UTC (permalink / raw)
To: Peter Xu; +Cc: qemu-devel, dgilbert, lvivier
Peter Xu <peterx@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:11PM +0200, Juan Quintela wrote:
>
> [...]
>
>> void migration_fd_process_incoming(QEMUFile *f)
>> {
>> - Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
>> -
>> + Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
>> + MigrationIncomingState *mis = migration_incoming_get_current();
>> +
>> + if (!mis->from_src_file) {
>> + mis->from_src_file = f;
>
> [1]
>
>> + }
>> qemu_file_set_blocking(f, false);
>> qemu_coroutine_enter(co);
>> }
>>
>> +void migration_ioc_process_incoming(QIOChannel *ioc)
>> +{
>> + MigrationIncomingState *mis = migration_incoming_get_current();
>> +
>> + if (!mis->from_src_file) {
>> + QEMUFile *f = qemu_fopen_channel_input(ioc);
>> + mis->from_src_file = f;
>
> Remove this line? Since migration_fd_process_incoming() will set it up
> as well below at [1].
>
> Then we can make sure there will be only one place to setup
> from_src_file.
This can be end being called from rdma code.
Yes, I know it is a mess.
Later, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming()
2017-09-06 11:07 ` Juan Quintela
@ 2017-09-08 4:16 ` Peter Xu
0 siblings, 0 replies; 61+ messages in thread
From: Peter Xu @ 2017-09-08 4:16 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, dgilbert, lvivier
On Wed, Sep 06, 2017 at 01:07:47PM +0200, Juan Quintela wrote:
> Peter Xu <peterx@redhat.com> wrote:
> > On Tue, Aug 08, 2017 at 06:26:11PM +0200, Juan Quintela wrote:
> >
> > [...]
> >
> >> void migration_fd_process_incoming(QEMUFile *f)
> >> {
> >> - Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
> >> -
> >> + Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
> >> + MigrationIncomingState *mis = migration_incoming_get_current();
> >> +
> >> + if (!mis->from_src_file) {
> >> + mis->from_src_file = f;
> >
> > [1]
> >
> >> + }
> >> qemu_file_set_blocking(f, false);
> >> qemu_coroutine_enter(co);
> >> }
> >>
> >> +void migration_ioc_process_incoming(QIOChannel *ioc)
> >> +{
> >> + MigrationIncomingState *mis = migration_incoming_get_current();
> >> +
> >> + if (!mis->from_src_file) {
> >> + QEMUFile *f = qemu_fopen_channel_input(ioc);
> >> + mis->from_src_file = f;
> >
> > Remove this line? Since migration_fd_process_incoming() will set it up
> > as well below at [1].
> >
> > Then we can make sure there will be only one place to setup
> > from_src_file.
>
> This can be end being called from rdma code.
>
> Yes, I know it is a mess.
But isn't RDMA code calling migration_fd_process_incoming()? Btw, I
meant removing the one in migration_ioc_process_incoming(), and keep
the one in migration_fd_process_incoming().
(did I miss anything?)
--
Peter Xu
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming()
2017-08-11 3:47 ` Peter Xu
2017-09-06 11:07 ` Juan Quintela
@ 2017-09-13 9:41 ` Juan Quintela
1 sibling, 0 replies; 61+ messages in thread
From: Juan Quintela @ 2017-09-13 9:41 UTC (permalink / raw)
To: Peter Xu; +Cc: qemu-devel, dgilbert, lvivier
Peter Xu <peterx@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:11PM +0200, Juan Quintela wrote:
>
> [...]
>
>> void migration_fd_process_incoming(QEMUFile *f)
>> {
>> - Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, f);
>> -
>> + Coroutine *co = qemu_coroutine_create(process_incoming_migration_co, NULL);
>> + MigrationIncomingState *mis = migration_incoming_get_current();
>> +
>> + if (!mis->from_src_file) {
>> + mis->from_src_file = f;
>
> [1]
>
>> + }
>> qemu_file_set_blocking(f, false);
>> qemu_coroutine_enter(co);
>> }
>>
>> +void migration_ioc_process_incoming(QIOChannel *ioc)
>> +{
>> + MigrationIncomingState *mis = migration_incoming_get_current();
>> +
>> + if (!mis->from_src_file) {
>> + QEMUFile *f = qemu_fopen_channel_input(ioc);
>> + mis->from_src_file = f;
>
> Remove this line? Since migration_fd_process_incoming() will set it up
> as well below at [1].
>
> Then we can make sure there will be only one place to setup
> from_src_file.
Done.
I still think that it is weird that we have both:
- mis
- f
in one function and wait to asign that to other function, but I agree
that it is also weird to set it up in two places.
Later, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side
2017-08-11 15:29 ` Daniel P. Berrange
@ 2017-09-13 9:53 ` Juan Quintela
2017-09-13 11:26 ` Juan Quintela
1 sibling, 0 replies; 61+ messages in thread
From: Juan Quintela @ 2017-09-13 9:53 UTC (permalink / raw)
To: Daniel P. Berrange; +Cc: qemu-devel, lvivier, dgilbert, peterx
"Daniel P. Berrange" <berrange@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:25PM +0200, Juan Quintela wrote:
>> 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 <quintela@redhat.com>
>>
>> --
>>
>> 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_group().
>> ---
>> 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/socket.c b/migration/socket.c
>> index 5dd6f42..3af9f7c 100644
>> --- a/migration/socket.c
>> +++ b/migration/socket.c
>> @@ -183,12 +183,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));
>
> AFAICT, migration_channel_process_incoming() acquires its own reference
> on 'sioc', so removing this object_unref means the code is now leaking a
> reference
Done.
Thanks, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter
2017-08-11 14:56 ` Daniel P. Berrange
@ 2017-09-13 10:12 ` Juan Quintela
2017-09-13 10:18 ` Daniel P. Berrange
0 siblings, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-09-13 10:12 UTC (permalink / raw)
To: Daniel P. Berrange; +Cc: qemu-devel, lvivier, dgilbert, peterx
"Daniel P. Berrange" <berrange@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:17PM +0200, Juan Quintela wrote:
>> Indicates the number of threads that we would create. By default we
>> create 2 threads.
>> # migrated and the destination must already have access to the
>> # same backing chain as was used on the source. (since 2.10)
>> #
>> +# @x-multifd-threads: Number of threads used to migrate data in
>> +# parallel. This is the same number that the
>> +# number of sockets used for migration.
>> +# The default value is 2 (since 2.11)
>
> The feature is called "multifd" but we're configuring number of threads,
> with the implicit suggestion that number of threads == number of sockets.
>
> I wonder if its better to call this "x-multifd-connections" so that it
> is explicit that we're determining the number of socket connections.
>
> This leaves room for adapting code to use a different number of threads,
> but same number of sockets. eg with post-copy, there's potential to have
> 2 threads for each socket (1 reading, 1 writing).
Changed to channels, is that ok for you?
Later, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter
2017-09-13 10:12 ` Juan Quintela
@ 2017-09-13 10:18 ` Daniel P. Berrange
0 siblings, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-09-13 10:18 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Wed, Sep 13, 2017 at 12:12:39PM +0200, Juan Quintela wrote:
> "Daniel P. Berrange" <berrange@redhat.com> wrote:
> > On Tue, Aug 08, 2017 at 06:26:17PM +0200, Juan Quintela wrote:
> >> Indicates the number of threads that we would create. By default we
> >> create 2 threads.
> >> # migrated and the destination must already have access to the
> >> # same backing chain as was used on the source. (since 2.10)
> >> #
> >> +# @x-multifd-threads: Number of threads used to migrate data in
> >> +# parallel. This is the same number that the
> >> +# number of sockets used for migration.
> >> +# The default value is 2 (since 2.11)
> >
> > The feature is called "multifd" but we're configuring number of threads,
> > with the implicit suggestion that number of threads == number of sockets.
> >
> > I wonder if its better to call this "x-multifd-connections" so that it
> > is explicit that we're determining the number of socket connections.
> >
> > This leaves room for adapting code to use a different number of threads,
> > but same number of sockets. eg with post-copy, there's potential to have
> > 2 threads for each socket (1 reading, 1 writing).
>
> Changed to channels, is that ok for you?
Sure
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter
2017-08-11 15:03 ` Daniel P. Berrange
@ 2017-09-13 10:25 ` Juan Quintela
0 siblings, 0 replies; 61+ messages in thread
From: Juan Quintela @ 2017-09-13 10:25 UTC (permalink / raw)
To: Daniel P. Berrange; +Cc: qemu-devel, lvivier, dgilbert, peterx
"Daniel P. Berrange" <berrange@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:18PM +0200, Juan Quintela wrote:
>> Indicates how many pages we are going to send in each batch to a multifd
>> thread.
>>
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
>>
>> --
>>
>> Be consistent with defaults and documentation
>> Use new DEFINE_PROP_*
>> @@ -1117,6 +1120,9 @@
>> # number of sockets used for migration.
>> # The default value is 2 (since 2.11)
>> #
>> +# @x-multifd-group: Number of pages sent together to a thread
>> +# The default value is 16 (since 2.11)
>
> Bikesheeding - how about "x-multifd-page-count" ? 'group' didn't
> give me much idea what it was controlling until reading the docs.
Done, thanks.
I was about to change it to pages, but once here, the length don't
matter.
Later, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side
2017-08-11 15:29 ` Daniel P. Berrange
2017-09-13 9:53 ` Juan Quintela
@ 2017-09-13 11:26 ` Juan Quintela
2017-09-13 11:45 ` Daniel P. Berrange
1 sibling, 1 reply; 61+ messages in thread
From: Juan Quintela @ 2017-09-13 11:26 UTC (permalink / raw)
To: Daniel P. Berrange; +Cc: qemu-devel, lvivier, dgilbert, peterx
"Daniel P. Berrange" <berrange@redhat.com> wrote:
> On Tue, Aug 08, 2017 at 06:26:25PM +0200, Juan Quintela wrote:
>> 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 <quintela@redhat.com>
>>
>> --
>>
>> 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_group().
>> ---
>> 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/socket.c b/migration/socket.c
>> index 5dd6f42..3af9f7c 100644
>> --- a/migration/socket.c
>> +++ b/migration/socket.c
>> @@ -183,12 +183,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));
>
> AFAICT, migration_channel_process_incoming() acquires its own reference
> on 'sioc', so removing this object_unref means the code is now leaking a
> reference
Nack.
I did it and ended with a segmentation fault because reference count was
bad.
(qemu)
Thread 6 "multifdrecv_0" received signal SIGSEGV, Segmentation fault.
[Switching to Thread 0x7fff9f9ff700 (LWP 10065)]
0x00005555559c12f8 in type_is_ancestor (type=0xb60f18247c894860,
target_type=0x555556531b30) at /mnt/kvm/qemu/cleanup/qom/object.c:217
217 while (type) {
(gdb) bt
#0 0x00005555559c12f8 in type_is_ancestor (type=0xb60f18247c894860, target_type=0x555556531b30) at /mnt/kvm/qemu/cleanup/qom/object.c:217
#1 0x00005555559c1dbb in object_class_dynamic_cast (class=class@entry=0x7ffff2cb3ce8 <main_arena+520>, typename=typename@entry=0x555555aa6d91 "qio-channel")
at /mnt/kvm/qemu/cleanup/qom/object.c:691
#2 0x00005555559c1ed2 in object_class_dynamic_cast_assert (class=0x7ffff2cb3ce8 <main_arena+520>, typename=typename@entry=0x555555aa6d91 "qio-channel", file=file@entry=0x555555b88208 "/mnt/kvm/qemu/cleanup/io/channel.c", line=line@entry=56, func=func@entry=0x555555b884e0 <__func__.22671> "qio_channel_readv_full")
at /mnt/kvm/qemu/cleanup/qom/object.c:723
#3 0x0000555555a3d4d7 in qio_channel_readv_full (ioc=0x5555568c5a00, iov=0x7fff900008c0, niov=15, fds=0x0, nfds=0x0, errp=0x7fff9f9fe950)
at /mnt/kvm/qemu/cleanup/io/channel.c:56
#4 0x0000555555a3ddc2 in qio_channel_readv (errp=0x7fff9f9fe950, niov=<optimized out>, iov=<optimized out>, ioc=0x5555568c5a00)
at /mnt/kvm/qemu/cleanup/io/channel.c:197
#5 0x0000555555a3ddc2 in qio_channel_readv_all_eof (ioc=0x5555568c5a00, iov=<optimized out>, niov=<optimized out>, errp=errp@entry=0x7fff9f9fe950)
at /mnt/kvm/qemu/cleanup/io/channel.c:106
#6 0x0000555555a3de79 in qio_channel_readv_all (ioc=<optimized out>, iov=<optimized out>, niov=<optimized out>, errp=errp@entry=0x7fff9f9fe950)
at /mnt/kvm/qemu/cleanup/io/channel.c:142
#7 0x0000555555794768 in multifd_recv_thread (opaque=0x5555570e5e00)
---Type <return> to continue, or q <return> to quit---up
at /mnt/kvm/qemu/cleanup/migration/ram.c:722
#8 0x00007ffff2cc036d in start_thread (arg=0x7fff9f9ff700)
at pthread_create.c:456
#9 0x00007ffff29f8bbf in clone ()
at ../sysdeps/unix/sysv/linux/x86_64/clone.S:97
(gdb) up
Removing this unref fixed things, so I think that the accounting is good
(famous last words).
Later, Juan.
^ permalink raw reply [flat|nested] 61+ messages in thread
* Re: [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side
2017-09-13 11:26 ` Juan Quintela
@ 2017-09-13 11:45 ` Daniel P. Berrange
0 siblings, 0 replies; 61+ messages in thread
From: Daniel P. Berrange @ 2017-09-13 11:45 UTC (permalink / raw)
To: Juan Quintela; +Cc: qemu-devel, lvivier, dgilbert, peterx
On Wed, Sep 13, 2017 at 01:26:07PM +0200, Juan Quintela wrote:
> "Daniel P. Berrange" <berrange@redhat.com> wrote:
> > On Tue, Aug 08, 2017 at 06:26:25PM +0200, Juan Quintela wrote:
> >> 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 <quintela@redhat.com>
> >>
> >> --
> >>
> >> 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_group().
> >> ---
> >> 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/socket.c b/migration/socket.c
> >> index 5dd6f42..3af9f7c 100644
> >> --- a/migration/socket.c
> >> +++ b/migration/socket.c
> >> @@ -183,12 +183,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));
> >
> > AFAICT, migration_channel_process_incoming() acquires its own reference
> > on 'sioc', so removing this object_unref means the code is now leaking a
> > reference
>
> Nack.
>
> I did it and ended with a segmentation fault because reference count was
> bad.
>
> (qemu)
> Thread 6 "multifdrecv_0" received signal SIGSEGV, Segmentation fault.
> [Switching to Thread 0x7fff9f9ff700 (LWP 10065)]
> 0x00005555559c12f8 in type_is_ancestor (type=0xb60f18247c894860,
> target_type=0x555556531b30) at /mnt/kvm/qemu/cleanup/qom/object.c:217
> 217 while (type) {
> (gdb) bt
> #0 0x00005555559c12f8 in type_is_ancestor (type=0xb60f18247c894860, target_type=0x555556531b30) at /mnt/kvm/qemu/cleanup/qom/object.c:217
> #1 0x00005555559c1dbb in object_class_dynamic_cast (class=class@entry=0x7ffff2cb3ce8 <main_arena+520>, typename=typename@entry=0x555555aa6d91 "qio-channel")
> at /mnt/kvm/qemu/cleanup/qom/object.c:691
> #2 0x00005555559c1ed2 in object_class_dynamic_cast_assert (class=0x7ffff2cb3ce8 <main_arena+520>, typename=typename@entry=0x555555aa6d91 "qio-channel", file=file@entry=0x555555b88208 "/mnt/kvm/qemu/cleanup/io/channel.c", line=line@entry=56, func=func@entry=0x555555b884e0 <__func__.22671> "qio_channel_readv_full")
> at /mnt/kvm/qemu/cleanup/qom/object.c:723
> #3 0x0000555555a3d4d7 in qio_channel_readv_full (ioc=0x5555568c5a00, iov=0x7fff900008c0, niov=15, fds=0x0, nfds=0x0, errp=0x7fff9f9fe950)
> at /mnt/kvm/qemu/cleanup/io/channel.c:56
> #4 0x0000555555a3ddc2 in qio_channel_readv (errp=0x7fff9f9fe950, niov=<optimized out>, iov=<optimized out>, ioc=0x5555568c5a00)
> at /mnt/kvm/qemu/cleanup/io/channel.c:197
> #5 0x0000555555a3ddc2 in qio_channel_readv_all_eof (ioc=0x5555568c5a00, iov=<optimized out>, niov=<optimized out>, errp=errp@entry=0x7fff9f9fe950)
> at /mnt/kvm/qemu/cleanup/io/channel.c:106
> #6 0x0000555555a3de79 in qio_channel_readv_all (ioc=<optimized out>, iov=<optimized out>, niov=<optimized out>, errp=errp@entry=0x7fff9f9fe950)
> at /mnt/kvm/qemu/cleanup/io/channel.c:142
> #7 0x0000555555794768 in multifd_recv_thread (opaque=0x5555570e5e00)
> ---Type <return> to continue, or q <return> to quit---up
> at /mnt/kvm/qemu/cleanup/migration/ram.c:722
> #8 0x00007ffff2cc036d in start_thread (arg=0x7fff9f9ff700)
> at pthread_create.c:456
> #9 0x00007ffff29f8bbf in clone ()
> at ../sysdeps/unix/sysv/linux/x86_64/clone.S:97
> (gdb) up
>
> Removing this unref fixed things, so I think that the accounting is good
> (famous last words).
No, this is just papering over a bug elsewhere. The multifd_new_channel
metohd needs to be acquiring a reference for the multifd_recv_thread to
hold onto.
Regards,
Daniel
--
|: https://berrange.com -o- https://www.flickr.com/photos/dberrange :|
|: https://libvirt.org -o- https://fstop138.berrange.com :|
|: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
^ permalink raw reply [flat|nested] 61+ messages in thread
* [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work
2017-08-08 16:22 Juan Quintela
@ 2017-08-08 16:22 ` Juan Quintela
0 siblings, 0 replies; 61+ messages in thread
From: Juan Quintela @ 2017-08-08 16:22 UTC (permalink / raw)
To: qemu-devel; +Cc: dgilbert, lvivier, peterx
We create new channels for each new thread created. We send through
them a string containing <uuid> multifd <channel number> so we are
sure that we connect the right channels in both sides.
Signed-off-by: Juan Quintela <quintela@redhat.com>
--
Split SocketArgs into incoming and outgoing args
Use UUID's on the initial message, so we are sure we are connecting to
the right channel.
Remove init semaphore. Now that we use uuids on the init message, we
know that this is our channel.
Fix recv socket destwroy, we were destroying send channels.
This was very interesting, because we were using an unreferred object
without problems.
Move to struct of pointers
init channel sooner.
split recv thread creation.
listen on main thread
We count the number of created threads to know when we need to stop listening
Use g_strdup_printf
report channel id on errors
---
migration/migration.c | 5 +++
migration/ram.c | 99 +++++++++++++++++++++++++++++++++++++++++++--------
migration/ram.h | 3 ++
migration/socket.c | 36 ++++++++++++++++++-
migration/socket.h | 10 ++++++
5 files changed, 138 insertions(+), 15 deletions(-)
diff --git a/migration/migration.c b/migration/migration.c
index e36e880..944d6e2 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -413,6 +413,11 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
*/
bool migration_has_all_channels(void)
{
+ if (migrate_use_multifd()) {
+ int thread_count = migrate_multifd_threads();
+
+ return thread_count == multifd_created_threads();
+ }
return true;
}
diff --git a/migration/ram.c b/migration/ram.c
index 9fb3496..e9fa556 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -36,6 +36,7 @@
#include "xbzrle.h"
#include "ram.h"
#include "migration.h"
+#include "socket.h"
#include "migration/register.h"
#include "migration/misc.h"
#include "qemu-file.h"
@@ -46,6 +47,8 @@
#include "exec/ram_addr.h"
#include "qemu/rcu_queue.h"
#include "migration/colo.h"
+#include "sysemu/sysemu.h"
+#include "qemu/uuid.h"
/***********************************************************/
/* ram save/restore */
@@ -361,6 +364,7 @@ static void compress_threads_save_setup(void)
struct MultiFDSendParams {
uint8_t id;
QemuThread thread;
+ QIOChannel *c;
QemuSemaphore sem;
QemuMutex mutex;
bool quit;
@@ -401,6 +405,7 @@ void multifd_save_cleanup(void)
qemu_thread_join(&p->thread);
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem);
+ socket_send_channel_destroy(p->c);
}
g_free(multifd_send_state->params);
multifd_send_state->params = NULL;
@@ -408,9 +413,26 @@ void multifd_save_cleanup(void)
multifd_send_state = NULL;
}
+/* Default uuid for multifd when qemu is not started with uuid */
+static char multifd_uuid[] = "5c49fd7e-af88-4a07-b6e8-091fd696ad40";
+/* strlen(multifd) + '-' + <channel id> + '-' + UUID_FMT + '\0' */
+#define MULTIFD_UUID_MSG (7 + 1 + 3 + 1 + UUID_FMT_LEN + 1)
+
static void *multifd_send_thread(void *opaque)
{
MultiFDSendParams *p = opaque;
+ char *string;
+ char *string_uuid;
+
+ if (qemu_uuid_set) {
+ string_uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
+ } else {
+ string_uuid = g_strdup(multifd_uuid);
+ }
+ string = g_strdup_printf("%s multifd %03d", string_uuid, p->id);
+ g_free(string_uuid);
+ qio_channel_write(p->c, string, MULTIFD_UUID_MSG, &error_abort);
+ g_free(string);
while (true) {
qemu_mutex_lock(&p->mutex);
@@ -445,6 +467,12 @@ int multifd_save_setup(void)
qemu_sem_init(&p->sem, 0);
p->quit = false;
p->id = i;
+ p->c = socket_send_channel_create();
+ if (!p->c) {
+ error_report("Error creating a send channel");
+ multifd_save_cleanup();
+ return -1;
+ }
snprintf(thread_name, sizeof(thread_name), "multifdsend_%d", i);
qemu_thread_create(&p->thread, thread_name, multifd_send_thread, p,
QEMU_THREAD_JOINABLE);
@@ -456,6 +484,7 @@ int multifd_save_setup(void)
struct MultiFDRecvParams {
uint8_t id;
QemuThread thread;
+ QIOChannel *c;
QemuSemaphore sem;
QemuMutex mutex;
bool quit;
@@ -466,12 +495,16 @@ struct {
MultiFDRecvParams **params;
/* number of created threads */
int count;
+ /* Should we finish */
+ bool quit;
} *multifd_recv_state;
static void terminate_multifd_recv_threads(void)
{
int i;
+ multifd_recv_state->quit = true;
+
for (i = 0; i < multifd_recv_state->count; i++) {
MultiFDRecvParams *p = multifd_recv_state->params[i];
@@ -496,6 +529,7 @@ void multifd_load_cleanup(void)
qemu_thread_join(&p->thread);
qemu_mutex_destroy(&p->mutex);
qemu_sem_destroy(&p->sem);
+ socket_recv_channel_destroy(p->c);
g_free(p);
multifd_recv_state->params[i] = NULL;
}
@@ -522,10 +556,54 @@ static void *multifd_recv_thread(void *opaque)
return NULL;
}
+void multifd_new_channel(QIOChannel *ioc)
+{
+ MultiFDRecvParams *p = g_new0(MultiFDRecvParams, 1);
+ MigrationState *s = migrate_get_current();
+ char string[MULTIFD_UUID_MSG];
+ char string_uuid[UUID_FMT_LEN];
+ char *uuid;
+ int id;
+
+ qio_channel_read(ioc, string, sizeof(string), &error_abort);
+ sscanf(string, "%s multifd %03d", string_uuid, &id);
+
+ if (qemu_uuid_set) {
+ uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
+ } else {
+ uuid = g_strdup(multifd_uuid);
+ }
+ if (strcmp(string_uuid, uuid)) {
+ error_report("multifd: received uuid '%s' and expected uuid '%s'"
+ " for channel %d", string_uuid, uuid, id);
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return;
+ }
+ g_free(uuid);
+
+ if (multifd_recv_state->params[id] != NULL) {
+ error_report("multifd: received id '%d' is already setup'", id);
+ migrate_set_state(&s->state, MIGRATION_STATUS_ACTIVE,
+ MIGRATION_STATUS_FAILED);
+ terminate_multifd_recv_threads();
+ return;
+ }
+ qemu_mutex_init(&p->mutex);
+ qemu_sem_init(&p->sem, 0);
+ p->quit = false;
+ p->id = id;
+ p->c = ioc;
+ atomic_set(&multifd_recv_state->params[id], p);
+ multifd_recv_state->count++;
+ qemu_thread_create(&p->thread, "multifd_recv", multifd_recv_thread, p,
+ QEMU_THREAD_JOINABLE);
+}
+
int multifd_load_setup(void)
{
int thread_count;
- uint8_t i;
if (!migrate_use_multifd()) {
return 0;
@@ -534,22 +612,15 @@ int multifd_load_setup(void)
multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
multifd_recv_state->params = g_new0(MultiFDRecvParams *, thread_count);
multifd_recv_state->count = 0;
- for (i = 0; i < thread_count; i++) {
- char thread_name[16];
- MultiFDRecvParams *p = multifd_recv_state->params[i];
-
- qemu_mutex_init(&p->mutex);
- qemu_sem_init(&p->sem, 0);
- p->quit = false;
- p->id = i;
- snprintf(thread_name, sizeof(thread_name), "multifdrecv_%d", i);
- qemu_thread_create(&p->thread, thread_name, multifd_recv_thread, p,
- QEMU_THREAD_JOINABLE);
- multifd_recv_state->count++;
- }
+ multifd_recv_state->quit = false;
return 0;
}
+int multifd_created_threads(void)
+{
+ return multifd_recv_state->count;
+}
+
/**
* save_page_header: write page header to wire
*
diff --git a/migration/ram.h b/migration/ram.h
index 93c2bb4..b3dabf2 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -31,6 +31,7 @@
#include "qemu-common.h"
#include "exec/cpu-common.h"
+#include "io/channel.h"
extern MigrationStats ram_counters;
extern XBZRLECacheStats xbzrle_counters;
@@ -43,6 +44,8 @@ int multifd_save_setup(void);
void multifd_save_cleanup(void);
int multifd_load_setup(void);
void multifd_load_cleanup(void);
+void multifd_new_channel(QIOChannel *ioc);
+int multifd_created_threads(void);
uint64_t ram_pagesize_summary(void);
int ram_save_queue_pages(const char *rbname, ram_addr_t start, ram_addr_t len);
diff --git a/migration/socket.c b/migration/socket.c
index dee8690..5dd6f42 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -26,6 +26,38 @@
#include "io/channel-socket.h"
#include "trace.h"
+int socket_recv_channel_destroy(QIOChannel *recv)
+{
+ /* Remove channel */
+ object_unref(OBJECT(recv));
+ return 0;
+}
+
+struct SocketOutgoingArgs {
+ SocketAddress *saddr;
+ Error **errp;
+} outgoing_args;
+
+QIOChannel *socket_send_channel_create(void)
+{
+ QIOChannelSocket *sioc = qio_channel_socket_new();
+
+ qio_channel_socket_connect_sync(sioc, outgoing_args.saddr,
+ outgoing_args.errp);
+ qio_channel_set_delay(QIO_CHANNEL(sioc), false);
+ return QIO_CHANNEL(sioc);
+}
+
+int socket_send_channel_destroy(QIOChannel *send)
+{
+ /* Remove channel */
+ object_unref(OBJECT(send));
+ if (outgoing_args.saddr) {
+ qapi_free_SocketAddress(outgoing_args.saddr);
+ outgoing_args.saddr = NULL;
+ }
+ return 0;
+}
static SocketAddress *tcp_build_address(const char *host_port, Error **errp)
{
@@ -96,6 +128,9 @@ static void socket_start_outgoing_migration(MigrationState *s,
struct SocketConnectData *data = g_new0(struct SocketConnectData, 1);
data->s = s;
+ outgoing_args.saddr = saddr;
+ outgoing_args.errp = errp;
+
if (saddr->type == SOCKET_ADDRESS_TYPE_INET) {
data->hostname = g_strdup(saddr->u.inet.host);
}
@@ -106,7 +141,6 @@ static void socket_start_outgoing_migration(MigrationState *s,
socket_outgoing_migration,
data,
socket_connect_data_free);
- qapi_free_SocketAddress(saddr);
}
void tcp_start_outgoing_migration(MigrationState *s,
diff --git a/migration/socket.h b/migration/socket.h
index 6b91e9d..dabce0e 100644
--- a/migration/socket.h
+++ b/migration/socket.h
@@ -16,6 +16,16 @@
#ifndef QEMU_MIGRATION_SOCKET_H
#define QEMU_MIGRATION_SOCKET_H
+
+#include "io/channel.h"
+
+QIOChannel *socket_recv_channel_create(void);
+int socket_recv_channel_destroy(QIOChannel *recv);
+
+QIOChannel *socket_send_channel_create(void);
+
+int socket_send_channel_destroy(QIOChannel *send);
+
void tcp_start_incoming_migration(const char *host_port, Error **errp);
void tcp_start_outgoing_migration(MigrationState *s, const char *host_port,
--
2.9.4
^ permalink raw reply related [flat|nested] 61+ messages in thread
end of thread, other threads:[~2017-09-13 11:46 UTC | newest]
Thread overview: 61+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2017-08-08 16:26 [Qemu-devel] [PATCH v6 00/19] Multifd Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 01/19] migration: Create migration_ioc_process_incoming() Juan Quintela
2017-08-11 3:47 ` Peter Xu
2017-09-06 11:07 ` Juan Quintela
2017-09-08 4:16 ` Peter Xu
2017-09-13 9:41 ` Juan Quintela
2017-08-11 14:50 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 02/19] migration: Teach it about G_SOURCE_REMOVE Juan Quintela
2017-08-10 14:56 ` Daniel P. Berrange
2017-08-11 3:49 ` Peter Xu
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 03/19] migration: Add comments to channel functions Juan Quintela
2017-08-10 14:57 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 04/19] migration: Create migration_has_all_channels Juan Quintela
2017-08-10 14:58 ` Daniel P. Berrange
2017-08-10 15:17 ` Eric Blake
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 05/19] qio: Create new qio_channel_{readv, writev}_all Juan Quintela
2017-08-10 15:04 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 06/19] migration: Add multifd capability Juan Quintela
2017-08-11 7:14 ` Peter Xu
2017-08-11 14:52 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 07/19] migration: Create x-multifd-threads parameter Juan Quintela
2017-08-11 7:15 ` Peter Xu
2017-08-11 14:56 ` Daniel P. Berrange
2017-09-13 10:12 ` Juan Quintela
2017-09-13 10:18 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 08/19] migration: Create x-multifd-group parameter Juan Quintela
2017-08-11 7:16 ` Peter Xu
2017-08-11 15:03 ` Daniel P. Berrange
2017-09-13 10:25 ` Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 09/19] migration: Create multifd migration threads Juan Quintela
2017-08-15 6:42 ` Peter Xu
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 10/19] migration: Split migration_fd_process_incoming Juan Quintela
2017-08-11 15:10 ` Daniel P. Berrange
2017-08-15 6:43 ` Peter Xu
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work Juan Quintela
2017-08-11 15:22 ` Daniel P. Berrange
2017-08-14 13:43 ` Juan Quintela
2017-08-14 13:50 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 12/19] migration: Create ram_multifd_page Juan Quintela
2017-08-11 15:25 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 13/19] migration: Really use multiple pages at a time Juan Quintela
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 14/19] migration: Send the fd number which we are going to use for this page Juan Quintela
2017-08-14 15:47 ` Dr. David Alan Gilbert
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 15/19] migration: Create thread infrastructure for multifd recv side Juan Quintela
2017-08-11 4:35 ` Peter Xu
2017-08-11 15:29 ` Daniel P. Berrange
2017-09-13 9:53 ` Juan Quintela
2017-09-13 11:26 ` Juan Quintela
2017-09-13 11:45 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 16/19] migration: Test new fd infrastructure Juan Quintela
2017-08-11 15:32 ` Daniel P. Berrange
2017-08-11 15:35 ` Daniel P. Berrange
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 17/19] migration: Rename initial_bytes Juan Quintela
2017-08-11 19:01 ` Dr. David Alan Gilbert
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 18/19] migration: Transfer pages over new channels Juan Quintela
2017-08-11 15:34 ` Daniel P. Berrange
2017-08-16 16:38 ` Dr. David Alan Gilbert
2017-08-08 16:26 ` [Qemu-devel] [PATCH v6 19/19] migration: Flush receive queue Juan Quintela
2017-08-11 4:16 ` Peter Xu
2017-08-08 23:37 ` [Qemu-devel] [PATCH v6 00/19] Multifd no-reply
-- strict thread matches above, loose matches on Subject: below --
2017-08-08 16:22 Juan Quintela
2017-08-08 16:22 ` [Qemu-devel] [PATCH v6 11/19] migration: Start of multiple fd work Juan Quintela
This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.