xref: /openbmc/qemu/net/colo-compare.c (revision 05caa062)
1 /*
2  * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
3  * (a.k.a. Fault Tolerance or Continuous Replication)
4  *
5  * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
6  * Copyright (c) 2016 FUJITSU LIMITED
7  * Copyright (c) 2016 Intel Corporation
8  *
9  * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
10  *
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.
13  */
14 
15 #include "qemu/osdep.h"
16 #include "qemu/error-report.h"
17 #include "trace.h"
18 #include "qapi/error.h"
19 #include "net/net.h"
20 #include "net/eth.h"
21 #include "qom/object_interfaces.h"
22 #include "qemu/iov.h"
23 #include "qom/object.h"
24 #include "net/queue.h"
25 #include "chardev/char-fe.h"
26 #include "qemu/sockets.h"
27 #include "colo.h"
28 #include "sysemu/iothread.h"
29 #include "net/colo-compare.h"
30 #include "migration/colo.h"
31 #include "util.h"
32 
33 #include "block/aio-wait.h"
34 #include "qemu/coroutine.h"
35 
36 #define TYPE_COLO_COMPARE "colo-compare"
37 typedef struct CompareState CompareState;
38 DECLARE_INSTANCE_CHECKER(CompareState, COLO_COMPARE,
39                          TYPE_COLO_COMPARE)
40 
41 static QTAILQ_HEAD(, CompareState) net_compares =
42        QTAILQ_HEAD_INITIALIZER(net_compares);
43 
44 static NotifierList colo_compare_notifiers =
45     NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
46 
47 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
48 #define MAX_QUEUE_SIZE 1024
49 
50 #define COLO_COMPARE_FREE_PRIMARY     0x01
51 #define COLO_COMPARE_FREE_SECONDARY   0x02
52 
53 #define REGULAR_PACKET_CHECK_MS 1000
54 #define DEFAULT_TIME_OUT_MS 3000
55 
56 /* #define DEBUG_COLO_PACKETS */
57 
58 static QemuMutex colo_compare_mutex;
59 static bool colo_compare_active;
60 static QemuMutex event_mtx;
61 static QemuCond event_complete_cond;
62 static int event_unhandled_count;
63 static uint32_t max_queue_size;
64 
65 /*
66  *  + CompareState ++
67  *  |               |
68  *  +---------------+   +---------------+         +---------------+
69  *  |   conn list   + - >      conn     + ------- >      conn     + -- > ......
70  *  +---------------+   +---------------+         +---------------+
71  *  |               |     |           |             |          |
72  *  +---------------+ +---v----+  +---v----+    +---v----+ +---v----+
73  *                    |primary |  |secondary    |primary | |secondary
74  *                    |packet  |  |packet  +    |packet  | |packet  +
75  *                    +--------+  +--------+    +--------+ +--------+
76  *                        |           |             |          |
77  *                    +---v----+  +---v----+    +---v----+ +---v----+
78  *                    |primary |  |secondary    |primary | |secondary
79  *                    |packet  |  |packet  +    |packet  | |packet  +
80  *                    +--------+  +--------+    +--------+ +--------+
81  *                        |           |             |          |
82  *                    +---v----+  +---v----+    +---v----+ +---v----+
83  *                    |primary |  |secondary    |primary | |secondary
84  *                    |packet  |  |packet  +    |packet  | |packet  +
85  *                    +--------+  +--------+    +--------+ +--------+
86  */
87 
88 typedef struct SendCo {
89     Coroutine *co;
90     struct CompareState *s;
91     CharBackend *chr;
92     GQueue send_list;
93     bool notify_remote_frame;
94     bool done;
95     int ret;
96 } SendCo;
97 
98 typedef struct SendEntry {
99     uint32_t size;
100     uint32_t vnet_hdr_len;
101     uint8_t *buf;
102 } SendEntry;
103 
104 struct CompareState {
105     Object parent;
106 
107     char *pri_indev;
108     char *sec_indev;
109     char *outdev;
110     char *notify_dev;
111     CharBackend chr_pri_in;
112     CharBackend chr_sec_in;
113     CharBackend chr_out;
114     CharBackend chr_notify_dev;
115     SocketReadState pri_rs;
116     SocketReadState sec_rs;
117     SocketReadState notify_rs;
118     SendCo out_sendco;
119     SendCo notify_sendco;
120     bool vnet_hdr;
121     uint64_t compare_timeout;
122     uint32_t expired_scan_cycle;
123 
124     /*
125      * Record the connection that through the NIC
126      * Element type: Connection
127      */
128     GQueue conn_list;
129     /* Record the connection without repetition */
130     GHashTable *connection_track_table;
131 
132     IOThread *iothread;
133     GMainContext *worker_context;
134     QEMUTimer *packet_check_timer;
135 
136     QEMUBH *event_bh;
137     enum colo_event event;
138 
139     QTAILQ_ENTRY(CompareState) next;
140 };
141 
142 typedef struct CompareClass {
143     ObjectClass parent_class;
144 } CompareClass;
145 
146 enum {
147     PRIMARY_IN = 0,
148     SECONDARY_IN,
149 };
150 
151 static const char *colo_mode[] = {
152     [PRIMARY_IN] = "primary",
153     [SECONDARY_IN] = "secondary",
154 };
155 
156 static int compare_chr_send(CompareState *s,
157                             uint8_t *buf,
158                             uint32_t size,
159                             uint32_t vnet_hdr_len,
160                             bool notify_remote_frame,
161                             bool zero_copy);
162 
163 static bool packet_matches_str(const char *str,
164                                const uint8_t *buf,
165                                uint32_t packet_len)
166 {
167     if (packet_len != strlen(str)) {
168         return false;
169     }
170 
171     return !memcmp(str, buf, packet_len);
172 }
173 
174 static void notify_remote_frame(CompareState *s)
175 {
176     char msg[] = "DO_CHECKPOINT";
177     int ret = 0;
178 
179     ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
180     if (ret < 0) {
181         error_report("Notify Xen COLO-frame failed");
182     }
183 }
184 
185 static void colo_compare_inconsistency_notify(CompareState *s)
186 {
187     if (s->notify_dev) {
188         notify_remote_frame(s);
189     } else {
190         notifier_list_notify(&colo_compare_notifiers,
191                              NULL);
192     }
193 }
194 
195 /* Use restricted to colo_insert_packet() */
196 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
197 {
198     return b->tcp_seq - a->tcp_seq;
199 }
200 
201 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
202 {
203     Packet *pkt = data;
204     struct tcp_hdr *tcphd;
205 
206     tcphd = (struct tcp_hdr *)pkt->transport_header;
207 
208     pkt->tcp_seq = ntohl(tcphd->th_seq);
209     pkt->tcp_ack = ntohl(tcphd->th_ack);
210     /* Need to consider ACK will bigger than uint32_t MAX */
211     *max_ack = pkt->tcp_ack - *max_ack > 0 ? pkt->tcp_ack : *max_ack;
212     pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
213                        + (tcphd->th_off << 2);
214     pkt->payload_size = pkt->size - pkt->header_size;
215     pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
216     pkt->flags = tcphd->th_flags;
217 }
218 
219 /*
220  * Return 1 on success, if return 0 means the
221  * packet will be dropped
222  */
223 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
224 {
225     if (g_queue_get_length(queue) <= max_queue_size) {
226         if (pkt->ip->ip_p == IPPROTO_TCP) {
227             fill_pkt_tcp_info(pkt, max_ack);
228             g_queue_insert_sorted(queue,
229                                   pkt,
230                                   (GCompareDataFunc)seq_sorter,
231                                   NULL);
232         } else {
233             g_queue_push_tail(queue, pkt);
234         }
235         return 1;
236     }
237     return 0;
238 }
239 
240 /*
241  * Return 0 on success, if return -1 means the pkt
242  * is unsupported(arp and ipv6) and will be sent later
243  */
244 static int packet_enqueue(CompareState *s, int mode, Connection **con)
245 {
246     ConnectionKey key;
247     Packet *pkt = NULL;
248     Connection *conn;
249     int ret;
250 
251     if (mode == PRIMARY_IN) {
252         pkt = packet_new(s->pri_rs.buf,
253                          s->pri_rs.packet_len,
254                          s->pri_rs.vnet_hdr_len);
255     } else {
256         pkt = packet_new(s->sec_rs.buf,
257                          s->sec_rs.packet_len,
258                          s->sec_rs.vnet_hdr_len);
259     }
260 
261     if (parse_packet_early(pkt)) {
262         packet_destroy(pkt, NULL);
263         pkt = NULL;
264         return -1;
265     }
266     fill_connection_key(pkt, &key, false);
267 
268     conn = connection_get(s->connection_track_table,
269                           &key,
270                           &s->conn_list);
271 
272     if (!conn->processing) {
273         g_queue_push_tail(&s->conn_list, conn);
274         conn->processing = true;
275     }
276 
277     if (mode == PRIMARY_IN) {
278         ret = colo_insert_packet(&conn->primary_list, pkt, &conn->pack);
279     } else {
280         ret = colo_insert_packet(&conn->secondary_list, pkt, &conn->sack);
281     }
282 
283     if (!ret) {
284         trace_colo_compare_drop_packet(colo_mode[mode],
285             "queue size too big, drop packet");
286         packet_destroy(pkt, NULL);
287         pkt = NULL;
288     }
289 
290     *con = conn;
291 
292     return 0;
293 }
294 
295 static inline bool after(uint32_t seq1, uint32_t seq2)
296 {
297         return (int32_t)(seq1 - seq2) > 0;
298 }
299 
300 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
301 {
302     int ret;
303     ret = compare_chr_send(s,
304                            pkt->data,
305                            pkt->size,
306                            pkt->vnet_hdr_len,
307                            false,
308                            true);
309     if (ret < 0) {
310         error_report("colo send primary packet failed");
311     }
312     trace_colo_compare_main("packet same and release packet");
313     packet_destroy_partial(pkt, NULL);
314 }
315 
316 /*
317  * The IP packets sent by primary and secondary
318  * will be compared in here
319  * TODO support ip fragment, Out-Of-Order
320  * return:    0  means packet same
321  *            > 0 || < 0 means packet different
322  */
323 static int colo_compare_packet_payload(Packet *ppkt,
324                                        Packet *spkt,
325                                        uint16_t poffset,
326                                        uint16_t soffset,
327                                        uint16_t len)
328 
329 {
330     if (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
331         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
332 
333         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
334         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
335         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
336         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
337 
338         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
339                                    pri_ip_dst, spkt->size,
340                                    sec_ip_src, sec_ip_dst);
341     }
342 
343     return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
344 }
345 
346 /*
347  * return true means that the payload is consist and
348  * need to make the next comparison, false means do
349  * the checkpoint
350 */
351 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
352                               int8_t *mark, uint32_t max_ack)
353 {
354     *mark = 0;
355 
356     if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
357         if (!colo_compare_packet_payload(ppkt, spkt,
358                                         ppkt->header_size, spkt->header_size,
359                                         ppkt->payload_size)) {
360             *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
361             return true;
362         }
363     }
364 
365     /* one part of secondary packet payload still need to be compared */
366     if (!after(ppkt->seq_end, spkt->seq_end)) {
367         if (!colo_compare_packet_payload(ppkt, spkt,
368                                         ppkt->header_size + ppkt->offset,
369                                         spkt->header_size + spkt->offset,
370                                         ppkt->payload_size - ppkt->offset)) {
371             if (!after(ppkt->tcp_ack, max_ack)) {
372                 *mark = COLO_COMPARE_FREE_PRIMARY;
373                 spkt->offset += ppkt->payload_size - ppkt->offset;
374                 return true;
375             } else {
376                 /* secondary guest hasn't ack the data, don't send
377                  * out this packet
378                  */
379                 return false;
380             }
381         }
382     } else {
383         /* primary packet is longer than secondary packet, compare
384          * the same part and mark the primary packet offset
385          */
386         if (!colo_compare_packet_payload(ppkt, spkt,
387                                         ppkt->header_size + ppkt->offset,
388                                         spkt->header_size + spkt->offset,
389                                         spkt->payload_size - spkt->offset)) {
390             *mark = COLO_COMPARE_FREE_SECONDARY;
391             ppkt->offset += spkt->payload_size - spkt->offset;
392             return true;
393         }
394     }
395 
396     return false;
397 }
398 
399 static void colo_compare_tcp(CompareState *s, Connection *conn)
400 {
401     Packet *ppkt = NULL, *spkt = NULL;
402     int8_t mark;
403 
404     /*
405      * If ppkt and spkt have the same payload, but ppkt's ACK
406      * is greater than spkt's ACK, in this case we can not
407      * send the ppkt because it will cause the secondary guest
408      * to miss sending some data in the next. Therefore, we
409      * record the maximum ACK in the current queue at both
410      * primary side and secondary side. Only when the ack is
411      * less than the smaller of the two maximum ack, then we
412      * can ensure that the packet's payload is acknowledged by
413      * primary and secondary.
414     */
415     uint32_t min_ack = conn->pack - conn->sack > 0 ?
416                        conn->sack : conn->pack;
417 
418 pri:
419     if (g_queue_is_empty(&conn->primary_list)) {
420         return;
421     }
422     ppkt = g_queue_pop_tail(&conn->primary_list);
423 sec:
424     if (g_queue_is_empty(&conn->secondary_list)) {
425         g_queue_push_tail(&conn->primary_list, ppkt);
426         return;
427     }
428     spkt = g_queue_pop_tail(&conn->secondary_list);
429 
430     if (ppkt->tcp_seq == ppkt->seq_end) {
431         colo_release_primary_pkt(s, ppkt);
432         ppkt = NULL;
433     }
434 
435     if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
436         trace_colo_compare_main("pri: this packet has compared");
437         colo_release_primary_pkt(s, ppkt);
438         ppkt = NULL;
439     }
440 
441     if (spkt->tcp_seq == spkt->seq_end) {
442         packet_destroy(spkt, NULL);
443         if (!ppkt) {
444             goto pri;
445         } else {
446             goto sec;
447         }
448     } else {
449         if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
450             trace_colo_compare_main("sec: this packet has compared");
451             packet_destroy(spkt, NULL);
452             if (!ppkt) {
453                 goto pri;
454             } else {
455                 goto sec;
456             }
457         }
458         if (!ppkt) {
459             g_queue_push_tail(&conn->secondary_list, spkt);
460             goto pri;
461         }
462     }
463 
464     if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
465         trace_colo_compare_tcp_info("pri",
466                                     ppkt->tcp_seq, ppkt->tcp_ack,
467                                     ppkt->header_size, ppkt->payload_size,
468                                     ppkt->offset, ppkt->flags);
469 
470         trace_colo_compare_tcp_info("sec",
471                                     spkt->tcp_seq, spkt->tcp_ack,
472                                     spkt->header_size, spkt->payload_size,
473                                     spkt->offset, spkt->flags);
474 
475         if (mark == COLO_COMPARE_FREE_PRIMARY) {
476             conn->compare_seq = ppkt->seq_end;
477             colo_release_primary_pkt(s, ppkt);
478             g_queue_push_tail(&conn->secondary_list, spkt);
479             goto pri;
480         } else if (mark == COLO_COMPARE_FREE_SECONDARY) {
481             conn->compare_seq = spkt->seq_end;
482             packet_destroy(spkt, NULL);
483             goto sec;
484         } else if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
485             conn->compare_seq = ppkt->seq_end;
486             colo_release_primary_pkt(s, ppkt);
487             packet_destroy(spkt, NULL);
488             goto pri;
489         }
490     } else {
491         g_queue_push_tail(&conn->primary_list, ppkt);
492         g_queue_push_tail(&conn->secondary_list, spkt);
493 
494 #ifdef DEBUG_COLO_PACKETS
495         qemu_hexdump(stderr, "colo-compare ppkt", ppkt->data, ppkt->size);
496         qemu_hexdump(stderr, "colo-compare spkt", spkt->data, spkt->size);
497 #endif
498 
499         colo_compare_inconsistency_notify(s);
500     }
501 }
502 
503 
504 /*
505  * Called from the compare thread on the primary
506  * for compare udp packet
507  */
508 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
509 {
510     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
511     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
512 
513     trace_colo_compare_main("compare udp");
514 
515     /*
516      * Because of ppkt and spkt are both in the same connection,
517      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
518      * same with spkt. In addition, IP header's Identification is a random
519      * field, we can handle it in IP fragmentation function later.
520      * COLO just concern the response net packet payload from primary guest
521      * and secondary guest are same or not, So we ignored all IP header include
522      * other field like TOS,TTL,IP Checksum. we only need to compare
523      * the ip payload here.
524      */
525     if (ppkt->size != spkt->size) {
526         trace_colo_compare_main("UDP: payload size of packets are different");
527         return -1;
528     }
529     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
530                                     ppkt->size - offset)) {
531         trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
532         trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
533 #ifdef DEBUG_COLO_PACKETS
534         qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
535         qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
536 #endif
537         return -1;
538     } else {
539         return 0;
540     }
541 }
542 
543 /*
544  * Called from the compare thread on the primary
545  * for compare icmp packet
546  */
547 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
548 {
549     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
550     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
551 
552     trace_colo_compare_main("compare icmp");
553 
554     /*
555      * Because of ppkt and spkt are both in the same connection,
556      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
557      * same with spkt. In addition, IP header's Identification is a random
558      * field, we can handle it in IP fragmentation function later.
559      * COLO just concern the response net packet payload from primary guest
560      * and secondary guest are same or not, So we ignored all IP header include
561      * other field like TOS,TTL,IP Checksum. we only need to compare
562      * the ip payload here.
563      */
564     if (ppkt->size != spkt->size) {
565         trace_colo_compare_main("ICMP: payload size of packets are different");
566         return -1;
567     }
568     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
569                                     ppkt->size - offset)) {
570         trace_colo_compare_icmp_miscompare("primary pkt size",
571                                            ppkt->size);
572         trace_colo_compare_icmp_miscompare("Secondary pkt size",
573                                            spkt->size);
574 #ifdef DEBUG_COLO_PACKETS
575         qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
576         qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
577 #endif
578         return -1;
579     } else {
580         return 0;
581     }
582 }
583 
584 /*
585  * Called from the compare thread on the primary
586  * for compare other packet
587  */
588 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
589 {
590     uint16_t offset = ppkt->vnet_hdr_len;
591 
592     trace_colo_compare_main("compare other");
593     if (ppkt->size != spkt->size) {
594         trace_colo_compare_main("Other: payload size of packets are different");
595         return -1;
596     }
597     return colo_compare_packet_payload(ppkt, spkt, offset, offset,
598                                        ppkt->size - offset);
599 }
600 
601 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
602 {
603     int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
604 
605     if ((now - pkt->creation_ms) > (*check_time)) {
606         trace_colo_old_packet_check_found(pkt->creation_ms);
607         return 0;
608     } else {
609         return 1;
610     }
611 }
612 
613 void colo_compare_register_notifier(Notifier *notify)
614 {
615     notifier_list_add(&colo_compare_notifiers, notify);
616 }
617 
618 void colo_compare_unregister_notifier(Notifier *notify)
619 {
620     notifier_remove(notify);
621 }
622 
623 static int colo_old_packet_check_one_conn(Connection *conn,
624                                           CompareState *s)
625 {
626     if (!g_queue_is_empty(&conn->primary_list)) {
627         if (g_queue_find_custom(&conn->primary_list,
628                                 &s->compare_timeout,
629                                 (GCompareFunc)colo_old_packet_check_one))
630             goto out;
631     }
632 
633     if (!g_queue_is_empty(&conn->secondary_list)) {
634         if (g_queue_find_custom(&conn->secondary_list,
635                                 &s->compare_timeout,
636                                 (GCompareFunc)colo_old_packet_check_one))
637             goto out;
638     }
639 
640     return 1;
641 
642 out:
643     /* Do checkpoint will flush old packet */
644     colo_compare_inconsistency_notify(s);
645     return 0;
646 }
647 
648 /*
649  * Look for old packets that the secondary hasn't matched,
650  * if we have some then we have to checkpoint to wake
651  * the secondary up.
652  */
653 static void colo_old_packet_check(void *opaque)
654 {
655     CompareState *s = opaque;
656 
657     /*
658      * If we find one old packet, stop finding job and notify
659      * COLO frame do checkpoint.
660      */
661     g_queue_find_custom(&s->conn_list, s,
662                         (GCompareFunc)colo_old_packet_check_one_conn);
663 }
664 
665 static void colo_compare_packet(CompareState *s, Connection *conn,
666                                 int (*HandlePacket)(Packet *spkt,
667                                 Packet *ppkt))
668 {
669     Packet *pkt = NULL;
670     GList *result = NULL;
671 
672     while (!g_queue_is_empty(&conn->primary_list) &&
673            !g_queue_is_empty(&conn->secondary_list)) {
674         pkt = g_queue_pop_tail(&conn->primary_list);
675         result = g_queue_find_custom(&conn->secondary_list,
676                  pkt, (GCompareFunc)HandlePacket);
677 
678         if (result) {
679             colo_release_primary_pkt(s, pkt);
680             packet_destroy(result->data, NULL);
681             g_queue_delete_link(&conn->secondary_list, result);
682         } else {
683             /*
684              * If one packet arrive late, the secondary_list or
685              * primary_list will be empty, so we can't compare it
686              * until next comparison. If the packets in the list are
687              * timeout, it will trigger a checkpoint request.
688              */
689             trace_colo_compare_main("packet different");
690             g_queue_push_tail(&conn->primary_list, pkt);
691 
692             colo_compare_inconsistency_notify(s);
693             break;
694         }
695     }
696 }
697 
698 /*
699  * Called from the compare thread on the primary
700  * for compare packet with secondary list of the
701  * specified connection when a new packet was
702  * queued to it.
703  */
704 static void colo_compare_connection(void *opaque, void *user_data)
705 {
706     CompareState *s = user_data;
707     Connection *conn = opaque;
708 
709     switch (conn->ip_proto) {
710     case IPPROTO_TCP:
711         colo_compare_tcp(s, conn);
712         break;
713     case IPPROTO_UDP:
714         colo_compare_packet(s, conn, colo_packet_compare_udp);
715         break;
716     case IPPROTO_ICMP:
717         colo_compare_packet(s, conn, colo_packet_compare_icmp);
718         break;
719     default:
720         colo_compare_packet(s, conn, colo_packet_compare_other);
721         break;
722     }
723 }
724 
725 static void coroutine_fn _compare_chr_send(void *opaque)
726 {
727     SendCo *sendco = opaque;
728     CompareState *s = sendco->s;
729     int ret = 0;
730 
731     while (!g_queue_is_empty(&sendco->send_list)) {
732         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
733         uint32_t len = htonl(entry->size);
734 
735         ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
736 
737         if (ret != sizeof(len)) {
738             g_free(entry->buf);
739             g_slice_free(SendEntry, entry);
740             goto err;
741         }
742 
743         if (!sendco->notify_remote_frame && s->vnet_hdr) {
744             /*
745              * We send vnet header len make other module(like filter-redirector)
746              * know how to parse net packet correctly.
747              */
748             len = htonl(entry->vnet_hdr_len);
749 
750             ret = qemu_chr_fe_write_all(sendco->chr,
751                                         (uint8_t *)&len,
752                                         sizeof(len));
753 
754             if (ret != sizeof(len)) {
755                 g_free(entry->buf);
756                 g_slice_free(SendEntry, entry);
757                 goto err;
758             }
759         }
760 
761         ret = qemu_chr_fe_write_all(sendco->chr,
762                                     (uint8_t *)entry->buf,
763                                     entry->size);
764 
765         if (ret != entry->size) {
766             g_free(entry->buf);
767             g_slice_free(SendEntry, entry);
768             goto err;
769         }
770 
771         g_free(entry->buf);
772         g_slice_free(SendEntry, entry);
773     }
774 
775     sendco->ret = 0;
776     goto out;
777 
778 err:
779     while (!g_queue_is_empty(&sendco->send_list)) {
780         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
781         g_free(entry->buf);
782         g_slice_free(SendEntry, entry);
783     }
784     sendco->ret = ret < 0 ? ret : -EIO;
785 out:
786     sendco->co = NULL;
787     sendco->done = true;
788     aio_wait_kick();
789 }
790 
791 static int compare_chr_send(CompareState *s,
792                             uint8_t *buf,
793                             uint32_t size,
794                             uint32_t vnet_hdr_len,
795                             bool notify_remote_frame,
796                             bool zero_copy)
797 {
798     SendCo *sendco;
799     SendEntry *entry;
800 
801     if (notify_remote_frame) {
802         sendco = &s->notify_sendco;
803     } else {
804         sendco = &s->out_sendco;
805     }
806 
807     if (!size) {
808         return -1;
809     }
810 
811     entry = g_slice_new(SendEntry);
812     entry->size = size;
813     entry->vnet_hdr_len = vnet_hdr_len;
814     if (zero_copy) {
815         entry->buf = buf;
816     } else {
817         entry->buf = g_malloc(size);
818         memcpy(entry->buf, buf, size);
819     }
820     g_queue_push_tail(&sendco->send_list, entry);
821 
822     if (sendco->done) {
823         sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
824         sendco->done = false;
825         qemu_coroutine_enter(sendco->co);
826         if (sendco->done) {
827             /* report early errors */
828             return sendco->ret;
829         }
830     }
831 
832     /* assume success */
833     return 0;
834 }
835 
836 static int compare_chr_can_read(void *opaque)
837 {
838     return COMPARE_READ_LEN_MAX;
839 }
840 
841 /*
842  * Called from the main thread on the primary for packets
843  * arriving over the socket from the primary.
844  */
845 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
846 {
847     CompareState *s = COLO_COMPARE(opaque);
848     int ret;
849 
850     ret = net_fill_rstate(&s->pri_rs, buf, size);
851     if (ret == -1) {
852         qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
853                                  NULL, NULL, true);
854         error_report("colo-compare primary_in error");
855     }
856 }
857 
858 /*
859  * Called from the main thread on the primary for packets
860  * arriving over the socket from the secondary.
861  */
862 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
863 {
864     CompareState *s = COLO_COMPARE(opaque);
865     int ret;
866 
867     ret = net_fill_rstate(&s->sec_rs, buf, size);
868     if (ret == -1) {
869         qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
870                                  NULL, NULL, true);
871         error_report("colo-compare secondary_in error");
872     }
873 }
874 
875 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
876 {
877     CompareState *s = COLO_COMPARE(opaque);
878     int ret;
879 
880     ret = net_fill_rstate(&s->notify_rs, buf, size);
881     if (ret == -1) {
882         qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
883                                  NULL, NULL, true);
884         error_report("colo-compare notify_dev error");
885     }
886 }
887 
888 /*
889  * Check old packet regularly so it can watch for any packets
890  * that the secondary hasn't produced equivalents of.
891  */
892 static void check_old_packet_regular(void *opaque)
893 {
894     CompareState *s = opaque;
895 
896     /* if have old packet we will notify checkpoint */
897     colo_old_packet_check(s);
898     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
899               s->expired_scan_cycle);
900 }
901 
902 /* Public API, Used for COLO frame to notify compare event */
903 void colo_notify_compares_event(void *opaque, int event, Error **errp)
904 {
905     CompareState *s;
906     qemu_mutex_lock(&colo_compare_mutex);
907 
908     if (!colo_compare_active) {
909         qemu_mutex_unlock(&colo_compare_mutex);
910         return;
911     }
912 
913     qemu_mutex_lock(&event_mtx);
914     QTAILQ_FOREACH(s, &net_compares, next) {
915         s->event = event;
916         qemu_bh_schedule(s->event_bh);
917         event_unhandled_count++;
918     }
919     /* Wait all compare threads to finish handling this event */
920     while (event_unhandled_count > 0) {
921         qemu_cond_wait(&event_complete_cond, &event_mtx);
922     }
923 
924     qemu_mutex_unlock(&event_mtx);
925     qemu_mutex_unlock(&colo_compare_mutex);
926 }
927 
928 static void colo_compare_timer_init(CompareState *s)
929 {
930     AioContext *ctx = iothread_get_aio_context(s->iothread);
931 
932     s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_HOST,
933                                 SCALE_MS, check_old_packet_regular,
934                                 s);
935     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
936               s->expired_scan_cycle);
937 }
938 
939 static void colo_compare_timer_del(CompareState *s)
940 {
941     if (s->packet_check_timer) {
942         timer_free(s->packet_check_timer);
943         s->packet_check_timer = NULL;
944     }
945  }
946 
947 static void colo_flush_packets(void *opaque, void *user_data);
948 
949 static void colo_compare_handle_event(void *opaque)
950 {
951     CompareState *s = opaque;
952 
953     switch (s->event) {
954     case COLO_EVENT_CHECKPOINT:
955         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
956         break;
957     case COLO_EVENT_FAILOVER:
958         break;
959     default:
960         break;
961     }
962 
963     qemu_mutex_lock(&event_mtx);
964     assert(event_unhandled_count > 0);
965     event_unhandled_count--;
966     qemu_cond_broadcast(&event_complete_cond);
967     qemu_mutex_unlock(&event_mtx);
968 }
969 
970 static void colo_compare_iothread(CompareState *s)
971 {
972     AioContext *ctx = iothread_get_aio_context(s->iothread);
973     object_ref(OBJECT(s->iothread));
974     s->worker_context = iothread_get_g_main_context(s->iothread);
975 
976     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
977                              compare_pri_chr_in, NULL, NULL,
978                              s, s->worker_context, true);
979     qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
980                              compare_sec_chr_in, NULL, NULL,
981                              s, s->worker_context, true);
982     if (s->notify_dev) {
983         qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
984                                  compare_notify_chr, NULL, NULL,
985                                  s, s->worker_context, true);
986     }
987 
988     colo_compare_timer_init(s);
989     s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
990 }
991 
992 static char *compare_get_pri_indev(Object *obj, Error **errp)
993 {
994     CompareState *s = COLO_COMPARE(obj);
995 
996     return g_strdup(s->pri_indev);
997 }
998 
999 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1000 {
1001     CompareState *s = COLO_COMPARE(obj);
1002 
1003     g_free(s->pri_indev);
1004     s->pri_indev = g_strdup(value);
1005 }
1006 
1007 static char *compare_get_sec_indev(Object *obj, Error **errp)
1008 {
1009     CompareState *s = COLO_COMPARE(obj);
1010 
1011     return g_strdup(s->sec_indev);
1012 }
1013 
1014 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1015 {
1016     CompareState *s = COLO_COMPARE(obj);
1017 
1018     g_free(s->sec_indev);
1019     s->sec_indev = g_strdup(value);
1020 }
1021 
1022 static char *compare_get_outdev(Object *obj, Error **errp)
1023 {
1024     CompareState *s = COLO_COMPARE(obj);
1025 
1026     return g_strdup(s->outdev);
1027 }
1028 
1029 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1030 {
1031     CompareState *s = COLO_COMPARE(obj);
1032 
1033     g_free(s->outdev);
1034     s->outdev = g_strdup(value);
1035 }
1036 
1037 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1038 {
1039     CompareState *s = COLO_COMPARE(obj);
1040 
1041     return s->vnet_hdr;
1042 }
1043 
1044 static void compare_set_vnet_hdr(Object *obj,
1045                                  bool value,
1046                                  Error **errp)
1047 {
1048     CompareState *s = COLO_COMPARE(obj);
1049 
1050     s->vnet_hdr = value;
1051 }
1052 
1053 static char *compare_get_notify_dev(Object *obj, Error **errp)
1054 {
1055     CompareState *s = COLO_COMPARE(obj);
1056 
1057     return g_strdup(s->notify_dev);
1058 }
1059 
1060 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1061 {
1062     CompareState *s = COLO_COMPARE(obj);
1063 
1064     g_free(s->notify_dev);
1065     s->notify_dev = g_strdup(value);
1066 }
1067 
1068 static void compare_get_timeout(Object *obj, Visitor *v,
1069                                 const char *name, void *opaque,
1070                                 Error **errp)
1071 {
1072     CompareState *s = COLO_COMPARE(obj);
1073     uint64_t value = s->compare_timeout;
1074 
1075     visit_type_uint64(v, name, &value, errp);
1076 }
1077 
1078 static void compare_set_timeout(Object *obj, Visitor *v,
1079                                 const char *name, void *opaque,
1080                                 Error **errp)
1081 {
1082     CompareState *s = COLO_COMPARE(obj);
1083     uint32_t value;
1084 
1085     if (!visit_type_uint32(v, name, &value, errp)) {
1086         return;
1087     }
1088     if (!value) {
1089         error_setg(errp, "Property '%s.%s' requires a positive value",
1090                    object_get_typename(obj), name);
1091         return;
1092     }
1093     s->compare_timeout = value;
1094 }
1095 
1096 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1097                                            const char *name, void *opaque,
1098                                            Error **errp)
1099 {
1100     CompareState *s = COLO_COMPARE(obj);
1101     uint32_t value = s->expired_scan_cycle;
1102 
1103     visit_type_uint32(v, name, &value, errp);
1104 }
1105 
1106 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1107                                            const char *name, void *opaque,
1108                                            Error **errp)
1109 {
1110     CompareState *s = COLO_COMPARE(obj);
1111     uint32_t value;
1112 
1113     if (!visit_type_uint32(v, name, &value, errp)) {
1114         return;
1115     }
1116     if (!value) {
1117         error_setg(errp, "Property '%s.%s' requires a positive value",
1118                    object_get_typename(obj), name);
1119         return;
1120     }
1121     s->expired_scan_cycle = value;
1122 }
1123 
1124 static void get_max_queue_size(Object *obj, Visitor *v,
1125                                const char *name, void *opaque,
1126                                Error **errp)
1127 {
1128     uint32_t value = max_queue_size;
1129 
1130     visit_type_uint32(v, name, &value, errp);
1131 }
1132 
1133 static void set_max_queue_size(Object *obj, Visitor *v,
1134                                const char *name, void *opaque,
1135                                Error **errp)
1136 {
1137     uint64_t value;
1138 
1139     if (!visit_type_uint64(v, name, &value, errp)) {
1140         return;
1141     }
1142     if (!value) {
1143         error_setg(errp, "Property '%s.%s' requires a positive value",
1144                    object_get_typename(obj), name);
1145         return;
1146     }
1147     max_queue_size = value;
1148 }
1149 
1150 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1151 {
1152     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1153     Connection *conn = NULL;
1154 
1155     if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1156         trace_colo_compare_main("primary: unsupported packet in");
1157         compare_chr_send(s,
1158                          pri_rs->buf,
1159                          pri_rs->packet_len,
1160                          pri_rs->vnet_hdr_len,
1161                          false,
1162                          false);
1163     } else {
1164         /* compare packet in the specified connection */
1165         colo_compare_connection(conn, s);
1166     }
1167 }
1168 
1169 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1170 {
1171     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1172     Connection *conn = NULL;
1173 
1174     if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1175         trace_colo_compare_main("secondary: unsupported packet in");
1176     } else {
1177         /* compare packet in the specified connection */
1178         colo_compare_connection(conn, s);
1179     }
1180 }
1181 
1182 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1183 {
1184     CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1185 
1186     const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1187     int ret;
1188 
1189     if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1190                            notify_rs->buf,
1191                            notify_rs->packet_len)) {
1192         ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1193         if (ret < 0) {
1194             error_report("Notify Xen COLO-frame INIT failed");
1195         }
1196     } else if (packet_matches_str("COLO_CHECKPOINT",
1197                                   notify_rs->buf,
1198                                   notify_rs->packet_len)) {
1199         /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1200         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1201     } else {
1202         error_report("COLO compare got unsupported instruction");
1203     }
1204 }
1205 
1206 /*
1207  * Return 0 is success.
1208  * Return 1 is failed.
1209  */
1210 static int find_and_check_chardev(Chardev **chr,
1211                                   char *chr_name,
1212                                   Error **errp)
1213 {
1214     *chr = qemu_chr_find(chr_name);
1215     if (*chr == NULL) {
1216         error_setg(errp, "Device '%s' not found",
1217                    chr_name);
1218         return 1;
1219     }
1220 
1221     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1222         error_setg(errp, "chardev \"%s\" is not reconnectable",
1223                    chr_name);
1224         return 1;
1225     }
1226 
1227     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1228         error_setg(errp, "chardev \"%s\" cannot switch context",
1229                    chr_name);
1230         return 1;
1231     }
1232 
1233     return 0;
1234 }
1235 
1236 /*
1237  * Called from the main thread on the primary
1238  * to setup colo-compare.
1239  */
1240 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1241 {
1242     CompareState *s = COLO_COMPARE(uc);
1243     Chardev *chr;
1244 
1245     if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1246         error_setg(errp, "colo compare needs 'primary_in' ,"
1247                    "'secondary_in','outdev','iothread' property set");
1248         return;
1249     } else if (!strcmp(s->pri_indev, s->outdev) ||
1250                !strcmp(s->sec_indev, s->outdev) ||
1251                !strcmp(s->pri_indev, s->sec_indev)) {
1252         error_setg(errp, "'indev' and 'outdev' could not be same "
1253                    "for compare module");
1254         return;
1255     }
1256 
1257     if (!s->compare_timeout) {
1258         /* Set default value to 3000 MS */
1259         s->compare_timeout = DEFAULT_TIME_OUT_MS;
1260     }
1261 
1262     if (!s->expired_scan_cycle) {
1263         /* Set default value to 1000 MS */
1264         s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1265     }
1266 
1267     if (!max_queue_size) {
1268         /* Set default queue size to 1024 */
1269         max_queue_size = MAX_QUEUE_SIZE;
1270     }
1271 
1272     if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1273         !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1274         return;
1275     }
1276 
1277     if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1278         !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1279         return;
1280     }
1281 
1282     if (find_and_check_chardev(&chr, s->outdev, errp) ||
1283         !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1284         return;
1285     }
1286 
1287     net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1288     net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1289 
1290     /* Try to enable remote notify chardev, currently just for Xen COLO */
1291     if (s->notify_dev) {
1292         if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1293             !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1294             return;
1295         }
1296 
1297         net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1298                            s->vnet_hdr);
1299     }
1300 
1301     s->out_sendco.s = s;
1302     s->out_sendco.chr = &s->chr_out;
1303     s->out_sendco.notify_remote_frame = false;
1304     s->out_sendco.done = true;
1305     g_queue_init(&s->out_sendco.send_list);
1306 
1307     if (s->notify_dev) {
1308         s->notify_sendco.s = s;
1309         s->notify_sendco.chr = &s->chr_notify_dev;
1310         s->notify_sendco.notify_remote_frame = true;
1311         s->notify_sendco.done = true;
1312         g_queue_init(&s->notify_sendco.send_list);
1313     }
1314 
1315     g_queue_init(&s->conn_list);
1316 
1317     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1318                                                       connection_key_equal,
1319                                                       g_free,
1320                                                       NULL);
1321 
1322     colo_compare_iothread(s);
1323 
1324     qemu_mutex_lock(&colo_compare_mutex);
1325     if (!colo_compare_active) {
1326         qemu_mutex_init(&event_mtx);
1327         qemu_cond_init(&event_complete_cond);
1328         colo_compare_active = true;
1329     }
1330     QTAILQ_INSERT_TAIL(&net_compares, s, next);
1331     qemu_mutex_unlock(&colo_compare_mutex);
1332 
1333     return;
1334 }
1335 
1336 static void colo_flush_packets(void *opaque, void *user_data)
1337 {
1338     CompareState *s = user_data;
1339     Connection *conn = opaque;
1340     Packet *pkt = NULL;
1341 
1342     while (!g_queue_is_empty(&conn->primary_list)) {
1343         pkt = g_queue_pop_tail(&conn->primary_list);
1344         compare_chr_send(s,
1345                          pkt->data,
1346                          pkt->size,
1347                          pkt->vnet_hdr_len,
1348                          false,
1349                          true);
1350         packet_destroy_partial(pkt, NULL);
1351     }
1352     while (!g_queue_is_empty(&conn->secondary_list)) {
1353         pkt = g_queue_pop_tail(&conn->secondary_list);
1354         packet_destroy(pkt, NULL);
1355     }
1356 }
1357 
1358 static void colo_compare_class_init(ObjectClass *oc, void *data)
1359 {
1360     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1361 
1362     ucc->complete = colo_compare_complete;
1363 }
1364 
1365 static void colo_compare_init(Object *obj)
1366 {
1367     CompareState *s = COLO_COMPARE(obj);
1368 
1369     object_property_add_str(obj, "primary_in",
1370                             compare_get_pri_indev, compare_set_pri_indev);
1371     object_property_add_str(obj, "secondary_in",
1372                             compare_get_sec_indev, compare_set_sec_indev);
1373     object_property_add_str(obj, "outdev",
1374                             compare_get_outdev, compare_set_outdev);
1375     object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1376                             (Object **)&s->iothread,
1377                             object_property_allow_set_link,
1378                             OBJ_PROP_LINK_STRONG);
1379     /* This parameter just for Xen COLO */
1380     object_property_add_str(obj, "notify_dev",
1381                             compare_get_notify_dev, compare_set_notify_dev);
1382 
1383     object_property_add(obj, "compare_timeout", "uint64",
1384                         compare_get_timeout,
1385                         compare_set_timeout, NULL, NULL);
1386 
1387     object_property_add(obj, "expired_scan_cycle", "uint32",
1388                         compare_get_expired_scan_cycle,
1389                         compare_set_expired_scan_cycle, NULL, NULL);
1390 
1391     object_property_add(obj, "max_queue_size", "uint32",
1392                         get_max_queue_size,
1393                         set_max_queue_size, NULL, NULL);
1394 
1395     s->vnet_hdr = false;
1396     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1397                              compare_set_vnet_hdr);
1398 }
1399 
1400 void colo_compare_cleanup(void)
1401 {
1402     CompareState *tmp = NULL;
1403     CompareState *n = NULL;
1404 
1405     QTAILQ_FOREACH_SAFE(tmp, &net_compares, next, n) {
1406         object_unparent(OBJECT(tmp));
1407     }
1408 }
1409 
1410 static void colo_compare_finalize(Object *obj)
1411 {
1412     CompareState *s = COLO_COMPARE(obj);
1413     CompareState *tmp = NULL;
1414 
1415     qemu_mutex_lock(&colo_compare_mutex);
1416     QTAILQ_FOREACH(tmp, &net_compares, next) {
1417         if (tmp == s) {
1418             QTAILQ_REMOVE(&net_compares, s, next);
1419             break;
1420         }
1421     }
1422     if (QTAILQ_EMPTY(&net_compares)) {
1423         colo_compare_active = false;
1424         qemu_mutex_destroy(&event_mtx);
1425         qemu_cond_destroy(&event_complete_cond);
1426     }
1427     qemu_mutex_unlock(&colo_compare_mutex);
1428 
1429     qemu_chr_fe_deinit(&s->chr_pri_in, false);
1430     qemu_chr_fe_deinit(&s->chr_sec_in, false);
1431     qemu_chr_fe_deinit(&s->chr_out, false);
1432     if (s->notify_dev) {
1433         qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1434     }
1435 
1436     colo_compare_timer_del(s);
1437 
1438     qemu_bh_delete(s->event_bh);
1439 
1440     AioContext *ctx = iothread_get_aio_context(s->iothread);
1441     AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1442     if (s->notify_dev) {
1443         AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1444     }
1445 
1446     /* Release all unhandled packets after compare thead exited */
1447     g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1448     AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1449 
1450     g_queue_clear(&s->conn_list);
1451     g_queue_clear(&s->out_sendco.send_list);
1452     if (s->notify_dev) {
1453         g_queue_clear(&s->notify_sendco.send_list);
1454     }
1455 
1456     if (s->connection_track_table) {
1457         g_hash_table_destroy(s->connection_track_table);
1458     }
1459 
1460     object_unref(OBJECT(s->iothread));
1461 
1462     g_free(s->pri_indev);
1463     g_free(s->sec_indev);
1464     g_free(s->outdev);
1465     g_free(s->notify_dev);
1466 }
1467 
1468 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1469 {
1470     colo_compare_active = false;
1471     qemu_mutex_init(&colo_compare_mutex);
1472 }
1473 
1474 static const TypeInfo colo_compare_info = {
1475     .name = TYPE_COLO_COMPARE,
1476     .parent = TYPE_OBJECT,
1477     .instance_size = sizeof(CompareState),
1478     .instance_init = colo_compare_init,
1479     .instance_finalize = colo_compare_finalize,
1480     .class_size = sizeof(CompareClass),
1481     .class_init = colo_compare_class_init,
1482     .interfaces = (InterfaceInfo[]) {
1483         { TYPE_USER_CREATABLE },
1484         { }
1485     }
1486 };
1487 
1488 static void register_types(void)
1489 {
1490     type_register_static(&colo_compare_info);
1491 }
1492 
1493 type_init(register_types);
1494