[Qemu-devel] [RFC v10 00/24] Multifd

[Qemu-devel] [RFC v10 00/24] Multifd

[Juan Quintela]

Hi  this is the v10 version of the multifd patches,

Lots of changes from previous versions:
a - everything is sent now through the multifd channels, nothing is sent through main channel
b - locking is band new, I was getting into a hole with the previous approach, right now, there is a single way to
    do locking (both source and destination)
       main thread : sets a ->sync variable for each thread and wakeps it
       multifd threads: clean the variable and signal (sem) back to main thread

    using this for either:
    - all threads have started
    - we need to synchronize after each round through memory
    - all threads have finished

c - I have to use a qio watcher for a thread to wait for ready data to read

d - lots of cleanups

e - to make things easier, I have included the missing tests stuff on
    this round of patches, because they build on top of them

f - lots of traces, it is now much easier to follow what is happening

Now, why it is an RFC:

- in the last patch, there is still race between the whatcher, the
  ->quit of the threads and the last synchronization.  Techinically they
  are done in oder, but in practice, they are hanging sometimes.

- I *know* I can optimize the synchronization of the threads sending
  the "we start a new round" through the multifd channels, have to add a flag here.

- Not having a thread on the incoming side  is a mess, I can't block waiting for things to happen :-(

- When doing the synchronization, I need to optimize the sending of the "not finished packet" of pages, working on that.

please, take a look and review.

Thanks, Juan.

[v9]

This series is on top of my migration test series just sent, only reject should be on the test system, though.

On v9 series for you:
- qobject_unref() as requested by dan

  Yes he was right, I had a reference leak for _non_ multifd, I
  *thought* he mean for multifd, and that took a while to understand
  (and then find when/where).

- multifd page count: it is dropped for good
- uuid handling: we use the default qemu uuid of 0000...
- uuid handling: using and struct and sending the struct
  * idea is to add a size field and add more parameter after that
  * anyone has a good idea how to "ouptut" info
    migrate_capabilities/parameters json into a string and how to read it back?
- changed how we test that all threads/channels are already created.
  Should be more robust.
- Add tests multifd.  Still not ported on top of migration-tests series sent early
  waiting for review on the ideas there.
- Rebase and remove al the integrated patches (back at 12)

Please, review.

Later, Juan.

[v8]
Things NOT done yet:

- drop x-multifd-page-count?  We can use performance to set a default value
- paolo suggestion of not having a control channel
  needs iyet more cleanups to be able to have more than one ramstate, trying it.
- still not performance done, but it has been very stable

On v8:
- use connect_async
- rename multifd-group to multifd-page-count (danp suggestion)
- rename multifd-threads to multifd-channels (danp suggestion)
- use new qio*channel functions
- Address rest of comments left


So, please review.

My idea will be to pull this changes and continue performance changes
for inside, basically everything is already reviewed.

Thanks, Juan.

On v7:
- tests fixed as danp wanted
- have to revert danp qio_*_all patches, as they break multifd, I have to investigate why.
- error_abort is gone.  After several tries about getting errors, I ended having a single error
  proceted by a lock and first error wins.
- Addressed basically all reviews (see on ToDo)
- Pointers to struct are done now
- fix lots of leaks
- lots of small fixes


[v6]
- 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.

*** BLURB HERE ***

Juan Quintela (24):
  tests: Add migration precopy test
  tests: Add migration xbzrle test
  migration: Create tcp_port parameter
  migration: Set the migration tcp port
  tests: Migration ppc now inlines its program
  tests: Add basic migration precopy tcp test
  [RFH] tests: Add migration compress threads tests
  migration: Add multifd test
  migration: Set error state in case of error
  migration: In case of error just end the migration
  migration: terminate_* can be called for other threads
  migration: Reference counting recv channels correctly
  migration: Introduce multifd_recv_new_channel()
  migration: Be sure all recv channels are created
  migration: Synchronize send threads
  migration: Synchronize recv threads
  migration: Export functions to create send channels
  migration: Add multifd traces for start/end thread
  migration: Create multifd channels
  migration: Delay start of migration main routines
  migration: Transmit initial package through the multifd channels
  migration: Create ram_multifd_page
  migration: Create pages structure for reception
  [RFC] migration: Send pages through the multifd channels

 hmp.c                  |   3 +
 migration/migration.c  |  31 ++-
 migration/migration.h  |   3 +
 migration/ram.c        | 504 +++++++++++++++++++++++++++++++++++++++++++++++--
 migration/ram.h        |   3 +
 migration/socket.c     |  83 +++++++-
 migration/socket.h     |  10 +
 migration/trace-events |   8 +
 qapi/migration.json    |  19 +-
 tests/migration-test.c | 309 ++++++++++++++++++++++++++----
 10 files changed, 898 insertions(+), 75 deletions(-)

-- 
2.14.3

[Qemu-devel] [PATCH v10 01/24] tests: Add migration precopy test

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
---
 tests/migration-test.c | 44 ++++++++++++++++++++++++++++++++++++++++++--
 1 file changed, 42 insertions(+), 2 deletions(-)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index 74f9361bdd..65ce3ea4ab 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -513,7 +513,7 @@ static void test_deprecated(void)
     qtest_quit(from);
 }
 
-static void test_migrate(void)
+static void test_postcopy(void)
 {
     char *uri = g_strdup_printf("unix:%s/migsocket", tmpfs);
     QTestState *from, *to;
@@ -584,6 +584,45 @@ static void test_baddest(void)
     test_migrate_end(from, to, false);
 }
 
+static void test_precopy_unix(void)
+{
+    char *uri = g_strdup_printf("unix:%s/migsocket", tmpfs);
+    QTestState *from, *to;
+
+    test_migrate_start(&from, &to, uri);
+
+    /* We want to pick a speed slow enough that the test completes
+     * quickly, but that it doesn't complete precopy even on a slow
+     * machine, so also set the downtime.
+     */
+    /* 1 ms should make it not converge*/
+    migrate_set_parameter(from, "downtime-limit", "1");
+    /* 1GB/s */
+    migrate_set_parameter(from, "max-bandwidth", "1000000000");
+
+    /* Wait for the first serial output from the source */
+    wait_for_serial("src_serial");
+
+    migrate(from, uri);
+
+    wait_for_migration_pass(from);
+
+    /* 300 ms should converge */
+    migrate_set_parameter(from, "downtime-limit", "300");
+
+    if (!got_stop) {
+        qtest_qmp_eventwait(from, "STOP");
+    }
+
+    qtest_qmp_eventwait(to, "RESUME");
+
+    wait_for_serial("dest_serial");
+    wait_for_migration_complete(from);
+
+    test_migrate_end(from, to, true);
+    g_free(uri);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -603,9 +642,10 @@ int main(int argc, char **argv)
 
     module_call_init(MODULE_INIT_QOM);
 
-    qtest_add_func("/migration/postcopy/unix", test_migrate);
+    qtest_add_func("/migration/postcopy/unix", test_postcopy);
     qtest_add_func("/migration/deprecated", test_deprecated);
     qtest_add_func("/migration/bad_dest", test_baddest);
+    qtest_add_func("/migration/precopy/unix", test_precopy_unix);
 
     ret = g_test_run();
 
-- 
2.14.3

[Qemu-devel] [PATCH v10 02/24] tests: Add migration xbzrle test

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
---
 tests/migration-test.c | 64 ++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 64 insertions(+)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index 65ce3ea4ab..fb67a88353 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -501,6 +501,20 @@ static void deprecated_set_speed(QTestState *who, const char *value)
     migrate_check_parameter(who, "max-bandwidth", value);
 }
 
+static void deprecated_set_cache_size(QTestState *who, const char *value)
+{
+    QDict *rsp;
+    gchar *cmd;
+
+    cmd = g_strdup_printf("{ 'execute': 'migrate-set-cache-size',"
+                          "'arguments': { 'value': %s } }", value);
+    rsp = qtest_qmp(who, cmd);
+    g_free(cmd);
+    g_assert(qdict_haskey(rsp, "return"));
+    QDECREF(rsp);
+    migrate_check_parameter(who, "xbzrle-cache-size", value);
+}
+
 static void test_deprecated(void)
 {
     QTestState *from;
@@ -509,6 +523,7 @@ static void test_deprecated(void)
 
     deprecated_set_downtime(from, 0.12345);
     deprecated_set_speed(from, "12345");
+    deprecated_set_cache_size(from, "4096");
 
     qtest_quit(from);
 }
@@ -623,6 +638,54 @@ static void test_precopy_unix(void)
     g_free(uri);
 }
 
+static void test_xbzrle(const char *uri)
+{
+    QTestState *from, *to;
+
+    test_migrate_start(&from, &to, uri);
+
+    /* We want to pick a speed slow enough that the test completes
+     * quickly, but that it doesn't complete precopy even on a slow
+     * machine, so also set the downtime.
+     */
+    /* 1 ms should make it not converge*/
+    migrate_set_parameter(from, "downtime-limit", "1");
+    /* 1GB/s */
+    migrate_set_parameter(from, "max-bandwidth", "1000000000");
+
+    migrate_set_parameter(from, "xbzrle-cache-size", "33554432");
+
+    migrate_set_capability(from, "xbzrle", "true");
+    migrate_set_capability(to, "xbzrle", "true");
+    /* Wait for the first serial output from the source */
+    wait_for_serial("src_serial");
+
+    migrate(from, uri);
+
+    wait_for_migration_pass(from);
+
+    /* 300ms should converge */
+    migrate_set_parameter(from, "downtime-limit", "300");
+
+    if (!got_stop) {
+        qtest_qmp_eventwait(from, "STOP");
+    }
+    qtest_qmp_eventwait(to, "RESUME");
+
+    wait_for_serial("dest_serial");
+    wait_for_migration_complete(from);
+
+    test_migrate_end(from, to, true);
+}
+
+static void test_xbzrle_unix(void)
+{
+    char *uri = g_strdup_printf("unix:%s/migsocket", tmpfs);
+
+    test_xbzrle(uri);
+    g_free(uri);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -646,6 +709,7 @@ int main(int argc, char **argv)
     qtest_add_func("/migration/deprecated", test_deprecated);
     qtest_add_func("/migration/bad_dest", test_baddest);
     qtest_add_func("/migration/precopy/unix", test_precopy_unix);
+    qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
 
     ret = g_test_run();
 
-- 
2.14.3

[Qemu-devel] [PATCH v10 03/24] migration: Create tcp_port parameter

[Juan Quintela]

It will be used to store the uri tcp_port parameter.  This is the only
parameter than can change and we can need to be able to connect to it.

Signed-off-by: Juan Quintela <quintela@redhat.com>

--

This used to be uri parameter, but it has so many troubles to
reproduce that it don't just make sense.
---
 hmp.c                 |  3 +++
 migration/migration.c |  8 ++++++++
 qapi/migration.json   | 19 ++++++++++++++++---
 3 files changed, 27 insertions(+), 3 deletions(-)

diff --git a/hmp.c b/hmp.c
index 016cb5c4f1..b37605d86a 100644
--- a/hmp.c
+++ b/hmp.c
@@ -355,6 +355,9 @@ void hmp_info_migrate_parameters(Monitor *mon, const QDict *qdict)
         monitor_printf(mon, "%s: %" PRIu64 "\n",
             MigrationParameter_str(MIGRATION_PARAMETER_XBZRLE_CACHE_SIZE),
             params->xbzrle_cache_size);
+        monitor_printf(mon, "%s: %d\n",
+            MigrationParameter_str(MIGRATION_PARAMETER_X_TCP_PORT),
+            params->x_tcp_port);
     }
 
     qapi_free_MigrationParameters(params);
diff --git a/migration/migration.c b/migration/migration.c
index e345d0cc7e..31b16a335b 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -545,6 +545,8 @@ MigrationParameters *qmp_query_migrate_parameters(Error **errp)
     params->x_multifd_page_count = s->parameters.x_multifd_page_count;
     params->has_xbzrle_cache_size = true;
     params->xbzrle_cache_size = s->parameters.xbzrle_cache_size;
+    params->has_x_tcp_port = true;
+    params->x_tcp_port = s->parameters.x_tcp_port;
 
     return params;
 }
@@ -912,6 +914,9 @@ static void migrate_params_test_apply(MigrateSetParameters *params,
     if (params->has_xbzrle_cache_size) {
         dest->xbzrle_cache_size = params->xbzrle_cache_size;
     }
+    if (params->has_x_tcp_port) {
+        dest->x_tcp_port = params->x_tcp_port;
+    }
 }
 
 static void migrate_params_apply(MigrateSetParameters *params, Error **errp)
@@ -984,6 +989,9 @@ static void migrate_params_apply(MigrateSetParameters *params, Error **errp)
         s->parameters.xbzrle_cache_size = params->xbzrle_cache_size;
         xbzrle_cache_resize(params->xbzrle_cache_size, errp);
     }
+    if (params->has_x_tcp_port) {
+        s->parameters.x_tcp_port = params->x_tcp_port;
+    }
 }
 
 void qmp_migrate_set_parameters(MigrateSetParameters *params, Error **errp)
diff --git a/qapi/migration.json b/qapi/migration.json
index 7f465a1902..b6ef193f47 100644
--- a/qapi/migration.json
+++ b/qapi/migration.json
@@ -490,6 +490,9 @@
 #                     and a power of 2
 #                     (Since 2.11)
 #
+# @x-tcp-port: Only used for tcp, to know what the real port is
+#                     (Since 2.12)
+#
 # Since: 2.4
 ##
 { 'enum': 'MigrationParameter',
@@ -498,7 +501,7 @@
            'tls-creds', 'tls-hostname', 'max-bandwidth',
            'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
            'x-multifd-channels', 'x-multifd-page-count',
-           'xbzrle-cache-size' ] }
+           'xbzrle-cache-size', 'x-tcp-port' ] }
 
 ##
 # @MigrateSetParameters:
@@ -566,6 +569,10 @@
 #                     needs to be a multiple of the target page size
 #                     and a power of 2
 #                     (Since 2.11)
+#
+# @x-tcp-port: Only used for tcp, to know what the real port is
+#                     (Since 2.12)
+#
 # Since: 2.4
 ##
 # TODO either fuse back into MigrationParameters, or make
@@ -584,7 +591,8 @@
             '*block-incremental': 'bool',
             '*x-multifd-channels': 'int',
             '*x-multifd-page-count': 'int',
-            '*xbzrle-cache-size': 'size' } }
+            '*xbzrle-cache-size': 'size',
+            '*x-tcp-port': 'uint16'} }
 
 ##
 # @migrate-set-parameters:
@@ -667,6 +675,10 @@
 #                     needs to be a multiple of the target page size
 #                     and a power of 2
 #                     (Since 2.11)
+#
+# @x-tcp-port: Only used for tcp, to know what the real port is
+#                     (Since 2.12)
+#
 # Since: 2.4
 ##
 { 'struct': 'MigrationParameters',
@@ -683,7 +695,8 @@
             '*block-incremental': 'bool' ,
             '*x-multifd-channels': 'uint8',
             '*x-multifd-page-count': 'uint32',
-            '*xbzrle-cache-size': 'size' } }
+            '*xbzrle-cache-size': 'size',
+            '*x-tcp-port': 'uint16'} }
 
 ##
 # @query-migrate-parameters:
-- 
2.14.3

[Qemu-devel] [PATCH v10 04/24] migration: Set the migration tcp port

[Juan Quintela]

We can set the port parameter as zero.  This patch lets us know what
port the system was choosen for us.  Now we can migrate to this place.

Signed-off-by: Juan Quintela <quintela@redhat.com>

--

This was migrate_set_uri(), but as we only need the tcp_port, change
to that one.
---
 migration/migration.c | 10 ++++++++++
 migration/migration.h |  2 ++
 migration/socket.c    | 35 ++++++++++++++++++++++++++++++-----
 3 files changed, 42 insertions(+), 5 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index 31b16a335b..c398665de7 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -268,6 +268,16 @@ int migrate_send_rp_req_pages(MigrationIncomingState *mis, const char *rbname,
     return migrate_send_rp_message(mis, msg_type, msglen, bufc);
 }
 
