From mboxrd@z Thu Jan 1 00:00:00 1970 Received: from eggs.gnu.org ([2001:4830:134:3::10]:56093) by lists.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1bVZpl-0008AS-7c for qemu-devel@nongnu.org; Fri, 05 Aug 2016 03:46:22 -0400 Received: from Debian-exim by eggs.gnu.org with spam-scanned (Exim 4.71) (envelope-from ) id 1bVZpg-0000rY-Ty for qemu-devel@nongnu.org; Fri, 05 Aug 2016 03:46:20 -0400 Received: from [59.151.112.132] (port=51090 helo=heian.cn.fujitsu.com) by eggs.gnu.org with esmtp (Exim 4.71) (envelope-from ) id 1bVZpd-0000nJ-Q1 for qemu-devel@nongnu.org; Fri, 05 Aug 2016 03:46:16 -0400 References: <1469497794-16976-1-git-send-email-zhangchen.fnst@cn.fujitsu.com> <1469497794-16976-7-git-send-email-zhangchen.fnst@cn.fujitsu.com> From: Zhang Chen Message-ID: <1e1a4127-92e0-a6ce-c920-32f7d0bfca54@cn.fujitsu.com> Date: Fri, 5 Aug 2016 15:45:34 +0800 MIME-Version: 1.0 In-Reply-To: Content-Type: text/plain; charset="utf-8"; format=flowed Content-Transfer-Encoding: 8bit Subject: Re: [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread List-Id: List-Unsubscribe: , List-Archive: List-Post: List-Help: List-Subscribe: , To: Jason Wang , qemu devel Cc: Li Zhijian , Wen Congyang , zhanghailiang , "eddie . dong" , "Dr . David Alan Gilbert" On 08/02/2016 03:52 PM, Jason Wang wrote: > > > On 2016年07月26日 09:49, 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 and secondary packet not, > > s/and/but/ and /packet not/packet does not/ I will fix it. > >> 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-base.c | 1 + >> net/colo-base.h | 3 + >> net/colo-compare.c | 212 >> +++++++++++++++++++++++++++++++++++++++++++++++++++++ >> trace-events | 2 + >> 4 files changed, 218 insertions(+) >> >> diff --git a/net/colo-base.c b/net/colo-base.c >> index 7e91dec..eb1b631 100644 >> --- a/net/colo-base.c >> +++ b/net/colo-base.c >> @@ -132,6 +132,7 @@ Packet *packet_new(const void *data, int size) >> pkt->data = g_memdup(data, size); >> pkt->size = size; >> + pkt->creation_ms = qemu_clock_get_ms(QEMU_CLOCK_HOST); >> return pkt; >> } >> diff --git a/net/colo-base.h b/net/colo-base.h >> index 0505608..06d6dca 100644 >> --- a/net/colo-base.h >> +++ b/net/colo-base.h >> @@ -17,6 +17,7 @@ >> #include "slirp/slirp.h" >> #include "qemu/jhash.h" >> +#include "qemu/timer.h" >> #define HASHTABLE_MAX_SIZE 16384 >> @@ -28,6 +29,8 @@ typedef struct Packet { >> }; >> uint8_t *transport_layer; >> int size; >> + /* Time of packet creation, in wall clock ms */ >> + int64_t creation_ms; >> } Packet; >> typedef struct ConnectionKey { >> diff --git a/net/colo-compare.c b/net/colo-compare.c >> index 5f87710..e020edc 100644 >> --- a/net/colo-compare.c >> +++ b/net/colo-compare.c >> @@ -36,6 +36,8 @@ >> #define COMPARE_READ_LEN_MAX NET_BUFSIZE >> #define MAX_QUEUE_SIZE 1024 >> +/* TODO: Should be configurable */ >> +#define REGULAR_PACKET_CHECK_MS 3000 >> /* >> + CompareState ++ >> @@ -83,6 +85,10 @@ typedef struct CompareState { >> GQueue unprocessed_connections; >> /* proxy current hash size */ >> uint32_t hashtable_size; >> + /* compare thread, a thread for each NIC */ >> + QemuThread thread; >> + /* Timer used on the primary to find packets that are never >> matched */ >> + QEMUTimer *timer; >> } CompareState; >> typedef struct CompareClass { >> @@ -170,6 +176,112 @@ static int packet_enqueue(CompareState *s, int >> mode) >> return 0; >> } >> +/* >> + * 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 == 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 = (bool *)opaque_found; >> + Packet *ppkt = (Packet *)opaque_packet; >> + >> + if (*found_old) { >> + /* Someone found an old packet earlier in the queue */ >> + return; >> + } >> + >> + now = 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 = true; >> + } >> +} >> + >> +static void colo_old_packet_check_one_conn(void *opaque, >> + void *user_data) >> +{ > > user_data is unused. Yes,g_queue_foreach() need this. void (*GFunc) (gpointer data, gpointer user_data); > >> + bool found_old = false; >> + Connection *conn = opaque; >> + >> + g_queue_foreach(&conn->primary_list, colo_old_packet_check_one, >> + &found_old); > > To avoid odd API for colo_old_packet_check_one, maybe you can try use > use QTAILQ and QTAILQ_FOREACH() or g_queue_find_custom() for avoiding > iterating each element? I will consider it, but I think glib are more general. > >> + 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 = opaque; >> + >> + g_queue_foreach(&s->conn_list, colo_old_packet_check_one_conn, >> NULL); >> +} >> + >> +/* >> + * called from the compare thread on the primary >> + * for compare connection >> + */ >> +static void colo_compare_connection(void *opaque, void *user_data) >> +{ >> + CompareState *s = user_data; >> + Connection *conn = opaque; >> + Packet *pkt = NULL; >> + GList *result = NULL; >> + int ret; >> + >> + while (!g_queue_is_empty(&conn->primary_list) && >> + !g_queue_is_empty(&conn->secondary_list)) { >> + pkt = g_queue_pop_tail(&conn->primary_list); >> + result = g_queue_find_custom(&conn->secondary_list, >> + pkt, >> (GCompareFunc)colo_packet_compare_all); >> + >> + if (result) { >> + ret = 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); > > Both pkt and result were leaked here? I will fix it in next version. > >> + } else { >> + trace_colo_compare_main("packet different"); >> + g_queue_push_tail(&conn->primary_list, pkt); >> + /* TODO: colo_notify_checkpoint();*/ > > Shouldn't we wait for a while consider packet may arrive late? If one packet arrive late, the secondary_list will be empty, so we can't compare it until next compare. > >> + break; >> + } >> + } >> +} >> + >> static int compare_chr_send(CharDriverState *out, >> const uint8_t *buf, >> uint32_t size) >> @@ -197,6 +309,65 @@ err: >> return ret < 0 ? ret : -EIO; >> } >> +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 >> size) >> +{ >> + CompareState *s = COLO_COMPARE(opaque); >> + int ret; >> + >> + ret = net_fill_rstate(&s->pri_rs, buf, size); >> + if (ret == -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 >> size) >> +{ >> + CompareState *s = COLO_COMPARE(opaque); >> + int ret; >> + >> + ret = net_fill_rstate(&s->sec_rs, buf, size); >> + if (ret == -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 = opaque; >> + >> + worker_context = 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 = 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 = COLO_COMPARE(obj); >> @@ -249,6 +420,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); >> } >> } >> @@ -258,16 +432,35 @@ static void >> compare_sec_rs_finalize(SocketReadState *sec_rs) >> 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); >> } >> } >> /* >> + * 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 = 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. >> */ >> static void colo_compare_complete(UserCreatable *uc, Error **errp) >> { >> CompareState *s = COLO_COMPARE(uc); >> + char thread_name[64]; >> + static int compare_id; >> if (!s->pri_indev || !s->sec_indev || !s->outdev) { >> error_setg(errp, "colo compare needs 'primary_in' ," >> @@ -319,6 +512,18 @@ static void colo_compare_complete(UserCreatable >> *uc, Error **errp) >> g_free, >> connection_destroy); >> + sprintf(thread_name, "compare %d", compare_id); > > colo-compare? OK, will rename it. > >> + 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 = timer_new_ms(QEMU_CLOCK_VIRTUAL, /* Only when guest >> runs */ >> + check_old_packet_regular, s); >> + timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) + >> + REGULAR_PACKET_CHECK_MS); > > I believe we need make sure this timer were processed in colo > comparing thread. But it looks not. > Yes, this timer just check whether have old packet and can't trigger colo_compare_connection() now. So, I think we needn't make it run in colo comparing thread. Thanks Zhang Chen >> + >> return; >> } >> @@ -360,6 +565,13 @@ static void colo_compare_finalize(Object *obj) >> g_queue_free(&s->conn_list); >> + if (s->thread.thread) { >> + /* 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 >> # 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 = %d, ip_src >> = %s, ip_dst = %s, spkt size = %d, ip_src = %s, ip_dst = %s" >> +colo_old_packet_check_found(int64_t old_time) "%" PRId64 > > > > . > -- Thanks zhangchen