In both sides.  We still don't transmit anything through them.

Signed-off-by: Juan Quintela <quintela@redhat.com>
---
 migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
 1 file changed, 42 insertions(+), 10 deletions(-)

diff --git a/migration/ram.c b/migration/ram.c
index b57d9fd667..7ef0c2b7e2 100644
--- a/migration/ram.c
+++ b/migration/ram.c
@@ -399,6 +399,7 @@ struct MultiFDSendParams {
     uint8_t id;
     char *name;
     QemuThread thread;
+    QIOChannel *c;
     QemuSemaphore sem;
     QemuMutex mutex;
     bool running;
@@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
             qemu_thread_join(&p->thread);
             p->running = false;
         }
+        socket_send_channel_destroy(p->c);
+        p->c = NULL;
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
@@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
     return NULL;
 }
 
+static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
+{
+    MultiFDSendParams *p = opaque;
+    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
+    Error *local_err = NULL;
+
+    if (qio_task_propagate_error(task, &local_err)) {
+        if (multifd_save_cleanup(&local_err) != 0) {
+            migrate_set_error(migrate_get_current(), local_err);
+        }
+    } else {
+        p->c = QIO_CHANNEL(sioc);
+        qio_channel_set_delay(p->c, false);
+        p->running = true;
+        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
+                           QEMU_THREAD_JOINABLE);
+
+        atomic_inc(&multifd_send_state->count);
+    }
+}
+
 int multifd_save_setup(void)
 {
     int thread_count;
@@ -536,11 +560,7 @@ int multifd_save_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdsend_%d", i);
-        p->running = true;
-        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
-                           QEMU_THREAD_JOINABLE);
-
-        atomic_inc(&multifd_send_state->count);
+        socket_send_channel_create(multifd_new_send_channel_async, p);
     }
 
     return 0;
@@ -550,6 +570,7 @@ struct MultiFDRecvParams {
     uint8_t id;
     char *name;
     QemuThread thread;
+    QIOChannel *c;
     QemuSemaphore sem;
     QemuMutex mutex;
     bool running;
@@ -606,6 +627,8 @@ int multifd_load_cleanup(Error **errp)
             qemu_thread_join(&p->thread);
             p->running = false;
         }
+        socket_recv_channel_unref(p->c);
+        p->c = NULL;
         qemu_mutex_destroy(&p->mutex);
         qemu_sem_destroy(&p->sem);
         g_free(p->name);
@@ -688,10 +711,6 @@ int multifd_load_setup(void)
         p->quit = false;
         p->id = i;
         p->name = g_strdup_printf("multifdrecv_%d", i);
-        p->running = true;
-        qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
-                           QEMU_THREAD_JOINABLE);
-        atomic_inc(&multifd_recv_state->count);
     }
 
     return 0;