+void migrate_set_port(const uint16_t port, Error **errp)
+{
+    MigrateSetParameters p = {
+        .has_x_tcp_port = true,
+        .x_tcp_port = port,
+    };
+
+    qmp_migrate_set_parameters(&p, errp);
+}
+
 void qemu_start_incoming_migration(const char *uri, Error **errp)
 {
     const char *p;
diff --git a/migration/migration.h b/migration/migration.h
index 08c5d2ded1..f40014cf94 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -234,4 +234,6 @@ void migrate_send_rp_pong(MigrationIncomingState *mis,
 int migrate_send_rp_req_pages(MigrationIncomingState *mis, const char* rbname,
                               ram_addr_t start, size_t len);
 
+void migrate_set_port(const uint16_t port, Error **errp);
+
 #endif
diff --git a/migration/socket.c b/migration/socket.c
index e090097077..08606c665d 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -15,6 +15,7 @@
  */
 
 #include "qemu/osdep.h"
+#include "qemu/cutils.h"
 
 #include "qemu-common.h"
 #include "qemu/error-report.h"
@@ -160,17 +161,24 @@ out:
 }
 
 
-static void socket_start_incoming_migration(SocketAddress *saddr,
-                                            Error **errp)
+static SocketAddress *socket_start_incoming_migration(SocketAddress *saddr,
+                                                      Error **errp)
 {
     QIOChannelSocket *listen_ioc = qio_channel_socket_new();
+    SocketAddress *address;
 
     qio_channel_set_name(QIO_CHANNEL(listen_ioc),
                          "migration-socket-listener");
 
     if (qio_channel_socket_listen_sync(listen_ioc, saddr, errp) < 0) {
         object_unref(OBJECT(listen_ioc));
-        return;
+        return NULL;
+    }
+
+    address = qio_channel_socket_get_local_address(listen_ioc, errp);
+    if (address < 0) {
+        object_unref(OBJECT(listen_ioc));
+        return NULL;
     }
 
     qio_channel_add_watch(QIO_CHANNEL(listen_ioc),
@@ -178,14 +186,28 @@ static void socket_start_incoming_migration(SocketAddress *saddr,
                           socket_accept_incoming_migration,
                           listen_ioc,
                           (GDestroyNotify)object_unref);
+    return address;
 }
 
 void tcp_start_incoming_migration(const char *host_port, Error **errp)
 {
     Error *err = NULL;
     SocketAddress *saddr = tcp_build_address(host_port, &err);
+
     if (!err) {
-        socket_start_incoming_migration(saddr, &err);
+        SocketAddress *address = socket_start_incoming_migration(saddr, &err);
+
+        if (address) {
+            unsigned long long port;
+
+            if (parse_uint_full(address->u.inet.port, &port, 10) < 0) {
+                error_setg(errp, "error parsing port in '%s'",
+                           address->u.inet.port);
+            } else {
+                migrate_set_port(port, errp);
+            }
+            qapi_free_SocketAddress(address);
+        }
     }
     qapi_free_SocketAddress(saddr);
     error_propagate(errp, err);
@@ -194,6 +216,9 @@ void tcp_start_incoming_migration(const char *host_port, Error **errp)
 void unix_start_incoming_migration(const char *path, Error **errp)
 {
     SocketAddress *saddr = unix_build_address(path);
-    socket_start_incoming_migration(saddr, errp);
+    SocketAddress *address;
+
+    address = socket_start_incoming_migration(saddr, errp);
+    qapi_free_SocketAddress(address);
     qapi_free_SocketAddress(saddr);
 }
-- 
2.14.3

[Qemu-devel] [PATCH v10 05/24] tests: Migration ppc now inlines its program

[Juan Quintela]

No need to write it to a file.  Just need a proper firmware O:-)

Signed-off-by: Juan Quintela <quintela@redhat.com>
CC: Laurent Vivier <lvivier@redhat.com>
---
 tests/migration-test.c | 41 +++++------------------------------------
 1 file changed, 5 insertions(+), 36 deletions(-)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index fb67a88353..a8db55ef83 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -19,9 +19,6 @@
 #include "qemu/sockets.h"
 #include "chardev/char.h"
 #include "sysemu/sysemu.h"
-#include "hw/nvram/chrp_nvram.h"
-
-#define MIN_NVRAM_SIZE 8192 /* from spapr_nvram.c */
 
 const unsigned start_address = 1024 * 1024;
 const unsigned end_address = 100 * 1024 * 1024;
@@ -90,36 +87,6 @@ static void init_bootfile_x86(const char *bootpath)
     fclose(bootfile);
 }
 
-static void init_bootfile_ppc(const char *bootpath)
-{
-    FILE *bootfile;
-    char buf[MIN_NVRAM_SIZE];
-    ChrpNvramPartHdr *header = (ChrpNvramPartHdr *)buf;
-
-    memset(buf, 0, MIN_NVRAM_SIZE);
-
-    /* Create a "common" partition in nvram to store boot-command property */
-
-    header->signature = CHRP_NVPART_SYSTEM;
-    memcpy(header->name, "common", 6);
-    chrp_nvram_finish_partition(header, MIN_NVRAM_SIZE);
-
-    /* FW_MAX_SIZE is 4MB, but slof.bin is only 900KB,
-     * so let's modify memory between 1MB and 100MB
-     * to do like PC bootsector
-     */
-
-    sprintf(buf + 16,
-            "boot-command=hex .\" _\" begin %x %x do i c@ 1 + i c! 1000 +loop "
-            ".\" B\" 0 until", end_address, start_address);
-
-    /* Write partition to the NVRAM file */
-
-    bootfile = fopen(bootpath, "wb");
-    g_assert_cmpint(fwrite(buf, MIN_NVRAM_SIZE, 1, bootfile), ==, 1);
-    fclose(bootfile);
-}
-
 /*
  * Wait for some output in the serial output file,
  * we get an 'A' followed by an endless string of 'B's
@@ -410,12 +377,14 @@ static void test_migrate_start(QTestState **from, QTestState **to,
         if (access("/sys/module/kvm_hv", F_OK)) {
             accel = "tcg";
         }
-        init_bootfile_ppc(bootpath);
         cmd_src = g_strdup_printf("-machine accel=%s -m 256M"
                                   " -name source,debug-threads=on"
                                   " -serial file:%s/src_serial"
-                                  " -drive file=%s,if=pflash,format=raw",
-                                  accel, tmpfs, bootpath);
+                                  " -prom-env '"
+                                  "boot-command=hex .\" _\" begin %x %x "
+                                  "do i c@ 1 + i c! 1000 +loop .\" B\" 0 "
+                                  "until'",  accel, tmpfs, end_address,
+                                  start_address);
         cmd_dst = g_strdup_printf("-machine accel=%s -m 256M"
                                   " -name target,debug-threads=on"
                                   " -serial file:%s/dest_serial"
-- 
2.14.3

[Qemu-devel] [PATCH v10 06/24] tests: Add basic migration precopy tcp test

[Juan Quintela]

Not sharing code from precopy/unix because we have to read back the
tcp parameter.

Signed-off-by: Juan Quintela <quintela@redhat.com>
Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>
Reviewed-by: Peter Xu <peterx@redhat.com>
---
 tests/migration-test.c | 58 +++++++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 55 insertions(+), 3 deletions(-)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index a8db55ef83..6b718bb5dd 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -277,8 +277,7 @@ static void cleanup(const char *filename)
     g_free(path);
 }
 
-static void migrate_check_parameter(QTestState *who, const char *parameter,
-                                    const char *value)
+static char *migrate_get_parameter(QTestState *who, const char *parameter)
 {
     QDict *rsp, *rsp_return;
     char *result;
@@ -287,9 +286,18 @@ static void migrate_check_parameter(QTestState *who, const char *parameter,
     rsp_return = qdict_get_qdict(rsp, "return");
     result = g_strdup_printf("%" PRId64,
                              qdict_get_try_int(rsp_return,  parameter, -1));
+    QDECREF(rsp);
+    return result;
+}
+
+static void migrate_check_parameter(QTestState *who, const char *parameter,
+                                    const char *value)
+{
+    char *result;
+
+    result = migrate_get_parameter(who, parameter);
     g_assert_cmpstr(result, ==, value);
     g_free(result);
-    QDECREF(rsp);
 }
 
 static void migrate_set_parameter(QTestState *who, const char *parameter,
@@ -655,6 +663,49 @@ static void test_xbzrle_unix(void)
     g_free(uri);
 }
 
+static void test_precopy_tcp(void)
+{
+    char *uri;
+    char *port;
+    QTestState *from, *to;
+
+    test_migrate_start(&from, &to, "tcp:127.0.0.1:0");
+
+    /* We want to pick a speed slow enough that the test completes
+     * quickly, but that it doesn't complete precopy even on a slow
+     * machine, so also set the downtime.
+     */
+    /* 1 ms should make it not converge*/
+    migrate_set_parameter(from, "downtime-limit", "1");
+    /* 1GB/s */
+    migrate_set_parameter(from, "max-bandwidth", "1000000000");
+
+    /* Wait for the first serial output from the source */
+    wait_for_serial("src_serial");
+
+    port = migrate_get_parameter(to, "x-tcp-port");
+    uri = g_strdup_printf("tcp:127.0.0.1:%s", port);
+
+    migrate(from, uri);
+
+    wait_for_migration_pass(from);
+
+    /* 300ms should converge */
+    migrate_set_parameter(from, "downtime-limit", "300");
+
+    if (!got_stop) {
+        qtest_qmp_eventwait(from, "STOP");
+    }
+    qtest_qmp_eventwait(to, "RESUME");
+
+    wait_for_serial("dest_serial");
+    wait_for_migration_complete(from);
+
+    test_migrate_end(from, to, true);
+    g_free(uri);
+    g_free(port);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -678,6 +729,7 @@ int main(int argc, char **argv)
     qtest_add_func("/migration/deprecated", test_deprecated);
     qtest_add_func("/migration/bad_dest", test_baddest);
     qtest_add_func("/migration/precopy/unix", test_precopy_unix);
+    qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
 
     ret = g_test_run();
-- 
2.14.3

[Qemu-devel] [PATCH v10 07/24] [RFH] tests: Add migration compress threads tests

[Juan Quintela]

Yeap, it is still not working. trying to learn how to debug threads
for guests running from the testt hardness.

For some reason, compression is not working at the moment, test is
disabled until I found why.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 tests/migration-test.c | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 52 insertions(+)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index 6b718bb5dd..6f9b4c8d7a 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -706,6 +706,55 @@ static void test_precopy_tcp(void)
     g_free(port);
 }
 
+static void test_compress(const char *uri)
+{
+    QTestState *from, *to;
+
+    test_migrate_start(&from, &to, uri);
+
+    /* We want to pick a speed slow enough that the test completes
+     * quickly, but that it doesn't complete precopy even on a slow
+     * machine, so also set the downtime.
+     */
+    /* 1 ms should make it not converge*/
+    migrate_set_parameter(from, "downtime-limit", "1");
+    /* 1GB/s */
+    migrate_set_parameter(from, "max-bandwidth", "1000000000");
+
+    migrate_set_parameter(from, "compress-threads", "4");
+    migrate_set_parameter(to, "decompress-threads", "3");
+
+    migrate_set_capability(from, "compress", "true");
+    migrate_set_capability(to, "compress", "true");
+    /* Wait for the first serial output from the source */
+    wait_for_serial("src_serial");
+
+    migrate(from, uri);
+
+    wait_for_migration_pass(from);
+
+    /* 300ms should converge */
+    migrate_set_parameter(from, "downtime-limit", "300");
+
+    if (!got_stop) {
+        qtest_qmp_eventwait(from, "STOP");
+    }
+    qtest_qmp_eventwait(to, "RESUME");
+
+    wait_for_serial("dest_serial");
+    wait_for_migration_complete(from);
+
+    test_migrate_end(from, to, true);
+}
+
+static void test_compress_unix(void)
+{
+    char *uri = g_strdup_printf("unix:%s/migsocket", tmpfs);
+
+    test_compress(uri);
+    g_free(uri);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -731,6 +780,9 @@ int main(int argc, char **argv)
     qtest_add_func("/migration/precopy/unix", test_precopy_unix);
     qtest_add_func("/migration/precopy/tcp", test_precopy_tcp);
     qtest_add_func("/migration/xbzrle/unix", test_xbzrle_unix);
+    if (0) {
+        qtest_add_func("/migration/compress/unix", test_compress_unix);
+    }
 
     ret = g_test_run();
 
-- 
2.14.3

[Qemu-devel] [PATCH v10 08/24] migration: Add multifd test

[Juan Quintela]

We set the x-multifd-page-count and x-multifd-channels.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 tests/migration-test.c | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 50 insertions(+)

diff --git a/tests/migration-test.c b/tests/migration-test.c
index 6f9b4c8d7a..97d35f979d 100644
--- a/tests/migration-test.c
+++ b/tests/migration-test.c
@@ -755,6 +755,55 @@ static void test_compress_unix(void)
     g_free(uri);
 }
 
+static void test_multifd_tcp(void)
+{
+    char *uri;
+    char *port;
+    QTestState *from, *to;
+
+    test_migrate_start(&from, &to, "tcp:127.0.0.1:0");
+
+    /* We want to pick a speed slow enough that the test completes
+     * quickly, but that it doesn't complete precopy even on a slow
+     * machine, so also set the downtime.
+     */
+    /* 1 ms should make it not converge*/
+    migrate_set_parameter(from, "downtime-limit", "1");
+    /* 1GB/s */
+    migrate_set_parameter(from, "max-bandwidth", "1000000000");
+
+    migrate_set_parameter(from, "x-multifd-channels", "4");
+    migrate_set_parameter(to, "x-multifd-channels", "4");
+
+    migrate_set_parameter(from, "x-multifd-page-count", "64");
+    migrate_set_parameter(to, "x-multifd-page-count", "64");
+
+    migrate_set_capability(from, "x-multifd", "true");
+    migrate_set_capability(to, "x-multifd", "true");
+    /* Wait for the first serial output from the source */
+    wait_for_serial("src_serial");
+
+    port = migrate_get_parameter(to, "x-tcp-port");
+    uri = g_strdup_printf("tcp:127.0.0.1:%s", port);
+
+    migrate(from, uri);
+
+    wait_for_migration_pass(from);
+
+    /* 300ms it should converge */
+    migrate_set_parameter(from, "downtime-limit", "300");
+
+    if (!got_stop) {
+        qtest_qmp_eventwait(from, "STOP");
+    }
+    qtest_qmp_eventwait(to, "RESUME");
+
+    wait_for_serial("dest_serial");
+    wait_for_migration_complete(from);
+
+    test_migrate_end(from, to, true);
+}
+
 int main(int argc, char **argv)
 {
     char template[] = "/tmp/migration-test-XXXXXX";
@@ -783,6 +832,7 @@ int main(int argc, char **argv)
     if (0) {
         qtest_add_func("/migration/compress/unix", test_compress_unix);
     }
+    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
 
     ret = g_test_run();
 
-- 
2.14.3

[Qemu-devel] [PATCH v10 09/24] migration: Set error state in case of error

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/migration/ram.c b/migration/ram.c
index 3b6c077964..4a56a85d53 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -414,6 +414,16 @@ static void terminate_multifd_send_threads(Error *errp)
 {
     int i;
 
+    if (errp) {
+        MigrationState *s = migrate_get_current();
+        migrate_set_error(s, errp);
+        if (s->state == MIGRATION_STATUS_SETUP ||
+            s->state == MIGRATION_STATUS_ACTIVE) {
+            migrate_set_state(&s->state, s->state,
+                              MIGRATION_STATUS_FAILED);
+        }
+    }
+
     for (i = 0; i < multifd_send_state->count; i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
 
@@ -514,6 +524,16 @@ static void terminate_multifd_recv_threads(Error *errp)
 {
     int i;
 
+    if (errp) {
+        MigrationState *s = migrate_get_current();
+        migrate_set_error(s, errp);
+        if (s->state == MIGRATION_STATUS_SETUP ||
+            s->state == MIGRATION_STATUS_ACTIVE) {
+            migrate_set_state(&s->state, s->state,
+                              MIGRATION_STATUS_FAILED);
+        }
+    }
+
     for (i = 0; i < multifd_recv_state->count; i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 
-- 
2.14.3

[Qemu-devel] [PATCH v10 10/24] migration: In case of error just end the migration

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/socket.c | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/migration/socket.c b/migration/socket.c
index 08606c665d..b12b0a462e 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -139,9 +139,8 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
     sioc = qio_channel_socket_accept(QIO_CHANNEL_SOCKET(ioc),
                                      &err);
     if (!sioc) {
-        error_report("could not accept migration connection (%s)",
-                     error_get_pretty(err));
-        goto out;
+        migrate_set_error(migrate_get_current(), err);
+        return G_SOURCE_REMOVE;
     }
 
     trace_migration_socket_incoming_accepted();
@@ -150,7 +149,6 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
     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);
-- 
2.14.3

[Qemu-devel] [PATCH v10 11/24] migration: terminate_* can be called for other threads

[Juan Quintela]

Once there, make  count field to always be accessed with atomic
operations.  To make blocking operations, we need to know that the
thread is running, so create a bool to indicate that.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 30 ++++++++++++++++++++----------
 1 file changed, 20 insertions(+), 10 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index 4a56a85d53..977e675f46 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -400,6 +400,7 @@ struct MultiFDSendParams {
     QemuThread thread;
     QemuSemaphore sem;
     QemuMutex mutex;
+    bool running;
     bool quit;
 };
 typedef struct MultiFDSendParams MultiFDSendParams;
@@ -424,7 +425,7 @@ static void terminate_multifd_send_threads(Error *errp)
         }
     }
 
-    for (i = 0; i < multifd_send_state->count; i++) {
+    for (i = 0; i < migrate_multifd_channels(); i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
 
         qemu_mutex_lock(&p->mutex);
@@ -443,10 +444,13 @@ int multifd_save_cleanup(Error **errp)
         return 0;
     }
     terminate_multifd_send_threads(NULL);
-    for (i = 0; i < multifd_send_state->count; i++) {
+    for (i = 0; i < migrate_multifd_channels(); i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
 
-        qemu_thread_join(&p->thread);
+        if (p->running) {
+            qemu_thread_join(&p->thread);
+            p->running = false;
+        }
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
@@ -487,7 +491,7 @@ int multifd_save_setup(void)
     thread_count = migrate_multifd_channels();
     multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
-    multifd_send_state->count = 0;
+    atomic_set(&multifd_send_state->count, 0);
     for (i = 0; i < thread_count; i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
 
@@ -496,10 +500,11 @@ int multifd_save_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdsend_%d", i);
+        p->running = true;
         qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
                            QEMU_THREAD_JOINABLE);
 
-        multifd_send_state->count++;
+        atomic_inc(&multifd_send_state->count);
     }
     return 0;
 }
@@ -510,6 +515,7 @@ struct MultiFDRecvParams {
     QemuThread thread;
     QemuSemaphore sem;
     QemuMutex mutex;
+    bool running;
     bool quit;
 };
 typedef struct MultiFDRecvParams MultiFDRecvParams;
@@ -534,7 +540,7 @@ static void terminate_multifd_recv_threads(Error *errp)
         }
     }
 
-    for (i = 0; i < multifd_recv_state->count; i++) {
+    for (i = 0; i < migrate_multifd_channels(); i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 
         qemu_mutex_lock(&p->mutex);
@@ -553,10 +559,13 @@ int multifd_load_cleanup(Error **errp)
         return 0;
     }
     terminate_multifd_recv_threads(NULL);
-    for (i = 0; i < multifd_recv_state->count; i++) {
+    for (i = 0; i < migrate_multifd_channels(); i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 
-        qemu_thread_join(&p->thread);
+        if (p->running) {
+            qemu_thread_join(&p->thread);
+            p->running = false;
+        }
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
@@ -598,7 +607,7 @@ int multifd_load_setup(void)
     thread_count = migrate_multifd_channels();
     multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
-    multifd_recv_state->count = 0;
+    atomic_set(&multifd_recv_state->count, 0);
     for (i = 0; i < thread_count; i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 
@@ -607,9 +616,10 @@ int multifd_load_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdrecv_%d", i);
+        p->running = true;
         qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
                            QEMU_THREAD_JOINABLE);
-        multifd_recv_state->count++;
+        atomic_inc(&multifd_recv_state->count);
     }
     return 0;
 }
-- 
2.14.3

[Qemu-devel] [PATCH v10 12/24] migration: Reference counting recv channels correctly

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/socket.c | 11 +++++++++++
 migration/socket.h |  7 +++++++
 2 files changed, 18 insertions(+)

diff --git a/migration/socket.c b/migration/socket.c
index b12b0a462e..26110739cf 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -27,6 +27,17 @@
 #include "io/channel-socket.h"
 #include "trace.h"
 
+int socket_recv_channel_ref(QIOChannel *recv)
+{
+    object_ref(OBJECT(recv));
+    return 0;
+}
+
+int socket_recv_channel_unref(QIOChannel *recv)
+{
+    object_unref(OBJECT(recv));
+    return 0;
+}
 
 static SocketAddress *tcp_build_address(const char *host_port, Error **errp)
 {
diff --git a/migration/socket.h b/migration/socket.h
index 6b91e9db38..638a85255a 100644
--- a/migration/socket.h
+++ b/migration/socket.h
@@ -16,6 +16,13 @@
 
 #ifndef QEMU_MIGRATION_SOCKET_H
 #define QEMU_MIGRATION_SOCKET_H
+
+#include "io/channel.h"
+#include "io/task.h"
+
+int socket_recv_channel_ref(QIOChannel *recv);
+int socket_recv_channel_unref(QIOChannel *recv);
+
 void tcp_start_incoming_migration(const char *host_port, Error **errp);
 
 void tcp_start_outgoing_migration(MigrationState *s, const char *host_port,
-- 
2.14.3

[Qemu-devel] [PATCH v10 13/24] migration: Introduce multifd_recv_new_channel()

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/migration.c | 3 ++-
 migration/ram.c       | 6 ++++++
 migration/ram.h       | 2 ++
 3 files changed, 10 insertions(+), 1 deletion(-)

diff --git a/migration/migration.c b/migration/migration.c
index c398665de7..919343232e 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -448,8 +448,9 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
     if (!mis->from_src_file) {
         QEMUFile *f = qemu_fopen_channel_input(ioc);
         migration_fd_process_incoming(f);
+        return;
     }
-    /* We still only have a single channel.  Nothing to do here yet */
+    multifd_recv_new_channel(ioc);
 }
 
 /**
diff --git a/migration/ram.c b/migration/ram.c
index 977e675f46..f48c74585f 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"
@@ -624,6 +625,11 @@ int multifd_load_setup(void)
     return 0;
 }
 
+void multifd_recv_new_channel(QIOChannel *ioc)
+{
+    /* nothing to do yet */
+}
+
 /**
  * save_page_header: write page header to wire
  *
diff --git a/migration/ram.h b/migration/ram.h
index 53f0021c51..a2031acf59 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -32,6 +32,7 @@
 #include "qemu-common.h"
 #include "qapi/qapi-types-migration.h"
 #include "exec/cpu-common.h"
+#include "io/channel.h"
 
 extern MigrationStats ram_counters;
 extern XBZRLECacheStats xbzrle_counters;
@@ -44,6 +45,7 @@ int multifd_save_setup(void);
 int multifd_save_cleanup(Error **errp);
 int multifd_load_setup(void);
 int multifd_load_cleanup(Error **errp);
+void multifd_recv_new_channel(QIOChannel *ioc);
 
 uint64_t ram_pagesize_summary(void);
 int ram_save_queue_pages(const char *rbname, ram_addr_t start, ram_addr_t len);
-- 
2.14.3

[Qemu-devel] [PATCH v10 14/24] migration: Be sure all recv channels are created

[Juan Quintela]

We need them before we start migration.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/migration.c |  6 +++++-
 migration/ram.c       | 11 +++++++++++
 migration/ram.h       |  1 +
 3 files changed, 17 insertions(+), 1 deletion(-)

diff --git a/migration/migration.c b/migration/migration.c
index 919343232e..c06c34ca0f 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -461,7 +461,11 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
  */
 bool migration_has_all_channels(void)
 {
-    return true;
+    bool all_channels;
+
+    all_channels = multifd_recv_all_channels_created();
+
+    return all_channels;
 }
 
 /*
diff --git a/migration/ram.c b/migration/ram.c
index f48c74585f..e502be5dda 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -625,6 +625,17 @@ int multifd_load_setup(void)
     return 0;
 }
 
+bool multifd_recv_all_channels_created(void)
+{
+    int thread_count = migrate_multifd_channels();
+
+    if (!migrate_use_multifd()) {
+        return true;
+    }
+
+    return thread_count == atomic_read(&multifd_recv_state->count);
+}
+
 void multifd_recv_new_channel(QIOChannel *ioc)
 {
     /* nothing to do yet */
diff --git a/migration/ram.h b/migration/ram.h
index a2031acf59..3daf074bcc 100644
--- a/migration/ram.h
+++ b/migration/ram.h
@@ -45,6 +45,7 @@ int multifd_save_setup(void);
 int multifd_save_cleanup(Error **errp);
 int multifd_load_setup(void);
 int multifd_load_cleanup(Error **errp);
+bool multifd_recv_all_channels_created(void);
 void multifd_recv_new_channel(QIOChannel *ioc);
 
 uint64_t ram_pagesize_summary(void);
-- 
2.14.3

[Qemu-devel] [PATCH v10 15/24] migration: Synchronize send threads

[Juan Quintela]

We synchronize all threads each RAM_SAVE_FLAG_EOS.  Bitmap
synchronizations don't happen inside a  ram section, so we are safe
about two channels trying to overwrite the same memory.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c        | 38 +++++++++++++++++++++++++++++++++++++-
 migration/trace-events |  1 +
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git a/migration/ram.c b/migration/ram.c
index e502be5dda..153c7560cb 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -403,6 +403,7 @@ struct MultiFDSendParams {
     QemuMutex mutex;
     bool running;
     bool quit;
+    bool sync;
 };
 typedef struct MultiFDSendParams MultiFDSendParams;
 
@@ -410,6 +411,8 @@ struct {
     MultiFDSendParams *params;
     /* number of created threads */
     int count;
+    /* syncs main thread and channels */
+    QemuSemaphore sem_main;
 } *multifd_send_state;
 
 static void terminate_multifd_send_threads(Error *errp)
@@ -457,6 +460,7 @@ int multifd_save_cleanup(Error **errp)
         g_free(p->name);
         p->name = NULL;
     }
+    qemu_sem_destroy(&multifd_send_state->sem_main);
     g_free(multifd_send_state->params);
     multifd_send_state->params = NULL;
     g_free(multifd_send_state);
@@ -464,18 +468,44 @@ int multifd_save_cleanup(Error **errp)
     return ret;
 }
 
