Re: [Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-02-15 Thread Daniel P. Berrange 
On Wed, Feb 15, 2017 at 02:46:15PM +, Dr. David Alan Gilbert wrote:
> * Daniel P. Berrange (berra...@redhat.com) wrote:
> > On Mon, Jan 23, 2017 at 10:32:13PM +0100, Juan Quintela wrote:
> > > We create new channels for each new thread created. We only send through
> > > them a character to be sure that we are creating the channels in the
> > > right order.
> > > 
> > > Note: Reference count/freeing of channels is not done
> > > 
> > > Signed-off-by: Juan Quintela 
> > > ---
> > >  include/migration/migration.h |  6 +
> > >  migration/ram.c   | 45 +-
> > >  migration/socket.c| 56 
> > > +--
> > 
> > BTW, right now libvirt never uses QEMU's tcp: protocol - it does everything
> > with the fd: protocol.  So either we need multi-fd support for fd: protocol,
> > or libvirt needs to switch to use tcp:
> 
> I thought using fd was safer than tcp: because of the race when something else
> could listen on the proposed port on the incoming side between the point of 
> libvirt
> picking the port number and qemu starting.

Hmm, good point.

> > In fact, having said that, we're going to have to switch to use  the tcp:
> > protocol anyway in order to support TLS, so this is just another good
> > reason for the switch.
> 
> I thought you had a way of allowing fd to work for TLS?

Oh yes, I forgot that I made that work :-)

Regards,
Daniel
-- 
|: http://berrange.com  -o-http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org  -o- http://virt-manager.org :|
|: http://entangle-photo.org   -o-http://search.cpan.org/~danberr/ :|

