Patchwork [14/41] migration: cleanup migration (including thread) in the iothread

login
register
mail settings
Submitter Paolo Bonzini
Date Feb. 15, 2013, 5:46 p.m.
Message ID <1360950433-17106-15-git-send-email-pbonzini@redhat.com>
Download mbox | patch
Permalink /patch/220815/
State New
Headers show

Comments

Paolo Bonzini - Feb. 15, 2013, 5:46 p.m.
Perform final cleanup in a bottom half, and add joining the thread to
the series of cleanup actions.

Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
---
 include/migration/migration.h |    1 +
 migration.c                   |   37 +++++++++++++++++++------------------
 2 files changed, 20 insertions(+), 18 deletions(-)
Orit Wasserman - Feb. 18, 2013, 10:53 a.m.
On 02/15/2013 07:46 PM, Paolo Bonzini wrote:
> Perform final cleanup in a bottom half, and add joining the thread to
> the series of cleanup actions.
> 
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  include/migration/migration.h |    1 +
>  migration.c                   |   37 +++++++++++++++++++------------------
>  2 files changed, 20 insertions(+), 18 deletions(-)
> 
> diff --git a/include/migration/migration.h b/include/migration/migration.h
> index 4928642..d78bbbb 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -38,6 +38,7 @@ struct MigrationState
>      size_t buffer_size;
>      size_t buffer_capacity;
>      QemuThread thread;
> +    QEMUBH *cleanup_bh;
>  
>      QEMUFile *file;
>      int fd;
> diff --git a/migration.c b/migration.c
> index a7f619b..8abaaea 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -260,8 +260,13 @@ void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
>  
>  /* shared migration helpers */
>  
> -static void migrate_fd_cleanup(MigrationState *s)
> +static void migrate_fd_cleanup(void *opaque)
>  {
> +    MigrationState *s = opaque;
> +
> +    qemu_bh_delete(s->cleanup_bh);
> +    s->cleanup_bh = NULL;
> +
>      if (s->file) {
>          DPRINTF("closing file\n");
>          qemu_fclose(s->file);
> @@ -281,15 +286,9 @@ static void migrate_fd_cleanup(MigrationState *s)
>  void migrate_fd_error(MigrationState *s)
>  {
>      DPRINTF("setting error state\n");
> -    __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_ERROR);
> -    migrate_fd_cleanup(s);
> -}
> -
> -static void migrate_fd_completed(MigrationState *s)
> -{
> -    DPRINTF("setting completed state\n");
> -    __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_COMPLETED);
> -    migrate_fd_cleanup(s);
> +    assert(s->file == NULL);
> +    s->state = MIG_STATE_ERROR;
> +    notifier_list_notify(&migration_state_notifiers, s);
>  }
>  
>  static ssize_t migrate_fd_put_buffer(MigrationState *s, const void *data,
> @@ -316,7 +315,6 @@ static void migrate_fd_cancel(MigrationState *s)
>      DPRINTF("cancelling migration\n");
>  
>      __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_CANCELLED);
> -    migrate_fd_cleanup(s);
>  }
>  
>  int migrate_fd_close(MigrationState *s)
> @@ -580,6 +578,11 @@ static int buffered_close(void *opaque)
>  
>      DPRINTF("closing\n");
>  
> +    qemu_mutex_unlock_iothread();
> +    qemu_thread_join(&s->thread);
> +    qemu_mutex_lock_iothread();
> +    assert (s->state != MIG_STATE_ACTIVE);
> +
>      return migrate_fd_close(s);
>  }
>  
> @@ -692,13 +695,9 @@ static void *buffered_file_thread(void *opaque)
>          }
>          buffered_flush(s);
>          if (qemu_file_get_error(s->file)) {
> -            qemu_mutex_lock_iothread();
> -            migrate_fd_error(s);
> -            qemu_mutex_unlock_iothread();
> +            __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_ERROR);
>          } else if (last_round && s->buffer_size == 0) {
> -            qemu_mutex_lock_iothread();
> -            migrate_fd_completed(s);
> -            qemu_mutex_unlock_iothread();
> +            __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_COMPLETED);
>          }
>      }
>  
> @@ -714,6 +713,7 @@ static void *buffered_file_thread(void *opaque)
>              vm_start();
>          }
>      }
> +    qemu_bh_schedule(s->cleanup_bh);
>      qemu_mutex_unlock_iothread();
>  
>      g_free(s->buffer);
> @@ -739,9 +739,10 @@ void migrate_fd_connect(MigrationState *s)
>  
>      s->xfer_limit = s->bandwidth_limit / XFER_LIMIT_RATIO;
>  
> +    s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
>      s->file = qemu_fopen_ops(s, &buffered_file_ops);
>  
>      qemu_thread_create(&s->thread, buffered_file_thread, s,
> -                       QEMU_THREAD_DETACHED);
> +                       QEMU_THREAD_JOINABLE);
>      notifier_list_notify(&migration_state_notifiers, s);
>  }
> 