+static void multifd_send_sync_main(void)
+{
+    int i;
+
+    if (!migrate_use_multifd()) {
+        return;
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+        qemu_mutex_lock(&p->mutex);
+        p->sync = true;
+        qemu_mutex_unlock(&p->mutex);
+        qemu_sem_post(&p->sem);
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        qemu_sem_wait(&multifd_send_state->sem_main);
+    }
+    trace_multifd_send_sync_main();
+}
+
 static void *multifd_send_thread(void *opaque)
 {
     MultiFDSendParams *p = opaque;
 
     while (true) {
+        qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
+        if (p->sync) {
+            p->sync = false;
+            qemu_mutex_unlock(&p->mutex);
+            qemu_sem_post(&multifd_send_state->sem_main);
+            continue;
+        }
         if (p->quit) {
             qemu_mutex_unlock(&p->mutex);
             break;
         }
         qemu_mutex_unlock(&p->mutex);
-        qemu_sem_wait(&p->sem);
     }
 
     return NULL;
@@ -493,6 +523,8 @@ int multifd_save_setup(void)
     multifd_send_state = g_malloc0(sizeof(*multifd_send_state));
     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
     atomic_set(&multifd_send_state->count, 0);
+    qemu_sem_init(&multifd_send_state->sem_main, 0);
+
     for (i = 0; i < thread_count; i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
 
@@ -507,6 +539,7 @@ int multifd_save_setup(void)
 
         atomic_inc(&multifd_send_state->count);
     }
+
     return 0;
 }
 
@@ -2283,6 +2316,7 @@ static int ram_save_setup(QEMUFile *f, void *opaque)
     ram_control_before_iterate(f, RAM_CONTROL_SETUP);
     ram_control_after_iterate(f, RAM_CONTROL_SETUP);
 
+    multifd_send_sync_main();
     qemu_put_be64(f, RAM_SAVE_FLAG_EOS);
 
     return 0;
@@ -2351,6 +2385,7 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
      */
     ram_control_after_iterate(f, RAM_CONTROL_ROUND);
 
+    multifd_send_sync_main();
     qemu_put_be64(f, RAM_SAVE_FLAG_EOS);
     ram_counters.transferred += 8;
 
@@ -2403,6 +2438,7 @@ static int ram_save_complete(QEMUFile *f, void *opaque)
 
     rcu_read_unlock();
 
+    multifd_send_sync_main();
     qemu_put_be64(f, RAM_SAVE_FLAG_EOS);
 
     return 0;
diff --git a/migration/trace-events b/migration/trace-events
index 93961dea16..97b5ac564f 100644
--- a/migration/trace-events
+++ b/migration/trace-events
@@ -77,6 +77,7 @@ ram_load_postcopy_loop(uint64_t addr, int flags) "@%" PRIx64 " %x"
 ram_postcopy_send_discard_bitmap(void) ""
 ram_save_page(const char *rbname, uint64_t offset, void *host) "%s: offset: 0x%" PRIx64 " host: %p"
 ram_save_queue_pages(const char *rbname, size_t start, size_t len) "%s: start: 0x%zx len: 0x%zx"
+multifd_send_sync_main(void) ""
 
 # migration/migration.c
 await_return_path_close_on_source_close(void) ""
-- 
2.14.3

[Qemu-devel] [PATCH v10 16/24] migration: Synchronize recv threads

[Juan Quintela]

We synchronize all threads each RAM_SAVE_FLAG_EOS.  Bitmap
synchronizations don't happen inside a  ram section, so we are safe
about two channels trying to overwrite the same memory.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c        | 37 ++++++++++++++++++++++++++++++++++++-
 migration/trace-events |  1 +
 2 files changed, 37 insertions(+), 1 deletion(-)

diff --git a/migration/ram.c b/migration/ram.c
index 153c7560cb..0266bd200c 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -551,6 +551,7 @@ struct MultiFDRecvParams {
     QemuMutex mutex;
     bool running;
     bool quit;
+    bool sync;
 };
 typedef struct MultiFDRecvParams MultiFDRecvParams;
 
@@ -558,6 +559,8 @@ struct {
     MultiFDRecvParams *params;
     /* number of created threads */
     int count;
+    /* syncs main thread and channels */
+    QemuSemaphore sem_main;
 } *multifd_recv_state;
 
 static void terminate_multifd_recv_threads(Error *errp)
@@ -605,6 +608,7 @@ int multifd_load_cleanup(Error **errp)
         g_free(p->name);
         p->name = NULL;
     }
+    qemu_sem_destroy(&multifd_recv_state->sem_main);
     g_free(multifd_recv_state->params);
     multifd_recv_state->params = NULL;
     g_free(multifd_recv_state);
@@ -613,18 +617,45 @@ int multifd_load_cleanup(Error **errp)
     return ret;
 }
 
+static void multifd_recv_sync_main(void)
+{
+    int i;
+
+    if (!migrate_use_multifd()) {
+        return;
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+
+        qemu_mutex_lock(&p->mutex);
+        p->sync = true;
+        qemu_mutex_unlock(&p->mutex);
+        qemu_sem_post(&p->sem);
+    }
+    for (i = 0; i < migrate_multifd_channels(); i++) {
+        qemu_sem_wait(&multifd_recv_state->sem_main);
+    }
+    trace_multifd_recv_sync_main();
+}
+
 static void *multifd_recv_thread(void *opaque)
 {
     MultiFDRecvParams *p = opaque;
 
     while (true) {
+        qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
+        if (p->sync) {
+            p->sync = false;
+            qemu_mutex_unlock(&p->mutex);
+            qemu_sem_post(&multifd_recv_state->sem_main);
+            continue;
+        }
         if (p->quit) {
             qemu_mutex_unlock(&p->mutex);
             break;
         }
         qemu_mutex_unlock(&p->mutex);
-        qemu_sem_wait(&p->sem);
     }
 
     return NULL;
@@ -642,6 +673,7 @@ int multifd_load_setup(void)
     multifd_recv_state = g_malloc0(sizeof(*multifd_recv_state));
     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
     atomic_set(&multifd_recv_state->count, 0);
+    qemu_sem_init(&multifd_recv_state->sem_main, 0);
     for (i = 0; i < thread_count; i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 
@@ -655,6 +687,7 @@ int multifd_load_setup(void)
                            QEMU_THREAD_JOINABLE);
         atomic_inc(&multifd_recv_state->count);
     }
+
     return 0;
 }
 
@@ -2868,6 +2901,7 @@ static int ram_load_postcopy(QEMUFile *f)
             break;
         case RAM_SAVE_FLAG_EOS:
             /* normal exit */
+            multifd_recv_sync_main();
             break;
         default:
             error_report("Unknown combination of migration flags: %#x"
@@ -3053,6 +3087,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
             break;
         case RAM_SAVE_FLAG_EOS:
             /* normal exit */
+            multifd_recv_sync_main();
             break;
         default:
             if (flags & RAM_SAVE_FLAG_HOOK) {
diff --git a/migration/trace-events b/migration/trace-events
index 97b5ac564f..76075c26bc 100644
--- a/migration/trace-events
+++ b/migration/trace-events
@@ -78,6 +78,7 @@ ram_postcopy_send_discard_bitmap(void) ""
 ram_save_page(const char *rbname, uint64_t offset, void *host) "%s: offset: 0x%" PRIx64 " host: %p"
 ram_save_queue_pages(const char *rbname, size_t start, size_t len) "%s: start: 0x%zx len: 0x%zx"
 multifd_send_sync_main(void) ""
+multifd_recv_sync_main(void) ""
 
 # migration/migration.c
 await_return_path_close_on_source_close(void) ""
-- 
2.14.3

[Qemu-devel] [PATCH v10 17/24] migration: Export functions to create send channels

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/socket.c | 28 +++++++++++++++++++++++++++-
 migration/socket.h |  3 +++
 2 files changed, 30 insertions(+), 1 deletion(-)

diff --git a/migration/socket.c b/migration/socket.c
index 26110739cf..b3b5571ebb 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -39,6 +39,28 @@ int socket_recv_channel_unref(QIOChannel *recv)
     return 0;
 }
 
+struct SocketOutgoingArgs {
+    SocketAddress *saddr;
+} outgoing_args;
+
+void socket_send_channel_create(void (*f)(QIOTask *, gpointer), void *data)
+{
+    QIOChannelSocket *sioc = qio_channel_socket_new();
+    qio_channel_socket_connect_async(sioc, outgoing_args.saddr,
+                                     f, data, NULL);
+}
+
+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)
 {
     SocketAddress *saddr;
@@ -106,6 +128,11 @@ static void socket_start_outgoing_migration(MigrationState *s,
     struct SocketConnectData *data = g_new0(struct SocketConnectData, 1);
 
     data->s = s;
+
+    /* in case previous migration leaked it */
+    qapi_free_SocketAddress(outgoing_args.saddr);
+    outgoing_args.saddr = saddr;
+
     if (saddr->type == SOCKET_ADDRESS_TYPE_INET) {
         data->hostname = g_strdup(saddr->u.inet.host);
     }
@@ -116,7 +143,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 638a85255a..cbdb8d64c3 100644
--- a/migration/socket.h
+++ b/migration/socket.h
@@ -23,6 +23,9 @@
 int socket_recv_channel_ref(QIOChannel *recv);
 int socket_recv_channel_unref(QIOChannel *recv);
 
+void socket_send_channel_create(void (*f)(QIOTask *, gpointer), void *data);
+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.14.3

[Qemu-devel] [PATCH v10 18/24] migration: Add multifd traces for start/end thread

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c        | 6 ++++++
 migration/trace-events | 4 ++++
 2 files changed, 10 insertions(+)

diff --git a/migration/ram.c b/migration/ram.c
index 0266bd200c..b57d9fd667 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -492,6 +492,8 @@ static void *multifd_send_thread(void *opaque)
 {
     MultiFDSendParams *p = opaque;
 
+    trace_multifd_send_thread_start(p->id);
+
     while (true) {
         qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
@@ -507,6 +509,7 @@ static void *multifd_send_thread(void *opaque)
         }
         qemu_mutex_unlock(&p->mutex);
     }
+    trace_multifd_send_thread_end(p->id);
 
     return NULL;
 }
@@ -642,6 +645,8 @@ static void *multifd_recv_thread(void *opaque)
 {
     MultiFDRecvParams *p = opaque;
 
+    trace_multifd_recv_thread_start(p->id);
+
     while (true) {
         qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
@@ -658,6 +663,7 @@ static void *multifd_recv_thread(void *opaque)
         qemu_mutex_unlock(&p->mutex);
     }
 
+    trace_multifd_recv_thread_end(p->id);
     return NULL;
 }
 
diff --git a/migration/trace-events b/migration/trace-events
index 76075c26bc..db88fa699f 100644
--- a/migration/trace-events
+++ b/migration/trace-events
@@ -79,6 +79,10 @@ ram_save_page(const char *rbname, uint64_t offset, void *host) "%s: offset: 0x%"
 ram_save_queue_pages(const char *rbname, size_t start, size_t len) "%s: start: 0x%zx len: 0x%zx"
 multifd_send_sync_main(void) ""
 multifd_recv_sync_main(void) ""
+multifd_send_thread_start(int id) "%d"
+multifd_send_thread_end(int id) "%d"
+multifd_recv_thread_start(int id) "%d"
+multifd_recv_thread_end(int id) "%d"
 
 # migration/migration.c
 await_return_path_close_on_source_close(void) ""
-- 
2.14.3

[Qemu-devel] [PATCH v10 19/24] migration: Create multifd channels

[Juan Quintela]

In both sides.  We still don't transmit anything through them.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 42 insertions(+), 10 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index b57d9fd667..7ef0c2b7e2 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -399,6 +399,7 @@ struct MultiFDSendParams {
     uint8_t id;
     char *name;
     QemuThread thread;
+    QIOChannel *c;
     QemuSemaphore sem;
     QemuMutex mutex;
     bool running;
@@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
             qemu_thread_join(&p->thread);
             p->running = false;
         }
+        socket_send_channel_destroy(p->c);
+        p->c = NULL;
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
@@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
     return NULL;
 }
 
+static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
+{
+    MultiFDSendParams *p = opaque;
+    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
+    Error *local_err = NULL;
+
+    if (qio_task_propagate_error(task, &local_err)) {
+        if (multifd_save_cleanup(&local_err) != 0) {
+            migrate_set_error(migrate_get_current(), local_err);
+        }
+    } else {
+        p->c = QIO_CHANNEL(sioc);
+        qio_channel_set_delay(p->c, false);
+        p->running = true;
+        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
+                           QEMU_THREAD_JOINABLE);
+
+        atomic_inc(&multifd_send_state->count);
+    }
+}
+
 int multifd_save_setup(void)
 {
     int thread_count;
@@ -536,11 +560,7 @@ int multifd_save_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdsend_%d", i);
-        p->running = true;
-        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
-                           QEMU_THREAD_JOINABLE);
-
-        atomic_inc(&multifd_send_state->count);
+        socket_send_channel_create(multifd_new_send_channel_async, p);
     }
 
     return 0;
@@ -550,6 +570,7 @@ struct MultiFDRecvParams {
     uint8_t id;
     char *name;
     QemuThread thread;
+    QIOChannel *c;
     QemuSemaphore sem;
     QemuMutex mutex;
     bool running;
@@ -606,6 +627,8 @@ int multifd_load_cleanup(Error **errp)
             qemu_thread_join(&p->thread);
             p->running = false;
         }
