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