hw/misc: Add Microchip PolarFire SoC DDR Memory Controller support
[qemu/ar7.git] / net / colo-compare.c
blob3a45d64175f00543f307a5234ea2b50487f50c97
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 #include "block/aio-wait.h"
36 #include "qemu/coroutine.h"
38 #define TYPE_COLO_COMPARE "colo-compare"
39 typedef struct CompareState CompareState;
40 DECLARE_INSTANCE_CHECKER(CompareState, COLO_COMPARE,
41 TYPE_COLO_COMPARE)
43 static QTAILQ_HEAD(, CompareState) net_compares =
44 QTAILQ_HEAD_INITIALIZER(net_compares);
46 static NotifierList colo_compare_notifiers =
47 NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
49 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
50 #define MAX_QUEUE_SIZE 1024
52 #define COLO_COMPARE_FREE_PRIMARY 0x01
53 #define COLO_COMPARE_FREE_SECONDARY 0x02
55 #define REGULAR_PACKET_CHECK_MS 3000
56 #define DEFAULT_TIME_OUT_MS 3000
58 /* #define DEBUG_COLO_PACKETS */
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;
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 * +--------+ +--------+ +--------+ +--------+
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;
100 typedef struct SendEntry {
101 uint32_t size;
102 uint32_t vnet_hdr_len;
103 uint8_t *buf;
104 } SendEntry;
106 struct CompareState {
107 Object parent;
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 uint32_t compare_timeout;
124 uint32_t expired_scan_cycle;
127 * Record the connection that through the NIC
128 * Element type: Connection
130 GQueue conn_list;
131 /* Record the connection without repetition */
132 GHashTable *connection_track_table;
134 IOThread *iothread;
135 GMainContext *worker_context;
136 QEMUTimer *packet_check_timer;
138 QEMUBH *event_bh;
139 enum colo_event event;
141 QTAILQ_ENTRY(CompareState) next;
144 typedef struct CompareClass {
145 ObjectClass parent_class;
146 } CompareClass;
148 enum {
149 PRIMARY_IN = 0,
150 SECONDARY_IN,
153 static const char *colo_mode[] = {
154 [PRIMARY_IN] = "primary",
155 [SECONDARY_IN] = "secondary",
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);
165 static bool packet_matches_str(const char *str,
166 const uint8_t *buf,
167 uint32_t packet_len)
169 if (packet_len != strlen(str)) {
170 return false;
173 return !memcmp(str, buf, strlen(str));
176 static void notify_remote_frame(CompareState *s)
178 char msg[] = "DO_CHECKPOINT";
179 int ret = 0;
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");
187 static void colo_compare_inconsistency_notify(CompareState *s)
189 if (s->notify_dev) {
190 notify_remote_frame(s);
191 } else {
192 notifier_list_notify(&colo_compare_notifiers,
193 migrate_get_current());
197 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
199 struct tcp_hdr *atcp, *btcp;
201 atcp = (struct tcp_hdr *)(a->transport_header);
202 btcp = (struct tcp_hdr *)(b->transport_header);
203 return ntohl(atcp->th_seq) - ntohl(btcp->th_seq);
206 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
208 Packet *pkt = data;
209 struct tcp_hdr *tcphd;
211 tcphd = (struct tcp_hdr *)pkt->transport_header;
213 pkt->tcp_seq = ntohl(tcphd->th_seq);
214 pkt->tcp_ack = ntohl(tcphd->th_ack);
215 *max_ack = *max_ack > pkt->tcp_ack ? *max_ack : pkt->tcp_ack;
216 pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
217 + (tcphd->th_off << 2) - pkt->vnet_hdr_len;
218 pkt->payload_size = pkt->size - pkt->header_size;
219 pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
220 pkt->flags = tcphd->th_flags;
224 * Return 1 on success, if return 0 means the
225 * packet will be dropped
227 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
229 if (g_queue_get_length(queue) <= max_queue_size) {
230 if (pkt->ip->ip_p == IPPROTO_TCP) {
231 fill_pkt_tcp_info(pkt, max_ack);
232 g_queue_insert_sorted(queue,
233 pkt,
234 (GCompareDataFunc)seq_sorter,
235 NULL);
236 } else {
237 g_queue_push_tail(queue, pkt);
239 return 1;
241 return 0;
245 * Return 0 on success, if return -1 means the pkt
246 * is unsupported(arp and ipv6) and will be sent later
248 static int packet_enqueue(CompareState *s, int mode, Connection **con)
250 ConnectionKey key;
251 Packet *pkt = NULL;
252 Connection *conn;
253 int ret;
255 if (mode == PRIMARY_IN) {
256 pkt = packet_new(s->pri_rs.buf,
257 s->pri_rs.packet_len,
258 s->pri_rs.vnet_hdr_len);
259 } else {
260 pkt = packet_new(s->sec_rs.buf,
261 s->sec_rs.packet_len,
262 s->sec_rs.vnet_hdr_len);
265 if (parse_packet_early(pkt)) {
266 packet_destroy(pkt, NULL);
267 pkt = NULL;
268 return -1;
270 fill_connection_key(pkt, &key);
272 conn = connection_get(s->connection_track_table,
273 &key,
274 &s->conn_list);
276 if (!conn->processing) {
277 g_queue_push_tail(&s->conn_list, conn);
278 conn->processing = true;
281 if (mode == PRIMARY_IN) {
282 ret = colo_insert_packet(&conn->primary_list, pkt, &conn->pack);
283 } else {
284 ret = colo_insert_packet(&conn->secondary_list, pkt, &conn->sack);
287 if (!ret) {
288 trace_colo_compare_drop_packet(colo_mode[mode],
289 "queue size too big, drop packet");
290 packet_destroy(pkt, NULL);
291 pkt = NULL;
294 *con = conn;
296 return 0;
299 static inline bool after(uint32_t seq1, uint32_t seq2)
301 return (int32_t)(seq1 - seq2) > 0;
304 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
306 int ret;
307 ret = compare_chr_send(s,
308 pkt->data,
309 pkt->size,
310 pkt->vnet_hdr_len,
311 false,
312 true);
313 if (ret < 0) {
314 error_report("colo send primary packet failed");
316 trace_colo_compare_main("packet same and release packet");
317 packet_destroy_partial(pkt, NULL);
321 * The IP packets sent by primary and secondary
322 * will be compared in here
323 * TODO support ip fragment, Out-Of-Order
324 * return: 0 means packet same
325 * > 0 || < 0 means packet different
327 static int colo_compare_packet_payload(Packet *ppkt,
328 Packet *spkt,
329 uint16_t poffset,
330 uint16_t soffset,
331 uint16_t len)
334 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
335 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
337 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
338 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
339 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
340 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
342 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
343 pri_ip_dst, spkt->size,
344 sec_ip_src, sec_ip_dst);
347 return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
351 * return true means that the payload is consist and
352 * need to make the next comparison, false means do
353 * the checkpoint
355 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
356 int8_t *mark, uint32_t max_ack)
358 *mark = 0;
360 if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
361 if (!colo_compare_packet_payload(ppkt, spkt,
362 ppkt->header_size, spkt->header_size,
363 ppkt->payload_size)) {
364 *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
365 return true;
369 /* one part of secondary packet payload still need to be compared */
370 if (!after(ppkt->seq_end, spkt->seq_end)) {
371 if (!colo_compare_packet_payload(ppkt, spkt,
372 ppkt->header_size + ppkt->offset,
373 spkt->header_size + spkt->offset,
374 ppkt->payload_size - ppkt->offset)) {
375 if (!after(ppkt->tcp_ack, max_ack)) {
376 *mark = COLO_COMPARE_FREE_PRIMARY;
377 spkt->offset += ppkt->payload_size - ppkt->offset;
378 return true;
379 } else {
380 /* secondary guest hasn't ack the data, don't send
381 * out this packet
383 return false;
386 } else {
387 /* primary packet is longer than secondary packet, compare
388 * the same part and mark the primary packet offset
390 if (!colo_compare_packet_payload(ppkt, spkt,
391 ppkt->header_size + ppkt->offset,
392 spkt->header_size + spkt->offset,
393 spkt->payload_size - spkt->offset)) {
394 *mark = COLO_COMPARE_FREE_SECONDARY;
395 ppkt->offset += spkt->payload_size - spkt->offset;
396 return true;
400 return false;
403 static void colo_compare_tcp(CompareState *s, Connection *conn)
405 Packet *ppkt = NULL, *spkt = NULL;
406 int8_t mark;
409 * If ppkt and spkt have the same payload, but ppkt's ACK
410 * is greater than spkt's ACK, in this case we can not
411 * send the ppkt because it will cause the secondary guest
412 * to miss sending some data in the next. Therefore, we
413 * record the maximum ACK in the current queue at both
414 * primary side and secondary side. Only when the ack is
415 * less than the smaller of the two maximum ack, then we
416 * can ensure that the packet's payload is acknowledged by
417 * primary and secondary.
419 uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
421 pri:
422 if (g_queue_is_empty(&conn->primary_list)) {
423 return;
425 ppkt = g_queue_pop_head(&conn->primary_list);
426 sec:
427 if (g_queue_is_empty(&conn->secondary_list)) {
428 g_queue_push_head(&conn->primary_list, ppkt);
429 return;
431 spkt = g_queue_pop_head(&conn->secondary_list);
433 if (ppkt->tcp_seq == ppkt->seq_end) {
434 colo_release_primary_pkt(s, ppkt);
435 ppkt = NULL;
438 if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
439 trace_colo_compare_main("pri: this packet has compared");
440 colo_release_primary_pkt(s, ppkt);
441 ppkt = NULL;
444 if (spkt->tcp_seq == spkt->seq_end) {
445 packet_destroy(spkt, NULL);
446 if (!ppkt) {
447 goto pri;
448 } else {
449 goto sec;
451 } else {
452 if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
453 trace_colo_compare_main("sec: this packet has compared");
454 packet_destroy(spkt, NULL);
455 if (!ppkt) {
456 goto pri;
457 } else {
458 goto sec;
461 if (!ppkt) {
462 g_queue_push_head(&conn->secondary_list, spkt);
463 goto pri;
467 if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
468 trace_colo_compare_tcp_info("pri",
469 ppkt->tcp_seq, ppkt->tcp_ack,
470 ppkt->header_size, ppkt->payload_size,
471 ppkt->offset, ppkt->flags);
473 trace_colo_compare_tcp_info("sec",
474 spkt->tcp_seq, spkt->tcp_ack,
475 spkt->header_size, spkt->payload_size,
476 spkt->offset, spkt->flags);
478 if (mark == COLO_COMPARE_FREE_PRIMARY) {
479 conn->compare_seq = ppkt->seq_end;
480 colo_release_primary_pkt(s, ppkt);
481 g_queue_push_head(&conn->secondary_list, spkt);
482 goto pri;
484 if (mark == COLO_COMPARE_FREE_SECONDARY) {
485 conn->compare_seq = spkt->seq_end;
486 packet_destroy(spkt, NULL);
487 goto sec;
489 if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
490 conn->compare_seq = ppkt->seq_end;
491 colo_release_primary_pkt(s, ppkt);
492 packet_destroy(spkt, NULL);
493 goto pri;
495 } else {
496 g_queue_push_head(&conn->primary_list, ppkt);
497 g_queue_push_head(&conn->secondary_list, spkt);
499 #ifdef DEBUG_COLO_PACKETS
500 qemu_hexdump(stderr, "colo-compare ppkt", ppkt->data, ppkt->size);
501 qemu_hexdump(stderr, "colo-compare spkt", spkt->data, spkt->size);
502 #endif
504 colo_compare_inconsistency_notify(s);
510 * Called from the compare thread on the primary
511 * for compare udp packet
513 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
515 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
516 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
518 trace_colo_compare_main("compare udp");
521 * Because of ppkt and spkt are both in the same connection,
522 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
523 * same with spkt. In addition, IP header's Identification is a random
524 * field, we can handle it in IP fragmentation function later.
525 * COLO just concern the response net packet payload from primary guest
526 * and secondary guest are same or not, So we ignored all IP header include
527 * other field like TOS,TTL,IP Checksum. we only need to compare
528 * the ip payload here.
530 if (ppkt->size != spkt->size) {
531 trace_colo_compare_main("UDP: payload size of packets are different");
532 return -1;
534 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
535 ppkt->size - offset)) {
536 trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
537 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
538 #ifdef DEBUG_COLO_PACKETS
539 qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
540 qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
541 #endif
542 return -1;
543 } else {
544 return 0;
549 * Called from the compare thread on the primary
550 * for compare icmp packet
552 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
554 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
555 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
557 trace_colo_compare_main("compare icmp");
560 * Because of ppkt and spkt are both in the same connection,
561 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
562 * same with spkt. In addition, IP header's Identification is a random
563 * field, we can handle it in IP fragmentation function later.
564 * COLO just concern the response net packet payload from primary guest
565 * and secondary guest are same or not, So we ignored all IP header include
566 * other field like TOS,TTL,IP Checksum. we only need to compare
567 * the ip payload here.
569 if (ppkt->size != spkt->size) {
570 trace_colo_compare_main("ICMP: payload size of packets are different");
571 return -1;
573 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
574 ppkt->size - offset)) {
575 trace_colo_compare_icmp_miscompare("primary pkt size",
576 ppkt->size);
577 trace_colo_compare_icmp_miscompare("Secondary pkt size",
578 spkt->size);
579 #ifdef DEBUG_COLO_PACKETS
580 qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
581 qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
582 #endif
583 return -1;
584 } else {
585 return 0;
590 * Called from the compare thread on the primary
591 * for compare other packet
593 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
595 uint16_t offset = ppkt->vnet_hdr_len;
597 trace_colo_compare_main("compare other");
598 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
599 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
601 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
602 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
603 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
604 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
606 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
607 pri_ip_dst, spkt->size,
608 sec_ip_src, sec_ip_dst);
611 if (ppkt->size != spkt->size) {
612 trace_colo_compare_main("Other: payload size of packets are different");
613 return -1;
615 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
616 ppkt->size - offset);
619 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
621 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
623 if ((now - pkt->creation_ms) > (*check_time)) {
624 trace_colo_old_packet_check_found(pkt->creation_ms);
625 return 0;
626 } else {
627 return 1;
631 void colo_compare_register_notifier(Notifier *notify)
633 notifier_list_add(&colo_compare_notifiers, notify);
636 void colo_compare_unregister_notifier(Notifier *notify)
638 notifier_remove(notify);
641 static int colo_old_packet_check_one_conn(Connection *conn,
642 CompareState *s)
644 GList *result = NULL;
646 result = g_queue_find_custom(&conn->primary_list,
647 &s->compare_timeout,
648 (GCompareFunc)colo_old_packet_check_one);
650 if (result) {
651 /* Do checkpoint will flush old packet */
652 colo_compare_inconsistency_notify(s);
653 return 0;
656 return 1;
660 * Look for old packets that the secondary hasn't matched,
661 * if we have some then we have to checkpoint to wake
662 * the secondary up.
664 static void colo_old_packet_check(void *opaque)
666 CompareState *s = opaque;
669 * If we find one old packet, stop finding job and notify
670 * COLO frame do checkpoint.
672 g_queue_find_custom(&s->conn_list, s,
673 (GCompareFunc)colo_old_packet_check_one_conn);
676 static void colo_compare_packet(CompareState *s, Connection *conn,
677 int (*HandlePacket)(Packet *spkt,
678 Packet *ppkt))
680 Packet *pkt = NULL;
681 GList *result = NULL;
683 while (!g_queue_is_empty(&conn->primary_list) &&
684 !g_queue_is_empty(&conn->secondary_list)) {
685 pkt = g_queue_pop_head(&conn->primary_list);
686 result = g_queue_find_custom(&conn->secondary_list,
687 pkt, (GCompareFunc)HandlePacket);
689 if (result) {
690 colo_release_primary_pkt(s, pkt);
691 g_queue_remove(&conn->secondary_list, result->data);
692 } else {
694 * If one packet arrive late, the secondary_list or
695 * primary_list will be empty, so we can't compare it
696 * until next comparison. If the packets in the list are
697 * timeout, it will trigger a checkpoint request.
699 trace_colo_compare_main("packet different");
700 g_queue_push_head(&conn->primary_list, pkt);
702 colo_compare_inconsistency_notify(s);
703 break;
709 * Called from the compare thread on the primary
710 * for compare packet with secondary list of the
711 * specified connection when a new packet was
712 * queued to it.
714 static void colo_compare_connection(void *opaque, void *user_data)
716 CompareState *s = user_data;
717 Connection *conn = opaque;
719 switch (conn->ip_proto) {
720 case IPPROTO_TCP:
721 colo_compare_tcp(s, conn);
722 break;
723 case IPPROTO_UDP:
724 colo_compare_packet(s, conn, colo_packet_compare_udp);
725 break;
726 case IPPROTO_ICMP:
727 colo_compare_packet(s, conn, colo_packet_compare_icmp);
728 break;
729 default:
730 colo_compare_packet(s, conn, colo_packet_compare_other);
731 break;
735 static void coroutine_fn _compare_chr_send(void *opaque)
737 SendCo *sendco = opaque;
738 CompareState *s = sendco->s;
739 int ret = 0;
741 while (!g_queue_is_empty(&sendco->send_list)) {
742 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
743 uint32_t len = htonl(entry->size);
745 ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
747 if (ret != sizeof(len)) {
748 g_free(entry->buf);
749 g_slice_free(SendEntry, entry);
750 goto err;
753 if (!sendco->notify_remote_frame && s->vnet_hdr) {
755 * We send vnet header len make other module(like filter-redirector)
756 * know how to parse net packet correctly.
758 len = htonl(entry->vnet_hdr_len);
760 ret = qemu_chr_fe_write_all(sendco->chr,
761 (uint8_t *)&len,
762 sizeof(len));
764 if (ret != sizeof(len)) {
765 g_free(entry->buf);
766 g_slice_free(SendEntry, entry);
767 goto err;
771 ret = qemu_chr_fe_write_all(sendco->chr,
772 (uint8_t *)entry->buf,
773 entry->size);
775 if (ret != entry->size) {
776 g_free(entry->buf);
777 g_slice_free(SendEntry, entry);
778 goto err;
781 g_free(entry->buf);
782 g_slice_free(SendEntry, entry);
785 sendco->ret = 0;
786 goto out;
788 err:
789 while (!g_queue_is_empty(&sendco->send_list)) {
790 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
791 g_free(entry->buf);
792 g_slice_free(SendEntry, entry);
794 sendco->ret = ret < 0 ? ret : -EIO;
795 out:
796 sendco->co = NULL;
797 sendco->done = true;
798 aio_wait_kick();
801 static int compare_chr_send(CompareState *s,
802 uint8_t *buf,
803 uint32_t size,
804 uint32_t vnet_hdr_len,
805 bool notify_remote_frame,
806 bool zero_copy)
808 SendCo *sendco;
809 SendEntry *entry;
811 if (notify_remote_frame) {
812 sendco = &s->notify_sendco;
813 } else {
814 sendco = &s->out_sendco;
817 if (!size) {
818 return 0;
821 entry = g_slice_new(SendEntry);
822 entry->size = size;
823 entry->vnet_hdr_len = vnet_hdr_len;
824 if (zero_copy) {
825 entry->buf = buf;
826 } else {
827 entry->buf = g_malloc(size);
828 memcpy(entry->buf, buf, size);
830 g_queue_push_head(&sendco->send_list, entry);
832 if (sendco->done) {
833 sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
834 sendco->done = false;
835 qemu_coroutine_enter(sendco->co);
836 if (sendco->done) {
837 /* report early errors */
838 return sendco->ret;
842 /* assume success */
843 return 0;
846 static int compare_chr_can_read(void *opaque)
848 return COMPARE_READ_LEN_MAX;
852 * Called from the main thread on the primary for packets
853 * arriving over the socket from the primary.
855 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
857 CompareState *s = COLO_COMPARE(opaque);
858 int ret;
860 ret = net_fill_rstate(&s->pri_rs, buf, size);
861 if (ret == -1) {
862 qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
863 NULL, NULL, true);
864 error_report("colo-compare primary_in error");
869 * Called from the main thread on the primary for packets
870 * arriving over the socket from the secondary.
872 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
874 CompareState *s = COLO_COMPARE(opaque);
875 int ret;
877 ret = net_fill_rstate(&s->sec_rs, buf, size);
878 if (ret == -1) {
879 qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
880 NULL, NULL, true);
881 error_report("colo-compare secondary_in error");
885 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
887 CompareState *s = COLO_COMPARE(opaque);
888 int ret;
890 ret = net_fill_rstate(&s->notify_rs, buf, size);
891 if (ret == -1) {
892 qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
893 NULL, NULL, true);
894 error_report("colo-compare notify_dev error");
899 * Check old packet regularly so it can watch for any packets
900 * that the secondary hasn't produced equivalents of.
902 static void check_old_packet_regular(void *opaque)
904 CompareState *s = opaque;
906 /* if have old packet we will notify checkpoint */
907 colo_old_packet_check(s);
908 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
909 s->expired_scan_cycle);
912 /* Public API, Used for COLO frame to notify compare event */
913 void colo_notify_compares_event(void *opaque, int event, Error **errp)
915 CompareState *s;
916 qemu_mutex_lock(&colo_compare_mutex);
918 if (!colo_compare_active) {
919 qemu_mutex_unlock(&colo_compare_mutex);
920 return;
923 qemu_mutex_lock(&event_mtx);
924 QTAILQ_FOREACH(s, &net_compares, next) {
925 s->event = event;
926 qemu_bh_schedule(s->event_bh);
927 event_unhandled_count++;
929 /* Wait all compare threads to finish handling this event */
930 while (event_unhandled_count > 0) {
931 qemu_cond_wait(&event_complete_cond, &event_mtx);
934 qemu_mutex_unlock(&event_mtx);
935 qemu_mutex_unlock(&colo_compare_mutex);
938 static void colo_compare_timer_init(CompareState *s)
940 AioContext *ctx = iothread_get_aio_context(s->iothread);
942 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
943 SCALE_MS, check_old_packet_regular,
945 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
946 s->expired_scan_cycle);
949 static void colo_compare_timer_del(CompareState *s)
951 if (s->packet_check_timer) {
952 timer_del(s->packet_check_timer);
953 timer_free(s->packet_check_timer);
954 s->packet_check_timer = NULL;
958 static void colo_flush_packets(void *opaque, void *user_data);
960 static void colo_compare_handle_event(void *opaque)
962 CompareState *s = opaque;
964 switch (s->event) {
965 case COLO_EVENT_CHECKPOINT:
966 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
967 break;
968 case COLO_EVENT_FAILOVER:
969 break;
970 default:
971 break;
974 qemu_mutex_lock(&event_mtx);
975 assert(event_unhandled_count > 0);
976 event_unhandled_count--;
977 qemu_cond_broadcast(&event_complete_cond);
978 qemu_mutex_unlock(&event_mtx);
981 static void colo_compare_iothread(CompareState *s)
983 AioContext *ctx = iothread_get_aio_context(s->iothread);
984 object_ref(OBJECT(s->iothread));
985 s->worker_context = iothread_get_g_main_context(s->iothread);
987 qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
988 compare_pri_chr_in, NULL, NULL,
989 s, s->worker_context, true);
990 qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
991 compare_sec_chr_in, NULL, NULL,
992 s, s->worker_context, true);
993 if (s->notify_dev) {
994 qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
995 compare_notify_chr, NULL, NULL,
996 s, s->worker_context, true);
999 colo_compare_timer_init(s);
1000 s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
1003 static char *compare_get_pri_indev(Object *obj, Error **errp)
1005 CompareState *s = COLO_COMPARE(obj);
1007 return g_strdup(s->pri_indev);
1010 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1012 CompareState *s = COLO_COMPARE(obj);
1014 g_free(s->pri_indev);
1015 s->pri_indev = g_strdup(value);
1018 static char *compare_get_sec_indev(Object *obj, Error **errp)
1020 CompareState *s = COLO_COMPARE(obj);
1022 return g_strdup(s->sec_indev);
1025 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1027 CompareState *s = COLO_COMPARE(obj);
1029 g_free(s->sec_indev);
1030 s->sec_indev = g_strdup(value);
1033 static char *compare_get_outdev(Object *obj, Error **errp)
1035 CompareState *s = COLO_COMPARE(obj);
1037 return g_strdup(s->outdev);
1040 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1042 CompareState *s = COLO_COMPARE(obj);
1044 g_free(s->outdev);
1045 s->outdev = g_strdup(value);
1048 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1050 CompareState *s = COLO_COMPARE(obj);
1052 return s->vnet_hdr;
1055 static void compare_set_vnet_hdr(Object *obj,
1056 bool value,
1057 Error **errp)
1059 CompareState *s = COLO_COMPARE(obj);
1061 s->vnet_hdr = value;
1064 static char *compare_get_notify_dev(Object *obj, Error **errp)
1066 CompareState *s = COLO_COMPARE(obj);
1068 return g_strdup(s->notify_dev);
1071 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1073 CompareState *s = COLO_COMPARE(obj);
1075 g_free(s->notify_dev);
1076 s->notify_dev = g_strdup(value);
1079 static void compare_get_timeout(Object *obj, Visitor *v,
1080 const char *name, void *opaque,
1081 Error **errp)
1083 CompareState *s = COLO_COMPARE(obj);
1084 uint32_t value = s->compare_timeout;
1086 visit_type_uint32(v, name, &value, errp);
1089 static void compare_set_timeout(Object *obj, Visitor *v,
1090 const char *name, void *opaque,
1091 Error **errp)
1093 CompareState *s = COLO_COMPARE(obj);
1094 uint32_t value;
1096 if (!visit_type_uint32(v, name, &value, errp)) {
1097 return;
1099 if (!value) {
1100 error_setg(errp, "Property '%s.%s' requires a positive value",
1101 object_get_typename(obj), name);
1102 return;
1104 s->compare_timeout = value;
1107 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1108 const char *name, void *opaque,
1109 Error **errp)
1111 CompareState *s = COLO_COMPARE(obj);
1112 uint32_t value = s->expired_scan_cycle;
1114 visit_type_uint32(v, name, &value, errp);
1117 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1118 const char *name, void *opaque,
1119 Error **errp)
1121 CompareState *s = COLO_COMPARE(obj);
1122 uint32_t value;
1124 if (!visit_type_uint32(v, name, &value, errp)) {
1125 return;
1127 if (!value) {
1128 error_setg(errp, "Property '%s.%s' requires a positive value",
1129 object_get_typename(obj), name);
1130 return;
1132 s->expired_scan_cycle = value;
1135 static void get_max_queue_size(Object *obj, Visitor *v,
1136 const char *name, void *opaque,
1137 Error **errp)
1139 uint32_t value = max_queue_size;
1141 visit_type_uint32(v, name, &value, errp);
1144 static void set_max_queue_size(Object *obj, Visitor *v,
1145 const char *name, void *opaque,
1146 Error **errp)
1148 Error *local_err = NULL;
1149 uint32_t value;
1151 visit_type_uint32(v, name, &value, &local_err);
1152 if (local_err) {
1153 goto out;
1155 if (!value) {
1156 error_setg(&local_err, "Property '%s.%s' requires a positive value",
1157 object_get_typename(obj), name);
1158 goto out;
1160 max_queue_size = value;
1162 out:
1163 error_propagate(errp, local_err);
1166 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1168 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1169 Connection *conn = NULL;
1171 if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1172 trace_colo_compare_main("primary: unsupported packet in");
1173 compare_chr_send(s,
1174 pri_rs->buf,
1175 pri_rs->packet_len,
1176 pri_rs->vnet_hdr_len,
1177 false,
1178 false);
1179 } else {
1180 /* compare packet in the specified connection */
1181 colo_compare_connection(conn, s);
1185 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1187 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1188 Connection *conn = NULL;
1190 if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1191 trace_colo_compare_main("secondary: unsupported packet in");
1192 } else {
1193 /* compare packet in the specified connection */
1194 colo_compare_connection(conn, s);
1198 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1200 CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1202 const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1203 int ret;
1205 if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1206 notify_rs->buf,
1207 notify_rs->packet_len)) {
1208 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1209 if (ret < 0) {
1210 error_report("Notify Xen COLO-frame INIT failed");
1212 } else if (packet_matches_str("COLO_CHECKPOINT",
1213 notify_rs->buf,
1214 notify_rs->packet_len)) {
1215 /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1216 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1217 } else {
1218 error_report("COLO compare got unsupported instruction");
1223 * Return 0 is success.
1224 * Return 1 is failed.
1226 static int find_and_check_chardev(Chardev **chr,
1227 char *chr_name,
1228 Error **errp)
1230 *chr = qemu_chr_find(chr_name);
1231 if (*chr == NULL) {
1232 error_setg(errp, "Device '%s' not found",
1233 chr_name);
1234 return 1;
1237 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1238 error_setg(errp, "chardev \"%s\" is not reconnectable",
1239 chr_name);
1240 return 1;
1243 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1244 error_setg(errp, "chardev \"%s\" cannot switch context",
1245 chr_name);
1246 return 1;
1249 return 0;
1253 * Called from the main thread on the primary
1254 * to setup colo-compare.
1256 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1258 CompareState *s = COLO_COMPARE(uc);
1259 Chardev *chr;
1261 if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1262 error_setg(errp, "colo compare needs 'primary_in' ,"
1263 "'secondary_in','outdev','iothread' property set");
1264 return;
1265 } else if (!strcmp(s->pri_indev, s->outdev) ||
1266 !strcmp(s->sec_indev, s->outdev) ||
1267 !strcmp(s->pri_indev, s->sec_indev)) {
1268 error_setg(errp, "'indev' and 'outdev' could not be same "
1269 "for compare module");
1270 return;
1273 if (!s->compare_timeout) {
1274 /* Set default value to 3000 MS */
1275 s->compare_timeout = DEFAULT_TIME_OUT_MS;
1278 if (!s->expired_scan_cycle) {
1279 /* Set default value to 3000 MS */
1280 s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1283 if (!max_queue_size) {
1284 /* Set default queue size to 1024 */
1285 max_queue_size = MAX_QUEUE_SIZE;
1288 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1289 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1290 return;
1293 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1294 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1295 return;
1298 if (find_and_check_chardev(&chr, s->outdev, errp) ||
1299 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1300 return;
1303 net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1304 net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1306 /* Try to enable remote notify chardev, currently just for Xen COLO */
1307 if (s->notify_dev) {
1308 if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1309 !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1310 return;
1313 net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1314 s->vnet_hdr);
1317 s->out_sendco.s = s;
1318 s->out_sendco.chr = &s->chr_out;
1319 s->out_sendco.notify_remote_frame = false;
1320 s->out_sendco.done = true;
1321 g_queue_init(&s->out_sendco.send_list);
1323 if (s->notify_dev) {
1324 s->notify_sendco.s = s;
1325 s->notify_sendco.chr = &s->chr_notify_dev;
1326 s->notify_sendco.notify_remote_frame = true;
1327 s->notify_sendco.done = true;
1328 g_queue_init(&s->notify_sendco.send_list);
1331 g_queue_init(&s->conn_list);
1333 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1334 connection_key_equal,
1335 g_free,
1336 connection_destroy);
1338 colo_compare_iothread(s);
1340 qemu_mutex_lock(&colo_compare_mutex);
1341 if (!colo_compare_active) {
1342 qemu_mutex_init(&event_mtx);
1343 qemu_cond_init(&event_complete_cond);
1344 colo_compare_active = true;
1346 QTAILQ_INSERT_TAIL(&net_compares, s, next);
1347 qemu_mutex_unlock(&colo_compare_mutex);
1349 return;
1352 static void colo_flush_packets(void *opaque, void *user_data)
1354 CompareState *s = user_data;
1355 Connection *conn = opaque;
1356 Packet *pkt = NULL;
1358 while (!g_queue_is_empty(&conn->primary_list)) {
1359 pkt = g_queue_pop_head(&conn->primary_list);
1360 compare_chr_send(s,
1361 pkt->data,
1362 pkt->size,
1363 pkt->vnet_hdr_len,
1364 false,
1365 true);
1366 packet_destroy_partial(pkt, NULL);
1368 while (!g_queue_is_empty(&conn->secondary_list)) {
1369 pkt = g_queue_pop_head(&conn->secondary_list);
1370 packet_destroy(pkt, NULL);
1374 static void colo_compare_class_init(ObjectClass *oc, void *data)
1376 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1378 ucc->complete = colo_compare_complete;
1381 static void colo_compare_init(Object *obj)
1383 CompareState *s = COLO_COMPARE(obj);
1385 object_property_add_str(obj, "primary_in",
1386 compare_get_pri_indev, compare_set_pri_indev);
1387 object_property_add_str(obj, "secondary_in",
1388 compare_get_sec_indev, compare_set_sec_indev);
1389 object_property_add_str(obj, "outdev",
1390 compare_get_outdev, compare_set_outdev);
1391 object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1392 (Object **)&s->iothread,
1393 object_property_allow_set_link,
1394 OBJ_PROP_LINK_STRONG);
1395 /* This parameter just for Xen COLO */
1396 object_property_add_str(obj, "notify_dev",
1397 compare_get_notify_dev, compare_set_notify_dev);
1399 object_property_add(obj, "compare_timeout", "uint32",
1400 compare_get_timeout,
1401 compare_set_timeout, NULL, NULL);
1403 object_property_add(obj, "expired_scan_cycle", "uint32",
1404 compare_get_expired_scan_cycle,
1405 compare_set_expired_scan_cycle, NULL, NULL);
1407 object_property_add(obj, "max_queue_size", "uint32",
1408 get_max_queue_size,
1409 set_max_queue_size, NULL, NULL);
1411 s->vnet_hdr = false;
1412 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1413 compare_set_vnet_hdr);
1416 static void colo_compare_finalize(Object *obj)
1418 CompareState *s = COLO_COMPARE(obj);
1419 CompareState *tmp = NULL;
1421 qemu_mutex_lock(&colo_compare_mutex);
1422 QTAILQ_FOREACH(tmp, &net_compares, next) {
1423 if (tmp == s) {
1424 QTAILQ_REMOVE(&net_compares, s, next);
1425 break;
1428 if (QTAILQ_EMPTY(&net_compares)) {
1429 colo_compare_active = false;
1430 qemu_mutex_destroy(&event_mtx);
1431 qemu_cond_destroy(&event_complete_cond);
1433 qemu_mutex_unlock(&colo_compare_mutex);
1435 qemu_chr_fe_deinit(&s->chr_pri_in, false);
1436 qemu_chr_fe_deinit(&s->chr_sec_in, false);
1437 qemu_chr_fe_deinit(&s->chr_out, false);
1438 if (s->notify_dev) {
1439 qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1442 colo_compare_timer_del(s);
1444 qemu_bh_delete(s->event_bh);
1446 AioContext *ctx = iothread_get_aio_context(s->iothread);
1447 aio_context_acquire(ctx);
1448 AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1449 if (s->notify_dev) {
1450 AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1452 aio_context_release(ctx);
1454 /* Release all unhandled packets after compare thead exited */
1455 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1456 AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1458 g_queue_clear(&s->conn_list);
1459 g_queue_clear(&s->out_sendco.send_list);
1460 if (s->notify_dev) {
1461 g_queue_clear(&s->notify_sendco.send_list);
1464 if (s->connection_track_table) {
1465 g_hash_table_destroy(s->connection_track_table);
1468 object_unref(OBJECT(s->iothread));
1470 g_free(s->pri_indev);
1471 g_free(s->sec_indev);
1472 g_free(s->outdev);
1473 g_free(s->notify_dev);
1476 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1478 colo_compare_active = false;
1479 qemu_mutex_init(&colo_compare_mutex);
1482 static const TypeInfo colo_compare_info = {
1483 .name = TYPE_COLO_COMPARE,
1484 .parent = TYPE_OBJECT,
1485 .instance_size = sizeof(CompareState),
1486 .instance_init = colo_compare_init,
1487 .instance_finalize = colo_compare_finalize,
1488 .class_size = sizeof(CompareClass),
1489 .class_init = colo_compare_class_init,
1490 .interfaces = (InterfaceInfo[]) {
1491 { TYPE_USER_CREATABLE },
1496 static void register_types(void)
1498 type_register_static(&colo_compare_info);
1501 type_init(register_types);