+        socket_recv_channel_unref(p->c);
+        p->c = NULL;
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
@@ -688,10 +711,6 @@ int multifd_load_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdrecv_%d", i);
-        p->running = true;
-        qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
-                           QEMU_THREAD_JOINABLE);
-        atomic_inc(&multifd_recv_state->count);
     }
 
     return 0;
@@ -710,7 +729,20 @@ bool multifd_recv_all_channels_created(void)
 
 void multifd_recv_new_channel(QIOChannel *ioc)
 {
-    /* nothing to do yet */
+    MultiFDRecvParams *p;
+    /* we need to invent channels id's until we transmit */
+    /* we will remove this on a later patch */
+    static int i = 0;
+
+    p = &multifd_recv_state->params[i];
+    i++;
+    p->c = ioc;
+    socket_recv_channel_ref(ioc);
+
+    p->running = true;
+    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
+                       QEMU_THREAD_JOINABLE);
+    atomic_inc(&multifd_recv_state->count);
 }
 
 /**
-- 
2.14.3

[Qemu-devel] [PATCH v10 20/24] migration: Delay start of migration main routines

[Juan Quintela]

We need to make sure that we have started all the multifd threads.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/migration.c | 4 ++--
 migration/migration.h | 1 +
 migration/ram.c       | 3 +++
 migration/socket.c    | 3 +++
 4 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/migration/migration.c b/migration/migration.c
index c06c34ca0f..a355618220 100644
--- a/migration/migration.c
+++ b/migration/migration.c
@@ -429,7 +429,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);
@@ -447,7 +447,7 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
 
     if (!mis->from_src_file) {
         QEMUFile *f = qemu_fopen_channel_input(ioc);
-        migration_fd_process_incoming(f);
+        migration_incoming_setup(f);
         return;
     }
     multifd_recv_new_channel(ioc);
diff --git a/migration/migration.h b/migration/migration.h
index f40014cf94..03a940831d 100644
--- a/migration/migration.h
+++ b/migration/migration.h
@@ -184,6 +184,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 7ef0c2b7e2..1aab96bd5e 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -743,6 +743,9 @@ void multifd_recv_new_channel(QIOChannel *ioc)
     qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
                        QEMU_THREAD_JOINABLE);
     atomic_inc(&multifd_recv_state->count);
+    if (multifd_recv_state->count == migrate_multifd_channels()) {
+        migration_incoming_process();
+    }
 }
 
 /**
diff --git a/migration/socket.c b/migration/socket.c
index b3b5571ebb..deda193de7 100644
--- a/migration/socket.c
+++ b/migration/socket.c
@@ -189,6 +189,9 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
     if (migration_has_all_channels()) {
         /* Close listening socket as its no longer needed */
         qio_channel_close(ioc, NULL);
+        if (!migrate_use_multifd()) {
+            migration_incoming_process();
+        }
         return G_SOURCE_REMOVE;
     } else {
         return G_SOURCE_CONTINUE;
-- 
2.14.3

[Qemu-devel] [PATCH v10 21/24] migration: Transmit initial package through the multifd channels

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 50 +++++++++++++++++++++++++++++++++++++++++++++-----
 1 file changed, 45 insertions(+), 5 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index 1aab96bd5e..4efac0c20c 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -52,6 +52,8 @@
 #include "qemu/rcu_queue.h"
 #include "migration/colo.h"
 #include "migration/block.h"
+#include "sysemu/sysemu.h"
+#include "qemu/uuid.h"
 
 /***********************************************************/
 /* ram save/restore */
@@ -491,12 +493,30 @@ static void multifd_send_sync_main(void)
     trace_multifd_send_sync_main();
 }
 
+typedef struct {
+    uint32_t version;
+    unsigned char uuid[16]; /* QemuUUID */
+    uint8_t id;
+} __attribute__((packed)) MultiFDInit_t;
+
 static void *multifd_send_thread(void *opaque)
 {
     MultiFDSendParams *p = opaque;
+    MultiFDInit_t msg;
+    Error *local_err = NULL;
+    size_t ret;
 
     trace_multifd_send_thread_start(p->id);
 
+    msg.version = 1;
+    msg.id = p->id;
+    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
+    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), &local_err);
+    if (ret != 0) {
+        terminate_multifd_send_threads(local_err);
+        return NULL;
+    }
+
     while (true) {
         qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
@@ -730,12 +750,32 @@ bool multifd_recv_all_channels_created(void)
 void multifd_recv_new_channel(QIOChannel *ioc)
 {
     MultiFDRecvParams *p;
-    /* we need to invent channels id's until we transmit */
-    /* we will remove this on a later patch */
-    static int i = 0;
+    MultiFDInit_t msg;
+    Error *local_err = NULL;
+    size_t ret;
 
-    p = &multifd_recv_state->params[i];
-    i++;
+    ret = qio_channel_read_all(ioc, (char *)&msg, sizeof(msg), &local_err);
+    if (ret != 0) {
+        terminate_multifd_recv_threads(local_err);
+        return;
+    }
+
+    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
+        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
+        error_setg(&local_err, "multifd: received uuid '%s' and expected "
+                   "uuid '%s' for channel %hhd", msg.uuid, uuid, msg.id);
+        g_free(uuid);
+        terminate_multifd_recv_threads(local_err);
+        return;
+    }
+
+    p = &multifd_recv_state->params[msg.id];
+    if (p->c != NULL) {
+        error_setg(&local_err, "multifd: received id '%d' already setup'",
+                   msg.id);
+        terminate_multifd_recv_threads(local_err);
+        return;
+    }
     p->c = ioc;
     socket_recv_channel_ref(ioc);
 
-- 
2.14.3

[Qemu-devel] [PATCH v10 22/24] migration: Create ram_multifd_page

[Juan Quintela]

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.

Signed-off-by: Juan Quintela <quintela@redhat.com>

--
Add last_page parameter
Add commets for done and address
Remove multifd field, it is the same than normal pages
Merge next patch, now we send multiple pages at a time
Remove counter for multifd pages, it is identical to normal pages
Use iovec's instead of creating the equivalent.
Clear memory used by pages (dave)
Use g_new0(danp)
define MULTIFD_CONTINUE
now pages member is a pointer
Fix off-by-one in number of pages in one packet
Remove RAM_SAVE_FLAG_MULTIFD_PAGE
---
 migration/ram.c        | 144 ++++++++++++++++++++++++++++++++++++++++++++++++-
 migration/trace-events |   3 +-
 2 files changed, 144 insertions(+), 3 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index 4efac0c20c..df9646ed2e 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -54,6 +54,7 @@
 #include "migration/block.h"
 #include "sysemu/sysemu.h"
 #include "qemu/uuid.h"
+#include "qemu/iov.h"
 
 /***********************************************************/
 /* ram save/restore */
@@ -397,7 +398,19 @@ static void compress_threads_save_setup(void)
 
 /* Multiple fd's */
 
+typedef struct {
+    /* number of used pages */
+    uint32_t used;
+    /* number of allocated pages */
+    uint32_t allocated;
+    /* global number of generated multifd packets */
+    uint32_t seq;
+    struct iovec *iov;
+    RAMBlock *block;
+} multifd_pages_t;
+
 struct MultiFDSendParams {
+    /* not changed */
     uint8_t id;
     char *name;
     QemuThread thread;
@@ -405,8 +418,15 @@ struct MultiFDSendParams {
     QemuSemaphore sem;
     QemuMutex mutex;
     bool running;
+    /* protected by param mutex */
     bool quit;
     bool sync;
+    multifd_pages_t *pages;
+    /* how many patches has sent this channel */
+    uint32_t packets_sent;
+    /* protected by multifd mutex */
+    /* has the thread finish the last submitted job */
+    bool done;
 };
 typedef struct MultiFDSendParams MultiFDSendParams;
 
@@ -416,8 +436,31 @@ struct {
     int count;
     /* syncs main thread and channels */
     QemuSemaphore sem_main;
+    QemuMutex mutex;
+    QemuSemaphore sem;
+    multifd_pages_t *pages;
 } *multifd_send_state;
 
+static void multifd_pages_init(multifd_pages_t **ppages, size_t size)
+{
+    multifd_pages_t *pages = g_new0(multifd_pages_t, 1);
+
+    pages->allocated = size;
+    pages->iov = g_new0(struct iovec, size);
+    *ppages = pages;
+}
+
+static void multifd_pages_clear(multifd_pages_t *pages)
+{
+    pages->used = 0;
+    pages->allocated = 0;
+    pages->seq = 0;
+    pages->block = NULL;
+    g_free(pages->iov);
+    pages->iov = NULL;
+    g_free(pages);
+}
+
 static void terminate_multifd_send_threads(Error *errp)
 {
     int i;
@@ -464,10 +507,14 @@ int multifd_save_cleanup(Error **errp)
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
         p->name = NULL;
+        multifd_pages_clear(p->pages);
+        p->pages = NULL;
     }
     qemu_sem_destroy(&multifd_send_state->sem_main);
     g_free(multifd_send_state->params);
     multifd_send_state->params = NULL;
+    multifd_pages_clear(multifd_send_state->pages);
+    multifd_send_state->pages = NULL;
     g_free(multifd_send_state);
     multifd_send_state = NULL;
     return ret;
@@ -516,6 +563,7 @@ static void *multifd_send_thread(void *opaque)
         terminate_multifd_send_threads(local_err);
         return NULL;
     }
+    qemu_sem_post(&multifd_send_state->sem);
 
     while (true) {
         qemu_sem_wait(&p->sem);
@@ -530,9 +578,23 @@ static void *multifd_send_thread(void *opaque)
             qemu_mutex_unlock(&p->mutex);
             break;
         }
+        if (p->pages->used) {
+            p->pages->used = 0;
+            qemu_mutex_unlock(&p->mutex);
+
+            trace_multifd_send(p->id, p->pages->seq, p->pages->used);
+            /* ToDo: send page here */
+
+            qemu_mutex_lock(&multifd_send_state->mutex);
+            p->done = true;
+            p->packets_sent++;
+            qemu_mutex_unlock(&multifd_send_state->mutex);
+            qemu_sem_post(&multifd_send_state->sem);
+            continue;
+        }
         qemu_mutex_unlock(&p->mutex);
     }
-    trace_multifd_send_thread_end(p->id);
+    trace_multifd_send_thread_end(p->id, p->packets_sent);
 
     return NULL;
 }
@@ -571,7 +633,10 @@ int multifd_save_setup(void)
     multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
     atomic_set(&multifd_send_state->count, 0);
     qemu_sem_init(&multifd_send_state->sem_main, 0);
-
+    qemu_mutex_init(&multifd_send_state->mutex);
+    qemu_sem_init(&multifd_send_state->sem, 0);
+    multifd_pages_init(&multifd_send_state->pages,
+                       migrate_multifd_page_count());
     for (i = 0; i < thread_count; i++) {
         MultiFDSendParams *p = &multifd_send_state->params[i];
 
@@ -579,6 +644,8 @@ int multifd_save_setup(void)
         qemu_sem_init(&p->sem, 0);
         p->quit = false;
         p->id = i;
+        p->done = true;
+        multifd_pages_init(&p->pages, migrate_multifd_page_count());
         p->name = g_strdup_printf("multifdsend_%d", i);
         socket_send_channel_create(multifd_new_send_channel_async, p);
     }
@@ -586,6 +653,50 @@ int multifd_save_setup(void)
     return 0;
 }
 
+static void multifd_send_page(RAMBlock *block, ram_addr_t offset,
+                                  bool last_page)
+{
+    int i;
+    static int next_channel = 0;
+    MultiFDSendParams *p = NULL; /* make happy gcc */
+    multifd_pages_t *pages = multifd_send_state->pages;
+
+    if (!pages->block) {
+        pages->block = block;
+    }
+
+    pages->iov[pages->used].iov_base = block->host + offset;
+    pages->iov[pages->used].iov_len = TARGET_PAGE_SIZE;
+    pages->used++;
+
+    if (!last_page) {
+        if (pages->used < pages->allocated) {
+            return;
+        }
+    }
+
+    qemu_sem_wait(&multifd_send_state->sem);
+    qemu_mutex_lock(&multifd_send_state->mutex);
+    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels() ) {
+        p = &multifd_send_state->params[i];
+
+        if (p->done) {
+            p->done = false;
+            next_channel = (i + 1) % migrate_multifd_channels();
+            break;
+        }
+    }
+    qemu_mutex_unlock(&multifd_send_state->mutex);
+    qemu_mutex_lock(&p->mutex);
+    p->pages->used = 0;
+    p->pages->seq = pages->seq + 1;
+    p->pages->block = NULL;
+    multifd_send_state->pages = p->pages;
+    p->pages = pages;
+    qemu_mutex_unlock(&p->mutex);
+    qemu_sem_post(&p->sem);
+}
+
 struct MultiFDRecvParams {
     uint8_t id;
     char *name;
@@ -1220,6 +1331,31 @@ 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);
+    if (pages == -1) {
+        ram_counters.transferred +=
+            save_page_header(rs, rs->f, block,
+                             offset | RAM_SAVE_FLAG_PAGE);
+        multifd_send_page(block, offset, rs->migration_dirty_pages == 1);
+        qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
+        ram_counters.transferred += TARGET_PAGE_SIZE;
+        pages = 1;
+        ram_counters.normal++;
+    }
+
+    return pages;
+}
+
 static int do_compress_ram_page(QEMUFile *f, RAMBlock *block,
                                 ram_addr_t offset)
 {
@@ -1648,6 +1784,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);
         }
@@ -3047,6 +3185,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
     if (!migrate_use_compression()) {
         invalid_flags |= RAM_SAVE_FLAG_COMPRESS_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
@@ -3166,6 +3305,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
                 break;
             }
             break;
+
         case RAM_SAVE_FLAG_EOS:
             /* normal exit */
             multifd_recv_sync_main();
diff --git a/migration/trace-events b/migration/trace-events
index db88fa699f..f6ab2c7bcb 100644
--- a/migration/trace-events
+++ b/migration/trace-events
@@ -80,9 +80,10 @@ ram_save_queue_pages(const char *rbname, size_t start, size_t len) "%s: start: 0
 multifd_send_sync_main(void) ""
 multifd_recv_sync_main(void) ""
 multifd_send_thread_start(int id) "%d"
-multifd_send_thread_end(int id) "%d"
+multifd_send_thread_end(char id, uint32_t packets) "channel %d packets %d"
 multifd_recv_thread_start(int id) "%d"
 multifd_recv_thread_end(int id) "%d"
+multifd_send(char id, int seq, int num) "channel %d sequence %d num pages %d"
 
 # migration/migration.c
 await_return_path_close_on_source_close(void) ""
-- 
2.14.3

[Qemu-devel] [PATCH v10 23/24] migration: Create pages structure for reception

[Juan Quintela]

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 15 +++++++++++++++
 1 file changed, 15 insertions(+)

diff --git a/migration/ram.c b/migration/ram.c
index df9646ed2e..264d2e462a 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -698,6 +698,7 @@ static void multifd_send_page(RAMBlock *block, ram_addr_t offset,
 }
 
 struct MultiFDRecvParams {
+    /* not changed */
     uint8_t id;
     char *name;
     QemuThread thread;
@@ -705,8 +706,13 @@ struct MultiFDRecvParams {
     QemuSemaphore sem;
     QemuMutex mutex;
     bool running;
+    /* protected by param mutex */
     bool quit;
     bool sync;
+    /* how many patckets has recv this channel */
+    uint32_t packets_recv;
+    multifd_pages_t *pages;
+    bool done;
 };
 typedef struct MultiFDRecvParams MultiFDRecvParams;
 
@@ -716,6 +722,7 @@ struct {
     int count;
     /* syncs main thread and channels */
     QemuSemaphore sem_main;
+    multifd_pages_t *pages;
 } *multifd_recv_state;
 
 static void terminate_multifd_recv_threads(Error *errp)
@@ -764,10 +771,14 @@ int multifd_load_cleanup(Error **errp)
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
         p->name = NULL;
+        multifd_pages_clear(p->pages);
+        p->pages = NULL;
     }
     qemu_sem_destroy(&multifd_recv_state->sem_main);
     g_free(multifd_recv_state->params);
     multifd_recv_state->params = NULL;
+    multifd_pages_clear(multifd_recv_state->pages);
+    multifd_recv_state->pages = NULL;
     g_free(multifd_recv_state);
     multifd_recv_state = NULL;
 
@@ -834,6 +845,9 @@ int multifd_load_setup(void)
     multifd_recv_state->params = g_new0(MultiFDRecvParams, thread_count);
     atomic_set(&multifd_recv_state->count, 0);
     qemu_sem_init(&multifd_recv_state->sem_main, 0);
+    multifd_pages_init(&multifd_recv_state->pages,
+                       migrate_multifd_page_count());
+
     for (i = 0; i < thread_count; i++) {
         MultiFDRecvParams *p = &multifd_recv_state->params[i];
 
@@ -842,6 +856,7 @@ int multifd_load_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdrecv_%d", i);
+        multifd_pages_init(&p->pages, migrate_multifd_page_count());
     }
 
     return 0;
-- 
2.14.3

[Qemu-devel] [PATCH v10 24/24] [RFC] migration: Send pages through the multifd channels

[Juan Quintela]

Migration ends correctly, but there is still a race between clean up
and last synchronization.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c        | 132 ++++++++++++++++++++++++++++++++++++++++++++++---
 migration/trace-events |   3 +-
 2 files changed, 126 insertions(+), 9 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index 264d2e462a..577b448db3 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -398,6 +398,19 @@ static void compress_threads_save_setup(void)
 
 /* Multiple fd's */
 
+#define MULTIFD_MAGIC 0x112233d
+#define MULTIFD_VERSION 1
+
+typedef struct {
+    uint32_t magic;
+    uint32_t version;
+    uint32_t size;
+    uint32_t used;
+    uint32_t seq;
+    char ramblock[256];
+    ram_addr_t offset[];
+} __attribute__((packed)) MultiFDPacket_t;
+
 typedef struct {
     /* number of used pages */
     uint32_t used;
@@ -407,6 +420,8 @@ typedef struct {
     uint32_t seq;
     struct iovec *iov;
     RAMBlock *block;
+    uint32_t packet_len;
+    MultiFDPacket_t *packet;
 } multifd_pages_t;
 
 struct MultiFDSendParams {
@@ -447,6 +462,8 @@ static void multifd_pages_init(multifd_pages_t **ppages, size_t size)
 
     pages->allocated = size;
     pages->iov = g_new0(struct iovec, size);
+    pages->packet_len = sizeof(MultiFDPacket_t) + sizeof(ram_addr_t) * size;
+    pages->packet = g_malloc0(pages->packet_len);
     *ppages = pages;
 }
 
@@ -458,6 +475,9 @@ static void multifd_pages_clear(multifd_pages_t *pages)
     pages->block = NULL;
     g_free(pages->iov);
     pages->iov = NULL;
+    pages->packet_len = 0;
+    g_free(pages->packet);
+    pages->packet = NULL;
     g_free(pages);
 }
 
