[Qemu-devel] [PATCHv5 00/03] Replace the COLO comparing thread with IOThread

[Qemu-devel] [PATCHv5 00/03] Replace the COLO comparing thread with IOThread

[Wang yong]

From: Wang Yong <wang.yong155@zte.com.cn>

It's a good idea to use IOThread instead of COLO comparing thread.
comparing thread can be completely replaced by IOThread, so this idea came.

This series of updates mainly include IOThread supports the GMainContext
event loop, then the old packet regularly check and primary/secondary network
packets compare all into the IOThread processing.

Please review,thanks.

wangyong(3):
qemu-iothread: IOThread supports the GMainContext event loop
colo-compare: Use IOThread to Check old packet regularly and
 Process pactkets of the primary
colo-compare: Update the COLO document to add the IOThread
 configuration

 include/sysemu/iothread.h | 10 +++++++++
 iothread.c                | 54 +++++++++++++++++++++++++++++++++++++++++++++++
 net/colo-compare.c | 75 ++++++++++++++++++++++++++++--------------------------
 docs/colo-proxy.txt | 3 ++- 
 4 file changed, 97 insertions(+), 39 deletions(-)

--
1.8.3.1

[Qemu-devel] [PATCHv5 01/03] qemu-iothread: IOThread supports the GMainContext event loop

[Wang yong]

From: Wang Yong <wang.yong155@zte.com.cn>

IOThread uses AioContext event loop and does not run a GMainContext.
Therefore,chardev cannot work in IOThread,such as the chardev is
used for colo-compare packets reception.

This patch makes the IOThread run the GMainContext event loop,
chardev and IOThread can work together.

Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
---
 include/sysemu/iothread.h |  4 ++++
 iothread.c                | 45 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 49 insertions(+)

diff --git a/include/sysemu/iothread.h b/include/sysemu/iothread.h
index e6da1a4..d2985b3 100644
--- a/include/sysemu/iothread.h
+++ b/include/sysemu/iothread.h
@@ -24,6 +24,9 @@ typedef struct {
 
     QemuThread thread;
     AioContext *ctx;
+    GMainContext *worker_context;
+    GMainLoop *main_loop;
+    GOnce once;
     QemuMutex init_done_lock;
     QemuCond init_done_cond;    /* is thread initialization done? */
     bool stopping;
@@ -41,5 +44,6 @@ typedef struct {
 char *iothread_get_id(IOThread *iothread);
 AioContext *iothread_get_aio_context(IOThread *iothread);
 void iothread_stop_all(void);
+GMainContext *iothread_get_g_main_context(IOThread *iothread);
 
 #endif /* IOTHREAD_H */
diff --git a/iothread.c b/iothread.c
index beeb870..44c8944 100644
--- a/iothread.c
+++ b/iothread.c
@@ -57,6 +57,23 @@ static void *iothread_run(void *opaque)
 
     while (!atomic_read(&iothread->stopping)) {
         aio_poll(iothread->ctx, true);
+
+        if (atomic_read(&iothread->worker_context)) {
+            GMainLoop *loop;
+
+            g_main_context_push_thread_default(iothread->worker_context);
+            iothread->main_loop =
+                g_main_loop_new(iothread->worker_context, TRUE);
+            loop = iothread->main_loop;
+
+            g_main_loop_run(iothread->main_loop);
+            iothread->main_loop = NULL;
+            g_main_loop_unref(loop);
+
+            g_main_context_pop_thread_default(iothread->worker_context);
+            g_main_context_unref(iothread->worker_context);
+            iothread->worker_context = NULL;
+        }
     }
 
     rcu_unregister_thread();
@@ -73,6 +90,9 @@ static int iothread_stop(Object *object, void *opaque)
     }
     iothread->stopping = true;
     aio_notify(iothread->ctx);
+    if (atomic_read(&iothread->main_loop)) {
+        g_main_loop_quit(iothread->main_loop);
+    }
     qemu_thread_join(&iothread->thread);
     return 0;
 }
@@ -125,6 +145,7 @@ static void iothread_complete(UserCreatable *obj, Error **errp)
 
     qemu_mutex_init(&iothread->init_done_lock);
     qemu_cond_init(&iothread->init_done_cond);
