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 "chardev/char-fe.h"
29 #include "qemu/sockets.h"
30 #include "qapi-visit.h"
32 #include "sysemu/iothread.h"
34 #define TYPE_COLO_COMPARE "colo-compare"
35 #define COLO_COMPARE(obj) \
36 OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
38 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
39 #define MAX_QUEUE_SIZE 1024
41 /* TODO: Should be configurable */
42 #define REGULAR_PACKET_CHECK_MS 3000
47 * +---------------+ +---------------+ +---------------+
48 * | conn list + - > conn + ------- > conn + -- > ......
49 * +---------------+ +---------------+ +---------------+
51 * +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
52 * |primary | |secondary |primary | |secondary
53 * |packet | |packet + |packet | |packet +
54 * +--------+ +--------+ +--------+ +--------+
56 * +---v----+ +---v----+ +---v----+ +---v----+
57 * |primary | |secondary |primary | |secondary
58 * |packet | |packet + |packet | |packet +
59 * +--------+ +--------+ +--------+ +--------+
61 * +---v----+ +---v----+ +---v----+ +---v----+
62 * |primary | |secondary |primary | |secondary
63 * |packet | |packet + |packet | |packet +
64 * +--------+ +--------+ +--------+ +--------+
66 typedef struct CompareState
{
72 CharBackend chr_pri_in
;
73 CharBackend chr_sec_in
;
75 SocketReadState pri_rs
;
76 SocketReadState sec_rs
;
80 * Record the connection that through the NIC
81 * Element type: Connection
84 /* Record the connection without repetition */
85 GHashTable
*connection_track_table
;
88 GMainContext
*worker_context
;
89 QEMUTimer
*packet_check_timer
;
92 typedef struct CompareClass
{
93 ObjectClass parent_class
;
101 static int compare_chr_send(CompareState
*s
,
104 uint32_t vnet_hdr_len
);
106 static gint
seq_sorter(Packet
*a
, Packet
*b
, gpointer data
)
108 struct tcphdr
*atcp
, *btcp
;
110 atcp
= (struct tcphdr
*)(a
->transport_header
);
111 btcp
= (struct tcphdr
*)(b
->transport_header
);
112 return ntohl(atcp
->th_seq
) - ntohl(btcp
->th_seq
);
116 * Return 1 on success, if return 0 means the
117 * packet will be dropped
119 static int colo_insert_packet(GQueue
*queue
, Packet
*pkt
)
121 if (g_queue_get_length(queue
) <= MAX_QUEUE_SIZE
) {
122 if (pkt
->ip
->ip_p
== IPPROTO_TCP
) {
123 g_queue_insert_sorted(queue
,
125 (GCompareDataFunc
)seq_sorter
,
128 g_queue_push_tail(queue
, pkt
);
136 * Return 0 on success, if return -1 means the pkt
137 * is unsupported(arp and ipv6) and will be sent later
139 static int packet_enqueue(CompareState
*s
, int mode
, Connection
**con
)
145 if (mode
== PRIMARY_IN
) {
146 pkt
= packet_new(s
->pri_rs
.buf
,
147 s
->pri_rs
.packet_len
,
148 s
->pri_rs
.vnet_hdr_len
);
150 pkt
= packet_new(s
->sec_rs
.buf
,
151 s
->sec_rs
.packet_len
,
152 s
->sec_rs
.vnet_hdr_len
);
155 if (parse_packet_early(pkt
)) {
156 packet_destroy(pkt
, NULL
);
160 fill_connection_key(pkt
, &key
);
162 conn
= connection_get(s
->connection_track_table
,
166 if (!conn
->processing
) {
167 g_queue_push_tail(&s
->conn_list
, conn
);
168 conn
->processing
= true;
171 if (mode
== PRIMARY_IN
) {
172 if (!colo_insert_packet(&conn
->primary_list
, pkt
)) {
173 error_report("colo compare primary queue size too big,"
177 if (!colo_insert_packet(&conn
->secondary_list
, pkt
)) {
178 error_report("colo compare secondary queue size too big,"
188 * The IP packets sent by primary and secondary
189 * will be compared in here
190 * TODO support ip fragment, Out-Of-Order
191 * return: 0 means packet same
192 * > 0 || < 0 means packet different
194 static int colo_packet_compare_common(Packet
*ppkt
,
199 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE
)) {
200 char pri_ip_src
[20], pri_ip_dst
[20], sec_ip_src
[20], sec_ip_dst
[20];
202 strcpy(pri_ip_src
, inet_ntoa(ppkt
->ip
->ip_src
));
203 strcpy(pri_ip_dst
, inet_ntoa(ppkt
->ip
->ip_dst
));
204 strcpy(sec_ip_src
, inet_ntoa(spkt
->ip
->ip_src
));
205 strcpy(sec_ip_dst
, inet_ntoa(spkt
->ip
->ip_dst
));
207 trace_colo_compare_ip_info(ppkt
->size
, pri_ip_src
,
208 pri_ip_dst
, spkt
->size
,
209 sec_ip_src
, sec_ip_dst
);
212 poffset
= ppkt
->vnet_hdr_len
+ poffset
;
213 soffset
= ppkt
->vnet_hdr_len
+ soffset
;
215 if (ppkt
->size
- poffset
== spkt
->size
- soffset
) {
216 return memcmp(ppkt
->data
+ poffset
,
217 spkt
->data
+ soffset
,
218 spkt
->size
- soffset
);
220 trace_colo_compare_main("Net packet size are not the same");
226 * Called from the compare thread on the primary
227 * for compare tcp packet
228 * compare_tcp copied from Dr. David Alan Gilbert's branch
230 static int colo_packet_compare_tcp(Packet
*spkt
, Packet
*ppkt
)
232 struct tcphdr
*ptcp
, *stcp
;
235 trace_colo_compare_main("compare tcp");
237 ptcp
= (struct tcphdr
*)ppkt
->transport_header
;
238 stcp
= (struct tcphdr
*)spkt
->transport_header
;
241 * The 'identification' field in the IP header is *very* random
242 * it almost never matches. Fudge this by ignoring differences in
243 * unfragmented packets; they'll normally sort themselves out if different
244 * anyway, and it should recover at the TCP level.
245 * An alternative would be to get both the primary and secondary to rewrite
246 * somehow; but that would need some sync traffic to sync the state
248 if (ntohs(ppkt
->ip
->ip_off
) & IP_DF
) {
249 spkt
->ip
->ip_id
= ppkt
->ip
->ip_id
;
250 /* and the sum will be different if the IDs were different */
251 spkt
->ip
->ip_sum
= ppkt
->ip
->ip_sum
;
255 * Check tcp header length for tcp option field.
256 * th_off > 5 means this tcp packet have options field.
257 * The tcp options maybe always different.
260 * TCP Timestamps option (TSopt):
265 * +-------+-------+---------------------+---------------------+
266 * |Kind=8 | 10 | TS Value (TSval) |TS Echo Reply (TSecr)|
267 * +-------+-------+---------------------+---------------------+
270 * In this case the primary guest's timestamp always different with
271 * the secondary guest's timestamp. COLO just focus on payload,
272 * so we just need skip this field.
274 if (ptcp
->th_off
> 5) {
275 ptrdiff_t ptcp_offset
, stcp_offset
;
277 ptcp_offset
= ppkt
->transport_header
- (uint8_t *)ppkt
->data
278 + (ptcp
->th_off
* 4) - ppkt
->vnet_hdr_len
;
279 stcp_offset
= spkt
->transport_header
- (uint8_t *)spkt
->data
280 + (stcp
->th_off
* 4) - spkt
->vnet_hdr_len
;
283 * When network is busy, some tcp options(like sack) will unpredictable
284 * occur in primary side or secondary side. it will make packet size
285 * not same, but the two packet's payload is identical. colo just
286 * care about packet payload, so we skip the option field.
288 res
= colo_packet_compare_common(ppkt
, spkt
, ptcp_offset
, stcp_offset
);
289 } else if (ptcp
->th_sum
== stcp
->th_sum
) {
290 res
= colo_packet_compare_common(ppkt
, spkt
, ETH_HLEN
, ETH_HLEN
);
296 trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE
)) {
297 char pri_ip_src
[20], pri_ip_dst
[20], sec_ip_src
[20], sec_ip_dst
[20];
299 strcpy(pri_ip_src
, inet_ntoa(ppkt
->ip
->ip_src
));
300 strcpy(pri_ip_dst
, inet_ntoa(ppkt
->ip
->ip_dst
));
301 strcpy(sec_ip_src
, inet_ntoa(spkt
->ip
->ip_src
));
302 strcpy(sec_ip_dst
, inet_ntoa(spkt
->ip
->ip_dst
));
304 trace_colo_compare_ip_info(ppkt
->size
, pri_ip_src
,
305 pri_ip_dst
, spkt
->size
,
306 sec_ip_src
, sec_ip_dst
);
308 trace_colo_compare_tcp_info("pri tcp packet",
314 trace_colo_compare_tcp_info("sec tcp packet",
320 qemu_hexdump((char *)ppkt
->data
, stderr
,
321 "colo-compare ppkt", ppkt
->size
);
322 qemu_hexdump((char *)spkt
->data
, stderr
,
323 "colo-compare spkt", spkt
->size
);
330 * Called from the compare thread on the primary
331 * for compare udp packet
333 static int colo_packet_compare_udp(Packet
*spkt
, Packet
*ppkt
)
336 int network_header_length
= ppkt
->ip
->ip_hl
* 4;
338 trace_colo_compare_main("compare udp");
341 * Because of ppkt and spkt are both in the same connection,
342 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
343 * same with spkt. In addition, IP header's Identification is a random
344 * field, we can handle it in IP fragmentation function later.
345 * COLO just concern the response net packet payload from primary guest
346 * and secondary guest are same or not, So we ignored all IP header include
347 * other field like TOS,TTL,IP Checksum. we only need to compare
348 * the ip payload here.
350 ret
= colo_packet_compare_common(ppkt
, spkt
,
351 network_header_length
+ ETH_HLEN
,
352 network_header_length
+ ETH_HLEN
);
355 trace_colo_compare_udp_miscompare("primary pkt size", ppkt
->size
);
356 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt
->size
);
357 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE
)) {
358 qemu_hexdump((char *)ppkt
->data
, stderr
, "colo-compare pri pkt",
360 qemu_hexdump((char *)spkt
->data
, stderr
, "colo-compare sec pkt",
369 * Called from the compare thread on the primary
370 * for compare icmp packet
372 static int colo_packet_compare_icmp(Packet
*spkt
, Packet
*ppkt
)
374 int network_header_length
= ppkt
->ip
->ip_hl
* 4;
376 trace_colo_compare_main("compare icmp");
379 * Because of ppkt and spkt are both in the same connection,
380 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
381 * same with spkt. In addition, IP header's Identification is a random
382 * field, we can handle it in IP fragmentation function later.
383 * COLO just concern the response net packet payload from primary guest
384 * and secondary guest are same or not, So we ignored all IP header include
385 * other field like TOS,TTL,IP Checksum. we only need to compare
386 * the ip payload here.
388 if (colo_packet_compare_common(ppkt
, spkt
,
389 network_header_length
+ ETH_HLEN
,
390 network_header_length
+ ETH_HLEN
)) {
391 trace_colo_compare_icmp_miscompare("primary pkt size",
393 trace_colo_compare_icmp_miscompare("Secondary pkt size",
395 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE
)) {
396 qemu_hexdump((char *)ppkt
->data
, stderr
, "colo-compare pri pkt",
398 qemu_hexdump((char *)spkt
->data
, stderr
, "colo-compare sec pkt",
408 * Called from the compare thread on the primary
409 * for compare other packet
411 static int colo_packet_compare_other(Packet
*spkt
, Packet
*ppkt
)
413 trace_colo_compare_main("compare other");
414 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE
)) {
415 char pri_ip_src
[20], pri_ip_dst
[20], sec_ip_src
[20], sec_ip_dst
[20];
417 strcpy(pri_ip_src
, inet_ntoa(ppkt
->ip
->ip_src
));
418 strcpy(pri_ip_dst
, inet_ntoa(ppkt
->ip
->ip_dst
));
419 strcpy(sec_ip_src
, inet_ntoa(spkt
->ip
->ip_src
));
420 strcpy(sec_ip_dst
, inet_ntoa(spkt
->ip
->ip_dst
));
422 trace_colo_compare_ip_info(ppkt
->size
, pri_ip_src
,
423 pri_ip_dst
, spkt
->size
,
424 sec_ip_src
, sec_ip_dst
);
427 return colo_packet_compare_common(ppkt
, spkt
, 0, 0);
430 static int colo_old_packet_check_one(Packet
*pkt
, int64_t *check_time
)
432 int64_t now
= qemu_clock_get_ms(QEMU_CLOCK_HOST
);
434 if ((now
- pkt
->creation_ms
) > (*check_time
)) {
435 trace_colo_old_packet_check_found(pkt
->creation_ms
);
442 static int colo_old_packet_check_one_conn(Connection
*conn
,
445 GList
*result
= NULL
;
446 int64_t check_time
= REGULAR_PACKET_CHECK_MS
;
448 result
= g_queue_find_custom(&conn
->primary_list
,
450 (GCompareFunc
)colo_old_packet_check_one
);
453 /* Do checkpoint will flush old packet */
455 * TODO: Notify colo frame to do checkpoint.
456 * colo_compare_inconsistent_notify();
465 * Look for old packets that the secondary hasn't matched,
466 * if we have some then we have to checkpoint to wake
469 static void colo_old_packet_check(void *opaque
)
471 CompareState
*s
= opaque
;
474 * If we find one old packet, stop finding job and notify
475 * COLO frame do checkpoint.
477 g_queue_find_custom(&s
->conn_list
, NULL
,
478 (GCompareFunc
)colo_old_packet_check_one_conn
);
482 * Called from the compare thread on the primary
483 * for compare packet with secondary list of the
484 * specified connection when a new packet was
487 static void colo_compare_connection(void *opaque
, void *user_data
)
489 CompareState
*s
= user_data
;
490 Connection
*conn
= opaque
;
492 GList
*result
= NULL
;
495 while (!g_queue_is_empty(&conn
->primary_list
) &&
496 !g_queue_is_empty(&conn
->secondary_list
)) {
497 pkt
= g_queue_pop_head(&conn
->primary_list
);
498 switch (conn
->ip_proto
) {
500 result
= g_queue_find_custom(&conn
->secondary_list
,
501 pkt
, (GCompareFunc
)colo_packet_compare_tcp
);
504 result
= g_queue_find_custom(&conn
->secondary_list
,
505 pkt
, (GCompareFunc
)colo_packet_compare_udp
);
508 result
= g_queue_find_custom(&conn
->secondary_list
,
509 pkt
, (GCompareFunc
)colo_packet_compare_icmp
);
512 result
= g_queue_find_custom(&conn
->secondary_list
,
513 pkt
, (GCompareFunc
)colo_packet_compare_other
);
518 ret
= compare_chr_send(s
,
523 error_report("colo_send_primary_packet failed");
525 trace_colo_compare_main("packet same and release packet");
526 g_queue_remove(&conn
->secondary_list
, result
->data
);
527 packet_destroy(pkt
, NULL
);
530 * If one packet arrive late, the secondary_list or
531 * primary_list will be empty, so we can't compare it
532 * until next comparison.
534 trace_colo_compare_main("packet different");
535 g_queue_push_head(&conn
->primary_list
, pkt
);
536 /* TODO: colo_notify_checkpoint();*/
542 static int compare_chr_send(CompareState
*s
,
545 uint32_t vnet_hdr_len
)
548 uint32_t len
= htonl(size
);
554 ret
= qemu_chr_fe_write_all(&s
->chr_out
, (uint8_t *)&len
, sizeof(len
));
555 if (ret
!= sizeof(len
)) {
561 * We send vnet header len make other module(like filter-redirector)
562 * know how to parse net packet correctly.
564 len
= htonl(vnet_hdr_len
);
565 ret
= qemu_chr_fe_write_all(&s
->chr_out
, (uint8_t *)&len
, sizeof(len
));
566 if (ret
!= sizeof(len
)) {
571 ret
= qemu_chr_fe_write_all(&s
->chr_out
, (uint8_t *)buf
, size
);
579 return ret
< 0 ? ret
: -EIO
;
582 static int compare_chr_can_read(void *opaque
)
584 return COMPARE_READ_LEN_MAX
;
588 * Called from the main thread on the primary for packets
589 * arriving over the socket from the primary.
591 static void compare_pri_chr_in(void *opaque
, const uint8_t *buf
, int size
)
593 CompareState
*s
= COLO_COMPARE(opaque
);
596 ret
= net_fill_rstate(&s
->pri_rs
, buf
, size
);
598 qemu_chr_fe_set_handlers(&s
->chr_pri_in
, NULL
, NULL
, NULL
, NULL
,
600 error_report("colo-compare primary_in error");
605 * Called from the main thread on the primary for packets
606 * arriving over the socket from the secondary.
608 static void compare_sec_chr_in(void *opaque
, const uint8_t *buf
, int size
)
610 CompareState
*s
= COLO_COMPARE(opaque
);
613 ret
= net_fill_rstate(&s
->sec_rs
, buf
, size
);
615 qemu_chr_fe_set_handlers(&s
->chr_sec_in
, NULL
, NULL
, NULL
, NULL
,
617 error_report("colo-compare secondary_in error");
622 * Check old packet regularly so it can watch for any packets
623 * that the secondary hasn't produced equivalents of.
625 static void check_old_packet_regular(void *opaque
)
627 CompareState
*s
= opaque
;
629 /* if have old packet we will notify checkpoint */
630 colo_old_packet_check(s
);
631 timer_mod(s
->packet_check_timer
, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL
) +
632 REGULAR_PACKET_CHECK_MS
);
635 static void colo_compare_timer_init(CompareState
*s
)
637 AioContext
*ctx
= iothread_get_aio_context(s
->iothread
);
639 s
->packet_check_timer
= aio_timer_new(ctx
, QEMU_CLOCK_VIRTUAL
,
640 SCALE_MS
, check_old_packet_regular
,
642 timer_mod(s
->packet_check_timer
, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL
) +
643 REGULAR_PACKET_CHECK_MS
);
646 static void colo_compare_timer_del(CompareState
*s
)
648 if (s
->packet_check_timer
) {
649 timer_del(s
->packet_check_timer
);
650 timer_free(s
->packet_check_timer
);
651 s
->packet_check_timer
= NULL
;
655 static void colo_compare_iothread(CompareState
*s
)
657 object_ref(OBJECT(s
->iothread
));
658 s
->worker_context
= iothread_get_g_main_context(s
->iothread
);
660 qemu_chr_fe_set_handlers(&s
->chr_pri_in
, compare_chr_can_read
,
661 compare_pri_chr_in
, NULL
, NULL
,
662 s
, s
->worker_context
, true);
663 qemu_chr_fe_set_handlers(&s
->chr_sec_in
, compare_chr_can_read
,
664 compare_sec_chr_in
, NULL
, NULL
,
665 s
, s
->worker_context
, true);
667 colo_compare_timer_init(s
);
670 static char *compare_get_pri_indev(Object
*obj
, Error
**errp
)
672 CompareState
*s
= COLO_COMPARE(obj
);
674 return g_strdup(s
->pri_indev
);
677 static void compare_set_pri_indev(Object
*obj
, const char *value
, Error
**errp
)
679 CompareState
*s
= COLO_COMPARE(obj
);
681 g_free(s
->pri_indev
);
682 s
->pri_indev
= g_strdup(value
);
685 static char *compare_get_sec_indev(Object
*obj
, Error
**errp
)
687 CompareState
*s
= COLO_COMPARE(obj
);
689 return g_strdup(s
->sec_indev
);
692 static void compare_set_sec_indev(Object
*obj
, const char *value
, Error
**errp
)
694 CompareState
*s
= COLO_COMPARE(obj
);
696 g_free(s
->sec_indev
);
697 s
->sec_indev
= g_strdup(value
);
700 static char *compare_get_outdev(Object
*obj
, Error
**errp
)
702 CompareState
*s
= COLO_COMPARE(obj
);
704 return g_strdup(s
->outdev
);
707 static void compare_set_outdev(Object
*obj
, const char *value
, Error
**errp
)
709 CompareState
*s
= COLO_COMPARE(obj
);
712 s
->outdev
= g_strdup(value
);
715 static bool compare_get_vnet_hdr(Object
*obj
, Error
**errp
)
717 CompareState
*s
= COLO_COMPARE(obj
);
722 static void compare_set_vnet_hdr(Object
*obj
,
726 CompareState
*s
= COLO_COMPARE(obj
);
731 static void compare_pri_rs_finalize(SocketReadState
*pri_rs
)
733 CompareState
*s
= container_of(pri_rs
, CompareState
, pri_rs
);
734 Connection
*conn
= NULL
;
736 if (packet_enqueue(s
, PRIMARY_IN
, &conn
)) {
737 trace_colo_compare_main("primary: unsupported packet in");
741 pri_rs
->vnet_hdr_len
);
743 /* compare packet in the specified connection */
744 colo_compare_connection(conn
, s
);
748 static void compare_sec_rs_finalize(SocketReadState
*sec_rs
)
750 CompareState
*s
= container_of(sec_rs
, CompareState
, sec_rs
);
751 Connection
*conn
= NULL
;
753 if (packet_enqueue(s
, SECONDARY_IN
, &conn
)) {
754 trace_colo_compare_main("secondary: unsupported packet in");
756 /* compare packet in the specified connection */
757 colo_compare_connection(conn
, s
);
763 * Return 0 is success.
764 * Return 1 is failed.
766 static int find_and_check_chardev(Chardev
**chr
,
770 *chr
= qemu_chr_find(chr_name
);
772 error_setg(errp
, "Device '%s' not found",
777 if (!qemu_chr_has_feature(*chr
, QEMU_CHAR_FEATURE_RECONNECTABLE
)) {
778 error_setg(errp
, "chardev \"%s\" is not reconnectable",
787 * Called from the main thread on the primary
788 * to setup colo-compare.
790 static void colo_compare_complete(UserCreatable
*uc
, Error
**errp
)
792 CompareState
*s
= COLO_COMPARE(uc
);
795 if (!s
->pri_indev
|| !s
->sec_indev
|| !s
->outdev
|| !s
->iothread
) {
796 error_setg(errp
, "colo compare needs 'primary_in' ,"
797 "'secondary_in','outdev','iothread' property set");
799 } else if (!strcmp(s
->pri_indev
, s
->outdev
) ||
800 !strcmp(s
->sec_indev
, s
->outdev
) ||
801 !strcmp(s
->pri_indev
, s
->sec_indev
)) {
802 error_setg(errp
, "'indev' and 'outdev' could not be same "
803 "for compare module");
807 if (find_and_check_chardev(&chr
, s
->pri_indev
, errp
) ||
808 !qemu_chr_fe_init(&s
->chr_pri_in
, chr
, errp
)) {
812 if (find_and_check_chardev(&chr
, s
->sec_indev
, errp
) ||
813 !qemu_chr_fe_init(&s
->chr_sec_in
, chr
, errp
)) {
817 if (find_and_check_chardev(&chr
, s
->outdev
, errp
) ||
818 !qemu_chr_fe_init(&s
->chr_out
, chr
, errp
)) {
822 net_socket_rs_init(&s
->pri_rs
, compare_pri_rs_finalize
, s
->vnet_hdr
);
823 net_socket_rs_init(&s
->sec_rs
, compare_sec_rs_finalize
, s
->vnet_hdr
);
825 g_queue_init(&s
->conn_list
);
827 s
->connection_track_table
= g_hash_table_new_full(connection_key_hash
,
828 connection_key_equal
,
832 colo_compare_iothread(s
);
836 static void colo_flush_packets(void *opaque
, void *user_data
)
838 CompareState
*s
= user_data
;
839 Connection
*conn
= opaque
;
842 while (!g_queue_is_empty(&conn
->primary_list
)) {
843 pkt
= g_queue_pop_head(&conn
->primary_list
);
848 packet_destroy(pkt
, NULL
);
850 while (!g_queue_is_empty(&conn
->secondary_list
)) {
851 pkt
= g_queue_pop_head(&conn
->secondary_list
);
852 packet_destroy(pkt
, NULL
);
856 static void colo_compare_class_init(ObjectClass
*oc
, void *data
)
858 UserCreatableClass
*ucc
= USER_CREATABLE_CLASS(oc
);
860 ucc
->complete
= colo_compare_complete
;
863 static void colo_compare_init(Object
*obj
)
865 CompareState
*s
= COLO_COMPARE(obj
);
867 object_property_add_str(obj
, "primary_in",
868 compare_get_pri_indev
, compare_set_pri_indev
,
870 object_property_add_str(obj
, "secondary_in",
871 compare_get_sec_indev
, compare_set_sec_indev
,
873 object_property_add_str(obj
, "outdev",
874 compare_get_outdev
, compare_set_outdev
,
876 object_property_add_link(obj
, "iothread", TYPE_IOTHREAD
,
877 (Object
**)&s
->iothread
,
878 object_property_allow_set_link
,
879 OBJ_PROP_LINK_UNREF_ON_RELEASE
, NULL
);
882 object_property_add_bool(obj
, "vnet_hdr_support", compare_get_vnet_hdr
,
883 compare_set_vnet_hdr
, NULL
);
886 static void colo_compare_finalize(Object
*obj
)
888 CompareState
*s
= COLO_COMPARE(obj
);
890 qemu_chr_fe_deinit(&s
->chr_pri_in
, false);
891 qemu_chr_fe_deinit(&s
->chr_sec_in
, false);
892 qemu_chr_fe_deinit(&s
->chr_out
, false);
894 colo_compare_timer_del(s
);
896 /* Release all unhandled packets after compare thead exited */
897 g_queue_foreach(&s
->conn_list
, colo_flush_packets
, s
);
899 g_queue_clear(&s
->conn_list
);
901 if (s
->connection_track_table
) {
902 g_hash_table_destroy(s
->connection_track_table
);
906 object_unref(OBJECT(s
->iothread
));
908 g_free(s
->pri_indev
);
909 g_free(s
->sec_indev
);
913 static const TypeInfo colo_compare_info
= {
914 .name
= TYPE_COLO_COMPARE
,
915 .parent
= TYPE_OBJECT
,
916 .instance_size
= sizeof(CompareState
),
917 .instance_init
= colo_compare_init
,
918 .instance_finalize
= colo_compare_finalize
,
919 .class_size
= sizeof(CompareClass
),
920 .class_init
= colo_compare_class_init
,
921 .interfaces
= (InterfaceInfo
[]) {
922 { TYPE_USER_CREATABLE
},
927 static void register_types(void)
929 type_register_static(&colo_compare_info
);
932 type_init(register_types
);