* [Qemu-devel] [RFC PATCH V7 1/7] colo-compare: introduce colo compare initialization
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 2/7] colo-base: add colo-base to define and handle packet Zhang Chen
` (5 subsequent siblings)
6 siblings, 0 replies; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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] 14+ messages in thread* [Qemu-devel] [RFC PATCH V7 2/7] colo-base: add colo-base to define and handle packet
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 1/7] colo-compare: introduce colo compare initialization Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 3/7] Jhash: add linux kernel jhashtable in qemu Zhang Chen
` (4 subsequent siblings)
6 siblings, 0 replies; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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] 14+ messages in thread* [Qemu-devel] [RFC PATCH V7 3/7] Jhash: add linux kernel jhashtable in qemu
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 1/7] colo-compare: introduce colo compare initialization Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 2/7] colo-base: add colo-base to define and handle packet Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 4/7] colo-compare: track connection and enqueue packet Zhang Chen
` (3 subsequent siblings)
6 siblings, 0 replies; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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] 14+ messages in thread* [Qemu-devel] [RFC PATCH V7 4/7] colo-compare: track connection and enqueue packet
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
` (2 preceding siblings ...)
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 3/7] Jhash: add linux kernel jhashtable in qemu Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 5/7] qemu-char: Fix context for g_source_attach() Zhang Chen
` (2 subsequent siblings)
6 siblings, 0 replies; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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] 14+ messages in thread* [Qemu-devel] [RFC PATCH V7 5/7] qemu-char: Fix context for g_source_attach()
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
` (3 preceding siblings ...)
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 4/7] colo-compare: track connection and enqueue packet Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
2016-07-18 8:37 ` Daniel P. Berrange
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread Zhang Chen
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 7/7] colo-compare: add TCP, UDP, ICMP packet comparison Zhang Chen
6 siblings, 1 reply; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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
We want to poll and handle chardev in another thread
other than main loop. But qemu_chr_add_handlers() can only
work for global default context other than thread default context.
So we use g_source_attach(xx, g_main_context_get_thread_default())
replace g_source_attach(xx, NULL) to attach g_source.
Comments from jason.
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: Jason Wang <jasowang@redhat.com>
---
io/channel.c | 2 +-
qemu-char.c | 6 +++---
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/io/channel.c b/io/channel.c
index 692eb17..cd25677 100644
--- a/io/channel.c
+++ b/io/channel.c
@@ -146,7 +146,7 @@ guint qio_channel_add_watch(QIOChannel *ioc,
g_source_set_callback(source, (GSourceFunc)func, user_data, notify);
- id = g_source_attach(source, NULL);
+ id = g_source_attach(source, g_main_context_get_thread_default());
g_source_unref(source);
return id;
diff --git a/qemu-char.c b/qemu-char.c
index b597ee1..ed7e1f7 100644
--- a/qemu-char.c
+++ b/qemu-char.c
@@ -860,7 +860,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, g_main_context_get_thread_default());
} else {
g_source_destroy(iwp->src);
g_source_unref(iwp->src);
@@ -919,7 +919,7 @@ static guint io_add_watch_poll(QIOChannel *ioc,
iwp->fd_read = (GSourceFunc) fd_read;
iwp->src = NULL;
- tag = g_source_attach(&iwp->parent, NULL);
+ tag = g_source_attach(&iwp->parent, g_main_context_get_thread_default());
g_source_unref(&iwp->parent);
return tag;
}
@@ -3983,7 +3983,7 @@ int qemu_chr_fe_add_watch(CharDriverState *s, GIOCondition cond,
}
g_source_set_callback(src, (GSourceFunc)func, user_data, NULL);
- tag = g_source_attach(src, NULL);
+ tag = g_source_attach(src, g_main_context_get_thread_default());
g_source_unref(src);
return tag;
--
2.7.4
^ permalink raw reply related [flat|nested] 14+ messages in thread* Re: [Qemu-devel] [RFC PATCH V7 5/7] qemu-char: Fix context for g_source_attach()
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 5/7] qemu-char: Fix context for g_source_attach() Zhang Chen
@ 2016-07-18 8:37 ` Daniel P. Berrange
0 siblings, 0 replies; 14+ messages in thread
From: Daniel P. Berrange @ 2016-07-18 8:37 UTC (permalink / raw)
To: Zhang Chen
Cc: qemu devel, Jason Wang, Li Zhijian, Wen Congyang, zhanghailiang,
eddie . dong, Dr . David Alan Gilbert, Paolo Bonzini
On Mon, Jul 18, 2016 at 03:40:48PM +0800, Zhang Chen wrote:
> We want to poll and handle chardev in another thread
> other than main loop. But qemu_chr_add_handlers() can only
> work for global default context other than thread default context.
> So we use g_source_attach(xx, g_main_context_get_thread_default())
> replace g_source_attach(xx, NULL) to attach g_source.
> Comments from jason.
>
> 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: Jason Wang <jasowang@redhat.com>
> ---
> io/channel.c | 2 +-
> qemu-char.c | 6 +++---
> 2 files changed, 4 insertions(+), 4 deletions(-)
So, on further reflection I'm against adding this patch - please see
my comments against the next patch for why I think this is not needed
nor desirable.
Regards,
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
^ permalink raw reply [flat|nested] 14+ messages in thread
* [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
` (4 preceding siblings ...)
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 5/7] qemu-char: Fix context for g_source_attach() Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
2016-07-18 8:37 ` Daniel P. Berrange
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 7/7] colo-compare: add TCP, UDP, ICMP packet comparison Zhang Chen
6 siblings, 1 reply; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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 | 216 +++++++++++++++++++++++++++++++++++++++++++++++++++++
trace-events | 2 +
4 files changed, 222 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..942e326 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,69 @@ 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();
+ g_assert(g_main_context_get_thread_default() == NULL);
+ g_main_context_push_thread_default(worker_context);
+ g_assert(g_main_context_get_thread_default() == worker_context);
+
+ qemu_chr_add_handlers(s->chr_pri_in, compare_chr_can_read,
+ compare_pri_chr_in, NULL, s);
+ qemu_chr_add_handlers(s->chr_sec_in, compare_chr_can_read,
+ compare_sec_chr_in, NULL, s);
+
+ compare_loop = g_main_loop_new(worker_context, FALSE);
+
+ g_main_loop_run(compare_loop);
+
+ g_main_loop_unref(compare_loop);
+ g_main_context_pop_thread_default(worker_context);
+ g_main_context_unref(worker_context);
+ return NULL;
+}
+
static char *compare_get_pri_indev(Object *obj, Error **errp)
{
CompareState *s = COLO_COMPARE(obj);
@@ -249,6 +424,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 +436,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 +516,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 +569,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] 14+ messages in thread* Re: [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread Zhang Chen
@ 2016-07-18 8:37 ` Daniel P. Berrange
2016-07-19 3:32 ` Zhang Chen
0 siblings, 1 reply; 14+ messages in thread
From: Daniel P. Berrange @ 2016-07-18 8:37 UTC (permalink / raw)
To: Zhang Chen
Cc: qemu devel, Jason Wang, Li Zhijian, eddie . dong,
Dr . David Alan Gilbert, zhanghailiang
On Mon, Jul 18, 2016 at 03:40:49PM +0800, 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,
> 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 | 216 +++++++++++++++++++++++++++++++++++++++++++++++++++++
> trace-events | 2 +
> 4 files changed, 222 insertions(+)
>
> +static void *colo_compare_thread(void *opaque)
> +{
> + GMainContext *worker_context;
> + GMainLoop *compare_loop;
> + CompareState *s = opaque;
> +
> + worker_context = g_main_context_new();
> + g_assert(g_main_context_get_thread_default() == NULL);
> + g_main_context_push_thread_default(worker_context);
> + g_assert(g_main_context_get_thread_default() == worker_context);
> +
> + qemu_chr_add_handlers(s->chr_pri_in, compare_chr_can_read,
> + compare_pri_chr_in, NULL, s);
> + qemu_chr_add_handlers(s->chr_sec_in, compare_chr_can_read,
> + compare_sec_chr_in, NULL, s);
Looking at this I see you've used g_main_context_push_thread_default
in order to avoid having to modify the qemu_chr_add_handlers() method
to accept a GMainContext. Personally I think I'd favour explicit
passing of a GMainContext, rather than modifying Qemu Char code &
QIOChannel to magically use per-thread main context. In particular
I'm concerned that one day other code may use the
g_main_context_push_thread_default method for some other purpose
but still want their char devs handled by the main thread. Your
solution makes this impossible and adds magic which could surprise
us in bad ways.
So I think I'd rather see qemu_chr_add_handlers() modified to accept
a GMainContext, or perhaps add a qemu_chr_add_handlers_full() method
which takes a GMainContext, so we can avoid modifying all existing
callers ofo qemu_chr_add_handlers. There should be no need to modify
QIOChannel at all, since you can use qio_channel_create_watch instead
of qio_channel_add_watch and simply attach to the context you want
directly.
Regards,
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
^ permalink raw reply [flat|nested] 14+ messages in thread* Re: [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread
2016-07-18 8:37 ` Daniel P. Berrange
@ 2016-07-19 3:32 ` Zhang Chen
2016-07-19 9:03 ` Daniel P. Berrange
0 siblings, 1 reply; 14+ messages in thread
From: Zhang Chen @ 2016-07-19 3:32 UTC (permalink / raw)
To: Daniel P. Berrange
Cc: qemu devel, Jason Wang, Li Zhijian, eddie . dong,
Dr . David Alan Gilbert, zhanghailiang
On 07/18/2016 04:37 PM, Daniel P. Berrange wrote:
> On Mon, Jul 18, 2016 at 03:40:49PM +0800, 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,
>> 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 | 216 +++++++++++++++++++++++++++++++++++++++++++++++++++++
>> trace-events | 2 +
>> 4 files changed, 222 insertions(+)
>>
>> +static void *colo_compare_thread(void *opaque)
>> +{
>> + GMainContext *worker_context;
>> + GMainLoop *compare_loop;
>> + CompareState *s = opaque;
>> +
>> + worker_context = g_main_context_new();
>> + g_assert(g_main_context_get_thread_default() == NULL);
>> + g_main_context_push_thread_default(worker_context);
>> + g_assert(g_main_context_get_thread_default() == worker_context);
>> +
>> + qemu_chr_add_handlers(s->chr_pri_in, compare_chr_can_read,
>> + compare_pri_chr_in, NULL, s);
>> + qemu_chr_add_handlers(s->chr_sec_in, compare_chr_can_read,
>> + compare_sec_chr_in, NULL, s);
> Looking at this I see you've used g_main_context_push_thread_default
> in order to avoid having to modify the qemu_chr_add_handlers() method
> to accept a GMainContext. Personally I think I'd favour explicit
> passing of a GMainContext, rather than modifying Qemu Char code &
> QIOChannel to magically use per-thread main context. In particular
> I'm concerned that one day other code may use the
> g_main_context_push_thread_default method for some other purpose
> but still want their char devs handled by the main thread. Your
> solution makes this impossible and adds magic which could surprise
> us in bad ways.
>
> So I think I'd rather see qemu_chr_add_handlers() modified to accept
> a GMainContext, or perhaps add a qemu_chr_add_handlers_full() method
> which takes a GMainContext, so we can avoid modifying all existing
> callers ofo qemu_chr_add_handlers. There should be no need to modify
> QIOChannel at all, since you can use qio_channel_create_watch instead
> of qio_channel_add_watch and simply attach to the context you want
> directly.
I get your point.
Considering current qemu code, I think add a flag
in CharDriverState to decide whether or not to
run g_source_attach(xxx, g_main_context_get_thread_default());
is a easy way for this.
How about this way?
Thanks for your review.
Zhang Chen
>
> Regards,
> Daniel
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 14+ messages in thread* Re: [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread
2016-07-19 3:32 ` Zhang Chen
@ 2016-07-19 9:03 ` Daniel P. Berrange
2016-07-20 3:29 ` Zhang Chen
0 siblings, 1 reply; 14+ messages in thread
From: Daniel P. Berrange @ 2016-07-19 9:03 UTC (permalink / raw)
To: Zhang Chen
Cc: qemu devel, Jason Wang, Li Zhijian, eddie . dong,
Dr . David Alan Gilbert, zhanghailiang
On Tue, Jul 19, 2016 at 11:32:41AM +0800, Zhang Chen wrote:
>
>
> On 07/18/2016 04:37 PM, Daniel P. Berrange wrote:
> > On Mon, Jul 18, 2016 at 03:40:49PM +0800, 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,
> > > 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 | 216 +++++++++++++++++++++++++++++++++++++++++++++++++++++
> > > trace-events | 2 +
> > > 4 files changed, 222 insertions(+)
> > >
> > > +static void *colo_compare_thread(void *opaque)
> > > +{
> > > + GMainContext *worker_context;
> > > + GMainLoop *compare_loop;
> > > + CompareState *s = opaque;
> > > +
> > > + worker_context = g_main_context_new();
> > > + g_assert(g_main_context_get_thread_default() == NULL);
> > > + g_main_context_push_thread_default(worker_context);
> > > + g_assert(g_main_context_get_thread_default() == worker_context);
> > > +
> > > + qemu_chr_add_handlers(s->chr_pri_in, compare_chr_can_read,
> > > + compare_pri_chr_in, NULL, s);
> > > + qemu_chr_add_handlers(s->chr_sec_in, compare_chr_can_read,
> > > + compare_sec_chr_in, NULL, s);
> > Looking at this I see you've used g_main_context_push_thread_default
> > in order to avoid having to modify the qemu_chr_add_handlers() method
> > to accept a GMainContext. Personally I think I'd favour explicit
> > passing of a GMainContext, rather than modifying Qemu Char code &
> > QIOChannel to magically use per-thread main context. In particular
> > I'm concerned that one day other code may use the
> > g_main_context_push_thread_default method for some other purpose
> > but still want their char devs handled by the main thread. Your
> > solution makes this impossible and adds magic which could surprise
> > us in bad ways.
> >
> > So I think I'd rather see qemu_chr_add_handlers() modified to accept
> > a GMainContext, or perhaps add a qemu_chr_add_handlers_full() method
> > which takes a GMainContext, so we can avoid modifying all existing
> > callers ofo qemu_chr_add_handlers. There should be no need to modify
> > QIOChannel at all, since you can use qio_channel_create_watch instead
> > of qio_channel_add_watch and simply attach to the context you want
> > directly.
>
> I get your point.
> Considering current qemu code, I think add a flag
> in CharDriverState to decide whether or not to
> run g_source_attach(xxx, g_main_context_get_thread_default());
> is a easy way for this.
> How about this way?
I really prefer to see the GMainContext passed in with the API(s)
which register callbacks, not provided via the backdoor via the
glib thread default global state, or via chardev global state.
Regards,
Daniel
--
|: http://berrange.com -o- http://www.flickr.com/photos/dberrange/ :|
|: http://libvirt.org -o- http://virt-manager.org :|
|: http://autobuild.org -o- http://search.cpan.org/~danberr/ :|
|: http://entangle-photo.org -o- http://live.gnome.org/gtk-vnc :|
^ permalink raw reply [flat|nested] 14+ messages in thread* Re: [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread
2016-07-19 9:03 ` Daniel P. Berrange
@ 2016-07-20 3:29 ` Zhang Chen
0 siblings, 0 replies; 14+ messages in thread
From: Zhang Chen @ 2016-07-20 3:29 UTC (permalink / raw)
To: Daniel P. Berrange
Cc: qemu devel, Jason Wang, Li Zhijian, eddie . dong,
Dr . David Alan Gilbert, zhanghailiang
On 07/19/2016 05:03 PM, Daniel P. Berrange wrote:
> On Tue, Jul 19, 2016 at 11:32:41AM +0800, Zhang Chen wrote:
>>
>> On 07/18/2016 04:37 PM, Daniel P. Berrange wrote:
>>> On Mon, Jul 18, 2016 at 03:40:49PM +0800, 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,
>>>> 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 | 216 +++++++++++++++++++++++++++++++++++++++++++++++++++++
>>>> trace-events | 2 +
>>>> 4 files changed, 222 insertions(+)
>>>>
>>>> +static void *colo_compare_thread(void *opaque)
>>>> +{
>>>> + GMainContext *worker_context;
>>>> + GMainLoop *compare_loop;
>>>> + CompareState *s = opaque;
>>>> +
>>>> + worker_context = g_main_context_new();
>>>> + g_assert(g_main_context_get_thread_default() == NULL);
>>>> + g_main_context_push_thread_default(worker_context);
>>>> + g_assert(g_main_context_get_thread_default() == worker_context);
>>>> +
>>>> + qemu_chr_add_handlers(s->chr_pri_in, compare_chr_can_read,
>>>> + compare_pri_chr_in, NULL, s);
>>>> + qemu_chr_add_handlers(s->chr_sec_in, compare_chr_can_read,
>>>> + compare_sec_chr_in, NULL, s);
>>> Looking at this I see you've used g_main_context_push_thread_default
>>> in order to avoid having to modify the qemu_chr_add_handlers() method
>>> to accept a GMainContext. Personally I think I'd favour explicit
>>> passing of a GMainContext, rather than modifying Qemu Char code &
>>> QIOChannel to magically use per-thread main context. In particular
>>> I'm concerned that one day other code may use the
>>> g_main_context_push_thread_default method for some other purpose
>>> but still want their char devs handled by the main thread. Your
>>> solution makes this impossible and adds magic which could surprise
>>> us in bad ways.
>>>
>>> So I think I'd rather see qemu_chr_add_handlers() modified to accept
>>> a GMainContext, or perhaps add a qemu_chr_add_handlers_full() method
>>> which takes a GMainContext, so we can avoid modifying all existing
>>> callers ofo qemu_chr_add_handlers. There should be no need to modify
>>> QIOChannel at all, since you can use qio_channel_create_watch instead
>>> of qio_channel_add_watch and simply attach to the context you want
>>> directly.
>> I get your point.
>> Considering current qemu code, I think add a flag
>> in CharDriverState to decide whether or not to
>> run g_source_attach(xxx, g_main_context_get_thread_default());
>> is a easy way for this.
>> How about this way?
> I really prefer to see the GMainContext passed in with the API(s)
> which register callbacks, not provided via the backdoor via the
> glib thread default global state, or via chardev global state.
>
OK,I will add this API in next version.
void qemu_chr_add_handlers_full(CharDriverState *s,
IOCanReadHandler *fd_can_read,
IOReadHandler *fd_read,
IOEventHandler *fd_event,
void *opaque,
GMainContext *context)
But maybe have some codes duplicate with qemu_chr_add_handlers(),
Do you have suggestion to avoid it?
Thanks
Zhang Chen
--
Thanks
zhangchen
^ permalink raw reply [flat|nested] 14+ messages in thread
* [Qemu-devel] [RFC PATCH V7 7/7] colo-compare: add TCP, UDP, ICMP packet comparison
2016-07-18 7:40 [Qemu-devel] [RFC PATCH V7 0/7] Introduce COLO-compare Zhang Chen
` (5 preceding siblings ...)
2016-07-18 7:40 ` [Qemu-devel] [RFC PATCH V7 6/7] colo-compare: introduce packet comparison thread Zhang Chen
@ 2016-07-18 7:40 ` Zhang Chen
6 siblings, 0 replies; 14+ messages in thread
From: Zhang Chen @ 2016-07-18 7:40 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 942e326..9737ec6 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] 14+ messages in thread