Merge remote-tracking branch 'bonzini/migr-coroutine' into staging

Signed-off-by: Anthony Liguori <aliguori@us.ibm.com>

* bonzini/migr-coroutine:
  migration: move process_incoming_migration to a coroutine
  migration: handle EAGAIN while reading QEMUFile
  migration: move qemu_fclose to process_incoming_migration
  migration: close socket QEMUFile from socket_close
  migration: xxx_close will only be called once
  migration: use closesocket, not close
  migration: use migrate_fd_close in migrate_fd_cleanup
  migration: clean up server sockets and handlers before invoking process_incoming_migration
  migration: replace qemu_stdio_fd with qemu_get_fd
  migration: add qemu_get_fd
  migration: consolidate QEMUFile methods in a single QEMUFileOps struct
  migration: unify stdio-based QEMUFile operations

1 files changed, 110 insertions, 78 deletions

diff --git a/savevm.c b/savevm.c
index 43d3d1bbeb..5d04d59688 100644
--- a/savevm.c
+++ b/savevm.c
@@ -163,12 +163,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;
 
@@ -193,28 +188,52 @@ 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;
     ssize_t len;
 
-    do {
+    for (;;) {
         len = qemu_recv(s->fd, buf, size, 0);
-    } while (len == -1 && socket_error() == EINTR);
+        if (len != -1) {
+            break;
+        }
+        if (socket_error() == EAGAIN) {
+            assert(qemu_in_coroutine());
+            qemu_coroutine_yield();
+        } else if (socket_error() != EINTR) {
+            break;
+        }
+    }
 
-    if (len == -1)
+    if (len == -1) {
         len = -socket_error();
-
+    }
     return len;
 }
 
 static int socket_close(void *opaque)
 {
     QEMUFileSocket *s = opaque;
+    closesocket(s->fd);
     g_free(s);
     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;
@@ -227,10 +246,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;
 }
 
@@ -257,6 +285,18 @@ static int stdio_fclose(void *opaque)
     return ret;
 }
 
+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
+};
+
 QEMUFile *qemu_popen(FILE *stdio_file, const char *mode)
 {
     QEMUFileStdio *s;
@@ -271,11 +311,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;
 }
@@ -292,16 +330,17 @@ 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);
+static const QEMUFileOps stdio_file_read_ops = {
+    .get_fd =     stdio_get_fd,
+    .get_buffer = stdio_get_buffer,
+    .close =      stdio_fclose
+};
 
-    return fd;
-}
+static const QEMUFileOps stdio_file_write_ops = {
+    .get_fd =     stdio_get_fd,
+    .put_buffer = stdio_put_buffer,
+    .close =      stdio_fclose
+};
 
 QEMUFile *qemu_fdopen(int fd, const char *mode)
 {
@@ -320,11 +359,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;
 
@@ -333,31 +370,21 @@ fail:
     return NULL;
 }
 
+static const QEMUFileOps socket_read_ops = {
+    .get_fd =     socket_get_fd,
+    .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;
 }
 
-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;
@@ -376,11 +403,9 @@ 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, 
-				 NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_file_write_ops);
     } else {
-        s->file = qemu_fopen_ops(s, NULL, file_get_buffer, stdio_fclose, 
-			       NULL, NULL, NULL);
+        s->file = qemu_fopen_ops(s, &stdio_file_read_ops);
     }
     return s->file;
 fail:
@@ -405,32 +430,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;
@@ -453,11 +477,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;
         }
@@ -471,7 +495,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)
@@ -484,7 +508,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;
@@ -495,6 +519,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
@@ -508,8 +540,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;
         }
@@ -526,7 +558,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)
@@ -673,16 +705,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;
 }
@@ -691,8 +723,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;
 }