[Qemu-devel] [PATCH 0/9] Incoming migration coroutine

[Qemu-devel] [PATCH 0/9] Incoming migration coroutine

[Paolo Bonzini]

Juan,

here is the implementation of migration-in-a-coroutine, applying on top
of your recently-posted 41-patch series.  With these patches, the monitor
(and in the future the NBD server) is responsive during migration.

The first seven patches are just cleanups, only the last two
actually implement the feature.  They are the opposite change of the
nonblocking->blocking change that you implemented for the migration
thread.  However, the change is much simpler because we have no timers,
and because of the use of coroutines.

Without coroutines (and as in non-threaded migration), you have
to proceed in two steps: first collect data in a buffer, then
write it.  This lets you handle EAGAIN only at precise points in
buffered_flush/buffered_put_buffer, so that you can restart writing
in migrate_fd_put_notify.  This "checkpointing" is the reason why
QEMUFileBuffered exists.  With coroutines, you can just stop whenever
you want with qemu_coroutine_yield.  As soon as select tells you that you
can read, you'll re-enter directly in qemu_get_buffer, read more data
and pass it to the loading routines.

Thanks,

Paolo

Paolo Bonzini (9):
  migration: unify stdio-based QEMUFile operations
  migration: consolidate QEMUFile methods in a single QEMUFileOps struct
  migration: add qemu_get_fd
  migration: replace qemu_stdio_fd with qemu_get_fd
  migration: clean up server sockets and handlers before invoking process_incoming_migration
  migration: close socket QEMUFile from socket_close
  migration: move qemu_fclose to process_incoming_migration
  migration: handle EAGAIN while reading QEMUFile
  migration: move process_incoming_migration to a coroutine

 buffered_file.c  |  21 +++++--
 migration-exec.c |   5 +-
 migration-fd.c   |   3 +-
 migration-tcp.c  |  10 +--
 migration-unix.c |  10 +--
 migration.c      |  27 +++++++-
 qemu-file.h      |  23 ++++---
 savevm.c         | 188 ++++++++++++++++++++++++++++++++-----------------------
 8 file modificati, 181 inserzioni(+), 106 rimozioni(-)

-- 
1.7.12

[Qemu-devel] [PATCH 1/9] migration: unify stdio-based QEMUFile operations

[Paolo Bonzini]

Now that qemu_fseek does not exist anymore, there is no reason to do
an fseek before fread/fwrite when operating on an stdio file.
Thus, unify the get/put_buffer callbacks used by qemu_fopen
with those used for pipes.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 savevm.c | 19 ++-----------------
 1 file modificato, 2 inserzioni(+), 17 rimozioni(-)

diff --git a/savevm.c b/savevm.c
index 2ea1fa6..5cd4432 100644
--- a/savevm.c
+++ b/savevm.c
@@ -342,21 +342,6 @@ QEMUFile *qemu_fopen_socket(int fd)
     return s->file;
 }
 
-static int file_put_buffer(void *opaque, const uint8_t *buf,
-                            int64_t pos, int size)
-{
-    QEMUFileStdio *s = opaque;
-    fseek(s->stdio_file, pos, SEEK_SET);
-    return fwrite(buf, 1, size, s->stdio_file);
-}
-
-static int file_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
-{
-    QEMUFileStdio *s = opaque;
-    fseek(s->stdio_file, pos, SEEK_SET);
-    return fread(buf, 1, size, s->stdio_file);
-}
-
 QEMUFile *qemu_fopen(const char *filename, const char *mode)
 {
     QEMUFileStdio *s;
@@ -375,10 +360,10 @@ QEMUFile *qemu_fopen(const char *filename, const char *mode)
         goto fail;
     
     if(mode[0] == 'w') {
-        s->file = qemu_fopen_ops(s, file_put_buffer, NULL, stdio_fclose, 
+        s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_fclose,
 				 NULL, NULL, NULL);
     } else {
-        s->file = qemu_fopen_ops(s, NULL, file_get_buffer, stdio_fclose, 
+        s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_fclose,
 			       NULL, NULL, NULL);
     }
     return s->file;
-- 
1.7.12

[Qemu-devel] [PATCH 2/9] migration: consolidate QEMUFile methods in a single QEMUFileOps struct

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 buffered_file.c |  13 ++++---
 qemu-file.h     |  16 +++++----
 savevm.c        | 108 +++++++++++++++++++++++++++++++-------------------------
 3 file modificati, 79 inserzioni(+), 58 rimozioni(-)

diff --git a/buffered_file.c b/buffered_file.c
index 318d0f0..535cdf5 100644
--- a/buffered_file.c
+++ b/buffered_file.c
@@ -233,6 +233,14 @@ static void buffered_rate_tick(void *opaque)
     buffered_put_buffer(s, NULL, 0, 0);
 }
 
