[Qemu-devel] [RFC PATCH v3 5/5] Making iothread block for migrate_cancel

[Umesh Deshpande]

Following patch makes iothread wait until the migration thread responds to the
migrate_cancel request and terminates its execution.

Signed-off-by: Umesh Deshpande <address@hidden>
---
 buffered_file.c     |   13 ++++++++++++-
 buffered_file.h     |    3 +++
 hw/hw.h             |    5 ++++-
 migration.c         |    1 +
 qemu-thread-posix.c |   10 ++++++++++
 qemu-thread.h       |    1 +
 savevm.c            |   31 +++++++++++++++++++++----------
 7 files changed, 52 insertions(+), 12 deletions(-)

diff --git a/buffered_file.c b/buffered_file.c
index bdcdf42..405b17f 100644
--- a/buffered_file.c
+++ b/buffered_file.c
@@ -223,6 +223,16 @@ static int64_t buffered_get_rate_limit(void *opaque)
     return s->xfer_limit;
 }
 
+static void buffered_wait_for_cancel(void *opaque)
+{
+    QEMUFileBuffered *s = opaque;
+    QemuThread thread = s->thread;
+
+    qemu_mutex_unlock_iothread();
+    qemu_thread_join(thread);
+    qemu_mutex_lock_iothread();
+}
+
 static void *migrate_vm(void *opaque)
 {
     QEMUFileBuffered *s = opaque;
@@ -296,7 +306,8 @@ QEMUFile *qemu_fopen_ops_buffered(void *opaque,
     s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
                              buffered_close, buffered_rate_limit,
                              buffered_set_rate_limit,
-                             buffered_get_rate_limit);
+                             buffered_get_rate_limit,
+                             buffered_wait_for_cancel);
 
     qemu_thread_create(&s->thread, migrate_vm, s);
 
diff --git a/buffered_file.h b/buffered_file.h
index 98d358b..413cc9f 100644
--- a/buffered_file.h
+++ b/buffered_file.h
@@ -20,6 +20,9 @@ typedef ssize_t (BufferedPutFunc)(void *opaque, const void 
*data, size_t size);
 typedef void (BufferedPutReadyFunc)(void *opaque);
 typedef void (BufferedWaitForUnfreezeFunc)(void *opaque);
 typedef int (BufferedCloseFunc)(void *opaque);
+typedef void (BufferedWaitForCancelFunc)(void *opaque);
+
+void wait_for_cancel(void *opaque);
 
 QEMUFile *qemu_fopen_ops_buffered(void *opaque, size_t xfer_limit,
                                   BufferedPutFunc *put_buffer,
diff --git a/hw/hw.h b/hw/hw.h
index 9dd7096..e1d5ea8 100644
--- a/hw/hw.h
+++ b/hw/hw.h
@@ -41,13 +41,15 @@ typedef int (QEMUFileRateLimit)(void *opaque);
  */
 typedef int64_t (QEMUFileSetRateLimit)(void *opaque, int64_t new_rate);
 typedef int64_t (QEMUFileGetRateLimit)(void *opaque);
+typedef void (QEMUFileWaitForCancel)(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);
+                         QEMUFileGetRateLimit *get_rate_limit,
+                         QEMUFileWaitForCancel *wait_for_cancel);
 QEMUFile *qemu_fopen(const char *filename, const char *mode);
 QEMUFile *qemu_fdopen(int fd, const char *mode);
 QEMUFile *qemu_fopen_socket(int fd);
@@ -56,6 +58,7 @@ QEMUFile *qemu_popen_cmd(const char *command, const char 
*mode);
 int qemu_stdio_fd(QEMUFile *f);
 void qemu_fflush(QEMUFile *f);
 int qemu_fclose(QEMUFile *f);
+void qemu_wait_for_cancel(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/migration.c b/migration.c
index b6ba690..0c5a484 100644
--- a/migration.c
+++ b/migration.c
@@ -423,6 +423,7 @@ void migrate_fd_cancel(MigrationState *mig_state)
     DPRINTF("cancelling migration\n");
 
     s->state = MIG_STATE_CANCELLED;
+    qemu_wait_for_cancel(s->file);
 }
 
 void migrate_fd_release(MigrationState *mig_state)
diff --git a/qemu-thread-posix.c b/qemu-thread-posix.c
index 2bd02ef..0d18b35 100644
--- a/qemu-thread-posix.c
+++ b/qemu-thread-posix.c
@@ -115,6 +115,16 @@ void qemu_cond_wait(QemuCond *cond, QemuMutex *mutex)
         error_exit(err, __func__);
 }
 
+void qemu_thread_join(QemuThread thread)
+{
+    int err;
+
+    err = pthread_join(thread.thread, NULL);
+    if (err) {
+        error_exit(err, __func__);
+    }
+}
+
 void qemu_thread_create(QemuThread *thread,
                        void *(*start_routine)(void*),
                        void *arg)
