[Qemu-devel] [PATCHv2 00/04] Replace the COLO comparing thread with IOThread

[Qemu-devel] [PATCHv2 00/04] Replace the COLO comparing thread with IOThread

[Yong Wang]

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 the old packet regularly check and
primary/secondary network packets compare all into the IOThread processing.

Please review,thanks.

wangyong(4):
colo-compare: Use IOThread context timer to Check old packet
 regularly
colo-compare: Process pactkets in the IOThread of the primary
colo-compare: Update the COLO document to add the IOThread
 configuration
colo-compare: Update the COLO document to fix the processing
 of secondary packets in the main thread

 docs/colo-proxy.txt |   5 +-
 net/colo-compare.c  | 175 ++++++++++++++++++++++++++++++++++++++--------------
 net/colo.h          |   1 +
 3 files changed, 134 insertions(+), 47 deletions(-)

--
1.8.3.1

[Qemu-devel] [PATCHv2 01/04] colo-compare: Use IOThread context timer to Check old packet regularly

[Yong Wang]

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.

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

diff --git a/net/colo-compare.c b/net/colo-compare.c
index 2639c7f..b0942a4 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) \
@@ -86,6 +87,12 @@ typedef struct CompareState {
 
     GMainContext *worker_context;
     GMainLoop *compare_loop;
+
+    /*compare iothread*/
+    IOThread *iothread;
+    /*iothread context*/
+    AioContext *ctx;
+    QEMUTimer *packet_check_timer;
 } CompareState;
 
 typedef struct CompareClass {
@@ -570,20 +577,37 @@ 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);
+}
 
-    return TRUE;
+static void colo_compare_timer_init(CompareState *s)
+{
+    s->packet_check_timer = aio_timer_new(s->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_timer_del(CompareState *s)
+{
+    if (s->packet_check_timer) {
+        timer_del(s->packet_check_timer);
+        timer_free(s->packet_check_timer);
+        s->packet_check_timer = NULL;
+    }
 }
 
 static void *colo_compare_thread(void *opaque)
 {
     CompareState *s = opaque;
-    GSource *timeout_source;
 
     s->worker_context = g_main_context_new();
 
@@ -594,20 +618,21 @@ static void *colo_compare_thread(void *opaque)
 
     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;
 }
 
+static void colo_compare_iothread(CompareState *s)
+{
+    object_ref(OBJECT(s->iothread));
+    s->ctx = iothread_get_aio_context(s->iothread);
+
+    colo_compare_timer_init(s);
+}
+
 static char *compare_get_pri_indev(Object *obj, Error **errp)
 {
     CompareState *s = COLO_COMPARE(obj);
@@ -714,9 +739,9 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
     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) ||
@@ -757,6 +782,8 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
                        QEMU_THREAD_JOINABLE);
     compare_id++;
 
+    colo_compare_iothread(s);
+
     return;
 }
 
@@ -786,6 +813,8 @@ static void colo_compare_class_init(ObjectClass *oc, void *data)
 
 static void colo_compare_init(Object *obj)
 {
+    CompareState *s = COLO_COMPARE(obj);
+
     object_property_add_str(obj, "primary_in",
                             compare_get_pri_indev, compare_set_pri_indev,
                             NULL);
@@ -795,6 +824,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);
 }
 
 static void colo_compare_finalize(Object *obj)
@@ -806,6 +839,7 @@ static void colo_compare_finalize(Object *obj)
     qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
                              s->worker_context, true);
     qemu_chr_fe_deinit(&s->chr_out);
+    colo_compare_timer_del(s);
 
     g_main_loop_quit(s->compare_loop);
     qemu_thread_join(&s->thread);
@@ -816,6 +850,10 @@ static void colo_compare_finalize(Object *obj)
     g_queue_clear(&s->conn_list);
 
     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

[Qemu-devel] [PATCHv2 02/04] colo-compare: Process pactkets in the IOThread of the primary

[Yong Wang]

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