+static const QEMUFileOps buffered_file_ops = {
+    .put_buffer =     buffered_put_buffer,
+    .close =          buffered_close,
+    .rate_limit =     buffered_rate_limit,
+    .get_rate_limit = buffered_get_rate_limit,
+    .set_rate_limit = buffered_set_rate_limit,
+};
+
 QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
 {
     QEMUFileBuffered *s;
@@ -242,10 +250,7 @@ QEMUFile *qemu_fopen_ops_buffered(MigrationState *migration_state)
     s->migration_state = migration_state;
     s->xfer_limit = migration_state->bandwidth_limit / 10;
 
-    s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
-                             buffered_close, buffered_rate_limit,
-                             buffered_set_rate_limit,
-			     buffered_get_rate_limit);
+    s->file = qemu_fopen_ops(s, &buffered_file_ops);
 
     s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
 
diff --git a/qemu-file.h b/qemu-file.h
index 9c8985b..c89e8e0 100644
--- a/qemu-file.h
+++ b/qemu-file.h
@@ -59,12 +59,16 @@ typedef int (QEMUFileRateLimit)(void *opaque);
 typedef int64_t (QEMUFileSetRateLimit)(void *opaque, int64_t new_rate);
 typedef int64_t (QEMUFileGetRateLimit)(void *opaque);
 
-QEMUFile *qemu_fopen_ops(void *opaque, QEMUFilePutBufferFunc *put_buffer,
-                         QEMUFileGetBufferFunc *get_buffer,
-                         QEMUFileCloseFunc *close,
-                         QEMUFileRateLimit *rate_limit,
-                         QEMUFileSetRateLimit *set_rate_limit,
-                         QEMUFileGetRateLimit *get_rate_limit);
+typedef struct QEMUFileOps {
+    QEMUFilePutBufferFunc *put_buffer;
+    QEMUFileGetBufferFunc *get_buffer;
+    QEMUFileCloseFunc *close;
+    QEMUFileRateLimit *rate_limit;
+    QEMUFileSetRateLimit *set_rate_limit;
+    QEMUFileGetRateLimit *get_rate_limit;
+} QEMUFileOps;
+
+QEMUFile *qemu_fopen_ops(void *opaque, const QEMUFileOps *ops);
 QEMUFile *qemu_fopen(const char *filename, const char *mode);
 QEMUFile *qemu_fdopen(int fd, const char *mode);
 QEMUFile *qemu_fopen_socket(int fd);
diff --git a/savevm.c b/savevm.c
index 5cd4432..0751422 100644
--- a/savevm.c
+++ b/savevm.c
@@ -162,12 +162,7 @@ void qemu_announce_self(void)
 #define IO_BUF_SIZE 32768
 
 struct QEMUFile {
-    QEMUFilePutBufferFunc *put_buffer;
-    QEMUFileGetBufferFunc *get_buffer;
-    QEMUFileCloseFunc *close;
-    QEMUFileRateLimit *rate_limit;
-    QEMUFileSetRateLimit *set_rate_limit;
-    QEMUFileGetRateLimit *get_rate_limit;
+    const QEMUFileOps *ops;
     void *opaque;
     int is_write;
 
@@ -256,6 +251,16 @@ static int stdio_fclose(void *opaque)
     return ret;
 }
 
+static const QEMUFileOps stdio_pipe_read_ops = {
+    .get_buffer = stdio_get_buffer,
+    .close =      stdio_pclose
+};
+
+static const QEMUFileOps stdio_pipe_write_ops = {
+    .put_buffer = stdio_put_buffer,
+    .close =      stdio_pclose
+};
+
 QEMUFile *qemu_popen(FILE *stdio_file, const char *mode)
 {
     QEMUFileStdio *s;
@@ -270,11 +275,9 @@ QEMUFile *qemu_popen(FILE *stdio_file, const char *mode)
     s->stdio_file = stdio_file;
 
     if(mode[0] == 'r') {
-        s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_pclose, 
-				 NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_pipe_read_ops);
     } else {
-        s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_pclose, 
-				 NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_pipe_write_ops);
     }
     return s->file;
 }
@@ -302,6 +305,16 @@ int qemu_stdio_fd(QEMUFile *f)
     return fd;
 }
 