+    iothread->once = (GOnce) G_ONCE_INIT;
 
     /* This assumes we are called from a thread with useful CPU affinity for us
      * to inherit.
@@ -309,3 +330,27 @@ void iothread_stop_all(void)
 
     object_child_foreach(container, iothread_stop, NULL);
 }
+
+static gpointer iothread_g_main_context_init(gpointer opaque)
+{
+    AioContext *ctx;
+    IOThread *iothread = opaque;
+    GSource *source;
+
+    iothread->worker_context = g_main_context_new();
+
+    ctx = iothread_get_aio_context(iothread);
+    source = aio_get_g_source(ctx);
+    g_source_attach(source, iothread->worker_context);
+    g_source_unref(source);
+
+    aio_notify(iothread->ctx);
+    return NULL;
+}
+
+GMainContext *iothread_get_g_main_context(IOThread *iothread)
+{
+    g_once(&iothread->once, iothread_g_main_context_init, iothread);
+
+    return iothread->worker_context;
+}
-- 
1.8.3.1

Re: [Qemu-devel] [PATCHv5 01/03] qemu-iothread: IOThread supports the GMainContext event loop

[Fam Zheng]

On Tue, 08/29 15:22, Wang yong wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
> 
> IOThread uses AioContext event loop and does not run a GMainContext.
> Therefore,chardev cannot work in IOThread,such as the chardev is
> used for colo-compare packets reception.
> 
> This patch makes the IOThread run the GMainContext event loop,
> chardev and IOThread can work together.
> 
> Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
> Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
> ---
>  include/sysemu/iothread.h |  4 ++++
>  iothread.c                | 45 +++++++++++++++++++++++++++++++++++++++++++++
>  2 files changed, 49 insertions(+)
> 
> diff --git a/include/sysemu/iothread.h b/include/sysemu/iothread.h
> index e6da1a4..d2985b3 100644
> --- a/include/sysemu/iothread.h
> +++ b/include/sysemu/iothread.h
> @@ -24,6 +24,9 @@ typedef struct {
>  
>      QemuThread thread;
>      AioContext *ctx;
> +    GMainContext *worker_context;
> +    GMainLoop *main_loop;
> +    GOnce once;
>      QemuMutex init_done_lock;
>      QemuCond init_done_cond;    /* is thread initialization done? */
>      bool stopping;
> @@ -41,5 +44,6 @@ typedef struct {
>  char *iothread_get_id(IOThread *iothread);
>  AioContext *iothread_get_aio_context(IOThread *iothread);
>  void iothread_stop_all(void);
> +GMainContext *iothread_get_g_main_context(IOThread *iothread);
>  
>  #endif /* IOTHREAD_H */
> diff --git a/iothread.c b/iothread.c
> index beeb870..44c8944 100644
> --- a/iothread.c
> +++ b/iothread.c
> @@ -57,6 +57,23 @@ static void *iothread_run(void *opaque)
>  
>      while (!atomic_read(&iothread->stopping)) {
>          aio_poll(iothread->ctx, true);
> +
> +        if (atomic_read(&iothread->worker_context)) {
> +            GMainLoop *loop;
> +
> +            g_main_context_push_thread_default(iothread->worker_context);
> +            iothread->main_loop =
> +                g_main_loop_new(iothread->worker_context, TRUE);
> +            loop = iothread->main_loop;
> +
> +            g_main_loop_run(iothread->main_loop);
> +            iothread->main_loop = NULL;
> +            g_main_loop_unref(loop);
> +
> +            g_main_context_pop_thread_default(iothread->worker_context);
> +            g_main_context_unref(iothread->worker_context);
> +            iothread->worker_context = NULL;
> +        }
>      }
>  
>      rcu_unregister_thread();
> @@ -73,6 +90,9 @@ static int iothread_stop(Object *object, void *opaque)
>      }
>      iothread->stopping = true;
>      aio_notify(iothread->ctx);
> +    if (atomic_read(&iothread->main_loop)) {
> +        g_main_loop_quit(iothread->main_loop);
> +    }
>      qemu_thread_join(&iothread->thread);
>      return 0;
>  }
> @@ -125,6 +145,7 @@ static void iothread_complete(UserCreatable *obj, Error **errp)
>  
>      qemu_mutex_init(&iothread->init_done_lock);
>      qemu_cond_init(&iothread->init_done_cond);
> +    iothread->once = (GOnce) G_ONCE_INIT;

In last review I suggested removing this type cast, otherwise looks good. Drop
it and please add

Reviewed-by: Fam Zheng <famz@redhat.com>