Process pactkets in the IOThread which arrived over the socket.
we use qio_channel_set_aio_fd_handler to set the handlers on the
IOThread AioContext.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 | 133 ++++++++++++++++++++++++++++++++++++-----------------
 net/colo.h         |   1 +
 2 files changed, 91 insertions(+), 43 deletions(-)

diff --git a/net/colo-compare.c b/net/colo-compare.c
index b0942a4..e3af791 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 "io/channel.h"
 #include "sysemu/iothread.h"
 
 #define TYPE_COLO_COMPARE "colo-compare"
@@ -82,11 +83,6 @@ typedef struct CompareState {
     GQueue conn_list;
     /* hashtable to save connection */
     GHashTable *connection_track_table;
-    /* compare thread, a thread for each NIC */
-    QemuThread thread;
-
-    GMainContext *worker_context;
-    GMainLoop *compare_loop;
 
     /*compare iothread*/
     IOThread *iothread;
@@ -95,6 +91,14 @@ typedef struct CompareState {
     QEMUTimer *packet_check_timer;
 } CompareState;
 
+typedef struct {
+    Chardev parent;
+    QIOChannel *ioc; /*I/O channel */
+} CompareChardev;
+
+#define COMPARE_CHARDEV(obj)                                     \
+    OBJECT_CHECK(CompareChardev, (obj), TYPE_CHARDEV_SOCKET)
+
 typedef struct CompareClass {
     ObjectClass parent_class;
 } CompareClass;
@@ -107,6 +111,12 @@ enum {
 static int compare_chr_send(CharBackend *out,
                             const uint8_t *buf,
                             uint32_t size);
+static void compare_chr_set_aio_fd_handlers(CharBackend *b,
+                                    AioContext *ctx,
+                                    IOCanReadHandler *fd_can_read,
+                                    IOReadHandler *fd_read,
+                                    IOEventHandler *fd_event,
+                                    void *opaque);
 
 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
 {
@@ -534,6 +544,30 @@ err:
     return ret < 0 ? ret : -EIO;
 }
 
+static void compare_chr_read(void *opaque)
+{
+    Chardev *chr = opaque;
+    uint8_t buf[CHR_READ_BUF_LEN];
+    int len, size;
+    int max_size;
+
+    max_size = qemu_chr_be_can_write(chr);
+    if (max_size <= 0) {
+        return;
+    }
+
+    len = sizeof(buf);
+    if (len > max_size) {
+        len = max_size;
+    }
+    size = CHARDEV_GET_CLASS(chr)->chr_sync_read(chr, (void *)buf, len);
+    if (size == 0) {
+        return;
+    } else if (size > 0) {
+        qemu_chr_be_write(chr, buf, size);
+    }
+}
+
 static int compare_chr_can_read(void *opaque)
 {
     return COMPARE_READ_LEN_MAX;
@@ -550,8 +584,8 @@ static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
 
     ret = net_fill_rstate(&s->pri_rs, buf, size);
     if (ret == -1) {
-        qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL,
-                                 NULL, NULL, true);
+        compare_chr_set_aio_fd_handlers(&s->chr_pri_in, s->ctx,
+                                    NULL, NULL, NULL, NULL);
         error_report("colo-compare primary_in error");
     }
 }
@@ -567,8 +601,8 @@ static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 
     ret = net_fill_rstate(&s->sec_rs, buf, size);
     if (ret == -1) {
-        qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL,
-                                 NULL, NULL, true);
+        compare_chr_set_aio_fd_handlers(&s->chr_sec_in, s->ctx,
+                                    NULL, NULL, NULL, NULL);
         error_report("colo-compare secondary_in error");
     }
 }
@@ -605,34 +639,57 @@ static void colo_compare_timer_del(CompareState *s)
     }
 }
 
