?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
s390x/tcg: Fix ALGSI
[qemu/kevin.git] / net / colo-compare.c
blob84db4978ac334b6371c987956ed129e89fe70357
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 typedef struct CompareState CompareState;
40 DECLARE_INSTANCE_CHECKER(CompareState, COLO_COMPARE,
41 TYPE_COLO_COMPARE)
42
43 static QTAILQ_HEAD(, CompareState) net_compares =
44 QTAILQ_HEAD_INITIALIZER(net_compares);
45
46 static NotifierList colo_compare_notifiers =
47 NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
48
49 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
50 #define MAX_QUEUE_SIZE 1024
51
52 #define COLO_COMPARE_FREE_PRIMARY 0x01
53 #define COLO_COMPARE_FREE_SECONDARY 0x02
54
55 #define REGULAR_PACKET_CHECK_MS 1000
56 #define DEFAULT_TIME_OUT_MS 3000
57
58 /* #define DEBUG_COLO_PACKETS */
59
60 static QemuMutex colo_compare_mutex;
61 static bool colo_compare_active;
62 static QemuMutex event_mtx;
63 static QemuCond event_complete_cond;
64 static int event_unhandled_count;
65 static uint32_t max_queue_size;
66
67 /*
68 * + CompareState ++
69 * | |
70 * +---------------+ +---------------+ +---------------+
71 * | conn list + - > conn + ------- > conn + -- > ......
72 * +---------------+ +---------------+ +---------------+
73 * | | | | | |
74 * +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
75 * |primary | |secondary |primary | |secondary
76 * |packet | |packet + |packet | |packet +
77 * +--------+ +--------+ +--------+ +--------+
78 * | | | |
79 * +---v----+ +---v----+ +---v----+ +---v----+
80 * |primary | |secondary |primary | |secondary
81 * |packet | |packet + |packet | |packet +
82 * +--------+ +--------+ +--------+ +--------+
83 * | | | |
84 * +---v----+ +---v----+ +---v----+ +---v----+
85 * |primary | |secondary |primary | |secondary
86 * |packet | |packet + |packet | |packet +
87 * +--------+ +--------+ +--------+ +--------+
88 */
89
90 typedef struct SendCo {
91 Coroutine *co;
92 struct CompareState *s;
93 CharBackend *chr;
94 GQueue send_list;
95 bool notify_remote_frame;
96 bool done;
97 int ret;
98 } SendCo;
99
100 typedef struct SendEntry {
101 uint32_t size;
102 uint32_t vnet_hdr_len;
103 uint8_t *buf;
104 } SendEntry;
105
106 struct CompareState {
107 Object parent;
108
109 char *pri_indev;
110 char *sec_indev;
111 char *outdev;
112 char *notify_dev;
113 CharBackend chr_pri_in;
114 CharBackend chr_sec_in;
115 CharBackend chr_out;
116 CharBackend chr_notify_dev;
117 SocketReadState pri_rs;
118 SocketReadState sec_rs;
119 SocketReadState notify_rs;
120 SendCo out_sendco;
121 SendCo notify_sendco;
122 bool vnet_hdr;
123 uint64_t compare_timeout;
124 uint32_t expired_scan_cycle;
125
126 /*
127 * Record the connection that through the NIC
128 * Element type: Connection
129 */
130 GQueue conn_list;
131 /* Record the connection without repetition */
132 GHashTable *connection_track_table;
133
134 IOThread *iothread;
135 GMainContext *worker_context;
136 QEMUTimer *packet_check_timer;
137
138 QEMUBH *event_bh;
139 enum colo_event event;
140
141 QTAILQ_ENTRY(CompareState) next;
142 };
143
144 typedef struct CompareClass {
145 ObjectClass parent_class;
146 } CompareClass;
147
148 enum {
149 PRIMARY_IN = 0,
150 SECONDARY_IN,
151 };
152
153 static const char *colo_mode[] = {
154 [PRIMARY_IN] = "primary",
155 [SECONDARY_IN] = "secondary",
156 };
157
158 static int compare_chr_send(CompareState *s,
159 uint8_t *buf,
160 uint32_t size,
161 uint32_t vnet_hdr_len,
162 bool notify_remote_frame,
163 bool zero_copy);
164
165 static bool packet_matches_str(const char *str,
166 const uint8_t *buf,
167 uint32_t packet_len)
168 {
169 if (packet_len != strlen(str)) {
170 return false;
171 }
172
173 return !memcmp(str, buf, strlen(str));
174 }
175
176 static void notify_remote_frame(CompareState *s)
177 {
178 char msg[] = "DO_CHECKPOINT";
179 int ret = 0;
180
181 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
182 if (ret < 0) {
183 error_report("Notify Xen COLO-frame failed");
184 }
185 }
186
187 static void colo_compare_inconsistency_notify(CompareState *s)
188 {
189 if (s->notify_dev) {
190 notify_remote_frame(s);
191 } else {
192 notifier_list_notify(&colo_compare_notifiers,
193 migrate_get_current());
194 }
195 }
196
197 /* Use restricted to colo_insert_packet() */
198 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
199 {
200 return a->tcp_seq - b->tcp_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_IP_INFO)) {
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 } 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_head(&conn->primary_list, ppkt);
492 g_queue_push_head(&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 (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
594 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
595
596 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
597 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
598 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
599 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
600
601 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
602 pri_ip_dst, spkt->size,
603 sec_ip_src, sec_ip_dst);
604 }
605
606 if (ppkt->size != spkt->size) {
607 trace_colo_compare_main("Other: payload size of packets are different");
608 return -1;
609 }
610 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
611 ppkt->size - offset);
612 }
613
614 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
615 {
616 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
617
618 if ((now - pkt->creation_ms) > (*check_time)) {
619 trace_colo_old_packet_check_found(pkt->creation_ms);
620 return 0;
621 } else {
622 return 1;
623 }
624 }
625
626 void colo_compare_register_notifier(Notifier *notify)
627 {
628 notifier_list_add(&colo_compare_notifiers, notify);
629 }
630
631 void colo_compare_unregister_notifier(Notifier *notify)
632 {
633 notifier_remove(notify);
634 }
635
636 static int colo_old_packet_check_one_conn(Connection *conn,
637 CompareState *s)
638 {
639 if (!g_queue_is_empty(&conn->primary_list)) {
640 if (g_queue_find_custom(&conn->primary_list,
641 &s->compare_timeout,
642 (GCompareFunc)colo_old_packet_check_one))
643 goto out;
644 }
645
646 if (!g_queue_is_empty(&conn->secondary_list)) {
647 if (g_queue_find_custom(&conn->secondary_list,
648 &s->compare_timeout,
649 (GCompareFunc)colo_old_packet_check_one))
650 goto out;
651 }
652
653 return 1;
654
655 out:
656 /* Do checkpoint will flush old packet */
657 colo_compare_inconsistency_notify(s);
658 return 0;
659 }
660
661 /*
662 * Look for old packets that the secondary hasn't matched,
663 * if we have some then we have to checkpoint to wake
664 * the secondary up.
665 */
666 static void colo_old_packet_check(void *opaque)
667 {
668 CompareState *s = opaque;
669
670 /*
671 * If we find one old packet, stop finding job and notify
672 * COLO frame do checkpoint.
673 */
674 g_queue_find_custom(&s->conn_list, s,
675 (GCompareFunc)colo_old_packet_check_one_conn);
676 }
677
678 static void colo_compare_packet(CompareState *s, Connection *conn,
679 int (*HandlePacket)(Packet *spkt,
680 Packet *ppkt))
681 {
682 Packet *pkt = NULL;
683 GList *result = NULL;
684
685 while (!g_queue_is_empty(&conn->primary_list) &&
686 !g_queue_is_empty(&conn->secondary_list)) {
687 pkt = g_queue_pop_head(&conn->primary_list);
688 result = g_queue_find_custom(&conn->secondary_list,
689 pkt, (GCompareFunc)HandlePacket);
690
691 if (result) {
692 colo_release_primary_pkt(s, pkt);
693 g_queue_remove(&conn->secondary_list, result->data);
694 } else {
695 /*
696 * If one packet arrive late, the secondary_list or
697 * primary_list will be empty, so we can't compare it
698 * until next comparison. If the packets in the list are
699 * timeout, it will trigger a checkpoint request.
700 */
701 trace_colo_compare_main("packet different");
702 g_queue_push_head(&conn->primary_list, pkt);
703
704 colo_compare_inconsistency_notify(s);
705 break;
706 }
707 }
708 }
709
710 /*
711 * Called from the compare thread on the primary
712 * for compare packet with secondary list of the
713 * specified connection when a new packet was
714 * queued to it.
715 */
716 static void colo_compare_connection(void *opaque, void *user_data)
717 {
718 CompareState *s = user_data;
719 Connection *conn = opaque;
720
721 switch (conn->ip_proto) {
722 case IPPROTO_TCP:
723 colo_compare_tcp(s, conn);
724 break;
725 case IPPROTO_UDP:
726 colo_compare_packet(s, conn, colo_packet_compare_udp);
727 break;
728 case IPPROTO_ICMP:
729 colo_compare_packet(s, conn, colo_packet_compare_icmp);
730 break;
731 default:
732 colo_compare_packet(s, conn, colo_packet_compare_other);
733 break;
734 }
735 }
736
737 static void coroutine_fn _compare_chr_send(void *opaque)
738 {
739 SendCo *sendco = opaque;
740 CompareState *s = sendco->s;
741 int ret = 0;
742
743 while (!g_queue_is_empty(&sendco->send_list)) {
744 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
745 uint32_t len = htonl(entry->size);
746
747 ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
748
749 if (ret != sizeof(len)) {
750 g_free(entry->buf);
751 g_slice_free(SendEntry, entry);
752 goto err;
753 }
754
755 if (!sendco->notify_remote_frame && s->vnet_hdr) {
756 /*
757 * We send vnet header len make other module(like filter-redirector)
758 * know how to parse net packet correctly.
759 */
760 len = htonl(entry->vnet_hdr_len);
761
762 ret = qemu_chr_fe_write_all(sendco->chr,
763 (uint8_t *)&len,
764 sizeof(len));
765
766 if (ret != sizeof(len)) {
767 g_free(entry->buf);
768 g_slice_free(SendEntry, entry);
769 goto err;
770 }
771 }
772
773 ret = qemu_chr_fe_write_all(sendco->chr,
774 (uint8_t *)entry->buf,
775 entry->size);
776
777 if (ret != entry->size) {
778 g_free(entry->buf);
779 g_slice_free(SendEntry, entry);
780 goto err;
781 }
782
783 g_free(entry->buf);
784 g_slice_free(SendEntry, entry);
785 }
786
787 sendco->ret = 0;
788 goto out;
789
790 err:
791 while (!g_queue_is_empty(&sendco->send_list)) {
792 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
793 g_free(entry->buf);
794 g_slice_free(SendEntry, entry);
795 }
796 sendco->ret = ret < 0 ? ret : -EIO;
797 out:
798 sendco->co = NULL;
799 sendco->done = true;
800 aio_wait_kick();
801 }
802
803 static int compare_chr_send(CompareState *s,
804 uint8_t *buf,
805 uint32_t size,
806 uint32_t vnet_hdr_len,
807 bool notify_remote_frame,
808 bool zero_copy)
809 {
810 SendCo *sendco;
811 SendEntry *entry;
812
813 if (notify_remote_frame) {
814 sendco = &s->notify_sendco;
815 } else {
816 sendco = &s->out_sendco;
817 }
818
819 if (!size) {
820 return 0;
821 }
822
823 entry = g_slice_new(SendEntry);
824 entry->size = size;
825 entry->vnet_hdr_len = vnet_hdr_len;
826 if (zero_copy) {
827 entry->buf = buf;
828 } else {
829 entry->buf = g_malloc(size);
830 memcpy(entry->buf, buf, size);
831 }
832 g_queue_push_head(&sendco->send_list, entry);
833
834 if (sendco->done) {
835 sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
836 sendco->done = false;
837 qemu_coroutine_enter(sendco->co);
838 if (sendco->done) {
839 /* report early errors */
840 return sendco->ret;
841 }
842 }
843
844 /* assume success */
845 return 0;
846 }
847
848 static int compare_chr_can_read(void *opaque)
849 {
850 return COMPARE_READ_LEN_MAX;
851 }
852
853 /*
854 * Called from the main thread on the primary for packets
855 * arriving over the socket from the primary.
856 */
857 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
858 {
859 CompareState *s = COLO_COMPARE(opaque);
860 int ret;
861
862 ret = net_fill_rstate(&s->pri_rs, buf, size);
863 if (ret == -1) {
864 qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
865 NULL, NULL, true);
866 error_report("colo-compare primary_in error");
867 }
868 }
869
870 /*
871 * Called from the main thread on the primary for packets
872 * arriving over the socket from the secondary.
873 */
874 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
875 {
876 CompareState *s = COLO_COMPARE(opaque);
877 int ret;
878
879 ret = net_fill_rstate(&s->sec_rs, buf, size);
880 if (ret == -1) {
881 qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
882 NULL, NULL, true);
883 error_report("colo-compare secondary_in error");
884 }
885 }
886
887 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
888 {
889 CompareState *s = COLO_COMPARE(opaque);
890 int ret;
891
892 ret = net_fill_rstate(&s->notify_rs, buf, size);
893 if (ret == -1) {
894 qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
895 NULL, NULL, true);
896 error_report("colo-compare notify_dev error");
897 }
898 }
899
900 /*
901 * Check old packet regularly so it can watch for any packets
902 * that the secondary hasn't produced equivalents of.
903 */
904 static void check_old_packet_regular(void *opaque)
905 {
906 CompareState *s = opaque;
907
908 /* if have old packet we will notify checkpoint */
909 colo_old_packet_check(s);
910 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
911 s->expired_scan_cycle);
912 }
913
914 /* Public API, Used for COLO frame to notify compare event */
915 void colo_notify_compares_event(void *opaque, int event, Error **errp)
916 {
917 CompareState *s;
918 qemu_mutex_lock(&colo_compare_mutex);
919
920 if (!colo_compare_active) {
921 qemu_mutex_unlock(&colo_compare_mutex);
922 return;
923 }
924
925 qemu_mutex_lock(&event_mtx);
926 QTAILQ_FOREACH(s, &net_compares, next) {
927 s->event = event;
928 qemu_bh_schedule(s->event_bh);
929 event_unhandled_count++;
930 }
931 /* Wait all compare threads to finish handling this event */
932 while (event_unhandled_count > 0) {
933 qemu_cond_wait(&event_complete_cond, &event_mtx);
934 }
935
936 qemu_mutex_unlock(&event_mtx);
937 qemu_mutex_unlock(&colo_compare_mutex);
938 }
939
940 static void colo_compare_timer_init(CompareState *s)
941 {
942 AioContext *ctx = iothread_get_aio_context(s->iothread);
943
944 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_HOST,
945 SCALE_MS, check_old_packet_regular,
946 s);
947 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
948 s->expired_scan_cycle);
949 }
950
951 static void colo_compare_timer_del(CompareState *s)
952 {
953 if (s->packet_check_timer) {
954 timer_free(s->packet_check_timer);
955 s->packet_check_timer = NULL;
956 }
957 }
958
959 static void colo_flush_packets(void *opaque, void *user_data);
960
961 static void colo_compare_handle_event(void *opaque)
962 {
963 CompareState *s = opaque;
964
965 switch (s->event) {
966 case COLO_EVENT_CHECKPOINT:
967 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
968 break;
969 case COLO_EVENT_FAILOVER:
970 break;
971 default:
972 break;
973 }
974
975 qemu_mutex_lock(&event_mtx);
976 assert(event_unhandled_count > 0);
977 event_unhandled_count--;
978 qemu_cond_broadcast(&event_complete_cond);
979 qemu_mutex_unlock(&event_mtx);
980 }
981
982 static void colo_compare_iothread(CompareState *s)
983 {
984 AioContext *ctx = iothread_get_aio_context(s->iothread);
985 object_ref(OBJECT(s->iothread));
986 s->worker_context = iothread_get_g_main_context(s->iothread);
987
988 qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
989 compare_pri_chr_in, NULL, NULL,
990 s, s->worker_context, true);
991 qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
992 compare_sec_chr_in, NULL, NULL,
993 s, s->worker_context, true);
994 if (s->notify_dev) {
995 qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
996 compare_notify_chr, NULL, NULL,
997 s, s->worker_context, true);
998 }
999
1000 colo_compare_timer_init(s);
1001 s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
1002 }
1003
1004 static char *compare_get_pri_indev(Object *obj, Error **errp)
1005 {
1006 CompareState *s = COLO_COMPARE(obj);
1007
1008 return g_strdup(s->pri_indev);
1009 }
1010
1011 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1012 {
1013 CompareState *s = COLO_COMPARE(obj);
1014
1015 g_free(s->pri_indev);
1016 s->pri_indev = g_strdup(value);
1017 }
1018
1019 static char *compare_get_sec_indev(Object *obj, Error **errp)
1020 {
1021 CompareState *s = COLO_COMPARE(obj);
1022
1023 return g_strdup(s->sec_indev);
1024 }
1025
1026 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1027 {
1028 CompareState *s = COLO_COMPARE(obj);
1029
1030 g_free(s->sec_indev);
1031 s->sec_indev = g_strdup(value);
1032 }
1033
1034 static char *compare_get_outdev(Object *obj, Error **errp)
1035 {
1036 CompareState *s = COLO_COMPARE(obj);
1037
1038 return g_strdup(s->outdev);
1039 }
1040
1041 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1042 {
1043 CompareState *s = COLO_COMPARE(obj);
1044
1045 g_free(s->outdev);
1046 s->outdev = g_strdup(value);
1047 }
1048
1049 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1050 {
1051 CompareState *s = COLO_COMPARE(obj);
1052
1053 return s->vnet_hdr;
1054 }
1055
1056 static void compare_set_vnet_hdr(Object *obj,
1057 bool value,
1058 Error **errp)
1059 {
1060 CompareState *s = COLO_COMPARE(obj);
1061
1062 s->vnet_hdr = value;
1063 }
1064
1065 static char *compare_get_notify_dev(Object *obj, Error **errp)
1066 {
1067 CompareState *s = COLO_COMPARE(obj);
1068
1069 return g_strdup(s->notify_dev);
1070 }
1071
1072 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1073 {
1074 CompareState *s = COLO_COMPARE(obj);
1075
1076 g_free(s->notify_dev);
1077 s->notify_dev = g_strdup(value);
1078 }
1079
1080 static void compare_get_timeout(Object *obj, Visitor *v,
1081 const char *name, void *opaque,
1082 Error **errp)
1083 {
1084 CompareState *s = COLO_COMPARE(obj);
1085 uint64_t value = s->compare_timeout;
1086
1087 visit_type_uint64(v, name, &value, errp);
1088 }
1089
1090 static void compare_set_timeout(Object *obj, Visitor *v,
1091 const char *name, void *opaque,
1092 Error **errp)
1093 {
1094 CompareState *s = COLO_COMPARE(obj);
1095 uint32_t value;
1096
1097 if (!visit_type_uint32(v, name, &value, errp)) {
1098 return;
1099 }
1100 if (!value) {
1101 error_setg(errp, "Property '%s.%s' requires a positive value",
1102 object_get_typename(obj), name);
1103 return;
1104 }
1105 s->compare_timeout = value;
1106 }
1107
1108 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1109 const char *name, void *opaque,
1110 Error **errp)
1111 {
1112 CompareState *s = COLO_COMPARE(obj);
1113 uint32_t value = s->expired_scan_cycle;
1114
1115 visit_type_uint32(v, name, &value, errp);
1116 }
1117
1118 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1119 const char *name, void *opaque,
1120 Error **errp)
1121 {
1122 CompareState *s = COLO_COMPARE(obj);
1123 uint32_t value;
1124
1125 if (!visit_type_uint32(v, name, &value, errp)) {
1126 return;
1127 }
1128 if (!value) {
1129 error_setg(errp, "Property '%s.%s' requires a positive value",
1130 object_get_typename(obj), name);
1131 return;
1132 }
1133 s->expired_scan_cycle = value;
1134 }
1135
1136 static void get_max_queue_size(Object *obj, Visitor *v,
1137 const char *name, void *opaque,
1138 Error **errp)
1139 {
1140 uint32_t value = max_queue_size;
1141
1142 visit_type_uint32(v, name, &value, errp);
1143 }
1144
1145 static void set_max_queue_size(Object *obj, Visitor *v,
1146 const char *name, void *opaque,
1147 Error **errp)
1148 {
1149 Error *local_err = NULL;
1150 uint64_t value;
1151
1152 visit_type_uint64(v, name, &value, &local_err);
1153 if (local_err) {
1154 goto out;
1155 }
1156 if (!value) {
1157 error_setg(&local_err, "Property '%s.%s' requires a positive value",
1158 object_get_typename(obj), name);
1159 goto out;
1160 }
1161 max_queue_size = value;
1162
1163 out:
1164 error_propagate(errp, local_err);
1165 }
1166
1167 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1168 {
1169 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1170 Connection *conn = NULL;
1171
1172 if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1173 trace_colo_compare_main("primary: unsupported packet in");
1174 compare_chr_send(s,
1175 pri_rs->buf,
1176 pri_rs->packet_len,
1177 pri_rs->vnet_hdr_len,
1178 false,
1179 false);
1180 } else {
1181 /* compare packet in the specified connection */
1182 colo_compare_connection(conn, s);
1183 }
1184 }
1185
1186 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1187 {
1188 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1189 Connection *conn = NULL;
1190
1191 if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1192 trace_colo_compare_main("secondary: unsupported packet in");
1193 } else {
1194 /* compare packet in the specified connection */
1195 colo_compare_connection(conn, s);
1196 }
1197 }
1198
1199 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1200 {
1201 CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1202
1203 const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1204 int ret;
1205
1206 if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1207 notify_rs->buf,
1208 notify_rs->packet_len)) {
1209 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1210 if (ret < 0) {
1211 error_report("Notify Xen COLO-frame INIT failed");
1212 }
1213 } else if (packet_matches_str("COLO_CHECKPOINT",
1214 notify_rs->buf,
1215 notify_rs->packet_len)) {
1216 /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1217 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1218 } else {
1219 error_report("COLO compare got unsupported instruction");
1220 }
1221 }
1222
1223 /*
1224 * Return 0 is success.
1225 * Return 1 is failed.
1226 */
1227 static int find_and_check_chardev(Chardev **chr,
1228 char *chr_name,
1229 Error **errp)
1230 {
1231 *chr = qemu_chr_find(chr_name);
1232 if (*chr == NULL) {
1233 error_setg(errp, "Device '%s' not found",
1234 chr_name);
1235 return 1;
1236 }
1237
1238 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1239 error_setg(errp, "chardev \"%s\" is not reconnectable",
1240 chr_name);
1241 return 1;
1242 }
1243
1244 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1245 error_setg(errp, "chardev \"%s\" cannot switch context",
1246 chr_name);
1247 return 1;
1248 }
1249
1250 return 0;
1251 }
1252
1253 /*
1254 * Called from the main thread on the primary
1255 * to setup colo-compare.
1256 */
1257 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1258 {
1259 CompareState *s = COLO_COMPARE(uc);
1260 Chardev *chr;
1261
1262 if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1263 error_setg(errp, "colo compare needs 'primary_in' ,"
1264 "'secondary_in','outdev','iothread' property set");
1265 return;
1266 } else if (!strcmp(s->pri_indev, s->outdev) ||
1267 !strcmp(s->sec_indev, s->outdev) ||
1268 !strcmp(s->pri_indev, s->sec_indev)) {
1269 error_setg(errp, "'indev' and 'outdev' could not be same "
1270 "for compare module");
1271 return;
1272 }
1273
1274 if (!s->compare_timeout) {
1275 /* Set default value to 3000 MS */
1276 s->compare_timeout = DEFAULT_TIME_OUT_MS;
1277 }
1278
1279 if (!s->expired_scan_cycle) {
1280 /* Set default value to 3000 MS */
1281 s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1282 }
1283
1284 if (!max_queue_size) {
1285 /* Set default queue size to 1024 */
1286 max_queue_size = MAX_QUEUE_SIZE;
1287 }
1288
1289 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1290 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1291 return;
1292 }
1293
1294 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1295 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1296 return;
1297 }
1298
1299 if (find_and_check_chardev(&chr, s->outdev, errp) ||
1300 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1301 return;
1302 }
1303
1304 net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1305 net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1306
1307 /* Try to enable remote notify chardev, currently just for Xen COLO */
1308 if (s->notify_dev) {
1309 if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1310 !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1311 return;
1312 }
1313
1314 net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1315 s->vnet_hdr);
1316 }
1317
1318 s->out_sendco.s = s;
1319 s->out_sendco.chr = &s->chr_out;
1320 s->out_sendco.notify_remote_frame = false;
1321 s->out_sendco.done = true;
1322 g_queue_init(&s->out_sendco.send_list);
1323
1324 if (s->notify_dev) {
1325 s->notify_sendco.s = s;
1326 s->notify_sendco.chr = &s->chr_notify_dev;
1327 s->notify_sendco.notify_remote_frame = true;
1328 s->notify_sendco.done = true;
1329 g_queue_init(&s->notify_sendco.send_list);
1330 }
1331
1332 g_queue_init(&s->conn_list);
1333
1334 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1335 connection_key_equal,
1336 g_free,
1337 connection_destroy);
1338
1339 colo_compare_iothread(s);
1340
1341 qemu_mutex_lock(&colo_compare_mutex);
1342 if (!colo_compare_active) {
1343 qemu_mutex_init(&event_mtx);
1344 qemu_cond_init(&event_complete_cond);
1345 colo_compare_active = true;
1346 }
1347 QTAILQ_INSERT_TAIL(&net_compares, s, next);
1348 qemu_mutex_unlock(&colo_compare_mutex);
1349
1350 return;
1351 }
1352
1353 static void colo_flush_packets(void *opaque, void *user_data)
1354 {
1355 CompareState *s = user_data;
1356 Connection *conn = opaque;
1357 Packet *pkt = NULL;
1358
1359 while (!g_queue_is_empty(&conn->primary_list)) {
1360 pkt = g_queue_pop_head(&conn->primary_list);
1361 compare_chr_send(s,
1362 pkt->data,
1363 pkt->size,
1364 pkt->vnet_hdr_len,
1365 false,
1366 true);
1367 packet_destroy_partial(pkt, NULL);
1368 }
1369 while (!g_queue_is_empty(&conn->secondary_list)) {
1370 pkt = g_queue_pop_head(&conn->secondary_list);
1371 packet_destroy(pkt, NULL);
1372 }
1373 }
1374
1375 static void colo_compare_class_init(ObjectClass *oc, void *data)
1376 {
1377 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1378
1379 ucc->complete = colo_compare_complete;
1380 }
1381
1382 static void colo_compare_init(Object *obj)
1383 {
1384 CompareState *s = COLO_COMPARE(obj);
1385
1386 object_property_add_str(obj, "primary_in",
1387 compare_get_pri_indev, compare_set_pri_indev);
1388 object_property_add_str(obj, "secondary_in",
1389 compare_get_sec_indev, compare_set_sec_indev);
1390 object_property_add_str(obj, "outdev",
1391 compare_get_outdev, compare_set_outdev);
1392 object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1393 (Object **)&s->iothread,
1394 object_property_allow_set_link,
1395 OBJ_PROP_LINK_STRONG);
1396 /* This parameter just for Xen COLO */
1397 object_property_add_str(obj, "notify_dev",
1398 compare_get_notify_dev, compare_set_notify_dev);
1399
1400 object_property_add(obj, "compare_timeout", "uint64",
1401 compare_get_timeout,
1402 compare_set_timeout, NULL, NULL);
1403
1404 object_property_add(obj, "expired_scan_cycle", "uint32",
1405 compare_get_expired_scan_cycle,
1406 compare_set_expired_scan_cycle, NULL, NULL);
1407
1408 object_property_add(obj, "max_queue_size", "uint32",
1409 get_max_queue_size,
1410 set_max_queue_size, NULL, NULL);
1411
1412 s->vnet_hdr = false;
1413 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1414 compare_set_vnet_hdr);
1415 }
1416
1417 static void colo_compare_finalize(Object *obj)
1418 {
1419 CompareState *s = COLO_COMPARE(obj);
1420 CompareState *tmp = NULL;
1421
1422 qemu_mutex_lock(&colo_compare_mutex);
1423 QTAILQ_FOREACH(tmp, &net_compares, next) {
1424 if (tmp == s) {
1425 QTAILQ_REMOVE(&net_compares, s, next);
1426 break;
1427 }
1428 }
1429 if (QTAILQ_EMPTY(&net_compares)) {
1430 colo_compare_active = false;
1431 qemu_mutex_destroy(&event_mtx);
1432 qemu_cond_destroy(&event_complete_cond);
1433 }
1434 qemu_mutex_unlock(&colo_compare_mutex);
1435
1436 qemu_chr_fe_deinit(&s->chr_pri_in, false);
1437 qemu_chr_fe_deinit(&s->chr_sec_in, false);
1438 qemu_chr_fe_deinit(&s->chr_out, false);
1439 if (s->notify_dev) {
1440 qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1441 }
1442
1443 colo_compare_timer_del(s);
1444
1445 qemu_bh_delete(s->event_bh);
1446
1447 AioContext *ctx = iothread_get_aio_context(s->iothread);
1448 aio_context_acquire(ctx);
1449 AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1450 if (s->notify_dev) {
1451 AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1452 }
1453 aio_context_release(ctx);
1454
1455 /* Release all unhandled packets after compare thead exited */
1456 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1457 AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1458
1459 g_queue_clear(&s->conn_list);
1460 g_queue_clear(&s->out_sendco.send_list);
1461 if (s->notify_dev) {
1462 g_queue_clear(&s->notify_sendco.send_list);
1463 }
1464
1465 if (s->connection_track_table) {
1466 g_hash_table_destroy(s->connection_track_table);
1467 }
1468
1469 object_unref(OBJECT(s->iothread));
1470
1471 g_free(s->pri_indev);
1472 g_free(s->sec_indev);
1473 g_free(s->outdev);
1474 g_free(s->notify_dev);
1475 }
1476
1477 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1478 {
1479 colo_compare_active = false;
1480 qemu_mutex_init(&colo_compare_mutex);
1481 }
1482
1483 static const TypeInfo colo_compare_info = {
1484 .name = TYPE_COLO_COMPARE,
1485 .parent = TYPE_OBJECT,
1486 .instance_size = sizeof(CompareState),
1487 .instance_init = colo_compare_init,
1488 .instance_finalize = colo_compare_finalize,
1489 .class_size = sizeof(CompareClass),
1490 .class_init = colo_compare_class_init,
1491 .interfaces = (InterfaceInfo[]) {
1492 { TYPE_USER_CREATABLE },
1493 { }
1494 }
1495 };
1496
1497 static void register_types(void)
1498 {
1499 type_register_static(&colo_compare_info);
1500 }
1501
1502 type_init(register_types);
Random patches