>  
>      /* This assumes we are called from a thread with useful CPU affinity for us
>       * to inherit.
> @@ -309,3 +330,27 @@ void iothread_stop_all(void)
>  
>      object_child_foreach(container, iothread_stop, NULL);
>  }
> +
> +static gpointer iothread_g_main_context_init(gpointer opaque)
> +{
> +    AioContext *ctx;
> +    IOThread *iothread = opaque;
> +    GSource *source;
> +
> +    iothread->worker_context = g_main_context_new();
> +
> +    ctx = iothread_get_aio_context(iothread);
> +    source = aio_get_g_source(ctx);
> +    g_source_attach(source, iothread->worker_context);
> +    g_source_unref(source);
> +
> +    aio_notify(iothread->ctx);
> +    return NULL;
> +}
> +
> +GMainContext *iothread_get_g_main_context(IOThread *iothread)
> +{
> +    g_once(&iothread->once, iothread_g_main_context_init, iothread);
> +
> +    return iothread->worker_context;
> +}
> -- 
> 1.8.3.1
> 
> 

Re: [Qemu-devel] [PATCHv5 01/03] qemu-iothread: IOThread supports the GMainContext event loop

[Fam Zheng]

On Thu, 08/31 10:18, Fam Zheng wrote:
> On Tue, 08/29 15:22, Wang yong wrote:
> > From: Wang Yong <wang.yong155@zte.com.cn>
> > 
> > IOThread uses AioContext event loop and does not run a GMainContext.
> > Therefore,chardev cannot work in IOThread,such as the chardev is
> > used for colo-compare packets reception.
> > 
> > This patch makes the IOThread run the GMainContext event loop,
> > chardev and IOThread can work together.
> > 
> > Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
> > Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
> > ---
> >  include/sysemu/iothread.h |  4 ++++
> >  iothread.c                | 45 +++++++++++++++++++++++++++++++++++++++++++++
> >  2 files changed, 49 insertions(+)
> > 
> > diff --git a/include/sysemu/iothread.h b/include/sysemu/iothread.h
> > index e6da1a4..d2985b3 100644
> > --- a/include/sysemu/iothread.h
> > +++ b/include/sysemu/iothread.h
> > @@ -24,6 +24,9 @@ typedef struct {
> >  
> >      QemuThread thread;
> >      AioContext *ctx;
> > +    GMainContext *worker_context;
> > +    GMainLoop *main_loop;
> > +    GOnce once;
> >      QemuMutex init_done_lock;
> >      QemuCond init_done_cond;    /* is thread initialization done? */
> >      bool stopping;
> > @@ -41,5 +44,6 @@ typedef struct {
> >  char *iothread_get_id(IOThread *iothread);
> >  AioContext *iothread_get_aio_context(IOThread *iothread);
> >  void iothread_stop_all(void);
> > +GMainContext *iothread_get_g_main_context(IOThread *iothread);
> >  
> >  #endif /* IOTHREAD_H */
> > diff --git a/iothread.c b/iothread.c
> > index beeb870..44c8944 100644
> > --- a/iothread.c
> > +++ b/iothread.c
> > @@ -57,6 +57,23 @@ static void *iothread_run(void *opaque)
> >  
> >      while (!atomic_read(&iothread->stopping)) {
> >          aio_poll(iothread->ctx, true);
> > +
> > +        if (atomic_read(&iothread->worker_context)) {
> > +            GMainLoop *loop;
> > +
> > +            g_main_context_push_thread_default(iothread->worker_context);
> > +            iothread->main_loop =
> > +                g_main_loop_new(iothread->worker_context, TRUE);
> > +            loop = iothread->main_loop;
> > +
> > +            g_main_loop_run(iothread->main_loop);
> > +            iothread->main_loop = NULL;
> > +            g_main_loop_unref(loop);
> > +
> > +            g_main_context_pop_thread_default(iothread->worker_context);
> > +            g_main_context_unref(iothread->worker_context);
> > +            iothread->worker_context = NULL;
> > +        }
> >      }
> >  
> >      rcu_unregister_thread();
> > @@ -73,6 +90,9 @@ static int iothread_stop(Object *object, void *opaque)
> >      }
> >      iothread->stopping = true;
> >      aio_notify(iothread->ctx);
> > +    if (atomic_read(&iothread->main_loop)) {
> > +        g_main_loop_quit(iothread->main_loop);
> > +    }
> >      qemu_thread_join(&iothread->thread);
> >      return 0;
> >  }
> > @@ -125,6 +145,7 @@ static void iothread_complete(UserCreatable *obj, Error **errp)
> >  
> >      qemu_mutex_init(&iothread->init_done_lock);
> >      qemu_cond_init(&iothread->init_done_cond);
> > +    iothread->once = (GOnce) G_ONCE_INIT;
> 
> In last review I suggested removing this type cast, otherwise looks good. Drop
> it and please add

No, Yong is right, because this is not varialble initializer, type cast is
required by C. Sorry for the noise.

Fam

Re: [Qemu-devel] [PATCHv5 01/03] qemu-iothread: IOThread supports the GMainContext event loop

[Stefan Hajnoczi]

On Tue, Aug 29, 2017 at 03:22:37PM +0800, Wang yong wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
> 
> IOThread uses AioContext event loop and does not run a GMainContext.
> Therefore,chardev cannot work in IOThread,such as the chardev is
> used for colo-compare packets reception.
> 
> This patch makes the IOThread run the GMainContext event loop,
> chardev and IOThread can work together.
> 
> Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
> Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
> ---
>  include/sysemu/iothread.h |  4 ++++
>  iothread.c                | 45 +++++++++++++++++++++++++++++++++++++++++++++
>  2 files changed, 49 insertions(+)

Reviewed-by: Stefan Hajnoczi <stefanha@redhat.com>

[Qemu-devel] [PATCHv5 02/03] colo-compare: Use IOThread to Check old packet regularly and Process pactkets of the primary

[Wang yong]

From: Wang Yong <wang.yong155@zte.com.cn>

Remove the task which check old packet in the comparing thread,
then use IOthread context timer to handle it.

Process pactkets in the IOThread which arrived over the socket.
we use iothread_get_g_main_context to create a new g_main_loop in
the IOThread.then the packets from the primary and the secondary
are processed in the IOThread.

Finally remove the colo-compare thread using the IOThread instead.

Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
---
 net/colo-compare.c | 83 +++++++++++++++++++++++++++++-------------------------
 1 file changed, 45 insertions(+), 38 deletions(-)