@@ -499,7 +519,6 @@ int multifd_save_cleanup(Error **errp)
 
         if (p->running) {
             qemu_thread_join(&p->thread);
-            p->running = false;
         }
         socket_send_channel_destroy(p->c);
         p->c = NULL;
@@ -535,7 +554,16 @@ static void multifd_send_sync_main(void)
         qemu_sem_post(&p->sem);
     }
     for (i = 0; i < migrate_multifd_channels(); i++) {
-        qemu_sem_wait(&multifd_send_state->sem_main);
+        MultiFDSendParams *p = &multifd_send_state->params[i];
+        bool wait = true;
+
+        qemu_mutex_lock(&p->mutex);
+        wait = p->running;
+        qemu_mutex_unlock(&p->mutex);
+
+        if (wait) {
+            qemu_sem_wait(&multifd_send_state->sem_main);
+        }
     }
     trace_multifd_send_sync_main();
 }
@@ -575,16 +603,37 @@ static void *multifd_send_thread(void *opaque)
             continue;
         }
         if (p->quit) {
+            p->running = false;
             qemu_mutex_unlock(&p->mutex);
             break;
         }
         if (p->pages->used) {
+            MultiFDPacket_t *packet = p->pages->packet;
+            Error *local_err = NULL;
+            size_t ret;
+
+            packet->used = p->pages->used;
             p->pages->used = 0;
             qemu_mutex_unlock(&p->mutex);
+            packet->magic = MULTIFD_MAGIC;
+            packet->version = MULTIFD_VERSION;
 
-            trace_multifd_send(p->id, p->pages->seq, p->pages->used);
-            /* ToDo: send page here */
-
+            strncpy(packet->ramblock, p->pages->block->idstr, 256);
+            packet->size = migrate_multifd_page_count();
+            packet->seq = p->pages->seq;
+            ret = qio_channel_write_all(p->c, (void *)packet,
+                                        p->pages->packet_len, &local_err);
+            if (ret != 0) {
+                terminate_multifd_send_threads(local_err);
+                return NULL;
+            }
+            trace_multifd_send(p->id, p->pages->seq, packet->used);
+            ret = qio_channel_writev_all(p->c, p->pages->iov,
+                                         packet->used, &local_err);
+            if (ret != 0) {
+                terminate_multifd_send_threads(local_err);
+                return NULL;
+            }
             qemu_mutex_lock(&multifd_send_state->mutex);
             p->done = true;
             p->packets_sent++;
@@ -763,7 +812,6 @@ int multifd_load_cleanup(Error **errp)
 
         if (p->running) {
             qemu_thread_join(&p->thread);
-            p->running = false;
         }
         socket_recv_channel_unref(p->c);
         p->c = NULL;
@@ -801,17 +849,48 @@ static void multifd_recv_sync_main(void)
         qemu_sem_post(&p->sem);
     }
     for (i = 0; i < migrate_multifd_channels(); i++) {
-        qemu_sem_wait(&multifd_recv_state->sem_main);
+        MultiFDRecvParams *p = &multifd_recv_state->params[i];
+        bool wait = true;
+
+        qemu_mutex_lock(&p->mutex);
+        wait = p->running && !p->quit;
+        qemu_mutex_unlock(&p->mutex);
+
+        if (wait) {
+            qemu_sem_wait(&multifd_recv_state->sem_main);
+        }
     }
     trace_multifd_recv_sync_main();
 }
 
+static gboolean recv_channel_ready(QIOChannel *ioc,
+                                   GIOCondition condition,
+                                   gpointer opaque)
+{
+    MultiFDRecvParams *p = opaque;
+
+    if (condition != G_IO_IN) {
+        return G_SOURCE_REMOVE;
+    }
+
+    qemu_mutex_lock(&p->mutex);
+    p->done = false;
+    qemu_mutex_unlock(&p->mutex);
+    qemu_sem_post(&p->sem);
+
+    return G_SOURCE_CONTINUE;
+
+}
+
 static void *multifd_recv_thread(void *opaque)
 {
     MultiFDRecvParams *p = opaque;
 
     trace_multifd_recv_thread_start(p->id);
 
+    qio_channel_add_watch(p->c, G_IO_IN | G_IO_HUP | G_IO_ERR,
+                          recv_channel_ready, p, NULL);
+
     while (true) {
         qemu_sem_wait(&p->sem);
         qemu_mutex_lock(&p->mutex);
@@ -821,14 +900,50 @@ static void *multifd_recv_thread(void *opaque)
             qemu_sem_post(&multifd_recv_state->sem_main);
             continue;
         }
+        if (!p->done) {
+            MultiFDPacket_t *packet = p->pages->packet;
+            RAMBlock *block;
+            Error *local_err = NULL;
+            size_t ret;
+            int i;
+
+            qemu_mutex_unlock(&p->mutex);
+
+            ret = qio_channel_read_all(p->c, (void *)packet,
+                                       p->pages->packet_len, &local_err);
+            if (ret != 0) {
+                terminate_multifd_recv_threads(local_err);
+                return NULL;
+            }
+            block = qemu_ram_block_by_name(packet->ramblock);
+            p->pages->seq = packet->seq;
+            for (i = 0; i < packet->used; i++) {
+                p->pages->iov[i].iov_base = block->host + packet->offset[i];
+                p->pages->iov[i].iov_len = TARGET_PAGE_SIZE;
+            }
+            trace_multifd_recv(p->id, p->pages->seq, packet->used);
+            ret = qio_channel_readv_all(p->c, p->pages->iov,
+                                        packet->used, &local_err);
+            if (ret != 0) {
+                terminate_multifd_recv_threads(local_err);
+                return NULL;
+            }
+            qemu_mutex_lock(&p->mutex);
+            p->done = true;
+            p->packets_recv++;
+            qemu_mutex_unlock(&p->mutex);
+
+            continue;
+        }
         if (p->quit) {
+            p->running = false;
             qemu_mutex_unlock(&p->mutex);
             break;
         }
         qemu_mutex_unlock(&p->mutex);
     }
 
-    trace_multifd_recv_thread_end(p->id);
+    trace_multifd_recv_thread_end(p->id, p->packets_recv);
     return NULL;
 }
 
@@ -854,6 +969,7 @@ int multifd_load_setup(void)
         qemu_mutex_init(&p->mutex);
         qemu_sem_init(&p->sem, 0);
         p->quit = false;
+        p->done = true;
         p->id = i;
         p->name = g_strdup_printf("multifdrecv_%d", i);
         multifd_pages_init(&p->pages, migrate_multifd_page_count());
diff --git a/migration/trace-events b/migration/trace-events
index f6ab2c7bcb..e9f1aae985 100644
--- a/migration/trace-events
+++ b/migration/trace-events
@@ -82,8 +82,9 @@ multifd_recv_sync_main(void) ""
 multifd_send_thread_start(int id) "%d"
 multifd_send_thread_end(char id, uint32_t packets) "channel %d packets %d"
 multifd_recv_thread_start(int id) "%d"
-multifd_recv_thread_end(int id) "%d"
+multifd_recv_thread_end(char id, uint32_t packets) "channel %d packets %d"
 multifd_send(char id, int seq, int num) "channel %d sequence %d num pages %d"
+multifd_recv(char id, int seq, int num) "channel %d sequence %d num pages %d"
 
 # migration/migration.c
 await_return_path_close_on_source_close(void) ""
-- 
2.14.3

Re: [Qemu-devel] [RFC v10 00/24] Multifd

[no-reply]

Hi,

This series seems to have some coding style problems. See output below for
more information:

Type: series
Message-id: 20180307110010.2205-1-quintela@redhat.com
Subject: [Qemu-devel] [RFC v10 00/24] Multifd

=== 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
git config --local diff.algorithm histogram

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'
0d4c2b3560 migration: Send pages through the multifd channels
dfd20c6d98 migration: Create pages structure for reception
24414a65b0 migration: Create ram_multifd_page
2d73c8a690 migration: Transmit initial package through the multifd channels
cfe7de3af1 migration: Delay start of migration main routines
c753eaf6af migration: Create multifd channels
87eeee0345 migration: Add multifd traces for start/end thread
17e961cc77 migration: Export functions to create send channels
94f824225d migration: Synchronize recv threads
7f34d667a9 migration: Synchronize send threads
4a4b00a54d migration: Be sure all recv channels are created
c97d843216 migration: Introduce multifd_recv_new_channel()
8a13787f66 migration: Reference counting recv channels correctly
577fbe0895 migration: terminate_* can be called for other threads
d4f0966d50 migration: In case of error just end the migration
c9443cf0c7 migration: Set error state in case of error
caea778e74 migration: Add multifd test
eab612ab0a tests: Add migration compress threads tests
849a3b52d2 tests: Add basic migration precopy tcp test
5ce68931c2 tests: Migration ppc now inlines its program
5f046ff880 migration: Set the migration tcp port
4a55e7ead3 migration: Create tcp_port parameter
0daf31aeec tests: Add migration xbzrle test
2d9ce85e32 tests: Add migration precopy test

=== OUTPUT BEGIN ===
Checking PATCH 1/24: tests: Add migration precopy test...
Checking PATCH 2/24: tests: Add migration xbzrle test...
Checking PATCH 3/24: migration: Create tcp_port parameter...
Checking PATCH 4/24: migration: Set the migration tcp port...
Checking PATCH 5/24: tests: Migration ppc now inlines its program...
Checking PATCH 6/24: tests: Add basic migration precopy tcp test...
Checking PATCH 7/24: tests: Add migration compress threads tests...
Checking PATCH 8/24: migration: Add multifd test...
Checking PATCH 9/24: migration: Set error state in case of error...
Checking PATCH 10/24: migration: In case of error just end the migration...
Checking PATCH 11/24: migration: terminate_* can be called for other threads...
Checking PATCH 12/24: migration: Reference counting recv channels correctly...
Checking PATCH 13/24: migration: Introduce multifd_recv_new_channel()...
Checking PATCH 14/24: migration: Be sure all recv channels are created...
Checking PATCH 15/24: migration: Synchronize send threads...
Checking PATCH 16/24: migration: Synchronize recv threads...
Checking PATCH 17/24: migration: Export functions to create send channels...
Checking PATCH 18/24: migration: Add multifd traces for start/end thread...
Checking PATCH 19/24: migration: Create multifd channels...
ERROR: do not initialise statics to 0 or NULL
#109: FILE: migration/ram.c:735:
+    static int i = 0;

total: 1 errors, 0 warnings, 100 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 20/24: migration: Delay start of migration main routines...
Checking PATCH 21/24: migration: Transmit initial package through the multifd channels...
Checking PATCH 22/24: migration: Create ram_multifd_page...
ERROR: do not initialise statics to 0 or NULL
#184: FILE: migration/ram.c:660:
+    static int next_channel = 0;

ERROR: space prohibited before that close parenthesis ')'
#204: FILE: migration/ram.c:680:
+    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels() ) {

total: 2 errors, 0 warnings, 250 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 23/24: migration: Create pages structure for reception...
Checking PATCH 24/24: migration: Send pages through the multifd channels...
=== 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

Re: [Qemu-devel] [PATCH v10 03/24] migration: Create tcp_port parameter

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 11:59:49AM +0100, Juan Quintela wrote:
> It will be used to store the uri tcp_port parameter.  This is the only
> parameter than can change and we can need to be able to connect to it.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> 
> --
> 
> This used to be uri parameter, but it has so many troubles to
> reproduce that it don't just make sense.
> ---
>  hmp.c                 |  3 +++
>  migration/migration.c |  8 ++++++++
>  qapi/migration.json   | 19 ++++++++++++++++---
>  3 files changed, 27 insertions(+), 3 deletions(-)

> diff --git a/qapi/migration.json b/qapi/migration.json
> index 7f465a1902..b6ef193f47 100644
> --- a/qapi/migration.json
> +++ b/qapi/migration.json
> @@ -490,6 +490,9 @@
>  #                     and a power of 2
>  #                     (Since 2.11)
>  #
> +# @x-tcp-port: Only used for tcp, to know what the real port is
> +#                     (Since 2.12)
> +#
>  # Since: 2.4
>  ##
>  { 'enum': 'MigrationParameter',
> @@ -498,7 +501,7 @@
>             'tls-creds', 'tls-hostname', 'max-bandwidth',
>             'downtime-limit', 'x-checkpoint-delay', 'block-incremental',
>             'x-multifd-channels', 'x-multifd-page-count',
> -           'xbzrle-cache-size' ] }
> +           'xbzrle-cache-size', 'x-tcp-port' ] }
>  
>  ##
>  # @MigrateSetParameters:
> @@ -566,6 +569,10 @@
>  #                     needs to be a multiple of the target page size
>  #                     and a power of 2
>  #                     (Since 2.11)
> +#
> +# @x-tcp-port: Only used for tcp, to know what the real port is
> +#                     (Since 2.12)
> +#
>  # Since: 2.4
>  ##
>  # TODO either fuse back into MigrationParameters, or make
> @@ -584,7 +591,8 @@
>              '*block-incremental': 'bool',
>              '*x-multifd-channels': 'int',
>              '*x-multifd-page-count': 'int',
> -            '*xbzrle-cache-size': 'size' } }
> +            '*xbzrle-cache-size': 'size',
> +            '*x-tcp-port': 'uint16'} }

This should not exist - this exposes this parameter in migate-set-parameters
as a end user settable property, which is definitely not desirable.

It is only something we should report with 'query-migrate' / 'info migrate'

>  # @migrate-set-parameters:
> @@ -667,6 +675,10 @@
>  #                     needs to be a multiple of the target page size
>  #                     and a power of 2
>  #                     (Since 2.11)
> +#
> +# @x-tcp-port: Only used for tcp, to know what the real port is
> +#                     (Since 2.12)
> +#
>  # Since: 2.4
>  ##
>  { 'struct': 'MigrationParameters',
> @@ -683,7 +695,8 @@
>              '*block-incremental': 'bool' ,
>              '*x-multifd-channels': 'uint8',
>              '*x-multifd-page-count': 'uint32',
> -            '*xbzrle-cache-size': 'size' } }
> +            '*xbzrle-cache-size': 'size',
> +            '*x-tcp-port': 'uint16'} }

As mentioned in previous review, IMHO we should be reporting the full
socket address, and allow an array of them, since we're going to have
more than one address available. i.e.

   '*socket-address': ['SocketAddress']

It doesn't cover non-socket based URIs, but that's fine, because for those
the mgmt app already knows how the channel is setup. We just need the
array of SocketAddress, because for socket URIs, the hostname, gets turned
into an array of addresses, and the mgmt app can't discover them.

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 :|

