* [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare
@ 2016-07-26 1:49 Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization Zhang Chen
` (6 more replies)
0 siblings, 7 replies; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
COLO-compare is a part of COLO project. It is used
to compare the network package to help COLO decide
whether to do checkpoint.
The full version in this github:
https://github.com/zhangckid/qemu/tree/colo-v2.7-proxy-mode-compare-with-colo-base-jul22
v10:
- fix typo
- Should we make patch 5 independent with this series?
This patch just add a API for qemu-char.
v9:
p5:
- use chr_update_read_handler_full() replace
the chr_update_read_handler()
- use io_watch_poll_prepare_full() replace
the io_watch_poll_prepare()
- use io_watch_poll_funcs_full replace
the io_watch_poll_funcs
- avoid code duplication
v8:
p5:
- add new patch:
qemu-char: Add qemu_chr_add_handlers_full() for GMaincontext
v7:
p5:
- add [PATCH]qemu-char: Fix context for g_source_attach()
in this patch series.
v6:
p6:
- add more commit log.
- fix icmp comparison to compare all packet.
p5:
- add more cpmments in commit log.
- change REGULAR_CHECK_MS to REGULAR_PACKET_CHECK_MS
- make check old packet independent to compare thread
- remove thread_status
p4:
- change this patch only about
Connection and ConnectionKey.
- add some comments in commit log.
- remove mode in fill_connection_key().
- fix some comments and bug.
- move colo_conn_state to patch of
"work with colo-frame"
- remove conn_list_lock.
- add MAX_QUEUE_SIZE, if primary_list or
secondary_list biger than MAX_QUEUE_SIZE
we will drop packet.
p3:
- add new independent kernel jhash patch.
p2:
- add new independent colo-base patch.
p1:
- add a ascii figure and some comments to explain it
- move trace.h to p2
- move QTAILQ_HEAD(, CompareState) net_compares to
patch of "work with colo-frame"
- add some comments in qemu-option.hx
v5:
p3:
- comments from Jason
we poll and handle chardev in comapre thread,
Through this way, there's no need for extra
synchronization with main loop
this depend on another patch:
qemu-char: Fix context for g_source_attach()
- remove QemuEvent
p2:
- remove conn->list_lock
p1:
- move compare_pri/sec_chr_in to p3
- move compare_chr_send to p2
v4:
p4:
- add some comments
- fix some trace-events
- fix tcp compare error
p3:
- add rcu_read_lock().
- fix trace name
- fix jason's other comments
- rebase some Dave's branch function
p2:
- colo_compare_connection() change g_queue_push_head() to
- g_queue_push_tail() match to sorted order.
- remove pkt->s
- move data structure to colo-base.h
- add colo-base.c reuse codes for filter-rewriter
- add some filter-rewriter needs struct
- depends on previous SocketReadState patch
p1:
- except move qemu_chr_add_handlers()
to colo thread
- remove class_finalize
- remove secondary arp codes
- depends on previous SocketReadState patch
v3:
- rebase colo-compare to colo-frame v2.7
- fix most of Dave's comments
(except RCU)
- add TCP,UDP,ICMP and other packet comparison
- add trace-event
- add some comments
- other bug fix
- add RFC index
- add usage in patch 1/4
v2:
- add jhash.h
v1:
- initial patch
Zhang Chen (7):
colo-compare: introduce colo compare initialization
colo-base: add colo-base to define and handle packet
Jhash: add linux kernel jhashtable in qemu
colo-compare: track connection and enqueue packet
qemu-char: Add qemu_chr_add_handlers_full() for GMaincontext
colo-compare: introduce packet comparison thread
colo-compare: add TCP,UDP,ICMP packet comparison
include/qemu/jhash.h | 61 ++++
include/sysemu/char.h | 11 +-
net/Makefile.objs | 2 +
net/colo-base.c | 183 ++++++++++++
net/colo-base.h | 71 +++++
net/colo-compare.c | 765 ++++++++++++++++++++++++++++++++++++++++++++++++++
qemu-char.c | 117 +++++---
qemu-options.hx | 38 +++
trace-events | 9 +
vl.c | 3 +-
10 files changed, 1214 insertions(+), 46 deletions(-)
create mode 100644 include/qemu/jhash.h
create mode 100644 net/colo-base.c
create mode 100644 net/colo-base.h
create mode 100644 net/colo-compare.c
--
2.7.4
^ permalink raw reply [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-08-02 6:26 ` Jason Wang
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet Zhang Chen
` (5 subsequent siblings)
6 siblings, 1 reply; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
This a COLO net ascii figure:
Primary qemu Secondary qemu
+----------------------------------------------------------+ +----------------------------------------------------------------+
| +-----------------------------------------------------+ | | +-----------------------------------------------------------+ |
| | | | | | | |
| | guest | | | | guest | |
| | | | | | | |
| +-------^------------------+--------------------------+ | | +---------------------+--------+----------------------------+ |
| | | | | ^ | |
| | | | | | | |
| | +------------------------------------------------------+ | | | |
|netfilter| | | | | | netfilter | | |
| +----------+ -----------------------+ | | | +-----------------------------------------------------------+ |
| | | | | | | | | | | | filter excute order | |
| | | | | | | | | | | | +-------------------> | |
| | | | | | | | | | | | TCP | |
| | +-----+--+--+ +------v-----+ | +------------+ | | | | +------------+ +---+----+---v+rewriter++ +------------+ | |
| | | | | | | | | | | | | | | | | | | | | |
| | | filter | | filter +----> colo <--------+ +--------> filter +--> adjust | adjust +--> filter | | |
| | | mirror | | redirector | | | compare | | | | | | redirector | | ack | seq | | redirector | | |
| | | | | | | | | | | | | | | | | | | | | |
| | +----^------+ +------------+ | +-----+------+ | | | | +------------+ +--------+--------------+ +---+--------+ | |
| | | tx rx | | | | | | tx all | rx | |
| | | | | | | | +-----------------------------------------------------------+ |
| | | | | | | | | |
| | | filter excute order | | | | | | |
| | | +-------------------> | | | +------------------------------------------------------------+ |
| +-----------------------------------+ | | | |
| | | | | |
+----------------------------------------------------------+ +----------------------------------------------------------------+
|guest receive |guest send
| |
+--------+------------------------------------v------------+
| | NOTE: filter direction is rx/tx/all
| tap | rx:receive packets sent to the netdev
| | tx:receive packets sent by the netdev
+----------------------------------------------------------+
In COLO-compare.
Packets coming from the primary char indev will be sent to outdev
Packets coming from the secondary char dev will be dropped
colo-comapre need two input chardev and one output chardev:
primary_in=chardev1-id
secondary_in=chardev2-id
outdev=chardev3-id
usage:
primary:
-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=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
-chardev socket,id=compare_out0,host=3.3.3.3,port=9005
-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
secondary:
-netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down script=/etc/qemu-ifdown
-device e1000,netdev=hn0,mac=52:a4:00:12:78:66
-chardev socket,id=red0,host=3.3.3.3,port=9003
-chardev socket,id=red1,host=3.3.3.3,port=9004
-object filter-redirector,id=f1,netdev=hn0,queue=tx,indev=red0
-object filter-redirector,id=f2,netdev=hn0,queue=rx,outdev=red1
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
net/Makefile.objs | 1 +
net/colo-compare.c | 222 +++++++++++++++++++++++++++++++++++++++++++++++++++++
qemu-options.hx | 38 +++++++++
vl.c | 3 +-
4 files changed, 263 insertions(+), 1 deletion(-)
create mode 100644 net/colo-compare.c
diff --git a/net/Makefile.objs b/net/Makefile.objs
index b7c22fd..ba92f73 100644
--- a/net/Makefile.objs
+++ b/net/Makefile.objs
@@ -16,3 +16,4 @@ common-obj-$(CONFIG_NETMAP) += netmap.o
common-obj-y += filter.o
common-obj-y += filter-buffer.o
common-obj-y += filter-mirror.o
+common-obj-y += colo-compare.o
diff --git a/net/colo-compare.c b/net/colo-compare.c
new file mode 100644
index 0000000..0402958
--- /dev/null
+++ b/net/colo-compare.c
@@ -0,0 +1,222 @@
+/*
+ * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
+ * (a.k.a. Fault Tolerance or Continuous Replication)
+ *
+ * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
+ * Copyright (c) 2016 FUJITSU LIMITED
+ * Copyright (c) 2016 Intel Corporation
+ *
+ * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or
+ * later. See the COPYING file in the top-level directory.
+ */
+
+#include "qemu/osdep.h"
+#include "qemu/error-report.h"
+#include "qemu-common.h"
+#include "qapi/qmp/qerror.h"
+#include "qapi/error.h"
+#include "net/net.h"
+#include "net/vhost_net.h"
+#include "qom/object_interfaces.h"
+#include "qemu/iov.h"
+#include "qom/object.h"
+#include "qemu/typedefs.h"
+#include "net/queue.h"
+#include "sysemu/char.h"
+#include "qemu/sockets.h"
+#include "qapi-visit.h"
+
+#define TYPE_COLO_COMPARE "colo-compare"
+#define COLO_COMPARE(obj) \
+ OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
+
+#define COMPARE_READ_LEN_MAX NET_BUFSIZE
+
+typedef struct CompareState {
+ Object parent;
+
+ char *pri_indev;
+ char *sec_indev;
+ char *outdev;
+ CharDriverState *chr_pri_in;
+ CharDriverState *chr_sec_in;
+ CharDriverState *chr_out;
+ QTAILQ_ENTRY(CompareState) next;
+ SocketReadState pri_rs;
+ SocketReadState sec_rs;
+} CompareState;
+
+typedef struct CompareClass {
+ ObjectClass parent_class;
+} CompareClass;
+
+static char *compare_get_pri_indev(Object *obj, Error **errp)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ return g_strdup(s->pri_indev);
+}
+
+static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ g_free(s->pri_indev);
+ s->pri_indev = g_strdup(value);
+}
+
+static char *compare_get_sec_indev(Object *obj, Error **errp)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ return g_strdup(s->sec_indev);
+}
+
+static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ g_free(s->sec_indev);
+ s->sec_indev = g_strdup(value);
+}
+
+static char *compare_get_outdev(Object *obj, Error **errp)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ return g_strdup(s->outdev);
+}
+
+static void compare_set_outdev(Object *obj, const char *value, Error **errp)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ g_free(s->outdev);
+ s->outdev = g_strdup(value);
+}
+
+static void compare_pri_rs_finalize(SocketReadState *pri_rs)
+{
+ /* if packet_enqueue pri pkt failed we will send unsupported packet */
+}
+
+static void compare_sec_rs_finalize(SocketReadState *sec_rs)
+{
+ /* if packet_enqueue sec pkt failed we will notify trace */
+}
+
+/*
+ * 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);
+
+ if (!s->pri_indev || !s->sec_indev || !s->outdev) {
+ error_setg(errp, "colo compare needs 'primary_in' ,"
+ "'secondary_in','outdev' property set");
+ return;
+ } else if (!strcmp(s->pri_indev, s->outdev) ||
+ !strcmp(s->sec_indev, s->outdev) ||
+ !strcmp(s->pri_indev, s->sec_indev)) {
+ error_setg(errp, "'indev' and 'outdev' could not be same "
+ "for compare module");
+ return;
+ }
+
+ s->chr_pri_in = qemu_chr_find(s->pri_indev);
+ if (s->chr_pri_in == NULL) {
+ error_setg(errp, "Primary IN Device '%s' not found",
+ s->pri_indev);
+ return;
+ }
+
+ s->chr_sec_in = qemu_chr_find(s->sec_indev);
+ if (s->chr_sec_in == NULL) {
+ error_setg(errp, "Secondary IN Device '%s' not found",
+ s->sec_indev);
+ return;
+ }
+
+ s->chr_out = qemu_chr_find(s->outdev);
+ if (s->chr_out == NULL) {
+ error_setg(errp, "OUT Device '%s' not found", s->outdev);
+ return;
+ }
+
+ qemu_chr_fe_claim_no_fail(s->chr_pri_in);
+
+ qemu_chr_fe_claim_no_fail(s->chr_sec_in);
+
+ qemu_chr_fe_claim_no_fail(s->chr_out);
+
+ net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
+ net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
+
+ return;
+}
+
+static void colo_compare_class_init(ObjectClass *oc, void *data)
+{
+ UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
+
+ ucc->complete = colo_compare_complete;
+}
+
+static void colo_compare_init(Object *obj)
+{
+ object_property_add_str(obj, "primary_in",
+ compare_get_pri_indev, compare_set_pri_indev,
+ NULL);
+ object_property_add_str(obj, "secondary_in",
+ compare_get_sec_indev, compare_set_sec_indev,
+ NULL);
+ object_property_add_str(obj, "outdev",
+ compare_get_outdev, compare_set_outdev,
+ NULL);
+}
+
+static void colo_compare_finalize(Object *obj)
+{
+ CompareState *s = COLO_COMPARE(obj);
+
+ if (s->chr_pri_in) {
+ qemu_chr_add_handlers(s->chr_pri_in, NULL, NULL, NULL, NULL);
+ qemu_chr_fe_release(s->chr_pri_in);
+ }
+ if (s->chr_sec_in) {
+ qemu_chr_add_handlers(s->chr_sec_in, NULL, NULL, NULL, NULL);
+ qemu_chr_fe_release(s->chr_sec_in);
+ }
+ if (s->chr_out) {
+ qemu_chr_fe_release(s->chr_out);
+ }
+
+ g_free(s->pri_indev);
+ g_free(s->sec_indev);
+ g_free(s->outdev);
+}
+
+static const TypeInfo colo_compare_info = {
+ .name = TYPE_COLO_COMPARE,
+ .parent = TYPE_OBJECT,
+ .instance_size = sizeof(CompareState),
+ .instance_init = colo_compare_init,
+ .instance_finalize = colo_compare_finalize,
+ .class_size = sizeof(CompareClass),
+ .class_init = colo_compare_class_init,
+ .interfaces = (InterfaceInfo[]) {
+ { TYPE_USER_CREATABLE },
+ { }
+ }
+};
+
+static void register_types(void)
+{
+ type_register_static(&colo_compare_info);
+}
+
+type_init(register_types);
diff --git a/qemu-options.hx b/qemu-options.hx
index 587de8f..79e5896 100644
--- a/qemu-options.hx
+++ b/qemu-options.hx
@@ -3866,6 +3866,44 @@ Dump the network traffic on netdev @var{dev} to the file specified by
The file format is libpcap, so it can be analyzed with tools such as tcpdump
or Wireshark.
+@item -object colo-compare,id=@var{id},primary_in=@var{chardevid},secondary_in=@var{chardevid},
+outdev=@var{chardevid}
+
+Colo-compare gets packet from primary_in@var{chardevid} and secondary_in@var{chardevid},than compare primary packet with
+secondary packet. if the packet same,we will output primary
+packet to outdev@var{chardevid},else we will notify colo-frame
+do checkpoint and send primary packet to outdev@var{chardevid}.
+
+we can use it with the help of filter-mirror and filter-redirector.
+
+The simple usage:
+
+@example
+
+primary:
+-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=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
+-chardev socket,id=compare_out0,host=3.3.3.3,port=9005
+-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
+
+secondary:
+-netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down script=/etc/qemu-ifdown
+-device e1000,netdev=hn0,mac=52:a4:00:12:78:66
+-chardev socket,id=red0,host=3.3.3.3,port=9003
+-chardev socket,id=red1,host=3.3.3.3,port=9004
+-object filter-redirector,id=f1,netdev=hn0,queue=tx,indev=red0
+-object filter-redirector,id=f2,netdev=hn0,queue=rx,outdev=red1
+
+@end example
+
@item -object secret,id=@var{id},data=@var{string},format=@var{raw|base64}[,keyid=@var{secretid},iv=@var{string}]
@item -object secret,id=@var{id},file=@var{filename},format=@var{raw|base64}[,keyid=@var{secretid},iv=@var{string}]
diff --git a/vl.c b/vl.c
index cbe51ac..c6b9a6f 100644
--- a/vl.c
+++ b/vl.c
@@ -2865,7 +2865,8 @@ static bool object_create_initial(const char *type)
if (g_str_equal(type, "filter-buffer") ||
g_str_equal(type, "filter-dump") ||
g_str_equal(type, "filter-mirror") ||
- g_str_equal(type, "filter-redirector")) {
+ g_str_equal(type, "filter-redirector") ||
+ g_str_equal(type, "colo-compare")) {
return false;
}
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-08-02 6:38 ` Jason Wang
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu Zhang Chen
` (4 subsequent siblings)
6 siblings, 1 reply; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
COLO-base used by colo-compare and filter-rewriter.
this can share common data structure like:net packet,
and share other functions.
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
net/Makefile.objs | 1 +
net/colo-base.c | 74 +++++++++++++++++++++++++++++++++
net/colo-base.h | 38 +++++++++++++++++
net/colo-compare.c | 119 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
trace-events | 3 ++
5 files changed, 233 insertions(+), 2 deletions(-)
create mode 100644 net/colo-base.c
create mode 100644 net/colo-base.h
diff --git a/net/Makefile.objs b/net/Makefile.objs
index ba92f73..119589f 100644
--- a/net/Makefile.objs
+++ b/net/Makefile.objs
@@ -17,3 +17,4 @@ common-obj-y += filter.o
common-obj-y += filter-buffer.o
common-obj-y += filter-mirror.o
common-obj-y += colo-compare.o
+common-obj-y += colo-base.o
diff --git a/net/colo-base.c b/net/colo-base.c
new file mode 100644
index 0000000..f5d5de9
--- /dev/null
+++ b/net/colo-base.c
@@ -0,0 +1,74 @@
+/*
+ * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
+ * (a.k.a. Fault Tolerance or Continuous Replication)
+ *
+ * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
+ * Copyright (c) 2016 FUJITSU LIMITED
+ * Copyright (c) 2016 Intel Corporation
+ *
+ * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or
+ * later. See the COPYING file in the top-level directory.
+ */
+
+#include "qemu/osdep.h"
+#include "qemu/error-report.h"
+#include "net/colo-base.h"
+
+int parse_packet_early(Packet *pkt)
+{
+ int network_length;
+ uint8_t *data = pkt->data;
+ uint16_t l3_proto;
+ ssize_t l2hdr_len = eth_get_l2_hdr_length(data);
+
+ if (pkt->size < ETH_HLEN) {
+ error_report("pkt->size < ETH_HLEN");
+ return 1;
+ }
+ pkt->network_layer = data + ETH_HLEN;
+ l3_proto = eth_get_l3_proto(data, l2hdr_len);
+ if (l3_proto != ETH_P_IP) {
+ return 1;
+ }
+
+ network_length = pkt->ip->ip_hl * 4;
+ if (pkt->size < ETH_HLEN + network_length) {
+ error_report("pkt->size < network_layer + network_length");
+ return 1;
+ }
+ pkt->transport_layer = pkt->network_layer + network_length;
+ if (!pkt->transport_layer) {
+ error_report("pkt->transport_layer is valid");
+ return 1;
+ }
+
+ return 0;
+}
+
+Packet *packet_new(const void *data, int size)
+{
+ Packet *pkt = g_slice_new(Packet);
+
+ pkt->data = g_memdup(data, size);
+ pkt->size = size;
+
+ return pkt;
+}
+
+void packet_destroy(void *opaque, void *user_data)
+{
+ Packet *pkt = opaque;
+
+ g_free(pkt->data);
+ g_slice_free(Packet, pkt);
+}
+
+/*
+ * Clear hashtable, stop this hash growing really huge
+ */
+void connection_hashtable_reset(GHashTable *connection_track_table)
+{
+ g_hash_table_remove_all(connection_track_table);
+}
diff --git a/net/colo-base.h b/net/colo-base.h
new file mode 100644
index 0000000..48835e7
--- /dev/null
+++ b/net/colo-base.h
@@ -0,0 +1,38 @@
+/*
+ * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
+ * (a.k.a. Fault Tolerance or Continuous Replication)
+ *
+ * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
+ * Copyright (c) 2016 FUJITSU LIMITED
+ * Copyright (c) 2016 Intel Corporation
+ *
+ * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
+ *
+ * This work is licensed under the terms of the GNU GPL, version 2 or
+ * later. See the COPYING file in the top-level directory.
+ */
+
+#ifndef QEMU_COLO_BASE_H
+#define QEMU_COLO_BASE_H
+
+#include "slirp/slirp.h"
+#include "qemu/jhash.h"
+
+#define HASHTABLE_MAX_SIZE 16384
+
+typedef struct Packet {
+ void *data;
+ union {
+ uint8_t *network_layer;
+ struct ip *ip;
+ };
+ uint8_t *transport_layer;
+ int size;
+} Packet;
+
+int parse_packet_early(Packet *pkt);
+void connection_hashtable_reset(GHashTable *connection_track_table);
+Packet *packet_new(const void *data, int size);
+void packet_destroy(void *opaque, void *user_data);
+
+#endif /* QEMU_COLO_BASE_H */
diff --git a/net/colo-compare.c b/net/colo-compare.c
index 0402958..7c52cc8 100644
--- a/net/colo-compare.c
+++ b/net/colo-compare.c
@@ -27,13 +27,38 @@
#include "sysemu/char.h"
#include "qemu/sockets.h"
#include "qapi-visit.h"
+#include "net/colo-base.h"
+#include "trace.h"
#define TYPE_COLO_COMPARE "colo-compare"
#define COLO_COMPARE(obj) \
OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
#define COMPARE_READ_LEN_MAX NET_BUFSIZE
+#define MAX_QUEUE_SIZE 1024
+/*
+ + CompareState ++
+ | |
+ +---------------+ +---------------+ +---------------+
+ |conn list +--->conn +--------->conn |
+ +---------------+ +---------------+ +---------------+
+ | | | | | |
+ +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
+ |primary | |secondary |primary | |secondary
+ |packet | |packet + |packet | |packet +
+ +--------+ +--------+ +--------+ +--------+
+ | | | |
+ +---v----+ +---v----+ +---v----+ +---v----+
+ |primary | |secondary |primary | |secondary
+ |packet | |packet + |packet | |packet +
+ +--------+ +--------+ +--------+ +--------+
+ | | | |
+ +---v----+ +---v----+ +---v----+ +---v----+
+ |primary | |secondary |primary | |secondary
+ |packet | |packet + |packet | |packet +
+ +--------+ +--------+ +--------+ +--------+
+*/
typedef struct CompareState {
Object parent;
@@ -46,12 +71,89 @@ typedef struct CompareState {
QTAILQ_ENTRY(CompareState) next;
SocketReadState pri_rs;
SocketReadState sec_rs;
+
+ /* hashtable to save connection */
+ GHashTable *connection_track_table;
+ /* to save unprocessed_connections */
+ GQueue unprocessed_connections;
+ /* proxy current hash size */
+ uint32_t hashtable_size;
} CompareState;
typedef struct CompareClass {
ObjectClass parent_class;
} CompareClass;
+enum {
+ PRIMARY_IN = 0,
+ SECONDARY_IN,
+};
+
+static int compare_chr_send(CharDriverState *out,
+ const uint8_t *buf,
+ uint32_t size);
+
+/*
+ * Return 0 on success, if return -1 means the pkt
+ * is unsupported(arp and ipv6) and will be sent later
+ */
+static int packet_enqueue(CompareState *s, int mode)
+{
+ Packet *pkt = NULL;
+
+ if (mode == PRIMARY_IN) {
+ pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
+ } else {
+ pkt = packet_new(s->sec_rs.buf, s->sec_rs.packet_len);
+ }
+
+ if (parse_packet_early(pkt)) {
+ packet_destroy(pkt, NULL);
+ pkt = NULL;
+ return -1;
+ }
+ /* TODO: get connection key from pkt */
+
+ /*
+ * TODO: use connection key get conn from
+ * connection_track_table
+ */
+
+ /*
+ * TODO: insert pkt to it's conn->primary_list
+ * or conn->secondary_list
+ */
+
+ return 0;
+}
+
+static int compare_chr_send(CharDriverState *out,
+ const uint8_t *buf,
+ uint32_t size)
+{
+ int ret = 0;
+ uint32_t len = htonl(size);
+
+ if (!size) {
+ return 0;
+ }
+
+ ret = qemu_chr_fe_write_all(out, (uint8_t *)&len, sizeof(len));
+ if (ret != sizeof(len)) {
+ goto err;
+ }
+
+ ret = qemu_chr_fe_write_all(out, (uint8_t *)buf, size);
+ if (ret != size) {
+ goto err;
+ }
+
+ return 0;
+
+err:
+ return ret < 0 ? ret : -EIO;
+}
+
static char *compare_get_pri_indev(Object *obj, Error **errp)
{
CompareState *s = COLO_COMPARE(obj);
@@ -99,12 +201,21 @@ static void compare_set_outdev(Object *obj, const char *value, Error **errp)
static void compare_pri_rs_finalize(SocketReadState *pri_rs)
{
- /* if packet_enqueue pri pkt failed we will send unsupported packet */
+ CompareState *s = container_of(pri_rs, CompareState, 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);
+ }
}
static void compare_sec_rs_finalize(SocketReadState *sec_rs)
{
- /* if packet_enqueue sec pkt failed we will notify trace */
+ CompareState *s = container_of(sec_rs, CompareState, sec_rs);
+
+ if (packet_enqueue(s, SECONDARY_IN)) {
+ trace_colo_compare_main("secondary: unsupported packet in");
+ }
}
/*
@@ -156,6 +267,10 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
+ s->hashtable_size = 0;
+
+ /* use g_hash_table_new_full() to new a hashtable */
+
return;
}
diff --git a/trace-events b/trace-events
index ca7211b..703de1a 100644
--- a/trace-events
+++ b/trace-events
@@ -1916,3 +1916,6 @@ aspeed_vic_update_fiq(int flags) "Raising FIQ: %d"
aspeed_vic_update_irq(int flags) "Raising IRQ: %d"
aspeed_vic_read(uint64_t offset, unsigned size, uint32_t value) "From 0x%" PRIx64 " of size %u: 0x%" PRIx32
aspeed_vic_write(uint64_t offset, unsigned size, uint32_t data) "To 0x%" PRIx64 " of size %u: 0x%" PRIx32
+
+# net/colo-compare.c
+colo_compare_main(const char *chr) ": %s"
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-08-02 6:40 ` Jason Wang
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet Zhang Chen
` (3 subsequent siblings)
6 siblings, 1 reply; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
Jhash used by colo-compare and filter-rewriter
to save and lookup net connection info
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
include/qemu/jhash.h | 61 ++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 61 insertions(+)
create mode 100644 include/qemu/jhash.h
diff --git a/include/qemu/jhash.h b/include/qemu/jhash.h
new file mode 100644
index 0000000..0fcd875
--- /dev/null
+++ b/include/qemu/jhash.h
@@ -0,0 +1,61 @@
+/* jhash.h: Jenkins hash support.
+ *
+ * Copyright (C) 2006. Bob Jenkins (bob_jenkins@burtleburtle.net)
+ *
+ * http://burtleburtle.net/bob/hash/
+ *
+ * These are the credits from Bob's sources:
+ *
+ * lookup3.c, by Bob Jenkins, May 2006, Public Domain.
+ *
+ * These are functions for producing 32-bit hashes for hash table lookup.
+ * hashword(), hashlittle(), hashlittle2(), hashbig(), mix(), and final()
+ * are externally useful functions. Routines to test the hash are
+included
+ * if SELF_TEST is defined. You can use this free for any purpose.
+It's in
+ * the public domain. It has no warranty.
+ *
+ * Copyright (C) 2009-2010 Jozsef Kadlecsik (kadlec@blackhole.kfki.hu)
+ *
+ * I've modified Bob's hash to be useful in the Linux kernel, and
+ * any bugs present are my fault.
+ * Jozsef
+ */
+
+#ifndef QEMU_JHASH_H__
+#define QEMU_JHASH_H__
+
+#include "qemu/bitops.h"
+
+/*
+ * hashtable relation copy from linux kernel jhash
+ */
+
+/* __jhash_mix -- mix 3 32-bit values reversibly. */
+#define __jhash_mix(a, b, c) \
+{ \
+ a -= c; a ^= rol32(c, 4); c += b; \
+ b -= a; b ^= rol32(a, 6); a += c; \
+ c -= b; c ^= rol32(b, 8); b += a; \
+ a -= c; a ^= rol32(c, 16); c += b; \
+ b -= a; b ^= rol32(a, 19); a += c; \
+ c -= b; c ^= rol32(b, 4); b += a; \
+}
+
+/* __jhash_final - final mixing of 3 32-bit values (a,b,c) into c */
+#define __jhash_final(a, b, c) \
+{ \
+ c ^= b; c -= rol32(b, 14); \
+ a ^= c; a -= rol32(c, 11); \
+ b ^= a; b -= rol32(a, 25); \
+ c ^= b; c -= rol32(b, 16); \
+ a ^= c; a -= rol32(c, 4); \
+ b ^= a; b -= rol32(a, 14); \
+ c ^= b; c -= rol32(b, 24); \
+}
+
+/* An arbitrary initial parameter */
+#define JHASH_INITVAL 0xdeadbeef
+
+#endif /* QEMU_JHASH_H__ */
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
` (2 preceding siblings ...)
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-08-02 7:14 ` Jason Wang
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 5/7] qemu-char: Add qemu_chr_add_handlers_full() for GMaincontext Zhang Chen
` (2 subsequent siblings)
6 siblings, 1 reply; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
In this patch we use kernel jhash table to track
connection, and then enqueue net packet like this:
+ CompareState ++
| |
+---------------+ +---------------+ +---------------+
|conn list +--->conn +--------->conn |
+---------------+ +---------------+ +---------------+
| | | | | |
+---------------+ +---v----+ +---v----+ +---v----+ +---v----+
|primary | |secondary |primary | |secondary
|packet | |packet + |packet | |packet +
+--------+ +--------+ +--------+ +--------+
| | | |
+---v----+ +---v----+ +---v----+ +---v----+
|primary | |secondary |primary | |secondary
|packet | |packet + |packet | |packet +
+--------+ +--------+ +--------+ +--------+
| | | |
+---v----+ +---v----+ +---v----+ +---v----+
|primary | |secondary |primary | |secondary
|packet | |packet + |packet | |packet +
+--------+ +--------+ +--------+ +--------+
We use conn_list to record connection info.
When we want to enqueue a packet, firstly get the
connection from connection_track_table. then push
the packet to g_queue(pri/sec) in it's own conn.
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
net/colo-base.c | 108 +++++++++++++++++++++++++++++++++++++++++++++++++++++
net/colo-base.h | 30 +++++++++++++++
net/colo-compare.c | 70 +++++++++++++++++++++++++++++-----
3 files changed, 198 insertions(+), 10 deletions(-)
diff --git a/net/colo-base.c b/net/colo-base.c
index f5d5de9..7e91dec 100644
--- a/net/colo-base.c
+++ b/net/colo-base.c
@@ -16,6 +16,29 @@
#include "qemu/error-report.h"
#include "net/colo-base.h"
+uint32_t connection_key_hash(const void *opaque)
+{
+ const ConnectionKey *key = opaque;
+ uint32_t a, b, c;
+
+ /* Jenkins hash */
+ a = b = c = JHASH_INITVAL + sizeof(*key);
+ a += key->src.s_addr;
+ b += key->dst.s_addr;
+ c += (key->src_port | key->dst_port << 16);
+ __jhash_mix(a, b, c);
+
+ a += key->ip_proto;
+ __jhash_final(a, b, c);
+
+ return c;
+}
+
+int connection_key_equal(const void *key1, const void *key2)
+{
+ return memcmp(key1, key2, sizeof(ConnectionKey)) == 0;
+}
+
int parse_packet_early(Packet *pkt)
{
int network_length;
@@ -47,6 +70,62 @@ int parse_packet_early(Packet *pkt)
return 0;
}
+void fill_connection_key(Packet *pkt, ConnectionKey *key)
+{
+ uint32_t tmp_ports;
+
+ key->ip_proto = pkt->ip->ip_p;
+
+ switch (key->ip_proto) {
+ case IPPROTO_TCP:
+ case IPPROTO_UDP:
+ case IPPROTO_DCCP:
+ case IPPROTO_ESP:
+ case IPPROTO_SCTP:
+ case IPPROTO_UDPLITE:
+ tmp_ports = *(uint32_t *)(pkt->transport_layer);
+ key->src = pkt->ip->ip_src;
+ key->dst = pkt->ip->ip_dst;
+ key->src_port = ntohs(tmp_ports & 0xffff);
+ key->dst_port = ntohs(tmp_ports >> 16);
+ break;
+ case IPPROTO_AH:
+ tmp_ports = *(uint32_t *)(pkt->transport_layer + 4);
+ key->src = pkt->ip->ip_src;
+ key->dst = pkt->ip->ip_dst;
+ key->src_port = ntohs(tmp_ports & 0xffff);
+ key->dst_port = ntohs(tmp_ports >> 16);
+ break;
+ default:
+ key->src_port = 0;
+ key->dst_port = 0;
+ break;
+ }
+}
+
+Connection *connection_new(ConnectionKey *key)
+{
+ Connection *conn = g_slice_new(Connection);
+
+ conn->ip_proto = key->ip_proto;
+ conn->processing = false;
+ g_queue_init(&conn->primary_list);
+ g_queue_init(&conn->secondary_list);
+
+ return conn;
+}
+
+void connection_destroy(void *opaque)
+{
+ Connection *conn = opaque;
+
+ g_queue_foreach(&conn->primary_list, packet_destroy, NULL);
+ g_queue_free(&conn->primary_list);
+ g_queue_foreach(&conn->secondary_list, packet_destroy, NULL);
+ g_queue_free(&conn->secondary_list);
+ g_slice_free(Connection, conn);
+}
+
Packet *packet_new(const void *data, int size)
{
Packet *pkt = g_slice_new(Packet);
@@ -72,3 +151,32 @@ void connection_hashtable_reset(GHashTable *connection_track_table)
{
g_hash_table_remove_all(connection_track_table);
}
+
+/* if not found, create a new connection and add to hash table */
+Connection *connection_get(GHashTable *connection_track_table,
+ ConnectionKey *key,
+ uint32_t *hashtable_size)
+{
+ Connection *conn = g_hash_table_lookup(connection_track_table, key);
+
+ if (conn == NULL) {
+ ConnectionKey *new_key = g_memdup(key, sizeof(*key));
+
+ conn = connection_new(key);
+
+ (*hashtable_size) += 1;
+ if (*hashtable_size > HASHTABLE_MAX_SIZE) {
+ error_report("colo proxy connection hashtable full, clear it");
+ connection_hashtable_reset(connection_track_table);
+ /*
+ * when hashtable_size == 0, clear the conn_list
+ * in place where be called.
+ */
+ *hashtable_size = 0;
+ }
+
+ g_hash_table_insert(connection_track_table, new_key, conn);
+ }
+
+ return conn;
+}
diff --git a/net/colo-base.h b/net/colo-base.h
index 48835e7..0505608 100644
--- a/net/colo-base.h
+++ b/net/colo-base.h
@@ -30,7 +30,37 @@ typedef struct Packet {
int size;
} Packet;
+typedef struct ConnectionKey {
+ /* (src, dst) must be grouped, in the same way than in IP header */
+ struct in_addr src;
+ struct in_addr dst;
+ uint16_t src_port;
+ uint16_t dst_port;
+ uint8_t ip_proto;
+} QEMU_PACKED ConnectionKey;
+
+typedef struct Connection {
+ /* connection primary send queue: element type: Packet */
+ GQueue primary_list;
+ /* connection secondary send queue: element type: Packet */
+ GQueue secondary_list;
+ /* flag to enqueue unprocessed_connections */
+ bool processing;
+ uint8_t ip_proto;
+ /* be used by filter-rewriter */
+ tcp_seq primary_seq;
+ tcp_seq secondary_seq;
+} Connection;
+
+uint32_t connection_key_hash(const void *opaque);
+int connection_key_equal(const void *opaque1, const void *opaque2);
int parse_packet_early(Packet *pkt);
+void fill_connection_key(Packet *pkt, ConnectionKey *key);
+Connection *connection_new(ConnectionKey *key);
+void connection_destroy(void *opaque);
+Connection *connection_get(GHashTable *connection_track_table,
+ ConnectionKey *key,
+ uint32_t *hashtable_size);
void connection_hashtable_reset(GHashTable *connection_track_table);
Packet *packet_new(const void *data, int size);
void packet_destroy(void *opaque, void *user_data);
diff --git a/net/colo-compare.c b/net/colo-compare.c
index 7c52cc8..5f87710 100644
--- a/net/colo-compare.c
+++ b/net/colo-compare.c
@@ -72,6 +72,11 @@ typedef struct CompareState {
SocketReadState pri_rs;
SocketReadState sec_rs;
+ /* connection list: the connections belonged to this NIC could be found
+ * in this list.
+ * element type: Connection
+ */
+ GQueue conn_list;
/* hashtable to save connection */
GHashTable *connection_track_table;
/* to save unprocessed_connections */
@@ -93,13 +98,30 @@ static int compare_chr_send(CharDriverState *out,
const uint8_t *buf,
uint32_t size);
+static void colo_rm_connection(void *opaque, void *user_data)
+{
+ Connection *conn = opaque;
+ Packet *pkt = NULL;
+
+ while (!g_queue_is_empty(&conn->primary_list)) {
+ pkt = g_queue_pop_head(&conn->primary_list);
+ packet_destroy(pkt, NULL);
+ }
+ while (!g_queue_is_empty(&conn->secondary_list)) {
+ pkt = g_queue_pop_head(&conn->secondary_list);
+ packet_destroy(pkt, NULL);
+ }
+}
+
/*
* Return 0 on success, if return -1 means the pkt
* is unsupported(arp and ipv6) and will be sent later
*/
static int packet_enqueue(CompareState *s, int mode)
{
+ ConnectionKey key = {{ 0 } };
Packet *pkt = NULL;
+ Connection *conn;
if (mode == PRIMARY_IN) {
pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
@@ -112,17 +134,38 @@ static int packet_enqueue(CompareState *s, int mode)
pkt = NULL;
return -1;
}
- /* TODO: get connection key from pkt */
+ fill_connection_key(pkt, &key);
- /*
- * TODO: use connection key get conn from
- * connection_track_table
- */
+ conn = connection_get(s->connection_track_table,
+ &key,
+ &s->hashtable_size);
- /*
- * TODO: insert pkt to it's conn->primary_list
- * or conn->secondary_list
- */
+ if (!s->hashtable_size) {
+ g_queue_foreach(&s->conn_list, colo_rm_connection, NULL);
+ }
+
+ if (!conn->processing) {
+ g_queue_push_tail(&s->conn_list, conn);
+ conn->processing = true;
+ }
+
+ if (mode == PRIMARY_IN) {
+ if (g_queue_get_length(&conn->primary_list) <
+ MAX_QUEUE_SIZE) {
+ g_queue_push_tail(&conn->primary_list, pkt);
+ } else {
+ error_report("colo compare primary queue size too big,"
+ "drop packet");
+ }
+ } else {
+ if (g_queue_get_length(&conn->secondary_list) <
+ MAX_QUEUE_SIZE) {
+ g_queue_push_tail(&conn->secondary_list, pkt);
+ } else {
+ error_report("colo compare secondary queue size too big,"
+ "drop packet");
+ }
+ }
return 0;
}
@@ -267,9 +310,14 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
+ g_queue_init(&s->conn_list);
+
s->hashtable_size = 0;
- /* use g_hash_table_new_full() to new a hashtable */
+ s->connection_track_table = g_hash_table_new_full(connection_key_hash,
+ connection_key_equal,
+ g_free,
+ connection_destroy);
return;
}
@@ -310,6 +358,8 @@ static void colo_compare_finalize(Object *obj)
qemu_chr_fe_release(s->chr_out);
}
+ g_queue_free(&s->conn_list);
+
g_free(s->pri_indev);
g_free(s->sec_indev);
g_free(s->outdev);
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 5/7] qemu-char: Add qemu_chr_add_handlers_full() for GMaincontext
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
` (3 preceding siblings ...)
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison Zhang Chen
6 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert, Daniel P . Berrange,
Paolo Bonzini
Add qemu_chr_add_handlers_full() API, we can use
this API pass in a GMainContext,make handler run
in the context rather than main_loop.
This comments from Daniel P . Berrange.
Cc: Daniel P . Berrange <berrange@redhat.com>
Cc: Paolo Bonzini <pbonzini@redhat.com>
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
include/sysemu/char.h | 11 ++++-
qemu-char.c | 117 +++++++++++++++++++++++++++++++-------------------
2 files changed, 83 insertions(+), 45 deletions(-)
diff --git a/include/sysemu/char.h b/include/sysemu/char.h
index 307fd8f..86888bc 100644
--- a/include/sysemu/char.h
+++ b/include/sysemu/char.h
@@ -65,7 +65,8 @@ struct CharDriverState {
int (*chr_sync_read)(struct CharDriverState *s,
const uint8_t *buf, int len);
GSource *(*chr_add_watch)(struct CharDriverState *s, GIOCondition cond);
- void (*chr_update_read_handler)(struct CharDriverState *s);
+ void (*chr_update_read_handler_full)(struct CharDriverState *s,
+ GMainContext *context);
int (*chr_ioctl)(struct CharDriverState *s, int cmd, void *arg);
int (*get_msgfds)(struct CharDriverState *s, int* fds, int num);
int (*set_msgfds)(struct CharDriverState *s, int *fds, int num);
@@ -388,6 +389,14 @@ void qemu_chr_add_handlers(CharDriverState *s,
IOEventHandler *fd_event,
void *opaque);
+/* This API can make handler run in the context what you pass to. */
+void qemu_chr_add_handlers_full(CharDriverState *s,
+ IOCanReadHandler *fd_can_read,
+ IOReadHandler *fd_read,
+ IOEventHandler *fd_event,
+ void *opaque,
+ GMainContext *context);
+
void qemu_chr_be_generic_open(CharDriverState *s);
void qemu_chr_accept_input(CharDriverState *s);
int qemu_chr_add_client(CharDriverState *s, int fd);
diff --git a/qemu-char.c b/qemu-char.c
index b597ee1..18e7d3c 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -448,11 +448,12 @@ void qemu_chr_fe_printf(CharDriverState *s, const char *fmt, ...)
static void remove_fd_in_watch(CharDriverState *chr);
-void qemu_chr_add_handlers(CharDriverState *s,
- IOCanReadHandler *fd_can_read,
- IOReadHandler *fd_read,
- IOEventHandler *fd_event,
- void *opaque)
+void qemu_chr_add_handlers_full(CharDriverState *s,
+ IOCanReadHandler *fd_can_read,
+ IOReadHandler *fd_read,
+ IOEventHandler *fd_event,
+ void *opaque,
+ GMainContext *context)
{
int fe_open;
@@ -466,8 +467,9 @@ void qemu_chr_add_handlers(CharDriverState *s,
s->chr_read = fd_read;
s->chr_event = fd_event;
s->handler_opaque = opaque;
- if (fe_open && s->chr_update_read_handler)
- s->chr_update_read_handler(s);
+ if (fe_open && s->chr_update_read_handler_full) {
+ s->chr_update_read_handler_full(s, context);
+ }
if (!s->explicit_fe_open) {
qemu_chr_fe_set_open(s, fe_open);
@@ -480,6 +482,16 @@ void qemu_chr_add_handlers(CharDriverState *s,
}
}
+void qemu_chr_add_handlers(CharDriverState *s,
+ IOCanReadHandler *fd_can_read,
+ IOReadHandler *fd_read,
+ IOEventHandler *fd_event,
+ void *opaque)
+{
+ qemu_chr_add_handlers_full(s, fd_can_read, fd_read,
+ fd_event, opaque, NULL);
+}
+
static int null_chr_write(CharDriverState *chr, const uint8_t *buf, int len)
{
return len;
@@ -717,7 +729,8 @@ static void mux_chr_event(void *opaque, int event)
mux_chr_send_event(d, i, event);
}
-static void mux_chr_update_read_handler(CharDriverState *chr)
+static void mux_chr_update_read_handler_full(CharDriverState *chr,
+ GMainContext *context)
{
MuxDriver *d = chr->opaque;
@@ -731,8 +744,10 @@ static void mux_chr_update_read_handler(CharDriverState *chr)
d->chr_event[d->mux_cnt] = chr->chr_event;
/* Fix up the real driver with mux routines */
if (d->mux_cnt == 0) {
- qemu_chr_add_handlers(d->drv, mux_chr_can_read, mux_chr_read,
- mux_chr_event, chr);
+ qemu_chr_add_handlers_full(d->drv, mux_chr_can_read,
+ mux_chr_read,
+ mux_chr_event,
+ chr, context);
}
if (d->focus != -1) {
mux_chr_send_event(d, d->focus, CHR_EVENT_MUX_OUT);
@@ -813,7 +828,7 @@ static CharDriverState *qemu_chr_open_mux(const char *id,
d->drv = drv;
d->focus = -1;
chr->chr_write = mux_chr_write;
- chr->chr_update_read_handler = mux_chr_update_read_handler;
+ chr->chr_update_read_handler_full = mux_chr_update_read_handler_full;
chr->chr_accept_input = mux_chr_accept_input;
/* Frontend guest-open / -close notification is not support with muxes */
chr->chr_set_fe_open = NULL;
@@ -840,6 +855,7 @@ typedef struct IOWatchPoll
IOCanReadHandler *fd_can_read;
GSourceFunc fd_read;
void *opaque;
+ GMainContext *context;
} IOWatchPoll;
static IOWatchPoll *io_watch_poll_from_source(GSource *source)
@@ -847,7 +863,8 @@ static IOWatchPoll *io_watch_poll_from_source(GSource *source)
return container_of(source, IOWatchPoll, parent);
}
-static gboolean io_watch_poll_prepare(GSource *source, gint *timeout_)
+static gboolean io_watch_poll_prepare_full(GSource *source,
+ gint *timeout_)
{
IOWatchPoll *iwp = io_watch_poll_from_source(source);
bool now_active = iwp->fd_can_read(iwp->opaque) > 0;
@@ -860,7 +877,7 @@ static gboolean io_watch_poll_prepare(GSource *source, gint *timeout_)
iwp->src = qio_channel_create_watch(
iwp->ioc, G_IO_IN | G_IO_ERR | G_IO_HUP | G_IO_NVAL);
g_source_set_callback(iwp->src, iwp->fd_read, iwp->opaque, NULL);
- g_source_attach(iwp->src, NULL);
+ g_source_attach(iwp->src, iwp->context);
} else {
g_source_destroy(iwp->src);
g_source_unref(iwp->src);
@@ -896,30 +913,33 @@ static void io_watch_poll_finalize(GSource *source)
assert(iwp->src == NULL);
}
-static GSourceFuncs io_watch_poll_funcs = {
- .prepare = io_watch_poll_prepare,
+static GSourceFuncs io_watch_poll_funcs_full = {
+ .prepare = io_watch_poll_prepare_full,
.check = io_watch_poll_check,
.dispatch = io_watch_poll_dispatch,
.finalize = io_watch_poll_finalize,
};
/* Can only be used for read */
-static guint io_add_watch_poll(QIOChannel *ioc,
- IOCanReadHandler *fd_can_read,
- QIOChannelFunc fd_read,
- gpointer user_data)
+static guint io_add_watch_poll_full(QIOChannel *ioc,
+ IOCanReadHandler *fd_can_read,
+ QIOChannelFunc fd_read,
+ gpointer user_data,
+ GMainContext *context)
{
IOWatchPoll *iwp;
int tag;
- iwp = (IOWatchPoll *) g_source_new(&io_watch_poll_funcs, sizeof(IOWatchPoll));
+ iwp = (IOWatchPoll *) g_source_new(&io_watch_poll_funcs_full,
+ sizeof(IOWatchPoll));
iwp->fd_can_read = fd_can_read;
iwp->opaque = user_data;
iwp->ioc = ioc;
iwp->fd_read = (GSourceFunc) fd_read;
iwp->src = NULL;
+ iwp->context = context;
- tag = g_source_attach(&iwp->parent, NULL);
+ tag = g_source_attach(&iwp->parent, context);
g_source_unref(&iwp->parent);
return tag;
}
@@ -1051,15 +1071,17 @@ static GSource *fd_chr_add_watch(CharDriverState *chr, GIOCondition cond)
return qio_channel_create_watch(s->ioc_out, cond);
}
-static void fd_chr_update_read_handler(CharDriverState *chr)
+static void fd_chr_update_read_handler_full(CharDriverState *chr,
+ GMainContext *context)
{
FDCharDriver *s = chr->opaque;
remove_fd_in_watch(chr);
if (s->ioc_in) {
- chr->fd_in_tag = io_add_watch_poll(s->ioc_in,
- fd_chr_read_poll,
- fd_chr_read, chr);
+ chr->fd_in_tag = io_add_watch_poll_full(s->ioc_in,
+ fd_chr_read_poll,
+ fd_chr_read, chr,
+ context);
}
}
@@ -1098,7 +1120,7 @@ static CharDriverState *qemu_chr_open_fd(int fd_in, int fd_out,
chr->opaque = s;
chr->chr_add_watch = fd_chr_add_watch;
chr->chr_write = fd_chr_write;
- chr->chr_update_read_handler = fd_chr_update_read_handler;
+ chr->chr_update_read_handler_full = fd_chr_update_read_handler_full;
chr->chr_close = fd_chr_close;
return chr;
@@ -1303,7 +1325,8 @@ static void pty_chr_update_read_handler_locked(CharDriverState *chr)
}
}
-static void pty_chr_update_read_handler(CharDriverState *chr)
+static void pty_chr_update_read_handler_full(CharDriverState *chr,
+ GMainContext *context)
{
qemu_mutex_lock(&chr->chr_write_lock);
pty_chr_update_read_handler_locked(chr);
@@ -1405,9 +1428,10 @@ static void pty_chr_state(CharDriverState *chr, int connected)
s->open_tag = g_idle_add(qemu_chr_be_generic_open_func, chr);
}
if (!chr->fd_in_tag) {
- chr->fd_in_tag = io_add_watch_poll(s->ioc,
- pty_chr_read_poll,
- pty_chr_read, chr);
+ chr->fd_in_tag = io_add_watch_poll_full(s->ioc,
+ pty_chr_read_poll,
+ pty_chr_read,
+ chr, NULL);
}
}
}
@@ -1464,7 +1488,7 @@ static CharDriverState *qemu_chr_open_pty(const char *id,
s = g_new0(PtyCharDriver, 1);
chr->opaque = s;
chr->chr_write = pty_chr_write;
- chr->chr_update_read_handler = pty_chr_update_read_handler;
+ chr->chr_update_read_handler_full = pty_chr_update_read_handler_full;
chr->chr_close = pty_chr_close;
chr->chr_add_watch = pty_chr_add_watch;
chr->explicit_be_open = true;
@@ -2546,15 +2570,17 @@ static gboolean udp_chr_read(QIOChannel *chan, GIOCondition cond, void *opaque)
return TRUE;
}
-static void udp_chr_update_read_handler(CharDriverState *chr)
+static void udp_chr_update_read_handler_full(CharDriverState *chr,
+ GMainContext *context)
{
NetCharDriver *s = chr->opaque;
remove_fd_in_watch(chr);
if (s->ioc) {
- chr->fd_in_tag = io_add_watch_poll(s->ioc,
- udp_chr_read_poll,
- udp_chr_read, chr);
+ chr->fd_in_tag = io_add_watch_poll_full(s->ioc,
+ udp_chr_read_poll,
+ udp_chr_read, chr,
+ context);
}
}
@@ -2588,7 +2614,7 @@ static CharDriverState *qemu_chr_open_udp(QIOChannelSocket *sioc,
s->bufptr = 0;
chr->opaque = s;
chr->chr_write = udp_chr_write;
- chr->chr_update_read_handler = udp_chr_update_read_handler;
+ chr->chr_update_read_handler_full = udp_chr_update_read_handler_full;
chr->chr_close = udp_chr_close;
/* be isn't opened until we get a connection */
chr->explicit_be_open = true;
@@ -2929,14 +2955,16 @@ static void tcp_chr_connect(void *opaque)
s->connected = 1;
if (s->ioc) {
- chr->fd_in_tag = io_add_watch_poll(s->ioc,
- tcp_chr_read_poll,
- tcp_chr_read, chr);
+ chr->fd_in_tag = io_add_watch_poll_full(s->ioc,
+ tcp_chr_read_poll,
+ tcp_chr_read,
+ chr, NULL);
}
qemu_chr_be_generic_open(chr);
}
-static void tcp_chr_update_read_handler(CharDriverState *chr)
+static void tcp_chr_update_read_handler_full(CharDriverState *chr,
+ GMainContext *context)
{
TCPCharDriver *s = chr->opaque;
@@ -2946,9 +2974,10 @@ static void tcp_chr_update_read_handler(CharDriverState *chr)
remove_fd_in_watch(chr);
if (s->ioc) {
- chr->fd_in_tag = io_add_watch_poll(s->ioc,
- tcp_chr_read_poll,
- tcp_chr_read, chr);
+ chr->fd_in_tag = io_add_watch_poll_full(s->ioc,
+ tcp_chr_read_poll,
+ tcp_chr_read, chr,
+ context);
}
}
@@ -4409,7 +4438,7 @@ static CharDriverState *qmp_chardev_open_socket(const char *id,
chr->set_msgfds = tcp_set_msgfds;
chr->chr_add_client = tcp_chr_add_client;
chr->chr_add_watch = tcp_chr_add_watch;
- chr->chr_update_read_handler = tcp_chr_update_read_handler;
+ chr->chr_update_read_handler_full = tcp_chr_update_read_handler_full;
/* be isn't opened until we get a connection */
chr->explicit_be_open = true;
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
` (4 preceding siblings ...)
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 5/7] qemu-char: Add qemu_chr_add_handlers_full() for GMaincontext Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-08-02 7:52 ` Jason Wang
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison Zhang Chen
6 siblings, 1 reply; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
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,
after REGULAR_PACKET_CHECK_MS milliseconds we set
the primary packet as old_packet,then do a checkpoint.
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
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)
+{
+ bool found_old = false;
+ Connection *conn = opaque;
+
+ g_queue_foreach(&conn->primary_list, colo_old_packet_check_one,
+ &found_old);
+ 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);
+ } else {
+ 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)
@@ -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);
+ 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);
+
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
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
` (5 preceding siblings ...)
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread Zhang Chen
@ 2016-07-26 1:49 ` Zhang Chen
2016-08-02 8:04 ` Jason Wang
6 siblings, 1 reply; 20+ messages in thread
From: Zhang Chen @ 2016-07-26 1:49 UTC (permalink / raw)
To: qemu devel, Jason Wang
Cc: Zhang Chen, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert
We add TCP,UDP,ICMP packet comparison to replace
IP packet comparison. This can increase the
accuracy of the package comparison.
less checkpoint more efficiency.
Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
---
net/colo-compare.c | 174 +++++++++++++++++++++++++++++++++++++++++++++++++++--
trace-events | 4 ++
2 files changed, 174 insertions(+), 4 deletions(-)
diff --git a/net/colo-compare.c b/net/colo-compare.c
index e020edc..c7bb5f7 100644
--- a/net/colo-compare.c
+++ b/net/colo-compare.c
@@ -18,6 +18,7 @@
#include "qapi/qmp/qerror.h"
#include "qapi/error.h"
#include "net/net.h"
+#include "net/eth.h"
#include "net/vhost_net.h"
#include "qom/object_interfaces.h"
#include "qemu/iov.h"
@@ -197,9 +198,158 @@ static int colo_packet_compare(Packet *ppkt, Packet *spkt)
}
}
-static int colo_packet_compare_all(Packet *spkt, Packet *ppkt)
+/*
+ * called from the compare thread on the primary
+ * for compare tcp packet
+ * compare_tcp copied from Dr. David Alan Gilbert's branch
+ */
+static int colo_packet_compare_tcp(Packet *spkt, Packet *ppkt)
+{
+ struct tcphdr *ptcp, *stcp;
+ int res;
+ char *sdebug, *ddebug;
+
+ trace_colo_compare_main("compare tcp");
+ if (ppkt->size != spkt->size) {
+ if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
+ trace_colo_compare_main("pkt size not same");
+ }
+ return -1;
+ }
+
+ ptcp = (struct tcphdr *)ppkt->transport_layer;
+ stcp = (struct tcphdr *)spkt->transport_layer;
+
+ if (ptcp->th_seq != stcp->th_seq) {
+ if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
+ trace_colo_compare_main("pkt tcp seq not same");
+ }
+ return -1;
+ }
+
+ /*
+ * The 'identification' field in the IP header is *very* random
+ * it almost never matches. Fudge this by ignoring differences in
+ * unfragmented packets; they'll normally sort themselves out if different
+ * anyway, and it should recover at the TCP level.
+ * An alternative would be to get both the primary and secondary to rewrite
+ * somehow; but that would need some sync traffic to sync the state
+ */
+ if (ntohs(ppkt->ip->ip_off) & IP_DF) {
+ spkt->ip->ip_id = ppkt->ip->ip_id;
+ /* and the sum will be different if the IDs were different */
+ spkt->ip->ip_sum = ppkt->ip->ip_sum;
+ }
+
+ res = memcmp(ppkt->data + ETH_HLEN, spkt->data + ETH_HLEN,
+ (spkt->size - ETH_HLEN));
+
+ if (res != 0 && trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
+ sdebug = strdup(inet_ntoa(ppkt->ip->ip_src));
+ ddebug = strdup(inet_ntoa(ppkt->ip->ip_dst));
+ fprintf(stderr, "%s: src/dst: %s/%s p: seq/ack=%u/%u"
+ " s: seq/ack=%u/%u res=%d flags=%x/%x\n", __func__,
+ sdebug, ddebug,
+ ntohl(ptcp->th_seq), ntohl(ptcp->th_ack),
+ ntohl(stcp->th_seq), ntohl(stcp->th_ack),
+ res, ptcp->th_flags, stcp->th_flags);
+
+ trace_colo_compare_tcp_miscompare("Primary len", ppkt->size);
+ qemu_hexdump((char *)ppkt->data, stderr, "colo-compare", ppkt->size);
+ trace_colo_compare_tcp_miscompare("Secondary len", spkt->size);
+ qemu_hexdump((char *)spkt->data, stderr, "colo-compare", spkt->size);
+
+ g_free(sdebug);
+ g_free(ddebug);
+ }
+
+ return res;
+}
+
+/*
+ * called from the compare thread on the primary
+ * for compare udp packet
+ */
+static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
+{
+ int ret;
+
+ trace_colo_compare_main("compare udp");
+ ret = colo_packet_compare(ppkt, spkt);
+
+ if (ret) {
+ trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
+ qemu_hexdump((char *)ppkt->data, stderr, "colo-compare", ppkt->size);
+ trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
+ qemu_hexdump((char *)spkt->data, stderr, "colo-compare", spkt->size);
+ }
+
+ return ret;
+}
+
+/*
+ * called from the compare thread on the primary
+ * for compare icmp packet
+ */
+static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
{
- trace_colo_compare_main("compare all");
+ int network_length;
+ struct icmp *icmp_ppkt, *icmp_spkt;
+
+ trace_colo_compare_main("compare icmp");
+ network_length = ppkt->ip->ip_hl * 4;
+ if (ppkt->size != spkt->size ||
+ ppkt->size < network_length + ETH_HLEN) {
+ return -1;
+ }
+ icmp_ppkt = (struct icmp *)(ppkt->data + network_length + ETH_HLEN);
+ icmp_spkt = (struct icmp *)(spkt->data + network_length + ETH_HLEN);
+
+ if ((icmp_ppkt->icmp_type == icmp_spkt->icmp_type) &&
+ (icmp_ppkt->icmp_code == icmp_spkt->icmp_code)) {
+ if (icmp_ppkt->icmp_type == ICMP_REDIRECT) {
+ if (icmp_ppkt->icmp_gwaddr.s_addr !=
+ icmp_spkt->icmp_gwaddr.s_addr) {
+ trace_colo_compare_main("icmp_gwaddr.s_addr not same");
+ return -1;
+ }
+ } else if ((icmp_ppkt->icmp_type == ICMP_UNREACH) &&
+ (icmp_ppkt->icmp_type == ICMP_UNREACH_NEEDFRAG)) {
+ if (icmp_ppkt->icmp_nextmtu != icmp_spkt->icmp_nextmtu) {
+ trace_colo_compare_main("icmp_nextmtu not same");
+ return -1;
+ }
+ }
+
+ if (colo_packet_compare(ppkt, spkt)) {
+ trace_colo_compare_icmp_miscompare("primary pkt size",
+ ppkt->size);
+ qemu_hexdump((char *)ppkt->data, stderr, "colo-compare",
+ ppkt->size);
+ trace_colo_compare_icmp_miscompare("Secondary pkt size",
+ spkt->size);
+ qemu_hexdump((char *)spkt->data, stderr, "colo-compare",
+ spkt->size);
+ return -1;
+ }
+ } else {
+ return -1;
+ }
+
+ return 0;
+}
+
+/*
+ * called from the compare thread on the primary
+ * for compare other packet
+ */
+static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
+{
+ trace_colo_compare_main("compare other");
+ 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));
return colo_packet_compare(ppkt, spkt);
}
@@ -263,8 +413,24 @@ static void colo_compare_connection(void *opaque, void *user_data)
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);
+ switch (conn->ip_proto) {
+ case IPPROTO_TCP:
+ result = g_queue_find_custom(&conn->secondary_list,
+ pkt, (GCompareFunc)colo_packet_compare_tcp);
+ break;
+ case IPPROTO_UDP:
+ result = g_queue_find_custom(&conn->secondary_list,
+ pkt, (GCompareFunc)colo_packet_compare_udp);
+ break;
+ case IPPROTO_ICMP:
+ result = g_queue_find_custom(&conn->secondary_list,
+ pkt, (GCompareFunc)colo_packet_compare_icmp);
+ break;
+ default:
+ result = g_queue_find_custom(&conn->secondary_list,
+ pkt, (GCompareFunc)colo_packet_compare_other);
+ break;
+ }
if (result) {
ret = compare_chr_send(s->chr_out, pkt->data, pkt->size);
diff --git a/trace-events b/trace-events
index 1537e91..ab22eb2 100644
--- a/trace-events
+++ b/trace-events
@@ -1919,5 +1919,9 @@ 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_tcp_miscompare(const char *sta, int size) ": %s = %d"
+colo_compare_udp_miscompare(const char *sta, int size) ": %s = %d"
+colo_compare_icmp_miscompare(const char *sta, int size) ": %s = %d"
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
+colo_compare_miscompare(void) ""
--
2.7.4
^ permalink raw reply related [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization Zhang Chen
@ 2016-08-02 6:26 ` Jason Wang
2016-08-03 2:51 ` Zhang Chen
0 siblings, 1 reply; 20+ messages in thread
From: Jason Wang @ 2016-08-02 6:26 UTC (permalink / raw)
To: Zhang Chen, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 2016年07月26日 09:49, Zhang Chen wrote:
> This a COLO net ascii figure:
>
> Primary qemu Secondary qemu
> +----------------------------------------------------------+ +----------------------------------------------------------------+
> | +-----------------------------------------------------+ | | +-----------------------------------------------------------+ |
> | | | | | | | |
> | | guest | | | | guest | |
> | | | | | | | |
> | +-------^------------------+--------------------------+ | | +---------------------+--------+----------------------------+ |
> | | | | | ^ | |
> | | | | | | | |
> | | +------------------------------------------------------+ | | | |
> |netfilter| | | | | | netfilter | | |
> | +----------+ -----------------------+ | | | +-----------------------------------------------------------+ |
> | | | | | | | | | | | | filter excute order | |
> | | | | | | | | | | | | +-------------------> | |
> | | | | | | | | | | | | TCP | |
> | | +-----+--+--+ +------v-----+ | +------------+ | | | | +------------+ +---+----+---v+rewriter++ +------------+ | |
> | | | | | | | | | | | | | | | | | | | | | |
> | | | filter | | filter +----> colo <--------+ +--------> filter +--> adjust | adjust +--> filter | | |
> | | | mirror | | redirector | | | compare | | | | | | redirector | | ack | seq | | redirector | | |
> | | | | | | | | | | | | | | | | | | | | | |
> | | +----^------+ +------------+ | +-----+------+ | | | | +------------+ +--------+--------------+ +---+--------+ | |
> | | | tx rx | | | | | | tx all | rx | |
> | | | | | | | | +-----------------------------------------------------------+ |
> | | | | | | | | | |
> | | | filter excute order | | | | | | |
> | | | +-------------------> | | | +------------------------------------------------------------+ |
> | +-----------------------------------+ | | | |
> | | | | | |
> +----------------------------------------------------------+ +----------------------------------------------------------------+
> |guest receive |guest send
> | |
> +--------+------------------------------------v------------+
> | | NOTE: filter direction is rx/tx/all
> | tap | rx:receive packets sent to the netdev
> | | tx:receive packets sent by the netdev
> +----------------------------------------------------------+
>
> In COLO-compare.
> Packets coming from the primary char indev will be sent to outdev
> Packets coming from the secondary char dev will be dropped
> colo-comapre need two input chardev and one output chardev:
> primary_in=chardev1-id
> secondary_in=chardev2-id
> outdev=chardev3-id
Though it has 'compare' in its name, this description needs some key
information still. e.g what it did (packet comparing). And then you can
describe where were the data sources from.
> usage:
>
> primary:
> -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=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
> -chardev socket,id=compare_out0,host=3.3.3.3,port=9005
> -object filter-mirror,id=m0,netdev=hn0,queue=tx,outdev=mirror0
> -object filter-redirector,netdev=hn0,id=redire0,queue=rx,indev=compare_out
Looks like this filter were missed in the above figure.
> -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
>
> secondary:
> -netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down script=/etc/qemu-ifdown
> -device e1000,netdev=hn0,mac=52:a4:00:12:78:66
> -chardev socket,id=red0,host=3.3.3.3,port=9003
> -chardev socket,id=red1,host=3.3.3.3,port=9004
> -object filter-redirector,id=f1,netdev=hn0,queue=tx,indev=red0
> -object filter-redirector,id=f2,netdev=hn0,queue=rx,outdev=red1
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
> net/Makefile.objs | 1 +
> net/colo-compare.c | 222 +++++++++++++++++++++++++++++++++++++++++++++++++++++
> qemu-options.hx | 38 +++++++++
> vl.c | 3 +-
> 4 files changed, 263 insertions(+), 1 deletion(-)
> create mode 100644 net/colo-compare.c
>
> diff --git a/net/Makefile.objs b/net/Makefile.objs
> index b7c22fd..ba92f73 100644
> --- a/net/Makefile.objs
> +++ b/net/Makefile.objs
> @@ -16,3 +16,4 @@ common-obj-$(CONFIG_NETMAP) += netmap.o
> common-obj-y += filter.o
> common-obj-y += filter-buffer.o
> common-obj-y += filter-mirror.o
> +common-obj-y += colo-compare.o
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> new file mode 100644
> index 0000000..0402958
> --- /dev/null
> +++ b/net/colo-compare.c
> @@ -0,0 +1,222 @@
> +/*
> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
> + * (a.k.a. Fault Tolerance or Continuous Replication)
> + *
> + * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
> + * Copyright (c) 2016 FUJITSU LIMITED
> + * Copyright (c) 2016 Intel Corporation
> + *
> + * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or
> + * later. See the COPYING file in the top-level directory.
> + */
> +
> +#include "qemu/osdep.h"
> +#include "qemu/error-report.h"
> +#include "qemu-common.h"
> +#include "qapi/qmp/qerror.h"
> +#include "qapi/error.h"
> +#include "net/net.h"
> +#include "net/vhost_net.h"
> +#include "qom/object_interfaces.h"
> +#include "qemu/iov.h"
> +#include "qom/object.h"
> +#include "qemu/typedefs.h"
> +#include "net/queue.h"
> +#include "sysemu/char.h"
> +#include "qemu/sockets.h"
> +#include "qapi-visit.h"
> +
> +#define TYPE_COLO_COMPARE "colo-compare"
> +#define COLO_COMPARE(obj) \
> + OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
> +
> +#define COMPARE_READ_LEN_MAX NET_BUFSIZE
> +
> +typedef struct CompareState {
> + Object parent;
> +
> + char *pri_indev;
> + char *sec_indev;
> + char *outdev;
> + CharDriverState *chr_pri_in;
> + CharDriverState *chr_sec_in;
> + CharDriverState *chr_out;
> + QTAILQ_ENTRY(CompareState) next;
> + SocketReadState pri_rs;
> + SocketReadState sec_rs;
> +} CompareState;
> +
> +typedef struct CompareClass {
> + ObjectClass parent_class;
> +} CompareClass;
> +
> +static char *compare_get_pri_indev(Object *obj, Error **errp)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + return g_strdup(s->pri_indev);
> +}
> +
> +static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + g_free(s->pri_indev);
> + s->pri_indev = g_strdup(value);
> +}
> +
> +static char *compare_get_sec_indev(Object *obj, Error **errp)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + return g_strdup(s->sec_indev);
> +}
> +
> +static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + g_free(s->sec_indev);
> + s->sec_indev = g_strdup(value);
> +}
> +
> +static char *compare_get_outdev(Object *obj, Error **errp)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + return g_strdup(s->outdev);
> +}
> +
> +static void compare_set_outdev(Object *obj, const char *value, Error **errp)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + g_free(s->outdev);
> + s->outdev = g_strdup(value);
> +}
> +
> +static void compare_pri_rs_finalize(SocketReadState *pri_rs)
> +{
> + /* if packet_enqueue pri pkt failed we will send unsupported packet */
> +}
> +
> +static void compare_sec_rs_finalize(SocketReadState *sec_rs)
> +{
> + /* if packet_enqueue sec pkt failed we will notify trace */
> +}
> +
> +/*
> + * 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);
> +
> + if (!s->pri_indev || !s->sec_indev || !s->outdev) {
> + error_setg(errp, "colo compare needs 'primary_in' ,"
> + "'secondary_in','outdev' property set");
> + return;
> + } else if (!strcmp(s->pri_indev, s->outdev) ||
> + !strcmp(s->sec_indev, s->outdev) ||
> + !strcmp(s->pri_indev, s->sec_indev)) {
> + error_setg(errp, "'indev' and 'outdev' could not be same "
> + "for compare module");
> + return;
Can relax this by reusing the socket chardev? Looks we only use one
direction during packet passing.
> + }
> +
> + s->chr_pri_in = qemu_chr_find(s->pri_indev);
> + if (s->chr_pri_in == NULL) {
> + error_setg(errp, "Primary IN Device '%s' not found",
> + s->pri_indev);
> + return;
> + }
> +
> + s->chr_sec_in = qemu_chr_find(s->sec_indev);
> + if (s->chr_sec_in == NULL) {
> + error_setg(errp, "Secondary IN Device '%s' not found",
> + s->sec_indev);
> + return;
> + }
> +
> + s->chr_out = qemu_chr_find(s->outdev);
> + if (s->chr_out == NULL) {
> + error_setg(errp, "OUT Device '%s' not found", s->outdev);
> + return;
> + }
Like I point out in the past iterations, we probably need some
inspection here and fail if it was socket chardev. (Similar to what
vhost-use did for udp socket).
> +
> + qemu_chr_fe_claim_no_fail(s->chr_pri_in);
> +
> + qemu_chr_fe_claim_no_fail(s->chr_sec_in);
> +
> + qemu_chr_fe_claim_no_fail(s->chr_out);
> +
> + net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
> + net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
> +
Why chr_out is left here?
> + return;
> +}
> +
> +static void colo_compare_class_init(ObjectClass *oc, void *data)
> +{
> + UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
> +
> + ucc->complete = colo_compare_complete;
> +}
> +
> +static void colo_compare_init(Object *obj)
> +{
> + object_property_add_str(obj, "primary_in",
> + compare_get_pri_indev, compare_set_pri_indev,
> + NULL);
> + object_property_add_str(obj, "secondary_in",
> + compare_get_sec_indev, compare_set_sec_indev,
> + NULL);
> + object_property_add_str(obj, "outdev",
> + compare_get_outdev, compare_set_outdev,
> + NULL);
> +}
> +
> +static void colo_compare_finalize(Object *obj)
> +{
> + CompareState *s = COLO_COMPARE(obj);
> +
> + if (s->chr_pri_in) {
> + qemu_chr_add_handlers(s->chr_pri_in, NULL, NULL, NULL, NULL);
> + qemu_chr_fe_release(s->chr_pri_in);
> + }
> + if (s->chr_sec_in) {
> + qemu_chr_add_handlers(s->chr_sec_in, NULL, NULL, NULL, NULL);
> + qemu_chr_fe_release(s->chr_sec_in);
> + }
> + if (s->chr_out) {
> + qemu_chr_fe_release(s->chr_out);
> + }
> +
> + g_free(s->pri_indev);
> + g_free(s->sec_indev);
> + g_free(s->outdev);
> +}
> +
> +static const TypeInfo colo_compare_info = {
> + .name = TYPE_COLO_COMPARE,
> + .parent = TYPE_OBJECT,
> + .instance_size = sizeof(CompareState),
> + .instance_init = colo_compare_init,
> + .instance_finalize = colo_compare_finalize,
> + .class_size = sizeof(CompareClass),
> + .class_init = colo_compare_class_init,
> + .interfaces = (InterfaceInfo[]) {
> + { TYPE_USER_CREATABLE },
> + { }
> + }
> +};
> +
> +static void register_types(void)
> +{
> + type_register_static(&colo_compare_info);
> +}
> +
> +type_init(register_types);
> diff --git a/qemu-options.hx b/qemu-options.hx
> index 587de8f..79e5896 100644
> --- a/qemu-options.hx
> +++ b/qemu-options.hx
> @@ -3866,6 +3866,44 @@ Dump the network traffic on netdev @var{dev} to the file specified by
> The file format is libpcap, so it can be analyzed with tools such as tcpdump
> or Wireshark.
>
> +@item -object colo-compare,id=@var{id},primary_in=@var{chardevid},secondary_in=@var{chardevid},
> +outdev=@var{chardevid}
> +
> +Colo-compare gets packet from primary_in@var{chardevid} and secondary_in@var{chardevid},than compare primary packet with
> +secondary packet. if the packet same,we will output primary
s/if/If/ and miss a blank before ','
> +packet to outdev@var{chardevid},else we will notify colo-frame
miss a blank before ','
> +do checkpoint and send primary packet to outdev@var{chardevid}.
> +
> +we can use it with the help of filter-mirror and filter-redirector.
> +
> +The simple usage:
This line could be removed since we have example below.
> +
> +@example
> +
> +primary:
> +-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=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
> +-chardev socket,id=compare_out0,host=3.3.3.3,port=9005
> +-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
> +
> +secondary:
> +-netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down script=/etc/qemu-ifdown
> +-device e1000,netdev=hn0,mac=52:a4:00:12:78:66
> +-chardev socket,id=red0,host=3.3.3.3,port=9003
> +-chardev socket,id=red1,host=3.3.3.3,port=9004
> +-object filter-redirector,id=f1,netdev=hn0,queue=tx,indev=red0
> +-object filter-redirector,id=f2,netdev=hn0,queue=rx,outdev=red1
> +
> +@end example
> +
Better add detail description on the above command line since it was
rather complex for starters.
> @item -object secret,id=@var{id},data=@var{string},format=@var{raw|base64}[,keyid=@var{secretid},iv=@var{string}]
> @item -object secret,id=@var{id},file=@var{filename},format=@var{raw|base64}[,keyid=@var{secretid},iv=@var{string}]
>
> diff --git a/vl.c b/vl.c
> index cbe51ac..c6b9a6f 100644
> --- a/vl.c
> +++ b/vl.c
> @@ -2865,7 +2865,8 @@ static bool object_create_initial(const char *type)
> if (g_str_equal(type, "filter-buffer") ||
> g_str_equal(type, "filter-dump") ||
> g_str_equal(type, "filter-mirror") ||
> - g_str_equal(type, "filter-redirector")) {
> + g_str_equal(type, "filter-redirector") ||
> + g_str_equal(type, "colo-compare")) {
> return false;
> }
>
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet Zhang Chen
@ 2016-08-02 6:38 ` Jason Wang
2016-08-03 6:34 ` Zhang Chen
0 siblings, 1 reply; 20+ messages in thread
From: Jason Wang @ 2016-08-02 6:38 UTC (permalink / raw)
To: Zhang Chen, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 2016年07月26日 09:49, Zhang Chen wrote:
> COLO-base used by colo-compare and filter-rewriter.
s/used/is used/
> this can share common data structure like:net packet,
s/this/This/ and ':' looks unnecessary.
> and share other functions.
s/share//
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
> net/Makefile.objs | 1 +
> net/colo-base.c | 74 +++++++++++++++++++++++++++++++++
> net/colo-base.h | 38 +++++++++++++++++
> net/colo-compare.c | 119 ++++++++++++++++++++++++++++++++++++++++++++++++++++-
> trace-events | 3 ++
> 5 files changed, 233 insertions(+), 2 deletions(-)
> create mode 100644 net/colo-base.c
> create mode 100644 net/colo-base.h
>
> diff --git a/net/Makefile.objs b/net/Makefile.objs
> index ba92f73..119589f 100644
> --- a/net/Makefile.objs
> +++ b/net/Makefile.objs
> @@ -17,3 +17,4 @@ common-obj-y += filter.o
> common-obj-y += filter-buffer.o
> common-obj-y += filter-mirror.o
> common-obj-y += colo-compare.o
> +common-obj-y += colo-base.o
> diff --git a/net/colo-base.c b/net/colo-base.c
> new file mode 100644
> index 0000000..f5d5de9
> --- /dev/null
> +++ b/net/colo-base.c
I was thinking maybe just use colo.c is better here.
> @@ -0,0 +1,74 @@
> +/*
> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
> + * (a.k.a. Fault Tolerance or Continuous Replication)
> + *
> + * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
> + * Copyright (c) 2016 FUJITSU LIMITED
> + * Copyright (c) 2016 Intel Corporation
> + *
> + * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or
> + * later. See the COPYING file in the top-level directory.
> + */
> +
> +#include "qemu/osdep.h"
> +#include "qemu/error-report.h"
> +#include "net/colo-base.h"
> +
> +int parse_packet_early(Packet *pkt)
> +{
> + int network_length;
> + uint8_t *data = pkt->data;
> + uint16_t l3_proto;
> + ssize_t l2hdr_len = eth_get_l2_hdr_length(data);
> +
> + if (pkt->size < ETH_HLEN) {
> + error_report("pkt->size < ETH_HLEN");
> + return 1;
> + }
> + pkt->network_layer = data + ETH_HLEN;
> + l3_proto = eth_get_l3_proto(data, l2hdr_len);
> + if (l3_proto != ETH_P_IP) {
> + return 1;
> + }
> +
> + network_length = pkt->ip->ip_hl * 4;
> + if (pkt->size < ETH_HLEN + network_length) {
> + error_report("pkt->size < network_layer + network_length");
> + return 1;
> + }
> + pkt->transport_layer = pkt->network_layer + network_length;
> + if (!pkt->transport_layer) {
Looks like at least pkt->network_layer can't be zero, so I wonder how
this condition were met.
> + error_report("pkt->transport_layer is valid");
> + return 1;
> + }
> +
> + return 0;
> +}
> +
> +Packet *packet_new(const void *data, int size)
> +{
> + Packet *pkt = g_slice_new(Packet);
> +
> + pkt->data = g_memdup(data, size);
> + pkt->size = size;
> +
> + return pkt;
> +}
> +
> +void packet_destroy(void *opaque, void *user_data)
What's the reason of using void * instead of Packet * directly? And
user_data were not used in this function.
> +{
> + Packet *pkt = opaque;
> +
> + g_free(pkt->data);
> + g_slice_free(Packet, pkt);
> +}
> +
> +/*
> + * Clear hashtable, stop this hash growing really huge
> + */
> +void connection_hashtable_reset(GHashTable *connection_track_table)
> +{
> + g_hash_table_remove_all(connection_track_table);
> +}
> diff --git a/net/colo-base.h b/net/colo-base.h
> new file mode 100644
> index 0000000..48835e7
> --- /dev/null
> +++ b/net/colo-base.h
> @@ -0,0 +1,38 @@
> +/*
> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
> + * (a.k.a. Fault Tolerance or Continuous Replication)
> + *
> + * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
> + * Copyright (c) 2016 FUJITSU LIMITED
> + * Copyright (c) 2016 Intel Corporation
> + *
> + * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> + *
> + * This work is licensed under the terms of the GNU GPL, version 2 or
> + * later. See the COPYING file in the top-level directory.
> + */
> +
> +#ifndef QEMU_COLO_BASE_H
> +#define QEMU_COLO_BASE_H
> +
> +#include "slirp/slirp.h"
> +#include "qemu/jhash.h"
> +
> +#define HASHTABLE_MAX_SIZE 16384
> +
> +typedef struct Packet {
> + void *data;
> + union {
> + uint8_t *network_layer;
I think it would be better to use "network_header" here.
> + struct ip *ip;
> + };
> + uint8_t *transport_layer;
And "transport_header" here.
And why not union tcp header here? I'm asking since ip header were
unioned above.
> + int size;
> +} Packet;
> +
> +int parse_packet_early(Packet *pkt);
> +void connection_hashtable_reset(GHashTable *connection_track_table);
> +Packet *packet_new(const void *data, int size);
> +void packet_destroy(void *opaque, void *user_data);
> +
> +#endif /* QEMU_COLO_BASE_H */
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index 0402958..7c52cc8 100644
> --- a/net/colo-compare.c
> +++ b/net/colo-compare.c
> @@ -27,13 +27,38 @@
> #include "sysemu/char.h"
> #include "qemu/sockets.h"
> #include "qapi-visit.h"
> +#include "net/colo-base.h"
> +#include "trace.h"
>
> #define TYPE_COLO_COMPARE "colo-compare"
> #define COLO_COMPARE(obj) \
> OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
>
> #define COMPARE_READ_LEN_MAX NET_BUFSIZE
> +#define MAX_QUEUE_SIZE 1024
>
> +/*
> + + CompareState ++
> + | |
> + +---------------+ +---------------+ +---------------+
> + |conn list +--->conn +--------->conn |
> + +---------------+ +---------------+ +---------------+
> + | | | | | |
> + +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
> + |primary | |secondary |primary | |secondary
> + |packet | |packet + |packet | |packet +
> + +--------+ +--------+ +--------+ +--------+
> + | | | |
> + +---v----+ +---v----+ +---v----+ +---v----+
> + |primary | |secondary |primary | |secondary
> + |packet | |packet + |packet | |packet +
> + +--------+ +--------+ +--------+ +--------+
> + | | | |
> + +---v----+ +---v----+ +---v----+ +---v----+
> + |primary | |secondary |primary | |secondary
> + |packet | |packet + |packet | |packet +
> + +--------+ +--------+ +--------+ +--------+
> +*/
> typedef struct CompareState {
> Object parent;
>
> @@ -46,12 +71,89 @@ typedef struct CompareState {
> QTAILQ_ENTRY(CompareState) next;
> SocketReadState pri_rs;
> SocketReadState sec_rs;
> +
> + /* hashtable to save connection */
> + GHashTable *connection_track_table;
> + /* to save unprocessed_connections */
> + GQueue unprocessed_connections;
> + /* proxy current hash size */
> + uint32_t hashtable_size;
> } CompareState;
>
> typedef struct CompareClass {
> ObjectClass parent_class;
> } CompareClass;
>
> +enum {
> + PRIMARY_IN = 0,
> + SECONDARY_IN,
> +};
> +
> +static int compare_chr_send(CharDriverState *out,
> + const uint8_t *buf,
> + uint32_t size);
> +
> +/*
> + * Return 0 on success, if return -1 means the pkt
> + * is unsupported(arp and ipv6) and will be sent later
> + */
> +static int packet_enqueue(CompareState *s, int mode)
> +{
> + Packet *pkt = NULL;
> +
> + if (mode == PRIMARY_IN) {
> + pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
> + } else {
> + pkt = packet_new(s->sec_rs.buf, s->sec_rs.packet_len);
> + }
> +
> + if (parse_packet_early(pkt)) {
> + packet_destroy(pkt, NULL);
> + pkt = NULL;
> + return -1;
> + }
> + /* TODO: get connection key from pkt */
> +
> + /*
> + * TODO: use connection key get conn from
> + * connection_track_table
> + */
> +
> + /*
> + * TODO: insert pkt to it's conn->primary_list
> + * or conn->secondary_list
> + */
> +
> + return 0;
> +}
> +
> +static int compare_chr_send(CharDriverState *out,
> + const uint8_t *buf,
> + uint32_t size)
> +{
> + int ret = 0;
> + uint32_t len = htonl(size);
> +
> + if (!size) {
> + return 0;
> + }
> +
> + ret = qemu_chr_fe_write_all(out, (uint8_t *)&len, sizeof(len));
> + if (ret != sizeof(len)) {
> + goto err;
> + }
> +
> + ret = qemu_chr_fe_write_all(out, (uint8_t *)buf, size);
> + if (ret != size) {
> + goto err;
> + }
> +
> + return 0;
> +
> +err:
> + return ret < 0 ? ret : -EIO;
> +}
> +
> static char *compare_get_pri_indev(Object *obj, Error **errp)
> {
> CompareState *s = COLO_COMPARE(obj);
> @@ -99,12 +201,21 @@ static void compare_set_outdev(Object *obj, const char *value, Error **errp)
>
> static void compare_pri_rs_finalize(SocketReadState *pri_rs)
> {
> - /* if packet_enqueue pri pkt failed we will send unsupported packet */
> + CompareState *s = container_of(pri_rs, CompareState, 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);
> + }
> }
>
> static void compare_sec_rs_finalize(SocketReadState *sec_rs)
> {
> - /* if packet_enqueue sec pkt failed we will notify trace */
> + CompareState *s = container_of(sec_rs, CompareState, sec_rs);
> +
> + if (packet_enqueue(s, SECONDARY_IN)) {
> + trace_colo_compare_main("secondary: unsupported packet in");
> + }
> }
>
> /*
> @@ -156,6 +267,10 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
> net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
> net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
>
> + s->hashtable_size = 0;
> +
> + /* use g_hash_table_new_full() to new a hashtable */
> +
> return;
> }
>
> diff --git a/trace-events b/trace-events
> index ca7211b..703de1a 100644
> --- a/trace-events
> +++ b/trace-events
> @@ -1916,3 +1916,6 @@ aspeed_vic_update_fiq(int flags) "Raising FIQ: %d"
> aspeed_vic_update_irq(int flags) "Raising IRQ: %d"
> aspeed_vic_read(uint64_t offset, unsigned size, uint32_t value) "From 0x%" PRIx64 " of size %u: 0x%" PRIx32
> aspeed_vic_write(uint64_t offset, unsigned size, uint32_t data) "To 0x%" PRIx64 " of size %u: 0x%" PRIx32
> +
> +# net/colo-compare.c
> +colo_compare_main(const char *chr) ": %s"
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu Zhang Chen
@ 2016-08-02 6:40 ` Jason Wang
2016-08-03 6:36 ` Zhang Chen
0 siblings, 1 reply; 20+ messages in thread
From: Jason Wang @ 2016-08-02 6:40 UTC (permalink / raw)
To: Zhang Chen, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 2016年07月26日 09:49, Zhang Chen wrote:
> Jhash used by colo-compare and filter-rewriter
> to save and lookup net connection info
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
> include/qemu/jhash.h | 61 ++++++++++++++++++++++++++++++++++++++++++++++++++++
> 1 file changed, 61 insertions(+)
> create mode 100644 include/qemu/jhash.h
>
> diff --git a/include/qemu/jhash.h b/include/qemu/jhash.h
> new file mode 100644
> index 0000000..0fcd875
> --- /dev/null
> +++ b/include/qemu/jhash.h
> @@ -0,0 +1,61 @@
> +/* jhash.h: Jenkins hash support.
> + *
> + * Copyright (C) 2006. Bob Jenkins (bob_jenkins@burtleburtle.net)
> + *
> + * http://burtleburtle.net/bob/hash/
> + *
> + * These are the credits from Bob's sources:
> + *
> + * lookup3.c, by Bob Jenkins, May 2006, Public Domain.
> + *
> + * These are functions for producing 32-bit hashes for hash table lookup.
> + * hashword(), hashlittle(), hashlittle2(), hashbig(), mix(), and final()
> + * are externally useful functions. Routines to test the hash are
> +included
> + * if SELF_TEST is defined. You can use this free for any purpose.
> +It's in
The indentation looks odd here, please check.
> + * the public domain. It has no warranty.
> + *
> + * Copyright (C) 2009-2010 Jozsef Kadlecsik (kadlec@blackhole.kfki.hu)
> + *
> + * I've modified Bob's hash to be useful in the Linux kernel, and
> + * any bugs present are my fault.
> + * Jozsef
> + */
> +
> +#ifndef QEMU_JHASH_H__
> +#define QEMU_JHASH_H__
> +
> +#include "qemu/bitops.h"
> +
> +/*
> + * hashtable relation copy from linux kernel jhash
> + */
> +
> +/* __jhash_mix -- mix 3 32-bit values reversibly. */
> +#define __jhash_mix(a, b, c) \
> +{ \
> + a -= c; a ^= rol32(c, 4); c += b; \
> + b -= a; b ^= rol32(a, 6); a += c; \
> + c -= b; c ^= rol32(b, 8); b += a; \
> + a -= c; a ^= rol32(c, 16); c += b; \
> + b -= a; b ^= rol32(a, 19); a += c; \
> + c -= b; c ^= rol32(b, 4); b += a; \
> +}
> +
> +/* __jhash_final - final mixing of 3 32-bit values (a,b,c) into c */
> +#define __jhash_final(a, b, c) \
> +{ \
> + c ^= b; c -= rol32(b, 14); \
> + a ^= c; a -= rol32(c, 11); \
> + b ^= a; b -= rol32(a, 25); \
> + c ^= b; c -= rol32(b, 16); \
> + a ^= c; a -= rol32(c, 4); \
> + b ^= a; b -= rol32(a, 14); \
> + c ^= b; c -= rol32(b, 24); \
> +}
> +
> +/* An arbitrary initial parameter */
> +#define JHASH_INITVAL 0xdeadbeef
> +
> +#endif /* QEMU_JHASH_H__ */
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet Zhang Chen
@ 2016-08-02 7:14 ` Jason Wang
2016-08-04 6:49 ` Zhang Chen
0 siblings, 1 reply; 20+ messages in thread
From: Jason Wang @ 2016-08-02 7:14 UTC (permalink / raw)
To: Zhang Chen, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 2016年07月26日 09:49, Zhang Chen wrote:
> In this patch we use kernel jhash table to track
> connection, and then enqueue net packet like this:
>
> + CompareState ++
> | |
> +---------------+ +---------------+ +---------------+
> |conn list +--->conn +--------->conn |
> +---------------+ +---------------+ +---------------+
> | | | | | |
> +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
> |primary | |secondary |primary | |secondary
> |packet | |packet + |packet | |packet +
> +--------+ +--------+ +--------+ +--------+
> | | | |
> +---v----+ +---v----+ +---v----+ +---v----+
> |primary | |secondary |primary | |secondary
> |packet | |packet + |packet | |packet +
> +--------+ +--------+ +--------+ +--------+
> | | | |
> +---v----+ +---v----+ +---v----+ +---v----+
> |primary | |secondary |primary | |secondary
> |packet | |packet + |packet | |packet +
> +--------+ +--------+ +--------+ +--------+
>
> We use conn_list to record connection info.
> When we want to enqueue a packet, firstly get the
> connection from connection_track_table. then push
> the packet to g_queue(pri/sec) in it's own conn.
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
> net/colo-base.c | 108 +++++++++++++++++++++++++++++++++++++++++++++++++++++
> net/colo-base.h | 30 +++++++++++++++
> net/colo-compare.c | 70 +++++++++++++++++++++++++++++-----
> 3 files changed, 198 insertions(+), 10 deletions(-)
>
> diff --git a/net/colo-base.c b/net/colo-base.c
> index f5d5de9..7e91dec 100644
> --- a/net/colo-base.c
> +++ b/net/colo-base.c
> @@ -16,6 +16,29 @@
> #include "qemu/error-report.h"
> #include "net/colo-base.h"
>
> +uint32_t connection_key_hash(const void *opaque)
> +{
> + const ConnectionKey *key = opaque;
> + uint32_t a, b, c;
> +
> + /* Jenkins hash */
> + a = b = c = JHASH_INITVAL + sizeof(*key);
> + a += key->src.s_addr;
> + b += key->dst.s_addr;
> + c += (key->src_port | key->dst_port << 16);
> + __jhash_mix(a, b, c);
> +
> + a += key->ip_proto;
> + __jhash_final(a, b, c);
> +
> + return c;
> +}
> +
> +int connection_key_equal(const void *key1, const void *key2)
> +{
> + return memcmp(key1, key2, sizeof(ConnectionKey)) == 0;
> +}
> +
> int parse_packet_early(Packet *pkt)
> {
> int network_length;
> @@ -47,6 +70,62 @@ int parse_packet_early(Packet *pkt)
> return 0;
> }
>
> +void fill_connection_key(Packet *pkt, ConnectionKey *key)
> +{
> + uint32_t tmp_ports;
> +
> + key->ip_proto = pkt->ip->ip_p;
> +
> + switch (key->ip_proto) {
> + case IPPROTO_TCP:
> + case IPPROTO_UDP:
> + case IPPROTO_DCCP:
> + case IPPROTO_ESP:
> + case IPPROTO_SCTP:
> + case IPPROTO_UDPLITE:
> + tmp_ports = *(uint32_t *)(pkt->transport_layer);
> + key->src = pkt->ip->ip_src;
> + key->dst = pkt->ip->ip_dst;
> + key->src_port = ntohs(tmp_ports & 0xffff);
> + key->dst_port = ntohs(tmp_ports >> 16);
> + break;
> + case IPPROTO_AH:
> + tmp_ports = *(uint32_t *)(pkt->transport_layer + 4);
> + key->src = pkt->ip->ip_src;
> + key->dst = pkt->ip->ip_dst;
> + key->src_port = ntohs(tmp_ports & 0xffff);
> + key->dst_port = ntohs(tmp_ports >> 16);
> + break;
> + default:
> + key->src_port = 0;
> + key->dst_port = 0;
> + break;
> + }
> +}
> +
> +Connection *connection_new(ConnectionKey *key)
> +{
> + Connection *conn = g_slice_new(Connection);
> +
> + conn->ip_proto = key->ip_proto;
> + conn->processing = false;
> + g_queue_init(&conn->primary_list);
> + g_queue_init(&conn->secondary_list);
> +
> + return conn;
> +}
> +
> +void connection_destroy(void *opaque)
> +{
> + Connection *conn = opaque;
> +
> + g_queue_foreach(&conn->primary_list, packet_destroy, NULL);
> + g_queue_free(&conn->primary_list);
> + g_queue_foreach(&conn->secondary_list, packet_destroy, NULL);
> + g_queue_free(&conn->secondary_list);
> + g_slice_free(Connection, conn);
> +}
> +
> Packet *packet_new(const void *data, int size)
> {
> Packet *pkt = g_slice_new(Packet);
> @@ -72,3 +151,32 @@ void connection_hashtable_reset(GHashTable *connection_track_table)
> {
> g_hash_table_remove_all(connection_track_table);
> }
> +
> +/* if not found, create a new connection and add to hash table */
> +Connection *connection_get(GHashTable *connection_track_table,
> + ConnectionKey *key,
> + uint32_t *hashtable_size)
> +{
> + Connection *conn = g_hash_table_lookup(connection_track_table, key);
> +
> + if (conn == NULL) {
> + ConnectionKey *new_key = g_memdup(key, sizeof(*key));
> +
> + conn = connection_new(key);
> +
> + (*hashtable_size) += 1;
> + if (*hashtable_size > HASHTABLE_MAX_SIZE) {
> + error_report("colo proxy connection hashtable full, clear it");
> + connection_hashtable_reset(connection_track_table);
> + /*
> + * when hashtable_size == 0, clear the conn_list
> + * in place where be called.
> + */
Does this mean it requires the caller to do this? If yes, seems not
good, why not simply do things here?
> + *hashtable_size = 0;
> + }
> +
> + g_hash_table_insert(connection_track_table, new_key, conn);
Then we lose the track of *hashtable_size here. It should be 1 but we
set it to zero if we are out of space.
> + }
> +
> + return conn;
> +}
> diff --git a/net/colo-base.h b/net/colo-base.h
> index 48835e7..0505608 100644
> --- a/net/colo-base.h
> +++ b/net/colo-base.h
> @@ -30,7 +30,37 @@ typedef struct Packet {
> int size;
> } Packet;
>
> +typedef struct ConnectionKey {
> + /* (src, dst) must be grouped, in the same way than in IP header */
> + struct in_addr src;
> + struct in_addr dst;
> + uint16_t src_port;
> + uint16_t dst_port;
> + uint8_t ip_proto;
> +} QEMU_PACKED ConnectionKey;
> +
> +typedef struct Connection {
> + /* connection primary send queue: element type: Packet */
> + GQueue primary_list;
> + /* connection secondary send queue: element type: Packet */
> + GQueue secondary_list;
> + /* flag to enqueue unprocessed_connections */
> + bool processing;
> + uint8_t ip_proto;
> + /* be used by filter-rewriter */
> + tcp_seq primary_seq;
> + tcp_seq secondary_seq;
> +} Connection;
> +
> +uint32_t connection_key_hash(const void *opaque);
> +int connection_key_equal(const void *opaque1, const void *opaque2);
> int parse_packet_early(Packet *pkt);
> +void fill_connection_key(Packet *pkt, ConnectionKey *key);
> +Connection *connection_new(ConnectionKey *key);
> +void connection_destroy(void *opaque);
> +Connection *connection_get(GHashTable *connection_track_table,
> + ConnectionKey *key,
> + uint32_t *hashtable_size);
> void connection_hashtable_reset(GHashTable *connection_track_table);
> Packet *packet_new(const void *data, int size);
> void packet_destroy(void *opaque, void *user_data);
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index 7c52cc8..5f87710 100644
> --- a/net/colo-compare.c
> +++ b/net/colo-compare.c
> @@ -72,6 +72,11 @@ typedef struct CompareState {
> SocketReadState pri_rs;
> SocketReadState sec_rs;
>
> + /* connection list: the connections belonged to this NIC could be found
> + * in this list.
> + * element type: Connection
> + */
> + GQueue conn_list;
> /* hashtable to save connection */
> GHashTable *connection_track_table;
> /* to save unprocessed_connections */
> @@ -93,13 +98,30 @@ static int compare_chr_send(CharDriverState *out,
> const uint8_t *buf,
> uint32_t size);
>
> +static void colo_rm_connection(void *opaque, void *user_data)
> +{
> + Connection *conn = opaque;
> + Packet *pkt = NULL;
> +
> + while (!g_queue_is_empty(&conn->primary_list)) {
> + pkt = g_queue_pop_head(&conn->primary_list);
> + packet_destroy(pkt, NULL);
> + }
> + while (!g_queue_is_empty(&conn->secondary_list)) {
> + pkt = g_queue_pop_head(&conn->secondary_list);
> + packet_destroy(pkt, NULL);
> + }
This looks rather similar to connection_destroy(), can we share some codes?
> +}
> +
> /*
> * Return 0 on success, if return -1 means the pkt
> * is unsupported(arp and ipv6) and will be sent later
> */
> static int packet_enqueue(CompareState *s, int mode)
> {
> + ConnectionKey key = {{ 0 } };
> Packet *pkt = NULL;
> + Connection *conn;
>
> if (mode == PRIMARY_IN) {
> pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
> @@ -112,17 +134,38 @@ static int packet_enqueue(CompareState *s, int mode)
> pkt = NULL;
> return -1;
> }
> - /* TODO: get connection key from pkt */
> + fill_connection_key(pkt, &key);
>
> - /*
> - * TODO: use connection key get conn from
> - * connection_track_table
> - */
> + conn = connection_get(s->connection_track_table,
> + &key,
> + &s->hashtable_size);
>
> - /*
> - * TODO: insert pkt to it's conn->primary_list
> - * or conn->secondary_list
> - */
> + if (!s->hashtable_size) {
> + g_queue_foreach(&s->conn_list, colo_rm_connection, NULL);
> + }
> +
> + if (!conn->processing) {
> + g_queue_push_tail(&s->conn_list, conn);
> + conn->processing = true;
Why not simply do this in connection_get(), you can save a
conn->processing flag.
> + }
> +
> + if (mode == PRIMARY_IN) {
> + if (g_queue_get_length(&conn->primary_list) <
> + MAX_QUEUE_SIZE) {
> + g_queue_push_tail(&conn->primary_list, pkt);
> + } else {
> + error_report("colo compare primary queue size too big,"
> + "drop packet");
I don't see how packet were dropped in this case?
> + }
> + } else {
> + if (g_queue_get_length(&conn->secondary_list) <
> + MAX_QUEUE_SIZE) {
> + g_queue_push_tail(&conn->secondary_list, pkt);
> + } else {
> + error_report("colo compare secondary queue size too big,"
> + "drop packet");
> + }
> + }
>
> return 0;
> }
> @@ -267,9 +310,14 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
> net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
> net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
>
> + g_queue_init(&s->conn_list);
> +
> s->hashtable_size = 0;
>
> - /* use g_hash_table_new_full() to new a hashtable */
> + s->connection_track_table = g_hash_table_new_full(connection_key_hash,
> + connection_key_equal,
> + g_free,
> + connection_destroy);
>
> return;
> }
> @@ -310,6 +358,8 @@ static void colo_compare_finalize(Object *obj)
> qemu_chr_fe_release(s->chr_out);
> }
>
> + g_queue_free(&s->conn_list);
> +
> g_free(s->pri_indev);
> g_free(s->sec_indev);
> g_free(s->outdev);
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread Zhang Chen
@ 2016-08-02 7:52 ` Jason Wang
2016-08-05 7:45 ` Zhang Chen
0 siblings, 1 reply; 20+ messages in thread
From: Jason Wang @ 2016-08-02 7:52 UTC (permalink / raw)
To: Zhang Chen, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
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/
> after REGULAR_PACKET_CHECK_MS milliseconds we set
> the primary packet as old_packet,then do a checkpoint.
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
> 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.
> + 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?
> + 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?
> + } 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?
> + 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?
> + 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.
> +
> 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
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison Zhang Chen
@ 2016-08-02 8:04 ` Jason Wang
2016-08-05 8:55 ` Zhang Chen
0 siblings, 1 reply; 20+ messages in thread
From: Jason Wang @ 2016-08-02 8:04 UTC (permalink / raw)
To: Zhang Chen, qemu devel
Cc: Li Zhijian, eddie . dong, Dr . David Alan Gilbert, zhanghailiang
On 2016年07月26日 09:49, Zhang Chen wrote:
> We add TCP,UDP,ICMP packet comparison to replace
> IP packet comparison. This can increase the
> accuracy of the package comparison.
> less checkpoint more efficiency.
s/less/Less/
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
> net/colo-compare.c | 174 +++++++++++++++++++++++++++++++++++++++++++++++++++--
> trace-events | 4 ++
> 2 files changed, 174 insertions(+), 4 deletions(-)
>
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index e020edc..c7bb5f7 100644
> --- a/net/colo-compare.c
> +++ b/net/colo-compare.c
> @@ -18,6 +18,7 @@
> #include "qapi/qmp/qerror.h"
> #include "qapi/error.h"
> #include "net/net.h"
> +#include "net/eth.h"
> #include "net/vhost_net.h"
> #include "qom/object_interfaces.h"
> #include "qemu/iov.h"
> @@ -197,9 +198,158 @@ static int colo_packet_compare(Packet *ppkt, Packet *spkt)
> }
> }
>
> -static int colo_packet_compare_all(Packet *spkt, Packet *ppkt)
> +/*
> + * called from the compare thread on the primary
> + * for compare tcp packet
> + * compare_tcp copied from Dr. David Alan Gilbert's branch
> + */
> +static int colo_packet_compare_tcp(Packet *spkt, Packet *ppkt)
> +{
> + struct tcphdr *ptcp, *stcp;
> + int res;
> + char *sdebug, *ddebug;
> +
> + trace_colo_compare_main("compare tcp");
> + if (ppkt->size != spkt->size) {
> + if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
> + trace_colo_compare_main("pkt size not same");
> + }
> + return -1;
> + }
> +
> + ptcp = (struct tcphdr *)ppkt->transport_layer;
> + stcp = (struct tcphdr *)spkt->transport_layer;
> +
> + if (ptcp->th_seq != stcp->th_seq) {
> + if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
> + trace_colo_compare_main("pkt tcp seq not same");
> + }
> + return -1;
> + }
> +
> + /*
> + * The 'identification' field in the IP header is *very* random
> + * it almost never matches. Fudge this by ignoring differences in
> + * unfragmented packets; they'll normally sort themselves out if different
> + * anyway, and it should recover at the TCP level.
> + * An alternative would be to get both the primary and secondary to rewrite
> + * somehow; but that would need some sync traffic to sync the state
> + */
> + if (ntohs(ppkt->ip->ip_off) & IP_DF) {
> + spkt->ip->ip_id = ppkt->ip->ip_id;
> + /* and the sum will be different if the IDs were different */
> + spkt->ip->ip_sum = ppkt->ip->ip_sum;
This looks dangerous. If packet were not logical same, ip cusm of
secondary were changed.
> + }
> +
> + res = memcmp(ppkt->data + ETH_HLEN, spkt->data + ETH_HLEN,
> + (spkt->size - ETH_HLEN));
> +
> + if (res != 0 && trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
> + sdebug = strdup(inet_ntoa(ppkt->ip->ip_src));
> + ddebug = strdup(inet_ntoa(ppkt->ip->ip_dst));
> + fprintf(stderr, "%s: src/dst: %s/%s p: seq/ack=%u/%u"
> + " s: seq/ack=%u/%u res=%d flags=%x/%x\n", __func__,
> + sdebug, ddebug,
> + ntohl(ptcp->th_seq), ntohl(ptcp->th_ack),
> + ntohl(stcp->th_seq), ntohl(stcp->th_ack),
> + res, ptcp->th_flags, stcp->th_flags);
> +
> + trace_colo_compare_tcp_miscompare("Primary len", ppkt->size);
> + qemu_hexdump((char *)ppkt->data, stderr, "colo-compare", ppkt->size);
> + trace_colo_compare_tcp_miscompare("Secondary len", spkt->size);
> + qemu_hexdump((char *)spkt->data, stderr, "colo-compare", spkt->size);
> +
> + g_free(sdebug);
> + g_free(ddebug);
> + }
> +
> + return res;
> +}
> +
> +/*
> + * called from the compare thread on the primary
> + * for compare udp packet
> + */
> +static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
> +{
> + int ret;
> +
> + trace_colo_compare_main("compare udp");
> + ret = colo_packet_compare(ppkt, spkt);
> +
> + if (ret) {
> + trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
> + qemu_hexdump((char *)ppkt->data, stderr, "colo-compare", ppkt->size);
> + trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
> + qemu_hexdump((char *)spkt->data, stderr, "colo-compare", spkt->size);
> + }
> +
> + return ret;
> +}
> +
> +/*
> + * called from the compare thread on the primary
> + * for compare icmp packet
> + */
> +static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
> {
Consider icmp packet were usually not big, why not simply use memcpy? (I
think I asked the same question in the past).
> - trace_colo_compare_main("compare all");
> + int network_length;
> + struct icmp *icmp_ppkt, *icmp_spkt;
> +
> + trace_colo_compare_main("compare icmp");
> + network_length = ppkt->ip->ip_hl * 4;
> + if (ppkt->size != spkt->size ||
> + ppkt->size < network_length + ETH_HLEN) {
> + return -1;
> + }
> + icmp_ppkt = (struct icmp *)(ppkt->data + network_length + ETH_HLEN);
> + icmp_spkt = (struct icmp *)(spkt->data + network_length + ETH_HLEN);
> +
> + if ((icmp_ppkt->icmp_type == icmp_spkt->icmp_type) &&
> + (icmp_ppkt->icmp_code == icmp_spkt->icmp_code)) {
> + if (icmp_ppkt->icmp_type == ICMP_REDIRECT) {
> + if (icmp_ppkt->icmp_gwaddr.s_addr !=
> + icmp_spkt->icmp_gwaddr.s_addr) {
> + trace_colo_compare_main("icmp_gwaddr.s_addr not same");
> + return -1;
> + }
> + } else if ((icmp_ppkt->icmp_type == ICMP_UNREACH) &&
> + (icmp_ppkt->icmp_type == ICMP_UNREACH_NEEDFRAG)) {
> + if (icmp_ppkt->icmp_nextmtu != icmp_spkt->icmp_nextmtu) {
> + trace_colo_compare_main("icmp_nextmtu not same");
> + return -1;
> + }
> + }
> +
> + if (colo_packet_compare(ppkt, spkt)) {
> + trace_colo_compare_icmp_miscompare("primary pkt size",
> + ppkt->size);
> + qemu_hexdump((char *)ppkt->data, stderr, "colo-compare",
> + ppkt->size);
> + trace_colo_compare_icmp_miscompare("Secondary pkt size",
> + spkt->size);
> + qemu_hexdump((char *)spkt->data, stderr, "colo-compare",
> + spkt->size);
> + return -1;
> + }
> + } else {
> + return -1;
> + }
> +
> + return 0;
> +}
> +
> +/*
> + * called from the compare thread on the primary
> + * for compare other packet
> + */
> +static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
> +{
> + trace_colo_compare_main("compare other");
> + 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));
> return colo_packet_compare(ppkt, spkt);
> }
>
> @@ -263,8 +413,24 @@ static void colo_compare_connection(void *opaque, void *user_data)
> 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);
> + switch (conn->ip_proto) {
> + case IPPROTO_TCP:
> + result = g_queue_find_custom(&conn->secondary_list,
> + pkt, (GCompareFunc)colo_packet_compare_tcp);
> + break;
> + case IPPROTO_UDP:
> + result = g_queue_find_custom(&conn->secondary_list,
> + pkt, (GCompareFunc)colo_packet_compare_udp);
> + break;
> + case IPPROTO_ICMP:
> + result = g_queue_find_custom(&conn->secondary_list,
> + pkt, (GCompareFunc)colo_packet_compare_icmp);
> + break;
> + default:
> + result = g_queue_find_custom(&conn->secondary_list,
> + pkt, (GCompareFunc)colo_packet_compare_other);
> + break;
> + }
>
> if (result) {
> ret = compare_chr_send(s->chr_out, pkt->data, pkt->size);
> diff --git a/trace-events b/trace-events
> index 1537e91..ab22eb2 100644
> --- a/trace-events
> +++ b/trace-events
> @@ -1919,5 +1919,9 @@ 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_tcp_miscompare(const char *sta, int size) ": %s = %d"
> +colo_compare_udp_miscompare(const char *sta, int size) ": %s = %d"
> +colo_compare_icmp_miscompare(const char *sta, int size) ": %s = %d"
> 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
> +colo_compare_miscompare(void) ""
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization
2016-08-02 6:26 ` Jason Wang
@ 2016-08-03 2:51 ` Zhang Chen
0 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-08-03 2:51 UTC (permalink / raw)
To: Jason Wang, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 08/02/2016 02:26 PM, Jason Wang wrote:
>
>
> On 2016年07月26日 09:49, Zhang Chen wrote:
>> This a COLO net ascii figure:
>>
>> Primary qemu Secondary qemu
>> +----------------------------------------------------------+
>> +----------------------------------------------------------------+
>> | +-----------------------------------------------------+ |
>> | +-----------------------------------------------------------+ |
>> | | | |
>> | | | |
>> | | guest | |
>> | | guest | |
>> | | | |
>> | | | |
>> | +-------^------------------+--------------------------+ |
>> | +---------------------+--------+----------------------------+ |
>> | | | | | ^
>> | |
>> | | | | | |
>> | |
>> | | +------------------------------------------------------+
>> | | | |
>> |netfilter| | | | | | netfilter |
>> | |
>> | +----------+ -----------------------+ | | |
>> +-----------------------------------------------------------+ |
>> | | | | | | | | |
>> | | | filter excute order | |
>> | | | | | | | | |
>> | | | +-------------------> | |
>> | | | | | | | | |
>> | | | TCP | |
>> | | +-----+--+--+ +------v-----+ | +------------+ | | |
>> | +------------+ +---+----+---v+rewriter++ +------------+ | |
>> | | | | | | | | | | | |
>> | | | | | | | | | |
>> | | | filter | | filter +----> colo <--------+
>> +--------> filter +--> adjust | adjust +--> filter | | |
>> | | | mirror | | redirector | | | compare | | |
>> | | | redirector | | ack | seq | | redirector | | |
>> | | | | | | | | | | |
>> | | | | | | | | | | |
>> | | +----^------+ +------------+ | +-----+------+ | |
>> | | +------------+ +--------+--------------+ +---+--------+ | |
>> | | | tx rx | | | |
>> | | tx all | rx | |
>> | | | | | | |
>> | +-----------------------------------------------------------+ |
>> | | | | | | |
>> | | |
>> | | | filter excute order | | | |
>> | | |
>> | | | +-------------------> | | |
>> +------------------------------------------------------------+ |
>> | +-----------------------------------+ | | | |
>> | | | | | |
>> +----------------------------------------------------------+
>> +----------------------------------------------------------------+
>> |guest receive |guest send
>> | |
>> +--------+------------------------------------v------------+
>> | | NOTE: filter direction is rx/tx/all
>> | tap |
>> rx:receive packets sent to the netdev
>> | | tx:receive packets sent by the netdev
>> +----------------------------------------------------------+
>>
>> In COLO-compare.
>> Packets coming from the primary char indev will be sent to outdev
>> Packets coming from the secondary char dev will be dropped
>> colo-comapre need two input chardev and one output chardev:
>> primary_in=chardev1-id
>> secondary_in=chardev2-id
>> outdev=chardev3-id
>
> Though it has 'compare' in its name, this description needs some key
> information still. e.g what it did (packet comparing). And then you
> can describe where were the data sources from.
OK, I will add some comments for this.
>
>> usage:
>>
>> primary:
>> -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=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
>> -chardev socket,id=compare_out0,host=3.3.3.3,port=9005
>> -object filter-mirror,id=m0,netdev=hn0,queue=tx,outdev=mirror0
>> -object
>> filter-redirector,netdev=hn0,id=redire0,queue=rx,indev=compare_out
>
> Looks like this filter were missed in the above figure.
>
Yes,I will update figure in next version.
>> -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
>>
>> secondary:
>> -netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down
>> script=/etc/qemu-ifdown
>> -device e1000,netdev=hn0,mac=52:a4:00:12:78:66
>> -chardev socket,id=red0,host=3.3.3.3,port=9003
>> -chardev socket,id=red1,host=3.3.3.3,port=9004
>> -object filter-redirector,id=f1,netdev=hn0,queue=tx,indev=red0
>> -object filter-redirector,id=f2,netdev=hn0,queue=rx,outdev=red1
>>
>> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
>> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
>> ---
>> net/Makefile.objs | 1 +
>> net/colo-compare.c | 222
>> +++++++++++++++++++++++++++++++++++++++++++++++++++++
>> qemu-options.hx | 38 +++++++++
>> vl.c | 3 +-
>> 4 files changed, 263 insertions(+), 1 deletion(-)
>> create mode 100644 net/colo-compare.c
>>
>> diff --git a/net/Makefile.objs b/net/Makefile.objs
>> index b7c22fd..ba92f73 100644
>> --- a/net/Makefile.objs
>> +++ b/net/Makefile.objs
>> @@ -16,3 +16,4 @@ common-obj-$(CONFIG_NETMAP) += netmap.o
>> common-obj-y += filter.o
>> common-obj-y += filter-buffer.o
>> common-obj-y += filter-mirror.o
>> +common-obj-y += colo-compare.o
>> diff --git a/net/colo-compare.c b/net/colo-compare.c
>> new file mode 100644
>> index 0000000..0402958
>> --- /dev/null
>> +++ b/net/colo-compare.c
>> @@ -0,0 +1,222 @@
>> +/*
>> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service
>> (COLO)
>> + * (a.k.a. Fault Tolerance or Continuous Replication)
>> + *
>> + * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
>> + * Copyright (c) 2016 FUJITSU LIMITED
>> + * Copyright (c) 2016 Intel Corporation
>> + *
>> + * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2 or
>> + * later. See the COPYING file in the top-level directory.
>> + */
>> +
>> +#include "qemu/osdep.h"
>> +#include "qemu/error-report.h"
>> +#include "qemu-common.h"
>> +#include "qapi/qmp/qerror.h"
>> +#include "qapi/error.h"
>> +#include "net/net.h"
>> +#include "net/vhost_net.h"
>> +#include "qom/object_interfaces.h"
>> +#include "qemu/iov.h"
>> +#include "qom/object.h"
>> +#include "qemu/typedefs.h"
>> +#include "net/queue.h"
>> +#include "sysemu/char.h"
>> +#include "qemu/sockets.h"
>> +#include "qapi-visit.h"
>> +
>> +#define TYPE_COLO_COMPARE "colo-compare"
>> +#define COLO_COMPARE(obj) \
>> + OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
>> +
>> +#define COMPARE_READ_LEN_MAX NET_BUFSIZE
>> +
>> +typedef struct CompareState {
>> + Object parent;
>> +
>> + char *pri_indev;
>> + char *sec_indev;
>> + char *outdev;
>> + CharDriverState *chr_pri_in;
>> + CharDriverState *chr_sec_in;
>> + CharDriverState *chr_out;
>> + QTAILQ_ENTRY(CompareState) next;
>> + SocketReadState pri_rs;
>> + SocketReadState sec_rs;
>> +} CompareState;
>> +
>> +typedef struct CompareClass {
>> + ObjectClass parent_class;
>> +} CompareClass;
>> +
>> +static char *compare_get_pri_indev(Object *obj, Error **errp)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + return g_strdup(s->pri_indev);
>> +}
>> +
>> +static void compare_set_pri_indev(Object *obj, const char *value,
>> Error **errp)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + g_free(s->pri_indev);
>> + s->pri_indev = g_strdup(value);
>> +}
>> +
>> +static char *compare_get_sec_indev(Object *obj, Error **errp)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + return g_strdup(s->sec_indev);
>> +}
>> +
>> +static void compare_set_sec_indev(Object *obj, const char *value,
>> Error **errp)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + g_free(s->sec_indev);
>> + s->sec_indev = g_strdup(value);
>> +}
>> +
>> +static char *compare_get_outdev(Object *obj, Error **errp)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + return g_strdup(s->outdev);
>> +}
>> +
>> +static void compare_set_outdev(Object *obj, const char *value, Error
>> **errp)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + g_free(s->outdev);
>> + s->outdev = g_strdup(value);
>> +}
>> +
>> +static void compare_pri_rs_finalize(SocketReadState *pri_rs)
>> +{
>> + /* if packet_enqueue pri pkt failed we will send unsupported
>> packet */
>> +}
>> +
>> +static void compare_sec_rs_finalize(SocketReadState *sec_rs)
>> +{
>> + /* if packet_enqueue sec pkt failed we will notify trace */
>> +}
>> +
>> +/*
>> + * 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);
>> +
>> + if (!s->pri_indev || !s->sec_indev || !s->outdev) {
>> + error_setg(errp, "colo compare needs 'primary_in' ,"
>> + "'secondary_in','outdev' property set");
>> + return;
>> + } else if (!strcmp(s->pri_indev, s->outdev) ||
>> + !strcmp(s->sec_indev, s->outdev) ||
>> + !strcmp(s->pri_indev, s->sec_indev)) {
>> + error_setg(errp, "'indev' and 'outdev' could not be same "
>> + "for compare module");
>> + return;
>
> Can relax this by reusing the socket chardev? Looks we only use one
> direction during packet passing.
We can't qemu_chr_fe_claim_no_fail(xxx) same one chardev twice.
Do you mean we should use -chardev socket,id=xx,....mux=on,... ?
>
>> + }
>> +
>> + s->chr_pri_in = qemu_chr_find(s->pri_indev);
>> + if (s->chr_pri_in == NULL) {
>> + error_setg(errp, "Primary IN Device '%s' not found",
>> + s->pri_indev);
>> + return;
>> + }
>> +
>> + s->chr_sec_in = qemu_chr_find(s->sec_indev);
>> + if (s->chr_sec_in == NULL) {
>> + error_setg(errp, "Secondary IN Device '%s' not found",
>> + s->sec_indev);
>> + return;
>> + }
>> +
>> + s->chr_out = qemu_chr_find(s->outdev);
>> + if (s->chr_out == NULL) {
>> + error_setg(errp, "OUT Device '%s' not found", s->outdev);
>> + return;
>> + }
>
> Like I point out in the past iterations, we probably need some
> inspection here and fail if it was socket chardev. (Similar to what
> vhost-use did for udp socket).
>
OK,I got it.
>> +
>> + qemu_chr_fe_claim_no_fail(s->chr_pri_in);
>> +
>> + qemu_chr_fe_claim_no_fail(s->chr_sec_in);
>> +
>> + qemu_chr_fe_claim_no_fail(s->chr_out);
>> +
>> + net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
>> + net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
>> +
>
> Why chr_out is left here?
Because chr_out needn't read data from chardev, just output.
>
>> + return;
>> +}
>> +
>> +static void colo_compare_class_init(ObjectClass *oc, void *data)
>> +{
>> + UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
>> +
>> + ucc->complete = colo_compare_complete;
>> +}
>> +
>> +static void colo_compare_init(Object *obj)
>> +{
>> + object_property_add_str(obj, "primary_in",
>> + compare_get_pri_indev,
>> compare_set_pri_indev,
>> + NULL);
>> + object_property_add_str(obj, "secondary_in",
>> + compare_get_sec_indev,
>> compare_set_sec_indev,
>> + NULL);
>> + object_property_add_str(obj, "outdev",
>> + compare_get_outdev, compare_set_outdev,
>> + NULL);
>> +}
>> +
>> +static void colo_compare_finalize(Object *obj)
>> +{
>> + CompareState *s = COLO_COMPARE(obj);
>> +
>> + if (s->chr_pri_in) {
>> + qemu_chr_add_handlers(s->chr_pri_in, NULL, NULL, NULL, NULL);
>> + qemu_chr_fe_release(s->chr_pri_in);
>> + }
>> + if (s->chr_sec_in) {
>> + qemu_chr_add_handlers(s->chr_sec_in, NULL, NULL, NULL, NULL);
>> + qemu_chr_fe_release(s->chr_sec_in);
>> + }
>> + if (s->chr_out) {
>> + qemu_chr_fe_release(s->chr_out);
>> + }
>> +
>> + g_free(s->pri_indev);
>> + g_free(s->sec_indev);
>> + g_free(s->outdev);
>> +}
>> +
>> +static const TypeInfo colo_compare_info = {
>> + .name = TYPE_COLO_COMPARE,
>> + .parent = TYPE_OBJECT,
>> + .instance_size = sizeof(CompareState),
>> + .instance_init = colo_compare_init,
>> + .instance_finalize = colo_compare_finalize,
>> + .class_size = sizeof(CompareClass),
>> + .class_init = colo_compare_class_init,
>> + .interfaces = (InterfaceInfo[]) {
>> + { TYPE_USER_CREATABLE },
>> + { }
>> + }
>> +};
>> +
>> +static void register_types(void)
>> +{
>> + type_register_static(&colo_compare_info);
>> +}
>> +
>> +type_init(register_types);
>> diff --git a/qemu-options.hx b/qemu-options.hx
>> index 587de8f..79e5896 100644
>> --- a/qemu-options.hx
>> +++ b/qemu-options.hx
>> @@ -3866,6 +3866,44 @@ Dump the network traffic on netdev @var{dev}
>> to the file specified by
>> The file format is libpcap, so it can be analyzed with tools such
>> as tcpdump
>> or Wireshark.
>> +@item -object
>> colo-compare,id=@var{id},primary_in=@var{chardevid},secondary_in=@var{chardevid},
>> +outdev=@var{chardevid}
>> +
>> +Colo-compare gets packet from primary_in@var{chardevid} and
>> secondary_in@var{chardevid},than compare primary packet with
>> +secondary packet. if the packet same,we will output primary
>
> s/if/If/ and miss a blank before ','
will fix~
>> +packet to outdev@var{chardevid},else we will notify colo-frame
>
> miss a blank before ','
will fix it.
>
>> +do checkpoint and send primary packet to outdev@var{chardevid}.
>> +
>> +we can use it with the help of filter-mirror and filter-redirector.
>> +
>> +The simple usage:
>
> This line could be removed since we have example below.
will remove it.
>
>> +
>> +@example
>> +
>> +primary:
>> +-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=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
>> +-chardev socket,id=compare_out0,host=3.3.3.3,port=9005
>> +-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
>> +
>> +secondary:
>> +-netdev tap,id=hn0,vhost=off,script=/etc/qemu-ifup,down
>> script=/etc/qemu-ifdown
>> +-device e1000,netdev=hn0,mac=52:a4:00:12:78:66
>> +-chardev socket,id=red0,host=3.3.3.3,port=9003
>> +-chardev socket,id=red1,host=3.3.3.3,port=9004
>> +-object filter-redirector,id=f1,netdev=hn0,queue=tx,indev=red0
>> +-object filter-redirector,id=f2,netdev=hn0,queue=rx,outdev=red1
>> +
>> +@end example
>> +
>
> Better add detail description on the above command line since it was
> rather complex for starters.
We have add detail description in commit log, need we repeat it again here?
Thanks
Zhang Chen
>
>
>> @item -object
>> secret,id=@var{id},data=@var{string},format=@var{raw|base64}[,keyid=@var{secretid},iv=@var{string}]
>> @item -object
>> secret,id=@var{id},file=@var{filename},format=@var{raw|base64}[,keyid=@var{secretid},iv=@var{string}]
>> diff --git a/vl.c b/vl.c
>> index cbe51ac..c6b9a6f 100644
>> --- a/vl.c
>> +++ b/vl.c
>> @@ -2865,7 +2865,8 @@ static bool object_create_initial(const char
>> *type)
>> if (g_str_equal(type, "filter-buffer") ||
>> g_str_equal(type, "filter-dump") ||
>> g_str_equal(type, "filter-mirror") ||
>> - g_str_equal(type, "filter-redirector")) {
>> + g_str_equal(type, "filter-redirector") ||
>> + g_str_equal(type, "colo-compare")) {
>> return false;
>> }
>
>
>
> .
>
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet
2016-08-02 6:38 ` Jason Wang
@ 2016-08-03 6:34 ` Zhang Chen
0 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-08-03 6:34 UTC (permalink / raw)
To: Jason Wang, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 08/02/2016 02:38 PM, Jason Wang wrote:
>
>
> On 2016年07月26日 09:49, Zhang Chen wrote:
>> COLO-base used by colo-compare and filter-rewriter.
>
> s/used/is used/
>
OK~
>> this can share common data structure like:net packet,
>
> s/this/This/ and ':' looks unnecessary.
OK~
>> and share other functions.
>
> s/share//
>
OK~
>>
>> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
>> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
>> ---
>> net/Makefile.objs | 1 +
>> net/colo-base.c | 74 +++++++++++++++++++++++++++++++++
>> net/colo-base.h | 38 +++++++++++++++++
>> net/colo-compare.c | 119
>> ++++++++++++++++++++++++++++++++++++++++++++++++++++-
>> trace-events | 3 ++
>> 5 files changed, 233 insertions(+), 2 deletions(-)
>> create mode 100644 net/colo-base.c
>> create mode 100644 net/colo-base.h
>>
>> diff --git a/net/Makefile.objs b/net/Makefile.objs
>> index ba92f73..119589f 100644
>> --- a/net/Makefile.objs
>> +++ b/net/Makefile.objs
>> @@ -17,3 +17,4 @@ common-obj-y += filter.o
>> common-obj-y += filter-buffer.o
>> common-obj-y += filter-mirror.o
>> common-obj-y += colo-compare.o
>> +common-obj-y += colo-base.o
>> diff --git a/net/colo-base.c b/net/colo-base.c
>> new file mode 100644
>> index 0000000..f5d5de9
>> --- /dev/null
>> +++ b/net/colo-base.c
>
> I was thinking maybe just use colo.c is better here.
>
If you think that OK, I will change the name in next version,
But it haven't COLO function in this .c,
just some packet and connection data structure or common function.
>> @@ -0,0 +1,74 @@
>> +/*
>> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service
>> (COLO)
>> + * (a.k.a. Fault Tolerance or Continuous Replication)
>> + *
>> + * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
>> + * Copyright (c) 2016 FUJITSU LIMITED
>> + * Copyright (c) 2016 Intel Corporation
>> + *
>> + * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2 or
>> + * later. See the COPYING file in the top-level directory.
>> + */
>> +
>> +#include "qemu/osdep.h"
>> +#include "qemu/error-report.h"
>> +#include "net/colo-base.h"
>> +
>> +int parse_packet_early(Packet *pkt)
>> +{
>> + int network_length;
>> + uint8_t *data = pkt->data;
>> + uint16_t l3_proto;
>> + ssize_t l2hdr_len = eth_get_l2_hdr_length(data);
>> +
>> + if (pkt->size < ETH_HLEN) {
>> + error_report("pkt->size < ETH_HLEN");
>> + return 1;
>> + }
>> + pkt->network_layer = data + ETH_HLEN;
>> + l3_proto = eth_get_l3_proto(data, l2hdr_len);
>> + if (l3_proto != ETH_P_IP) {
>> + return 1;
>> + }
>> +
>> + network_length = pkt->ip->ip_hl * 4;
>> + if (pkt->size < ETH_HLEN + network_length) {
>> + error_report("pkt->size < network_layer + network_length");
>> + return 1;
>> + }
>> + pkt->transport_layer = pkt->network_layer + network_length;
>> + if (!pkt->transport_layer) {
>
> Looks like at least pkt->network_layer can't be zero, so I wonder how
> this condition were met.
I will remove this check in next version.
>
>> + error_report("pkt->transport_layer is valid");
>> + return 1;
>> + }
>> +
>> + return 0;
>> +}
>> +
>> +Packet *packet_new(const void *data, int size)
>> +{
>> + Packet *pkt = g_slice_new(Packet);
>> +
>> + pkt->data = g_memdup(data, size);
>> + pkt->size = size;
>> +
>> + return pkt;
>> +}
>> +
>> +void packet_destroy(void *opaque, void *user_data)
>
> What's the reason of using void * instead of Packet * directly? And
> user_data were not used in this function.
>
If we change void *opaque to Packet *pkt make it have this warning...
net/colo-base.c:138: warning: passing argument 2 of 'g_queue_foreach'
from incompatible pointer type
g_queue_foreach(&conn->primary_list, packet_destroy, NULL);
>> +{
>> + Packet *pkt = opaque;
>> +
>> + g_free(pkt->data);
>> + g_slice_free(Packet, pkt);
>> +}
>> +
>> +/*
>> + * Clear hashtable, stop this hash growing really huge
>> + */
>> +void connection_hashtable_reset(GHashTable *connection_track_table)
>> +{
>> + g_hash_table_remove_all(connection_track_table);
>> +}
>> diff --git a/net/colo-base.h b/net/colo-base.h
>> new file mode 100644
>> index 0000000..48835e7
>> --- /dev/null
>> +++ b/net/colo-base.h
>> @@ -0,0 +1,38 @@
>> +/*
>> + * COarse-grain LOck-stepping Virtual Machines for Non-stop Service
>> (COLO)
>> + * (a.k.a. Fault Tolerance or Continuous Replication)
>> + *
>> + * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
>> + * Copyright (c) 2016 FUJITSU LIMITED
>> + * Copyright (c) 2016 Intel Corporation
>> + *
>> + * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> + *
>> + * This work is licensed under the terms of the GNU GPL, version 2 or
>> + * later. See the COPYING file in the top-level directory.
>> + */
>> +
>> +#ifndef QEMU_COLO_BASE_H
>> +#define QEMU_COLO_BASE_H
>> +
>> +#include "slirp/slirp.h"
>> +#include "qemu/jhash.h"
>> +
>> +#define HASHTABLE_MAX_SIZE 16384
>> +
>> +typedef struct Packet {
>> + void *data;
>> + union {
>> + uint8_t *network_layer;
>
> I think it would be better to use "network_header" here.
OK,will fix in next version.
>
>> + struct ip *ip;
>> + };
>> + uint8_t *transport_layer;
>
> And "transport_header" here.
OK.
>
> And why not union tcp header here? I'm asking since ip header were
> unioned above.
I don't understand what you mean, can you give a example?
Thanks
Zhang Chen
>
>> + int size;
>> +} Packet;
>> +
>> +int parse_packet_early(Packet *pkt);
>> +void connection_hashtable_reset(GHashTable *connection_track_table);
>> +Packet *packet_new(const void *data, int size);
>> +void packet_destroy(void *opaque, void *user_data);
>> +
>> +#endif /* QEMU_COLO_BASE_H */
>> diff --git a/net/colo-compare.c b/net/colo-compare.c
>> index 0402958..7c52cc8 100644
>> --- a/net/colo-compare.c
>> +++ b/net/colo-compare.c
>> @@ -27,13 +27,38 @@
>> #include "sysemu/char.h"
>> #include "qemu/sockets.h"
>> #include "qapi-visit.h"
>> +#include "net/colo-base.h"
>> +#include "trace.h"
>> #define TYPE_COLO_COMPARE "colo-compare"
>> #define COLO_COMPARE(obj) \
>> OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
>> #define COMPARE_READ_LEN_MAX NET_BUFSIZE
>> +#define MAX_QUEUE_SIZE 1024
>> +/*
>> + + CompareState ++
>> + | |
>> + +---------------+ +---------------+ +---------------+
>> + |conn list +--->conn +--------->conn |
>> + +---------------+ +---------------+ +---------------+
>> + | | | | | |
>> + +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
>> + |primary | |secondary |primary | |secondary
>> + |packet | |packet + |packet | |packet +
>> + +--------+ +--------+ +--------+ +--------+
>> + | | | |
>> + +---v----+ +---v----+ +---v----+ +---v----+
>> + |primary | |secondary |primary | |secondary
>> + |packet | |packet + |packet | |packet +
>> + +--------+ +--------+ +--------+ +--------+
>> + | | | |
>> + +---v----+ +---v----+ +---v----+ +---v----+
>> + |primary | |secondary |primary | |secondary
>> + |packet | |packet + |packet | |packet +
>> + +--------+ +--------+ +--------+ +--------+
>> +*/
>> typedef struct CompareState {
>> Object parent;
>> @@ -46,12 +71,89 @@ typedef struct CompareState {
>> QTAILQ_ENTRY(CompareState) next;
>> SocketReadState pri_rs;
>> SocketReadState sec_rs;
>> +
>> + /* hashtable to save connection */
>> + GHashTable *connection_track_table;
>> + /* to save unprocessed_connections */
>> + GQueue unprocessed_connections;
>> + /* proxy current hash size */
>> + uint32_t hashtable_size;
>> } CompareState;
>> typedef struct CompareClass {
>> ObjectClass parent_class;
>> } CompareClass;
>> +enum {
>> + PRIMARY_IN = 0,
>> + SECONDARY_IN,
>> +};
>> +
>> +static int compare_chr_send(CharDriverState *out,
>> + const uint8_t *buf,
>> + uint32_t size);
>> +
>> +/*
>> + * Return 0 on success, if return -1 means the pkt
>> + * is unsupported(arp and ipv6) and will be sent later
>> + */
>> +static int packet_enqueue(CompareState *s, int mode)
>> +{
>> + Packet *pkt = NULL;
>> +
>> + if (mode == PRIMARY_IN) {
>> + pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
>> + } else {
>> + pkt = packet_new(s->sec_rs.buf, s->sec_rs.packet_len);
>> + }
>> +
>> + if (parse_packet_early(pkt)) {
>> + packet_destroy(pkt, NULL);
>> + pkt = NULL;
>> + return -1;
>> + }
>> + /* TODO: get connection key from pkt */
>> +
>> + /*
>> + * TODO: use connection key get conn from
>> + * connection_track_table
>> + */
>> +
>> + /*
>> + * TODO: insert pkt to it's conn->primary_list
>> + * or conn->secondary_list
>> + */
>> +
>> + return 0;
>> +}
>> +
>> +static int compare_chr_send(CharDriverState *out,
>> + const uint8_t *buf,
>> + uint32_t size)
>> +{
>> + int ret = 0;
>> + uint32_t len = htonl(size);
>> +
>> + if (!size) {
>> + return 0;
>> + }
>> +
>> + ret = qemu_chr_fe_write_all(out, (uint8_t *)&len, sizeof(len));
>> + if (ret != sizeof(len)) {
>> + goto err;
>> + }
>> +
>> + ret = qemu_chr_fe_write_all(out, (uint8_t *)buf, size);
>> + if (ret != size) {
>> + goto err;
>> + }
>> +
>> + return 0;
>> +
>> +err:
>> + return ret < 0 ? ret : -EIO;
>> +}
>> +
>> static char *compare_get_pri_indev(Object *obj, Error **errp)
>> {
>> CompareState *s = COLO_COMPARE(obj);
>> @@ -99,12 +201,21 @@ static void compare_set_outdev(Object *obj,
>> const char *value, Error **errp)
>> static void compare_pri_rs_finalize(SocketReadState *pri_rs)
>> {
>> - /* if packet_enqueue pri pkt failed we will send unsupported
>> packet */
>> + CompareState *s = container_of(pri_rs, CompareState, 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);
>> + }
>> }
>> static void compare_sec_rs_finalize(SocketReadState *sec_rs)
>> {
>> - /* if packet_enqueue sec pkt failed we will notify trace */
>> + CompareState *s = container_of(sec_rs, CompareState, sec_rs);
>> +
>> + if (packet_enqueue(s, SECONDARY_IN)) {
>> + trace_colo_compare_main("secondary: unsupported packet in");
>> + }
>> }
>> /*
>> @@ -156,6 +267,10 @@ static void colo_compare_complete(UserCreatable
>> *uc, Error **errp)
>> net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
>> net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
>> + s->hashtable_size = 0;
>> +
>> + /* use g_hash_table_new_full() to new a hashtable */
>> +
>> return;
>> }
>> diff --git a/trace-events b/trace-events
>> index ca7211b..703de1a 100644
>> --- a/trace-events
>> +++ b/trace-events
>> @@ -1916,3 +1916,6 @@ aspeed_vic_update_fiq(int flags) "Raising FIQ: %d"
>> aspeed_vic_update_irq(int flags) "Raising IRQ: %d"
>> aspeed_vic_read(uint64_t offset, unsigned size, uint32_t value)
>> "From 0x%" PRIx64 " of size %u: 0x%" PRIx32
>> aspeed_vic_write(uint64_t offset, unsigned size, uint32_t data) "To
>> 0x%" PRIx64 " of size %u: 0x%" PRIx32
>> +
>> +# net/colo-compare.c
>> +colo_compare_main(const char *chr) ": %s"
>
>
>
> .
>
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu
2016-08-02 6:40 ` Jason Wang
@ 2016-08-03 6:36 ` Zhang Chen
0 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-08-03 6:36 UTC (permalink / raw)
To: Jason Wang, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 08/02/2016 02:40 PM, Jason Wang wrote:
>
>
> On 2016年07月26日 09:49, Zhang Chen wrote:
>> Jhash used by colo-compare and filter-rewriter
>> to save and lookup net connection info
>>
>> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
>> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
>> ---
>> include/qemu/jhash.h | 61
>> ++++++++++++++++++++++++++++++++++++++++++++++++++++
>> 1 file changed, 61 insertions(+)
>> create mode 100644 include/qemu/jhash.h
>>
>> diff --git a/include/qemu/jhash.h b/include/qemu/jhash.h
>> new file mode 100644
>> index 0000000..0fcd875
>> --- /dev/null
>> +++ b/include/qemu/jhash.h
>> @@ -0,0 +1,61 @@
>> +/* jhash.h: Jenkins hash support.
>> + *
>> + * Copyright (C) 2006. Bob Jenkins (bob_jenkins@burtleburtle.net)
>> + *
>> + * http://burtleburtle.net/bob/hash/
>> + *
>> + * These are the credits from Bob's sources:
>> + *
>> + * lookup3.c, by Bob Jenkins, May 2006, Public Domain.
>> + *
>> + * These are functions for producing 32-bit hashes for hash table
>> lookup.
>> + * hashword(), hashlittle(), hashlittle2(), hashbig(), mix(), and
>> final()
>> + * are externally useful functions. Routines to test the hash are
>> +included
>> + * if SELF_TEST is defined. You can use this free for any purpose.
>> +It's in
>
> The indentation looks odd here, please check.
OK, I will fix it in next version.
Thanks
Zhang Chen
>> + * the public domain. It has no warranty.
>> + *
>> + * Copyright (C) 2009-2010 Jozsef Kadlecsik (kadlec@blackhole.kfki.hu)
>> + *
>> + * I've modified Bob's hash to be useful in the Linux kernel, and
>> + * any bugs present are my fault.
>> + * Jozsef
>> + */
>> +
>> +#ifndef QEMU_JHASH_H__
>> +#define QEMU_JHASH_H__
>> +
>> +#include "qemu/bitops.h"
>> +
>> +/*
>> + * hashtable relation copy from linux kernel jhash
>> + */
>> +
>> +/* __jhash_mix -- mix 3 32-bit values reversibly. */
>> +#define __jhash_mix(a, b, c) \
>> +{ \
>> + a -= c; a ^= rol32(c, 4); c += b; \
>> + b -= a; b ^= rol32(a, 6); a += c; \
>> + c -= b; c ^= rol32(b, 8); b += a; \
>> + a -= c; a ^= rol32(c, 16); c += b; \
>> + b -= a; b ^= rol32(a, 19); a += c; \
>> + c -= b; c ^= rol32(b, 4); b += a; \
>> +}
>> +
>> +/* __jhash_final - final mixing of 3 32-bit values (a,b,c) into c */
>> +#define __jhash_final(a, b, c) \
>> +{ \
>> + c ^= b; c -= rol32(b, 14); \
>> + a ^= c; a -= rol32(c, 11); \
>> + b ^= a; b -= rol32(a, 25); \
>> + c ^= b; c -= rol32(b, 16); \
>> + a ^= c; a -= rol32(c, 4); \
>> + b ^= a; b -= rol32(a, 14); \
>> + c ^= b; c -= rol32(b, 24); \
>> +}
>> +
>> +/* An arbitrary initial parameter */
>> +#define JHASH_INITVAL 0xdeadbeef
>> +
>> +#endif /* QEMU_JHASH_H__ */
>
>
>
> .
>
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet
2016-08-02 7:14 ` Jason Wang
@ 2016-08-04 6:49 ` Zhang Chen
0 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-08-04 6:49 UTC (permalink / raw)
To: Jason Wang, qemu devel
Cc: Li Zhijian, Wen Congyang, zhanghailiang, eddie . dong,
Dr . David Alan Gilbert
On 08/02/2016 03:14 PM, Jason Wang wrote:
>
>
> On 2016年07月26日 09:49, Zhang Chen wrote:
>> In this patch we use kernel jhash table to track
>> connection, and then enqueue net packet like this:
>>
>> + CompareState ++
>> | |
>> +---------------+ +---------------+ +---------------+
>> |conn list +--->conn +--------->conn |
>> +---------------+ +---------------+ +---------------+
>> | | | | | |
>> +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
>> |primary | |secondary |primary | |secondary
>> |packet | |packet + |packet | |packet +
>> +--------+ +--------+ +--------+ +--------+
>> | | | |
>> +---v----+ +---v----+ +---v----+ +---v----+
>> |primary | |secondary |primary | |secondary
>> |packet | |packet + |packet | |packet +
>> +--------+ +--------+ +--------+ +--------+
>> | | | |
>> +---v----+ +---v----+ +---v----+ +---v----+
>> |primary | |secondary |primary | |secondary
>> |packet | |packet + |packet | |packet +
>> +--------+ +--------+ +--------+ +--------+
>>
>> We use conn_list to record connection info.
>> When we want to enqueue a packet, firstly get the
>> connection from connection_track_table. then push
>> the packet to g_queue(pri/sec) in it's own conn.
>>
>> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
>> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
>> ---
>> net/colo-base.c | 108
>> +++++++++++++++++++++++++++++++++++++++++++++++++++++
>> net/colo-base.h | 30 +++++++++++++++
>> net/colo-compare.c | 70 +++++++++++++++++++++++++++++-----
>> 3 files changed, 198 insertions(+), 10 deletions(-)
>>
>> diff --git a/net/colo-base.c b/net/colo-base.c
>> index f5d5de9..7e91dec 100644
>> --- a/net/colo-base.c
>> +++ b/net/colo-base.c
>> @@ -16,6 +16,29 @@
>> #include "qemu/error-report.h"
>> #include "net/colo-base.h"
>> +uint32_t connection_key_hash(const void *opaque)
>> +{
>> + const ConnectionKey *key = opaque;
>> + uint32_t a, b, c;
>> +
>> + /* Jenkins hash */
>> + a = b = c = JHASH_INITVAL + sizeof(*key);
>> + a += key->src.s_addr;
>> + b += key->dst.s_addr;
>> + c += (key->src_port | key->dst_port << 16);
>> + __jhash_mix(a, b, c);
>> +
>> + a += key->ip_proto;
>> + __jhash_final(a, b, c);
>> +
>> + return c;
>> +}
>> +
>> +int connection_key_equal(const void *key1, const void *key2)
>> +{
>> + return memcmp(key1, key2, sizeof(ConnectionKey)) == 0;
>> +}
>> +
>> int parse_packet_early(Packet *pkt)
>> {
>> int network_length;
>> @@ -47,6 +70,62 @@ int parse_packet_early(Packet *pkt)
>> return 0;
>> }
>> +void fill_connection_key(Packet *pkt, ConnectionKey *key)
>> +{
>> + uint32_t tmp_ports;
>> +
>> + key->ip_proto = pkt->ip->ip_p;
>> +
>> + switch (key->ip_proto) {
>> + case IPPROTO_TCP:
>> + case IPPROTO_UDP:
>> + case IPPROTO_DCCP:
>> + case IPPROTO_ESP:
>> + case IPPROTO_SCTP:
>> + case IPPROTO_UDPLITE:
>> + tmp_ports = *(uint32_t *)(pkt->transport_layer);
>> + key->src = pkt->ip->ip_src;
>> + key->dst = pkt->ip->ip_dst;
>> + key->src_port = ntohs(tmp_ports & 0xffff);
>> + key->dst_port = ntohs(tmp_ports >> 16);
>> + break;
>> + case IPPROTO_AH:
>> + tmp_ports = *(uint32_t *)(pkt->transport_layer + 4);
>> + key->src = pkt->ip->ip_src;
>> + key->dst = pkt->ip->ip_dst;
>> + key->src_port = ntohs(tmp_ports & 0xffff);
>> + key->dst_port = ntohs(tmp_ports >> 16);
>> + break;
>> + default:
>> + key->src_port = 0;
>> + key->dst_port = 0;
>> + break;
>> + }
>> +}
>> +
>> +Connection *connection_new(ConnectionKey *key)
>> +{
>> + Connection *conn = g_slice_new(Connection);
>> +
>> + conn->ip_proto = key->ip_proto;
>> + conn->processing = false;
>> + g_queue_init(&conn->primary_list);
>> + g_queue_init(&conn->secondary_list);
>> +
>> + return conn;
>> +}
>> +
>> +void connection_destroy(void *opaque)
>> +{
>> + Connection *conn = opaque;
>> +
>> + g_queue_foreach(&conn->primary_list, packet_destroy, NULL);
>> + g_queue_free(&conn->primary_list);
>> + g_queue_foreach(&conn->secondary_list, packet_destroy, NULL);
>> + g_queue_free(&conn->secondary_list);
>> + g_slice_free(Connection, conn);
>> +}
>> +
>> Packet *packet_new(const void *data, int size)
>> {
>> Packet *pkt = g_slice_new(Packet);
>> @@ -72,3 +151,32 @@ void connection_hashtable_reset(GHashTable
>> *connection_track_table)
>> {
>> g_hash_table_remove_all(connection_track_table);
>> }
>> +
>> +/* if not found, create a new connection and add to hash table */
>> +Connection *connection_get(GHashTable *connection_track_table,
>> + ConnectionKey *key,
>> + uint32_t *hashtable_size)
>> +{
>> + Connection *conn = g_hash_table_lookup(connection_track_table,
>> key);
>> +
>> + if (conn == NULL) {
>> + ConnectionKey *new_key = g_memdup(key, sizeof(*key));
>> +
>> + conn = connection_new(key);
>> +
>> + (*hashtable_size) += 1;
>> + if (*hashtable_size > HASHTABLE_MAX_SIZE) {
>> + error_report("colo proxy connection hashtable full,
>> clear it");
>> + connection_hashtable_reset(connection_track_table);
>> + /*
>> + * when hashtable_size == 0, clear the conn_list
>> + * in place where be called.
>> + */
>
> Does this mean it requires the caller to do this? If yes, seems not
> good, why not simply do things here?
OK, I will add conn_list as parameter input to this function then clear it.
>
>> + *hashtable_size = 0;
>> + }
>> +
>> + g_hash_table_insert(connection_track_table, new_key, conn);
>
> Then we lose the track of *hashtable_size here. It should be 1 but we
> set it to zero if we are out of space.
After clear conn_list I will fix it.
>
>> + }
>> +
>> + return conn;
>> +}
>> diff --git a/net/colo-base.h b/net/colo-base.h
>> index 48835e7..0505608 100644
>> --- a/net/colo-base.h
>> +++ b/net/colo-base.h
>> @@ -30,7 +30,37 @@ typedef struct Packet {
>> int size;
>> } Packet;
>> +typedef struct ConnectionKey {
>> + /* (src, dst) must be grouped, in the same way than in IP header */
>> + struct in_addr src;
>> + struct in_addr dst;
>> + uint16_t src_port;
>> + uint16_t dst_port;
>> + uint8_t ip_proto;
>> +} QEMU_PACKED ConnectionKey;
>> +
>> +typedef struct Connection {
>> + /* connection primary send queue: element type: Packet */
>> + GQueue primary_list;
>> + /* connection secondary send queue: element type: Packet */
>> + GQueue secondary_list;
>> + /* flag to enqueue unprocessed_connections */
>> + bool processing;
>> + uint8_t ip_proto;
>> + /* be used by filter-rewriter */
>> + tcp_seq primary_seq;
>> + tcp_seq secondary_seq;
>> +} Connection;
>> +
>> +uint32_t connection_key_hash(const void *opaque);
>> +int connection_key_equal(const void *opaque1, const void *opaque2);
>> int parse_packet_early(Packet *pkt);
>> +void fill_connection_key(Packet *pkt, ConnectionKey *key);
>> +Connection *connection_new(ConnectionKey *key);
>> +void connection_destroy(void *opaque);
>> +Connection *connection_get(GHashTable *connection_track_table,
>> + ConnectionKey *key,
>> + uint32_t *hashtable_size);
>> void connection_hashtable_reset(GHashTable *connection_track_table);
>> Packet *packet_new(const void *data, int size);
>> void packet_destroy(void *opaque, void *user_data);
>> diff --git a/net/colo-compare.c b/net/colo-compare.c
>> index 7c52cc8..5f87710 100644
>> --- a/net/colo-compare.c
>> +++ b/net/colo-compare.c
>> @@ -72,6 +72,11 @@ typedef struct CompareState {
>> SocketReadState pri_rs;
>> SocketReadState sec_rs;
>> + /* connection list: the connections belonged to this NIC could
>> be found
>> + * in this list.
>> + * element type: Connection
>> + */
>> + GQueue conn_list;
>> /* hashtable to save connection */
>> GHashTable *connection_track_table;
>> /* to save unprocessed_connections */
>> @@ -93,13 +98,30 @@ static int compare_chr_send(CharDriverState *out,
>> const uint8_t *buf,
>> uint32_t size);
>> +static void colo_rm_connection(void *opaque, void *user_data)
>> +{
>> + Connection *conn = opaque;
>> + Packet *pkt = NULL;
>> +
>> + while (!g_queue_is_empty(&conn->primary_list)) {
>> + pkt = g_queue_pop_head(&conn->primary_list);
>> + packet_destroy(pkt, NULL);
>> + }
>> + while (!g_queue_is_empty(&conn->secondary_list)) {
>> + pkt = g_queue_pop_head(&conn->secondary_list);
>> + packet_destroy(pkt, NULL);
>> + }
>
> This looks rather similar to connection_destroy(), can we share some
> codes?
Good idea~
>
>> +}
>> +
>> /*
>> * Return 0 on success, if return -1 means the pkt
>> * is unsupported(arp and ipv6) and will be sent later
>> */
>> static int packet_enqueue(CompareState *s, int mode)
>> {
>> + ConnectionKey key = {{ 0 } };
>> Packet *pkt = NULL;
>> + Connection *conn;
>> if (mode == PRIMARY_IN) {
>> pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
>> @@ -112,17 +134,38 @@ static int packet_enqueue(CompareState *s, int
>> mode)
>> pkt = NULL;
>> return -1;
>> }
>> - /* TODO: get connection key from pkt */
>> + fill_connection_key(pkt, &key);
>> - /*
>> - * TODO: use connection key get conn from
>> - * connection_track_table
>> - */
>> + conn = connection_get(s->connection_track_table,
>> + &key,
>> + &s->hashtable_size);
>> - /*
>> - * TODO: insert pkt to it's conn->primary_list
>> - * or conn->secondary_list
>> - */
>> + if (!s->hashtable_size) {
>> + g_queue_foreach(&s->conn_list, colo_rm_connection, NULL);
>> + }
>> +
>> + if (!conn->processing) {
>> + g_queue_push_tail(&s->conn_list, conn);
>> + conn->processing = true;
>
> Why not simply do this in connection_get(), you can save a
> conn->processing flag.
Because filter-rewriter need connection_get() without do this work,
We make connection_get() more common.
>
>> + }
>> +
>> + if (mode == PRIMARY_IN) {
>> + if (g_queue_get_length(&conn->primary_list) <
>> + MAX_QUEUE_SIZE) {
>> + g_queue_push_tail(&conn->primary_list, pkt);
>> + } else {
>> + error_report("colo compare primary queue size too big,"
>> + "drop packet");
>
> I don't see how packet were dropped in this case?
In here we didn't push the packet to primary_list, then we comparing
queued packet can't
get this packet, so,the packet be dropped.
Thanks
Zhang Chen
>
>> + }
>> + } else {
>> + if (g_queue_get_length(&conn->secondary_list) <
>> + MAX_QUEUE_SIZE) {
>> + g_queue_push_tail(&conn->secondary_list, pkt);
>> + } else {
>> + error_report("colo compare secondary queue size too big,"
>> + "drop packet");
>> + }
>> + }
>> return 0;
>> }
>> @@ -267,9 +310,14 @@ static void colo_compare_complete(UserCreatable
>> *uc, Error **errp)
>> net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
>> net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
>> + g_queue_init(&s->conn_list);
>> +
>> s->hashtable_size = 0;
>> - /* use g_hash_table_new_full() to new a hashtable */
>> + s->connection_track_table =
>> g_hash_table_new_full(connection_key_hash,
>> + connection_key_equal,
>> + g_free,
>> + connection_destroy);
>> return;
>> }
>> @@ -310,6 +358,8 @@ static void colo_compare_finalize(Object *obj)
>> qemu_chr_fe_release(s->chr_out);
>> }
>> + g_queue_free(&s->conn_list);
>> +
>> g_free(s->pri_indev);
>> g_free(s->sec_indev);
>> g_free(s->outdev);
>
>
>
> .
>
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread
2016-08-02 7:52 ` Jason Wang
@ 2016-08-05 7:45 ` Zhang Chen
0 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-08-05 7:45 UTC (permalink / raw)
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 <zhangchen.fnst@cn.fujitsu.com>
>> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
>> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
>> ---
>> 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
^ permalink raw reply [flat|nested] 20+ messages in thread
* Re: [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison
2016-08-02 8:04 ` Jason Wang
@ 2016-08-05 8:55 ` Zhang Chen
0 siblings, 0 replies; 20+ messages in thread
From: Zhang Chen @ 2016-08-05 8:55 UTC (permalink / raw)
To: Jason Wang, qemu devel
Cc: Li Zhijian, eddie . dong, Dr . David Alan Gilbert, zhanghailiang
On 08/02/2016 04:04 PM, Jason Wang wrote:
>
>
> On 2016年07月26日 09:49, Zhang Chen wrote:
>> We add TCP,UDP,ICMP packet comparison to replace
>> IP packet comparison. This can increase the
>> accuracy of the package comparison.
>> less checkpoint more efficiency.
>
> s/less/Less/
OK.
>
>>
>> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
>> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
>> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
>> ---
>> net/colo-compare.c | 174
>> +++++++++++++++++++++++++++++++++++++++++++++++++++--
>> trace-events | 4 ++
>> 2 files changed, 174 insertions(+), 4 deletions(-)
>>
>> diff --git a/net/colo-compare.c b/net/colo-compare.c
>> index e020edc..c7bb5f7 100644
>> --- a/net/colo-compare.c
>> +++ b/net/colo-compare.c
>> @@ -18,6 +18,7 @@
>> #include "qapi/qmp/qerror.h"
>> #include "qapi/error.h"
>> #include "net/net.h"
>> +#include "net/eth.h"
>> #include "net/vhost_net.h"
>> #include "qom/object_interfaces.h"
>> #include "qemu/iov.h"
>> @@ -197,9 +198,158 @@ static int colo_packet_compare(Packet *ppkt,
>> Packet *spkt)
>> }
>> }
>> -static int colo_packet_compare_all(Packet *spkt, Packet *ppkt)
>> +/*
>> + * called from the compare thread on the primary
>> + * for compare tcp packet
>> + * compare_tcp copied from Dr. David Alan Gilbert's branch
>> + */
>> +static int colo_packet_compare_tcp(Packet *spkt, Packet *ppkt)
>> +{
>> + struct tcphdr *ptcp, *stcp;
>> + int res;
>> + char *sdebug, *ddebug;
>> +
>> + trace_colo_compare_main("compare tcp");
>> + if (ppkt->size != spkt->size) {
>> + if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
>> + trace_colo_compare_main("pkt size not same");
>> + }
>> + return -1;
>> + }
>> +
>> + ptcp = (struct tcphdr *)ppkt->transport_layer;
>> + stcp = (struct tcphdr *)spkt->transport_layer;
>> +
>> + if (ptcp->th_seq != stcp->th_seq) {
>> + if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
>> + trace_colo_compare_main("pkt tcp seq not same");
>> + }
>> + return -1;
>> + }
>> +
>> + /*
>> + * The 'identification' field in the IP header is *very* random
>> + * it almost never matches. Fudge this by ignoring differences in
>> + * unfragmented packets; they'll normally sort themselves out if
>> different
>> + * anyway, and it should recover at the TCP level.
>> + * An alternative would be to get both the primary and secondary
>> to rewrite
>> + * somehow; but that would need some sync traffic to sync the state
>> + */
>> + if (ntohs(ppkt->ip->ip_off) & IP_DF) {
>> + spkt->ip->ip_id = ppkt->ip->ip_id;
>> + /* and the sum will be different if the IDs were different */
>> + spkt->ip->ip_sum = ppkt->ip->ip_sum;
>
> This looks dangerous. If packet were not logical same, ip cusm of
> secondary were changed.
If packet not same, we will do checkpoint.
>
>> + }
>> +
>> + res = memcmp(ppkt->data + ETH_HLEN, spkt->data + ETH_HLEN,
>> + (spkt->size - ETH_HLEN));
>> +
>> + if (res != 0 &&
>> trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
>> + sdebug = strdup(inet_ntoa(ppkt->ip->ip_src));
>> + ddebug = strdup(inet_ntoa(ppkt->ip->ip_dst));
>> + fprintf(stderr, "%s: src/dst: %s/%s p: seq/ack=%u/%u"
>> + " s: seq/ack=%u/%u res=%d flags=%x/%x\n", __func__,
>> + sdebug, ddebug,
>> + ntohl(ptcp->th_seq), ntohl(ptcp->th_ack),
>> + ntohl(stcp->th_seq), ntohl(stcp->th_ack),
>> + res, ptcp->th_flags, stcp->th_flags);
>> +
>> + trace_colo_compare_tcp_miscompare("Primary len", ppkt->size);
>> + qemu_hexdump((char *)ppkt->data, stderr, "colo-compare",
>> ppkt->size);
>> + trace_colo_compare_tcp_miscompare("Secondary len", spkt->size);
>> + qemu_hexdump((char *)spkt->data, stderr, "colo-compare",
>> spkt->size);
>> +
>> + g_free(sdebug);
>> + g_free(ddebug);
>> + }
>> +
>> + return res;
>> +}
>> +
>> +/*
>> + * called from the compare thread on the primary
>> + * for compare udp packet
>> + */
>> +static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
>> +{
>> + int ret;
>> +
>> + trace_colo_compare_main("compare udp");
>> + ret = colo_packet_compare(ppkt, spkt);
>> +
>> + if (ret) {
>> + trace_colo_compare_udp_miscompare("primary pkt size",
>> ppkt->size);
>> + qemu_hexdump((char *)ppkt->data, stderr, "colo-compare",
>> ppkt->size);
>> + trace_colo_compare_udp_miscompare("Secondary pkt size",
>> spkt->size);
>> + qemu_hexdump((char *)spkt->data, stderr, "colo-compare",
>> spkt->size);
>> + }
>> +
>> + return ret;
>> +}
>> +
>> +/*
>> + * called from the compare thread on the primary
>> + * for compare icmp packet
>> + */
>> +static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
>> {
>
> Consider icmp packet were usually not big, why not simply use memcpy?
> (I think I asked the same question in the past).
>
In past we add the colo_packet_compare(ppkt, spkt) in tail to compare
all packet,
by the head of this function we can know what part of icmp pkt
difference when packet not same.
So, for this comments I will remove head of this function in next version.
Thanks
Zhang Chen
>> - trace_colo_compare_main("compare all");
>> + int network_length;
>> + struct icmp *icmp_ppkt, *icmp_spkt;
>> +
>> + trace_colo_compare_main("compare icmp");
>> + network_length = ppkt->ip->ip_hl * 4;
>> + if (ppkt->size != spkt->size ||
>> + ppkt->size < network_length + ETH_HLEN) {
>> + return -1;
>> + }
>> + icmp_ppkt = (struct icmp *)(ppkt->data + network_length +
>> ETH_HLEN);
>> + icmp_spkt = (struct icmp *)(spkt->data + network_length +
>> ETH_HLEN);
>> +
>> + if ((icmp_ppkt->icmp_type == icmp_spkt->icmp_type) &&
>> + (icmp_ppkt->icmp_code == icmp_spkt->icmp_code)) {
>> + if (icmp_ppkt->icmp_type == ICMP_REDIRECT) {
>> + if (icmp_ppkt->icmp_gwaddr.s_addr !=
>> + icmp_spkt->icmp_gwaddr.s_addr) {
>> + trace_colo_compare_main("icmp_gwaddr.s_addr not same");
>> + return -1;
>> + }
>> + } else if ((icmp_ppkt->icmp_type == ICMP_UNREACH) &&
>> + (icmp_ppkt->icmp_type == ICMP_UNREACH_NEEDFRAG)) {
>> + if (icmp_ppkt->icmp_nextmtu != icmp_spkt->icmp_nextmtu) {
>> + trace_colo_compare_main("icmp_nextmtu not same");
>> + return -1;
>> + }
>> + }
>> +
>> + if (colo_packet_compare(ppkt, spkt)) {
>> + trace_colo_compare_icmp_miscompare("primary pkt size",
>> + ppkt->size);
>> + qemu_hexdump((char *)ppkt->data, stderr, "colo-compare",
>> + ppkt->size);
>> + trace_colo_compare_icmp_miscompare("Secondary pkt size",
>> + spkt->size);
>> + qemu_hexdump((char *)spkt->data, stderr, "colo-compare",
>> + spkt->size);
>> + return -1;
>> + }
>> + } else {
>> + return -1;
>> + }
>> +
>> + return 0;
>> +}
>> +
>> +/*
>> + * called from the compare thread on the primary
>> + * for compare other packet
>> + */
>> +static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
>> +{
>> + trace_colo_compare_main("compare other");
>> + 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));
>> return colo_packet_compare(ppkt, spkt);
>> }
>> @@ -263,8 +413,24 @@ static void colo_compare_connection(void
>> *opaque, void *user_data)
>> 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);
>> + switch (conn->ip_proto) {
>> + case IPPROTO_TCP:
>> + result = g_queue_find_custom(&conn->secondary_list,
>> + pkt, (GCompareFunc)colo_packet_compare_tcp);
>> + break;
>> + case IPPROTO_UDP:
>> + result = g_queue_find_custom(&conn->secondary_list,
>> + pkt, (GCompareFunc)colo_packet_compare_udp);
>> + break;
>> + case IPPROTO_ICMP:
>> + result = g_queue_find_custom(&conn->secondary_list,
>> + pkt, (GCompareFunc)colo_packet_compare_icmp);
>> + break;
>> + default:
>> + result = g_queue_find_custom(&conn->secondary_list,
>> + pkt, (GCompareFunc)colo_packet_compare_other);
>> + break;
>> + }
>> if (result) {
>> ret = compare_chr_send(s->chr_out, pkt->data, pkt->size);
>> diff --git a/trace-events b/trace-events
>> index 1537e91..ab22eb2 100644
>> --- a/trace-events
>> +++ b/trace-events
>> @@ -1919,5 +1919,9 @@ 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_tcp_miscompare(const char *sta, int size) ": %s = %d"
>> +colo_compare_udp_miscompare(const char *sta, int size) ": %s = %d"
>> +colo_compare_icmp_miscompare(const char *sta, int size) ": %s = %d"
>> 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
>> +colo_compare_miscompare(void) ""
>
>
>
> .
>
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 20+ messages in thread
end of thread, other threads:[~2016-08-05 8:55 UTC | newest]
Thread overview: 20+ messages (download: mbox.gz / follow: Atom feed)
-- links below jump to the message on this page --
2016-07-26 1:49 [Qemu-devel] [RFC PATCH V10 0/7] Introduce COLO-compare Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 1/7] colo-compare: introduce colo compare initialization Zhang Chen
2016-08-02 6:26 ` Jason Wang
2016-08-03 2:51 ` Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 2/7] colo-base: add colo-base to define and handle packet Zhang Chen
2016-08-02 6:38 ` Jason Wang
2016-08-03 6:34 ` Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 3/7] Jhash: add linux kernel jhashtable in qemu Zhang Chen
2016-08-02 6:40 ` Jason Wang
2016-08-03 6:36 ` Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet Zhang Chen
2016-08-02 7:14 ` Jason Wang
2016-08-04 6:49 ` Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 5/7] qemu-char: Add qemu_chr_add_handlers_full() for GMaincontext Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 6/7] colo-compare: introduce packet comparison thread Zhang Chen
2016-08-02 7:52 ` Jason Wang
2016-08-05 7:45 ` Zhang Chen
2016-07-26 1:49 ` [Qemu-devel] [RFC PATCH V10 7/7] colo-compare: add TCP, UDP, ICMP packet comparison Zhang Chen
2016-08-02 8:04 ` Jason Wang
2016-08-05 8:55 ` Zhang Chen
This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.