[Qemu-devel] [PATCH FOR REVIEW] migration to/from file (v3 aio) -- additions

[Uri Lublin]

Hello,

The first patch-for-review that implemented migration-to-file using aio, used 
polling to figure out an aio operation completed.

The attached two patches make the migration-to-file not poll for aio completion, 
but use signal instead (SIGUSR1).

As Anthony suggested I've added a set_fd callback to the fd-migration code.
Now upon EAGAIN we do not automatically add the fd to qemu-select rfds set (but 
the default callback does that). Instead we call set_fd callback. Most migration 
protocols would use the default. Current migration-to-file implementation (v3 
aio) fake EAGAIN, and thus make its set_fd empty. When aio is completed a signal 
is received and only then the fd is added to the qemu-select rfds.

This version is quite slow (takes a few minutes to save a stopped 128M guest). I 
think if we add more aio calls (currently only 1), we can get better performance.

Alternatively we can try to open a pipe and create a "writing-thread" use the 
fd-migration code as is (v4).

Regards,
    Uri.

>From 41ff718ab6f08a83e81c45e4da6ec5b250680a14 Mon Sep 17 00:00:00 2001
From: Uri Lublin <address@hidden>
Date: Tue, 24 Feb 2009 07:03:40 +0200
Subject: [PATCH FOR REVIEW 1/2] migration: adding set_fd callback to 
fd-migration

The default fd_migrate_set_fd just add the fd to "qemu-big-select".
Can be implemented differently for other migration protocols.

Signed-off-by: Uri Lublin <address@hidden>
---
 migration-exec.c |    1 +
 migration-tcp.c  |    1 +
 migration.c      |    9 +++++++--
 migration.h      |    3 +++
 4 files changed, 12 insertions(+), 2 deletions(-)

diff --git a/migration-exec.c b/migration-exec.c
index 6ed322a..ede63ae 100644
--- a/migration-exec.c
+++ b/migration-exec.c
@@ -84,6 +84,7 @@ MigrationState *exec_start_outgoing_migration(const char 
*command,
     s->close = exec_close;
     s->get_error = file_errno;
     s->write = file_write;
+    s->set_fd =  migrate_fd_set_fd;
     s->mig_state.cancel = migrate_fd_cancel;
     s->mig_state.get_status = migrate_fd_get_status;
     s->mig_state.release = migrate_fd_release;
diff --git a/migration-tcp.c b/migration-tcp.c
index 3f5b104..5772d21 100644
--- a/migration-tcp.c
+++ b/migration-tcp.c
@@ -93,6 +93,7 @@ MigrationState *tcp_start_outgoing_migration(const char 
*host_port,
     s->get_error = socket_errno;
     s->write = socket_write;
     s->close = tcp_close;
+    s->set_fd = migrate_fd_set_fd;
     s->mig_state.cancel = migrate_fd_cancel;
     s->mig_state.get_status = migrate_fd_get_status;
     s->mig_state.release = migrate_fd_release;
diff --git a/migration.c b/migration.c
index 234dcf6..e4961ea 100644
--- a/migration.c
+++ b/migration.c
@@ -169,6 +169,11 @@ void migrate_fd_put_notify(void *opaque)
     qemu_file_put_notify(s->file);
 }
 
+int migrate_fd_set_fd(FdMigrationState *s)
+{
+    return qemu_set_fd_handler2(s->fd, NULL, NULL, migrate_fd_put_notify, s);
+}
+
 ssize_t migrate_fd_put_buffer(void *opaque, const void *data, size_t size)
 {
     FdMigrationState *s = opaque;
@@ -176,13 +181,13 @@ ssize_t migrate_fd_put_buffer(void *opaque, const void 
*data, size_t size)
 
     do {
         ret = s->write(s, data, size);
-    } while (ret == -1 && ((s->get_error(s)) == EINTR || (s->get_error(s)) == 
EWOULDBLOCK));
+    } while (ret == -1 && (s->get_error(s) == EINTR));
 
     if (ret == -1)
         ret = -(s->get_error(s));
 
     if (ret == -EAGAIN)
-        qemu_set_fd_handler2(s->fd, NULL, NULL, migrate_fd_put_notify, s);
+        s->set_fd(s);
 
     return ret;
 }
