From mboxrd@z Thu Jan 1 00:00:00 1970 Received: from eggs.gnu.org ([2001:4830:134:3::10]:53563) by lists.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1bLNKe-0000hs-OT for qemu-devel@nongnu.org; Fri, 08 Jul 2016 00:24:06 -0400 Received: from Debian-exim by eggs.gnu.org with spam-scanned (Exim 4.71) (envelope-from ) id 1bLNKZ-0004WU-2y for qemu-devel@nongnu.org; Fri, 08 Jul 2016 00:24:03 -0400 Received: from mx1.redhat.com ([209.132.183.28]:38149) by eggs.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1bLNKY-0004WP-QZ for qemu-devel@nongnu.org; Fri, 08 Jul 2016 00:23:59 -0400 References: <1466681677-30487-1-git-send-email-zhangchen.fnst@cn.fujitsu.com> <1466681677-30487-4-git-send-email-zhangchen.fnst@cn.fujitsu.com> From: Jason Wang Message-ID: <577F2AD6.40004@redhat.com> Date: Fri, 8 Jul 2016 12:23:50 +0800 MIME-Version: 1.0 In-Reply-To: <1466681677-30487-4-git-send-email-zhangchen.fnst@cn.fujitsu.com> Content-Type: text/plain; charset=UTF-8; format=flowed Content-Transfer-Encoding: quoted-printable Subject: Re: [Qemu-devel] [RFC PATCH V5 3/4] colo-compare: introduce packet comparison thread List-Id: List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: Zhang Chen , qemu devel Cc: Li Zhijian , Wen Congyang , zhanghailiang , "eddie . dong" , "Dr . David Alan Gilbert" On 2016=E5=B9=B406=E6=9C=8823=E6=97=A5 19:34, Zhang Chen wrote: > if packets are same, we send primary packet and drop secondary > packet, otherwise notify COLO do checkpoint. More verbose please, e.g how to handle each case of exception (or maybe=20 comment in the code). > > Signed-off-by: Zhang Chen > Signed-off-by: Li Zhijian > Signed-off-by: Wen Congyang > --- > net/colo-base.c | 1 + > net/colo-base.h | 3 + > net/colo-compare.c | 214 ++++++++++++++++++++++++++++++++++++++++++++= +++++++++ > trace-events | 2 + > 4 files changed, 220 insertions(+) > > diff --git a/net/colo-base.c b/net/colo-base.c > index 7e263e8..9673661 100644 > --- a/net/colo-base.c > +++ b/net/colo-base.c > @@ -146,6 +146,7 @@ Packet *packet_new(const void *data, int size) > =20 > pkt->data =3D g_memdup(data, size); > pkt->size =3D size; > + pkt->creation_ms =3D qemu_clock_get_ms(QEMU_CLOCK_HOST); > =20 > return pkt; > } > diff --git a/net/colo-base.h b/net/colo-base.h > index 01c1a5d..8bb1043 100644 > --- a/net/colo-base.h > +++ b/net/colo-base.h > @@ -18,6 +18,7 @@ > #include "slirp/slirp.h" > #include "qemu/jhash.h" > #include "qemu/rcu.h" > +#include "qemu/timer.h" > =20 > #define HASHTABLE_MAX_SIZE 16384 > =20 > @@ -47,6 +48,8 @@ typedef struct Packet { > }; > uint8_t *transport_layer; > int size; > + /* Time of packet creation, in wall clock ms */ > + int64_t creation_ms; > } Packet; > =20 > typedef struct ConnectionKey { > diff --git a/net/colo-compare.c b/net/colo-compare.c > index 4231fe7..928d729 100644 > --- a/net/colo-compare.c > +++ b/net/colo-compare.c > @@ -35,6 +35,8 @@ > OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE) > =20 > #define COMPARE_READ_LEN_MAX NET_BUFSIZE > +/* TODO: Should be configurable */ > +#define REGULAR_CHECK_MS 400 "REGULAR" seems to generic, need a better name. > =20 > static QTAILQ_HEAD(, CompareState) net_compares =3D > QTAILQ_HEAD_INITIALIZER(net_compares); > @@ -86,6 +88,11 @@ typedef struct CompareState { > GQueue unprocessed_connections; > /* proxy current hash size */ > uint32_t hashtable_size; > + /* compare thread, a thread for each NIC */ > + QemuThread thread; > + int thread_status; > + /* Timer used on the primary to find packets that are never matche= d */ > + QEMUTimer *timer; > } CompareState; > =20 > typedef struct CompareClass { > @@ -97,6 +104,15 @@ enum { > SECONDARY_IN, > }; > =20 > +enum { > + /* compare thread isn't started */ > + COMPARE_THREAD_NONE, > + /* compare thread is running */ > + COMPARE_THREAD_RUNNING, > + /* compare thread exit */ > + COMPARE_THREAD_EXIT, > +}; > + > static int compare_chr_send(CharDriverState *out, > const uint8_t *buf, > uint32_t size); > @@ -143,6 +159,98 @@ static int packet_enqueue(CompareState *s, int mod= e) > return 0; > } > =20 > +/* > + * The IP packets sent by primary and secondary > + * will be compared in here > + * TODO support ip fragment, Out-Of-Order > + * return: 0 means packet same > + * > 0 || < 0 means packet different > + */ > +static int colo_packet_compare(Packet *ppkt, Packet *spkt) > +{ > + trace_colo_compare_ip_info(ppkt->size, inet_ntoa(ppkt->ip->ip_src)= , > + inet_ntoa(ppkt->ip->ip_dst), spkt->size= , > + inet_ntoa(spkt->ip->ip_src), > + inet_ntoa(spkt->ip->ip_dst)); > + > + if (ppkt->size =3D=3D spkt->size) { > + return memcmp(ppkt->data, spkt->data, spkt->size); > + } else { > + return -1; > + } > +} > + > +static int colo_packet_compare_all(Packet *spkt, Packet *ppkt) > +{ > + trace_colo_compare_main("compare all"); > + return colo_packet_compare(ppkt, spkt); > +} > + > +static void colo_old_packet_check(void *opaque_packet, void *opaque_fo= und) > +{ > + int64_t now; > + bool *found_old =3D (bool *)opaque_found; > + Packet *ppkt =3D (Packet *)opaque_packet; > + > + if (*found_old) { > + /* Someone found an old packet earlier in the queue */ > + return; > + } > + > + now =3D qemu_clock_get_ms(QEMU_CLOCK_HOST); > + if ((ppkt->creation_ms < now) && Any case that ppkt->creation_ms >=3D now? > + ((now - ppkt->creation_ms) > REGULAR_CHECK_MS)) { > + trace_colo_old_packet_check_found(ppkt->creation_ms); > + *found_old =3D true; > + } > +} > + > +/* > + * called from the compare thread on the primary > + * for compare connection > + */ > +static void colo_compare_connection(void *opaque, void *user_data) > +{ > + CompareState *s =3D user_data; > + Connection *conn =3D opaque; > + Packet *pkt =3D NULL; > + GList *result =3D NULL; > + bool found_old; > + int ret; > + > + while (!g_queue_is_empty(&conn->primary_list) && > + !g_queue_is_empty(&conn->secondary_list)) { > + pkt =3D g_queue_pop_tail(&conn->primary_list); > + result =3D g_queue_find_custom(&conn->secondary_list, > + pkt, (GCompareFunc)colo_packet_compare_a= ll); > + > + if (result) { > + ret =3D compare_chr_send(s->chr_out, pkt->data, pkt->size)= ; > + if (ret < 0) { > + error_report("colo_send_primary_packet failed"); > + } > + trace_colo_compare_main("packet same and release packet"); > + g_queue_remove(&conn->secondary_list, result->data); > + } else { A question I forget the answer, so may ask again. What if secondary=20 packet comes late? > + trace_colo_compare_main("packet different"); > + g_queue_push_tail(&conn->primary_list, pkt); > + /* TODO: colo_notify_checkpoint();*/ > + break; > + } > + } > + > + /* > + * Look for old packets that the secondary hasn't matched, > + * if we have some then we have to checkpoint to wake > + * the secondary up. > + */ > + found_old =3D false; > + g_queue_foreach(&conn->primary_list, colo_old_packet_check, &found= _old); > + if (found_old) { > + /* TODO: colo_notify_checkpoint();*/ Shouldn't we need to remove all "old" packets here? > + } > +} > + > static int compare_chr_send(CharDriverState *out, > const uint8_t *buf, > uint32_t size) > @@ -170,6 +278,69 @@ err: > return ret < 0 ? ret : -EIO; > } > =20 > +static int compare_chr_can_read(void *opaque) > +{ > + return COMPARE_READ_LEN_MAX; > +} > + > +/* > + * called from the main thread on the primary for packets > + * arriving over the socket from the primary. > + */ > +static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int s= ize) > +{ > + CompareState *s =3D COLO_COMPARE(opaque); > + int ret; > + > + ret =3D net_fill_rstate(&s->pri_rs, buf, size); > + if (ret =3D=3D -1) { > + qemu_chr_add_handlers(s->chr_pri_in, NULL, NULL, NULL, NULL); > + error_report("colo-compare primary_in error"); > + } > +} > + > +/* > + * called from the main thread on the primary for packets > + * arriving over the socket from the secondary. > + */ > +static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int s= ize) > +{ > + CompareState *s =3D COLO_COMPARE(opaque); > + int ret; > + > + ret =3D net_fill_rstate(&s->sec_rs, buf, size); > + if (ret =3D=3D -1) { > + qemu_chr_add_handlers(s->chr_sec_in, NULL, NULL, NULL, NULL); > + error_report("colo-compare secondary_in error"); > + } > +} > + > +static void *colo_compare_thread(void *opaque) > +{ > + GMainContext *worker_context; > + GMainLoop *compare_loop; > + CompareState *s =3D opaque; > + > + worker_context =3D g_main_context_new(); > + g_assert(g_main_context_get_thread_default() =3D=3D NULL); > + g_main_context_push_thread_default(worker_context); > + g_assert(g_main_context_get_thread_default() =3D=3D worker_context= ); > + > + qemu_chr_add_handlers(s->chr_pri_in, compare_chr_can_read, > + compare_pri_chr_in, NULL, s); > + qemu_chr_add_handlers(s->chr_sec_in, compare_chr_can_read, > + compare_sec_chr_in, NULL, s); > + > + compare_loop =3D g_main_loop_new(worker_context, FALSE); > + > + g_main_loop_run(compare_loop); > + > + g_main_loop_unref(compare_loop); > + g_main_context_pop_thread_default(worker_context); > + g_main_context_unref(worker_context); > + return NULL; > +} > + > static char *compare_get_pri_indev(Object *obj, Error **errp) > { > CompareState *s =3D COLO_COMPARE(obj); > @@ -222,6 +393,9 @@ static void compare_pri_rs_finalize(SocketReadState= *pri_rs) > if (packet_enqueue(s, PRIMARY_IN)) { > trace_colo_compare_main("primary: unsupported packet in"); > compare_chr_send(s->chr_out, pri_rs->buf, pri_rs->packet_len)= ; > + } else { > + /* compare connection */ > + g_queue_foreach(&s->conn_list, colo_compare_connection, s); > } > } > =20 > @@ -231,16 +405,35 @@ static void compare_sec_rs_finalize(SocketReadSta= te *sec_rs) > =20 > if (packet_enqueue(s, SECONDARY_IN)) { > trace_colo_compare_main("secondary: unsupported packet in"); > + } else { > + /* compare connection */ > + g_queue_foreach(&s->conn_list, colo_compare_connection, s); > } > } > =20 > /* > + * Prod the compare thread regularly so it can watch for any packets > + * that the secondary hasn't produced equivalents of. > + */ > +static void colo_compare_regular(void *opaque) > +{ > + CompareState *s =3D opaque; > + > + timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) + > + REGULAR_CHECK_MS); > + /* compare connection */ > + g_queue_foreach(&s->conn_list, colo_compare_connection, s); > +} We need make sure this function was called from colo thread, but it=20 looks not? > + > +/* > * called from the main thread on the primary > * to setup colo-compare. > */ > static void colo_compare_complete(UserCreatable *uc, Error **errp) > { > CompareState *s =3D COLO_COMPARE(uc); > + char thread_name[64]; > + static int compare_id; > =20 > if (!s->pri_indev || !s->sec_indev || !s->outdev) { > error_setg(errp, "colo compare needs 'primary_in' ," > @@ -293,6 +486,19 @@ static void colo_compare_complete(UserCreatable *u= c, Error **errp) > g_free, > connection_dest= roy); > =20 > + s->thread_status =3D COMPARE_THREAD_RUNNING; > + sprintf(thread_name, "compare %d", compare_id); > + qemu_thread_create(&s->thread, thread_name, > + colo_compare_thread, s, > + QEMU_THREAD_JOINABLE); > + compare_id++; > + > + /* A regular timer to kick any packets that the secondary doesn't = match */ > + s->timer =3D timer_new_ms(QEMU_CLOCK_VIRTUAL, /* Only when guest r= uns */ > + colo_compare_regular, s); > + timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) + > + REGULAR_CHECK_MS); > + > return; > } > =20 > @@ -338,6 +544,14 @@ static void colo_compare_finalize(Object *obj) > qemu_mutex_destroy(&s->conn_list_lock); > g_queue_free(&s->conn_list); > =20 > + if (s->thread.thread) { > + s->thread_status =3D COMPARE_THREAD_EXIT; Looks like there's not any code that depends on the status, so why need=20 to this> > + /* compare connection */ > + g_queue_foreach(&s->conn_list, colo_compare_connection, s); > + qemu_thread_join(&s->thread); > + } > + timer_del(s->timer); > + > g_free(s->pri_indev); > g_free(s->sec_indev); > g_free(s->outdev); > diff --git a/trace-events b/trace-events > index 703de1a..1537e91 100644 > --- a/trace-events > +++ b/trace-events > @@ -1919,3 +1919,5 @@ aspeed_vic_write(uint64_t offset, unsigned size, = uint32_t data) "To 0x%" PRIx64 > =20 > # net/colo-compare.c > colo_compare_main(const char *chr) ": %s" > +colo_compare_ip_info(int psize, const char *sta, const char *stb, int = ssize, const char *stc, const char *std) "ppkt size =3D %d, ip_src =3D %s= , ip_dst =3D %s, spkt size =3D %d, ip_src =3D %s, ip_dst =3D %s" > +colo_old_packet_check_found(int64_t old_time) "%" PRId64