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
packet_matches_str(const char * str,const uint8_t * buf,uint32_t packet_len)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
notify_remote_frame(CompareState * s)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
colo_compare_inconsistency_notify(CompareState * s)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() */
seq_sorter(Packet * a,Packet * b,gpointer data)196 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
197 {
198 return b->tcp_seq - a->tcp_seq;
199 }
200
fill_pkt_tcp_info(void * data,uint32_t * max_ack)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 */
colo_insert_packet(GQueue * queue,Packet * pkt,uint32_t * max_ack)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 */
packet_enqueue(CompareState * s,int mode,Connection ** con)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
after(uint32_t seq1,uint32_t seq2)295 static inline bool after(uint32_t seq1, uint32_t seq2)
296 {
297 return (int32_t)(seq1 - seq2) > 0;
298 }
299
colo_release_primary_pkt(CompareState * s,Packet * pkt)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 */
colo_compare_packet_payload(Packet * ppkt,Packet * spkt,uint16_t poffset,uint16_t soffset,uint16_t len)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 */
colo_mark_tcp_pkt(Packet * ppkt,Packet * spkt,int8_t * mark,uint32_t max_ack)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
colo_compare_tcp(CompareState * s,Connection * conn)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 = MIN(conn->pack, conn->sack);
416
417 pri:
418 if (g_queue_is_empty(&conn->primary_list)) {
419 return;
420 }
421 ppkt = g_queue_pop_tail(&conn->primary_list);
422 sec:
423 if (g_queue_is_empty(&conn->secondary_list)) {
424 g_queue_push_tail(&conn->primary_list, ppkt);
425 return;
426 }
427 spkt = g_queue_pop_tail(&conn->secondary_list);
428
429 if (ppkt->tcp_seq == ppkt->seq_end) {
430 colo_release_primary_pkt(s, ppkt);
431 ppkt = NULL;
432 }
433
434 if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
435 trace_colo_compare_main("pri: this packet has compared");
436 colo_release_primary_pkt(s, ppkt);
437 ppkt = NULL;
438 }
439
440 if (spkt->tcp_seq == spkt->seq_end) {
441 packet_destroy(spkt, NULL);
442 if (!ppkt) {
443 goto pri;
444 } else {
445 goto sec;
446 }
447 } else {
448 if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
449 trace_colo_compare_main("sec: this packet has compared");
450 packet_destroy(spkt, NULL);
451 if (!ppkt) {
452 goto pri;
453 } else {
454 goto sec;
455 }
456 }
457 if (!ppkt) {
458 g_queue_push_tail(&conn->secondary_list, spkt);
459 goto pri;
460 }
461 }
462
463 if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
464 trace_colo_compare_tcp_info("pri",
465 ppkt->tcp_seq, ppkt->tcp_ack,
466 ppkt->header_size, ppkt->payload_size,
467 ppkt->offset, ppkt->flags);
468
469 trace_colo_compare_tcp_info("sec",
470 spkt->tcp_seq, spkt->tcp_ack,
471 spkt->header_size, spkt->payload_size,
472 spkt->offset, spkt->flags);
473
474 if (mark == COLO_COMPARE_FREE_PRIMARY) {
475 conn->compare_seq = ppkt->seq_end;
476 colo_release_primary_pkt(s, ppkt);
477 g_queue_push_tail(&conn->secondary_list, spkt);
478 goto pri;
479 } else if (mark == COLO_COMPARE_FREE_SECONDARY) {
480 conn->compare_seq = spkt->seq_end;
481 packet_destroy(spkt, NULL);
482 goto sec;
483 } else if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
484 conn->compare_seq = ppkt->seq_end;
485 colo_release_primary_pkt(s, ppkt);
486 packet_destroy(spkt, NULL);
487 goto pri;
488 }
489 } else {
490 g_queue_push_tail(&conn->primary_list, ppkt);
491 g_queue_push_tail(&conn->secondary_list, spkt);
492
493 #ifdef DEBUG_COLO_PACKETS
494 qemu_hexdump(stderr, "colo-compare ppkt", ppkt->data, ppkt->size);
495 qemu_hexdump(stderr, "colo-compare spkt", spkt->data, spkt->size);
496 #endif
497
498 colo_compare_inconsistency_notify(s);
499 }
500 }
501
502
503 /*
504 * Called from the compare thread on the primary
505 * for compare udp packet
506 */
colo_packet_compare_udp(Packet * spkt,Packet * ppkt)507 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
508 {
509 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
510 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
511
512 trace_colo_compare_main("compare udp");
513
514 /*
515 * Because of ppkt and spkt are both in the same connection,
516 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
517 * same with spkt. In addition, IP header's Identification is a random
518 * field, we can handle it in IP fragmentation function later.
519 * COLO just concern the response net packet payload from primary guest
520 * and secondary guest are same or not, So we ignored all IP header include
521 * other field like TOS,TTL,IP Checksum. we only need to compare
522 * the ip payload here.
523 */
524 if (ppkt->size != spkt->size) {
525 trace_colo_compare_main("UDP: payload size of packets are different");
526 return -1;
527 }
528 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
529 ppkt->size - offset)) {
530 trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
531 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
532 #ifdef DEBUG_COLO_PACKETS
533 qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
534 qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
535 #endif
536 return -1;
537 } else {
538 return 0;
539 }
540 }
541
542 /*
543 * Called from the compare thread on the primary
544 * for compare icmp packet
545 */
colo_packet_compare_icmp(Packet * spkt,Packet * ppkt)546 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
547 {
548 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
549 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
550
551 trace_colo_compare_main("compare icmp");
552
553 /*
554 * Because of ppkt and spkt are both in the same connection,
555 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
556 * same with spkt. In addition, IP header's Identification is a random
557 * field, we can handle it in IP fragmentation function later.
558 * COLO just concern the response net packet payload from primary guest
559 * and secondary guest are same or not, So we ignored all IP header include
560 * other field like TOS,TTL,IP Checksum. we only need to compare
561 * the ip payload here.
562 */
563 if (ppkt->size != spkt->size) {
564 trace_colo_compare_main("ICMP: payload size of packets are different");
565 return -1;
566 }
567 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
568 ppkt->size - offset)) {
569 trace_colo_compare_icmp_miscompare("primary pkt size",
570 ppkt->size);
571 trace_colo_compare_icmp_miscompare("Secondary pkt size",
572 spkt->size);
573 #ifdef DEBUG_COLO_PACKETS
574 qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
575 qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
576 #endif
577 return -1;
578 } else {
579 return 0;
580 }
581 }
582
583 /*
584 * Called from the compare thread on the primary
585 * for compare other packet
586 */
colo_packet_compare_other(Packet * spkt,Packet * ppkt)587 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
588 {
589 uint16_t offset = ppkt->vnet_hdr_len;
590
591 trace_colo_compare_main("compare other");
592 if (ppkt->size != spkt->size) {
593 trace_colo_compare_main("Other: payload size of packets are different");
594 return -1;
595 }
596 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
597 ppkt->size - offset);
598 }
599
colo_old_packet_check_one(Packet * pkt,int64_t * check_time)600 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
601 {
602 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
603
604 if ((now - pkt->creation_ms) > (*check_time)) {
605 trace_colo_old_packet_check_found(pkt->creation_ms);
606 return 0;
607 } else {
608 return 1;
609 }
610 }
611
colo_compare_register_notifier(Notifier * notify)612 void colo_compare_register_notifier(Notifier *notify)
613 {
614 notifier_list_add(&colo_compare_notifiers, notify);
615 }
616
colo_compare_unregister_notifier(Notifier * notify)617 void colo_compare_unregister_notifier(Notifier *notify)
618 {
619 notifier_remove(notify);
620 }
621
colo_old_packet_check_one_conn(Connection * conn,CompareState * s)622 static int colo_old_packet_check_one_conn(Connection *conn,
623 CompareState *s)
624 {
625 if (!g_queue_is_empty(&conn->primary_list)) {
626 if (g_queue_find_custom(&conn->primary_list,
627 &s->compare_timeout,
628 (GCompareFunc)colo_old_packet_check_one))
629 goto out;
630 }
631
632 if (!g_queue_is_empty(&conn->secondary_list)) {
633 if (g_queue_find_custom(&conn->secondary_list,
634 &s->compare_timeout,
635 (GCompareFunc)colo_old_packet_check_one))
636 goto out;
637 }
638
639 return 1;
640
641 out:
642 /* Do checkpoint will flush old packet */
643 colo_compare_inconsistency_notify(s);
644 return 0;
645 }
646
647 /*
648 * Look for old packets that the secondary hasn't matched,
649 * if we have some then we have to checkpoint to wake
650 * the secondary up.
651 */
colo_old_packet_check(void * opaque)652 static void colo_old_packet_check(void *opaque)
653 {
654 CompareState *s = opaque;
655
656 /*
657 * If we find one old packet, stop finding job and notify
658 * COLO frame do checkpoint.
659 */
660 g_queue_find_custom(&s->conn_list, s,
661 (GCompareFunc)colo_old_packet_check_one_conn);
662 }
663
colo_compare_packet(CompareState * s,Connection * conn,int (* HandlePacket)(Packet * spkt,Packet * ppkt))664 static void colo_compare_packet(CompareState *s, Connection *conn,
665 int (*HandlePacket)(Packet *spkt,
666 Packet *ppkt))
667 {
668 Packet *pkt = NULL;
669 GList *result = NULL;
670
671 while (!g_queue_is_empty(&conn->primary_list) &&
672 !g_queue_is_empty(&conn->secondary_list)) {
673 pkt = g_queue_pop_tail(&conn->primary_list);
674 result = g_queue_find_custom(&conn->secondary_list,
675 pkt, (GCompareFunc)HandlePacket);
676
677 if (result) {
678 colo_release_primary_pkt(s, pkt);
679 packet_destroy(result->data, NULL);
680 g_queue_delete_link(&conn->secondary_list, result);
681 } else {
682 /*
683 * If one packet arrive late, the secondary_list or
684 * primary_list will be empty, so we can't compare it
685 * until next comparison. If the packets in the list are
686 * timeout, it will trigger a checkpoint request.
687 */
688 trace_colo_compare_main("packet different");
689 g_queue_push_tail(&conn->primary_list, pkt);
690
691 colo_compare_inconsistency_notify(s);
692 break;
693 }
694 }
695 }
696
697 /*
698 * Called from the compare thread on the primary
699 * for compare packet with secondary list of the
700 * specified connection when a new packet was
701 * queued to it.
702 */
colo_compare_connection(void * opaque,void * user_data)703 static void colo_compare_connection(void *opaque, void *user_data)
704 {
705 CompareState *s = user_data;
706 Connection *conn = opaque;
707
708 switch (conn->ip_proto) {
709 case IPPROTO_TCP:
710 colo_compare_tcp(s, conn);
711 break;
712 case IPPROTO_UDP:
713 colo_compare_packet(s, conn, colo_packet_compare_udp);
714 break;
715 case IPPROTO_ICMP:
716 colo_compare_packet(s, conn, colo_packet_compare_icmp);
717 break;
718 default:
719 colo_compare_packet(s, conn, colo_packet_compare_other);
720 break;
721 }
722 }
723
_compare_chr_send(void * opaque)724 static void coroutine_fn _compare_chr_send(void *opaque)
725 {
726 SendCo *sendco = opaque;
727 CompareState *s = sendco->s;
728 int ret = 0;
729
730 while (!g_queue_is_empty(&sendco->send_list)) {
731 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
732 uint32_t len = htonl(entry->size);
733
734 ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
735
736 if (ret != sizeof(len)) {
737 g_free(entry->buf);
738 g_slice_free(SendEntry, entry);
739 goto err;
740 }
741
742 if (!sendco->notify_remote_frame && s->vnet_hdr) {
743 /*
744 * We send vnet header len make other module(like filter-redirector)
745 * know how to parse net packet correctly.
746 */
747 len = htonl(entry->vnet_hdr_len);
748
749 ret = qemu_chr_fe_write_all(sendco->chr,
750 (uint8_t *)&len,
751 sizeof(len));
752
753 if (ret != sizeof(len)) {
754 g_free(entry->buf);
755 g_slice_free(SendEntry, entry);
756 goto err;
757 }
758 }
759
760 ret = qemu_chr_fe_write_all(sendco->chr,
761 (uint8_t *)entry->buf,
762 entry->size);
763
764 if (ret != entry->size) {
765 g_free(entry->buf);
766 g_slice_free(SendEntry, entry);
767 goto err;
768 }
769
770 g_free(entry->buf);
771 g_slice_free(SendEntry, entry);
772 }
773
774 sendco->ret = 0;
775 goto out;
776
777 err:
778 while (!g_queue_is_empty(&sendco->send_list)) {
779 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
780 g_free(entry->buf);
781 g_slice_free(SendEntry, entry);
782 }
783 sendco->ret = ret < 0 ? ret : -EIO;
784 out:
785 sendco->co = NULL;
786 sendco->done = true;
787 aio_wait_kick();
788 }
789
compare_chr_send(CompareState * s,uint8_t * buf,uint32_t size,uint32_t vnet_hdr_len,bool notify_remote_frame,bool zero_copy)790 static int compare_chr_send(CompareState *s,
791 uint8_t *buf,
792 uint32_t size,
793 uint32_t vnet_hdr_len,
794 bool notify_remote_frame,
795 bool zero_copy)
796 {
797 SendCo *sendco;
798 SendEntry *entry;
799
800 if (notify_remote_frame) {
801 sendco = &s->notify_sendco;
802 } else {
803 sendco = &s->out_sendco;
804 }
805
806 if (!size) {
807 return -1;
808 }
809
810 entry = g_slice_new(SendEntry);
811 entry->size = size;
812 entry->vnet_hdr_len = vnet_hdr_len;
813 if (zero_copy) {
814 entry->buf = buf;
815 } else {
816 entry->buf = g_malloc(size);
817 memcpy(entry->buf, buf, size);
818 }
819 g_queue_push_tail(&sendco->send_list, entry);
820
821 if (sendco->done) {
822 sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
823 sendco->done = false;
824 qemu_coroutine_enter(sendco->co);
825 if (sendco->done) {
826 /* report early errors */
827 return sendco->ret;
828 }
829 }
830
831 /* assume success */
832 return 0;
833 }
834
compare_chr_can_read(void * opaque)835 static int compare_chr_can_read(void *opaque)
836 {
837 return COMPARE_READ_LEN_MAX;
838 }
839
840 /*
841 * Called from the main thread on the primary for packets
842 * arriving over the socket from the primary.
843 */
compare_pri_chr_in(void * opaque,const uint8_t * buf,int size)844 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
845 {
846 CompareState *s = COLO_COMPARE(opaque);
847 int ret;
848
849 ret = net_fill_rstate(&s->pri_rs, buf, size);
850 if (ret == -1) {
851 qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
852 NULL, NULL, true);
853 error_report("colo-compare primary_in error");
854 }
855 }
856
857 /*
858 * Called from the main thread on the primary for packets
859 * arriving over the socket from the secondary.
860 */
compare_sec_chr_in(void * opaque,const uint8_t * buf,int size)861 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
862 {
863 CompareState *s = COLO_COMPARE(opaque);
864 int ret;
865
866 ret = net_fill_rstate(&s->sec_rs, buf, size);
867 if (ret == -1) {
868 qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
869 NULL, NULL, true);
870 error_report("colo-compare secondary_in error");
871 }
872 }
873
compare_notify_chr(void * opaque,const uint8_t * buf,int size)874 static void compare_notify_chr(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->notify_rs, buf, size);
880 if (ret == -1) {
881 qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
882 NULL, NULL, true);
883 error_report("colo-compare notify_dev error");
884 }
885 }
886
887 /*
888 * Check old packet regularly so it can watch for any packets
889 * that the secondary hasn't produced equivalents of.
890 */
check_old_packet_regular(void * opaque)891 static void check_old_packet_regular(void *opaque)
892 {
893 CompareState *s = opaque;
894
895 /* if have old packet we will notify checkpoint */
896 colo_old_packet_check(s);
897 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
898 s->expired_scan_cycle);
899 }
900
901 /* Public API, Used for COLO frame to notify compare event */
colo_notify_compares_event(void * opaque,int event,Error ** errp)902 void colo_notify_compares_event(void *opaque, int event, Error **errp)
903 {
904 CompareState *s;
905 qemu_mutex_lock(&colo_compare_mutex);
906
907 if (!colo_compare_active) {
908 qemu_mutex_unlock(&colo_compare_mutex);
909 return;
910 }
911
912 qemu_mutex_lock(&event_mtx);
913 QTAILQ_FOREACH(s, &net_compares, next) {
914 s->event = event;
915 qemu_bh_schedule(s->event_bh);
916 event_unhandled_count++;
917 }
918 /* Wait all compare threads to finish handling this event */
919 while (event_unhandled_count > 0) {
920 qemu_cond_wait(&event_complete_cond, &event_mtx);
921 }
922
923 qemu_mutex_unlock(&event_mtx);
924 qemu_mutex_unlock(&colo_compare_mutex);
925 }
926
colo_compare_timer_init(CompareState * s)927 static void colo_compare_timer_init(CompareState *s)
928 {
929 AioContext *ctx = iothread_get_aio_context(s->iothread);
930
931 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_HOST,
932 SCALE_MS, check_old_packet_regular,
933 s);
934 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_HOST) +
935 s->expired_scan_cycle);
936 }
937
colo_compare_timer_del(CompareState * s)938 static void colo_compare_timer_del(CompareState *s)
939 {
940 if (s->packet_check_timer) {
941 timer_free(s->packet_check_timer);
942 s->packet_check_timer = NULL;
943 }
944 }
945
946 static void colo_flush_packets(void *opaque, void *user_data);
947
colo_compare_handle_event(void * opaque)948 static void colo_compare_handle_event(void *opaque)
949 {
950 CompareState *s = opaque;
951
952 switch (s->event) {
953 case COLO_EVENT_CHECKPOINT:
954 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
955 break;
956 case COLO_EVENT_FAILOVER:
957 break;
958 default:
959 break;
960 }
961
962 qemu_mutex_lock(&event_mtx);
963 assert(event_unhandled_count > 0);
964 event_unhandled_count--;
965 qemu_cond_broadcast(&event_complete_cond);
966 qemu_mutex_unlock(&event_mtx);
967 }
968
colo_compare_iothread(CompareState * s)969 static void colo_compare_iothread(CompareState *s)
970 {
971 AioContext *ctx = iothread_get_aio_context(s->iothread);
972 object_ref(OBJECT(s->iothread));
973 s->worker_context = iothread_get_g_main_context(s->iothread);
974
975 qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
976 compare_pri_chr_in, NULL, NULL,
977 s, s->worker_context, true);
978 qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
979 compare_sec_chr_in, NULL, NULL,
980 s, s->worker_context, true);
981 if (s->notify_dev) {
982 qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
983 compare_notify_chr, NULL, NULL,
984 s, s->worker_context, true);
985 }
986
987 colo_compare_timer_init(s);
988 s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
989 }
990
compare_get_pri_indev(Object * obj,Error ** errp)991 static char *compare_get_pri_indev(Object *obj, Error **errp)
992 {
993 CompareState *s = COLO_COMPARE(obj);
994
995 return g_strdup(s->pri_indev);
996 }
997
compare_set_pri_indev(Object * obj,const char * value,Error ** errp)998 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
999 {
1000 CompareState *s = COLO_COMPARE(obj);
1001
1002 g_free(s->pri_indev);
1003 s->pri_indev = g_strdup(value);
1004 }
1005
compare_get_sec_indev(Object * obj,Error ** errp)1006 static char *compare_get_sec_indev(Object *obj, Error **errp)
1007 {
1008 CompareState *s = COLO_COMPARE(obj);
1009
1010 return g_strdup(s->sec_indev);
1011 }
1012
compare_set_sec_indev(Object * obj,const char * value,Error ** errp)1013 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1014 {
1015 CompareState *s = COLO_COMPARE(obj);
1016
1017 g_free(s->sec_indev);
1018 s->sec_indev = g_strdup(value);
1019 }
1020
compare_get_outdev(Object * obj,Error ** errp)1021 static char *compare_get_outdev(Object *obj, Error **errp)
1022 {
1023 CompareState *s = COLO_COMPARE(obj);
1024
1025 return g_strdup(s->outdev);
1026 }
1027
compare_set_outdev(Object * obj,const char * value,Error ** errp)1028 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1029 {
1030 CompareState *s = COLO_COMPARE(obj);
1031
1032 g_free(s->outdev);
1033 s->outdev = g_strdup(value);
1034 }
1035
compare_get_vnet_hdr(Object * obj,Error ** errp)1036 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1037 {
1038 CompareState *s = COLO_COMPARE(obj);
1039
1040 return s->vnet_hdr;
1041 }
1042
compare_set_vnet_hdr(Object * obj,bool value,Error ** errp)1043 static void compare_set_vnet_hdr(Object *obj,
1044 bool value,
1045 Error **errp)
1046 {
1047 CompareState *s = COLO_COMPARE(obj);
1048
1049 s->vnet_hdr = value;
1050 }
1051
compare_get_notify_dev(Object * obj,Error ** errp)1052 static char *compare_get_notify_dev(Object *obj, Error **errp)
1053 {
1054 CompareState *s = COLO_COMPARE(obj);
1055
1056 return g_strdup(s->notify_dev);
1057 }
1058
compare_set_notify_dev(Object * obj,const char * value,Error ** errp)1059 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1060 {
1061 CompareState *s = COLO_COMPARE(obj);
1062
1063 g_free(s->notify_dev);
1064 s->notify_dev = g_strdup(value);
1065 }
1066
compare_get_timeout(Object * obj,Visitor * v,const char * name,void * opaque,Error ** errp)1067 static void compare_get_timeout(Object *obj, Visitor *v,
1068 const char *name, void *opaque,
1069 Error **errp)
1070 {
1071 CompareState *s = COLO_COMPARE(obj);
1072 uint64_t value = s->compare_timeout;
1073
1074 visit_type_uint64(v, name, &value, errp);
1075 }
1076
compare_set_timeout(Object * obj,Visitor * v,const char * name,void * opaque,Error ** errp)1077 static void compare_set_timeout(Object *obj, Visitor *v,
1078 const char *name, void *opaque,
1079 Error **errp)
1080 {
1081 CompareState *s = COLO_COMPARE(obj);
1082 uint32_t value;
1083
1084 if (!visit_type_uint32(v, name, &value, errp)) {
1085 return;
1086 }
1087 if (!value) {
1088 error_setg(errp, "Property '%s.%s' requires a positive value",
1089 object_get_typename(obj), name);
1090 return;
1091 }
1092 s->compare_timeout = value;
1093 }
1094
compare_get_expired_scan_cycle(Object * obj,Visitor * v,const char * name,void * opaque,Error ** errp)1095 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1096 const char *name, void *opaque,
1097 Error **errp)
1098 {
1099 CompareState *s = COLO_COMPARE(obj);
1100 uint32_t value = s->expired_scan_cycle;
1101
1102 visit_type_uint32(v, name, &value, errp);
1103 }
1104
compare_set_expired_scan_cycle(Object * obj,Visitor * v,const char * name,void * opaque,Error ** errp)1105 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1106 const char *name, void *opaque,
1107 Error **errp)
1108 {
1109 CompareState *s = COLO_COMPARE(obj);
1110 uint32_t value;
1111
1112 if (!visit_type_uint32(v, name, &value, errp)) {
1113 return;
1114 }
1115 if (!value) {
1116 error_setg(errp, "Property '%s.%s' requires a positive value",
1117 object_get_typename(obj), name);
1118 return;
1119 }
1120 s->expired_scan_cycle = value;
1121 }
1122
get_max_queue_size(Object * obj,Visitor * v,const char * name,void * opaque,Error ** errp)1123 static void get_max_queue_size(Object *obj, Visitor *v,
1124 const char *name, void *opaque,
1125 Error **errp)
1126 {
1127 uint32_t value = max_queue_size;
1128
1129 visit_type_uint32(v, name, &value, errp);
1130 }
1131
set_max_queue_size(Object * obj,Visitor * v,const char * name,void * opaque,Error ** errp)1132 static void set_max_queue_size(Object *obj, Visitor *v,
1133 const char *name, void *opaque,
1134 Error **errp)
1135 {
1136 uint64_t value;
1137
1138 if (!visit_type_uint64(v, name, &value, errp)) {
1139 return;
1140 }
1141 if (!value) {
1142 error_setg(errp, "Property '%s.%s' requires a positive value",
1143 object_get_typename(obj), name);
1144 return;
1145 }
1146 max_queue_size = value;
1147 }
1148
compare_pri_rs_finalize(SocketReadState * pri_rs)1149 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1150 {
1151 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1152 Connection *conn = NULL;
1153
1154 if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1155 trace_colo_compare_main("primary: unsupported packet in");
1156 compare_chr_send(s,
1157 pri_rs->buf,
1158 pri_rs->packet_len,
1159 pri_rs->vnet_hdr_len,
1160 false,
1161 false);
1162 } else {
1163 /* compare packet in the specified connection */
1164 colo_compare_connection(conn, s);
1165 }
1166 }
1167
compare_sec_rs_finalize(SocketReadState * sec_rs)1168 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1169 {
1170 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1171 Connection *conn = NULL;
1172
1173 if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1174 trace_colo_compare_main("secondary: unsupported packet in");
1175 } else {
1176 /* compare packet in the specified connection */
1177 colo_compare_connection(conn, s);
1178 }
1179 }
1180
compare_notify_rs_finalize(SocketReadState * notify_rs)1181 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1182 {
1183 CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1184
1185 const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1186 int ret;
1187
1188 if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1189 notify_rs->buf,
1190 notify_rs->packet_len)) {
1191 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1192 if (ret < 0) {
1193 error_report("Notify Xen COLO-frame INIT failed");
1194 }
1195 } else if (packet_matches_str("COLO_CHECKPOINT",
1196 notify_rs->buf,
1197 notify_rs->packet_len)) {
1198 /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1199 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1200 } else {
1201 error_report("COLO compare got unsupported instruction");
1202 }
1203 }
1204
1205 /*
1206 * Return 0 is success.
1207 * Return 1 is failed.
1208 */
find_and_check_chardev(Chardev ** chr,char * chr_name,Error ** errp)1209 static int find_and_check_chardev(Chardev **chr,
1210 char *chr_name,
1211 Error **errp)
1212 {
1213 *chr = qemu_chr_find(chr_name);
1214 if (*chr == NULL) {
1215 error_setg(errp, "Device '%s' not found",
1216 chr_name);
1217 return 1;
1218 }
1219
1220 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1221 error_setg(errp, "chardev \"%s\" is not reconnectable",
1222 chr_name);
1223 return 1;
1224 }
1225
1226 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1227 error_setg(errp, "chardev \"%s\" cannot switch context",
1228 chr_name);
1229 return 1;
1230 }
1231
1232 return 0;
1233 }
1234
1235 /*
1236 * Called from the main thread on the primary
1237 * to setup colo-compare.
1238 */
colo_compare_complete(UserCreatable * uc,Error ** errp)1239 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1240 {
1241 CompareState *s = COLO_COMPARE(uc);
1242 Chardev *chr;
1243
1244 if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1245 error_setg(errp, "colo compare needs 'primary_in' ,"
1246 "'secondary_in','outdev','iothread' property set");
1247 return;
1248 } else if (!strcmp(s->pri_indev, s->outdev) ||
1249 !strcmp(s->sec_indev, s->outdev) ||
1250 !strcmp(s->pri_indev, s->sec_indev)) {
1251 error_setg(errp, "'indev' and 'outdev' could not be same "
1252 "for compare module");
1253 return;
1254 }
1255
1256 if (!s->compare_timeout) {
1257 /* Set default value to 3000 MS */
1258 s->compare_timeout = DEFAULT_TIME_OUT_MS;
1259 }
1260
1261 if (!s->expired_scan_cycle) {
1262 /* Set default value to 1000 MS */
1263 s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1264 }
1265
1266 if (!max_queue_size) {
1267 /* Set default queue size to 1024 */
1268 max_queue_size = MAX_QUEUE_SIZE;
1269 }
1270
1271 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1272 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1273 return;
1274 }
1275
1276 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1277 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1278 return;
1279 }
1280
1281 if (find_and_check_chardev(&chr, s->outdev, errp) ||
1282 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1283 return;
1284 }
1285
1286 net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1287 net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1288
1289 /* Try to enable remote notify chardev, currently just for Xen COLO */
1290 if (s->notify_dev) {
1291 if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1292 !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1293 return;
1294 }
1295
1296 net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1297 s->vnet_hdr);
1298 }
1299
1300 s->out_sendco.s = s;
1301 s->out_sendco.chr = &s->chr_out;
1302 s->out_sendco.notify_remote_frame = false;
1303 s->out_sendco.done = true;
1304 g_queue_init(&s->out_sendco.send_list);
1305
1306 if (s->notify_dev) {
1307 s->notify_sendco.s = s;
1308 s->notify_sendco.chr = &s->chr_notify_dev;
1309 s->notify_sendco.notify_remote_frame = true;
1310 s->notify_sendco.done = true;
1311 g_queue_init(&s->notify_sendco.send_list);
1312 }
1313
1314 g_queue_init(&s->conn_list);
1315
1316 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1317 connection_key_equal,
1318 g_free,
1319 NULL);
1320
1321 colo_compare_iothread(s);
1322
1323 qemu_mutex_lock(&colo_compare_mutex);
1324 if (!colo_compare_active) {
1325 qemu_mutex_init(&event_mtx);
1326 qemu_cond_init(&event_complete_cond);
1327 colo_compare_active = true;
1328 }
1329 QTAILQ_INSERT_TAIL(&net_compares, s, next);
1330 qemu_mutex_unlock(&colo_compare_mutex);
1331
1332 return;
1333 }
1334
colo_flush_packets(void * opaque,void * user_data)1335 static void colo_flush_packets(void *opaque, void *user_data)
1336 {
1337 CompareState *s = user_data;
1338 Connection *conn = opaque;
1339 Packet *pkt = NULL;
1340
1341 while (!g_queue_is_empty(&conn->primary_list)) {
1342 pkt = g_queue_pop_tail(&conn->primary_list);
1343 compare_chr_send(s,
1344 pkt->data,
1345 pkt->size,
1346 pkt->vnet_hdr_len,
1347 false,
1348 true);
1349 packet_destroy_partial(pkt, NULL);
1350 }
1351 while (!g_queue_is_empty(&conn->secondary_list)) {
1352 pkt = g_queue_pop_tail(&conn->secondary_list);
1353 packet_destroy(pkt, NULL);
1354 }
1355 }
1356
colo_compare_class_init(ObjectClass * oc,void * data)1357 static void colo_compare_class_init(ObjectClass *oc, void *data)
1358 {
1359 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1360
1361 ucc->complete = colo_compare_complete;
1362 }
1363
colo_compare_init(Object * obj)1364 static void colo_compare_init(Object *obj)
1365 {
1366 CompareState *s = COLO_COMPARE(obj);
1367
1368 object_property_add_str(obj, "primary_in",
1369 compare_get_pri_indev, compare_set_pri_indev);
1370 object_property_add_str(obj, "secondary_in",
1371 compare_get_sec_indev, compare_set_sec_indev);
1372 object_property_add_str(obj, "outdev",
1373 compare_get_outdev, compare_set_outdev);
1374 object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1375 (Object **)&s->iothread,
1376 object_property_allow_set_link,
1377 OBJ_PROP_LINK_STRONG);
1378 /* This parameter just for Xen COLO */
1379 object_property_add_str(obj, "notify_dev",
1380 compare_get_notify_dev, compare_set_notify_dev);
1381
1382 object_property_add(obj, "compare_timeout", "uint64",
1383 compare_get_timeout,
1384 compare_set_timeout, NULL, NULL);
1385
1386 object_property_add(obj, "expired_scan_cycle", "uint32",
1387 compare_get_expired_scan_cycle,
1388 compare_set_expired_scan_cycle, NULL, NULL);
1389
1390 object_property_add(obj, "max_queue_size", "uint32",
1391 get_max_queue_size,
1392 set_max_queue_size, NULL, NULL);
1393
1394 s->vnet_hdr = false;
1395 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1396 compare_set_vnet_hdr);
1397 }
1398
colo_compare_cleanup(void)1399 void colo_compare_cleanup(void)
1400 {
1401 CompareState *tmp = NULL;
1402 CompareState *n = NULL;
1403
1404 QTAILQ_FOREACH_SAFE(tmp, &net_compares, next, n) {
1405 object_unparent(OBJECT(tmp));
1406 }
1407 }
1408
colo_compare_finalize(Object * obj)1409 static void colo_compare_finalize(Object *obj)
1410 {
1411 CompareState *s = COLO_COMPARE(obj);
1412 CompareState *tmp = NULL;
1413
1414 qemu_mutex_lock(&colo_compare_mutex);
1415 QTAILQ_FOREACH(tmp, &net_compares, next) {
1416 if (tmp == s) {
1417 QTAILQ_REMOVE(&net_compares, s, next);
1418 break;
1419 }
1420 }
1421 if (QTAILQ_EMPTY(&net_compares)) {
1422 colo_compare_active = false;
1423 qemu_mutex_destroy(&event_mtx);
1424 qemu_cond_destroy(&event_complete_cond);
1425 }
1426 qemu_mutex_unlock(&colo_compare_mutex);
1427
1428 qemu_chr_fe_deinit(&s->chr_pri_in, false);
1429 qemu_chr_fe_deinit(&s->chr_sec_in, false);
1430 qemu_chr_fe_deinit(&s->chr_out, false);
1431 if (s->notify_dev) {
1432 qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1433 }
1434
1435 colo_compare_timer_del(s);
1436
1437 qemu_bh_delete(s->event_bh);
1438
1439 AioContext *ctx = iothread_get_aio_context(s->iothread);
1440 AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1441 if (s->notify_dev) {
1442 AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1443 }
1444
1445 /* Release all unhandled packets after compare thead exited */
1446 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1447 AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1448
1449 g_queue_clear(&s->conn_list);
1450 g_queue_clear(&s->out_sendco.send_list);
1451 if (s->notify_dev) {
1452 g_queue_clear(&s->notify_sendco.send_list);
1453 }
1454
1455 if (s->connection_track_table) {
1456 g_hash_table_destroy(s->connection_track_table);
1457 }
1458
1459 object_unref(OBJECT(s->iothread));
1460
1461 g_free(s->pri_indev);
1462 g_free(s->sec_indev);
1463 g_free(s->outdev);
1464 g_free(s->notify_dev);
1465 }
1466
colo_compare_init_globals(void)1467 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1468 {
1469 colo_compare_active = false;
1470 qemu_mutex_init(&colo_compare_mutex);
1471 }
1472
1473 static const TypeInfo colo_compare_info = {
1474 .name = TYPE_COLO_COMPARE,
1475 .parent = TYPE_OBJECT,
1476 .instance_size = sizeof(CompareState),
1477 .instance_init = colo_compare_init,
1478 .instance_finalize = colo_compare_finalize,
1479 .class_size = sizeof(CompareClass),
1480 .class_init = colo_compare_class_init,
1481 .interfaces = (InterfaceInfo[]) {
1482 { TYPE_USER_CREATABLE },
1483 { }
1484 }
1485 };
1486
register_types(void)1487 static void register_types(void)
1488 {
1489 type_register_static(&colo_compare_info);
1490 }
1491
1492 type_init(register_types);
1493