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

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 */