diff --git a/qemu-thread.h b/qemu-thread.h
index 0a73d50..909529f 100644
--- a/qemu-thread.h
+++ b/qemu-thread.h
@@ -30,6 +30,7 @@ void qemu_cond_destroy(QemuCond *cond);
 void qemu_cond_signal(QemuCond *cond);
 void qemu_cond_broadcast(QemuCond *cond);
 void qemu_cond_wait(QemuCond *cond, QemuMutex *mutex);
+void qemu_thread_join(QemuThread thread);
 
 void qemu_thread_create(QemuThread *thread,
                        void *(*start_routine)(void*),
diff --git a/savevm.c b/savevm.c
index f54f555..8003411 100644
--- a/savevm.c
+++ b/savevm.c
@@ -164,6 +164,7 @@ struct QEMUFile {
     QEMUFileRateLimit *rate_limit;
     QEMUFileSetRateLimit *set_rate_limit;
     QEMUFileGetRateLimit *get_rate_limit;
+    QEMUFileWaitForCancel *wait_for_cancel;
     void *opaque;
     int is_write;
 
@@ -261,10 +262,10 @@ QEMUFile *qemu_popen(FILE *stdio_file, const char *mode)
 
     if(mode[0] == 'r') {
         s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_pclose, 
-                                NULL, NULL, NULL);
+                                 NULL, NULL, NULL, NULL);
     } else {
         s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_pclose, 
-                                NULL, NULL, NULL);
+                                 NULL, NULL, NULL, NULL);
     }
     return s->file;
 }
@@ -310,10 +311,10 @@ QEMUFile *qemu_fdopen(int fd, const char *mode)
 
     if(mode[0] == 'r') {
         s->file = qemu_fopen_ops(s, NULL, stdio_get_buffer, stdio_fclose, 
-                                NULL, NULL, NULL);
+                                 NULL, NULL, NULL, NULL);
     } else {
         s->file = qemu_fopen_ops(s, stdio_put_buffer, NULL, stdio_fclose, 
-                                NULL, NULL, NULL);
+                                 NULL, NULL, NULL, NULL);
     }
     return s->file;
 
@@ -328,7 +329,7 @@ QEMUFile *qemu_fopen_socket(int fd)
 
     s->fd = fd;
     s->file = qemu_fopen_ops(s, NULL, socket_get_buffer, socket_close, 
-                            NULL, NULL, NULL);
+                             NULL, NULL, NULL, NULL);
     return s->file;
 }
 
@@ -366,10 +367,10 @@ QEMUFile *qemu_fopen(const char *filename, const char 
*mode)
     
     if(mode[0] == 'w') {
         s->file = qemu_fopen_ops(s, file_put_buffer, NULL, stdio_fclose, 
-                                NULL, NULL, NULL);
+                                 NULL, NULL, NULL, NULL);
     } else {
         s->file = qemu_fopen_ops(s, NULL, file_get_buffer, stdio_fclose, 
-                              NULL, NULL, NULL);
+                                 NULL, NULL, NULL, NULL);
     }
     return s->file;
 fail:
@@ -398,8 +399,9 @@ 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);
+                              NULL, NULL, NULL, NULL);
+    return qemu_fopen_ops(bs, NULL, block_get_buffer, bdrv_fclose, NULL, NULL,
+                          NULL, NULL);
 }
 
 QEMUFile *qemu_fopen_ops(void *opaque, QEMUFilePutBufferFunc *put_buffer,
@@ -407,7 +409,8 @@ QEMUFile *qemu_fopen_ops(void *opaque, 
QEMUFilePutBufferFunc *put_buffer,
                          QEMUFileCloseFunc *close,
                          QEMUFileRateLimit *rate_limit,
                          QEMUFileSetRateLimit *set_rate_limit,
-                         QEMUFileGetRateLimit *get_rate_limit)
+                         QEMUFileGetRateLimit *get_rate_limit,
+                         QEMUFileWaitForCancel *wait_for_cancel)
 {
     QEMUFile *f;
 
@@ -420,6 +423,7 @@ QEMUFile *qemu_fopen_ops(void *opaque, 
QEMUFilePutBufferFunc *put_buffer,
     f->rate_limit = rate_limit;
     f->set_rate_limit = set_rate_limit;
     f->get_rate_limit = get_rate_limit;
+    f->wait_for_cancel = wait_for_cancel;
     f->is_write = 0;
 
     return f;
@@ -481,6 +485,13 @@ int qemu_fclose(QEMUFile *f)
     return ret;
 }
 
+void qemu_wait_for_cancel(QEMUFile *f)
+{
+    if (f->wait_for_cancel) {
+        f->wait_for_cancel(f->opaque);
+    }
+}
+
 void qemu_put_buffer(QEMUFile *f, const uint8_t *buf, int size)
 {
     int l;
-- 
1.7.4.1