[Qemu-devel] [PATCH 0/4] Asynchronous tcp incoming migration.

[Qemu-devel] [PATCH 0/4] Asynchronous tcp incoming migration.

[Yoshiaki Tamura]

Hi,

This series add asynchronous tcp incoming migration.  Currently, tcp
migration on incoming side is blocked when outgoing has started on the
remote side, and you can't do anything.  With this series you can get
info of incoming migration by calling "info migrate" like on outgoing
side.

This series is based on discussion below.

http://permalink.gmane.org/gmane.comp.emulators.qemu/72401

Yoshiaki Tamura (4):
  savevm: refactor qemu_loadvm_state().
  migration-tcp: add support for asynchronous incoming migration.
  arch_init: calculate transferred bytes at ram_load().
  migration: add support to print migration info on incoming side.

 arch_init.c     |    6 ++-
 migration-tcp.c |  112 +++++++++++++++++++++---------
 migration.c     |   18 ++++-
 migration.h     |    2 +-
 savevm.c        |  205 +++++++++++++++++++++++++++++++++++++-----------------
 sysemu.h        |    2 +
 6 files changed, 242 insertions(+), 103 deletions(-)

[Qemu-devel] [PATCH 1/4] savevm: refactor qemu_loadvm_state().

[Yoshiaki Tamura]

Split qemu_loadvm_state(), and introduce
qemu_loadvm_state_{begin,iterate,complete,async}.
qemu_loadvm_state_async() is a function to handle a single incoming
section.

Signed-off-by: Yoshiaki Tamura <tamura.yoshiaki@lab.ntt.co.jp>
---
 savevm.c |  206 +++++++++++++++++++++++++++++++++++++++++++-------------------
 sysemu.h |    2 +
 2 files changed, 146 insertions(+), 62 deletions(-)