diff --git a/net/colo-compare.c b/net/colo-compare.c
index 5fe8e3f..b2a2a13 100644
--- a/net/colo-compare.c
+++ b/net/colo-compare.c
@@ -29,6 +29,7 @@
 #include "qemu/sockets.h"
 #include "qapi-visit.h"
 #include "net/colo.h"
+#include "sysemu/iothread.h"
 
 #define TYPE_COLO_COMPARE "colo-compare"
 #define COLO_COMPARE(obj) \
@@ -82,11 +83,10 @@ typedef struct CompareState {
     GQueue conn_list;
     /* hashtable to save connection */
     GHashTable *connection_track_table;
-    /* compare thread, a thread for each NIC */
-    QemuThread thread;
 
+    IOThread *iothread;
     GMainContext *worker_context;
-    GMainLoop *compare_loop;
+    QEMUTimer *packet_check_timer;
 } CompareState;
 
 typedef struct CompareClass {
@@ -597,22 +597,40 @@ static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
  * Check old packet regularly so it can watch for any packets
  * that the secondary hasn't produced equivalents of.
  */
-static gboolean check_old_packet_regular(void *opaque)
+static void check_old_packet_regular(void *opaque)
 {
     CompareState *s = opaque;
 
     /* if have old packet we will notify checkpoint */
     colo_old_packet_check(s);
+    timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
+                REGULAR_PACKET_CHECK_MS);
+}
+
+static void colo_compare_timer_init(CompareState *s)
+{
+    AioContext *ctx = iothread_get_aio_context(s->iothread);
 
-    return TRUE;
+    s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
+                                SCALE_MS, check_old_packet_regular,
+                                s);
+    timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
+                    REGULAR_PACKET_CHECK_MS);
 }
 
-static void *colo_compare_thread(void *opaque)
+static void colo_compare_timer_del(CompareState *s)
 {
-    CompareState *s = opaque;
-    GSource *timeout_source;
+    if (s->packet_check_timer) {
+        timer_del(s->packet_check_timer);
+        timer_free(s->packet_check_timer);
+        s->packet_check_timer = NULL;
+    }
+ }
 
-    s->worker_context = g_main_context_new();
+static void colo_compare_iothread(CompareState *s)
+{
+    object_ref(OBJECT(s->iothread));
+    s->worker_context = iothread_get_g_main_context(s->iothread);
 
     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
                              compare_pri_chr_in, NULL, NULL,
@@ -621,20 +639,7 @@ static void *colo_compare_thread(void *opaque)
                              compare_sec_chr_in, NULL, NULL,
                              s, s->worker_context, true);
 
-    s->compare_loop = g_main_loop_new(s->worker_context, FALSE);
-
-    /* To kick any packets that the secondary doesn't match */
-    timeout_source = g_timeout_source_new(REGULAR_PACKET_CHECK_MS);
-    g_source_set_callback(timeout_source,
-                          (GSourceFunc)check_old_packet_regular, s, NULL);
-    g_source_attach(timeout_source, s->worker_context);
-
-    g_main_loop_run(s->compare_loop);
-
-    g_source_unref(timeout_source);
-    g_main_loop_unref(s->compare_loop);
-    g_main_context_unref(s->worker_context);
-    return NULL;
+    colo_compare_timer_init(s);
 }
 
 static char *compare_get_pri_indev(Object *obj, Error **errp)
@@ -759,12 +764,10 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
 {
     CompareState *s = COLO_COMPARE(uc);
     Chardev *chr;
-    char thread_name[64];
-    static int compare_id;
 
-    if (!s->pri_indev || !s->sec_indev || !s->outdev) {
+    if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
         error_setg(errp, "colo compare needs 'primary_in' ,"
-                   "'secondary_in','outdev' property set");
+                   "'secondary_in','outdev','iothread' property set");
         return;
     } else if (!strcmp(s->pri_indev, s->outdev) ||
                !strcmp(s->sec_indev, s->outdev) ||
@@ -799,12 +802,7 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
                                                       g_free,
                                                       connection_destroy);
 
-    sprintf(thread_name, "colo-compare %d", compare_id);
-    qemu_thread_create(&s->thread, thread_name,
-                       colo_compare_thread, s,
-                       QEMU_THREAD_JOINABLE);
-    compare_id++;
-
+    colo_compare_iothread(s);
     return;
 }
 
@@ -848,6 +846,10 @@ static void colo_compare_init(Object *obj)
     object_property_add_str(obj, "outdev",
                             compare_get_outdev, compare_set_outdev,
                             NULL);
+    object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
+                            (Object **)&s->iothread,
+                            object_property_allow_set_link,
+                            OBJ_PROP_LINK_UNREF_ON_RELEASE, NULL);
 
     s->vnet_hdr = false;
     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
@@ -861,16 +863,21 @@ static void colo_compare_finalize(Object *obj)
     qemu_chr_fe_deinit(&s->chr_pri_in, false);
     qemu_chr_fe_deinit(&s->chr_sec_in, false);
     qemu_chr_fe_deinit(&s->chr_out, false);
