monitor/hmp: Move hmp_drive_add_node to block-hmp-cmds.c
[qemu/ar7.git] / net / colo-compare.c
blob7ee17f2cf8026e335dfc4f314cc92d3338d837a4
1 /*
2 * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
3 * (a.k.a. Fault Tolerance or Continuous Replication)
5 * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
6 * Copyright (c) 2016 FUJITSU LIMITED
7 * Copyright (c) 2016 Intel Corporation
9 * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
11 * This work is licensed under the terms of the GNU GPL, version 2 or
12 * later. See the COPYING file in the top-level directory.
15 #include "qemu/osdep.h"
16 #include "qemu-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"
35 #define TYPE_COLO_COMPARE "colo-compare"
36 #define COLO_COMPARE(obj) \
37 OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
39 static QTAILQ_HEAD(, CompareState) net_compares =
40 QTAILQ_HEAD_INITIALIZER(net_compares);
42 static NotifierList colo_compare_notifiers =
43 NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
45 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
46 #define MAX_QUEUE_SIZE 1024
48 #define COLO_COMPARE_FREE_PRIMARY 0x01
49 #define COLO_COMPARE_FREE_SECONDARY 0x02
51 /* TODO: Should be configurable */
52 #define REGULAR_PACKET_CHECK_MS 3000
54 static QemuMutex event_mtx;
55 static QemuCond event_complete_cond;
56 static int event_unhandled_count;
59 * + CompareState ++
60 * | |
61 * +---------------+ +---------------+ +---------------+
62 * | conn list + - > conn + ------- > conn + -- > ......
63 * +---------------+ +---------------+ +---------------+
64 * | | | | | |
65 * +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
66 * |primary | |secondary |primary | |secondary
67 * |packet | |packet + |packet | |packet +
68 * +--------+ +--------+ +--------+ +--------+
69 * | | | |
70 * +---v----+ +---v----+ +---v----+ +---v----+
71 * |primary | |secondary |primary | |secondary
72 * |packet | |packet + |packet | |packet +
73 * +--------+ +--------+ +--------+ +--------+
74 * | | | |
75 * +---v----+ +---v----+ +---v----+ +---v----+
76 * |primary | |secondary |primary | |secondary
77 * |packet | |packet + |packet | |packet +
78 * +--------+ +--------+ +--------+ +--------+
80 typedef struct CompareState {
81 Object parent;
83 char *pri_indev;
84 char *sec_indev;
85 char *outdev;
86 char *notify_dev;
87 CharBackend chr_pri_in;
88 CharBackend chr_sec_in;
89 CharBackend chr_out;
90 CharBackend chr_notify_dev;
91 SocketReadState pri_rs;
92 SocketReadState sec_rs;
93 SocketReadState notify_rs;
94 bool vnet_hdr;
97 * Record the connection that through the NIC
98 * Element type: Connection
100 GQueue conn_list;
101 /* Record the connection without repetition */
102 GHashTable *connection_track_table;
104 IOThread *iothread;
105 GMainContext *worker_context;
106 QEMUTimer *packet_check_timer;
108 QEMUBH *event_bh;
109 enum colo_event event;
111 QTAILQ_ENTRY(CompareState) next;
112 } CompareState;
114 typedef struct CompareClass {
115 ObjectClass parent_class;
116 } CompareClass;
118 enum {
119 PRIMARY_IN = 0,
120 SECONDARY_IN,
124 static int compare_chr_send(CompareState *s,
125 const uint8_t *buf,
126 uint32_t size,
127 uint32_t vnet_hdr_len,
128 bool notify_remote_frame);
130 static bool packet_matches_str(const char *str,
131 const uint8_t *buf,
132 uint32_t packet_len)
134 if (packet_len != strlen(str)) {
135 return false;
138 return !memcmp(str, buf, strlen(str));
141 static void notify_remote_frame(CompareState *s)
143 char msg[] = "DO_CHECKPOINT";
144 int ret = 0;
146 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true);
147 if (ret < 0) {
148 error_report("Notify Xen COLO-frame failed");
152 static void colo_compare_inconsistency_notify(CompareState *s)
154 if (s->notify_dev) {
155 notify_remote_frame(s);
156 } else {
157 notifier_list_notify(&colo_compare_notifiers,
158 migrate_get_current());
162 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
164 struct tcp_hdr *atcp, *btcp;
166 atcp = (struct tcp_hdr *)(a->transport_header);
167 btcp = (struct tcp_hdr *)(b->transport_header);
168 return ntohl(atcp->th_seq) - ntohl(btcp->th_seq);
171 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
173 Packet *pkt = data;
174 struct tcp_hdr *tcphd;
176 tcphd = (struct tcp_hdr *)pkt->transport_header;
178 pkt->tcp_seq = ntohl(tcphd->th_seq);
179 pkt->tcp_ack = ntohl(tcphd->th_ack);
180 *max_ack = *max_ack > pkt->tcp_ack ? *max_ack : pkt->tcp_ack;
181 pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
182 + (tcphd->th_off << 2) - pkt->vnet_hdr_len;
183 pkt->payload_size = pkt->size - pkt->header_size;
184 pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
185 pkt->flags = tcphd->th_flags;
189 * Return 1 on success, if return 0 means the
190 * packet will be dropped
192 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
194 if (g_queue_get_length(queue) <= MAX_QUEUE_SIZE) {
195 if (pkt->ip->ip_p == IPPROTO_TCP) {
196 fill_pkt_tcp_info(pkt, max_ack);
197 g_queue_insert_sorted(queue,
198 pkt,
199 (GCompareDataFunc)seq_sorter,
200 NULL);
201 } else {
202 g_queue_push_tail(queue, pkt);
204 return 1;
206 return 0;
210 * Return 0 on success, if return -1 means the pkt
211 * is unsupported(arp and ipv6) and will be sent later
213 static int packet_enqueue(CompareState *s, int mode, Connection **con)
215 ConnectionKey key;
216 Packet *pkt = NULL;
217 Connection *conn;
219 if (mode == PRIMARY_IN) {
220 pkt = packet_new(s->pri_rs.buf,
221 s->pri_rs.packet_len,
222 s->pri_rs.vnet_hdr_len);
223 } else {
224 pkt = packet_new(s->sec_rs.buf,
225 s->sec_rs.packet_len,
226 s->sec_rs.vnet_hdr_len);
229 if (parse_packet_early(pkt)) {
230 packet_destroy(pkt, NULL);
231 pkt = NULL;
232 return -1;
234 fill_connection_key(pkt, &key);
236 conn = connection_get(s->connection_track_table,
237 &key,
238 &s->conn_list);
240 if (!conn->processing) {
241 g_queue_push_tail(&s->conn_list, conn);
242 conn->processing = true;
245 if (mode == PRIMARY_IN) {
246 if (!colo_insert_packet(&conn->primary_list, pkt, &conn->pack)) {
247 error_report("colo compare primary queue size too big,"
248 "drop packet");
250 } else {
251 if (!colo_insert_packet(&conn->secondary_list, pkt, &conn->sack)) {
252 error_report("colo compare secondary queue size too big,"
253 "drop packet");
256 *con = conn;
258 return 0;
261 static inline bool after(uint32_t seq1, uint32_t seq2)
263 return (int32_t)(seq1 - seq2) > 0;
266 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
268 int ret;
269 ret = compare_chr_send(s,
270 pkt->data,
271 pkt->size,
272 pkt->vnet_hdr_len,
273 false);
274 if (ret < 0) {
275 error_report("colo send primary packet failed");
277 trace_colo_compare_main("packet same and release packet");
278 packet_destroy(pkt, NULL);
282 * The IP packets sent by primary and secondary
283 * will be compared in here
284 * TODO support ip fragment, Out-Of-Order
285 * return: 0 means packet same
286 * > 0 || < 0 means packet different
288 static int colo_compare_packet_payload(Packet *ppkt,
289 Packet *spkt,
290 uint16_t poffset,
291 uint16_t soffset,
292 uint16_t len)
295 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
296 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
298 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
299 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
300 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
301 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
303 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
304 pri_ip_dst, spkt->size,
305 sec_ip_src, sec_ip_dst);
308 return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
312 * return true means that the payload is consist and
313 * need to make the next comparison, false means do
314 * the checkpoint
316 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
317 int8_t *mark, uint32_t max_ack)
319 *mark = 0;
321 if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
322 if (!colo_compare_packet_payload(ppkt, spkt,
323 ppkt->header_size, spkt->header_size,
324 ppkt->payload_size)) {
325 *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
326 return true;
330 /* one part of secondary packet payload still need to be compared */
331 if (!after(ppkt->seq_end, spkt->seq_end)) {
332 if (!colo_compare_packet_payload(ppkt, spkt,
333 ppkt->header_size + ppkt->offset,
334 spkt->header_size + spkt->offset,
335 ppkt->payload_size - ppkt->offset)) {
336 if (!after(ppkt->tcp_ack, max_ack)) {
337 *mark = COLO_COMPARE_FREE_PRIMARY;
338 spkt->offset += ppkt->payload_size - ppkt->offset;
339 return true;
340 } else {
341 /* secondary guest hasn't ack the data, don't send
342 * out this packet
344 return false;
347 } else {
348 /* primary packet is longer than secondary packet, compare
349 * the same part and mark the primary packet offset
351 if (!colo_compare_packet_payload(ppkt, spkt,
352 ppkt->header_size + ppkt->offset,
353 spkt->header_size + spkt->offset,
354 spkt->payload_size - spkt->offset)) {
355 *mark = COLO_COMPARE_FREE_SECONDARY;
356 ppkt->offset += spkt->payload_size - spkt->offset;
357 return true;
361 return false;
364 static void colo_compare_tcp(CompareState *s, Connection *conn)
366 Packet *ppkt = NULL, *spkt = NULL;
367 int8_t mark;
370 * If ppkt and spkt have the same payload, but ppkt's ACK
371 * is greater than spkt's ACK, in this case we can not
372 * send the ppkt because it will cause the secondary guest
373 * to miss sending some data in the next. Therefore, we
374 * record the maximum ACK in the current queue at both
375 * primary side and secondary side. Only when the ack is
376 * less than the smaller of the two maximum ack, then we
377 * can ensure that the packet's payload is acknowledged by
378 * primary and secondary.
380 uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
382 pri:
383 if (g_queue_is_empty(&conn->primary_list)) {
384 return;
386 ppkt = g_queue_pop_head(&conn->primary_list);
387 sec:
388 if (g_queue_is_empty(&conn->secondary_list)) {
389 g_queue_push_head(&conn->primary_list, ppkt);
390 return;
392 spkt = g_queue_pop_head(&conn->secondary_list);
394 if (ppkt->tcp_seq == ppkt->seq_end) {
395 colo_release_primary_pkt(s, ppkt);
396 ppkt = NULL;
399 if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
400 trace_colo_compare_main("pri: this packet has compared");
401 colo_release_primary_pkt(s, ppkt);
402 ppkt = NULL;
405 if (spkt->tcp_seq == spkt->seq_end) {
406 packet_destroy(spkt, NULL);
407 if (!ppkt) {
408 goto pri;
409 } else {
410 goto sec;
412 } else {
413 if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
414 trace_colo_compare_main("sec: this packet has compared");
415 packet_destroy(spkt, NULL);
416 if (!ppkt) {
417 goto pri;
418 } else {
419 goto sec;
422 if (!ppkt) {
423 g_queue_push_head(&conn->secondary_list, spkt);
424 goto pri;
428 if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
429 trace_colo_compare_tcp_info("pri",
430 ppkt->tcp_seq, ppkt->tcp_ack,
431 ppkt->header_size, ppkt->payload_size,
432 ppkt->offset, ppkt->flags);
434 trace_colo_compare_tcp_info("sec",
435 spkt->tcp_seq, spkt->tcp_ack,
436 spkt->header_size, spkt->payload_size,
437 spkt->offset, spkt->flags);
439 if (mark == COLO_COMPARE_FREE_PRIMARY) {
440 conn->compare_seq = ppkt->seq_end;
441 colo_release_primary_pkt(s, ppkt);
442 g_queue_push_head(&conn->secondary_list, spkt);
443 goto pri;
445 if (mark == COLO_COMPARE_FREE_SECONDARY) {
446 conn->compare_seq = spkt->seq_end;
447 packet_destroy(spkt, NULL);
448 goto sec;
450 if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
451 conn->compare_seq = ppkt->seq_end;
452 colo_release_primary_pkt(s, ppkt);
453 packet_destroy(spkt, NULL);
454 goto pri;
456 } else {
457 g_queue_push_head(&conn->primary_list, ppkt);
458 g_queue_push_head(&conn->secondary_list, spkt);
460 qemu_hexdump((char *)ppkt->data, stderr,
461 "colo-compare ppkt", ppkt->size);
462 qemu_hexdump((char *)spkt->data, stderr,
463 "colo-compare spkt", spkt->size);
465 colo_compare_inconsistency_notify(s);
471 * Called from the compare thread on the primary
472 * for compare udp packet
474 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
476 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
477 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
479 trace_colo_compare_main("compare udp");
482 * Because of ppkt and spkt are both in the same connection,
483 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
484 * same with spkt. In addition, IP header's Identification is a random
485 * field, we can handle it in IP fragmentation function later.
486 * COLO just concern the response net packet payload from primary guest
487 * and secondary guest are same or not, So we ignored all IP header include
488 * other field like TOS,TTL,IP Checksum. we only need to compare
489 * the ip payload here.
491 if (ppkt->size != spkt->size) {
492 trace_colo_compare_main("UDP: payload size of packets are different");
493 return -1;
495 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
496 ppkt->size - offset)) {
497 trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
498 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
499 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
500 qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
501 ppkt->size);
502 qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
503 spkt->size);
505 return -1;
506 } else {
507 return 0;
512 * Called from the compare thread on the primary
513 * for compare icmp packet
515 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
517 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
518 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
520 trace_colo_compare_main("compare icmp");
523 * Because of ppkt and spkt are both in the same connection,
524 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
525 * same with spkt. In addition, IP header's Identification is a random
526 * field, we can handle it in IP fragmentation function later.
527 * COLO just concern the response net packet payload from primary guest
528 * and secondary guest are same or not, So we ignored all IP header include
529 * other field like TOS,TTL,IP Checksum. we only need to compare
530 * the ip payload here.
532 if (ppkt->size != spkt->size) {
533 trace_colo_compare_main("ICMP: payload size of packets are different");
534 return -1;
536 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
537 ppkt->size - offset)) {
538 trace_colo_compare_icmp_miscompare("primary pkt size",
539 ppkt->size);
540 trace_colo_compare_icmp_miscompare("Secondary pkt size",
541 spkt->size);
542 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
543 qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
544 ppkt->size);
545 qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
546 spkt->size);
548 return -1;
549 } else {
550 return 0;
555 * Called from the compare thread on the primary
556 * for compare other packet
558 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
560 uint16_t offset = ppkt->vnet_hdr_len;
562 trace_colo_compare_main("compare other");
563 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
564 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
566 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
567 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
568 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
569 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
571 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
572 pri_ip_dst, spkt->size,
573 sec_ip_src, sec_ip_dst);
576 if (ppkt->size != spkt->size) {
577 trace_colo_compare_main("Other: payload size of packets are different");
578 return -1;
580 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
581 ppkt->size - offset);
584 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
586 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
588 if ((now - pkt->creation_ms) > (*check_time)) {
589 trace_colo_old_packet_check_found(pkt->creation_ms);
590 return 0;
591 } else {
592 return 1;
596 void colo_compare_register_notifier(Notifier *notify)
598 notifier_list_add(&colo_compare_notifiers, notify);
601 void colo_compare_unregister_notifier(Notifier *notify)
603 notifier_remove(notify);
606 static int colo_old_packet_check_one_conn(Connection *conn,
607 CompareState *s)
609 GList *result = NULL;
610 int64_t check_time = REGULAR_PACKET_CHECK_MS;
612 result = g_queue_find_custom(&conn->primary_list,
613 &check_time,
614 (GCompareFunc)colo_old_packet_check_one);
616 if (result) {
617 /* Do checkpoint will flush old packet */
618 colo_compare_inconsistency_notify(s);
619 return 0;
622 return 1;
626 * Look for old packets that the secondary hasn't matched,
627 * if we have some then we have to checkpoint to wake
628 * the secondary up.
630 static void colo_old_packet_check(void *opaque)
632 CompareState *s = opaque;
635 * If we find one old packet, stop finding job and notify
636 * COLO frame do checkpoint.
638 g_queue_find_custom(&s->conn_list, s,
639 (GCompareFunc)colo_old_packet_check_one_conn);
642 static void colo_compare_packet(CompareState *s, Connection *conn,
643 int (*HandlePacket)(Packet *spkt,
644 Packet *ppkt))
646 Packet *pkt = NULL;
647 GList *result = NULL;
649 while (!g_queue_is_empty(&conn->primary_list) &&
650 !g_queue_is_empty(&conn->secondary_list)) {
651 pkt = g_queue_pop_head(&conn->primary_list);
652 result = g_queue_find_custom(&conn->secondary_list,
653 pkt, (GCompareFunc)HandlePacket);
655 if (result) {
656 colo_release_primary_pkt(s, pkt);
657 g_queue_remove(&conn->secondary_list, result->data);
658 } else {
660 * If one packet arrive late, the secondary_list or
661 * primary_list will be empty, so we can't compare it
662 * until next comparison. If the packets in the list are
663 * timeout, it will trigger a checkpoint request.
665 trace_colo_compare_main("packet different");
666 g_queue_push_head(&conn->primary_list, pkt);
668 colo_compare_inconsistency_notify(s);
669 break;
675 * Called from the compare thread on the primary
676 * for compare packet with secondary list of the
677 * specified connection when a new packet was
678 * queued to it.
680 static void colo_compare_connection(void *opaque, void *user_data)
682 CompareState *s = user_data;
683 Connection *conn = opaque;
685 switch (conn->ip_proto) {
686 case IPPROTO_TCP:
687 colo_compare_tcp(s, conn);
688 break;
689 case IPPROTO_UDP:
690 colo_compare_packet(s, conn, colo_packet_compare_udp);
691 break;
692 case IPPROTO_ICMP:
693 colo_compare_packet(s, conn, colo_packet_compare_icmp);
694 break;
695 default:
696 colo_compare_packet(s, conn, colo_packet_compare_other);
697 break;
701 static int compare_chr_send(CompareState *s,
702 const uint8_t *buf,
703 uint32_t size,
704 uint32_t vnet_hdr_len,
705 bool notify_remote_frame)
707 int ret = 0;
708 uint32_t len = htonl(size);
710 if (!size) {
711 return 0;
714 if (notify_remote_frame) {
715 ret = qemu_chr_fe_write_all(&s->chr_notify_dev,
716 (uint8_t *)&len,
717 sizeof(len));
718 } else {
719 ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)&len, sizeof(len));
722 if (ret != sizeof(len)) {
723 goto err;
726 if (s->vnet_hdr) {
728 * We send vnet header len make other module(like filter-redirector)
729 * know how to parse net packet correctly.
731 len = htonl(vnet_hdr_len);
733 if (!notify_remote_frame) {
734 ret = qemu_chr_fe_write_all(&s->chr_out,
735 (uint8_t *)&len,
736 sizeof(len));
739 if (ret != sizeof(len)) {
740 goto err;
744 if (notify_remote_frame) {
745 ret = qemu_chr_fe_write_all(&s->chr_notify_dev,
746 (uint8_t *)buf,
747 size);
748 } else {
749 ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)buf, size);
752 if (ret != size) {
753 goto err;
756 return 0;
758 err:
759 return ret < 0 ? ret : -EIO;
762 static int compare_chr_can_read(void *opaque)
764 return COMPARE_READ_LEN_MAX;
768 * Called from the main thread on the primary for packets
769 * arriving over the socket from the primary.
771 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
773 CompareState *s = COLO_COMPARE(opaque);
774 int ret;
776 ret = net_fill_rstate(&s->pri_rs, buf, size);
777 if (ret == -1) {
778 qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
779 NULL, NULL, true);
780 error_report("colo-compare primary_in error");
785 * Called from the main thread on the primary for packets
786 * arriving over the socket from the secondary.
788 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
790 CompareState *s = COLO_COMPARE(opaque);
791 int ret;
793 ret = net_fill_rstate(&s->sec_rs, buf, size);
794 if (ret == -1) {
795 qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
796 NULL, NULL, true);
797 error_report("colo-compare secondary_in error");
801 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
803 CompareState *s = COLO_COMPARE(opaque);
804 int ret;
806 ret = net_fill_rstate(&s->notify_rs, buf, size);
807 if (ret == -1) {
808 qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
809 NULL, NULL, true);
810 error_report("colo-compare notify_dev error");
815 * Check old packet regularly so it can watch for any packets
816 * that the secondary hasn't produced equivalents of.
818 static void check_old_packet_regular(void *opaque)
820 CompareState *s = opaque;
822 /* if have old packet we will notify checkpoint */
823 colo_old_packet_check(s);
824 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
825 REGULAR_PACKET_CHECK_MS);
828 /* Public API, Used for COLO frame to notify compare event */
829 void colo_notify_compares_event(void *opaque, int event, Error **errp)
831 CompareState *s;
833 qemu_mutex_lock(&event_mtx);
834 QTAILQ_FOREACH(s, &net_compares, next) {
835 s->event = event;
836 qemu_bh_schedule(s->event_bh);
837 event_unhandled_count++;
839 /* Wait all compare threads to finish handling this event */
840 while (event_unhandled_count > 0) {
841 qemu_cond_wait(&event_complete_cond, &event_mtx);
844 qemu_mutex_unlock(&event_mtx);
847 static void colo_compare_timer_init(CompareState *s)
849 AioContext *ctx = iothread_get_aio_context(s->iothread);
851 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
852 SCALE_MS, check_old_packet_regular,
854 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
855 REGULAR_PACKET_CHECK_MS);
858 static void colo_compare_timer_del(CompareState *s)
860 if (s->packet_check_timer) {
861 timer_del(s->packet_check_timer);
862 timer_free(s->packet_check_timer);
863 s->packet_check_timer = NULL;
867 static void colo_flush_packets(void *opaque, void *user_data);
869 static void colo_compare_handle_event(void *opaque)
871 CompareState *s = opaque;
873 switch (s->event) {
874 case COLO_EVENT_CHECKPOINT:
875 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
876 break;
877 case COLO_EVENT_FAILOVER:
878 break;
879 default:
880 break;
883 qemu_mutex_lock(&event_mtx);
884 assert(event_unhandled_count > 0);
885 event_unhandled_count--;
886 qemu_cond_broadcast(&event_complete_cond);
887 qemu_mutex_unlock(&event_mtx);
890 static void colo_compare_iothread(CompareState *s)
892 object_ref(OBJECT(s->iothread));
893 s->worker_context = iothread_get_g_main_context(s->iothread);
895 qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
896 compare_pri_chr_in, NULL, NULL,
897 s, s->worker_context, true);
898 qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
899 compare_sec_chr_in, NULL, NULL,
900 s, s->worker_context, true);
901 if (s->notify_dev) {
902 qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
903 compare_notify_chr, NULL, NULL,
904 s, s->worker_context, true);
907 colo_compare_timer_init(s);
908 s->event_bh = qemu_bh_new(colo_compare_handle_event, s);
911 static char *compare_get_pri_indev(Object *obj, Error **errp)
913 CompareState *s = COLO_COMPARE(obj);
915 return g_strdup(s->pri_indev);
918 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
920 CompareState *s = COLO_COMPARE(obj);
922 g_free(s->pri_indev);
923 s->pri_indev = g_strdup(value);
926 static char *compare_get_sec_indev(Object *obj, Error **errp)
928 CompareState *s = COLO_COMPARE(obj);
930 return g_strdup(s->sec_indev);
933 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
935 CompareState *s = COLO_COMPARE(obj);
937 g_free(s->sec_indev);
938 s->sec_indev = g_strdup(value);
941 static char *compare_get_outdev(Object *obj, Error **errp)
943 CompareState *s = COLO_COMPARE(obj);
945 return g_strdup(s->outdev);
948 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
950 CompareState *s = COLO_COMPARE(obj);
952 g_free(s->outdev);
953 s->outdev = g_strdup(value);
956 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
958 CompareState *s = COLO_COMPARE(obj);
960 return s->vnet_hdr;
963 static void compare_set_vnet_hdr(Object *obj,
964 bool value,
965 Error **errp)
967 CompareState *s = COLO_COMPARE(obj);
969 s->vnet_hdr = value;
972 static char *compare_get_notify_dev(Object *obj, Error **errp)
974 CompareState *s = COLO_COMPARE(obj);
976 return g_strdup(s->notify_dev);
979 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
981 CompareState *s = COLO_COMPARE(obj);
983 g_free(s->notify_dev);
984 s->notify_dev = g_strdup(value);
987 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
989 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
990 Connection *conn = NULL;
992 if (packet_enqueue(s, PRIMARY_IN, &conn)) {
993 trace_colo_compare_main("primary: unsupported packet in");
994 compare_chr_send(s,
995 pri_rs->buf,
996 pri_rs->packet_len,
997 pri_rs->vnet_hdr_len,
998 false);
999 } else {
1000 /* compare packet in the specified connection */
1001 colo_compare_connection(conn, s);
1005 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1007 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1008 Connection *conn = NULL;
1010 if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1011 trace_colo_compare_main("secondary: unsupported packet in");
1012 } else {
1013 /* compare packet in the specified connection */
1014 colo_compare_connection(conn, s);
1018 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1020 CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1022 const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1023 int ret;
1025 if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1026 notify_rs->buf,
1027 notify_rs->packet_len)) {
1028 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true);
1029 if (ret < 0) {
1030 error_report("Notify Xen COLO-frame INIT failed");
1032 } else if (packet_matches_str("COLO_CHECKPOINT",
1033 notify_rs->buf,
1034 notify_rs->packet_len)) {
1035 /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1036 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1037 } else {
1038 error_report("COLO compare got unsupported instruction");
1043 * Return 0 is success.
1044 * Return 1 is failed.
1046 static int find_and_check_chardev(Chardev **chr,
1047 char *chr_name,
1048 Error **errp)
1050 *chr = qemu_chr_find(chr_name);
1051 if (*chr == NULL) {
1052 error_setg(errp, "Device '%s' not found",
1053 chr_name);
1054 return 1;
1057 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1058 error_setg(errp, "chardev \"%s\" is not reconnectable",
1059 chr_name);
1060 return 1;
1063 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1064 error_setg(errp, "chardev \"%s\" cannot switch context",
1065 chr_name);
1066 return 1;
1069 return 0;
1073 * Called from the main thread on the primary
1074 * to setup colo-compare.
1076 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1078 CompareState *s = COLO_COMPARE(uc);
1079 Chardev *chr;
1081 if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1082 error_setg(errp, "colo compare needs 'primary_in' ,"
1083 "'secondary_in','outdev','iothread' property set");
1084 return;
1085 } else if (!strcmp(s->pri_indev, s->outdev) ||
1086 !strcmp(s->sec_indev, s->outdev) ||
1087 !strcmp(s->pri_indev, s->sec_indev)) {
1088 error_setg(errp, "'indev' and 'outdev' could not be same "
1089 "for compare module");
1090 return;
1093 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1094 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1095 return;
1098 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1099 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1100 return;
1103 if (find_and_check_chardev(&chr, s->outdev, errp) ||
1104 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1105 return;
1108 net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1109 net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1111 /* Try to enable remote notify chardev, currently just for Xen COLO */
1112 if (s->notify_dev) {
1113 if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1114 !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1115 return;
1118 net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1119 s->vnet_hdr);
1122 QTAILQ_INSERT_TAIL(&net_compares, s, next);
1124 g_queue_init(&s->conn_list);
1126 qemu_mutex_init(&event_mtx);
1127 qemu_cond_init(&event_complete_cond);
1129 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1130 connection_key_equal,
1131 g_free,
1132 connection_destroy);
1134 colo_compare_iothread(s);
1135 return;
1138 static void colo_flush_packets(void *opaque, void *user_data)
1140 CompareState *s = user_data;
1141 Connection *conn = opaque;
1142 Packet *pkt = NULL;
1144 while (!g_queue_is_empty(&conn->primary_list)) {
1145 pkt = g_queue_pop_head(&conn->primary_list);
1146 compare_chr_send(s,
1147 pkt->data,
1148 pkt->size,
1149 pkt->vnet_hdr_len,
1150 false);
1151 packet_destroy(pkt, NULL);
1153 while (!g_queue_is_empty(&conn->secondary_list)) {
1154 pkt = g_queue_pop_head(&conn->secondary_list);
1155 packet_destroy(pkt, NULL);
1159 static void colo_compare_class_init(ObjectClass *oc, void *data)
1161 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1163 ucc->complete = colo_compare_complete;
1166 static void colo_compare_init(Object *obj)
1168 CompareState *s = COLO_COMPARE(obj);
1170 object_property_add_str(obj, "primary_in",
1171 compare_get_pri_indev, compare_set_pri_indev,
1172 NULL);
1173 object_property_add_str(obj, "secondary_in",
1174 compare_get_sec_indev, compare_set_sec_indev,
1175 NULL);
1176 object_property_add_str(obj, "outdev",
1177 compare_get_outdev, compare_set_outdev,
1178 NULL);
1179 object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1180 (Object **)&s->iothread,
1181 object_property_allow_set_link,
1182 OBJ_PROP_LINK_STRONG, NULL);
1183 /* This parameter just for Xen COLO */
1184 object_property_add_str(obj, "notify_dev",
1185 compare_get_notify_dev, compare_set_notify_dev,
1186 NULL);
1188 s->vnet_hdr = false;
1189 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1190 compare_set_vnet_hdr, NULL);
1193 static void colo_compare_finalize(Object *obj)
1195 CompareState *s = COLO_COMPARE(obj);
1196 CompareState *tmp = NULL;
1198 qemu_chr_fe_deinit(&s->chr_pri_in, false);
1199 qemu_chr_fe_deinit(&s->chr_sec_in, false);
1200 qemu_chr_fe_deinit(&s->chr_out, false);
1201 if (s->notify_dev) {
1202 qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1205 if (s->iothread) {
1206 colo_compare_timer_del(s);
1209 qemu_bh_delete(s->event_bh);
1211 QTAILQ_FOREACH(tmp, &net_compares, next) {
1212 if (tmp == s) {
1213 QTAILQ_REMOVE(&net_compares, s, next);
1214 break;
1218 /* Release all unhandled packets after compare thead exited */
1219 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1221 g_queue_clear(&s->conn_list);
1223 if (s->connection_track_table) {
1224 g_hash_table_destroy(s->connection_track_table);
1227 if (s->iothread) {
1228 object_unref(OBJECT(s->iothread));
1231 qemu_mutex_destroy(&event_mtx);
1232 qemu_cond_destroy(&event_complete_cond);
1234 g_free(s->pri_indev);
1235 g_free(s->sec_indev);
1236 g_free(s->outdev);
1237 g_free(s->notify_dev);
1240 static const TypeInfo colo_compare_info = {
1241 .name = TYPE_COLO_COMPARE,
1242 .parent = TYPE_OBJECT,
1243 .instance_size = sizeof(CompareState),
1244 .instance_init = colo_compare_init,
1245 .instance_finalize = colo_compare_finalize,
1246 .class_size = sizeof(CompareClass),
1247 .class_init = colo_compare_class_init,
1248 .interfaces = (InterfaceInfo[]) {
1249 { TYPE_USER_CREATABLE },
1254 static void register_types(void)
1256 type_register_static(&colo_compare_info);
1259 type_init(register_types);