@@ -710,7 +729,20 @@ bool multifd_recv_all_channels_created(void)
 
 void multifd_recv_new_channel(QIOChannel *ioc)
 {
-    /* nothing to do yet */
+    MultiFDRecvParams *p;
+    /* we need to invent channels id's until we transmit */
+    /* we will remove this on a later patch */
+    static int i = 0;
+
+    p = &multifd_recv_state->params[i];
+    i++;
+    p->c = ioc;
+    socket_recv_channel_ref(ioc);
+
+    p->running = true;
+    qemu_thread_create(&p->thread, p->name, multifd_recv_thread, p,
+                       QEMU_THREAD_JOINABLE);
+    atomic_inc(&multifd_recv_state->count);
 }
 
 /**
-- 
2.14.3

On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
> In both sides.  We still don't transmit anything through them.

s/In/On/?

> 
> Signed-off-by: Juan Quintela <quintela@redhat.com>
> ---
>  migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
>  1 file changed, 42 insertions(+), 10 deletions(-)
> 
> diff --git a/migration/ram.c b/migration/ram.c
> index b57d9fd667..7ef0c2b7e2 100644
> --- a/migration/ram.c
> +++ b/migration/ram.c
> @@ -399,6 +399,7 @@ struct MultiFDSendParams {
>      uint8_t id;
>      char *name;
>      QemuThread thread;
> +    QIOChannel *c;
>      QemuSemaphore sem;
>      QemuMutex mutex;
>      bool running;
> @@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
>              qemu_thread_join(&p->thread);
>              p->running = false;
>          }
> +        socket_send_channel_destroy(p->c);
> +        p->c = NULL;
>          qemu_mutex_destroy(&p->mutex);
>          qemu_sem_destroy(&p->sem);
>          g_free(p->name);
> @@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
>      return NULL;
>  }
>  
> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
> +{
> +    MultiFDSendParams *p = opaque;
> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
> +    Error *local_err = NULL;
> +
> +    if (qio_task_propagate_error(task, &local_err)) {
> +        if (multifd_save_cleanup(&local_err) != 0) {

Do we need to call multifd_save_cleanup() explicitly here?

Asked since I saw that it would also be called in migrate_fd_cleanup(),
and it seems that we should call migrate_fd_cleanup() soon too when
this happens?

Otherwise it looks fine to me.  Thanks,

> +            migrate_set_error(migrate_get_current(), local_err);
> +        }
> +    } else {
> +        p->c = QIO_CHANNEL(sioc);
> +        qio_channel_set_delay(p->c, false);
> +        p->running = true;
> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
> +                           QEMU_THREAD_JOINABLE);
> +
> +        atomic_inc(&multifd_send_state->count);
> +    }
> +}

[...]


-- 
Peter Xu

Peter Xu <peterx@redhat.com> wrote:
> On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
>> In both sides.  We still don't transmit anything through them.
>
> s/In/On/?
>
>> 
>> Signed-off-by: Juan Quintela <quintela@redhat.com>
>> ---
>>  migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
>>  1 file changed, 42 insertions(+), 10 deletions(-)
>> 
>> diff --git a/migration/ram.c b/migration/ram.c
>> index b57d9fd667..7ef0c2b7e2 100644
>> --- a/migration/ram.c
>> +++ b/migration/ram.c
>> @@ -399,6 +399,7 @@ struct MultiFDSendParams {
>>      uint8_t id;
>>      char *name;
>>      QemuThread thread;
>> +    QIOChannel *c;
>>      QemuSemaphore sem;
>>      QemuMutex mutex;
>>      bool running;
>> @@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
>>              qemu_thread_join(&p->thread);
>>              p->running = false;
>>          }
>> +        socket_send_channel_destroy(p->c);
>> +        p->c = NULL;
>>          qemu_mutex_destroy(&p->mutex);
>>          qemu_sem_destroy(&p->sem);
>>          g_free(p->name);
>> @@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
>>      return NULL;
>>  }
>>  
>> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
>> +{
>> +    MultiFDSendParams *p = opaque;
>> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
>> +    Error *local_err = NULL;
>> +
>> +    if (qio_task_propagate_error(task, &local_err)) {
>> +        if (multifd_save_cleanup(&local_err) != 0) {
>
> Do we need to call multifd_save_cleanup() explicitly here?

Is the easiest way of stoping all multifd threads, no?

> Asked since I saw that it would also be called in migrate_fd_cleanup(),
> and it seems that we should call migrate_fd_cleanup() soon too when
> this happens?

We need to stop migraiton. thtat migrate_set_error() is only used for
reporting in info migrate, it is not acted upon.

Yes, perhaps it should, but as it is, it is not.  So, I think it is
right O:-)

Later, Juan.

> Otherwise it looks fine to me.  Thanks,
>
>> +            migrate_set_error(migrate_get_current(), local_err);
>> +        }
>> +    } else {
>> +        p->c = QIO_CHANNEL(sioc);
>> +        qio_channel_set_delay(p->c, false);
>> +        p->running = true;
>> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
>> +                           QEMU_THREAD_JOINABLE);
>> +
>> +        atomic_inc(&multifd_send_state->count);
>> +    }
>> +}
>
> [...]

On Thu, Mar 15, 2018 at 01:57:54PM +0100, Juan Quintela wrote:
> Peter Xu <peterx@redhat.com> wrote:
> > On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
> >> In both sides.  We still don't transmit anything through them.
> >
> > s/In/On/?
> >
> >> 
> >> Signed-off-by: Juan Quintela <quintela@redhat.com>
> >> ---
> >>  migration/ram.c | 52 ++++++++++++++++++++++++++++++++++++++++++----------
> >>  1 file changed, 42 insertions(+), 10 deletions(-)
> >> 
> >> diff --git a/migration/ram.c b/migration/ram.c
> >> index b57d9fd667..7ef0c2b7e2 100644
> >> --- a/migration/ram.c
> >> +++ b/migration/ram.c
> >> @@ -399,6 +399,7 @@ struct MultiFDSendParams {
> >>      uint8_t id;
> >>      char *name;
> >>      QemuThread thread;
> >> +    QIOChannel *c;
> >>      QemuSemaphore sem;
> >>      QemuMutex mutex;
> >>      bool running;
> >> @@ -455,6 +456,8 @@ int multifd_save_cleanup(Error **errp)
> >>              qemu_thread_join(&p->thread);
> >>              p->running = false;
> >>          }
> >> +        socket_send_channel_destroy(p->c);
> >> +        p->c = NULL;
> >>          qemu_mutex_destroy(&p->mutex);
> >>          qemu_sem_destroy(&p->sem);
> >>          g_free(p->name);
> >> @@ -514,6 +517,27 @@ static void *multifd_send_thread(void *opaque)
> >>      return NULL;
> >>  }
> >>  
> >> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
> >> +{
> >> +    MultiFDSendParams *p = opaque;
> >> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
> >> +    Error *local_err = NULL;
> >> +
> >> +    if (qio_task_propagate_error(task, &local_err)) {
> >> +        if (multifd_save_cleanup(&local_err) != 0) {
> >
> > Do we need to call multifd_save_cleanup() explicitly here?
> 
> Is the easiest way of stoping all multifd threads, no?

Yeah, but again, I thought it would be called later too, since...

> 
> > Asked since I saw that it would also be called in migrate_fd_cleanup(),
> > and it seems that we should call migrate_fd_cleanup() soon too when
> > this happens?
> 
> We need to stop migraiton. thtat migrate_set_error() is only used for
> reporting in info migrate, it is not acted upon.
> 
> Yes, perhaps it should, but as it is, it is not.  So, I think it is
> right O:-)

... after Dave's 688a3dcba9 ("migration: Route errors down through
migration_channel_connect", 2018-02-06), all these channel errors
should finally be routed to migrate_fd_connect(), and in that we have:

void migrate_fd_connect(MigrationState *s, Error *error_in)
{
    s->expected_downtime = s->parameters.downtime_limit;
    s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
    if (error_in) {
        migrate_fd_error(s, error_in);
        migrate_fd_cleanup(s);
        return;
    }
    ...
}

Then, in migrate_fd_cleanup() we have multifd_save_cleanup().  That's
why I thought we can skip the cleanup here since after all we'll do it
in other places (and we can keep the cleanup code unified).

Thanks,

> 
> Later, Juan.
> 
> > Otherwise it looks fine to me.  Thanks,
> >
> >> +            migrate_set_error(migrate_get_current(), local_err);
> >> +        }
> >> +    } else {
> >> +        p->c = QIO_CHANNEL(sioc);
> >> +        qio_channel_set_delay(p->c, false);
> >> +        p->running = true;
> >> +        qemu_thread_create(&p->thread, p->name, multifd_send_thread, p,
> >> +                           QEMU_THREAD_JOINABLE);
> >> +
> >> +        atomic_inc(&multifd_send_state->count);
> >> +    }
> >> +}
> >
> > [...]

-- 
Peter Xu

Peter Xu <peterx@redhat.com> wrote:
> On Thu, Mar 15, 2018 at 01:57:54PM +0100, Juan Quintela wrote:
>> Peter Xu <peterx@redhat.com> wrote:
>> > On Wed, Mar 07, 2018 at 12:00:05PM +0100, Juan Quintela wrote:
>> >> In both sides.  We still don't transmit anything through them.
>> >
>> > s/In/On/?

>> >> +static void multifd_new_send_channel_async(QIOTask *task, gpointer opaque)
>> >> +{
>> >> +    MultiFDSendParams *p = opaque;
>> >> +    QIOChannel *sioc = QIO_CHANNEL(qio_task_get_source(task));
>> >> +    Error *local_err = NULL;
>> >> +
>> >> +    if (qio_task_propagate_error(task, &local_err)) {
>> >> +        if (multifd_save_cleanup(&local_err) != 0) {
>> >
>> > Do we need to call multifd_save_cleanup() explicitly here?
>> 
>> Is the easiest way of stoping all multifd threads, no?
>
> Yeah, but again, I thought it would be called later too, since...

But we are not stopping the threads.  Only if you preffer to call there
terminate_multifd_send_threads().  Probably it is better to call
terminate_multifd_send_threads(), it is what we do on the rest of
errors.

Changing it.

>> > Asked since I saw that it would also be called in migrate_fd_cleanup(),
>> > and it seems that we should call migrate_fd_cleanup() soon too when
>> > this happens?
>> 
>> We need to stop migraiton. thtat migrate_set_error() is only used for
>> reporting in info migrate, it is not acted upon.
>> 
>> Yes, perhaps it should, but as it is, it is not.  So, I think it is
>> right O:-)
>
> ... after Dave's 688a3dcba9 ("migration: Route errors down through
> migration_channel_connect", 2018-02-06), all these channel errors
> should finally be routed to migrate_fd_connect(), and in that we have:
>
> void migrate_fd_connect(MigrationState *s, Error *error_in)
> {
>     s->expected_downtime = s->parameters.downtime_limit;
>     s->cleanup_bh = qemu_bh_new(migrate_fd_cleanup, s);
>     if (error_in) {
>         migrate_fd_error(s, error_in);
>         migrate_fd_cleanup(s);
>         return;
>     }
>     ...
> }
>
> Then, in migrate_fd_cleanup() we have multifd_save_cleanup().  That's
> why I thought we can skip the cleanup here since after all we'll do it
> in other places (and we can keep the cleanup code unified).

Ok, it is better, changing to terminate_multifd_send_threads.

Thanks, Juan.