2 * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
3 * (a.k.a. Fault Tolerance or Continuous Replication)
5 * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
6 * Copyright (c) 2016 FUJITSU LIMITED
7 * Copyright (c) 2016 Intel Corporation
9 * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
11 * This work is licensed under the terms of the GNU GPL, version 2 or
12 * later. See the COPYING file in the top-level directory.
15 #include "qemu/osdep.h"
16 #include "qemu/error-report.h"
18 #include "qemu-common.h"
19 #include "qapi/qmp/qerror.h"
20 #include "qapi/error.h"
23 #include "qom/object_interfaces.h"
25 #include "qom/object.h"
26 #include "qemu/typedefs.h"
27 #include "net/queue.h"
28 #include "sysemu/char.h"
29 #include "qemu/sockets.h"
30 #include "qapi-visit.h"
33 #define TYPE_COLO_COMPARE "colo-compare"
34 #define COLO_COMPARE(obj) \
35 OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
37 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
38 #define MAX_QUEUE_SIZE 1024
40 /* TODO: Should be configurable */
41 #define REGULAR_PACKET_CHECK_MS 3000
46 +---------------+ +---------------+ +---------------+
47 |conn list +--->conn +--------->conn |
48 +---------------+ +---------------+ +---------------+
50 +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
51 |primary | |secondary |primary | |secondary
52 |packet | |packet + |packet | |packet +
53 +--------+ +--------+ +--------+ +--------+
55 +---v----+ +---v----+ +---v----+ +---v----+
56 |primary | |secondary |primary | |secondary
57 |packet | |packet + |packet | |packet +
58 +--------+ +--------+ +--------+ +--------+
60 +---v----+ +---v----+ +---v----+ +---v----+
61 |primary | |secondary |primary | |secondary
62 |packet | |packet + |packet | |packet +
63 +--------+ +--------+ +--------+ +--------+
65 typedef struct CompareState
{
71 CharBackend chr_pri_in
;
72 CharBackend chr_sec_in
;
74 SocketReadState pri_rs
;
75 SocketReadState sec_rs
;
77 /* connection list: the connections belonged to this NIC could be found
79 * element type: Connection
82 /* hashtable to save connection */
83 GHashTable
*connection_track_table
;
84 /* compare thread, a thread for each NIC */
87 GMainContext
*worker_context
;
88 GMainLoop
*compare_loop
;
91 typedef struct CompareClass
{
92 ObjectClass parent_class
;
100 static int compare_chr_send(CharBackend
*out
,
104 static gint
seq_sorter(Packet
*a
, Packet
*b
, gpointer data
)
106 struct tcphdr
*atcp
, *btcp
;
108 atcp
= (struct tcphdr
*)(a
->transport_header
);
109 btcp
= (struct tcphdr
*)(b
->transport_header
);
110 return ntohl(atcp
->th_seq
) - ntohl(btcp
->th_seq
);
114 * Return 0 on success, if return -1 means the pkt
115 * is unsupported(arp and ipv6) and will be sent later
117 static int packet_enqueue(CompareState
*s
, int mode
)
123 if (mode
== PRIMARY_IN
) {
124 pkt
= packet_new(s
->pri_rs
.buf
, s
->pri_rs
.packet_len
);
126 pkt
= packet_new(s
->sec_rs
.buf
, s
->sec_rs
.packet_len
);
129 if (parse_packet_early(pkt
)) {
130 packet_destroy(pkt
, NULL
);
134 fill_connection_key(pkt
, &key
);
136 conn
= connection_get(s
->connection_track_table
,
140 if (!conn
->processing
) {
141 g_queue_push_tail(&s
->conn_list
, conn
);
142 conn
->processing
= true;
145 if (mode
== PRIMARY_IN
) {
146 if (g_queue_get_length(&conn
->primary_list
) <=
148 g_queue_push_tail(&conn
->primary_list
, pkt
);
149 if (conn
->ip_proto
== IPPROTO_TCP
) {
150 g_queue_sort(&conn
->primary_list
,
151 (GCompareDataFunc
)seq_sorter
,
155 error_report("colo compare primary queue size too big,"
159 if (g_queue_get_length(&conn
->secondary_list
) <=
161 g_queue_push_tail(&conn
->secondary_list
, pkt
);
162 if (conn
->ip_proto
== IPPROTO_TCP
) {
163 g_queue_sort(&conn
->secondary_list
,
164 (GCompareDataFunc
)seq_sorter
,
168 error_report("colo compare secondary queue size too big,"
177 * The IP packets sent by primary and secondary
178 * will be compared in here
179 * TODO support ip fragment, Out-Of-Order
180 * return: 0 means packet same
181 * > 0 || < 0 means packet different
183 static int colo_packet_compare(Packet
*ppkt
, Packet
*spkt
)
185 trace_colo_compare_ip_info(ppkt
->size
, inet_ntoa(ppkt
->ip
->ip_src
),
186 inet_ntoa(ppkt
->ip
->ip_dst
), spkt
->size
,
187 inet_ntoa(spkt
->ip
->ip_src
),
188 inet_ntoa(spkt
->ip
->ip_dst
));
190 if (ppkt
->size
== spkt
->size
) {
191 return memcmp(ppkt
->data
, spkt
->data
, spkt
->size
);
198 * Called from the compare thread on the primary
199 * for compare tcp packet
200 * compare_tcp copied from Dr. David Alan Gilbert's branch
202 static int colo_packet_compare_tcp(Packet
*spkt
, Packet
*ppkt
)
204 struct tcphdr
*ptcp
, *stcp
;
207 trace_colo_compare_main("compare tcp");
208 if (ppkt
->size
!= spkt
->size
) {
209 if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE
)) {
210 trace_colo_compare_main("pkt size not same");
215 ptcp
= (struct tcphdr
*)ppkt
->transport_header
;
216 stcp
= (struct tcphdr
*)spkt
->transport_header
;
219 * The 'identification' field in the IP header is *very* random
220 * it almost never matches. Fudge this by ignoring differences in
221 * unfragmented packets; they'll normally sort themselves out if different
222 * anyway, and it should recover at the TCP level.
223 * An alternative would be to get both the primary and secondary to rewrite
224 * somehow; but that would need some sync traffic to sync the state
226 if (ntohs(ppkt
->ip
->ip_off
) & IP_DF
) {
227 spkt
->ip
->ip_id
= ppkt
->ip
->ip_id
;
228 /* and the sum will be different if the IDs were different */
229 spkt
->ip
->ip_sum
= ppkt
->ip
->ip_sum
;
232 res
= memcmp(ppkt
->data
+ ETH_HLEN
, spkt
->data
+ ETH_HLEN
,
233 (spkt
->size
- ETH_HLEN
));
235 if (res
!= 0 && trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE
)) {
236 trace_colo_compare_pkt_info_src(inet_ntoa(ppkt
->ip
->ip_src
),
242 trace_colo_compare_pkt_info_dst(inet_ntoa(ppkt
->ip
->ip_dst
),
248 qemu_hexdump((char *)ppkt
->data
, stderr
,
249 "colo-compare ppkt", ppkt
->size
);
250 qemu_hexdump((char *)spkt
->data
, stderr
,
251 "colo-compare spkt", spkt
->size
);
258 * Called from the compare thread on the primary
259 * for compare udp packet
261 static int colo_packet_compare_udp(Packet
*spkt
, Packet
*ppkt
)
265 trace_colo_compare_main("compare udp");
266 ret
= colo_packet_compare(ppkt
, spkt
);
269 trace_colo_compare_udp_miscompare("primary pkt size", ppkt
->size
);
270 qemu_hexdump((char *)ppkt
->data
, stderr
, "colo-compare", ppkt
->size
);
271 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt
->size
);
272 qemu_hexdump((char *)spkt
->data
, stderr
, "colo-compare", spkt
->size
);
279 * Called from the compare thread on the primary
280 * for compare icmp packet
282 static int colo_packet_compare_icmp(Packet
*spkt
, Packet
*ppkt
)
286 trace_colo_compare_main("compare icmp");
287 network_length
= ppkt
->ip
->ip_hl
* 4;
288 if (ppkt
->size
!= spkt
->size
||
289 ppkt
->size
< network_length
+ ETH_HLEN
) {
293 if (colo_packet_compare(ppkt
, spkt
)) {
294 trace_colo_compare_icmp_miscompare("primary pkt size",
296 qemu_hexdump((char *)ppkt
->data
, stderr
, "colo-compare",
298 trace_colo_compare_icmp_miscompare("Secondary pkt size",
300 qemu_hexdump((char *)spkt
->data
, stderr
, "colo-compare",
309 * Called from the compare thread on the primary
310 * for compare other packet
312 static int colo_packet_compare_other(Packet
*spkt
, Packet
*ppkt
)
314 trace_colo_compare_main("compare other");
315 trace_colo_compare_ip_info(ppkt
->size
, inet_ntoa(ppkt
->ip
->ip_src
),
316 inet_ntoa(ppkt
->ip
->ip_dst
), spkt
->size
,
317 inet_ntoa(spkt
->ip
->ip_src
),
318 inet_ntoa(spkt
->ip
->ip_dst
));
319 return colo_packet_compare(ppkt
, spkt
);
322 static int colo_old_packet_check_one(Packet
*pkt
, int64_t *check_time
)
324 int64_t now
= qemu_clock_get_ms(QEMU_CLOCK_HOST
);
326 if ((now
- pkt
->creation_ms
) > (*check_time
)) {
327 trace_colo_old_packet_check_found(pkt
->creation_ms
);
334 static void colo_old_packet_check_one_conn(void *opaque
,
337 Connection
*conn
= opaque
;
338 GList
*result
= NULL
;
339 int64_t check_time
= REGULAR_PACKET_CHECK_MS
;
341 result
= g_queue_find_custom(&conn
->primary_list
,
343 (GCompareFunc
)colo_old_packet_check_one
);
346 /* do checkpoint will flush old packet */
347 /* TODO: colo_notify_checkpoint();*/
352 * Look for old packets that the secondary hasn't matched,
353 * if we have some then we have to checkpoint to wake
356 static void colo_old_packet_check(void *opaque
)
358 CompareState
*s
= opaque
;
360 g_queue_foreach(&s
->conn_list
, colo_old_packet_check_one_conn
, NULL
);
364 * Called from the compare thread on the primary
365 * for compare connection
367 static void colo_compare_connection(void *opaque
, void *user_data
)
369 CompareState
*s
= user_data
;
370 Connection
*conn
= opaque
;
372 GList
*result
= NULL
;
375 while (!g_queue_is_empty(&conn
->primary_list
) &&
376 !g_queue_is_empty(&conn
->secondary_list
)) {
377 pkt
= g_queue_pop_tail(&conn
->primary_list
);
378 switch (conn
->ip_proto
) {
380 result
= g_queue_find_custom(&conn
->secondary_list
,
381 pkt
, (GCompareFunc
)colo_packet_compare_tcp
);
384 result
= g_queue_find_custom(&conn
->secondary_list
,
385 pkt
, (GCompareFunc
)colo_packet_compare_udp
);
388 result
= g_queue_find_custom(&conn
->secondary_list
,
389 pkt
, (GCompareFunc
)colo_packet_compare_icmp
);
392 result
= g_queue_find_custom(&conn
->secondary_list
,
393 pkt
, (GCompareFunc
)colo_packet_compare_other
);
398 ret
= compare_chr_send(&s
->chr_out
, pkt
->data
, pkt
->size
);
400 error_report("colo_send_primary_packet failed");
402 trace_colo_compare_main("packet same and release packet");
403 g_queue_remove(&conn
->secondary_list
, result
->data
);
404 packet_destroy(pkt
, NULL
);
407 * If one packet arrive late, the secondary_list or
408 * primary_list will be empty, so we can't compare it
409 * until next comparison.
411 trace_colo_compare_main("packet different");
412 g_queue_push_tail(&conn
->primary_list
, pkt
);
413 /* TODO: colo_notify_checkpoint();*/
419 static int compare_chr_send(CharBackend
*out
,
424 uint32_t len
= htonl(size
);
430 ret
= qemu_chr_fe_write_all(out
, (uint8_t *)&len
, sizeof(len
));
431 if (ret
!= sizeof(len
)) {
435 ret
= qemu_chr_fe_write_all(out
, (uint8_t *)buf
, size
);
443 return ret
< 0 ? ret
: -EIO
;
446 static int compare_chr_can_read(void *opaque
)
448 return COMPARE_READ_LEN_MAX
;
452 * Called from the main thread on the primary for packets
453 * arriving over the socket from the primary.
455 static void compare_pri_chr_in(void *opaque
, const uint8_t *buf
, int size
)
457 CompareState
*s
= COLO_COMPARE(opaque
);
460 ret
= net_fill_rstate(&s
->pri_rs
, buf
, size
);
462 qemu_chr_fe_set_handlers(&s
->chr_pri_in
, NULL
, NULL
, NULL
,
464 error_report("colo-compare primary_in error");
469 * Called from the main thread on the primary for packets
470 * arriving over the socket from the secondary.
472 static void compare_sec_chr_in(void *opaque
, const uint8_t *buf
, int size
)
474 CompareState
*s
= COLO_COMPARE(opaque
);
477 ret
= net_fill_rstate(&s
->sec_rs
, buf
, size
);
479 qemu_chr_fe_set_handlers(&s
->chr_sec_in
, NULL
, NULL
, NULL
,
481 error_report("colo-compare secondary_in error");
486 * Check old packet regularly so it can watch for any packets
487 * that the secondary hasn't produced equivalents of.
489 static gboolean
check_old_packet_regular(void *opaque
)
491 CompareState
*s
= opaque
;
493 /* if have old packet we will notify checkpoint */
494 colo_old_packet_check(s
);
499 static void *colo_compare_thread(void *opaque
)
501 CompareState
*s
= opaque
;
502 GSource
*timeout_source
;
504 s
->worker_context
= g_main_context_new();
506 qemu_chr_fe_set_handlers(&s
->chr_pri_in
, compare_chr_can_read
,
507 compare_pri_chr_in
, NULL
, s
, s
->worker_context
, true);
508 qemu_chr_fe_set_handlers(&s
->chr_sec_in
, compare_chr_can_read
,
509 compare_sec_chr_in
, NULL
, s
, s
->worker_context
, true);
511 s
->compare_loop
= g_main_loop_new(s
->worker_context
, FALSE
);
513 /* To kick any packets that the secondary doesn't match */
514 timeout_source
= g_timeout_source_new(REGULAR_PACKET_CHECK_MS
);
515 g_source_set_callback(timeout_source
,
516 (GSourceFunc
)check_old_packet_regular
, s
, NULL
);
517 g_source_attach(timeout_source
, s
->worker_context
);
519 g_main_loop_run(s
->compare_loop
);
521 g_source_unref(timeout_source
);
522 g_main_loop_unref(s
->compare_loop
);
523 g_main_context_unref(s
->worker_context
);
527 static char *compare_get_pri_indev(Object
*obj
, Error
**errp
)
529 CompareState
*s
= COLO_COMPARE(obj
);
531 return g_strdup(s
->pri_indev
);
534 static void compare_set_pri_indev(Object
*obj
, const char *value
, Error
**errp
)
536 CompareState
*s
= COLO_COMPARE(obj
);
538 g_free(s
->pri_indev
);
539 s
->pri_indev
= g_strdup(value
);
542 static char *compare_get_sec_indev(Object
*obj
, Error
**errp
)
544 CompareState
*s
= COLO_COMPARE(obj
);
546 return g_strdup(s
->sec_indev
);
549 static void compare_set_sec_indev(Object
*obj
, const char *value
, Error
**errp
)
551 CompareState
*s
= COLO_COMPARE(obj
);
553 g_free(s
->sec_indev
);
554 s
->sec_indev
= g_strdup(value
);
557 static char *compare_get_outdev(Object
*obj
, Error
**errp
)
559 CompareState
*s
= COLO_COMPARE(obj
);
561 return g_strdup(s
->outdev
);
564 static void compare_set_outdev(Object
*obj
, const char *value
, Error
**errp
)
566 CompareState
*s
= COLO_COMPARE(obj
);
569 s
->outdev
= g_strdup(value
);
572 static void compare_pri_rs_finalize(SocketReadState
*pri_rs
)
574 CompareState
*s
= container_of(pri_rs
, CompareState
, pri_rs
);
576 if (packet_enqueue(s
, PRIMARY_IN
)) {
577 trace_colo_compare_main("primary: unsupported packet in");
578 compare_chr_send(&s
->chr_out
, pri_rs
->buf
, pri_rs
->packet_len
);
580 /* compare connection */
581 g_queue_foreach(&s
->conn_list
, colo_compare_connection
, s
);
585 static void compare_sec_rs_finalize(SocketReadState
*sec_rs
)
587 CompareState
*s
= container_of(sec_rs
, CompareState
, sec_rs
);
589 if (packet_enqueue(s
, SECONDARY_IN
)) {
590 trace_colo_compare_main("secondary: unsupported packet in");
592 /* compare connection */
593 g_queue_foreach(&s
->conn_list
, colo_compare_connection
, s
);
599 * Return 0 is success.
600 * Return 1 is failed.
602 static int find_and_check_chardev(Chardev
**chr
,
606 *chr
= qemu_chr_find(chr_name
);
608 error_setg(errp
, "Device '%s' not found",
613 if (!qemu_chr_has_feature(*chr
, QEMU_CHAR_FEATURE_RECONNECTABLE
)) {
614 error_setg(errp
, "chardev \"%s\" is not reconnectable",
623 * Called from the main thread on the primary
624 * to setup colo-compare.
626 static void colo_compare_complete(UserCreatable
*uc
, Error
**errp
)
628 CompareState
*s
= COLO_COMPARE(uc
);
630 char thread_name
[64];
631 static int compare_id
;
633 if (!s
->pri_indev
|| !s
->sec_indev
|| !s
->outdev
) {
634 error_setg(errp
, "colo compare needs 'primary_in' ,"
635 "'secondary_in','outdev' property set");
637 } else if (!strcmp(s
->pri_indev
, s
->outdev
) ||
638 !strcmp(s
->sec_indev
, s
->outdev
) ||
639 !strcmp(s
->pri_indev
, s
->sec_indev
)) {
640 error_setg(errp
, "'indev' and 'outdev' could not be same "
641 "for compare module");
645 if (find_and_check_chardev(&chr
, s
->pri_indev
, errp
) ||
646 !qemu_chr_fe_init(&s
->chr_pri_in
, chr
, errp
)) {
650 if (find_and_check_chardev(&chr
, s
->sec_indev
, errp
) ||
651 !qemu_chr_fe_init(&s
->chr_sec_in
, chr
, errp
)) {
655 if (find_and_check_chardev(&chr
, s
->outdev
, errp
) ||
656 !qemu_chr_fe_init(&s
->chr_out
, chr
, errp
)) {
660 net_socket_rs_init(&s
->pri_rs
, compare_pri_rs_finalize
);
661 net_socket_rs_init(&s
->sec_rs
, compare_sec_rs_finalize
);
663 g_queue_init(&s
->conn_list
);
665 s
->connection_track_table
= g_hash_table_new_full(connection_key_hash
,
666 connection_key_equal
,
670 sprintf(thread_name
, "colo-compare %d", compare_id
);
671 qemu_thread_create(&s
->thread
, thread_name
,
672 colo_compare_thread
, s
,
673 QEMU_THREAD_JOINABLE
);
679 static void colo_flush_packets(void *opaque
, void *user_data
)
681 CompareState
*s
= user_data
;
682 Connection
*conn
= opaque
;
685 while (!g_queue_is_empty(&conn
->primary_list
)) {
686 pkt
= g_queue_pop_head(&conn
->primary_list
);
687 compare_chr_send(&s
->chr_out
, pkt
->data
, pkt
->size
);
688 packet_destroy(pkt
, NULL
);
690 while (!g_queue_is_empty(&conn
->secondary_list
)) {
691 pkt
= g_queue_pop_head(&conn
->secondary_list
);
692 packet_destroy(pkt
, NULL
);
696 static void colo_compare_class_init(ObjectClass
*oc
, void *data
)
698 UserCreatableClass
*ucc
= USER_CREATABLE_CLASS(oc
);
700 ucc
->complete
= colo_compare_complete
;
703 static void colo_compare_init(Object
*obj
)
705 object_property_add_str(obj
, "primary_in",
706 compare_get_pri_indev
, compare_set_pri_indev
,
708 object_property_add_str(obj
, "secondary_in",
709 compare_get_sec_indev
, compare_set_sec_indev
,
711 object_property_add_str(obj
, "outdev",
712 compare_get_outdev
, compare_set_outdev
,
716 static void colo_compare_finalize(Object
*obj
)
718 CompareState
*s
= COLO_COMPARE(obj
);
720 qemu_chr_fe_set_handlers(&s
->chr_pri_in
, NULL
, NULL
, NULL
, NULL
,
721 s
->worker_context
, true);
722 qemu_chr_fe_set_handlers(&s
->chr_sec_in
, NULL
, NULL
, NULL
, NULL
,
723 s
->worker_context
, true);
724 qemu_chr_fe_deinit(&s
->chr_out
);
726 g_main_loop_quit(s
->compare_loop
);
727 qemu_thread_join(&s
->thread
);
729 /* Release all unhandled packets after compare thead exited */
730 g_queue_foreach(&s
->conn_list
, colo_flush_packets
, s
);
732 g_queue_clear(&s
->conn_list
);
734 g_hash_table_destroy(s
->connection_track_table
);
735 g_free(s
->pri_indev
);
736 g_free(s
->sec_indev
);
740 static const TypeInfo colo_compare_info
= {
741 .name
= TYPE_COLO_COMPARE
,
742 .parent
= TYPE_OBJECT
,
743 .instance_size
= sizeof(CompareState
),
744 .instance_init
= colo_compare_init
,
745 .instance_finalize
= colo_compare_finalize
,
746 .class_size
= sizeof(CompareClass
),
747 .class_init
= colo_compare_class_init
,
748 .interfaces
= (InterfaceInfo
[]) {
749 { TYPE_USER_CREATABLE
},
754 static void register_types(void)
756 type_register_static(&colo_compare_info
);
759 type_init(register_types
);