diff --git a/savevm.c b/savevm.c
index dc20390..aa4f98c 100644
--- a/savevm.c
+++ b/savevm.c
@@ -1005,6 +1005,8 @@ typedef struct SaveStateEntry {
 
 static QTAILQ_HEAD(savevm_handlers, SaveStateEntry) savevm_handlers =
     QTAILQ_HEAD_INITIALIZER(savevm_handlers);
+static QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
+    QLIST_HEAD_INITIALIZER(loadvm_handlers);
 static int global_section_id;
 
 static int calculate_new_instance_id(const char *idstr)
@@ -1460,14 +1462,9 @@ typedef struct LoadStateEntry {
     int version_id;
 } LoadStateEntry;
 
-int qemu_loadvm_state(QEMUFile *f)
+int qemu_loadvm_state_begin(QEMUFile *f)
 {
-    QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
-        QLIST_HEAD_INITIALIZER(loadvm_handlers);
-    LoadStateEntry *le, *new_le;
-    uint8_t section_type;
     unsigned int v;
-    int ret;
 
     v = qemu_get_be32(f);
     if (v != QEMU_VM_FILE_MAGIC)
@@ -1481,73 +1478,157 @@ int qemu_loadvm_state(QEMUFile *f)
     if (v != QEMU_VM_FILE_VERSION)
         return -ENOTSUP;
 
-    while ((section_type = qemu_get_byte(f)) != QEMU_VM_EOF) {
-        uint32_t instance_id, version_id, section_id;
-        SaveStateEntry *se;
-        char idstr[257];
-        int len;
+    return 0;
+}
 
-        switch (section_type) {
-        case QEMU_VM_SECTION_START:
-        case QEMU_VM_SECTION_FULL:
-            /* Read section start */
-            section_id = qemu_get_be32(f);
-            len = qemu_get_byte(f);
-            qemu_get_buffer(f, (uint8_t *)idstr, len);
-            idstr[len] = 0;
-            instance_id = qemu_get_be32(f);
-            version_id = qemu_get_be32(f);
-
-            /* Find savevm section */
-            se = find_se(idstr, instance_id);
-            if (se == NULL) {
-                fprintf(stderr, "Unknown savevm section or instance '%s' %d\n", idstr, instance_id);
-                ret = -EINVAL;
-                goto out;
-            }
+static int qemu_loadvm_state_iterate(QEMUFile *f)
+{
+    LoadStateEntry *le;
+    uint32_t section_id;
+    int ret;
 
-            /* Validate version */
-            if (version_id > se->version_id) {
-                fprintf(stderr, "savevm: unsupported version %d for '%s' v%d\n",
-                        version_id, idstr, se->version_id);
-                ret = -EINVAL;
-                goto out;
-            }
+    section_id = qemu_get_be32(f);
+
+    QLIST_FOREACH(le, &loadvm_handlers, entry) {
+        if (le->section_id == section_id) {
+            break;
+        }
+    }
+    if (le == NULL) {
+        fprintf(stderr, "Unknown savevm section %d\n", section_id);
+        return -EINVAL;
+    }
+
+    ret = vmstate_load(f, le->se, le->version_id);
+    if (ret < 0) {
+        fprintf(stderr, "qemu: warning: error while loading state section id %d\n",
+                section_id);
+        return ret;
+    }
+
+    return 0;
+}
+
+static int qemu_loadvm_state_complete(QEMUFile *f)
+{
+    LoadStateEntry *le;
+    uint32_t instance_id, version_id, section_id;
+    SaveStateEntry *se;
+    char idstr[257];
+    int ret = -1, len;
+
+    /* Read section start */
+    section_id = qemu_get_be32(f);
+    len = qemu_get_byte(f);
+    qemu_get_buffer(f, (uint8_t *)idstr, len);
+    idstr[len] = 0;
+    instance_id = qemu_get_be32(f);
+    version_id = qemu_get_be32(f);
+
+    /* Find savevm section */
+    se = find_se(idstr, instance_id);
+    if (se == NULL) {
+        fprintf(stderr, "Unknown savevm section or instance '%s' %d\n", idstr, instance_id);
+        return -EINVAL;
+    }
+
+    /* Validate version */
+    if (version_id > se->version_id) {
+        fprintf(stderr, "savevm: unsupported version %d for '%s' v%d\n",
+                version_id, idstr, se->version_id);
+        return -EINVAL;
+    }
+
+    /* Add entry */
+    le = qemu_mallocz(sizeof(*le));
+
+    le->se = se;
+    le->section_id = section_id;
+    le->version_id = version_id;
+    QLIST_INSERT_HEAD(&loadvm_handlers, le, entry);
+
+    ret = vmstate_load(f, le->se, le->version_id);
+    if (ret < 0) {
+        fprintf(stderr, "qemu: warning: error while loading state for instance 0x%x of device '%s'\n",
+                instance_id, idstr);
+        return ret;
+    }
+
+    return 0;
+}
 
-            /* Add entry */
-            le = qemu_mallocz(sizeof(*le));
+int qemu_loadvm_state_async(QEMUFile *f)
+{
+    LoadStateEntry *le, *new_le;
+    int ret;
+    uint8_t section_type;
 
-            le->se = se;
-            le->section_id = section_id;
-            le->version_id = version_id;
-            QLIST_INSERT_HEAD(&loadvm_handlers, le, entry);
+    section_type = qemu_get_byte(f);
+    switch (section_type) {
+    case QEMU_VM_SECTION_START:
+    case QEMU_VM_SECTION_FULL:
+        ret = qemu_loadvm_state_complete(f);
+        if (ret < 0) {
+            goto out_free;
+        }
+        break;
+    case QEMU_VM_SECTION_PART:
+    case QEMU_VM_SECTION_END:
+        ret = qemu_loadvm_state_iterate(f);
+        if (ret < 0) {
+            goto out_free;
+        }
+        break;
+    case QEMU_VM_EOF:
+        cpu_synchronize_all_post_init();
+        break;
+    default:
+        fprintf(stderr, "Unknown savevm section type %d\n", section_type);
+        ret = -EINVAL;
+        goto out_free;
+    }
+ 
+    ret = section_type;
+    if (ret != QEMU_VM_EOF) {
+        goto out;
+    }
 
-            ret = vmstate_load(f, le->se, le->version_id);
+out_free:
+    QLIST_FOREACH_SAFE(le, &loadvm_handlers, entry, new_le) {
+        QLIST_REMOVE(le, entry);
+        qemu_free(le);
+    }
+out:
+    if (qemu_file_has_error(f)) {
+        ret = -EIO;
+    }
+
+    return ret;
+}
+
+int qemu_loadvm_state(QEMUFile *f)
+{
+    LoadStateEntry *le, *new_le;
+    int ret;
+    uint8_t section_type;
+
+    ret = qemu_loadvm_state_begin(f);
+    if (ret < 0)
+        goto out;
+
+    while ((section_type = qemu_get_byte(f)) != QEMU_VM_EOF) {
+        switch (section_type) {
+        case QEMU_VM_SECTION_START:
+        case QEMU_VM_SECTION_FULL:
+            ret = qemu_loadvm_state_complete(f);
             if (ret < 0) {
-                fprintf(stderr, "qemu: warning: error while loading state for instance 0x%x of device '%s'\n",
-                        instance_id, idstr);
                 goto out;
             }
             break;
         case QEMU_VM_SECTION_PART:
         case QEMU_VM_SECTION_END:
-            section_id = qemu_get_be32(f);
-
-            QLIST_FOREACH(le, &loadvm_handlers, entry) {
-                if (le->section_id == section_id) {
-                    break;
-                }
-            }
-            if (le == NULL) {
-                fprintf(stderr, "Unknown savevm section %d\n", section_id);
-                ret = -EINVAL;
-                goto out;
-            }
-
-            ret = vmstate_load(f, le->se, le->version_id);
+            ret = qemu_loadvm_state_iterate(f);
             if (ret < 0) {
-                fprintf(stderr, "qemu: warning: error while loading state section id %d\n",
-                        section_id);
                 goto out;
             }
             break;
@@ -1568,8 +1649,9 @@ out:
         qemu_free(le);
     }
 
-    if (qemu_file_has_error(f))
+    if (qemu_file_has_error(f)) {
         ret = -EIO;
+    }
 
     return ret;
 }
diff --git a/sysemu.h b/sysemu.h
index 879446a..c576840 100644
--- a/sysemu.h
+++ b/sysemu.h
@@ -69,6 +69,8 @@ int qemu_savevm_state_begin(Monitor *mon, QEMUFile *f, int blk_enable,
 int qemu_savevm_state_iterate(Monitor *mon, QEMUFile *f);
 int qemu_savevm_state_complete(Monitor *mon, QEMUFile *f);
 void qemu_savevm_state_cancel(Monitor *mon, QEMUFile *f);
+int qemu_loadvm_state_begin(QEMUFile *f);
+int qemu_loadvm_state_async(QEMUFile *f);
 int qemu_loadvm_state(QEMUFile *f);
 
 #ifdef _WIN32
-- 
1.7.0.31.g1df487

Re: [Qemu-devel] [PATCH 1/4] savevm: refactor qemu_loadvm_state().

[Anthony Liguori]

On 06/03/2010 02:22 AM, Yoshiaki Tamura wrote:
> Split qemu_loadvm_state(), and introduce
> qemu_loadvm_state_{begin,iterate,complete,async}.
> qemu_loadvm_state_async() is a function to handle a single incoming
> section.
>
> Signed-off-by: Yoshiaki Tamura<tamura.yoshiaki@lab.ntt.co.jp>
> ---
>   savevm.c |  206 +++++++++++++++++++++++++++++++++++++++++++-------------------
>   sysemu.h |    2 +
>   2 files changed, 146 insertions(+), 62 deletions(-)
>
> diff --git a/savevm.c b/savevm.c
> index dc20390..aa4f98c 100644
> --- a/savevm.c
> +++ b/savevm.c
> @@ -1005,6 +1005,8 @@ typedef struct SaveStateEntry {
>
>   static QTAILQ_HEAD(savevm_handlers, SaveStateEntry) savevm_handlers =
>       QTAILQ_HEAD_INITIALIZER(savevm_handlers);
> +static QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
> +    QLIST_HEAD_INITIALIZER(loadvm_handlers);
>   static int global_section_id;
>
>   static int calculate_new_instance_id(const char *idstr)
> @@ -1460,14 +1462,9 @@ typedef struct LoadStateEntry {
>       int version_id;
>   } LoadStateEntry;
>
> -int qemu_loadvm_state(QEMUFile *f)
> +int qemu_loadvm_state_begin(QEMUFile *f)
>   {
> -    QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
> -        QLIST_HEAD_INITIALIZER(loadvm_handlers);
> -    LoadStateEntry *le, *new_le;
> -    uint8_t section_type;
>       unsigned int v;
> -    int ret;
>
>       v = qemu_get_be32(f);
>       if (v != QEMU_VM_FILE_MAGIC)
> @@ -1481,73 +1478,157 @@ int qemu_loadvm_state(QEMUFile *f)
>       if (v != QEMU_VM_FILE_VERSION)
>           return -ENOTSUP;
>
> -    while ((section_type = qemu_get_byte(f)) != QEMU_VM_EOF) {
> -        uint32_t instance_id, version_id, section_id;
> -        SaveStateEntry *se;
> -        char idstr[257];
> -        int len;
> +    return 0;
> +}
>
> -        switch (section_type) {
> -        case QEMU_VM_SECTION_START:
> -        case QEMU_VM_SECTION_FULL:
> -            /* Read section start */
> -            section_id = qemu_get_be32(f);
> -            len = qemu_get_byte(f);
> -            qemu_get_buffer(f, (uint8_t *)idstr, len);
> -            idstr[len] = 0;
> -            instance_id = qemu_get_be32(f);
> -            version_id = qemu_get_be32(f);
> -
> -            /* Find savevm section */
> -            se = find_se(idstr, instance_id);
> -            if (se == NULL) {
> -                fprintf(stderr, "Unknown savevm section or instance '%s' %d\n", idstr, instance_id);
> -                ret = -EINVAL;
> -                goto out;
> -            }
> +static int qemu_loadvm_state_iterate(QEMUFile *f)
> +{
> +    LoadStateEntry *le;
> +    uint32_t section_id;
> +    int ret;
>
> -            /* Validate version */
> -            if (version_id>  se->version_id) {
> -                fprintf(stderr, "savevm: unsupported version %d for '%s' v%d\n",
> -                        version_id, idstr, se->version_id);
> -                ret = -EINVAL;
> -                goto out;
> -            }
> +    section_id = qemu_get_be32(f);
>    

But we're still blocking in qemu_get_be32() so I don't see how this 
makes it async.

Regards,

Anthony Liguori

Re: [Qemu-devel] [PATCH 1/4] savevm: refactor qemu_loadvm_state().

[Yoshiaki Tamura]

Anthony Liguori wrote:
> On 06/03/2010 02:22 AM, Yoshiaki Tamura wrote:
>> Split qemu_loadvm_state(), and introduce
>> qemu_loadvm_state_{begin,iterate,complete,async}.
>> qemu_loadvm_state_async() is a function to handle a single incoming
>> section.
>>
>> Signed-off-by: Yoshiaki Tamura<tamura.yoshiaki@lab.ntt.co.jp>
>> ---
>> savevm.c | 206
>> +++++++++++++++++++++++++++++++++++++++++++-------------------
>> sysemu.h | 2 +
>> 2 files changed, 146 insertions(+), 62 deletions(-)
>>
>> diff --git a/savevm.c b/savevm.c
>> index dc20390..aa4f98c 100644
>> --- a/savevm.c
>> +++ b/savevm.c
>> @@ -1005,6 +1005,8 @@ typedef struct SaveStateEntry {
>>
>> static QTAILQ_HEAD(savevm_handlers, SaveStateEntry) savevm_handlers =
>> QTAILQ_HEAD_INITIALIZER(savevm_handlers);
>> +static QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
>> + QLIST_HEAD_INITIALIZER(loadvm_handlers);
>> static int global_section_id;
>>
>> static int calculate_new_instance_id(const char *idstr)
>> @@ -1460,14 +1462,9 @@ typedef struct LoadStateEntry {
>> int version_id;
>> } LoadStateEntry;
>>
>> -int qemu_loadvm_state(QEMUFile *f)
>> +int qemu_loadvm_state_begin(QEMUFile *f)
>> {
>> - QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
>> - QLIST_HEAD_INITIALIZER(loadvm_handlers);
>> - LoadStateEntry *le, *new_le;
>> - uint8_t section_type;
>> unsigned int v;
>> - int ret;
>>
>> v = qemu_get_be32(f);
>> if (v != QEMU_VM_FILE_MAGIC)
>> @@ -1481,73 +1478,157 @@ int qemu_loadvm_state(QEMUFile *f)
>> if (v != QEMU_VM_FILE_VERSION)
>> return -ENOTSUP;
>>
>> - while ((section_type = qemu_get_byte(f)) != QEMU_VM_EOF) {
>> - uint32_t instance_id, version_id, section_id;
>> - SaveStateEntry *se;
>> - char idstr[257];
>> - int len;
>> + return 0;
>> +}
>>
>> - switch (section_type) {
>> - case QEMU_VM_SECTION_START:
>> - case QEMU_VM_SECTION_FULL:
>> - /* Read section start */
>> - section_id = qemu_get_be32(f);
>> - len = qemu_get_byte(f);
>> - qemu_get_buffer(f, (uint8_t *)idstr, len);
>> - idstr[len] = 0;
>> - instance_id = qemu_get_be32(f);
>> - version_id = qemu_get_be32(f);
>> -
>> - /* Find savevm section */
>> - se = find_se(idstr, instance_id);
>> - if (se == NULL) {
>> - fprintf(stderr, "Unknown savevm section or instance '%s' %d\n",
>> idstr, instance_id);
>> - ret = -EINVAL;
>> - goto out;
>> - }
>> +static int qemu_loadvm_state_iterate(QEMUFile *f)
>> +{
>> + LoadStateEntry *le;
>> + uint32_t section_id;
>> + int ret;
>>
>> - /* Validate version */
>> - if (version_id> se->version_id) {
>> - fprintf(stderr, "savevm: unsupported version %d for '%s' v%d\n",
>> - version_id, idstr, se->version_id);
>> - ret = -EINVAL;
>> - goto out;
>> - }
>> + section_id = qemu_get_be32(f);
>
> But we're still blocking in qemu_get_be32() so I don't see how this
> makes it async.

In that sense, it's not completely async, but still better than being in the 
while loop that doesn't accept any commands on the incoming side.
To make things completely async, I think we should solve the lock issue and add 
a thread for incoming migration, but I don't think that's your opinion.

Did I misunderstand your comment for the previous approach?

Thanks,

Yoshi

>
> Regards,
>
> Anthony Liguori
>
>
>
>

Re: [Qemu-devel] [PATCH 1/4] savevm: refactor qemu_loadvm_state().

[Anthony Liguori]

On 06/14/2010 09:01 PM, Yoshiaki Tamura wrote:
> Anthony Liguori wrote:
>> On 06/03/2010 02:22 AM, Yoshiaki Tamura wrote:
>>> Split qemu_loadvm_state(), and introduce
>>> qemu_loadvm_state_{begin,iterate,complete,async}.
>>> qemu_loadvm_state_async() is a function to handle a single incoming
>>> section.
>>>
>>> Signed-off-by: Yoshiaki Tamura<tamura.yoshiaki@lab.ntt.co.jp>
>>> ---
>>> savevm.c | 206
>>> +++++++++++++++++++++++++++++++++++++++++++-------------------
>>> sysemu.h | 2 +
>>> 2 files changed, 146 insertions(+), 62 deletions(-)
>>>
>>> diff --git a/savevm.c b/savevm.c
>>> index dc20390..aa4f98c 100644
>>> --- a/savevm.c
>>> +++ b/savevm.c
>>> @@ -1005,6 +1005,8 @@ typedef struct SaveStateEntry {
>>>
>>> static QTAILQ_HEAD(savevm_handlers, SaveStateEntry) savevm_handlers =
>>> QTAILQ_HEAD_INITIALIZER(savevm_handlers);
>>> +static QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
>>> + QLIST_HEAD_INITIALIZER(loadvm_handlers);
>>> static int global_section_id;
>>>
>>> static int calculate_new_instance_id(const char *idstr)
>>> @@ -1460,14 +1462,9 @@ typedef struct LoadStateEntry {
>>> int version_id;
>>> } LoadStateEntry;
>>>
>>> -int qemu_loadvm_state(QEMUFile *f)
>>> +int qemu_loadvm_state_begin(QEMUFile *f)
>>> {
>>> - QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
>>> - QLIST_HEAD_INITIALIZER(loadvm_handlers);
>>> - LoadStateEntry *le, *new_le;
>>> - uint8_t section_type;
>>> unsigned int v;
>>> - int ret;
>>>
>>> v = qemu_get_be32(f);
>>> if (v != QEMU_VM_FILE_MAGIC)
>>> @@ -1481,73 +1478,157 @@ int qemu_loadvm_state(QEMUFile *f)
>>> if (v != QEMU_VM_FILE_VERSION)
>>> return -ENOTSUP;
>>>
>>> - while ((section_type = qemu_get_byte(f)) != QEMU_VM_EOF) {
>>> - uint32_t instance_id, version_id, section_id;
>>> - SaveStateEntry *se;
>>> - char idstr[257];
>>> - int len;
>>> + return 0;
>>> +}
>>>
>>> - switch (section_type) {
>>> - case QEMU_VM_SECTION_START:
>>> - case QEMU_VM_SECTION_FULL:
>>> - /* Read section start */
>>> - section_id = qemu_get_be32(f);
>>> - len = qemu_get_byte(f);
>>> - qemu_get_buffer(f, (uint8_t *)idstr, len);
>>> - idstr[len] = 0;
>>> - instance_id = qemu_get_be32(f);
>>> - version_id = qemu_get_be32(f);
>>> -
>>> - /* Find savevm section */
>>> - se = find_se(idstr, instance_id);
>>> - if (se == NULL) {
>>> - fprintf(stderr, "Unknown savevm section or instance '%s' %d\n",
>>> idstr, instance_id);
>>> - ret = -EINVAL;
>>> - goto out;
>>> - }
>>> +static int qemu_loadvm_state_iterate(QEMUFile *f)
>>> +{
>>> + LoadStateEntry *le;
>>> + uint32_t section_id;
>>> + int ret;
>>>
>>> - /* Validate version */
>>> - if (version_id> se->version_id) {
>>> - fprintf(stderr, "savevm: unsupported version %d for '%s' v%d\n",
>>> - version_id, idstr, se->version_id);
>>> - ret = -EINVAL;
>>> - goto out;
>>> - }
>>> + section_id = qemu_get_be32(f);
>>
>> But we're still blocking in qemu_get_be32() so I don't see how this
>> makes it async.
>
> In that sense, it's not completely async, but still better than being 
> in the while loop that doesn't accept any commands on the incoming side.

What's confusing me is I don't understand how it's accepting command on 
the incoming side.

By my reading of the code, you set a callback on read and then in the 
read callback, you invoke iterate().  When iterate completes, you fall 
back to the main loop.  This sort of works only because the effect is 
that after each iteration, by falling back to the main loop you can run 
all handlers (including the monitor's handlers).

But there are some serious problems with this approach.  When iterate() 
completes, you've not necessarily exhausted the QEMUFile's buffer.  If 
the buffer is holding the full contents of the final stage of migration, 
then there is not going to be any more data available to read from the 
socket which means that when you drop to the main loop, you'll never 
execute async again.  You could probably easily reproduce this problem 
by making the QEMUFile buffer very large.  I think you're getting lucky 
because the final stage is probably larger than 32k in most circumstances.

In the very least, you should use a bottom half instead 
qemu_set_fd_handler2.  It's still not async but I'm not sure whether 
it's a good enough solution.

Regards,

Anthony Liguori

> To make things completely async, I think we should solve the lock 
> issue and add a thread for incoming migration, but I don't think 
> that's your opinion.
>
> Did I misunderstand your comment for the previous approach?
>
> Thanks,
>
> Yoshi
>
>>
>> Regards,
>>
>> Anthony Liguori
>>
>>
>>
>>
>

Re: [Qemu-devel] [PATCH 1/4] savevm: refactor qemu_loadvm_state().

[Yoshiaki Tamura]

Anthony Liguori wrote:
> On 06/14/2010 09:01 PM, Yoshiaki Tamura wrote:
>> Anthony Liguori wrote:
>>> On 06/03/2010 02:22 AM, Yoshiaki Tamura wrote:
>>>> Split qemu_loadvm_state(), and introduce
>>>> qemu_loadvm_state_{begin,iterate,complete,async}.
>>>> qemu_loadvm_state_async() is a function to handle a single incoming
>>>> section.
>>>>
>>>> Signed-off-by: Yoshiaki Tamura<tamura.yoshiaki@lab.ntt.co.jp>
>>>> ---
>>>> savevm.c | 206
>>>> +++++++++++++++++++++++++++++++++++++++++++-------------------
>>>> sysemu.h | 2 +
>>>> 2 files changed, 146 insertions(+), 62 deletions(-)
>>>>
>>>> diff --git a/savevm.c b/savevm.c
>>>> index dc20390..aa4f98c 100644
>>>> --- a/savevm.c
>>>> +++ b/savevm.c
>>>> @@ -1005,6 +1005,8 @@ typedef struct SaveStateEntry {
>>>>
>>>> static QTAILQ_HEAD(savevm_handlers, SaveStateEntry) savevm_handlers =
>>>> QTAILQ_HEAD_INITIALIZER(savevm_handlers);
>>>> +static QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
>>>> + QLIST_HEAD_INITIALIZER(loadvm_handlers);
>>>> static int global_section_id;
>>>>
>>>> static int calculate_new_instance_id(const char *idstr)
>>>> @@ -1460,14 +1462,9 @@ typedef struct LoadStateEntry {
>>>> int version_id;
>>>> } LoadStateEntry;
>>>>
>>>> -int qemu_loadvm_state(QEMUFile *f)
>>>> +int qemu_loadvm_state_begin(QEMUFile *f)
>>>> {
>>>> - QLIST_HEAD(, LoadStateEntry) loadvm_handlers =
>>>> - QLIST_HEAD_INITIALIZER(loadvm_handlers);
>>>> - LoadStateEntry *le, *new_le;
>>>> - uint8_t section_type;
>>>> unsigned int v;
>>>> - int ret;
>>>>
>>>> v = qemu_get_be32(f);
>>>> if (v != QEMU_VM_FILE_MAGIC)
>>>> @@ -1481,73 +1478,157 @@ int qemu_loadvm_state(QEMUFile *f)
>>>> if (v != QEMU_VM_FILE_VERSION)
>>>> return -ENOTSUP;
>>>>
>>>> - while ((section_type = qemu_get_byte(f)) != QEMU_VM_EOF) {
>>>> - uint32_t instance_id, version_id, section_id;
>>>> - SaveStateEntry *se;
>>>> - char idstr[257];
>>>> - int len;
>>>> + return 0;
>>>> +}
>>>>
>>>> - switch (section_type) {
>>>> - case QEMU_VM_SECTION_START:
>>>> - case QEMU_VM_SECTION_FULL:
>>>> - /* Read section start */
>>>> - section_id = qemu_get_be32(f);
>>>> - len = qemu_get_byte(f);
>>>> - qemu_get_buffer(f, (uint8_t *)idstr, len);
>>>> - idstr[len] = 0;
>>>> - instance_id = qemu_get_be32(f);
>>>> - version_id = qemu_get_be32(f);
>>>> -
>>>> - /* Find savevm section */
>>>> - se = find_se(idstr, instance_id);
>>>> - if (se == NULL) {
>>>> - fprintf(stderr, "Unknown savevm section or instance '%s' %d\n",
>>>> idstr, instance_id);
>>>> - ret = -EINVAL;
>>>> - goto out;
>>>> - }
>>>> +static int qemu_loadvm_state_iterate(QEMUFile *f)
>>>> +{
>>>> + LoadStateEntry *le;
>>>> + uint32_t section_id;
>>>> + int ret;
>>>>
>>>> - /* Validate version */
>>>> - if (version_id> se->version_id) {
>>>> - fprintf(stderr, "savevm: unsupported version %d for '%s' v%d\n",
>>>> - version_id, idstr, se->version_id);
>>>> - ret = -EINVAL;
>>>> - goto out;
>>>> - }
>>>> + section_id = qemu_get_be32(f);
>>>
>>> But we're still blocking in qemu_get_be32() so I don't see how this
>>> makes it async.
>>
>> In that sense, it's not completely async, but still better than being
>> in the while loop that doesn't accept any commands on the incoming side.
>
> What's confusing me is I don't understand how it's accepting command on
> the incoming side.
>
> By my reading of the code, you set a callback on read and then in the
> read callback, you invoke iterate(). When iterate completes, you fall
> back to the main loop. This sort of works only because the effect is
> that after each iteration, by falling back to the main loop you can run
> all handlers (including the monitor's handlers).
>
> But there are some serious problems with this approach. When iterate()
> completes, you've not necessarily exhausted the QEMUFile's buffer. If
> the buffer is holding the full contents of the final stage of migration,
> then there is not going to be any more data available to read from the
> socket which means that when you drop to the main loop, you'll never
> execute async again. You could probably easily reproduce this problem by
> making the QEMUFile buffer very large. I think you're getting lucky
> because the final stage is probably larger than 32k in most circumstances.
>
> In the very least, you should use a bottom half instead
> qemu_set_fd_handler2. It's still not async but I'm not sure whether it's
> a good enough solution.

Thank you for pointing out the problem.
So instead of waiting data with select which is event driven, bottom half would 
at least give us a chance to check the QEMUFile's buffer with specific interval. 
  This should fix the problem you mentioned above.  However, if the network 
connection is lost, as you imagine, we'll still be blocked at fill_buffer(), I 
guess.

What I eventually need is a mechanism to prevent from blocking during migration, 
so using bottom half instead of qemu_set_fd_handler2 isn't perfect solution for 
me.  Probably, introducing a header which describes the data to be received into 
the savevm format, and select until incoming side receives the expected data 
could be a solution.  But changing the savevm format just for this doesn't seem 
to be good.

I believe some people also needs async incoming migration framework too. (Juan?)
What I would like to ask is, should I keep working to improve this approach?
If it's not a good approach for making other people happy, I would like to drop 
this work for now, and discuss a better solution in the future.

Thanks,

Yoshi

>
> Regards,
>
> Anthony Liguori
>
>> To make things completely async, I think we should solve the lock
>> issue and add a thread for incoming migration, but I don't think
>> that's your opinion.
>>
>> Did I misunderstand your comment for the previous approach?
>>
>> Thanks,
>>
>> Yoshi
>>
>>>
>>> Regards,
>>>
>>> Anthony Liguori
>>>
>>>
>>>
>>>
>>
>
>
>
>

[Qemu-devel] [PATCH 2/4] migration-tcp: add support for asynchronous incoming migration.

[Yoshiaki Tamura]

Introduce tcp_incoming_migration_async(), and register it handle
incoming migration asynchronously.  MigrateInocming is used to pass
argument to tcp_incoming_migration_async().
tcp_start_incoming_migration() allocates FdMigrationState, and returns
MigrationState to print incoming migration info.

Signed-off-by: Yoshiaki Tamura <tamura.yoshiaki@lab.ntt.co.jp>
---
 migration-tcp.c |  109 ++++++++++++++++++++++++++++++++++++++++---------------
 migration.h     |    2 +-
 2 files changed, 80 insertions(+), 31 deletions(-)

diff --git a/migration-tcp.c b/migration-tcp.c
index 95ce722..b55e891 100644
--- a/migration-tcp.c
+++ b/migration-tcp.c
@@ -137,16 +137,54 @@ MigrationState *tcp_start_outgoing_migration(Monitor *mon,
     return &s->mig_state;
 }
 
+typedef struct MigrateIncoming {
+    FdMigrationState *s;
+    int c;
+} MigrateIncoming;
+
+static void tcp_incoming_migration_async(void *opaque)
+{
+    MigrateIncoming *p = opaque;
+    FdMigrationState *s = p->s;
+    QEMUFile *f = s->file;
+    int ret;
+
+    ret = qemu_loadvm_state_async(f);
+    if (ret < 0) {
+        s->state = MIG_STATE_ERROR;
+        fprintf(stderr, "load of migration failed\n");
+        goto out;
+    } else if (ret == 0) {
+        qemu_announce_self();
+        s->state = MIG_STATE_COMPLETED;
+        DPRINTF("successfully loaded vm state\n");
+
+        if (autostart) {
+            vm_start();
+        }
+
+        goto out;
+    }
+
+    return;
+
+out:
+    qemu_set_fd_handler2(p->c, NULL, NULL, NULL, NULL);
+    close(p->c);
+    migrate_fd_cleanup(s);
+    qemu_free(p);
+}
+
 static void tcp_accept_incoming_migration(void *opaque)
 {
     struct sockaddr_in addr;
     socklen_t addrlen = sizeof(addr);
-    int s = (unsigned long)opaque;
-    QEMUFile *f;
+    FdMigrationState *s = opaque;
+    MigrateIncoming *p;
     int c, ret;
 
     do {
-        c = qemu_accept(s, (struct sockaddr *)&addr, &addrlen);
+        c = qemu_accept(s->fd, (struct sockaddr *)&addr, &addrlen);
     } while (c == -1 && socket_error() == EINTR);
 
     DPRINTF("accepted migration\n");
@@ -156,61 +194,72 @@ static void tcp_accept_incoming_migration(void *opaque)
         return;
     }
 
-    f = qemu_fopen_socket(c);
-    if (f == NULL) {
+    p = qemu_mallocz(sizeof(MigrateIncoming));
+    p->s = s;
+    p->c = c;    
+
+    s->file = qemu_fopen_socket(c);
+    if (s->file == NULL) {
         fprintf(stderr, "could not qemu_fopen socket\n");
         goto out;
     }
 
-    ret = qemu_loadvm_state(f);
+    ret = qemu_loadvm_state_begin(s->file);
     if (ret < 0) {
-        fprintf(stderr, "load of migration failed\n");
-        goto out_fopen;
+        goto out;
     }
-    qemu_announce_self();
-    DPRINTF("successfully loaded vm state\n");
 
-    if (autostart)
-        vm_start();
+    qemu_set_fd_handler2(c, NULL, tcp_incoming_migration_async, NULL, p);
+
+    return;
 
-out_fopen:
-    qemu_fclose(f);
 out:
-    qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
-    close(s);
     close(c);
+    migrate_fd_release(&s->mig_state);
+    qemu_free(p);
 }
 
-int tcp_start_incoming_migration(const char *host_port)
+MigrationState *tcp_start_incoming_migration(const char *host_port)
 {
     struct sockaddr_in addr;
+    FdMigrationState *s;
     int val;
-    int s;
 
     if (parse_host_port(&addr, host_port) < 0) {
         fprintf(stderr, "invalid host/port combination: %s\n", host_port);
-        return -EINVAL;
+        return NULL;
     }
 
-    s = qemu_socket(PF_INET, SOCK_STREAM, 0);
-    if (s == -1)
-        return -socket_error();
+    s = qemu_mallocz(sizeof(*s));
+
+    s->get_error = socket_errno;
+    s->close = tcp_close;
+    s->mig_state.cancel = migrate_fd_cancel;
+    s->mig_state.get_status = migrate_fd_get_status;
+    s->mig_state.release = migrate_fd_release;
+    s->state = MIG_STATE_ACTIVE;
+
+    s->fd = qemu_socket(PF_INET, SOCK_STREAM, 0);
+    if (s->fd == -1) {
+        qemu_free(s);
+        return NULL;
+    }
 
     val = 1;
-    setsockopt(s, SOL_SOCKET, SO_REUSEADDR, (const char *)&val, sizeof(val));
+    setsockopt(s->fd, SOL_SOCKET, SO_REUSEADDR, (const char *)&val,
+               sizeof(val));
 
-    if (bind(s, (struct sockaddr *)&addr, sizeof(addr)) == -1)
+    if (bind(s->fd, (struct sockaddr *)&addr, sizeof(addr)) == -1)
         goto err;
 
-    if (listen(s, 1) == -1)
+    if (listen(s->fd, 1) == -1)
         goto err;
 
-    qemu_set_fd_handler2(s, NULL, tcp_accept_incoming_migration, NULL,
-                         (void *)(unsigned long)s);
+    qemu_set_fd_handler2(s->fd, NULL, tcp_accept_incoming_migration, NULL, s);
 
-    return 0;
+    return &s->mig_state;
 
 err:
-    close(s);
-    return -socket_error();
+    migrate_fd_release(&s->mig_state);
+    return NULL;
 }
diff --git a/migration.h b/migration.h
index 385423f..c11e6db 100644
--- a/migration.h
+++ b/migration.h
@@ -76,7 +76,7 @@ MigrationState *exec_start_outgoing_migration(Monitor *mon,
 					      int blk,
 					      int inc);
 
-int tcp_start_incoming_migration(const char *host_port);
+MigrationState *tcp_start_incoming_migration(const char *host_port);
 
 MigrationState *tcp_start_outgoing_migration(Monitor *mon,
                                              const char *host_port,
-- 
1.7.0.31.g1df487

[Qemu-devel] [PATCH 3/4] arch_init: calculate transferred bytes at ram_load().

[Yoshiaki Tamura]

Currently incoming side of migration doesn't know how many bytes are
transferred.  To print "info migrate" on incoming side, we need to
calculate transferred bytes at ram_load().

Signed-off-by: Yoshiaki Tamura <tamura.yoshiaki@lab.ntt.co.jp>
---
 arch_init.c |    6 +++++-
 1 files changed, 5 insertions(+), 1 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 8e849a8..1bdfa58 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -246,7 +246,7 @@ int ram_save_live(Monitor *mon, QEMUFile *f, int stage, void *opaque)
 int ram_load(QEMUFile *f, void *opaque, int version_id)
 {
     ram_addr_t addr;
-    int flags;
+    int flags, bytes_sent;
 
     if (version_id != 3) {
         return -EINVAL;
@@ -274,12 +274,16 @@ int ram_load(QEMUFile *f, void *opaque, int version_id)
                         MADV_DONTNEED);
             }
 #endif
+            bytes_sent = 1;
         } else if (flags & RAM_SAVE_FLAG_PAGE) {
             qemu_get_buffer(f, qemu_get_ram_ptr(addr), TARGET_PAGE_SIZE);
+            bytes_sent = TARGET_PAGE_SIZE;
         }
         if (qemu_file_has_error(f)) {
             return -EIO;
         }
+
+        bytes_transferred += bytes_sent;
     } while (!(flags & RAM_SAVE_FLAG_EOS));
 
     return 0;
-- 
1.7.0.31.g1df487

[Qemu-devel] [PATCH 4/4] migration: add support to print migration info on incoming side.

[Yoshiaki Tamura]

Set current_migration after calling tcp_star_incoming_migration().  On
incoming side, we don't have to print remaining rams, so introduce
incoming flag to switch messages.

Signed-off-by: Yoshiaki Tamura <tamura.yoshiaki@lab.ntt.co.jp>
---
 migration.c |   18 ++++++++++++++----
 1 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/migration.c b/migration.c
index 706fe55..9b6aa28 100644
--- a/migration.c
+++ b/migration.c
@@ -36,12 +36,14 @@ static uint32_t max_throttle = (32 << 20);
 
 static MigrationState *current_migration;
 
+static int incoming;
+
 void qemu_start_incoming_migration(const char *uri)
 {
     const char *p;
 
     if (strstart(uri, "tcp:", &p))
-        tcp_start_incoming_migration(p);
+        current_migration = tcp_start_incoming_migration(p);
 #if !defined(WIN32)
     else if (strstart(uri, "exec:", &p))
         exec_start_incoming_migration(p);
@@ -50,8 +52,12 @@ void qemu_start_incoming_migration(const char *uri)
     else if (strstart(uri, "fd:", &p))
         fd_start_incoming_migration(p);
 #endif
-    else
+    else {
         fprintf(stderr, "unknown migration protocol: %s\n", uri);
+        return;
+    }
+
+    incoming = 1;
 }
 
 int do_migrate(Monitor *mon, const QDict *qdict, QObject **ret_data)
@@ -162,8 +168,12 @@ static void migrate_print_status(Monitor *mon, const char *name,
 
     monitor_printf(mon, "transferred %s: %" PRIu64 " kbytes\n", name,
                         qdict_get_int(qdict, "transferred") >> 10);
-    monitor_printf(mon, "remaining %s: %" PRIu64 " kbytes\n", name,
-                        qdict_get_int(qdict, "remaining") >> 10);
+    if (incoming) {
+        monitor_printf(mon, "remaining %s: n/a\n", name);
+    } else {
+        monitor_printf(mon, "remaining %s: %" PRIu64 " kbytes\n", name,
+                            qdict_get_int(qdict, "remaining") >> 10);
+    }
     monitor_printf(mon, "total %s: %" PRIu64 " kbytes\n", name,
                         qdict_get_int(qdict, "total") >> 10);
 }
-- 
1.7.0.31.g1df487