+static const QEMUFileOps stdio_file_read_ops = {
+    .get_buffer = stdio_get_buffer,
+    .close =      stdio_fclose
+};
+
+static const QEMUFileOps stdio_file_write_ops = {
+    .put_buffer = stdio_put_buffer,
+    .close =      stdio_fclose
+};
+
 QEMUFile *qemu_fdopen(int fd, const char *mode)
 {
     QEMUFileStdio *s;
@@ -319,11 +332,9 @@ QEMUFile *qemu_fdopen(int fd, const char *mode)
         goto fail;
 
     if(mode[0] == 'r') {
-        s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_fclose, 
-				 NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_file_read_ops);
     } else {
-        s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_fclose, 
-				 NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_file_write_ops);
     }
     return s->file;
 
@@ -332,13 +343,17 @@ fail:
     return NULL;
 }
 
+static const QEMUFileOps socket_read_ops = {
+    .get_buffer = socket_get_buffer,
+    .close =      socket_close
+};
+
 QEMUFile *qemu_fopen_socket(int fd)
 {
     QEMUFileSocket *s = g_malloc0(sizeof(QEMUFileSocket));
 
     s->fd = fd;
-    s->file = qemu_fopen_ops(s, NULL, socket_get_buffer, socket_close, 
-			     NULL, NULL, NULL);
+    s->file = qemu_fopen_ops(s, &socket_read_ops);
     return s->file;
 }
 
@@ -360,11 +375,9 @@ QEMUFile *qemu_fopen(const char *filename, const char *mode)
         goto fail;
     
     if(mode[0] == 'w') {
-        s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_fclose,
-				 NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_file_write_ops);
     } else {
-        s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_fclose,
-			       NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_file_read_ops);
     }
     return s->file;
 fail:
@@ -389,32 +402,31 @@ static int bdrv_fclose(void *opaque)
     return bdrv_flush(opaque);
 }
 
+static const QEMUFileOps bdrv_read_ops = {
+    .get_buffer = block_get_buffer,
+    .close =      bdrv_fclose
+};
+
+static const QEMUFileOps bdrv_write_ops = {
+    .put_buffer = block_put_buffer,
+    .close =      bdrv_fclose
+};
+
 static QEMUFile *qemu_fopen_bdrv(BlockDriverState *bs, int is_writable)
 {
     if (is_writable)
-        return qemu_fopen_ops(bs, block_put_buffer, NULL, bdrv_fclose, 
-			      NULL, NULL, NULL);
-    return qemu_fopen_ops(bs, NULL, block_get_buffer, bdrv_fclose, NULL, NULL, NULL);
+        return qemu_fopen_ops(bs, &bdrv_write_ops);
+    return qemu_fopen_ops(bs, &bdrv_read_ops);
 }
 