-static void *colo_compare_thread(void *opaque)
-{
-    CompareState *s = opaque;
-
-    s->worker_context = g_main_context_new();
-
-    qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
-                          compare_pri_chr_in, NULL, s, s->worker_context, true);
-    qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
-                          compare_sec_chr_in, NULL, s, s->worker_context, true);
-
-    s->compare_loop = g_main_loop_new(s->worker_context, FALSE);
-
-    g_main_loop_run(s->compare_loop);
-
-    g_main_loop_unref(s->compare_loop);
-    g_main_context_unref(s->worker_context);
-    return NULL;
-}
 
 static void colo_compare_iothread(CompareState *s)
 {
     object_ref(OBJECT(s->iothread));
     s->ctx = iothread_get_aio_context(s->iothread);
 
+    compare_chr_set_aio_fd_handlers(&s->chr_pri_in, s->ctx,
+                    compare_chr_can_read,
+                    compare_pri_chr_in,
+                    NULL,
+                    s);
+    compare_chr_set_aio_fd_handlers(&s->chr_sec_in, s->ctx,
+                    compare_chr_can_read,
+                    compare_sec_chr_in,
+                    NULL,
+                    s);
+
     colo_compare_timer_init(s);
 }
 
+static void compare_chr_set_aio_fd_handlers(CharBackend *b,
+                                    AioContext *ctx,
+                                    IOCanReadHandler *fd_can_read,
+                                    IOReadHandler *fd_read,
+                                    IOEventHandler *fd_event,
+                                    void *opaque)
+{
+    CompareChardev *s;
+
+    if (!b->chr) {
+        return;
+    }
+    s = COMPARE_CHARDEV(b->chr);
+    if (!s->ioc) {
+        return;
+    }
+
+    b->chr_can_read = fd_can_read;
+    b->chr_read = fd_read;
+    b->chr_event = fd_event;
+    b->opaque = opaque;
+    remove_fd_in_watch(b->chr);
+
+    if (b->chr_read) {
+        qio_channel_set_aio_fd_handler(s->ioc, ctx,
+                                compare_chr_read, NULL, b->chr);
+    } else {
+        qio_channel_set_aio_fd_handler(s->ioc, ctx, NULL, NULL, NULL);
+    }
+}
+
 static char *compare_get_pri_indev(Object *obj, Error **errp)
 {
     CompareState *s = COLO_COMPARE(obj);
@@ -736,8 +793,6 @@ 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 || !s->iothread) {
         error_setg(errp, "colo compare needs 'primary_in' ,"
@@ -776,12 +831,6 @@ 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;
@@ -834,16 +883,14 @@ static void colo_compare_finalize(Object *obj)
 {
     CompareState *s = COLO_COMPARE(obj);
 
-    qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
-                             s->worker_context, true);
-    qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
-                             s->worker_context, true);
+    compare_chr_set_aio_fd_handlers(&s->chr_pri_in, s->ctx,
+                                    NULL, NULL, NULL, NULL);
+    compare_chr_set_aio_fd_handlers(&s->chr_sec_in, s->ctx,
+                                    NULL, NULL, NULL, NULL);
+
     qemu_chr_fe_deinit(&s->chr_out);
     colo_compare_timer_del(s);
 
-    g_main_loop_quit(s->compare_loop);
-    qemu_thread_join(&s->thread);
-
     /* Release all unhandled packets after compare thead exited */
     g_queue_foreach(&s->conn_list, colo_flush_packets, s);
 
diff --git a/net/colo.h b/net/colo.h
index 7c524f3..936dea1 100644
--- a/net/colo.h
+++ b/net/colo.h
@@ -84,5 +84,6 @@ Connection *connection_get(GHashTable *connection_track_table,
 void connection_hashtable_reset(GHashTable *connection_track_table);
 Packet *packet_new(const void *data, int size);
 void packet_destroy(void *opaque, void *user_data);
+void remove_fd_in_watch(Chardev *chr);
 
 #endif /* QEMU_COLO_PROXY_H */
-- 
1.8.3.1

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

[Yong Wang]

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

[Qemu-devel] [PATCHv2 04/04] colo-compare: Update the COLO document to fix the processing of secondary packets in the main thread

[Yong Wang]

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

In my test, secondary does not process the packets comparing in the IOThread
but in the qemu main thread processing.
secondary's configuration " -chardev socket,id=compare1,host=3.3.3.3,port=9004,server,nowait"
here,we should use 'wait' instead of 'nowait' configuration .
after use  'wait' configuration ,packets can be process in the IOThread from the secondary,
not in the qemu main thread.

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

diff --git a/docs/colo-proxy.txt b/docs/colo-proxy.txt
index ce3f783..567cc8b 100644
--- a/docs/colo-proxy.txt
+++ b/docs/colo-proxy.txt
@@ -165,7 +165,7 @@ Primary(ip:3.3.3.3):
 -netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,downscript=/etc/qemu-ifdown
 -device e1000,id=e0,netdev=hn0,mac=52:a4:00:12:78:66
 -chardev socket,id=mirror0,host=3.3.3.3,port=9003,server,nowait
--chardev socket,id=compare1,host=3.3.3.3,port=9004,server,nowait
+-chardev socket,id=compare1,host=3.3.3.3,port=9004,server,wait
 -chardev socket,id=compare0,host=3.3.3.3,port=9001,server,nowait
 -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
-- 
1.8.3.1

Re: [Qemu-devel] [PATCHv2 01/04] colo-compare: Use IOThread context timer to Check old packet regularly

[Jason Wang]

On 2017年06月05日 18:44, Yong Wang 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.
>
> Signed-off-by: Wang Yong<wang.yong155@zte.com.cn>
> Signed-off-by: Wang Guang<wang.guang55@zte.com.cn>
> ---
>   net/colo-compare.c | 62 +++++++++++++++++++++++++++++++++++++++++++-----------
>   1 file changed, 50 insertions(+), 12 deletions(-)

I suggest to squash this into patch 2, since we will have 2 threads now.

Thanks

Re: [Qemu-devel] [PATCHv2 02/04] colo-compare: Process pactkets in the IOThread of the primary

[Jason Wang]

On 2017年06月05日 18:44, Yong Wang wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
>
> Process pactkets in the IOThread which arrived over the socket.
> we use qio_channel_set_aio_fd_handler to set the handlers on the
> IOThread AioContext.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 | 133 ++++++++++++++++++++++++++++++++++++-----------------
>   net/colo.h         |   1 +
>   2 files changed, 91 insertions(+), 43 deletions(-)
>
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index b0942a4..e3af791 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 "io/channel.h"
>   #include "sysemu/iothread.h"
>   
>   #define TYPE_COLO_COMPARE "colo-compare"
> @@ -82,11 +83,6 @@ typedef struct CompareState {
>       GQueue conn_list;
>       /* hashtable to save connection */
>       GHashTable *connection_track_table;
> -    /* compare thread, a thread for each NIC */
> -    QemuThread thread;
> -
> -    GMainContext *worker_context;
> -    GMainLoop *compare_loop;
>   
>       /*compare iothread*/
>       IOThread *iothread;
> @@ -95,6 +91,14 @@ typedef struct CompareState {
>       QEMUTimer *packet_check_timer;
>   } CompareState;
>   
> +typedef struct {
> +    Chardev parent;
> +    QIOChannel *ioc; /*I/O channel */

We probably don't want to manipulate char backend's internal io channel. 
All need here is to access the frontend API (char-fe.c) I believe, and 
hide the internal implementation.

> +} CompareChardev;
> +
> +#define COMPARE_CHARDEV(obj)                                     \
> +    OBJECT_CHECK(CompareChardev, (obj), TYPE_CHARDEV_SOCKET)
> +
>   typedef struct CompareClass {
>       ObjectClass parent_class;
>   } CompareClass;
> @@ -107,6 +111,12 @@ enum {
>   static int compare_chr_send(CharBackend *out,
>                               const uint8_t *buf,
>                               uint32_t size);
> +static void compare_chr_set_aio_fd_handlers(CharBackend *b,
> +                                    AioContext *ctx,
> +                                    IOCanReadHandler *fd_can_read,
> +                                    IOReadHandler *fd_read,
> +                                    IOEventHandler *fd_event,
> +                                    void *opaque);
>   
>   static gint seq_sorter(Packet *a, Packet *b, gpointer data)
>   {
> @@ -534,6 +544,30 @@ err:
>       return ret < 0 ? ret : -EIO;
>   }
>   
> +static void compare_chr_read(void *opaque)
> +{
> +    Chardev *chr = opaque;
> +    uint8_t buf[CHR_READ_BUF_LEN];
> +    int len, size;
> +    int max_size;
> +
> +    max_size = qemu_chr_be_can_write(chr);
> +    if (max_size <= 0) {
> +        return;
> +    }
> +
> +    len = sizeof(buf);
> +    if (len > max_size) {
> +        len = max_size;
> +    }
> +    size = CHARDEV_GET_CLASS(chr)->chr_sync_read(chr, (void *)buf, len);
> +    if (size == 0) {
> +        return;
> +    } else if (size > 0) {
> +        qemu_chr_be_write(chr, buf, size);
> +    }
> +}
> +
>   static int compare_chr_can_read(void *opaque)
>   {
>       return COMPARE_READ_LEN_MAX;
> @@ -550,8 +584,8 @@ static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
>   
>       ret = net_fill_rstate(&s->pri_rs, buf, size);
>       if (ret == -1) {
> -        qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL,
> -                                 NULL, NULL, true);
> +        compare_chr_set_aio_fd_handlers(&s->chr_pri_in, s->ctx,
> +                                    NULL, NULL, NULL, NULL);
>           error_report("colo-compare primary_in error");
>       }
>   }
> @@ -567,8 +601,8 @@ static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
>   
>       ret = net_fill_rstate(&s->sec_rs, buf, size);
>       if (ret == -1) {
> -        qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL,
> -                                 NULL, NULL, true);
> +        compare_chr_set_aio_fd_handlers(&s->chr_sec_in, s->ctx,
> +                                    NULL, NULL, NULL, NULL);
>           error_report("colo-compare secondary_in error");
>       }
>   }
> @@ -605,34 +639,57 @@ static void colo_compare_timer_del(CompareState *s)
>       }
>   }
>   
> -static void *colo_compare_thread(void *opaque)
> -{
> -    CompareState *s = opaque;
> -
> -    s->worker_context = g_main_context_new();
> -
> -    qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
> -                          compare_pri_chr_in, NULL, s, s->worker_context, true);
> -    qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
> -                          compare_sec_chr_in, NULL, s, s->worker_context, true);
> -
> -    s->compare_loop = g_main_loop_new(s->worker_context, FALSE);
> -
> -    g_main_loop_run(s->compare_loop);
> -
> -    g_main_loop_unref(s->compare_loop);
> -    g_main_context_unref(s->worker_context);
> -    return NULL;
> -}
>   
>   static void colo_compare_iothread(CompareState *s)
>   {
>       object_ref(OBJECT(s->iothread));
>       s->ctx = iothread_get_aio_context(s->iothread);
>   
> +    compare_chr_set_aio_fd_handlers(&s->chr_pri_in, s->ctx,
> +                    compare_chr_can_read,
> +                    compare_pri_chr_in,
> +                    NULL,
> +                    s);
> +    compare_chr_set_aio_fd_handlers(&s->chr_sec_in, s->ctx,
> +                    compare_chr_can_read,
> +                    compare_sec_chr_in,
> +                    NULL,
> +                    s);
> +
>       colo_compare_timer_init(s);
>   }
>   
> +static void compare_chr_set_aio_fd_handlers(CharBackend *b,
> +                                    AioContext *ctx,
> +                                    IOCanReadHandler *fd_can_read,
> +                                    IOReadHandler *fd_read,
> +                                    IOEventHandler *fd_event,
> +                                    void *opaque)
> +{
> +    CompareChardev *s;
> +
> +    if (!b->chr) {
> +        return;
> +    }
> +    s = COMPARE_CHARDEV(b->chr);
> +    if (!s->ioc) {
> +        return;
> +    }

So this is hacky, you can refer how vhost-user validate udp socket char 
backend.

> +
> +    b->chr_can_read = fd_can_read;
> +    b->chr_read = fd_read;
> +    b->chr_event = fd_event;
> +    b->opaque = opaque;
> +    remove_fd_in_watch(b->chr);
> +
> +    if (b->chr_read) {
> +        qio_channel_set_aio_fd_handler(s->ioc, ctx,
> +                                compare_chr_read, NULL, b->chr);
> +    } else {
> +        qio_channel_set_aio_fd_handler(s->ioc, ctx, NULL, NULL, NULL);

So instead of doing such hack, how about passing a AioContext * instead 
of GMainContext * to qemu_chr_fe_set_handlers?

Thanks

> +    }
> +}
> +
>   static char *compare_get_pri_indev(Object *obj, Error **errp)
>   {
>       CompareState *s = COLO_COMPARE(obj);
> @@ -736,8 +793,6 @@ 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 || !s->iothread) {
>           error_setg(errp, "colo compare needs 'primary_in' ,"
> @@ -776,12 +831,6 @@ 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;
> @@ -834,16 +883,14 @@ static void colo_compare_finalize(Object *obj)
>   {
>       CompareState *s = COLO_COMPARE(obj);
>   
> -    qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
> -                             s->worker_context, true);
> -    qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
> -                             s->worker_context, true);
> +    compare_chr_set_aio_fd_handlers(&s->chr_pri_in, s->ctx,
> +                                    NULL, NULL, NULL, NULL);
> +    compare_chr_set_aio_fd_handlers(&s->chr_sec_in, s->ctx,
> +                                    NULL, NULL, NULL, NULL);
> +
>       qemu_chr_fe_deinit(&s->chr_out);
>       colo_compare_timer_del(s);
>   
> -    g_main_loop_quit(s->compare_loop);
> -    qemu_thread_join(&s->thread);
> -
>       /* Release all unhandled packets after compare thead exited */
>       g_queue_foreach(&s->conn_list, colo_flush_packets, s);
>   
> diff --git a/net/colo.h b/net/colo.h
> index 7c524f3..936dea1 100644
> --- a/net/colo.h
> +++ b/net/colo.h
> @@ -84,5 +84,6 @@ Connection *connection_get(GHashTable *connection_track_table,
>   void connection_hashtable_reset(GHashTable *connection_track_table);
>   Packet *packet_new(const void *data, int size);
>   void packet_destroy(void *opaque, void *user_data);
> +void remove_fd_in_watch(Chardev *chr);
>   
>   #endif /* QEMU_COLO_PROXY_H */

Re: [Qemu-devel] [PATCHv2 04/04] colo-compare: Update the COLO document to fix the processing of secondary packets in the main thread

[Jason Wang]

On 2017年06月05日 18:44, Yong Wang wrote:
> From: Wang Yong <wang.yong155@zte.com.cn>
>
> In my test, secondary does not process the packets comparing in the IOThread
> but in the qemu main thread processing.
> secondary's configuration " -chardev socket,id=compare1,host=3.3.3.3,port=9004,server,nowait"
> here,we should use 'wait' instead of 'nowait' configuration .
> after use  'wait' configuration ,packets can be process in the IOThread from the secondary,
> not in the qemu main thread.
>
> Signed-off-by: Wang Yong<wang.yong155@zte.com.cn>
> Signed-off-by: Wang Guang<wang.guang55@zte.com.cn>
> ---

This is a hint probably for something wrong in the code. Need to figure 
out the root cause.

Thanks

>   docs/colo-proxy.txt | 2 +-
>   1 file changed, 1 insertion(+), 1 deletion(-)
>
> diff --git a/docs/colo-proxy.txt b/docs/colo-proxy.txt
> index ce3f783..567cc8b 100644
> --- a/docs/colo-proxy.txt
> +++ b/docs/colo-proxy.txt
> @@ -165,7 +165,7 @@ Primary(ip:3.3.3.3):
>   -netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,downscript=/etc/qemu-ifdown
>   -device e1000,id=e0,netdev=hn0,mac=52:a4:00:12:78:66
>   -chardev socket,id=mirror0,host=3.3.3.3,port=9003,server,nowait
> --chardev socket,id=compare1,host=3.3.3.3,port=9004,server,nowait
> +-chardev socket,id=compare1,host=3.3.3.3,port=9004,server,wait
>   -chardev socket,id=compare0,host=3.3.3.3,port=9001,server,nowait
>   -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