Re: [Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-02-15 Thread Dr. David Alan Gilbert 
* Daniel P. Berrange (berra...@redhat.com) wrote:
> On Mon, Jan 23, 2017 at 10:32:13PM +0100, Juan Quintela wrote:
> > We create new channels for each new thread created. We only send through
> > them a character to be sure that we are creating the channels in the
> > right order.
> > 
> > Note: Reference count/freeing of channels is not done
> > 
> > Signed-off-by: Juan Quintela 
> > ---
> >  include/migration/migration.h |  6 +
> >  migration/ram.c   | 45 +-
> >  migration/socket.c| 56 
> > +--
> 
> BTW, right now libvirt never uses QEMU's tcp: protocol - it does everything
> with the fd: protocol.  So either we need multi-fd support for fd: protocol,
> or libvirt needs to switch to use tcp:

I thought using fd was safer than tcp: because of the race when something else
could listen on the proposed port on the incoming side between the point of 
libvirt
picking the port number and qemu starting.

> In fact, having said that, we're going to have to switch to use  the tcp:
> protocol anyway in order to support TLS, so this is just another good
> reason for the switch.

I thought you had a way of allowing fd to work for TLS?

Dave

> 
> We avoided tcp: in the past because QEMU was incapable of reporting error
> messages when the connection failed. That's fixed since
> 
>   commit d59ce6f34434bf47a9b26138c908650bf9a24be1
>   Author: Daniel P. Berrange 
>   Date:   Wed Apr 27 11:05:00 2016 +0100
> 
> migration: add reporting of errors for outgoing migration
> 
> so libvirt should be ok to use tcp: now.
> 
> >  3 files changed, 104 insertions(+), 3 deletions(-)
> > 
> > diff --git a/include/migration/migration.h b/include/migration/migration.h
> > index f119ba0..3989bd6 100644
> > --- a/include/migration/migration.h
> > +++ b/include/migration/migration.h
> > @@ -22,6 +22,7 @@
> >  #include "qapi-types.h"
> >  #include "exec/cpu-common.h"
> >  #include "qemu/coroutine_int.h"
> > +#include "io/channel.h"
> > 
> >  #define QEMU_VM_FILE_MAGIC   0x5145564d
> >  #define QEMU_VM_FILE_VERSION_COMPAT  0x0002
> > @@ -218,6 +219,11 @@ void tcp_start_incoming_migration(const char 
> > *host_port, Error **errp);
> > 
> >  void tcp_start_outgoing_migration(MigrationState *s, const char 
> > *host_port, Error **errp);
> > 
> > +QIOChannel *socket_recv_channel_create(void);
> > +int socket_recv_channel_destroy(QIOChannel *recv);
> > +QIOChannel *socket_send_channel_create(void);
> > +int socket_send_channel_destroy(QIOChannel *send);
> > +
> >  void unix_start_incoming_migration(const char *path, Error **errp);
> > 
> >  void unix_start_outgoing_migration(MigrationState *s, const char *path, 
> > Error **errp);
> > diff --git a/migration/ram.c b/migration/ram.c
> > index 939f364..5ad7cb3 100644
> > --- a/migration/ram.c
> > +++ b/migration/ram.c
> > @@ -386,9 +386,11 @@ void migrate_compress_threads_create(void)
> > 
> >  struct MultiFDSendParams {
> >  QemuThread thread;
> > +QIOChannel *c;
> >  QemuCond cond;
> >  QemuMutex mutex;
> >  bool quit;
> > +bool started;
> >  };
> >  typedef struct MultiFDSendParams MultiFDSendParams;
> > 
> > @@ -397,6 +399,13 @@ static MultiFDSendParams *multifd_send;
> >  static void *multifd_send_thread(void *opaque)
> >  {
> >  MultiFDSendParams *params = opaque;
> > +char start = 's';
> > +
> > +qio_channel_write(params->c, , 1, _abort);
> > +qemu_mutex_lock(>mutex);
> > +params->started = true;
> > +qemu_cond_signal(>cond);
> > +qemu_mutex_unlock(>mutex);
> > 
> >  qemu_mutex_lock(>mutex);
> >  while (!params->quit){
> > @@ -433,6 +442,7 @@ void migrate_multifd_send_threads_join(void)
> >  qemu_thread_join(_send[i].thread);
> >  qemu_mutex_destroy(_send[i].mutex);
> >  qemu_cond_destroy(_send[i].cond);
> > +socket_send_channel_destroy(multifd_send[i].c);
> >  }
> >  g_free(multifd_send);
> >  multifd_send = NULL;
> > @@ -452,18 +462,31 @@ void migrate_multifd_send_threads_create(void)
> >  qemu_mutex_init(_send[i].mutex);
> >  qemu_cond_init(_send[i].cond);
> >  multifd_send[i].quit = false;
> > +multifd_send[i].started = false;
> > +multifd_send[i].c = socket_send_channel_create();
> > +if(!multifd_send[i].c) {
> > +error_report("Error creating a send channel");
> > +exit(0);
> > +}
> >  snprintf(thread_name, 15, "multifd_send_%d", i);
> >  qemu_thread_create(_send[i].thread, thread_name,
> > multifd_send_thread, _send[i],
> > QEMU_THREAD_JOINABLE);
> > +qemu_mutex_lock(_send[i].mutex);
> > +while (!multifd_send[i].started) {
> > +qemu_cond_wait(_send[i].cond, _send[i].mutex);
> > +}
> > +qemu_mutex_unlock(_send[i].mutex);
> >  }
> >

Re: [Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-02-13 Thread Daniel P. Berrange 
On Mon, Jan 23, 2017 at 10:32:13PM +0100, Juan Quintela wrote:
> We create new channels for each new thread created. We only send through
> them a character to be sure that we are creating the channels in the
> right order.
> 
> Note: Reference count/freeing of channels is not done
> 
> Signed-off-by: Juan Quintela 
> ---
>  include/migration/migration.h |  6 +
>  migration/ram.c   | 45 +-
>  migration/socket.c| 56 
> +--

BTW, right now libvirt never uses QEMU's tcp: protocol - it does everything
with the fd: protocol.  So either we need multi-fd support for fd: protocol,
or libvirt needs to switch to use tcp:

In fact, having said that, we're going to have to switch to use  the tcp:
protocol anyway in order to support TLS, so this is just another good
reason for the switch.

We avoided tcp: in the past because QEMU was incapable of reporting error
messages when the connection failed. That's fixed since

  commit d59ce6f34434bf47a9b26138c908650bf9a24be1
  Author: Daniel P. Berrange 
  Date:   Wed Apr 27 11:05:00 2016 +0100

migration: add reporting of errors for outgoing migration

so libvirt should be ok to use tcp: now.

>  3 files changed, 104 insertions(+), 3 deletions(-)
> 
> diff --git a/include/migration/migration.h b/include/migration/migration.h
> index f119ba0..3989bd6 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -22,6 +22,7 @@
>  #include "qapi-types.h"
>  #include "exec/cpu-common.h"
>  #include "qemu/coroutine_int.h"
> +#include "io/channel.h"
> 
>  #define QEMU_VM_FILE_MAGIC   0x5145564d
>  #define QEMU_VM_FILE_VERSION_COMPAT  0x0002
> @@ -218,6 +219,11 @@ void tcp_start_incoming_migration(const char *host_port, 
> Error **errp);
> 
>  void tcp_start_outgoing_migration(MigrationState *s, const char *host_port, 
> Error **errp);
> 
> +QIOChannel *socket_recv_channel_create(void);
> +int socket_recv_channel_destroy(QIOChannel *recv);
> +QIOChannel *socket_send_channel_create(void);
> +int socket_send_channel_destroy(QIOChannel *send);
> +
>  void unix_start_incoming_migration(const char *path, Error **errp);
> 
>  void unix_start_outgoing_migration(MigrationState *s, const char *path, 
> Error **errp);
> diff --git a/migration/ram.c b/migration/ram.c
> index 939f364..5ad7cb3 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -386,9 +386,11 @@ void migrate_compress_threads_create(void)
> 
>  struct MultiFDSendParams {
>  QemuThread thread;
> +QIOChannel *c;
>  QemuCond cond;
>  QemuMutex mutex;
>  bool quit;
> +bool started;
>  };
>  typedef struct MultiFDSendParams MultiFDSendParams;
> 
> @@ -397,6 +399,13 @@ static MultiFDSendParams *multifd_send;
>  static void *multifd_send_thread(void *opaque)
>  {
>  MultiFDSendParams *params = opaque;
> +char start = 's';
> +
> +qio_channel_write(params->c, , 1, _abort);
> +qemu_mutex_lock(>mutex);
> +params->started = true;
> +qemu_cond_signal(>cond);
> +qemu_mutex_unlock(>mutex);
> 
>  qemu_mutex_lock(>mutex);
>  while (!params->quit){
> @@ -433,6 +442,7 @@ void migrate_multifd_send_threads_join(void)
>  qemu_thread_join(_send[i].thread);
>  qemu_mutex_destroy(_send[i].mutex);
>  qemu_cond_destroy(_send[i].cond);
> +socket_send_channel_destroy(multifd_send[i].c);
>  }
>  g_free(multifd_send);
>  multifd_send = NULL;
> @@ -452,18 +462,31 @@ void migrate_multifd_send_threads_create(void)
>  qemu_mutex_init(_send[i].mutex);
>  qemu_cond_init(_send[i].cond);
>  multifd_send[i].quit = false;
> +multifd_send[i].started = false;
> +multifd_send[i].c = socket_send_channel_create();
> +if(!multifd_send[i].c) {
> +error_report("Error creating a send channel");
> +exit(0);
> +}
>  snprintf(thread_name, 15, "multifd_send_%d", i);
>  qemu_thread_create(_send[i].thread, thread_name,
> multifd_send_thread, _send[i],
> QEMU_THREAD_JOINABLE);
> +qemu_mutex_lock(_send[i].mutex);
> +while (!multifd_send[i].started) {
> +qemu_cond_wait(_send[i].cond, _send[i].mutex);
> +}
> +qemu_mutex_unlock(_send[i].mutex);
>  }
>  }
> 
>  struct MultiFDRecvParams {
>  QemuThread thread;
> +QIOChannel *c;
>  QemuCond cond;
>  QemuMutex mutex;
>  bool quit;
> +bool started;
>  };
>  typedef struct MultiFDRecvParams MultiFDRecvParams;
> 
> @@ -472,7 +495,14 @@ static MultiFDRecvParams *multifd_recv;
>  static void *multifd_recv_thread(void *opaque)
>  {
>  MultiFDRecvParams *params = opaque;
> - 
> +char start;
> +
> +qio_channel_read(params->c, , 1, _abort);
> +qemu_mutex_lock(>mutex);
> +params->started = true;
> +

Re: [Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-02-13 Thread Dr. David Alan Gilbert 
* Juan Quintela (quint...@redhat.com) wrote:
> "Dr. David Alan Gilbert"  wrote:
> > * Juan Quintela (quint...@redhat.com) wrote:
> >> We create new channels for each new thread created. We only send through
> >> them a character to be sure that we are creating the channels in the
> >> right order.
> >> 
> >> Note: Reference count/freeing of channels is not done
> >> 
> >> Signed-off-by: Juan Quintela 
> >> ---
> >>  include/migration/migration.h |  6 +
> >>  migration/ram.c   | 45 +-
> >>  migration/socket.c| 56 
> >> +--
> >>  3 files changed, 104 insertions(+), 3 deletions(-)
> >
> > One thing not direclt in here, you should probably look at the migration 
> > cancel
> > code to get it to call shutdown() on all your extra sockets, it stops things
> > blocking in any one of them.
> 
> Will do.
> 
> >> 
> >> diff --git a/include/migration/migration.h b/include/migration/migration.h
> >> index f119ba0..3989bd6 100644
> >> --- a/include/migration/migration.h
> >> +++ b/include/migration/migration.h
> >> @@ -22,6 +22,7 @@
> >>  #include "qapi-types.h"
> >>  #include "exec/cpu-common.h"
> >>  #include "qemu/coroutine_int.h"
> >> +#include "io/channel.h"
> >> 
> >>  #define QEMU_VM_FILE_MAGIC   0x5145564d
> >>  #define QEMU_VM_FILE_VERSION_COMPAT  0x0002
> >> @@ -218,6 +219,11 @@ void tcp_start_incoming_migration(const char 
> >> *host_port, Error **errp);
> >> 
> >>  void tcp_start_outgoing_migration(MigrationState *s, const char 
> >> *host_port, Error **errp);
> >> 
> >> +QIOChannel *socket_recv_channel_create(void);
> >> +int socket_recv_channel_destroy(QIOChannel *recv);
> >> +QIOChannel *socket_send_channel_create(void);
> >> +int socket_send_channel_destroy(QIOChannel *send);
> >> +
> >>  void unix_start_incoming_migration(const char *path, Error **errp);
> >> 
> >>  void unix_start_outgoing_migration(MigrationState *s, const char *path, 
> >> Error **errp);
> >> diff --git a/migration/ram.c b/migration/ram.c
> >> index 939f364..5ad7cb3 100644
> >> --- a/migration/ram.c
> >> +++ b/migration/ram.c
> >> @@ -386,9 +386,11 @@ void migrate_compress_threads_create(void)
> >> 
> >>  struct MultiFDSendParams {
> >>  QemuThread thread;
> >> +QIOChannel *c;
> >>  QemuCond cond;
> >>  QemuMutex mutex;
> >>  bool quit;
> >> +bool started;
> >>  };
> >>  typedef struct MultiFDSendParams MultiFDSendParams;
> >> 
> >> @@ -397,6 +399,13 @@ static MultiFDSendParams *multifd_send;
> >>  static void *multifd_send_thread(void *opaque)
> >>  {
> >>  MultiFDSendParams *params = opaque;
> >> +char start = 's';
> >> +
> >> +qio_channel_write(params->c, , 1, _abort);
> >
> > I'd be tempted to send something stronger as a guarantee
> > that you're connecting the right thing to the right place;
> > maybe something like a QEMU + UUID + fd index?
> > I guarantee someone is going to mess up the fd's in the wrong
> > order or connect some random other process to one of them.
> 
> which UUID? I can put anything there.

I was thinking just something to stop two migrations getting mixed
up; but I see there is a qemu_uuid variable defined, might be as good
as anything.

> >> +qemu_mutex_lock(>mutex);
> >> +params->started = true;
> >> +qemu_cond_signal(>cond);
> >> +qemu_mutex_unlock(>mutex);
> >> 
> >>  qemu_mutex_lock(>mutex);
> >
> > That unlock/lock pair is odd.
> 
> Fixed.
> 
> >
> >>  while (!params->quit){
> >> @@ -433,6 +442,7 @@ void migrate_multifd_send_threads_join(void)
> >>  qemu_thread_join(_send[i].thread);
> >>  qemu_mutex_destroy(_send[i].mutex);
> >>  qemu_cond_destroy(_send[i].cond);
> >> +socket_send_channel_destroy(multifd_send[i].c);
> >>  }
> >>  g_free(multifd_send);
> >>  multifd_send = NULL;
> >> @@ -452,18 +462,31 @@ void migrate_multifd_send_threads_create(void)
> >>  qemu_mutex_init(_send[i].mutex);
> >>  qemu_cond_init(_send[i].cond);
> >>  multifd_send[i].quit = false;
> >> +multifd_send[i].started = false;
> >> +multifd_send[i].c = socket_send_channel_create();
> >> +if(!multifd_send[i].c) {
> >> +error_report("Error creating a send channel");
> >> +exit(0);
> >
> > Hmm no exit!
> 
> I have to add the error path before to the callers :-(
> 
> 
> >
> > We need to be careful there since that's a sync; it depends what
> > calls this, if I'm reading the code above correctly then it gets called
> > from the main thread and that would be bad if it blocked; it's ok if it
> > was the fd threads or the migration thread though.
> 
> I think it is from the migration thread, no?
> /me checks

Probably worth a comment saying which thread it comes from :-)

> I stand corrected.  It is called from the main thread.  It works if
> destination is not up.  It segfaults if destination is launched but

Re: [Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-02-13 Thread Juan Quintela 
"Dr. David Alan Gilbert"  wrote:
> * Juan Quintela (quint...@redhat.com) wrote:
>> We create new channels for each new thread created. We only send through
>> them a character to be sure that we are creating the channels in the
>> right order.
>> 
>> Note: Reference count/freeing of channels is not done
>> 
>> Signed-off-by: Juan Quintela 
>> ---
>>  include/migration/migration.h |  6 +
>>  migration/ram.c   | 45 +-
>>  migration/socket.c| 56 
>> +--
>>  3 files changed, 104 insertions(+), 3 deletions(-)
>
> One thing not direclt in here, you should probably look at the migration 
> cancel
> code to get it to call shutdown() on all your extra sockets, it stops things
> blocking in any one of them.

Will do.

>> 
>> diff --git a/include/migration/migration.h b/include/migration/migration.h
>> index f119ba0..3989bd6 100644
>> --- a/include/migration/migration.h
>> +++ b/include/migration/migration.h
>> @@ -22,6 +22,7 @@
>>  #include "qapi-types.h"
>>  #include "exec/cpu-common.h"
>>  #include "qemu/coroutine_int.h"
>> +#include "io/channel.h"
>> 
>>  #define QEMU_VM_FILE_MAGIC   0x5145564d
>>  #define QEMU_VM_FILE_VERSION_COMPAT  0x0002
>> @@ -218,6 +219,11 @@ void tcp_start_incoming_migration(const char 
>> *host_port, Error **errp);
>> 
>>  void tcp_start_outgoing_migration(MigrationState *s, const char *host_port, 
>> Error **errp);
>> 
>> +QIOChannel *socket_recv_channel_create(void);
>> +int socket_recv_channel_destroy(QIOChannel *recv);
>> +QIOChannel *socket_send_channel_create(void);
>> +int socket_send_channel_destroy(QIOChannel *send);
>> +
>>  void unix_start_incoming_migration(const char *path, Error **errp);
>> 
>>  void unix_start_outgoing_migration(MigrationState *s, const char *path, 
>> Error **errp);
>> diff --git a/migration/ram.c b/migration/ram.c
>> index 939f364..5ad7cb3 100644
>> --- a/migration/ram.c
>> +++ b/migration/ram.c
>> @@ -386,9 +386,11 @@ void migrate_compress_threads_create(void)
>> 
>>  struct MultiFDSendParams {
>>  QemuThread thread;
>> +QIOChannel *c;
>>  QemuCond cond;
>>  QemuMutex mutex;
>>  bool quit;
>> +bool started;
>>  };
>>  typedef struct MultiFDSendParams MultiFDSendParams;
>> 
>> @@ -397,6 +399,13 @@ static MultiFDSendParams *multifd_send;
>>  static void *multifd_send_thread(void *opaque)
>>  {
>>  MultiFDSendParams *params = opaque;
>> +char start = 's';
>> +
>> +qio_channel_write(params->c, , 1, _abort);
>
> I'd be tempted to send something stronger as a guarantee
> that you're connecting the right thing to the right place;
> maybe something like a QEMU + UUID + fd index?
> I guarantee someone is going to mess up the fd's in the wrong
> order or connect some random other process to one of them.

which UUID? I can put anything there.

>> +qemu_mutex_lock(>mutex);
>> +params->started = true;
>> +qemu_cond_signal(>cond);
>> +qemu_mutex_unlock(>mutex);
>> 
>>  qemu_mutex_lock(>mutex);
>
> That unlock/lock pair is odd.

Fixed.

>
>>  while (!params->quit){
>> @@ -433,6 +442,7 @@ void migrate_multifd_send_threads_join(void)
>>  qemu_thread_join(_send[i].thread);
>>  qemu_mutex_destroy(_send[i].mutex);
>>  qemu_cond_destroy(_send[i].cond);
>> +socket_send_channel_destroy(multifd_send[i].c);
>>  }
>>  g_free(multifd_send);
>>  multifd_send = NULL;
>> @@ -452,18 +462,31 @@ void migrate_multifd_send_threads_create(void)
>>  qemu_mutex_init(_send[i].mutex);
>>  qemu_cond_init(_send[i].cond);
>>  multifd_send[i].quit = false;
>> +multifd_send[i].started = false;
>> +multifd_send[i].c = socket_send_channel_create();
>> +if(!multifd_send[i].c) {
>> +error_report("Error creating a send channel");
>> +exit(0);
>
> Hmm no exit!

I have to add the error path before to the callers :-(


>
> We need to be careful there since that's a sync; it depends what
> calls this, if I'm reading the code above correctly then it gets called
> from the main thread and that would be bad if it blocked; it's ok if it
> was the fd threads or the migration thread though.

I think it is from the migration thread, no?
/me checks

I stand corrected.  It is called from the main thread.  It works if
destination is not up.  It segfaults if destination is launched but not
conffigured for multithread.

Will post fix later.

Later, Juan.

Re: [Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-01-27 Thread Dr. David Alan Gilbert 
* Juan Quintela (quint...@redhat.com) wrote:
> We create new channels for each new thread created. We only send through
> them a character to be sure that we are creating the channels in the
> right order.
> 
> Note: Reference count/freeing of channels is not done
> 
> Signed-off-by: Juan Quintela 
> ---
>  include/migration/migration.h |  6 +
>  migration/ram.c   | 45 +-
>  migration/socket.c| 56 
> +--
>  3 files changed, 104 insertions(+), 3 deletions(-)

One thing not direclt in here, you should probably look at the migration cancel
code to get it to call shutdown() on all your extra sockets, it stops things
blocking in any one of them.

> 
> diff --git a/include/migration/migration.h b/include/migration/migration.h
> index f119ba0..3989bd6 100644
> --- a/include/migration/migration.h
> +++ b/include/migration/migration.h
> @@ -22,6 +22,7 @@
>  #include "qapi-types.h"
>  #include "exec/cpu-common.h"
>  #include "qemu/coroutine_int.h"
> +#include "io/channel.h"
> 
>  #define QEMU_VM_FILE_MAGIC   0x5145564d
>  #define QEMU_VM_FILE_VERSION_COMPAT  0x0002
> @@ -218,6 +219,11 @@ void tcp_start_incoming_migration(const char *host_port, 
> Error **errp);
> 
>  void tcp_start_outgoing_migration(MigrationState *s, const char *host_port, 
> Error **errp);
> 
> +QIOChannel *socket_recv_channel_create(void);
> +int socket_recv_channel_destroy(QIOChannel *recv);
> +QIOChannel *socket_send_channel_create(void);
> +int socket_send_channel_destroy(QIOChannel *send);
> +
>  void unix_start_incoming_migration(const char *path, Error **errp);
> 
>  void unix_start_outgoing_migration(MigrationState *s, const char *path, 
> Error **errp);
> diff --git a/migration/ram.c b/migration/ram.c
> index 939f364..5ad7cb3 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -386,9 +386,11 @@ void migrate_compress_threads_create(void)
> 
>  struct MultiFDSendParams {
>  QemuThread thread;
> +QIOChannel *c;
>  QemuCond cond;
>  QemuMutex mutex;
>  bool quit;
> +bool started;
>  };
>  typedef struct MultiFDSendParams MultiFDSendParams;
> 
> @@ -397,6 +399,13 @@ static MultiFDSendParams *multifd_send;
>  static void *multifd_send_thread(void *opaque)
>  {
>  MultiFDSendParams *params = opaque;
> +char start = 's';
> +
> +qio_channel_write(params->c, , 1, _abort);

I'd be tempted to send something stronger as a guarantee
that you're connecting the right thing to the right place;
maybe something like a QEMU + UUID + fd index?
I guarantee someone is going to mess up the fd's in the wrong
order or connect some random other process to one of them.

> +qemu_mutex_lock(>mutex);
> +params->started = true;
> +qemu_cond_signal(>cond);
> +qemu_mutex_unlock(>mutex);
> 
>  qemu_mutex_lock(>mutex);

That unlock/lock pair is odd.

>  while (!params->quit){
> @@ -433,6 +442,7 @@ void migrate_multifd_send_threads_join(void)
>  qemu_thread_join(_send[i].thread);
>  qemu_mutex_destroy(_send[i].mutex);
>  qemu_cond_destroy(_send[i].cond);
> +socket_send_channel_destroy(multifd_send[i].c);
>  }
>  g_free(multifd_send);
>  multifd_send = NULL;
> @@ -452,18 +462,31 @@ void migrate_multifd_send_threads_create(void)
>  qemu_mutex_init(_send[i].mutex);
>  qemu_cond_init(_send[i].cond);
>  multifd_send[i].quit = false;
> +multifd_send[i].started = false;
> +multifd_send[i].c = socket_send_channel_create();
> +if(!multifd_send[i].c) {
> +error_report("Error creating a send channel");
> +exit(0);

Hmm no exit!

> +}
>  snprintf(thread_name, 15, "multifd_send_%d", i);
>  qemu_thread_create(_send[i].thread, thread_name,
> multifd_send_thread, _send[i],
> QEMU_THREAD_JOINABLE);
> +qemu_mutex_lock(_send[i].mutex);
> +while (!multifd_send[i].started) {
> +qemu_cond_wait(_send[i].cond, _send[i].mutex);
> +}
> +qemu_mutex_unlock(_send[i].mutex);
>  }
>  }
> 
>  struct MultiFDRecvParams {
>  QemuThread thread;
> +QIOChannel *c;
>  QemuCond cond;
>  QemuMutex mutex;
>  bool quit;
> +bool started;
>  };
>  typedef struct MultiFDRecvParams MultiFDRecvParams;
> 
> @@ -472,7 +495,14 @@ static MultiFDRecvParams *multifd_recv;
>  static void *multifd_recv_thread(void *opaque)
>  {
>  MultiFDRecvParams *params = opaque;
> - 
> +char start;
> +
> +qio_channel_read(params->c, , 1, _abort);
> +qemu_mutex_lock(>mutex);
> +params->started = true;
> +qemu_cond_signal(>cond);
> +qemu_mutex_unlock(>mutex);
> +
>  qemu_mutex_lock(>mutex);
>  while (!params->quit){
>  qemu_cond_wait(>cond, >mutex);
> @@ -508,6 +538,7 @@ void migrate_multifd_recv_threads_join(void)
>

[Qemu-devel] [PATCH 09/17] migration: Start of multiple fd work

2017-01-23 Thread Juan Quintela 
We create new channels for each new thread created. We only send through
them a character to be sure that we are creating the channels in the
right order.

Note: Reference count/freeing of channels is not done

Signed-off-by: Juan Quintela 
---
 include/migration/migration.h |  6 +
 migration/ram.c   | 45 +-
 migration/socket.c| 56 +--
 3 files changed, 104 insertions(+), 3 deletions(-)

diff --git a/include/migration/migration.h b/include/migration/migration.h
index f119ba0..3989bd6 100644
--- a/include/migration/migration.h
+++ b/include/migration/migration.h
@@ -22,6 +22,7 @@
 #include "qapi-types.h"
 #include "exec/cpu-common.h"
 #include "qemu/coroutine_int.h"
+#include "io/channel.h"

 #define QEMU_VM_FILE_MAGIC   0x5145564d
 #define QEMU_VM_FILE_VERSION_COMPAT  0x0002
@@ -218,6 +219,11 @@ void tcp_start_incoming_migration(const char *host_port, 
Error **errp);

 void tcp_start_outgoing_migration(MigrationState *s, const char *host_port, 
Error **errp);

+QIOChannel *socket_recv_channel_create(void);
+int socket_recv_channel_destroy(QIOChannel *recv);
+QIOChannel *socket_send_channel_create(void);
+int socket_send_channel_destroy(QIOChannel *send);
+
 void unix_start_incoming_migration(const char *path, Error **errp);

 void unix_start_outgoing_migration(MigrationState *s, const char *path, Error 
**errp);
diff --git a/migration/ram.c b/migration/ram.c
index 939f364..5ad7cb3 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -386,9 +386,11 @@ void migrate_compress_threads_create(void)

 struct MultiFDSendParams {
 QemuThread thread;
+QIOChannel *c;
 QemuCond cond;
 QemuMutex mutex;
 bool quit;
+bool started;
 };
 typedef struct MultiFDSendParams MultiFDSendParams;

@@ -397,6 +399,13 @@ static MultiFDSendParams *multifd_send;
 static void *multifd_send_thread(void *opaque)
 {
 MultiFDSendParams *params = opaque;
+char start = 's';
+
+qio_channel_write(params->c, , 1, _abort);
+qemu_mutex_lock(>mutex);
+params->started = true;
+qemu_cond_signal(>cond);
+qemu_mutex_unlock(>mutex);

 qemu_mutex_lock(>mutex);
 while (!params->quit){
@@ -433,6 +442,7 @@ void migrate_multifd_send_threads_join(void)
 qemu_thread_join(_send[i].thread);
 qemu_mutex_destroy(_send[i].mutex);
 qemu_cond_destroy(_send[i].cond);
+socket_send_channel_destroy(multifd_send[i].c);
 }
 g_free(multifd_send);
 multifd_send = NULL;
@@ -452,18 +462,31 @@ void migrate_multifd_send_threads_create(void)
 qemu_mutex_init(_send[i].mutex);
 qemu_cond_init(_send[i].cond);
 multifd_send[i].quit = false;
+multifd_send[i].started = false;
+multifd_send[i].c = socket_send_channel_create();
+if(!multifd_send[i].c) {
+error_report("Error creating a send channel");
+exit(0);
+}
 snprintf(thread_name, 15, "multifd_send_%d", i);
 qemu_thread_create(_send[i].thread, thread_name,
multifd_send_thread, _send[i],
QEMU_THREAD_JOINABLE);
+qemu_mutex_lock(_send[i].mutex);
+while (!multifd_send[i].started) {
+qemu_cond_wait(_send[i].cond, _send[i].mutex);
+}
+qemu_mutex_unlock(_send[i].mutex);
 }
 }

 struct MultiFDRecvParams {
 QemuThread thread;
+QIOChannel *c;
 QemuCond cond;
 QemuMutex mutex;
 bool quit;
+bool started;
 };
 typedef struct MultiFDRecvParams MultiFDRecvParams;

@@ -472,7 +495,14 @@ static MultiFDRecvParams *multifd_recv;
 static void *multifd_recv_thread(void *opaque)
 {
 MultiFDRecvParams *params = opaque;
- 
+char start;
+
+qio_channel_read(params->c, , 1, _abort);
+qemu_mutex_lock(>mutex);
+params->started = true;
+qemu_cond_signal(>cond);
+qemu_mutex_unlock(>mutex);
+
 qemu_mutex_lock(>mutex);
 while (!params->quit){
 qemu_cond_wait(>cond, >mutex);
@@ -508,6 +538,7 @@ void migrate_multifd_recv_threads_join(void)
 qemu_thread_join(_recv[i].thread);
 qemu_mutex_destroy(_recv[i].mutex);
 qemu_cond_destroy(_recv[i].cond);
+socket_send_channel_destroy(multifd_recv[i].c);
 }
 g_free(multifd_recv);
 multifd_recv = NULL;
@@ -526,9 +557,21 @@ void migrate_multifd_recv_threads_create(void)
 qemu_mutex_init(_recv[i].mutex);
 qemu_cond_init(_recv[i].cond);
 multifd_recv[i].quit = false;
+multifd_recv[i].started = false;
+multifd_recv[i].c = socket_recv_channel_create();
+
+if(!multifd_recv[i].c) {
+error_report("Error creating a recv channel");
+exit(0);
+}
 qemu_thread_create(_recv[i].thread, "multifd_recv",
multifd_recv_thread, _recv[i],
QEMU_THREAD_JOINABLE);
+