-QEMUFile *qemu_fopen_ops(void *opaque, QEMUFilePutBufferFunc *put_buffer,
-                         QEMUFileGetBufferFunc *get_buffer,
-                         QEMUFileCloseFunc *close,
-                         QEMUFileRateLimit *rate_limit,
-                         QEMUFileSetRateLimit *set_rate_limit,
-                         QEMUFileGetRateLimit *get_rate_limit)
+QEMUFile *qemu_fopen_ops(void *opaque, const QEMUFileOps *ops)
 {
     QEMUFile *f;
 
     f = g_malloc0(sizeof(QEMUFile));
 
     f->opaque = opaque;
-    f->put_buffer = put_buffer;
-    f->get_buffer = get_buffer;
-    f->close = close;
-    f->rate_limit = rate_limit;
-    f->set_rate_limit = set_rate_limit;
-    f->get_rate_limit = get_rate_limit;
+    f->ops = ops;
     f->is_write = 0;
 
     return f;
@@ -437,11 +449,11 @@ static int qemu_fflush(QEMUFile *f)
 {
     int ret = 0;
 
-    if (!f->put_buffer)
+    if (!f->ops->put_buffer)
         return 0;
 
     if (f->is_write && f->buf_index > 0) {
-        ret = f->put_buffer(f->opaque, f->buf, f->buf_offset, f->buf_index);
+        ret = f->ops->put_buffer(f->opaque, f->buf, f->buf_offset, f->buf_index);
         if (ret >= 0) {
             f->buf_offset += f->buf_index;
         }
@@ -455,7 +467,7 @@ static void qemu_fill_buffer(QEMUFile *f)
     int len;
     int pending;
 
-    if (!f->get_buffer)
+    if (!f->ops->get_buffer)
         return;
 
     if (f->is_write)
@@ -468,7 +480,7 @@ static void qemu_fill_buffer(QEMUFile *f)
     f->buf_index = 0;
     f->buf_size = pending;
 
-    len = f->get_buffer(f->opaque, f->buf + pending, f->buf_offset,
+    len = f->ops->get_buffer(f->opaque, f->buf + pending, f->buf_offset,
                         IO_BUF_SIZE - pending);
     if (len > 0) {
         f->buf_size += len;
@@ -492,8 +504,8 @@ int qemu_fclose(QEMUFile *f)
     int ret;
     ret = qemu_fflush(f);
 
-    if (f->close) {
-        int ret2 = f->close(f->opaque);
+    if (f->ops->close) {
+        int ret2 = f->ops->close(f->opaque);
         if (ret >= 0) {
             ret = ret2;
         }
@@ -510,7 +522,7 @@ int qemu_fclose(QEMUFile *f)
 
 int qemu_file_put_notify(QEMUFile *f)
 {
-    return f->put_buffer(f->opaque, NULL, 0, 0);
+    return f->ops->put_buffer(f->opaque, NULL, 0, 0);
 }
 
 void qemu_put_buffer(QEMUFile *f, const uint8_t *buf, int size)
@@ -657,16 +669,16 @@ static int64_t qemu_ftell(QEMUFile *f)
 
 int qemu_file_rate_limit(QEMUFile *f)
 {
-    if (f->rate_limit)
-        return f->rate_limit(f->opaque);
+    if (f->ops->rate_limit)
+        return f->ops->rate_limit(f->opaque);
 
     return 0;
 }
 
 int64_t qemu_file_get_rate_limit(QEMUFile *f)
 {
-    if (f->get_rate_limit)
-        return f->get_rate_limit(f->opaque);
+    if (f->ops->get_rate_limit)
+        return f->ops->get_rate_limit(f->opaque);
 
     return 0;
 }
@@ -675,8 +687,8 @@ int64_t qemu_file_set_rate_limit(QEMUFile *f, int64_t new_rate)
 {
     /* any failed or completed migration keeps its state to allow probing of
      * migration data, but has no associated file anymore */
-    if (f && f->set_rate_limit)
-        return f->set_rate_limit(f->opaque, new_rate);
+    if (f && f->ops->set_rate_limit)
+        return f->ops->set_rate_limit(f->opaque, new_rate);
 
     return 0;
 }
-- 
1.7.12

[Qemu-devel] [PATCH 3/9] migration: add qemu_get_fd

[Paolo Bonzini]

Add a callback that lets incoming migration retrieve the file descriptor
for a QEMUFile.  This will replace the type-unsafe qemu_stdio_fd.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 buffered_file.c |  8 ++++++++
 qemu-file.h     |  6 ++++++
 savevm.c        | 27 +++++++++++++++++++++++++++
 3 file modificati, 41 inserzioni(+)

diff --git a/buffered_file.c b/buffered_file.c
index 535cdf5..2943f2e 100644
--- a/buffered_file.c
+++ b/buffered_file.c
@@ -173,6 +173,13 @@ static int buffered_close(void *opaque)
  *   1: Time to stop
  *   negative: There has been an error
  */
+static int buffered_get_fd(void *opaque)
+{
+    QEMUFileBuffered *s = opaque;
+
+    return qemu_get_fd(s->file);
+}
+
 static int buffered_rate_limit(void *opaque)
 {
     QEMUFileBuffered *s = opaque;
@@ -234,6 +241,7 @@ static void buffered_rate_tick(void *opaque)
 }
 
 static const QEMUFileOps buffered_file_ops = {
+    .get_fd =         buffered_get_fd,
     .put_buffer =     buffered_put_buffer,
     .close =          buffered_close,
     .rate_limit =     buffered_rate_limit,
diff --git a/qemu-file.h b/qemu-file.h
index c89e8e0..d552f5d 100644
--- a/qemu-file.h
+++ b/qemu-file.h
@@ -47,6 +47,10 @@ typedef int (QEMUFileGetBufferFunc)(void *opaque, uint8_t *buf,
  */
 typedef int (QEMUFileCloseFunc)(void *opaque);
 
+/* Called to return the OS file descriptor associated to the QEMUFile.
+ */
+typedef int (QEMUFileGetFD)(void *opaque);
+
 /* Called to determine if the file has exceeded its bandwidth allocation.  The
  * bandwidth capping is a soft limit, not a hard limit.
  */
@@ -63,6 +67,7 @@ typedef struct QEMUFileOps {
     QEMUFilePutBufferFunc *put_buffer;
     QEMUFileGetBufferFunc *get_buffer;
     QEMUFileCloseFunc *close;
+    QEMUFileGetFD *get_fd;
     QEMUFileRateLimit *rate_limit;
     QEMUFileSetRateLimit *set_rate_limit;
     QEMUFileGetRateLimit *get_rate_limit;
@@ -74,6 +79,7 @@ QEMUFile *qemu_fdopen(int fd, const char *mode);
 QEMUFile *qemu_fopen_socket(int fd);
 QEMUFile *qemu_popen(FILE *popen_file, const char *mode);
 QEMUFile *qemu_popen_cmd(const char *command, const char *mode);
+int qemu_get_fd(QEMUFile *f);
 int qemu_stdio_fd(QEMUFile *f);
 int qemu_fclose(QEMUFile *f);
 void qemu_put_buffer(QEMUFile *f, const uint8_t *buf, int size);
diff --git a/savevm.c b/savevm.c
index 0751422..334848a 100644
--- a/savevm.c
+++ b/savevm.c
@@ -187,6 +187,13 @@ typedef struct QEMUFileSocket
     QEMUFile *file;
 } QEMUFileSocket;
 
+static int socket_get_fd(void *opaque)
+{
+    QEMUFileSocket *s = opaque;
+
+    return s->fd;
+}
+
 static int socket_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
 {
     QEMUFileSocket *s = opaque;
@@ -209,6 +216,13 @@ static int socket_close(void *opaque)
     return 0;
 }
 
+static int stdio_get_fd(void *opaque)
+{
+    QEMUFileStdio *s = opaque;
+
+    return fileno(s->stdio_file);
+}
+
 static int stdio_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, int size)
 {
     QEMUFileStdio *s = opaque;
@@ -252,11 +266,13 @@ static int stdio_fclose(void *opaque)
 }
 
 static const QEMUFileOps stdio_pipe_read_ops = {
+    .get_fd =     stdio_get_fd,
     .get_buffer = stdio_get_buffer,
     .close =      stdio_pclose
 };
 
 static const QEMUFileOps stdio_pipe_write_ops = {
+    .get_fd =     stdio_get_fd,
     .put_buffer = stdio_put_buffer,
     .close =      stdio_pclose
 };
@@ -306,11 +322,13 @@ int qemu_stdio_fd(QEMUFile *f)
 }
 
 static const QEMUFileOps stdio_file_read_ops = {
+    .get_fd =     stdio_get_fd,
     .get_buffer = stdio_get_buffer,
     .close =      stdio_fclose
 };
 
 static const QEMUFileOps stdio_file_write_ops = {
+    .get_fd =     stdio_get_fd,
     .put_buffer = stdio_put_buffer,
     .close =      stdio_fclose
 };
@@ -344,6 +362,7 @@ fail:
 }
 
 static const QEMUFileOps socket_read_ops = {
+    .get_fd =     socket_get_fd,
     .get_buffer = socket_get_buffer,
     .close =      socket_close
 };
@@ -491,6 +510,14 @@ static void qemu_fill_buffer(QEMUFile *f)
         qemu_file_set_error(f, len);
 }
 
+int qemu_get_fd(QEMUFile *f)
+{
+    if (f->ops->get_fd) {
+        return f->ops->get_fd(f->opaque);
+    }
+    return -1;
+}
+
 /** Closes the file
  *
  * Returns negative error value if any error happened on previous operations or
-- 
1.7.12

[Qemu-devel] [PATCH 4/9] migration: replace qemu_stdio_fd with qemu_get_fd

[Paolo Bonzini]

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 migration-exec.c |  4 ++--
 migration-fd.c   |  2 +-
 qemu-file.h      |  1 -
 savevm.c         | 11 -----------
 4 file modificati, 3 inserzioni(+), 15 rimozioni(-)

diff --git a/migration-exec.c b/migration-exec.c
index 6c97db9..f81e390 100644
--- a/migration-exec.c
+++ b/migration-exec.c
@@ -98,7 +98,7 @@ static void exec_accept_incoming_migration(void *opaque)
     QEMUFile *f = opaque;
 
     process_incoming_migration(f);
-    qemu_set_fd_handler2(qemu_stdio_fd(f), NULL, NULL, NULL, NULL);
+    qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
     qemu_fclose(f);
 }
 
@@ -113,7 +113,7 @@ int exec_start_incoming_migration(const char *command)
         return -errno;
     }
 
-    qemu_set_fd_handler2(qemu_stdio_fd(f), NULL,
+    qemu_set_fd_handler2(qemu_get_fd(f), NULL,
 			 exec_accept_incoming_migration, NULL, f);
 
     return 0;
diff --git a/migration-fd.c b/migration-fd.c
index 50138ed..7ad98d9 100644
--- a/migration-fd.c
+++ b/migration-fd.c
@@ -104,7 +104,7 @@ static void fd_accept_incoming_migration(void *opaque)
     QEMUFile *f = opaque;
 
     process_incoming_migration(f);
-    qemu_set_fd_handler2(qemu_stdio_fd(f), NULL, NULL, NULL, NULL);
+    qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
     qemu_fclose(f);
 }
 
diff --git a/qemu-file.h b/qemu-file.h
index d552f5d..d64bdbb 100644
--- a/qemu-file.h
+++ b/qemu-file.h
@@ -80,7 +80,6 @@ QEMUFile *qemu_fopen_socket(int fd);
 QEMUFile *qemu_popen(FILE *popen_file, const char *mode);
 QEMUFile *qemu_popen_cmd(const char *command, const char *mode);
 int qemu_get_fd(QEMUFile *f);
-int qemu_stdio_fd(QEMUFile *f);
 int qemu_fclose(QEMUFile *f);
 void qemu_put_buffer(QEMUFile *f, const uint8_t *buf, int size);
 void qemu_put_byte(QEMUFile *f, int v);
diff --git a/savevm.c b/savevm.c
index 334848a..040068d 100644
--- a/savevm.c
+++ b/savevm.c
@@ -310,17 +310,6 @@ QEMUFile *qemu_popen_cmd(const char *command, const char *mode)
     return qemu_popen(popen_file, mode);
 }
 
-int qemu_stdio_fd(QEMUFile *f)
-{
-    QEMUFileStdio *p;
-    int fd;
-
-    p = (QEMUFileStdio *)f->opaque;
-    fd = fileno(p->stdio_file);
-
-    return fd;
-}
-
 static const QEMUFileOps stdio_file_read_ops = {
     .get_fd =     stdio_get_fd,
     .get_buffer = stdio_get_buffer,
-- 
1.7.12

[Qemu-devel] [PATCH 5/9] migration: clean up server sockets and handlers before invoking process_incoming_migration

[Paolo Bonzini]

A first step towards making a common "suffix" for all migration protocols,
and moving it to process_incoming_migration.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 migration-exec.c | 2 +-
 migration-fd.c   | 2 +-
 migration-tcp.c  | 7 +++----
 migration-unix.c | 7 +++----
 4 file modificati, 8 inserzioni(+), 10 rimozioni(-)

diff --git a/migration-exec.c b/migration-exec.c
index f81e390..0964dbb 100644
--- a/migration-exec.c
+++ b/migration-exec.c
@@ -97,8 +97,8 @@ static void exec_accept_incoming_migration(void *opaque)
 {
     QEMUFile *f = opaque;
 
-    process_incoming_migration(f);
     qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
+    process_incoming_migration(f);
     qemu_fclose(f);
 }
 
diff --git a/migration-fd.c b/migration-fd.c
index 7ad98d9..62ef593 100644
--- a/migration-fd.c
+++ b/migration-fd.c
@@ -103,8 +103,8 @@ static void fd_accept_incoming_migration(void *opaque)
 {
     QEMUFile *f = opaque;
 
-    process_incoming_migration(f);
     qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
+    process_incoming_migration(f);
     qemu_fclose(f);
 }
 
diff --git a/migration-tcp.c b/migration-tcp.c
index ac891c3..3436e84 100644
--- a/migration-tcp.c
+++ b/migration-tcp.c
@@ -115,12 +115,14 @@ static void tcp_accept_incoming_migration(void *opaque)
     do {
         c = qemu_accept(s, (struct sockaddr *)&addr, &addrlen);
     } while (c == -1 && socket_error() == EINTR);
+    qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
+    close(s);
 
     DPRINTF("accepted migration\n");
 
     if (c == -1) {
         fprintf(stderr, "could not accept migration connection\n");
-        goto out2;
+        goto out;
     }
 
     f = qemu_fopen_socket(c);
@@ -133,9 +135,6 @@ static void tcp_accept_incoming_migration(void *opaque)
     qemu_fclose(f);
 out:
     close(c);
-out2:
-    qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
-    close(s);
 }
 
 int tcp_start_incoming_migration(const char *host_port, Error **errp)
diff --git a/migration-unix.c b/migration-unix.c
index 169de88..a407af2 100644
--- a/migration-unix.c
+++ b/migration-unix.c
@@ -129,12 +129,14 @@ static void unix_accept_incoming_migration(void *opaque)
     do {
         c = qemu_accept(s, (struct sockaddr *)&addr, &addrlen);
     } while (c == -1 && errno == EINTR);
+    qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
+    close(s);
 
     DPRINTF("accepted migration\n");
 
     if (c == -1) {
         fprintf(stderr, "could not accept migration connection\n");
-        goto out2;
+        goto out;
     }
 
     f = qemu_fopen_socket(c);
@@ -147,9 +149,6 @@ static void unix_accept_incoming_migration(void *opaque)
     qemu_fclose(f);
 out:
     close(c);
-out2:
-    qemu_set_fd_handler2(s, NULL, NULL, NULL, NULL);
-    close(s);
 }
 
 int unix_start_incoming_migration(const char *path)
-- 
1.7.12

[Qemu-devel] [PATCH 6/9] migration: close socket QEMUFile from socket_close

[Paolo Bonzini]

The common suffix now is process_incoming_migration+qemu_fclose.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 migration-tcp.c  | 2 ++
 migration-unix.c | 2 ++
 savevm.c         | 1 +
 3 file modificati, 5 inserzioni(+)

diff --git a/migration-tcp.c b/migration-tcp.c
index 3436e84..58d6884 100644
--- a/migration-tcp.c
+++ b/migration-tcp.c
@@ -133,6 +133,8 @@ static void tcp_accept_incoming_migration(void *opaque)
 
     process_incoming_migration(f);
     qemu_fclose(f);
+    return;
+
 out:
     close(c);
 }
diff --git a/migration-unix.c b/migration-unix.c
index a407af2..57aca11 100644
--- a/migration-unix.c
+++ b/migration-unix.c
@@ -147,6 +147,8 @@ static void unix_accept_incoming_migration(void *opaque)
 
     process_incoming_migration(f);
     qemu_fclose(f);
+    return;
+
 out:
     close(c);
 }
diff --git a/savevm.c b/savevm.c
index 040068d..949c38c 100644
--- a/savevm.c
+++ b/savevm.c
@@ -212,6 +212,7 @@ static int socket_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
 static int socket_close(void *opaque)
 {
     QEMUFileSocket *s = opaque;
+    close(s->fd);
     g_free(s);
     return 0;
 }
-- 
1.7.12

[Qemu-devel] [PATCH 7/9] migration: move qemu_fclose to process_incoming_migration

[Paolo Bonzini]

The common suffix is now just process_incoming_migration.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 migration-exec.c | 1 -
 migration-fd.c   | 1 -
 migration-tcp.c  | 1 -
 migration-unix.c | 1 -
 migration.c      | 6 +++++-
 5 file modificati, 5 inserzioni(+), 5 rimozioni(-)

diff --git a/migration-exec.c b/migration-exec.c
index 0964dbb..bb91917 100644
--- a/migration-exec.c
+++ b/migration-exec.c
@@ -99,7 +99,6 @@ static void exec_accept_incoming_migration(void *opaque)
 
     qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
     process_incoming_migration(f);
-    qemu_fclose(f);
 }
 
 int exec_start_incoming_migration(const char *command)
diff --git a/migration-fd.c b/migration-fd.c
index 62ef593..97136cb 100644
--- a/migration-fd.c
+++ b/migration-fd.c
@@ -105,7 +105,6 @@ static void fd_accept_incoming_migration(void *opaque)
 
     qemu_set_fd_handler2(qemu_get_fd(f), NULL, NULL, NULL, NULL);
     process_incoming_migration(f);
-    qemu_fclose(f);
 }
 
 int fd_start_incoming_migration(const char *infd)
diff --git a/migration-tcp.c b/migration-tcp.c
index 58d6884..a1766a4 100644
--- a/migration-tcp.c
+++ b/migration-tcp.c
@@ -132,7 +132,6 @@ static void tcp_accept_incoming_migration(void *opaque)
     }
 
     process_incoming_migration(f);
-    qemu_fclose(f);
     return;
 
 out:
diff --git a/migration-unix.c b/migration-unix.c
index 57aca11..d8d5dfa 100644
--- a/migration-unix.c
+++ b/migration-unix.c
@@ -146,7 +146,6 @@ static void unix_accept_incoming_migration(void *opaque)
     }
 
     process_incoming_migration(f);
-    qemu_fclose(f);
     return;
 
 out:
diff --git a/migration.c b/migration.c
index 2c29d04..2bc4bac 100644
--- a/migration.c
+++ b/migration.c
@@ -88,7 +88,11 @@ int qemu_start_incoming_migration(const char *uri, Error **errp)
 
 void process_incoming_migration(QEMUFile *f)
 {
-    if (qemu_loadvm_state(f) < 0) {
+    int ret;
+
+    ret = qemu_loadvm_state(f);
+    qemu_fclose(f);
+    if (ret < 0) {
         fprintf(stderr, "load of migration failed\n");
         exit(0);
     }
-- 
1.7.12

[Qemu-devel] [PATCH 8/9] migration: handle EAGAIN while reading QEMUFile

[Paolo Bonzini]

This will never happen right now (the assertion would fail).  The
next patch will set the socket or pipe in non-blocking mode, thus
enabling this part of the code.

Coroutines can just stop whenever they want with qemu_coroutine_yield.
As soon as select tells the main loop that the migration stream is
readable, the coroutine is re-entered directly in qemu_get_buffer,
where it will read more data and pass it to the loading routines.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 savevm.c | 30 ++++++++++++++++++++++++------
 1 file modificato, 24 inserzioni(+), 6 rimozioni(-)

diff --git a/savevm.c b/savevm.c
index 949c38c..90e0cfb 100644
--- a/savevm.c
+++ b/savevm.c
@@ -199,13 +199,22 @@ static int socket_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
     QEMUFileSocket *s = opaque;
     ssize_t len;
 
-    do {
+    for (;;) {
         len = qemu_recv(s->fd, buf, size, 0);
-    } while (len == -1 && socket_error() == EINTR);
+        if (len != -1) {
+            break;
+        }
+        if (errno == EAGAIN) {
+            assert(qemu_in_coroutine());
+            qemu_coroutine_yield();
+        } else if (errno != EINTR) {
+            break;
+        }
+    }
 
-    if (len == -1)
+    if (len == -1) {
         len = -socket_error();
-
+    }
     return len;
 }
 
@@ -236,10 +245,19 @@ static int stdio_get_buffer(void *opaque, uint8_t *buf, int64_t pos, int size)
     FILE *fp = s->stdio_file;
     int bytes;
 
-    do {
+    for (;;) {
         clearerr(fp);
         bytes = fread(buf, 1, size, fp);
-    } while ((bytes == 0) && ferror(fp) && (errno == EINTR));
+        if (bytes != 0 || !ferror(fp)) {
+            break;
+        }
+        if (errno == EAGAIN) {
+            assert(qemu_in_coroutine());
+            qemu_coroutine_yield();
+        } else if (errno != EINTR) {
+            break;
+        }
+    }
     return bytes;
 }
 
-- 
1.7.12

[Qemu-devel] [PATCH 9/9] migration: move process_incoming_migration to a coroutine

[Paolo Bonzini]

The final part of incoming migration, which now consists of
process_incoming_migration for all protocols, is thus made non-blocking.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 migration.c | 21 ++++++++++++++++++++-
 1 file modificato, 20 inserzioni(+). 1 rimozione(-)

diff --git a/migration.c b/migration.c
index 2bc4bac..66390bb 100644
--- a/migration.c
+++ b/migration.c
@@ -86,11 +86,13 @@ int qemu_start_incoming_migration(const char *uri, Error **errp)
     return ret;
 }
 
-void process_incoming_migration(QEMUFile *f)
+static void process_incoming_migration_co(void *opaque)
 {
+    QEMUFile *f = opaque;
     int ret;
 
     ret = qemu_loadvm_state(f);
+    qemu_set_fd_handler(qemu_get_fd(f), NULL, NULL, NULL);
     qemu_fclose(f);
     if (ret < 0) {
         fprintf(stderr, "load of migration failed\n");
@@ -110,6 +112,23 @@ void process_incoming_migration(QEMUFile *f)
     }
 }
 
+static void enter_migration_coroutine(void *opaque)
+{
+    Coroutine *co = opaque;
+    qemu_coroutine_enter(co, NULL);
+}
+
+void process_incoming_migration(QEMUFile *f)
+{
+    Coroutine *co = qemu_coroutine_create(process_incoming_migration_co);
+    int fd = qemu_get_fd(f);
+
+    assert(fd != -1);
+    socket_set_nonblock(fd);
+    qemu_set_fd_handler(fd, enter_migration_coroutine, NULL, co);
+    qemu_coroutine_enter(co, f);
+}
+
 /* amount of nanoseconds we are willing to wait for migration to be down.
  * the choice of nanoseconds is because it is the maximum resolution that
  * get_clock() can achieve. It is an internal measure. All user-visible
-- 
1.7.12