Re: [Qemu-devel] [PATCH v10 04/24] migration: Set the migration tcp port

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 11:59:50AM +0100, Juan Quintela wrote:
> We can set the port parameter as zero.  This patch lets us know what
> port the system was choosen for us.  Now we can migrate to this place.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> 
> --
> 
> This was migrate_set_uri(), but as we only need the tcp_port, change
> to that one.
> ---
>  migration/migration.c | 10 ++++++++++
>  migration/migration.h |  2 ++
>  migration/socket.c    | 35 ++++++++++++++++++++++++++++++-----
>  3 files changed, 42 insertions(+), 5 deletions(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index 31b16a335b..c398665de7 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -268,6 +268,16 @@ int migrate_send_rp_req_pages(MigrationIncomingState *mis, const char *rbname,
>      return migrate_send_rp_message(mis, msg_type, msglen, bufc);
>  }
>  
> +void migrate_set_port(const uint16_t port, Error **errp)
> +{
> +    MigrateSetParameters p = {
> +        .has_x_tcp_port = true,
> +        .x_tcp_port = port,
> +    };
> +
> +    qmp_migrate_set_parameters(&p, errp);
> +}

This is really not nice - it is requiring the QMP  'migrate-set-parameters'
command to accept an extra field that is never something we want the end
user to be allowed to set. We should not use the public QMP schema for
data items we are just passing between 2 internal pieces of code.

>  void qemu_start_incoming_migration(const char *uri, Error **errp)
>  {
>      const char *p;
> diff --git a/migration/migration.h b/migration/migration.h
> index 08c5d2ded1..f40014cf94 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -234,4 +234,6 @@ void migrate_send_rp_pong(MigrationIncomingState *mis,
>  int migrate_send_rp_req_pages(MigrationIncomingState *mis, const char* rbname,
>                                ram_addr_t start, size_t len);
>  
> +void migrate_set_port(const uint16_t port, Error **errp);
> +
>  #endif
> diff --git a/migration/socket.c b/migration/socket.c
> index e090097077..08606c665d 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -15,6 +15,7 @@
>   */
>  
>  #include "qemu/osdep.h"
> +#include "qemu/cutils.h"
>  
>  #include "qemu-common.h"
>  #include "qemu/error-report.h"
> @@ -160,17 +161,24 @@ out:
>  }
>  
>  
> -static void socket_start_incoming_migration(SocketAddress *saddr,
> -                                            Error **errp)
> +static SocketAddress *socket_start_incoming_migration(SocketAddress *saddr,
> +                                                      Error **errp)
>  {
>      QIOChannelSocket *listen_ioc = qio_channel_socket_new();
> +    SocketAddress *address;
>  
>      qio_channel_set_name(QIO_CHANNEL(listen_ioc),
>                           "migration-socket-listener");
>  
>      if (qio_channel_socket_listen_sync(listen_ioc, saddr, errp) < 0) {
>          object_unref(OBJECT(listen_ioc));
> -        return;
> +        return NULL;
> +    }
> +
> +    address = qio_channel_socket_get_local_address(listen_ioc, errp);
> +    if (address < 0) {
> +        object_unref(OBJECT(listen_ioc));
> +        return NULL;
>      }
>  
>      qio_channel_add_watch(QIO_CHANNEL(listen_ioc),
> @@ -178,14 +186,28 @@ static void socket_start_incoming_migration(SocketAddress *saddr,
>                            socket_accept_incoming_migration,
>                            listen_ioc,
>                            (GDestroyNotify)object_unref);
> +    return address;
>  }
>  
>  void tcp_start_incoming_migration(const char *host_port, Error **errp)
>  {
>      Error *err = NULL;
>      SocketAddress *saddr = tcp_build_address(host_port, &err);
> +
>      if (!err) {
> -        socket_start_incoming_migration(saddr, &err);
> +        SocketAddress *address = socket_start_incoming_migration(saddr, &err);
> +
> +        if (address) {
> +            unsigned long long port;
> +
> +            if (parse_uint_full(address->u.inet.port, &port, 10) < 0) {
> +                error_setg(errp, "error parsing port in '%s'",
> +                           address->u.inet.port);
> +            } else {
> +                migrate_set_port(port, errp);
> +            }
> +            qapi_free_SocketAddress(address);
> +        }
>      }
>      qapi_free_SocketAddress(saddr);
>      error_propagate(errp, err);
> @@ -194,6 +216,9 @@ void tcp_start_incoming_migration(const char *host_port, Error **errp)
>  void unix_start_incoming_migration(const char *path, Error **errp)
>  {
>      SocketAddress *saddr = unix_build_address(path);
> -    socket_start_incoming_migration(saddr, errp);
> +    SocketAddress *address;
> +
> +    address = socket_start_incoming_migration(saddr, errp);
> +    qapi_free_SocketAddress(address);
>      qapi_free_SocketAddress(saddr);
>  }
> -- 
> 2.14.3
> 
> 

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 :|

Re: [Qemu-devel] [PATCH v10 10/24] migration: In case of error just end the migration

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 11:59:56AM +0100, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/socket.c | 6 ++----
>  1 file changed, 2 insertions(+), 4 deletions(-)

Reviewed-by: Daniel P. Berrangé <berrange@redhat.com>

> diff --git a/migration/socket.c b/migration/socket.c
> index 08606c665d..b12b0a462e 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -139,9 +139,8 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
>      sioc = qio_channel_socket_accept(QIO_CHANNEL_SOCKET(ioc),
>                                       &err);
>      if (!sioc) {
> -        error_report("could not accept migration connection (%s)",
> -                     error_get_pretty(err));
> -        goto out;
> +        migrate_set_error(migrate_get_current(), err);
> +        return G_SOURCE_REMOVE;

It will only return NULL if a client connected & then went away. This should
not happen with a "normal" mgmt app usage. On the flip side this allows a
malicious network attacker to inflict a denial of service on the migration
by simply connecting to target QEMU & immediately exiting.

Our "authentication" for migration relies on being able to validate the TLS
certs during TLS handshake. So in general we ought to allow repeated incoming
connections until we get a successful handshake. 

So in fact, I think a better fix here is to simply remove the original
'error_report' line, and ensure we return G_SOURCE_CONTINUE to wait for
another incoming connection from the real mgmt app.

>      }
>  
>      trace_migration_socket_incoming_accepted();
> @@ -150,7 +149,6 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
>      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);
> -- 
> 2.14.3
> 
> 

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 :|

Re: [Qemu-devel] [PATCH v10 12/24] migration: Reference counting recv channels correctly

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 11:59:58AM +0100, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/socket.c | 11 +++++++++++
>  migration/socket.h |  7 +++++++
>  2 files changed, 18 insertions(+)
> 
> diff --git a/migration/socket.c b/migration/socket.c
> index b12b0a462e..26110739cf 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -27,6 +27,17 @@
>  #include "io/channel-socket.h"
>  #include "trace.h"
>  
> +int socket_recv_channel_ref(QIOChannel *recv)
> +{
> +    object_ref(OBJECT(recv));
> +    return 0;
> +}
> +
> +int socket_recv_channel_unref(QIOChannel *recv)
> +{
> +    object_unref(OBJECT(recv));
> +    return 0;
> +}

These helpers don't really add any value IMHO  - just call object_ref/unref
directly where needed. We don't provide explicit qio_channel_ref() wrappers
around object_ref because they add no value.

>  
>  static SocketAddress *tcp_build_address(const char *host_port, Error **errp)
>  {
> diff --git a/migration/socket.h b/migration/socket.h
> index 6b91e9db38..638a85255a 100644
> --- a/migration/socket.h
> +++ b/migration/socket.h
> @@ -16,6 +16,13 @@
>  
>  #ifndef QEMU_MIGRATION_SOCKET_H
>  #define QEMU_MIGRATION_SOCKET_H
> +
> +#include "io/channel.h"
> +#include "io/task.h"
> +
> +int socket_recv_channel_ref(QIOChannel *recv);
> +int socket_recv_channel_unref(QIOChannel *recv);
> +
>  void tcp_start_incoming_migration(const char *host_port, Error **errp);
>  
>  void tcp_start_outgoing_migration(MigrationState *s, const char *host_port,
> -- 
> 2.14.3
> 
> 

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 :|

Re: [Qemu-devel] [PATCH v10 17/24] migration: Export functions to create send channels

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 12:00:03PM +0100, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/socket.c | 28 +++++++++++++++++++++++++++-
>  migration/socket.h |  3 +++
>  2 files changed, 30 insertions(+), 1 deletion(-)
> 
> diff --git a/migration/socket.c b/migration/socket.c
> index 26110739cf..b3b5571ebb 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -39,6 +39,28 @@ int socket_recv_channel_unref(QIOChannel *recv)
>      return 0;
>  }
>  
> +struct SocketOutgoingArgs {
> +    SocketAddress *saddr;
> +} outgoing_args;
> +
> +void socket_send_channel_create(void (*f)(QIOTask *, gpointer), void *data)

This should use the proper typedef

   socket_send_channel_create(QIOTaskFunc f, void *data)

> +{
> +    QIOChannelSocket *sioc = qio_channel_socket_new();
> +    qio_channel_socket_connect_async(sioc, outgoing_args.saddr,
> +                                     f, data, NULL);
> +}
> +
> +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)
>  {
>      SocketAddress *saddr;
> @@ -106,6 +128,11 @@ static void socket_start_outgoing_migration(MigrationState *s,
>      struct SocketConnectData *data = g_new0(struct SocketConnectData, 1);
>  
>      data->s = s;
> +
> +    /* in case previous migration leaked it */
> +    qapi_free_SocketAddress(outgoing_args.saddr);
> +    outgoing_args.saddr = saddr;
> +
>      if (saddr->type == SOCKET_ADDRESS_TYPE_INET) {
>          data->hostname = g_strdup(saddr->u.inet.host);
>      }
> @@ -116,7 +143,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 638a85255a..cbdb8d64c3 100644
> --- a/migration/socket.h
> +++ b/migration/socket.h
> @@ -23,6 +23,9 @@
>  int socket_recv_channel_ref(QIOChannel *recv);
>  int socket_recv_channel_unref(QIOChannel *recv);
>  
> +void socket_send_channel_create(void (*f)(QIOTask *, gpointer), void *data);

Again use the proper typedef for the callback


> +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.14.3
> 
> 

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 :|

Re: [Qemu-devel] [PATCH v10 18/24] migration: Add multifd traces for start/end thread

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 12:00:04PM +0100, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c        | 6 ++++++
>  migration/trace-events | 4 ++++
>  2 files changed, 10 insertions(+)

Reviewed-by: Daniel P. Berrangé <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 :|

Re: [Qemu-devel] [PATCH v10 21/24] migration: Transmit initial package through the multifd channels

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 12:00:07PM +0100, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c | 50 +++++++++++++++++++++++++++++++++++++++++++++-----
>  1 file changed, 45 insertions(+), 5 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index 1aab96bd5e..4efac0c20c 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -52,6 +52,8 @@
>  #include "qemu/rcu_queue.h"
>  #include "migration/colo.h"
>  #include "migration/block.h"
> +#include "sysemu/sysemu.h"
> +#include "qemu/uuid.h"
>  
>  /***********************************************************/
>  /* ram save/restore */
> @@ -491,12 +493,30 @@ static void multifd_send_sync_main(void)
>      trace_multifd_send_sync_main();
>  }
>  
> +typedef struct {
> +    uint32_t version;
> +    unsigned char uuid[16]; /* QemuUUID */
> +    uint8_t id;
> +} __attribute__((packed)) MultiFDInit_t;
> +
>  static void *multifd_send_thread(void *opaque)
>  {
>      MultiFDSendParams *p = opaque;
> +    MultiFDInit_t msg;
> +    Error *local_err = NULL;
> +    size_t ret;
>  
>      trace_multifd_send_thread_start(p->id);
>  
> +    msg.version = 1;

I'm thinking we should standardize byte-order for this, as you could be
migrating between qemu-system-x86_64 running TCG on PPC, to a another
qemu-system-x86_64 running TCG on AArch64, and so have mixed-endianness.

Just a 'msg.version = htonl(1) call to set network byte order ?

> +    msg.id = p->id;
> +    memcpy(msg.uuid, &qemu_uuid.data, sizeof(msg.uuid));
> +    ret = qio_channel_write_all(p->c, (char *)&msg, sizeof(msg), &local_err);
> +    if (ret != 0) {
> +        terminate_multifd_send_threads(local_err);
> +        return NULL;
> +    }
> +
>      while (true) {
>          qemu_sem_wait(&p->sem);
>          qemu_mutex_lock(&p->mutex);
> @@ -730,12 +750,32 @@ bool multifd_recv_all_channels_created(void)
>  void multifd_recv_new_channel(QIOChannel *ioc)
>  {
>      MultiFDRecvParams *p;
> -    /* we need to invent channels id's until we transmit */
> -    /* we will remove this on a later patch */
> -    static int i = 0;
> +    MultiFDInit_t msg;
> +    Error *local_err = NULL;
> +    size_t ret;
>  
> -    p = &multifd_recv_state->params[i];
> -    i++;
> +    ret = qio_channel_read_all(ioc, (char *)&msg, sizeof(msg), &local_err);
> +    if (ret != 0) {
> +        terminate_multifd_recv_threads(local_err);
> +        return;
> +    }
> +
> +    if (memcmp(msg.uuid, &qemu_uuid, sizeof(qemu_uuid))) {
> +        char *uuid = qemu_uuid_unparse_strdup(&qemu_uuid);
> +        error_setg(&local_err, "multifd: received uuid '%s' and expected "
> +                   "uuid '%s' for channel %hhd", msg.uuid, uuid, msg.id);
> +        g_free(uuid);
> +        terminate_multifd_recv_threads(local_err);
> +        return;
> +    }
> +
> +    p = &multifd_recv_state->params[msg.id];

And  ntohl(msg.id) here

Also, since we're indexnig into an array using data off the network,
we should validate the index is in range to avoid out of bounds memory
access.

> +    if (p->c != NULL) {
> +        error_setg(&local_err, "multifd: received id '%d' already setup'",
> +                   msg.id);
> +        terminate_multifd_recv_threads(local_err);
> +        return;
> +    }
>      p->c = ioc;
>      socket_recv_channel_ref(ioc);


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 :|

Re: [Qemu-devel] [PATCH v10 18/24] migration: Add multifd traces for start/end thread

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 12:00:04PM +0100, Juan Quintela wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c        | 6 ++++++
>  migration/trace-events | 4 ++++
>  2 files changed, 10 insertions(+)
> 

> diff --git a/migration/trace-events b/migration/trace-events
> index 76075c26bc..db88fa699f 100644
> --- a/migration/trace-events
> +++ b/migration/trace-events
> @@ -79,6 +79,10 @@ ram_save_page(const char *rbname, uint64_t offset, void *host) "%s: offset: 0x%"
>  ram_save_queue_pages(const char *rbname, size_t start, size_t len) "%s: start: 0x%zx len: 0x%zx"
>  multifd_send_sync_main(void) ""
>  multifd_recv_sync_main(void) ""
> +multifd_send_thread_start(int id) "%d"
> +multifd_send_thread_end(int id) "%d"
> +multifd_recv_thread_start(int id) "%d"
> +multifd_recv_thread_end(int id) "%d"

Seems these should have been 'uint8_t' rather than 'int', avoiding the
change of type in later patches


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 :|

Re: [Qemu-devel] [PATCH v10 22/24] migration: Create ram_multifd_page

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 12:00:08PM +0100, 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.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> 
> --
> Add last_page parameter
> Add commets for done and address
> Remove multifd field, it is the same than normal pages
> Merge next patch, now we send multiple pages at a time
> Remove counter for multifd pages, it is identical to normal pages
> Use iovec's instead of creating the equivalent.
> Clear memory used by pages (dave)
> Use g_new0(danp)
> define MULTIFD_CONTINUE
> now pages member is a pointer
> Fix off-by-one in number of pages in one packet
> Remove RAM_SAVE_FLAG_MULTIFD_PAGE
> ---
>  migration/ram.c        | 144 ++++++++++++++++++++++++++++++++++++++++++++++++-
>  migration/trace-events |   3 +-
>  2 files changed, 144 insertions(+), 3 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index 4efac0c20c..df9646ed2e 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -54,6 +54,7 @@
>  #include "migration/block.h"
>  #include "sysemu/sysemu.h"
>  #include "qemu/uuid.h"
> +#include "qemu/iov.h"
>  
>  /***********************************************************/
>  /* ram save/restore */
> @@ -397,7 +398,19 @@ static void compress_threads_save_setup(void)
>  
>  /* Multiple fd's */
>  
> +typedef struct {
> +    /* number of used pages */
> +    uint32_t used;
> +    /* number of allocated pages */
> +    uint32_t allocated;
> +    /* global number of generated multifd packets */
> +    uint32_t seq;
> +    struct iovec *iov;
> +    RAMBlock *block;
> +} multifd_pages_t;
> +
>  struct MultiFDSendParams {
> +    /* not changed */
>      uint8_t id;
>      char *name;
>      QemuThread thread;
> @@ -405,8 +418,15 @@ struct MultiFDSendParams {
>      QemuSemaphore sem;
>      QemuMutex mutex;
>      bool running;
> +    /* protected by param mutex */
>      bool quit;
>      bool sync;
> +    multifd_pages_t *pages;
> +    /* how many patches has sent this channel */
> +    uint32_t packets_sent;
> +    /* protected by multifd mutex */
> +    /* has the thread finish the last submitted job */
> +    bool done;
>  };
>  typedef struct MultiFDSendParams MultiFDSendParams;
>  
> @@ -416,8 +436,31 @@ struct {
>      int count;
>      /* syncs main thread and channels */
>      QemuSemaphore sem_main;
> +    QemuMutex mutex;
> +    QemuSemaphore sem;
> +    multifd_pages_t *pages;
>  } *multifd_send_state;
>  
> +static void multifd_pages_init(multifd_pages_t **ppages, size_t size)
> +{
> +    multifd_pages_t *pages = g_new0(multifd_pages_t, 1);
> +
> +    pages->allocated = size;
> +    pages->iov = g_new0(struct iovec, size);
> +    *ppages = pages;
> +}
> +
> +static void multifd_pages_clear(multifd_pages_t *pages)
> +{
> +    pages->used = 0;
> +    pages->allocated = 0;
> +    pages->seq = 0;
> +    pages->block = NULL;
> +    g_free(pages->iov);
> +    pages->iov = NULL;
> +    g_free(pages);
> +}
> +
>  static void terminate_multifd_send_threads(Error *errp)
>  {
>      int i;
> @@ -464,10 +507,14 @@ int multifd_save_cleanup(Error **errp)
>          qemu_sem_destroy(&p->sem);
>          g_free(p->name);
>          p->name = NULL;
> +        multifd_pages_clear(p->pages);
> +        p->pages = NULL;
>      }
>      qemu_sem_destroy(&multifd_send_state->sem_main);
>      g_free(multifd_send_state->params);
>      multifd_send_state->params = NULL;
> +    multifd_pages_clear(multifd_send_state->pages);
> +    multifd_send_state->pages = NULL;
>      g_free(multifd_send_state);
>      multifd_send_state = NULL;
>      return ret;
> @@ -516,6 +563,7 @@ static void *multifd_send_thread(void *opaque)
>          terminate_multifd_send_threads(local_err);
>          return NULL;
>      }
> +    qemu_sem_post(&multifd_send_state->sem);
>  
>      while (true) {
>          qemu_sem_wait(&p->sem);
> @@ -530,9 +578,23 @@ static void *multifd_send_thread(void *opaque)
>              qemu_mutex_unlock(&p->mutex);
>              break;
>          }
> +        if (p->pages->used) {
> +            p->pages->used = 0;
> +            qemu_mutex_unlock(&p->mutex);
> +
> +            trace_multifd_send(p->id, p->pages->seq, p->pages->used);
> +            /* ToDo: send page here */
> +
> +            qemu_mutex_lock(&multifd_send_state->mutex);
> +            p->done = true;
> +            p->packets_sent++;
> +            qemu_mutex_unlock(&multifd_send_state->mutex);
> +            qemu_sem_post(&multifd_send_state->sem);
> +            continue;
> +        }
>          qemu_mutex_unlock(&p->mutex);
>      }
> -    trace_multifd_send_thread_end(p->id);
> +    trace_multifd_send_thread_end(p->id, p->packets_sent);
>  
>      return NULL;
>  }
> @@ -571,7 +633,10 @@ int multifd_save_setup(void)
>      multifd_send_state->params = g_new0(MultiFDSendParams, thread_count);
>      atomic_set(&multifd_send_state->count, 0);
>      qemu_sem_init(&multifd_send_state->sem_main, 0);
> -
> +    qemu_mutex_init(&multifd_send_state->mutex);
> +    qemu_sem_init(&multifd_send_state->sem, 0);
> +    multifd_pages_init(&multifd_send_state->pages,
> +                       migrate_multifd_page_count());
>      for (i = 0; i < thread_count; i++) {
>          MultiFDSendParams *p = &multifd_send_state->params[i];
>  
> @@ -579,6 +644,8 @@ int multifd_save_setup(void)
>          qemu_sem_init(&p->sem, 0);
>          p->quit = false;
>          p->id = i;
> +        p->done = true;
> +        multifd_pages_init(&p->pages, migrate_multifd_page_count());
>          p->name = g_strdup_printf("multifdsend_%d", i);
>          socket_send_channel_create(multifd_new_send_channel_async, p);
>      }
> @@ -586,6 +653,50 @@ int multifd_save_setup(void)
>      return 0;
>  }
>  
> +static void multifd_send_page(RAMBlock *block, ram_addr_t offset,
> +                                  bool last_page)
> +{
> +    int i;
> +    static int next_channel = 0;
> +    MultiFDSendParams *p = NULL; /* make happy gcc */
> +    multifd_pages_t *pages = multifd_send_state->pages;
> +
> +    if (!pages->block) {
> +        pages->block = block;
> +    }
> +
> +    pages->iov[pages->used].iov_base = block->host + offset;
> +    pages->iov[pages->used].iov_len = TARGET_PAGE_SIZE;
> +    pages->used++;
> +
> +    if (!last_page) {
> +        if (pages->used < pages->allocated) {
> +            return;
> +        }
> +    }
> +
> +    qemu_sem_wait(&multifd_send_state->sem);
> +    qemu_mutex_lock(&multifd_send_state->mutex);
> +    for (i = next_channel;; i = (i + 1) % migrate_multifd_channels() ) {
> +        p = &multifd_send_state->params[i];
> +
> +        if (p->done) {
> +            p->done = false;
> +            next_channel = (i + 1) % migrate_multifd_channels();
> +            break;
> +        }
> +    }
> +    qemu_mutex_unlock(&multifd_send_state->mutex);
> +    qemu_mutex_lock(&p->mutex);
> +    p->pages->used = 0;
> +    p->pages->seq = pages->seq + 1;
> +    p->pages->block = NULL;
> +    multifd_send_state->pages = p->pages;
> +    p->pages = pages;
> +    qemu_mutex_unlock(&p->mutex);
> +    qemu_sem_post(&p->sem);
> +}
> +
>  struct MultiFDRecvParams {
>      uint8_t id;
>      char *name;
> @@ -1220,6 +1331,31 @@ 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);
> +    if (pages == -1) {
> +        ram_counters.transferred +=
> +            save_page_header(rs, rs->f, block,
> +                             offset | RAM_SAVE_FLAG_PAGE);
> +        multifd_send_page(block, offset, rs->migration_dirty_pages == 1);
> +        qemu_put_buffer(rs->f, p, TARGET_PAGE_SIZE);
> +        ram_counters.transferred += TARGET_PAGE_SIZE;
> +        pages = 1;
> +        ram_counters.normal++;
> +    }
> +
> +    return pages;
> +}
> +
>  static int do_compress_ram_page(QEMUFile *f, RAMBlock *block,
>                                  ram_addr_t offset)
>  {
> @@ -1648,6 +1784,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);
>          }
> @@ -3047,6 +3185,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
>      if (!migrate_use_compression()) {
>          invalid_flags |= RAM_SAVE_FLAG_COMPRESS_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
> @@ -3166,6 +3305,7 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
>                  break;
>              }
>              break;
> +
>          case RAM_SAVE_FLAG_EOS:
>              /* normal exit */
>              multifd_recv_sync_main();
> diff --git a/migration/trace-events b/migration/trace-events
> index db88fa699f..f6ab2c7bcb 100644
> --- a/migration/trace-events
> +++ b/migration/trace-events
> @@ -80,9 +80,10 @@ ram_save_queue_pages(const char *rbname, size_t start, size_t len) "%s: start: 0
>  multifd_send_sync_main(void) ""
>  multifd_recv_sync_main(void) ""
>  multifd_send_thread_start(int id) "%d"
> -multifd_send_thread_end(int id) "%d"
> +multifd_send_thread_end(char id, uint32_t packets) "channel %d packets %d"
>  multifd_recv_thread_start(int id) "%d"
>  multifd_recv_thread_end(int id) "%d"
> +multifd_send(char id, int seq, int num) "channel %d sequence %d num pages %d"

Drop this, and just set 'id' to be 'uint8_t' in the earlier patches

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 :|

Re: [Qemu-devel] [PATCH v10 24/24] [RFC] migration: Send pages through the multifd channels

[Daniel P. Berrangé]

On Wed, Mar 07, 2018 at 12:00:10PM +0100, Juan Quintela wrote:
> Migration ends correctly, but there is still a race between clean up
> and last synchronization.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c        | 132 ++++++++++++++++++++++++++++++++++++++++++++++---
>  migration/trace-events |   3 +-
>  2 files changed, 126 insertions(+), 9 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index 264d2e462a..577b448db3 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -398,6 +398,19 @@ static void compress_threads_save_setup(void)
>  
>  /* Multiple fd's */
>  
> +#define MULTIFD_MAGIC 0x112233d
> +#define MULTIFD_VERSION 1
> +
> +typedef struct {
> +    uint32_t magic;
> +    uint32_t version;
> +    uint32_t size;
> +    uint32_t used;
> +    uint32_t seq;
> +    char ramblock[256];
> +    ram_addr_t offset[];
> +} __attribute__((packed)) MultiFDPacket_t;

Same question about byte ordering as earlier patches - if we have two
qemu-system-x86_64 binaries running TCG mode on hosts with different
endianness we need byte swapping

> +
>  typedef struct {
>      /* number of used pages */
>      uint32_t used;
> @@ -407,6 +420,8 @@ typedef struct {
>      uint32_t seq;
>      struct iovec *iov;
>      RAMBlock *block;
> +    uint32_t packet_len;
> +    MultiFDPacket_t *packet;
>  } multifd_pages_t;
>  
>  struct MultiFDSendParams {
> @@ -447,6 +462,8 @@ static void multifd_pages_init(multifd_pages_t **ppages, size_t size)
>  
>      pages->allocated = size;
>      pages->iov = g_new0(struct iovec, size);
> +    pages->packet_len = sizeof(MultiFDPacket_t) + sizeof(ram_addr_t) * size;
> +    pages->packet = g_malloc0(pages->packet_len);
>      *ppages = pages;
>  }
>  
> @@ -458,6 +475,9 @@ static void multifd_pages_clear(multifd_pages_t *pages)
>      pages->block = NULL;
>      g_free(pages->iov);
>      pages->iov = NULL;
> +    pages->packet_len = 0;
> +    g_free(pages->packet);
> +    pages->packet = NULL;
>      g_free(pages);
>  }
>  
> @@ -499,7 +519,6 @@ int multifd_save_cleanup(Error **errp)
>  
>          if (p->running) {
>              qemu_thread_join(&p->thread);
> -            p->running = false;
>          }
>          socket_send_channel_destroy(p->c);
>          p->c = NULL;
> @@ -535,7 +554,16 @@ static void multifd_send_sync_main(void)
>          qemu_sem_post(&p->sem);
>      }
>      for (i = 0; i < migrate_multifd_channels(); i++) {
> -        qemu_sem_wait(&multifd_send_state->sem_main);
> +        MultiFDSendParams *p = &multifd_send_state->params[i];
> +        bool wait = true;
> +
> +        qemu_mutex_lock(&p->mutex);
> +        wait = p->running;
> +        qemu_mutex_unlock(&p->mutex);
> +
> +        if (wait) {
> +            qemu_sem_wait(&multifd_send_state->sem_main);
> +        }
>      }
>      trace_multifd_send_sync_main();
>  }
> @@ -575,16 +603,37 @@ static void *multifd_send_thread(void *opaque)
>              continue;
>          }
>          if (p->quit) {
> +            p->running = false;
>              qemu_mutex_unlock(&p->mutex);
>              break;
>          }
>          if (p->pages->used) {
> +            MultiFDPacket_t *packet = p->pages->packet;
> +            Error *local_err = NULL;
> +            size_t ret;
> +
> +            packet->used = p->pages->used;
>              p->pages->used = 0;
>              qemu_mutex_unlock(&p->mutex);
> +            packet->magic = MULTIFD_MAGIC;
> +            packet->version = MULTIFD_VERSION;
>  
> -            trace_multifd_send(p->id, p->pages->seq, p->pages->used);
> -            /* ToDo: send page here */
> -
> +            strncpy(packet->ramblock, p->pages->block->idstr, 256);
> +            packet->size = migrate_multifd_page_count();
> +            packet->seq = p->pages->seq;
> +            ret = qio_channel_write_all(p->c, (void *)packet,
> +                                        p->pages->packet_len, &local_err);
> +            if (ret != 0) {
> +                terminate_multifd_send_threads(local_err);
> +                return NULL;
> +            }
> +            trace_multifd_send(p->id, p->pages->seq, packet->used);
> +            ret = qio_channel_writev_all(p->c, p->pages->iov,
> +                                         packet->used, &local_err);
> +            if (ret != 0) {
> +                terminate_multifd_send_threads(local_err);
> +                return NULL;
> +            }
>              qemu_mutex_lock(&multifd_send_state->mutex);
>              p->done = true;
>              p->packets_sent++;
> @@ -763,7 +812,6 @@ int multifd_load_cleanup(Error **errp)
>  
>          if (p->running) {
>              qemu_thread_join(&p->thread);
> -            p->running = false;
>          }
>          socket_recv_channel_unref(p->c);
>          p->c = NULL;
> @@ -801,17 +849,48 @@ static void multifd_recv_sync_main(void)
>          qemu_sem_post(&p->sem);
>      }
>      for (i = 0; i < migrate_multifd_channels(); i++) {
> -        qemu_sem_wait(&multifd_recv_state->sem_main);
> +        MultiFDRecvParams *p = &multifd_recv_state->params[i];
> +        bool wait = true;
> +
> +        qemu_mutex_lock(&p->mutex);
> +        wait = p->running && !p->quit;
> +        qemu_mutex_unlock(&p->mutex);
> +
> +        if (wait) {
> +            qemu_sem_wait(&multifd_recv_state->sem_main);
> +        }
>      }
>      trace_multifd_recv_sync_main();
>  }
>  
> +static gboolean recv_channel_ready(QIOChannel *ioc,
> +                                   GIOCondition condition,
> +                                   gpointer opaque)
> +{
> +    MultiFDRecvParams *p = opaque;
> +
> +    if (condition != G_IO_IN) {
> +        return G_SOURCE_REMOVE;
> +    }
> +
> +    qemu_mutex_lock(&p->mutex);
> +    p->done = false;
> +    qemu_mutex_unlock(&p->mutex);
> +    qemu_sem_post(&p->sem);
> +
> +    return G_SOURCE_CONTINUE;
> +
> +}
> +
>  static void *multifd_recv_thread(void *opaque)
>  {
>      MultiFDRecvParams *p = opaque;
>  
>      trace_multifd_recv_thread_start(p->id);
>  
> +    qio_channel_add_watch(p->c, G_IO_IN | G_IO_HUP | G_IO_ERR,
> +                          recv_channel_ready, p, NULL);
> +
>      while (true) {
>          qemu_sem_wait(&p->sem);
>          qemu_mutex_lock(&p->mutex);
> @@ -821,14 +900,50 @@ static void *multifd_recv_thread(void *opaque)
>              qemu_sem_post(&multifd_recv_state->sem_main);
>              continue;
>          }
> +        if (!p->done) {
> +            MultiFDPacket_t *packet = p->pages->packet;
> +            RAMBlock *block;
> +            Error *local_err = NULL;
> +            size_t ret;
> +            int i;
> +
> +            qemu_mutex_unlock(&p->mutex);
> +
> +            ret = qio_channel_read_all(p->c, (void *)packet,
> +                                       p->pages->packet_len, &local_err);
> +            if (ret != 0) {
> +                terminate_multifd_recv_threads(local_err);
> +                return NULL;
> +            }
> +            block = qemu_ram_block_by_name(packet->ramblock);
> +            p->pages->seq = packet->seq;
> +            for (i = 0; i < packet->used; i++) {
> +                p->pages->iov[i].iov_base = block->host + packet->offset[i];
> +                p->pages->iov[i].iov_len = TARGET_PAGE_SIZE;
> +            }
> +            trace_multifd_recv(p->id, p->pages->seq, packet->used);
> +            ret = qio_channel_readv_all(p->c, p->pages->iov,
> +                                        packet->used, &local_err);
> +            if (ret != 0) {
> +                terminate_multifd_recv_threads(local_err);
> +                return NULL;
> +            }
> +            qemu_mutex_lock(&p->mutex);
> +            p->done = true;
> +            p->packets_recv++;
> +            qemu_mutex_unlock(&p->mutex);
> +
> +            continue;
> +        }
>          if (p->quit) {
> +            p->running = false;
>              qemu_mutex_unlock(&p->mutex);
>              break;
>          }
>          qemu_mutex_unlock(&p->mutex);
>      }
>  
> -    trace_multifd_recv_thread_end(p->id);
> +    trace_multifd_recv_thread_end(p->id, p->packets_recv);
>      return NULL;
>  }
>  
> @@ -854,6 +969,7 @@ int multifd_load_setup(void)
>          qemu_mutex_init(&p->mutex);
>          qemu_sem_init(&p->sem, 0);
>          p->quit = false;
> +        p->done = true;
>          p->id = i;
>          p->name = g_strdup_printf("multifdrecv_%d", i);
>          multifd_pages_init(&p->pages, migrate_multifd_page_count());
> diff --git a/migration/trace-events b/migration/trace-events
> index f6ab2c7bcb..e9f1aae985 100644
> --- a/migration/trace-events
> +++ b/migration/trace-events
> @@ -82,8 +82,9 @@ multifd_recv_sync_main(void) ""
>  multifd_send_thread_start(int id) "%d"
>  multifd_send_thread_end(char id, uint32_t packets) "channel %d packets %d"
>  multifd_recv_thread_start(int id) "%d"
> -multifd_recv_thread_end(int id) "%d"
> +multifd_recv_thread_end(char id, uint32_t packets) "channel %d packets %d"
>  multifd_send(char id, int seq, int num) "channel %d sequence %d num pages %d"
> +multifd_recv(char id, int seq, int num) "channel %d sequence %d num pages %d"
>  
>  # migration/migration.c
>  await_return_path_close_on_source_close(void) ""
> -- 
> 2.14.3
> 
> 

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 :|