Reviewed-by: Orit Wasserman <owasserm@redhat.com>
Juan Quintela - Feb. 22, 2013, 10:50 a.m.
Paolo Bonzini <pbonzini@redhat.com> wrote:
> Perform final cleanup in a bottom half, and add joining the thread to
> the series of cleanup actions.
>
> Signed-off-by: Paolo Bonzini <pbonzini@redhat.com>
> ---
>  include/migration/migration.h |    1 +
>  migration.c                   |   37 +++++++++++++++++++------------------
>  2 files changed, 20 insertions(+), 18 deletions(-)
>

Reviewed-by: Juan Quintela <quintela@redhat.com>

Previously, I have had trouble with bottom half here because they added
a lot of latency between we made it runnable until they decided to run.
This was the reason that I didn't used them on my previous tries.

But I haven't hit the problem with this series.

Patch

diff --git a/include/migration/migration.h b/include/migration/migration.h
index 4928642..d78bbbb 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -38,6 +38,7 @@  struct MigrationState
     size_t buffer_size;
     size_t buffer_capacity;
     QemuThread thread;
+    QEMUBH *cleanup_bh;
 
     QEMUFile *file;
     int fd;
diff --git a/migration.c b/migration.c
index a7f619b..8abaaea 100644
--- a/migration.c
+++ b/migration.c
@@ -260,8 +260,13 @@  void qmp_migrate_set_capabilities(MigrationCapabilityStatusList *params,
 
 /* shared migration helpers */
 
-static void migrate_fd_cleanup(MigrationState *s)
+static void migrate_fd_cleanup(void *opaque)
 {
+    MigrationState *s = opaque;
+
+    qemu_bh_delete(s->cleanup_bh);
+    s->cleanup_bh = NULL;
+
     if (s->file) {
         DPRINTF("closing file\n");
         qemu_fclose(s->file);
@@ -281,15 +286,9 @@  static void migrate_fd_cleanup(MigrationState *s)
 void migrate_fd_error(MigrationState *s)
 {
     DPRINTF("setting error state\n");
-    __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_ERROR);
-    migrate_fd_cleanup(s);
-}
-
-static void migrate_fd_completed(MigrationState *s)
-{
-    DPRINTF("setting completed state\n");
-    __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_COMPLETED);
-    migrate_fd_cleanup(s);
+    assert(s->file == NULL);
+    s->state = MIG_STATE_ERROR;
+    notifier_list_notify(&migration_state_notifiers, s);
 }
 
 static ssize_t migrate_fd_put_buffer(MigrationState *s, const void *data,
@@ -316,7 +315,6 @@  static void migrate_fd_cancel(MigrationState *s)
     DPRINTF("cancelling migration\n");
 
     __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_CANCELLED);
-    migrate_fd_cleanup(s);
 }
 
 int migrate_fd_close(MigrationState *s)
@@ -580,6 +578,11 @@  static int buffered_close(void *opaque)
 
     DPRINTF("closing\n");
 
+    qemu_mutex_unlock_iothread();
+    qemu_thread_join(&s->thread);
+    qemu_mutex_lock_iothread();
+    assert (s->state != MIG_STATE_ACTIVE);
+
     return migrate_fd_close(s);
 }
 
@@ -692,13 +695,9 @@  static void *buffered_file_thread(void *opaque)
         }
         buffered_flush(s);
         if (qemu_file_get_error(s->file)) {
-            qemu_mutex_lock_iothread();
-            migrate_fd_error(s);
-            qemu_mutex_unlock_iothread();
+            __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_ERROR);
         } else if (last_round && s->buffer_size == 0) {
-            qemu_mutex_lock_iothread();
-            migrate_fd_completed(s);
-            qemu_mutex_unlock_iothread();
+            __sync_val_compare_and_swap(&s->state, MIG_STATE_ACTIVE, MIG_STATE_COMPLETED);
         }
     }
 
@@ -714,6 +713,7 @@  static void *buffered_file_thread(void *opaque)
             vm_start();
         }
     }
+    qemu_bh_schedule(s->cleanup_bh);
     qemu_mutex_unlock_iothread();
 
     g_free(s->buffer);
@@ -739,9 +739,10 @@  void migrate_fd_connect(MigrationState *s)
 
     s->xfer_limit = s->bandwidth_limit / XFER_LIMIT_RATIO;
 
+    s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
     s->file = qemu_fopen_ops(s, &buffered_file_ops);
 
     qemu_thread_create(&s->thread, buffered_file_thread, s,
-                       QEMU_THREAD_DETACHED);
+                       QEMU_THREAD_JOINABLE);
     notifier_list_notify(&migration_state_notifiers, s);
 }