-
-    g_main_loop_quit(s->compare_loop);
-    qemu_thread_join(&s->thread);
-
+    if (s->iothread) {
+        colo_compare_timer_del(s);
+    }
     /* Release all unhandled packets after compare thead exited */
     g_queue_foreach(&s->conn_list, colo_flush_packets, s);
 
     g_queue_clear(&s->conn_list);
 
-    g_hash_table_destroy(s->connection_track_table);
+    if (s->connection_track_table) {
+        g_hash_table_destroy(s->connection_track_table);
+    }
+
+    if (s->iothread) {
+        object_unref(OBJECT(s->iothread));
+    }
     g_free(s->pri_indev);
     g_free(s->sec_indev);
     g_free(s->outdev);
-- 
1.8.3.1

Re: [Qemu-devel] [PATCHv5 02/03] colo-compare: Use IOThread to Check old packet regularly and Process pactkets of the primary

[Zhang Chen]

On 08/29/2017 03:22 PM, Wang yong wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
>
> Remove the task which check old packet in the comparing thread,
> then use IOthread context timer to handle it.
>
> Process pactkets in the IOThread which arrived over the socket.
> we use iothread_get_g_main_context to create a new g_main_loop in
> the IOThread.then the packets from the primary and the secondary
> are processed in the IOThread.
>
> Finally remove the colo-compare thread using the IOThread instead.
>
> Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
> Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>

Reviewed-by: Zhang Chen<zhangchen.fnst@cn.fujitsu.com>

Hi~ Jason.
Have any comments to this series?

Thanks
Zhang Chen

> ---
>   net/colo-compare.c | 83 +++++++++++++++++++++++++++++-------------------------
>   1 file changed, 45 insertions(+), 38 deletions(-)
>
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index 5fe8e3f..b2a2a13 100644
> --- a/net/colo-compare.c
> +++ b/net/colo-compare.c
> @@ -29,6 +29,7 @@
>   #include "qemu/sockets.h"
>   #include "qapi-visit.h"
>   #include "net/colo.h"
> +#include "sysemu/iothread.h"
>   
>   #define TYPE_COLO_COMPARE "colo-compare"
>   #define COLO_COMPARE(obj) \
> @@ -82,11 +83,10 @@ typedef struct CompareState {
>       GQueue conn_list;
>       /* hashtable to save connection */
>       GHashTable *connection_track_table;
> -    /* compare thread, a thread for each NIC */
> -    QemuThread thread;
>   
> +    IOThread *iothread;
>       GMainContext *worker_context;
> -    GMainLoop *compare_loop;
> +    QEMUTimer *packet_check_timer;
>   } CompareState;
>   
>   typedef struct CompareClass {
> @@ -597,22 +597,40 @@ static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
>    * Check old packet regularly so it can watch for any packets
>    * that the secondary hasn't produced equivalents of.
>    */
> -static gboolean check_old_packet_regular(void *opaque)
> +static void check_old_packet_regular(void *opaque)
>   {
>       CompareState *s = opaque;
>   
>       /* if have old packet we will notify checkpoint */
>       colo_old_packet_check(s);
> +    timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
> +                REGULAR_PACKET_CHECK_MS);
> +}
> +
> +static void colo_compare_timer_init(CompareState *s)
> +{
> +    AioContext *ctx = iothread_get_aio_context(s->iothread);
>   
> -    return TRUE;
> +    s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
> +                                SCALE_MS, check_old_packet_regular,
> +                                s);
> +    timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
> +                    REGULAR_PACKET_CHECK_MS);
>   }
>   
> -static void *colo_compare_thread(void *opaque)
> +static void colo_compare_timer_del(CompareState *s)
>   {
> -    CompareState *s = opaque;
> -    GSource *timeout_source;
> +    if (s->packet_check_timer) {
> +        timer_del(s->packet_check_timer);
> +        timer_free(s->packet_check_timer);
> +        s->packet_check_timer = NULL;
> +    }
> + }
>   
> -    s->worker_context = g_main_context_new();
> +static void colo_compare_iothread(CompareState *s)
> +{
> +    object_ref(OBJECT(s->iothread));
> +    s->worker_context = iothread_get_g_main_context(s->iothread);
>   
>       qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
>                                compare_pri_chr_in, NULL, NULL,
> @@ -621,20 +639,7 @@ static void *colo_compare_thread(void *opaque)
>                                compare_sec_chr_in, NULL, NULL,
>                                s, s->worker_context, true);
>   
> -    s->compare_loop = g_main_loop_new(s->worker_context, FALSE);
> -
> -    /* To kick any packets that the secondary doesn't match */
> -    timeout_source = g_timeout_source_new(REGULAR_PACKET_CHECK_MS);
> -    g_source_set_callback(timeout_source,
> -                          (GSourceFunc)check_old_packet_regular, s, NULL);
> -    g_source_attach(timeout_source, s->worker_context);
> -
> -    g_main_loop_run(s->compare_loop);
> -
> -    g_source_unref(timeout_source);
> -    g_main_loop_unref(s->compare_loop);
> -    g_main_context_unref(s->worker_context);
> -    return NULL;
> +    colo_compare_timer_init(s);
>   }
>   
>   static char *compare_get_pri_indev(Object *obj, Error **errp)
> @@ -759,12 +764,10 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
>   {
>       CompareState *s = COLO_COMPARE(uc);
>       Chardev *chr;
> -    char thread_name[64];
> -    static int compare_id;
>   
> -    if (!s->pri_indev || !s->sec_indev || !s->outdev) {
> +    if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
>           error_setg(errp, "colo compare needs 'primary_in' ,"
> -                   "'secondary_in','outdev' property set");
> +                   "'secondary_in','outdev','iothread' property set");
>           return;
>       } else if (!strcmp(s->pri_indev, s->outdev) ||
>                  !strcmp(s->sec_indev, s->outdev) ||
> @@ -799,12 +802,7 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
>                                                         g_free,
>                                                         connection_destroy);
>   
> -    sprintf(thread_name, "colo-compare %d", compare_id);
> -    qemu_thread_create(&s->thread, thread_name,
> -                       colo_compare_thread, s,
> -                       QEMU_THREAD_JOINABLE);
> -    compare_id++;
> -
> +    colo_compare_iothread(s);
>       return;
>   }
>   
> @@ -848,6 +846,10 @@ static void colo_compare_init(Object *obj)
>       object_property_add_str(obj, "outdev",
>                               compare_get_outdev, compare_set_outdev,
>                               NULL);
> +    object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
> +                            (Object **)&s->iothread,
> +                            object_property_allow_set_link,
> +                            OBJ_PROP_LINK_UNREF_ON_RELEASE, NULL);
>   
>       s->vnet_hdr = false;
>       object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
> @@ -861,16 +863,21 @@ static void colo_compare_finalize(Object *obj)
>       qemu_chr_fe_deinit(&s->chr_pri_in, false);
>       qemu_chr_fe_deinit(&s->chr_sec_in, false);
>       qemu_chr_fe_deinit(&s->chr_out, false);
> -
> -    g_main_loop_quit(s->compare_loop);
> -    qemu_thread_join(&s->thread);
> -
> +    if (s->iothread) {
> +        colo_compare_timer_del(s);
> +    }
>       /* Release all unhandled packets after compare thead exited */
>       g_queue_foreach(&s->conn_list, colo_flush_packets, s);
>   
>       g_queue_clear(&s->conn_list);
>   
> -    g_hash_table_destroy(s->connection_track_table);
> +    if (s->connection_track_table) {
> +        g_hash_table_destroy(s->connection_track_table);
> +    }
> +
> +    if (s->iothread) {
> +        object_unref(OBJECT(s->iothread));
> +    }
>       g_free(s->pri_indev);
>       g_free(s->sec_indev);
>       g_free(s->outdev);