diff --git a/migration.h b/migration.h
index ae78792..5c3d5eb 100644
--- a/migration.h
+++ b/migration.h
@@ -42,6 +42,7 @@ struct FdMigrationState
     int (*get_error)(struct FdMigrationState*);
     int (*close)(struct FdMigrationState*);
     int (*write)(struct FdMigrationState*, const void *, size_t);
+    int (*set_fd)(struct FdMigrationState*);
     void *opaque;
 };
 
@@ -79,6 +80,8 @@ void migrate_fd_cleanup(FdMigrationState *s);
 
 void migrate_fd_put_notify(void *opaque);
 
+int migrate_fd_set_fd(FdMigrationState *s);
+
 ssize_t migrate_fd_put_buffer(void *opaque, const void *data, size_t size);
 
 void migrate_fd_connect(FdMigrationState *s);
-- 
1.6.0.6

>From f765d3baaf10174e7942f15dd801975125bc3d34 Mon Sep 17 00:00:00 2001
From: Uri Lublin <address@hidden>
Date: Tue, 24 Feb 2009 07:12:18 +0200
Subject: [PATCH 2/2 FOR REVIEW] migration to file: use signal for async-write 
notification

Instead of looping by io-handler called by writeable fd, do not
register handler, and use signal SIGUSR1 for aio completion notifications.

Signed-off-by: Uri Lublin <address@hidden>
---
 migration-file.c |  124 +++++++++++++++++++++++++++++++++++++++++++++++++-----
 1 files changed, 113 insertions(+), 11 deletions(-)

diff --git a/migration-file.c b/migration-file.c
index 6d3dbf5..02726fc 100644
--- a/migration-file.c
+++ b/migration-file.c
@@ -38,17 +38,19 @@ typedef struct FileMigrationState_s {
     off_t write_offset;
     struct qemu_paiocb aiocb;
     int aio_in_use;
+    int aio_rfd, aio_wfd;
 } FileMigrationState;
 
+static FileMigrationState *fm = NULL;
 
-static int file_errno(FdMigrationState *fds)
-{
-    return errno;
-}
 
