Re: [Qemu-devel] [PATCH V8 02/17] colo-compare: implement the process of checkpoint

[Zhang Chen]

On Mon, Jun 4, 2018 at 2:31 PM, Jason Wang <address@hidden> wrote:

>
>
> On 2018年06月03日 13:05, Zhang Chen wrote:
>
>> While do checkpoint, we need to flush all the unhandled packets,
>> By using the filter notifier mechanism, we can easily to notify
>> every compare object to do this process, which runs inside
>> of compare threads as a coroutine.
>>
>> Signed-off-by: zhanghailiang <address@hidden>
>> Signed-off-by: Zhang Chen <address@hidden>
>> ---
>>   include/migration/colo.h |  6 ++++
>>   net/colo-compare.c       | 76 ++++++++++++++++++++++++++++++++++++++++
>>   net/colo-compare.h       | 22 ++++++++++++
>>   3 files changed, 104 insertions(+)
>>   create mode 100644 net/colo-compare.h
>>
>> diff --git a/include/migration/colo.h b/include/migration/colo.h
>> index 2fe48ad353..fefb2fcf4c 100644
>> --- a/include/migration/colo.h
>> +++ b/include/migration/colo.h
>> @@ -16,6 +16,12 @@
>>   #include "qemu-common.h"
>>   #include "qapi/qapi-types-migration.h"
>>   +enum colo_event {
>> +    COLO_EVENT_NONE,
>> +    COLO_EVENT_CHECKPOINT,
>> +    COLO_EVENT_FAILOVER,
>> +};
>> +
>>   void colo_info_init(void);
>>     void migrate_start_colo_process(MigrationState *s);
>> diff --git a/net/colo-compare.c b/net/colo-compare.c
>> index 23b2d2c4cc..7ff3ae8904 100644
>> --- a/net/colo-compare.c
>> +++ b/net/colo-compare.c
>> @@ -27,11 +27,16 @@
>>   #include "qemu/sockets.h"
>>   #include "net/colo.h"
>>   #include "sysemu/iothread.h"
>> +#include "net/colo-compare.h"
>> +#include "migration/colo.h"
>>     #define TYPE_COLO_COMPARE "colo-compare"
>>   #define COLO_COMPARE(obj) \
>>       OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
>>   +static QTAILQ_HEAD(, CompareState) net_compares =
>> +       QTAILQ_HEAD_INITIALIZER(net_compares);
>> +
>>   #define COMPARE_READ_LEN_MAX NET_BUFSIZE
>>   #define MAX_QUEUE_SIZE 1024
>>   @@ -41,6 +46,10 @@
>>   /* TODO: Should be configurable */
>>   #define REGULAR_PACKET_CHECK_MS 3000
>>   +static QemuMutex event_mtx;
>> +static QemuCond event_complete_cond;
>> +static int event_unhandled_count;
>> +
>>   /*
>>    *  + CompareState ++
>>    *  |               |
>> @@ -87,6 +96,11 @@ typedef struct CompareState {
>>       IOThread *iothread;
>>       GMainContext *worker_context;
>>       QEMUTimer *packet_check_timer;
>> +
>> +    QEMUBH *event_bh;
>> +    enum colo_event event;
>> +
>> +    QTAILQ_ENTRY(CompareState) next;
>>   } CompareState;
>>     typedef struct CompareClass {
>> @@ -736,6 +750,25 @@ static void check_old_packet_regular(void *opaque)
>>                   REGULAR_PACKET_CHECK_MS);
>>   }
>>   +/* Public API, Used for COLO frame to notify compare event */
>> +void colo_notify_compares_event(void *opaque, int event, Error **errp)
>> +{
>> +    CompareState *s;
>> +
>> +    qemu_mutex_lock(&event_mtx);
>> +    QTAILQ_FOREACH(s, &net_compares, next) {
>> +        s->event = event;
>> +        qemu_bh_schedule(s->event_bh);
>> +        event_unhandled_count++;
>> +    }
>> +    /* Wait all compare threads to finish handling this event */
>> +    while (event_unhandled_count > 0) {
>> +        qemu_cond_wait(&event_complete_cond, &event_mtx);
>> +    }
>> +
>> +    qemu_mutex_unlock(&event_mtx);
>> +}
>> +
>>   static void colo_compare_timer_init(CompareState *s)
>>   {
>>       AioContext *ctx = iothread_get_aio_context(s->iothread);
>> @@ -756,6 +789,28 @@ static void colo_compare_timer_del(CompareState *s)
>>       }
>>    }
>>   +static void colo_flush_packets(void *opaque, void *user_data);
>> +
>> +static void colo_compare_handle_event(void *opaque)
>> +{
>> +    CompareState *s = opaque;
>> +
>> +    switch (s->event) {
>> +    case COLO_EVENT_CHECKPOINT:
>> +        g_queue_foreach(&s->conn_list, colo_flush_packets, s);
>> +        break;
>> +    case COLO_EVENT_FAILOVER:
>> +        break;
>> +    default:
>> +        break;
>> +    }
>> +    qemu_mutex_lock(&event_mtx);
>>
>
> Isn't this a deadlock? Since colo_notify_compares_event() won't release
> event_mtx until event_unhandled_count reaches zero.
>
>
Good catch!
I will fix it in next version.


>
> +    assert(event_unhandled_count > 0);
>> +    event_unhandled_count--;
>> +    qemu_cond_broadcast(&event_complete_cond);
>> +    qemu_mutex_unlock(&event_mtx);
>> +}
>> +
>>   static void colo_compare_iothread(CompareState *s)
>>   {
>>       object_ref(OBJECT(s->iothread));
>> @@ -769,6 +824,7 @@ static void colo_compare_iothread(CompareState *s)
>>                                s, s->worker_context, true);
>>         colo_compare_timer_init(s);
>> +    s->event_bh = qemu_bh_new(colo_compare_handle_event, s);
>>   }
>>     static char *compare_get_pri_indev(Object *obj, Error **errp)
>> @@ -926,8 +982,13 @@ static void colo_compare_complete(UserCreatable
>> *uc, Error **errp)
>>       net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize,
>> s->vnet_hdr);
>>       net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize,
>> s->vnet_hdr);
>>   +    QTAILQ_INSERT_TAIL(&net_compares, s, next);
>> +
>>       g_queue_init(&s->conn_list);
>>   +    qemu_mutex_init(&event_mtx);
>> +    qemu_cond_init(&event_complete_cond);
>> +
>>       s->connection_track_table = g_hash_table_new_full(connecti
>> on_key_hash,
>>
>> connection_key_equal,
>>                                                         g_free,
>> @@ -990,6 +1051,7 @@ static void colo_compare_init(Object *obj)
>>   static void colo_compare_finalize(Object *obj)
>>   {
>>       CompareState *s = COLO_COMPARE(obj);
>> +    CompareState *tmp = NULL;
>>         qemu_chr_fe_deinit(&s->chr_pri_in, false);
>>       qemu_chr_fe_deinit(&s->chr_sec_in, false);
>> @@ -997,6 +1059,16 @@ static void colo_compare_finalize(Object *obj)
>>       if (s->iothread) {
>>           colo_compare_timer_del(s);
>>       }
>> +
>> +    qemu_bh_delete(s->event_bh);
>> +
>> +    QTAILQ_FOREACH(tmp, &net_compares, next) {
>> +        if (!strcmp(tmp->outdev, s->outdev)) {
>>
>
> This looks not elegant, can we compare by address or just use QLIST?
>
>
OK, I will compare by address in next version.

Thanks
Zhang Chen


> Thanks
>
>
> +            QTAILQ_REMOVE(&net_compares, s, next);
>> +            break;
>> +        }
>> +    }
>> +
>>       /* Release all unhandled packets after compare thead exited */
>>       g_queue_foreach(&s->conn_list, colo_flush_packets, s);
>>   @@ -1009,6 +1081,10 @@ static void colo_compare_finalize(Object *obj)
>>       if (s->iothread) {
>>           object_unref(OBJECT(s->iothread));
>>       }
>> +
>> +    qemu_mutex_destroy(&event_mtx);
>> +    qemu_cond_destroy(&event_complete_cond);
>> +
>>       g_free(s->pri_indev);
>>       g_free(s->sec_indev);
>>       g_free(s->outdev);
>> diff --git a/net/colo-compare.h b/net/colo-compare.h
>> new file mode 100644
>> index 0000000000..1b1ce76aea
>> --- /dev/null
>> +++ b/net/colo-compare.h
>> @@ -0,0 +1,22 @@
>> +/*
>> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service
>> (COLO)
>> + * (a.k.a. Fault Tolerance or Continuous Replication)
>> + *
>> + * Copyright (c) 2017 HUAWEI TECHNOLOGIES CO., LTD.
>> + * Copyright (c) 2017 FUJITSU LIMITED
>> + * Copyright (c) 2017 Intel Corporation
>> + *
>> + * Authors:
>> + *    zhanghailiang <address@hidden>
>> + *    Zhang Chen <address@hidden>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2 or
>> + * later.  See the COPYING file in the top-level directory.
>> + */
>> +
>> +#ifndef QEMU_COLO_COMPARE_H
>> +#define QEMU_COLO_COMPARE_H
>> +
>> +void colo_notify_compares_event(void *opaque, int event, Error **errp);
>> +
>> +#endif /* QEMU_COLO_COMPARE_H */
>>
>
>