-- 
Thanks
Zhang Chen

[Qemu-devel] [PATCHv5 03/03] colo-compare: Update the COLO document to add the IOThread configuration

[Wang yong]

From: Wang Yong <wang.yong155@zte.com.cn>

Update colo-proxy.txt,add IOThread configuration.
Later we have to configure IOThread,if not COLO can not work.

Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
---
 docs/colo-proxy.txt | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/docs/colo-proxy.txt b/docs/colo-proxy.txt
index c4941de..ce3f783 100644
--- a/docs/colo-proxy.txt
+++ b/docs/colo-proxy.txt
@@ -170,10 +170,11 @@ Primary(ip:3.3.3.3):
 -chardev socket,id=compare0-0,host=3.3.3.3,port=9001
 -chardev socket,id=compare_out,host=3.3.3.3,port=9005,server,nowait
 -chardev socket,id=compare_out0,host=3.3.3.3,port=9005
+-object iothread,id=iothread1
 -object filter-mirror,id=m0,netdev=hn0,queue=tx,outdev=mirror0
 -object filter-redirector,netdev=hn0,id=redire0,queue=rx,indev=compare_out
 -object filter-redirector,netdev=hn0,id=redire1,queue=rx,outdev=compare0
--object colo-compare,id=comp0,primary_in=compare0-0,secondary_in=compare1,outdev=compare_out0
+-object colo-compare,id=comp0,primary_in=compare0-0,secondary_in=compare1,outdev=compare_out0,iothread=iothread1
 
 Secondary(ip:3.3.3.8):
 -netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down script=/etc/qemu-ifdown
-- 
1.8.3.1

Re: [Qemu-devel] [PATCHv5 03/03] colo-compare: Update the COLO document to add the IOThread configuration

[Zhang Chen]

On 08/29/2017 03:22 PM, Wang yong wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
>
> Update colo-proxy.txt,add IOThread configuration.
> Later we have to configure IOThread,if not COLO can not work.
>
> Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
> Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>

Reviewed-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>