Re: [Qemu-devel] [PATCH v10 10/24] migration: In case of error just end the migration

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> Signed-off-by: Juan Quintela <quintela@redhat.com>

It's probably worth keeping an eye on what happens in the case
of Peter's postcopy recovery where a new incoming connection
happens later.

Dave

> ---
>  migration/socket.c | 6 ++----
>  1 file changed, 2 insertions(+), 4 deletions(-)
> 
> diff --git a/migration/socket.c b/migration/socket.c
> index 08606c665d..b12b0a462e 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -139,9 +139,8 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
>      sioc = qio_channel_socket_accept(QIO_CHANNEL_SOCKET(ioc),
>                                       &err);
>      if (!sioc) {
> -        error_report("could not accept migration connection (%s)",
> -                     error_get_pretty(err));
> -        goto out;
> +        migrate_set_error(migrate_get_current(), err);
> +        return G_SOURCE_REMOVE;
>      }
>  
>      trace_migration_socket_incoming_accepted();
> @@ -150,7 +149,6 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
>      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);
> -- 
> 2.14.3
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v10 10/24] migration: In case of error just end the migration

[Eric Blake]

On 03/07/2018 05:52 AM, Daniel P. Berrangé wrote:
> On Wed, Mar 07, 2018 at 11:59:56AM +0100, Juan Quintela wrote:
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>   migration/socket.c | 6 ++----
>>   1 file changed, 2 insertions(+), 4 deletions(-)
> 
> Reviewed-by: Daniel P. Berrangé <berrange@redhat.com>
> 

> It will only return NULL if a client connected & then went away. This should
> not happen with a "normal" mgmt app usage. On the flip side this allows a
> malicious network attacker to inflict a denial of service on the migration
> by simply connecting to target QEMU & immediately exiting.
> 
> Our "authentication" for migration relies on being able to validate the TLS
> certs during TLS handshake. So in general we ought to allow repeated incoming
> connections until we get a successful handshake.

Indeed, our NBD code had some CVE fixes last year where a rogue 'nc' 
process could cause denial of service by connecting and hanging up 
immediately, until we fixed it to retry until the first client that 
actually got past the handshake.  We don't need to repeat CVEs like that.

> 
> So in fact, I think a better fix here is to simply remove the original
> 'error_report' line, and ensure we return G_SOURCE_CONTINUE to wait for
> another incoming connection from the real mgmt app.
> 

-- 
Eric Blake, Principal Software Engineer
Red Hat, Inc.           +1-919-301-3266
Virtualization:  qemu.org | libvirt.org

Re: [Qemu-devel] [PATCH v10 19/24] migration: Create multifd channels

[Peter Xu]

On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
> In both sides.  We still don't transmit anything through them.

s/In/On/?

> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
>  1 file changed, 42 insertions(+), 10 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index b57d9fd667..7ef0c2b7e2 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -399,6 +399,7 @@ struct MultiFDSendParams {
>      uint8_t id;
>      char *name;
>      QemuThread thread;
> +    QIOChannel *c;
>      QemuSemaphore sem;
>      QemuMutex mutex;
>      bool running;
> @@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
>              qemu_thread_join(&p->thread);
>              p->running = false;
>          }
> +        socket_send_channel_destroy(p->c);
> +        p->c = NULL;
>          qemu_mutex_destroy(&p->mutex);
>          qemu_sem_destroy(&p->sem);
>          g_free(p->name);
> @@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
>      return NULL;
>  }
>  
> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
> +{
> +    MultiFDSendParams *p = opaque;
> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
> +    Error *local_err = NULL;
> +
> +    if (qio_task_propagate_error(task, &local_err)) {
> +        if (multifd_save_cleanup(&local_err) != 0) {

Do we need to call multifd_save_cleanup() explicitly here?

Asked since I saw that it would also be called in migrate_fd_cleanup(),
and it seems that we should call migrate_fd_cleanup() soon too when
this happens?

Otherwise it looks fine to me.  Thanks,

> +            migrate_set_error(migrate_get_current(), local_err);
> +        }
> +    } else {
> +        p->c = QIO_CHANNEL(sioc);
> +        qio_channel_set_delay(p->c, false);
> +        p->running = true;
> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
> +                           QEMU_THREAD_JOINABLE);
> +
> +        atomic_inc(&multifd_send_state->count);
> +    }
> +}

[...]


-- 
Peter Xu

Re: [Qemu-devel] [PATCH v10 20/24] migration: Delay start of migration main routines

[Peter Xu]

