From mboxrd@z Thu Jan 1 00:00:00 1970 Received: from eggs.gnu.org ([2001:4830:134:3::10]:36808) by lists.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1bf1aQ-0000uX-9o for qemu-devel@nongnu.org; Wed, 31 Aug 2016 05:13:36 -0400 Received: from Debian-exim by eggs.gnu.org with spam-scanned (Exim 4.71) (envelope-from ) id 1bf1aK-0007TZ-IX for qemu-devel@nongnu.org; Wed, 31 Aug 2016 05:13:33 -0400 Received: from mx1.redhat.com ([209.132.183.28]:54880) by eggs.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1bf1aK-0007TH-9q for qemu-devel@nongnu.org; Wed, 31 Aug 2016 05:13:28 -0400 References: <1471421428-26379-1-git-send-email-zhangchen.fnst@cn.fujitsu.com> <1471421428-26379-7-git-send-email-zhangchen.fnst@cn.fujitsu.com> From: Jason Wang Message-ID: <31cb7123-0b74-df62-d269-a06b4fd7c0b3@redhat.com> Date: Wed, 31 Aug 2016 17:13:19 +0800 MIME-Version: 1.0 In-Reply-To: <1471421428-26379-7-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] [PATCH V12 06/10] 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=B408=E6=9C=8817=E6=97=A5 16:10, Zhang Chen wrote: > If primary packet is same with secondary packet, > we will send primary packet and drop secondary > packet, otherwise notify COLO frame to do checkpoint. > If primary packet comes but secondary packet does not, > after REGULAR_PACKET_CHECK_MS milliseconds we set > the primary packet as old_packet,then do a checkpoint. > > Signed-off-by: Zhang Chen > Signed-off-by: Li Zhijian > Signed-off-by: Wen Congyang > --- > net/colo-compare.c | 216 ++++++++++++++++++++++++++++++++++++++++++++= +++++++++ > net/colo.c | 1 + > net/colo.h | 3 + > trace-events | 2 + > 4 files changed, 222 insertions(+) > > diff --git a/net/colo-compare.c b/net/colo-compare.c > index bab215b..b90cf1f 100644 > --- a/net/colo-compare.c > +++ b/net/colo-compare.c > @@ -36,6 +36,8 @@ > =20 > #define COMPARE_READ_LEN_MAX NET_BUFSIZE > #define MAX_QUEUE_SIZE 1024 > +/* TODO: Should be configurable */ > +#define REGULAR_PACKET_CHECK_MS 3000 > =20 > /* > + CompareState ++ > @@ -79,6 +81,10 @@ typedef struct CompareState { > GQueue conn_list; > /* hashtable to save connection */ > GHashTable *connection_track_table; > + /* compare thread, a thread for each NIC */ > + QemuThread thread; > + /* Timer used on the primary to find packets that are never matche= d */ > + QEMUTimer *timer; > } CompareState; > =20 > typedef struct CompareClass { > @@ -152,6 +158,113 @@ static int packet_enqueue(CompareState *s, int mo= de) > 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_one(void *opaque_packet, > + void *opaque_found) > +{ > + 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 ((now - ppkt->creation_ms) > REGULAR_PACKET_CHECK_MS) { > + trace_colo_old_packet_check_found(ppkt->creation_ms); > + *found_old =3D true; > + } > +} > + > +static void colo_old_packet_check_one_conn(void *opaque, > + void *user_data) > +{ > + bool found_old =3D false; > + Connection *conn =3D opaque; > + > + g_queue_foreach(&conn->primary_list, colo_old_packet_check_one, > + &found_old); As I mentioned in last version, can we avoid iterating all packets by=20 using g_queue_find_custom() here? > + if (found_old) { > + /* do checkpoint will flush old packet */ > + /* TODO: colo_notify_checkpoint();*/ > + } > +} > + > +/* > + * Look for old packets that the secondary hasn't matched, > + * if we have some then we have to checkpoint to wake > + * the secondary up. > + */ > +static void colo_old_packet_check(void *opaque) > +{ > + CompareState *s =3D opaque; > + > + g_queue_foreach(&s->conn_list, colo_old_packet_check_one_conn, NUL= L); > +} > + > +/* > + * 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; > + 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); > + packet_destroy(pkt, NULL); > + } else { Better add a comment to explain the case when secondary packet comes a=20 little bit late here. > + trace_colo_compare_main("packet different"); > + g_queue_push_tail(&conn->primary_list, pkt); > + /* TODO: colo_notify_checkpoint();*/ > + break; > + } > + } > +} > + > static int compare_chr_send(CharDriverState *out, > const uint8_t *buf, > uint32_t size) > @@ -179,6 +292,65 @@ 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(); > + > + qemu_chr_add_handlers_full(s->chr_pri_in, compare_chr_can_read, > + compare_pri_chr_in, NULL, s, worker_context)= ; > + qemu_chr_add_handlers_full(s->chr_sec_in, compare_chr_can_read, > + compare_sec_chr_in, NULL, s, worker_context)= ; > + > + 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_unref(worker_context); > + return NULL; > +} > + > static char *compare_get_pri_indev(Object *obj, Error **errp) > { > CompareState *s =3D COLO_COMPARE(obj); > @@ -231,6 +403,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 > @@ -240,6 +415,9 @@ static void compare_sec_rs_finalize(SocketReadState= *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 > @@ -266,6 +444,20 @@ static int compare_chardev_opts(void *opaque, > } > =20 > /* > + * Check old packet regularly so it can watch for any packets > + * that the secondary hasn't produced equivalents of. > + */ > +static void check_old_packet_regular(void *opaque) > +{ > + CompareState *s =3D opaque; > + > + timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) + > + REGULAR_PACKET_CHECK_MS); > + /* if have old packet we will notify checkpoint */ > + colo_old_packet_check(s); > +} > + > +/* > * called from the main thread on the primary > * to setup colo-compare. > */ > @@ -273,6 +465,8 @@ static void colo_compare_complete(UserCreatable *uc= , Error **errp) > { > CompareState *s =3D COLO_COMPARE(uc); > CompareChardevProps props; > + 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' ," > @@ -356,6 +550,18 @@ static void colo_compare_complete(UserCreatable *u= c, Error **errp) > g_free, > connection_dest= roy); > =20 > + sprintf(thread_name, "colo-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 */ > + check_old_packet_regular, s); > + timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) + > + REGULAR_PACKET_CHECK_MS); I still think we need to make sure the timer were processed in colo=20 thread. Since check_old_packet_regular may iterate conn_list which may=20 be modified by colo thread at the same time. > + > return; > } > =20 > @@ -397,6 +603,16 @@ static void colo_compare_finalize(Object *obj) > =20 > g_queue_free(&s->conn_list); > =20 > + if (s->thread.thread) { > + /* compare connection */ > + g_queue_foreach(&s->conn_list, colo_compare_connection, s); > + qemu_thread_join(&s->thread); > + } > + > + if (s->timer) { > + timer_del(s->timer); > + } > + > g_free(s->pri_indev); > g_free(s->sec_indev); > g_free(s->outdev); > diff --git a/net/colo.c b/net/colo.c > index bc86553..da4b771 100644 > --- a/net/colo.c > +++ b/net/colo.c > @@ -128,6 +128,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.h b/net/colo.h > index 9cbc14e..6b395a3 100644 > --- a/net/colo.h > +++ b/net/colo.h > @@ -17,6 +17,7 @@ > =20 > #include "slirp/slirp.h" > #include "qemu/jhash.h" > +#include "qemu/timer.h" > =20 > #define HASHTABLE_MAX_SIZE 16384 > =20 > @@ -28,6 +29,8 @@ typedef struct Packet { > }; > uint8_t *transport_header; > int size; > + /* Time of packet creation, in wall clock ms */ > + int64_t creation_ms; > } Packet; > =20 > typedef struct ConnectionKey { > 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