> ---
>   docs/colo-proxy.txt | 3 ++-
>   1 file changed, 2 insertions(+), 1 deletion(-)
>
> diff --git a/docs/colo-proxy.txt b/docs/colo-proxy.txt
> index c4941de..ce3f783 100644
> --- a/docs/colo-proxy.txt
> +++ b/docs/colo-proxy.txt
> @@ -170,10 +170,11 @@ Primary(ip:3.3.3.3):
>   -chardev socket,id=compare0-0,host=3.3.3.3,port=9001
>   -chardev socket,id=compare_out,host=3.3.3.3,port=9005,server,nowait
>   -chardev socket,id=compare_out0,host=3.3.3.3,port=9005
> +-object iothread,id=iothread1
>   -object filter-mirror,id=m0,netdev=hn0,queue=tx,outdev=mirror0
>   -object filter-redirector,netdev=hn0,id=redire0,queue=rx,indev=compare_out
>   -object filter-redirector,netdev=hn0,id=redire1,queue=rx,outdev=compare0
> --object colo-compare,id=comp0,primary_in=compare0-0,secondary_in=compare1,outdev=compare_out0
> +-object colo-compare,id=comp0,primary_in=compare0-0,secondary_in=compare1,outdev=compare_out0,iothread=iothread1
>   
>   Secondary(ip:3.3.3.8):
>   -netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down script=/etc/qemu-ifdown

-- 
Thanks
Zhang Chen

Re: [Qemu-devel] [PATCHv5 00/03] Replace the COLO comparing thread with IOThread

[Jason Wang]

On 2017年08月29日 15:22, Wang yong wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
>
> It's a good idea to use IOThread instead of COLO comparing thread.
> comparing thread can be completely replaced by IOThread, so this idea came.
>
> This series of updates mainly include IOThread supports the GMainContext
> event loop, then the old packet regularly check and primary/secondary network
> packets compare all into the IOThread processing.
>
> Please review,thanks.
>
> wangyong(3):
> qemu-iothread: IOThread supports the GMainContext event loop
> colo-compare: Use IOThread to Check old packet regularly and
>   Process pactkets of the primary
> colo-compare: Update the COLO document to add the IOThread
>   configuration
>
>   include/sysemu/iothread.h | 10 +++++++++
>   iothread.c                | 54 +++++++++++++++++++++++++++++++++++++++++++++++
>   net/colo-compare.c | 75 ++++++++++++++++++++++++++++--------------------------
>   docs/colo-proxy.txt | 3 ++-
>   4 file changed, 97 insertions(+), 39 deletions(-)
>
> --
> 1.8.3.1
>

Applied.

Thanks

Re: [Qemu-devel] [PATCHv5 02/03] colo-compare: Use IOThread to Check old packet regularly and Process pactkets of the primary

[wang.yong155]

>Remove the task which check old packet in the comparing thread,

>then use IOthread context timer to handle it.

>

>Process pactkets in the IOThread which arrived over the socket.

>we use iothread_get_g_main_context to create a new g_main_loop in

>the IOThread.then the packets from the primary and the secondary

>are processed in the IOThread.

>

>Finally remove the colo-compare thread using the IOThread instead.

>

>Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>

>Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>

>---

> net/colo-compare.c | 83 +++++++++++++++++++++++++++++-------------------------

> 1 file changed, 45 insertions(+), 38 deletions(-)




Hi Jason,

How about this series?




Thanks






原始邮件



发件人：王勇10170530
收件人： <pbonzini@redhat.com> <stefanha@redhat.com> <famz@redhat.com> <jasowang@redhat.com> <zhangchen.fnst@cn.fujitsu.com> <zhang.zhanghailiang@huawei.com>
抄送人：王勇10170530王广10165992 <lizhijian@cn.fujitsu.com> <qemu-devel@nongnu.org>
日 期 ：2017年08月29日 15:22
主 题 ：[PATCHv5 02/03] colo-compare: Use IOThread to Check old packet regularly and Process pactkets of the primary





From: Wang Yong <wang.yong155@zte.com.cn>

Remove the task which check old packet in the comparing thread,
then use IOthread context timer to handle it.

Process pactkets in the IOThread which arrived over the socket.
we use iothread_get_g_main_context to create a new g_main_loop in
the IOThread.then the packets from the primary and the secondary
are processed in the IOThread.

Finally remove the colo-compare thread using the IOThread instead.

Signed-off-by: Wang Yong <wang.yong155@zte.com.cn>
Signed-off-by: Wang Guang <wang.guang55@zte.com.cn>
---
 net/colo-compare.c | 83 +++++++++++++++++++++++++++++-------------------------
 1 file changed, 45 insertions(+), 38 deletions(-)

diff --git a/net/colo-compare.c b/net/colo-compare.c
index 5fe8e3f..b2a2a13 100644
--- a/net/colo-compare.c
+++ b/net/colo-compare.c
@@ -29,6 +29,7 @@
 #include "qemu/sockets.h"
 #include "qapi-visit.h"
 #include "net/colo.h"
+#include "sysemu/iothread.h"
 
 #define TYPE_COLO_COMPARE "colo-compare"
 #define COLO_COMPARE(obj) \
