Re: [PATCH 5/5] Separate migration thread

[Marcelo Tosatti <mtosatti@redhat.com>]

On Sat, Aug 27, 2011 at 02:09:48PM -0400, Umesh Deshpande wrote:
> This patch creates a separate thread for the guest migration on the source side.
> All exits (on completion/error) from the migration thread are handled by a
> bottom handler, which is called from the iothread.
> 
> Signed-off-by: Umesh Deshpande <udeshpan@redhat.com>
> ---
>  buffered_file.c     |   76 ++++++++++++++++++++----------------
>  migration.c         |  105 ++++++++++++++++++++++++++++++--------------------
>  migration.h         |    8 ++++
>  qemu-thread-posix.c |   10 +++++
>  qemu-thread.h       |    1 +
>  5 files changed, 124 insertions(+), 76 deletions(-)
> 
> diff --git a/buffered_file.c b/buffered_file.c
> index 41b42c3..c31852e 100644
> --- a/buffered_file.c
> +++ b/buffered_file.c
> @@ -16,6 +16,8 @@
>  #include "qemu-timer.h"
>  #include "qemu-char.h"
>  #include "buffered_file.h"
> +#include "migration.h"
> +#include "qemu-thread.h"
>  
>  //#define DEBUG_BUFFERED_FILE
>  
> @@ -28,13 +30,14 @@ typedef struct QEMUFileBuffered
>      void *opaque;
>      QEMUFile *file;
>      int has_error;
> +    int closed;
>      int freeze_output;
>      size_t bytes_xfer;
>      size_t xfer_limit;
>      uint8_t *buffer;
>      size_t buffer_size;
>      size_t buffer_capacity;
> -    QEMUTimer *timer;
> +    QemuThread thread;
>  } QEMUFileBuffered;
>  
>  #ifdef DEBUG_BUFFERED_FILE
> @@ -155,14 +158,6 @@ static int buffered_put_buffer(void *opaque, const uint8_t *buf, int64_t pos, in
>          offset = size;
>      }
>  
> -    if (pos == 0 && size == 0) {
> -        DPRINTF("file is ready\n");
> -        if (s->bytes_xfer <= s->xfer_limit) {
> -            DPRINTF("notifying client\n");
> -            s->put_ready(s->opaque);
> -        }
> -    }
> -
>      return offset;
>  }
>  
> @@ -173,22 +168,25 @@ static int buffered_close(void *opaque)
>  
>      DPRINTF("closing\n");
>  
> -    while (!s->has_error && s->buffer_size) {
> -        buffered_flush(s);
> -        if (s->freeze_output)
> -            s->wait_for_unfreeze(s);
> -    }
> +    s->closed = 1;
>  
> -    ret = s->close(s->opaque);
> +    qemu_mutex_unlock_migrate_ram();
> +    qemu_mutex_unlock_iothread();

This is using the ram mutex to protect migration thread specific data.
A new lock should be introduced for that purpose.

> -    qemu_del_timer(s->timer);
> -    qemu_free_timer(s->timer);
> +    qemu_thread_join(&s->thread);
> +    /* Waits for the completion of the migration thread */
> +
> +    qemu_mutex_lock_iothread();
> +    qemu_mutex_lock_migrate_ram();
> +
> +    ret = s->close(s->opaque);
>      qemu_free(s->buffer);
>      qemu_free(s);
>  
>      return ret;
>  }
>  
> +
>  static int buffered_rate_limit(void *opaque)
>  {
>      QEMUFileBuffered *s = opaque;
> @@ -228,26 +226,37 @@ static int64_t buffered_get_rate_limit(void *opaque)
>      return s->xfer_limit;
>  }
>  
> -static void buffered_rate_tick(void *opaque)
> +static void *migrate_vm(void *opaque)
>  {

buffered_file.c was generic code that has now become migration specific
(although migration was the only user). So it should either stop
pretending to be generic code, by rename to migration_thread.c along
with un-exporting interfaces, or it should remain generic and therefore
all migration specific knowledge moved somewhere else.

Anthony?

> +    int64_t current_time, expire_time = qemu_get_clock_ms(rt_clock) + 100;
> +    struct timeval tv = { .tv_sec = 0, .tv_usec = 100000};

qemu_get_clock_ms should happen under iothread lock.

> -    if (s->freeze_output)
> -        return;
> +        current_time = qemu_get_clock_ms(rt_clock);
> +        if (!s->closed && (expire_time > current_time)) {
> +            tv.tv_usec = 1000 * (expire_time - current_time);
> +            select(0, NULL, NULL, NULL, &tv);
> +            continue;
> +        }
>  
> -    s->bytes_xfer = 0;
> +        s->bytes_xfer = 0;
>  
> -    buffered_flush(s);
> +        expire_time = qemu_get_clock_ms(rt_clock) + 100;
> +        if (!s->closed) {
> +            s->put_ready(s->opaque);
> +        } else {
> +            buffered_flush(s);
> +        }
> +    }
>  
> -    /* Add some checks around this */
> -    s->put_ready(s->opaque);
> +    return NULL;
>  }
>  
>  QEMUFile *qemu_fopen_ops_buffered(void *opaque,
> @@ -267,15 +276,14 @@ QEMUFile *qemu_fopen_ops_buffered(void *opaque,
>      s->put_ready = put_ready;
>      s->wait_for_unfreeze = wait_for_unfreeze;
>      s->close = close;
> +    s->closed = 0;
>  
>      s->file = qemu_fopen_ops(s, buffered_put_buffer, NULL,
>                               buffered_close, buffered_rate_limit,
>                               buffered_set_rate_limit,
> -			     buffered_get_rate_limit);
> -
> -    s->timer = qemu_new_timer_ms(rt_clock, buffered_rate_tick, s);
> +                             buffered_get_rate_limit);
>  
> -    qemu_mod_timer(s->timer, qemu_get_clock_ms(rt_clock) + 100);
> +    qemu_thread_create(&s->thread, migrate_vm, s);
>  
>      return s->file;
>  }
> diff --git a/migration.c b/migration.c
> index af3a1f2..5df186d 100644
> --- a/migration.c
> +++ b/migration.c
> @@ -149,10 +149,12 @@ int do_migrate_set_speed(Monitor *mon, const QDict *qdict, QObject **ret_data)
>      }
>      max_throttle = d;
>  
> +    qemu_mutex_lock_migrate_ram();
>      s = migrate_to_fms(current_migration);
>      if (s && s->file) {
>          qemu_file_set_rate_limit(s->file, max_throttle);
>      }
> +    qemu_mutex_unlock_migrate_ram();

This lock protects the RAMlist, and only the RAMlist, but here its      
being used to protect migration thread data. As noted above, a new lock 
should be introduced.

>      int ret = 0;
>  
> -    qemu_set_fd_handler2(s->fd, NULL, NULL, NULL, NULL);
> -
>      if (s->file) {
>          DPRINTF("closing file\n");
> +        qemu_mutex_lock_migrate_ram();
>          if (qemu_fclose(s->file) != 0) {
>              ret = -1;
>          }
> +        qemu_mutex_unlock_migrate_ram();
>          s->file = NULL;
>      }

Again.