-static void mig_file_check_aiocb(FileMigrationState *s)
+/* returns 0 if done nothing
+ * returns >0 if an aio operation was completed
+ */
+static int mig_file_check_aiocb(FileMigrationState *s)
 {
     ssize_t ret;
+    int rc = 0;
 
     if (s->aio_in_use) {
         ret = qemu_paio_return(&s->aiocb);
@@ -56,17 +58,102 @@ static void mig_file_check_aiocb(FileMigrationState *s)
                 &s->aiocb, s->aiocb.aio_nbytes, ret);
 
         if (ret == -EINPROGRESS)
-            return;
-        if (ret == s->aiocb.aio_nbytes) {
+            ; /* do nothing --> return 0 */
+        else if(ret == s->aiocb.aio_nbytes) {
             s->aio_in_use = 0;
+            rc = 1;
         }
         else {
-            dprintf("posix aio write failed, returned %ld expected %ld\n",
+            fprintf(stderr, "posix aio write failed, returned %ld expected 
%ld\n",
                     ret, s->aiocb.aio_nbytes);
             if (s->fds.state == MIG_STATE_ACTIVE)
                 s->fds.state = MIG_STATE_ERROR;
         }
     }
+    return rc;
+}
+
+static void mig_file_aio_read(void *opaque)
+{
+    FileMigrationState *s = (FileMigrationState *)opaque;
+    ssize_t len;
+
+    dprintf("mig_file_aio_read: IN\n");
+    /* read all bytes from signal pipe */
+    for (;;) {
+        char bytes[16];
+
+        len = read(s->aio_rfd, bytes, sizeof(bytes));
+        if (len == -1 && errno == EINTR)
+            continue; /* try again */
+        if (len == sizeof(bytes))
+            continue; /* more to read */
+        break;
+    }
+    
+    if (mig_file_check_aiocb(s))
+        migrate_fd_set_fd(&s->fds);
+}
+
+static void mig_file_aio_signal_handler(int signum)
+{
+    if (fm) {
+        char byte = 0;
+
+        write(fm->aio_wfd, &byte, sizeof(byte));
+    }
+
+    qemu_service_io();
+}
+
+static int mig_file_aio_init(FileMigrationState *s)
+{
+    struct sigaction act;
+    int fds[2];
+ 
+    sigfillset(&act.sa_mask);
+    act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
+    act.sa_handler = mig_file_aio_signal_handler;
+    sigaction(SIGUSR1, &act, NULL);
+
+    if (pipe(fds) == -1) {
+        perror("file-migration: failed to create pipe\n");
+        return -errno;
+    }
+
+    s->aio_rfd = fds[0];
+    s->aio_wfd = fds[1];
+
+    fcntl(s->aio_rfd, F_SETFL, O_NONBLOCK);
+    fcntl(s->aio_wfd, F_SETFL, O_NONBLOCK);
+
+    qemu_set_fd_handler2(s->aio_rfd, NULL, mig_file_aio_read, NULL, s);
+
+    return 0;
+}
+
+static void mig_file_aio_cleanup(FileMigrationState *s)
+{
+    struct sigaction act;
+
+    sigfillset(&act.sa_mask);
+    act.sa_flags = 0; /* do not restart syscalls to interrupt select() */
+    act.sa_handler = SIG_DFL;
+    sigaction(SIGUSR1, &act, NULL);
+
+    qemu_aio_set_fd_handler(s->aio_rfd, NULL, NULL, NULL, NULL);
+    close(s->aio_rfd);
+    close(s->aio_wfd);
+    s->aio_rfd = s->aio_wfd = -1;
+    
+    fm = NULL;
+}
+
+
+
+static int file_errno(FdMigrationState *fds)
+{
+    return errno;
 }
 
 static struct qemu_paiocb* mig_file_get_aiocb(FileMigrationState *s)
@@ -115,7 +202,7 @@ static int write_async(FileMigrationState *s, const void 
*buf, size_t size)
 
     paiocb->aio_fildes = s->fds.fd;
     paiocb->aio_nbytes = size;
-    paiocb->ev_signo = 0; /* No notification is needed */
+    paiocb->ev_signo = SIGUSR1;
     paiocb->aio_offset = s->write_offset;
 
     s->write_offset += size;
@@ -150,6 +237,11 @@ static int file_write(FdMigrationState *fds, const void * 
buf, size_t size)
     return offset;
 }
 
+static int file_set_fd(FdMigrationState *s)
+{
+    return 0;
+}
+
 static int file_close(FdMigrationState *fds)
 {
     FileMigrationState *s = (FileMigrationState*)fds;
@@ -162,6 +254,8 @@ static int file_close(FdMigrationState *fds)
             break;
         sleep(1);
     }
+
+    mig_file_aio_cleanup(s);
     close(fds->fd);
 
     free_aio_buf(s);
@@ -178,20 +272,25 @@ MigrationState *file_start_outgoing_migration(const char 
*filename,
     int fd;
 
     s = qemu_mallocz(sizeof(*s));
+    fm = s;
 
     fd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0600);
     if (fd < 0) {
         perror("file_migration: failed to open filename");
         term_printf("file_migration: failed to open filename %s\n", filename);
-        goto err;
+        goto err1;
     }
 
+    if (mig_file_aio_init(s) < 0)
+        goto err2;
+
     fds = &s->fds;
 
     fds->fd = fd;
     fds->close = file_close;
     fds->get_error = file_errno;
     fds->write = file_write;
+    fds->set_fd = file_set_fd;
     fds->mig_state.cancel = migrate_fd_cancel;
     fds->mig_state.get_status = migrate_fd_get_status;
     fds->mig_state.release = migrate_fd_release;
@@ -209,8 +308,11 @@ MigrationState *file_start_outgoing_migration(const char 
*filename,
     migrate_fd_connect(fds);
     return &fds->mig_state;
 
-err:
+err2:
+    close(fd);
+err1:
     qemu_free(s);
+    fm = NULL;
     return NULL;
 }
 
-- 
1.6.0.6