@@ -82,11 +83,10 @@ typedef struct CompareState {
     GQueue conn_list
     /* hashtable to save connection */
     GHashTable *connection_track_table
-    /* compare thread, a thread for each NIC */
-    QemuThread thread
 
+    IOThread *iothread
     GMainContext *worker_context
-    GMainLoop *compare_loop
+    QEMUTimer *packet_check_timer
 } CompareState
 
 typedef struct CompareClass {
@@ -597,22 +597,40 @@ static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
  * Check old packet regularly so it can watch for any packets
  * that the secondary hasn't produced equivalents of.
  */
-static gboolean check_old_packet_regular(void *opaque)
+static void check_old_packet_regular(void *opaque)
 {
     CompareState *s = opaque
 
     /* if have old packet we will notify checkpoint */
     colo_old_packet_check(s)
+    timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
+                REGULAR_PACKET_CHECK_MS)
+}
+
+static void colo_compare_timer_init(CompareState *s)
+{
+    AioContext *ctx = iothread_get_aio_context(s->iothread)
 
-    return TRUE
+    s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
+                                SCALE_MS, check_old_packet_regular,
+                                s)
+    timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
+                    REGULAR_PACKET_CHECK_MS)
 }
 
-static void *colo_compare_thread(void *opaque)
+static void colo_compare_timer_del(CompareState *s)
 {
-    CompareState *s = opaque
-    GSource *timeout_source
+    if (s->packet_check_timer) {
+        timer_del(s->packet_check_timer)
+        timer_free(s->packet_check_timer)
+        s->packet_check_timer = NULL
+    }
+ }
 
-    s->worker_context = g_main_context_new()
+static void colo_compare_iothread(CompareState *s)
+{
+    object_ref(OBJECT(s->iothread))
+    s->worker_context = iothread_get_g_main_context(s->iothread)
 
     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
                              compare_pri_chr_in, NULL, NULL,
@@ -621,20 +639,7 @@ static void *colo_compare_thread(void *opaque)
                              compare_sec_chr_in, NULL, NULL,
                              s, s->worker_context, true)
 
-    s->compare_loop = g_main_loop_new(s->worker_context, FALSE)
-
-    /* To kick any packets that the secondary doesn't match */
-    timeout_source = g_timeout_source_new(REGULAR_PACKET_CHECK_MS)
-    g_source_set_callback(timeout_source,
-                          (GSourceFunc)check_old_packet_regular, s, NULL)
-    g_source_attach(timeout_source, s->worker_context)
-
-    g_main_loop_run(s->compare_loop)
-
-    g_source_unref(timeout_source)
-    g_main_loop_unref(s->compare_loop)
-    g_main_context_unref(s->worker_context)
-    return NULL
+    colo_compare_timer_init(s)
 }
 
 static char *compare_get_pri_indev(Object *obj, Error **errp)
@@ -759,12 +764,10 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
 {
     CompareState *s = COLO_COMPARE(uc)
     Chardev *chr
-    char thread_name[64]
-    static int compare_id
 
-    if (!s->pri_indev || !s->sec_indev || !s->outdev) {
+    if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
         error_setg(errp, "colo compare needs 'primary_in' ,"
-                   "'secondary_in','outdev' property set")
+                   "'secondary_in','outdev','iothread' property set")
         return
     } else if (!strcmp(s->pri_indev, s->outdev) ||
                !strcmp(s->sec_indev, s->outdev) ||
@@ -799,12 +802,7 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
                                                       g_free,
                                                       connection_destroy)
 
-    sprintf(thread_name, "colo-compare %d", compare_id)
-    qemu_thread_create(&s->thread, thread_name,
-                       colo_compare_thread, s,
-                       QEMU_THREAD_JOINABLE)
-    compare_id++
-
+    colo_compare_iothread(s)
     return
 }
 
@@ -848,6 +846,10 @@ static void colo_compare_init(Object *obj)
     object_property_add_str(obj, "outdev",
                             compare_get_outdev, compare_set_outdev,
                             NULL)
+    object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
+                            (Object **)&s->iothread,
+                            object_property_allow_set_link,
+                            OBJ_PROP_LINK_UNREF_ON_RELEASE, NULL)
 
     s->vnet_hdr = false
     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
@@ -861,16 +863,21 @@ static void colo_compare_finalize(Object *obj)
     qemu_chr_fe_deinit(&s->chr_pri_in, false)
     qemu_chr_fe_deinit(&s->chr_sec_in, false)
     qemu_chr_fe_deinit(&s->chr_out, false)
-
-    g_main_loop_quit(s->compare_loop)
-    qemu_thread_join(&s->thread)
-
+    if (s->iothread) {
+        colo_compare_timer_del(s)
+    }
     /* Release all unhandled packets after compare thead exited */
     g_queue_foreach(&s->conn_list, colo_flush_packets, s)
 
     g_queue_clear(&s->conn_list)
 
-    g_hash_table_destroy(s->connection_track_table)
+    if (s->connection_track_table) {
+        g_hash_table_destroy(s->connection_track_table)
+    }
+
+    if (s->iothread) {
+        object_unref(OBJECT(s->iothread))
+    }
     g_free(s->pri_indev)
     g_free(s->sec_indev)
     g_free(s->outdev)
-- 
1.8.3.1