On Wed, Mar 07, 2018 at 12:00:06PM +0100, Juan Quintela wrote:
> We need to make sure that we have started all the multifd threads.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/migration.c | 4 ++--
>  migration/migration.h | 1 +
>  migration/ram.c       | 3 +++
>  migration/socket.c    | 3 +++
>  4 files changed, 9 insertions(+), 2 deletions(-)
> 
> diff --git a/migration/migration.c b/migration/migration.c
> index c06c34ca0f..a355618220 100644
> --- a/migration/migration.c
> +++ b/migration/migration.c
> @@ -429,7 +429,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);
> @@ -447,7 +447,7 @@ void migration_ioc_process_incoming(QIOChannel *ioc)
>  
>      if (!mis->from_src_file) {
>          QEMUFile *f = qemu_fopen_channel_input(ioc);
> -        migration_fd_process_incoming(f);
> +        migration_incoming_setup(f);
>          return;
>      }
>      multifd_recv_new_channel(ioc);
> diff --git a/migration/migration.h b/migration/migration.h
> index f40014cf94..03a940831d 100644
> --- a/migration/migration.h
> +++ b/migration/migration.h
> @@ -184,6 +184,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 7ef0c2b7e2..1aab96bd5e 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -743,6 +743,9 @@ void multifd_recv_new_channel(QIOChannel *ioc)
>      qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
>                         QEMU_THREAD_JOINABLE);
>      atomic_inc(&multifd_recv_state->count);
> +    if (multifd_recv_state->count == migrate_multifd_channels()) {
> +        migration_incoming_process();
> +    }
>  }
>  
>  /**
> diff --git a/migration/socket.c b/migration/socket.c
> index b3b5571ebb..deda193de7 100644
> --- a/migration/socket.c
> +++ b/migration/socket.c
> @@ -189,6 +189,9 @@ static gboolean socket_accept_incoming_migration(QIOChannel *ioc,
>      if (migration_has_all_channels()) {
>          /* Close listening socket as its no longer needed */
>          qio_channel_close(ioc, NULL);
> +        if (!migrate_use_multifd()) {
> +            migration_incoming_process();

Calling this in socket code seems a bit odd to me.

Can we do something like this?

void migration_ioc_process_incoming(QIOChannel *ioc)
{
    MigrationIncomingState *mis = migration_incoming_get_current();

    if (!mis->from_src_file) {
        /* This is the main channel */
        QEMUFile *f = qemu_fopen_channel_input(ioc);
        migration_incoming_setup(f);
    } else {
        /* This is one of the multifd channels */
        assert(migrate_use_multifd());
        multifd_recv_new_channel(ioc);
    }

    /*
     * Trigger the migration either if:
     * (1) we are not using multifd, or
     * (2) we have setup all the multifd channels
     */
    if (!migrate_use_multifd() || multifd_recv_all_channels_created()) {
        migration_incoming_process();
    }
}

Then we possibly won't need patch 13 as well.  Thanks,

-- 
Peter Xu

Re: [Qemu-devel] [PATCH v10 08/24] migration: Add multifd test

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> We set the x-multifd-page-count and x-multifd-channels.
> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>


This should probably go nearer the end of the series;
we've also got the problem that things are a bit delicate with TCG so
adding more migration tests probably shouldn't happen until Paolo's
TCG fixes are worked out.

Also, should we be checking for some stats to show all 4 channels were
used?

Reviewed-by: Dr. David Alan Gilbert <dgilbert@redhat.com>

> ---
>  tests/migration-test.c | 50 ++++++++++++++++++++++++++++++++++++++++++++++++++
>  1 file changed, 50 insertions(+)
> 
> diff --git a/tests/migration-test.c b/tests/migration-test.c
> index 6f9b4c8d7a..97d35f979d 100644
> --- a/tests/migration-test.c
> +++ b/tests/migration-test.c
> @@ -755,6 +755,55 @@ static void test_compress_unix(void)
>      g_free(uri);
>  }
>  
> +static void test_multifd_tcp(void)
> +{
> +    char *uri;
> +    char *port;
> +    QTestState *from, *to;
> +
> +    test_migrate_start(&from, &to, "tcp:127.0.0.1:0");
> +
> +    /* We want to pick a speed slow enough that the test completes
> +     * quickly, but that it doesn't complete precopy even on a slow
> +     * machine, so also set the downtime.
> +     */
> +    /* 1 ms should make it not converge*/
> +    migrate_set_parameter(from, "downtime-limit", "1");
> +    /* 1GB/s */
> +    migrate_set_parameter(from, "max-bandwidth", "1000000000");
> +
> +    migrate_set_parameter(from, "x-multifd-channels", "4");
> +    migrate_set_parameter(to, "x-multifd-channels", "4");
> +
> +    migrate_set_parameter(from, "x-multifd-page-count", "64");
> +    migrate_set_parameter(to, "x-multifd-page-count", "64");
> +
> +    migrate_set_capability(from, "x-multifd", "true");
> +    migrate_set_capability(to, "x-multifd", "true");
> +    /* Wait for the first serial output from the source */
> +    wait_for_serial("src_serial");
> +
> +    port = migrate_get_parameter(to, "x-tcp-port");
> +    uri = g_strdup_printf("tcp:127.0.0.1:%s", port);
> +
> +    migrate(from, uri);
> +
> +    wait_for_migration_pass(from);
> +
> +    /* 300ms it should converge */
> +    migrate_set_parameter(from, "downtime-limit", "300");
> +
> +    if (!got_stop) {
> +        qtest_qmp_eventwait(from, "STOP");
> +    }
> +    qtest_qmp_eventwait(to, "RESUME");
> +
> +    wait_for_serial("dest_serial");
> +    wait_for_migration_complete(from);
> +
> +    test_migrate_end(from, to, true);
> +}
> +
>  int main(int argc, char **argv)
>  {
>      char template[] = "/tmp/migration-test-XXXXXX";
> @@ -783,6 +832,7 @@ int main(int argc, char **argv)
>      if (0) {
>          qtest_add_func("/migration/compress/unix", test_compress_unix);
>      }
> +    qtest_add_func("/migration/multifd/tcp", test_multifd_tcp);
>  
>      ret = g_test_run();
>  
> -- 
> 2.14.3
> 
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v10 03/24] migration: Create tcp_port parameter

[Juan Quintela]

Daniel P. Berrange <berrange@redhat.com> wrote:
> On Wed, Mar 07, 2018 at 11:59:49AM +0100, Juan Quintela wrote:
>> It will be used to store the uri tcp_port parameter.  This is the only
>> parameter than can change and we can need to be able to connect to it.
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> 
>> --
>>  # TODO either fuse back into MigrationParameters, or make
>> @@ -584,7 +591,8 @@
>>              '*block-incremental': 'bool',
>>              '*x-multifd-channels': 'int',
>>              '*x-multifd-page-count': 'int',
>> -            '*xbzrle-cache-size': 'size' } }
>> +            '*xbzrle-cache-size': 'size',
>> +            '*x-tcp-port': 'uint16'} }
>
> This should not exist - this exposes this parameter in migate-set-parameters
> as a end user settable property, which is definitely not desirable.
>
> It is only something we should report with 'query-migrate' / 'info migrate'

Oops, my understanding was that the three places have to be in sync.
Now I stand corrected.  Thanks.


>
>>  # @migrate-set-parameters:
>> @@ -667,6 +675,10 @@
>>  #                     needs to be a multiple of the target page size
>>  #                     and a power of 2
>>  #                     (Since 2.11)
>> +#
>> +# @x-tcp-port: Only used for tcp, to know what the real port is
>> +#                     (Since 2.12)
>> +#
>>  # Since: 2.4
>>  ##
>>  { 'struct': 'MigrationParameters',
>> @@ -683,7 +695,8 @@
>>              '*block-incremental': 'bool' ,
>>              '*x-multifd-channels': 'uint8',
>>              '*x-multifd-page-count': 'uint32',
>> -            '*xbzrle-cache-size': 'size' } }
>> +            '*xbzrle-cache-size': 'size',
>> +            '*x-tcp-port': 'uint16'} }
>
> As mentioned in previous review, IMHO we should be reporting the full
> socket address, and allow an array of them, since we're going to have
> more than one address available. i.e.
>
>    '*socket-address': ['SocketAddress']

This is weird, really weird.  But was done.

- this needs to be copied, because it is a pointer, so we end needing
  QAPI_CLONE(), yes, I know.
- it is really weird that I have to:

    rsp_return = qdict_get_qdict(rsp, "return");
    object = qdict_get(rsp_return, parameter);

    iv = qobject_input_visitor_new(object);
    visit_type_SocketAddress(iv, NULL, &saddr, &local_err);
    result = g_strdup_printf("%s",
                             SocketAddress_to_str("", saddr, false, false));
    QDECREF(rsp);

  Remember, it is a *series* of strings in the ddict, we have code to
  _print_ qdicts, as info migrate works.  But I haven't found an easier
  way of getting from a qdict to an string than:
    * parsing it
    * convert it back to text
  sniff

> It doesn't cover non-socket based URIs, but that's fine, because for those
> the mgmt app already knows how the channel is setup. We just need the
> array of SocketAddress, because for socket URIs, the hostname, gets turned
> into an array of addresses, and the mgmt app can't discover them.

SocketAddress are a mess, there is not a single example that I can find
on how to use it on the whole tree.  I *think* that I did that right,
will see your comments on the next post.

Later, Juan.

Re: [Qemu-devel] [PATCH v10 04/24] migration: Set the migration tcp port

[Juan Quintela]

Daniel P. Berrange <berrange@redhat.com> wrote:
> On Wed, Mar 07, 2018 at 11:59:50AM +0100, Juan Quintela wrote:
>> We can set the port parameter as zero.  This patch lets us know what
>> port the system was choosen for us.  Now we can migrate to this place.
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>

>> +void migrate_set_port(const uint16_t port, Error **errp)
>> +{
>> +    MigrateSetParameters p = {
>> +        .has_x_tcp_port = true,
>> +        .x_tcp_port = port,
>> +    };
>> +
>> +    qmp_migrate_set_parameters(&p, errp);
>> +}
>
> This is really not nice - it is requiring the QMP  'migrate-set-parameters'
> command to accept an extra field that is never something we want the end
> user to be allowed to set. We should not use the public QMP schema for
> data items we are just passing between 2 internal pieces of code.

void migrate_set_address(SocketAddress *address)
{
    MigrationState *s = migrate_get_current();

    s->parameters.has_x_socket_address = true;
    s->parameters.x_socket_address = address;
}


I hope that is ok with you O:-)

Later, Juan.

PD.  Yes, I agree about not using QMP inside two pieces of code, but on
     the other hand, it make this so *future* proof O:-)

Re: [Qemu-devel] [PATCH v10 08/24] migration: Add multifd test

[Juan Quintela]

"Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> * Juan Quintela (quintela@redhat.com) wrote:
>> We set the x-multifd-page-count and x-multifd-channels.
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>
>
> This should probably go nearer the end of the series;

it is _much_ better here.  It makes so much easier to test that I don't
break neither migration nor multifd while developing O:-)

> we've also got the problem that things are a bit delicate with TCG so
> adding more migration tests probably shouldn't happen until Paolo's
> TCG fixes are worked out.



> Also, should we be checking for some stats to show all 4 channels were
> used?

There are traces now, I can add new counters is that is what you want.

Later, Juan.

Re: [Qemu-devel] [PATCH v10 08/24] migration: Add multifd test

[Dr. David Alan Gilbert]

* Juan Quintela (quintela@redhat.com) wrote:
> "Dr. David Alan Gilbert" <dgilbert@redhat.com> wrote:
> > * Juan Quintela (quintela@redhat.com) wrote:
> >> We set the x-multifd-page-count and x-multifd-channels.
> >> 
> >> Signed-off-by: Juan Quintela <quintela@redhat.com>
> >
> >
> > This should probably go nearer the end of the series;
> 
> it is _much_ better here.  It makes so much easier to test that I don't
> break neither migration nor multifd while developing O:-)

OK, not a biggy.

> > we've also got the problem that things are a bit delicate with TCG so
> > adding more migration tests probably shouldn't happen until Paolo's
> > TCG fixes are worked out.
> 
> 
> 
> > Also, should we be checking for some stats to show all 4 channels were
> > used?
> 
> There are traces now, I can add new counters is that is what you want.

If it's easy then it's worth it.

Dave

> Later, Juan.
--
Dr. David Alan Gilbert / dgilbert@redhat.com / Manchester, UK

Re: [Qemu-devel] [PATCH v10 19/24] migration: Create multifd channels

[Juan Quintela]

Peter Xu <peterx@redhat.com> wrote:
> On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
>> In both sides.  We still don't transmit anything through them.
>
> s/In/On/?
>
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
>>  1 file changed, 42 insertions(+), 10 deletions(-)
>> 
>> diff --git a/migration/ram.c b/migration/ram.c
>> index b57d9fd667..7ef0c2b7e2 100644
>> --- a/migration/ram.c
>> +++ b/migration/ram.c
>> @@ -399,6 +399,7 @@ struct MultiFDSendParams {
>>      uint8_t id;
>>      char *name;
>>      QemuThread thread;
>> +    QIOChannel *c;
>>      QemuSemaphore sem;
>>      QemuMutex mutex;
>>      bool running;
>> @@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
>>              qemu_thread_join(&p->thread);
>>              p->running = false;
>>          }
>> +        socket_send_channel_destroy(p->c);
>> +        p->c = NULL;
>>          qemu_mutex_destroy(&p->mutex);
>>          qemu_sem_destroy(&p->sem);
>>          g_free(p->name);
>> @@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
>>      return NULL;
>>  }
>>  
>> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
>> +{
>> +    MultiFDSendParams *p = opaque;
>> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
>> +    Error *local_err = NULL;
>> +
>> +    if (qio_task_propagate_error(task, &local_err)) {
>> +        if (multifd_save_cleanup(&local_err) != 0) {
>
> Do we need to call multifd_save_cleanup() explicitly here?

Is the easiest way of stoping all multifd threads, no?

> Asked since I saw that it would also be called in migrate_fd_cleanup(),
> and it seems that we should call migrate_fd_cleanup() soon too when
> this happens?

We need to stop migraiton. thtat migrate_set_error() is only used for
reporting in info migrate, it is not acted upon.

Yes, perhaps it should, but as it is, it is not.  So, I think it is
right O:-)

Later, Juan.

> Otherwise it looks fine to me.  Thanks,
>
>> +            migrate_set_error(migrate_get_current(), local_err);
>> +        }
>> +    } else {
>> +        p->c = QIO_CHANNEL(sioc);
>> +        qio_channel_set_delay(p->c, false);
>> +        p->running = true;
>> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
>> +                           QEMU_THREAD_JOINABLE);
>> +
>> +        atomic_inc(&multifd_send_state->count);
>> +    }
>> +}
>
> [...]

Re: [Qemu-devel] [PATCH v10 19/24] migration: Create multifd channels

[Peter Xu]

On Thu, Mar 15, 2018 at 01:57:54PM +0100, Juan Quintela wrote:
> Peter Xu <peterx@redhat.com> wrote:
> > On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
> >> In both sides.  We still don't transmit anything through them.
> >
> > s/In/On/?
> >
> >> 
> >> Signed-off-by: Juan Quintela <quintela@redhat.com>
> >> ---
> >>  migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
> >>  1 file changed, 42 insertions(+), 10 deletions(-)
> >> 
> >> diff --git a/migration/ram.c b/migration/ram.c
> >> index b57d9fd667..7ef0c2b7e2 100644
> >> --- a/migration/ram.c
> >> +++ b/migration/ram.c
> >> @@ -399,6 +399,7 @@ struct MultiFDSendParams {
> >>      uint8_t id;
> >>      char *name;
> >>      QemuThread thread;
> >> +    QIOChannel *c;
> >>      QemuSemaphore sem;
> >>      QemuMutex mutex;
> >>      bool running;
> >> @@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
> >>              qemu_thread_join(&p->thread);
> >>              p->running = false;
> >>          }
> >> +        socket_send_channel_destroy(p->c);
> >> +        p->c = NULL;
> >>          qemu_mutex_destroy(&p->mutex);
> >>          qemu_sem_destroy(&p->sem);
> >>          g_free(p->name);
> >> @@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
> >>      return NULL;
> >>  }
> >>  
> >> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
> >> +{
> >> +    MultiFDSendParams *p = opaque;
> >> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
> >> +    Error *local_err = NULL;
> >> +
> >> +    if (qio_task_propagate_error(task, &local_err)) {
> >> +        if (multifd_save_cleanup(&local_err) != 0) {
> >
> > Do we need to call multifd_save_cleanup() explicitly here?
> 
> Is the easiest way of stoping all multifd threads, no?

Yeah, but again, I thought it would be called later too, since...

> 
> > Asked since I saw that it would also be called in migrate_fd_cleanup(),
> > and it seems that we should call migrate_fd_cleanup() soon too when
> > this happens?
> 
> We need to stop migraiton. thtat migrate_set_error() is only used for
> reporting in info migrate, it is not acted upon.
> 
> Yes, perhaps it should, but as it is, it is not.  So, I think it is
> right O:-)

... after Dave's 688a3dcba9 ("migration: Route errors down through
migration_channel_connect", 2018-02-06), all these channel errors
should finally be routed to migrate_fd_connect(), and in that we have:

void migrate_fd_connect(MigrationState *s, Error *error_in)
{
    s->expected_downtime = s->parameters.downtime_limit;
    s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
    if (error_in) {
        migrate_fd_error(s, error_in);
        migrate_fd_cleanup(s);
        return;
    }
    ...
}

Then, in migrate_fd_cleanup() we have multifd_save_cleanup().  That's
why I thought we can skip the cleanup here since after all we'll do it
in other places (and we can keep the cleanup code unified).

Thanks,

> 
> Later, Juan.
> 
> > Otherwise it looks fine to me.  Thanks,
> >
> >> +            migrate_set_error(migrate_get_current(), local_err);
> >> +        }
> >> +    } else {
> >> +        p->c = QIO_CHANNEL(sioc);
> >> +        qio_channel_set_delay(p->c, false);
> >> +        p->running = true;
> >> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
> >> +                           QEMU_THREAD_JOINABLE);
> >> +
> >> +        atomic_inc(&multifd_send_state->count);
> >> +    }
> >> +}
> >
> > [...]

-- 
Peter Xu

Re: [Qemu-devel] [PATCH v10 19/24] migration: Create multifd channels

[Juan Quintela]

Peter Xu <peterx@redhat.com> wrote:
> On Thu, Mar 15, 2018 at 01:57:54PM +0100, Juan Quintela wrote:
>> Peter Xu <peterx@redhat.com> wrote:
>> > On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
>> >> In both sides.  We still don't transmit anything through them.
>> >
>> > s/In/On/?

>> >> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
>> >> +{
>> >> +    MultiFDSendParams *p = opaque;
>> >> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
>> >> +    Error *local_err = NULL;
>> >> +
>> >> +    if (qio_task_propagate_error(task, &local_err)) {
>> >> +        if (multifd_save_cleanup(&local_err) != 0) {
>> >
>> > Do we need to call multifd_save_cleanup() explicitly here?
>> 
>> Is the easiest way of stoping all multifd threads, no?
>
> Yeah, but again, I thought it would be called later too, since...

But we are not stopping the threads.  Only if you preffer to call there
terminate_multifd_send_threads().  Probably it is better to call
terminate_multifd_send_threads(), it is what we do on the rest of
errors.

Changing it.

>> > Asked since I saw that it would also be called in migrate_fd_cleanup(),
>> > and it seems that we should call migrate_fd_cleanup() soon too when
>> > this happens?
>> 
>> We need to stop migraiton. thtat migrate_set_error() is only used for
>> reporting in info migrate, it is not acted upon.
>> 
>> Yes, perhaps it should, but as it is, it is not.  So, I think it is
>> right O:-)
>
> ... after Dave's 688a3dcba9 ("migration: Route errors down through
> migration_channel_connect", 2018-02-06), all these channel errors
> should finally be routed to migrate_fd_connect(), and in that we have:
>
> void migrate_fd_connect(MigrationState *s, Error *error_in)
> {
>     s->expected_downtime = s->parameters.downtime_limit;
>     s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
>     if (error_in) {
>         migrate_fd_error(s, error_in);
>         migrate_fd_cleanup(s);
>         return;
>     }
>     ...
> }
>
> Then, in migrate_fd_cleanup() we have multifd_save_cleanup().  That's
> why I thought we can skip the cleanup here since after all we'll do it
> in other places (and we can keep the cleanup code unified).

Ok, it is better, changing to terminate_multifd_send_threads.

Thanks, Juan.