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"
19 #include "qapi/error.h"
22 #include "qom/object_interfaces.h"
24 #include "qom/object.h"
25 #include "net/queue.h"
26 #include "chardev/char-fe.h"
27 #include "qemu/sockets.h"
29 #include "sysemu/iothread.h"
30 #include "net/colo-compare.h"
31 #include "migration/colo.h"
32 #include "migration/migration.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,
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;
70 * +---------------+ +---------------+ +---------------+
71 * | conn list + - > conn + ------- > conn + -- > ......
72 * +---------------+ +---------------+ +---------------+
74 * +---------------+ +---v----+ +---v----+ +---v----+ +---v----+
75 * |primary | |secondary |primary | |secondary
76 * |packet | |packet + |packet | |packet +
77 * +--------+ +--------+ +--------+ +--------+
79 * +---v----+ +---v----+ +---v----+ +---v----+
80 * |primary | |secondary |primary | |secondary
81 * |packet | |packet + |packet | |packet +
82 * +--------+ +--------+ +--------+ +--------+
84 * +---v----+ +---v----+ +---v----+ +---v----+
85 * |primary | |secondary |primary | |secondary
86 * |packet | |packet + |packet | |packet +
87 * +--------+ +--------+ +--------+ +--------+
90 typedef struct SendCo {
92 struct CompareState *s;
95 bool notify_remote_frame;
100 typedef struct SendEntry {
102 uint32_t vnet_hdr_len;
106 struct CompareState {
113 CharBackend chr_pri_in;
114 CharBackend chr_sec_in;
116 CharBackend chr_notify_dev;
117 SocketReadState pri_rs;
118 SocketReadState sec_rs;
119 SocketReadState notify_rs;
121 SendCo notify_sendco;
123 uint32_t compare_timeout;
124 uint32_t expired_scan_cycle;
127 * Record the connection that through the NIC
128 * Element type: Connection
131 /* Record the connection without repetition */
132 GHashTable *connection_track_table;
135 GMainContext *worker_context;
136 QEMUTimer *packet_check_timer;
139 enum colo_event event;
141 QTAILQ_ENTRY(CompareState) next;
144 typedef struct CompareClass {
145 ObjectClass parent_class;
153 static const char *colo_mode[] = {
154 [PRIMARY_IN] = "primary",
155 [SECONDARY_IN] = "secondary",
158 static int compare_chr_send(CompareState *s,
161 uint32_t vnet_hdr_len,
162 bool notify_remote_frame,
165 static bool packet_matches_str(const char *str,
169 if (packet_len != strlen(str)) {
173 return !memcmp(str, buf, strlen(str));
176 static void notify_remote_frame(CompareState *s)
178 char msg[] = "DO_CHECKPOINT";
181 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
183 error_report("Notify Xen COLO-frame failed");
187 static void colo_compare_inconsistency_notify(CompareState *s)
190 notify_remote_frame(s);
192 notifier_list_notify(&colo_compare_notifiers,
193 migrate_get_current());
197 /* Use restricted to colo_insert_packet() */
198 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
200 return a->tcp_seq - b->tcp_seq;
203 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
206 struct tcp_hdr *tcphd;
208 tcphd = (struct tcp_hdr *)pkt->transport_header;
210 pkt->tcp_seq = ntohl(tcphd->th_seq);
211 pkt->tcp_ack = ntohl(tcphd->th_ack);
212 *max_ack = *max_ack > pkt->tcp_ack ? *max_ack : pkt->tcp_ack;
213 pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
214 + (tcphd->th_off << 2) - pkt->vnet_hdr_len;
215 pkt->payload_size = pkt->size - pkt->header_size;
216 pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
217 pkt->flags = tcphd->th_flags;
221 * Return 1 on success, if return 0 means the
222 * packet will be dropped
224 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
226 if (g_queue_get_length(queue) <= max_queue_size) {
227 if (pkt->ip->ip_p == IPPROTO_TCP) {
228 fill_pkt_tcp_info(pkt, max_ack);
229 g_queue_insert_sorted(queue,
231 (GCompareDataFunc)seq_sorter,
234 g_queue_push_tail(queue, pkt);
242 * Return 0 on success, if return -1 means the pkt
243 * is unsupported(arp and ipv6) and will be sent later
245 static int packet_enqueue(CompareState *s, int mode, Connection **con)
252 if (mode == PRIMARY_IN) {
253 pkt = packet_new(s->pri_rs.buf,
254 s->pri_rs.packet_len,
255 s->pri_rs.vnet_hdr_len);
257 pkt = packet_new(s->sec_rs.buf,
258 s->sec_rs.packet_len,
259 s->sec_rs.vnet_hdr_len);
262 if (parse_packet_early(pkt)) {
263 packet_destroy(pkt, NULL);
267 fill_connection_key(pkt, &key);
269 conn = connection_get(s->connection_track_table,
273 if (!conn->processing) {
274 g_queue_push_tail(&s->conn_list, conn);
275 conn->processing = true;
278 if (mode == PRIMARY_IN) {
279 ret = colo_insert_packet(&conn->primary_list, pkt, &conn->pack);
281 ret = colo_insert_packet(&conn->secondary_list, pkt, &conn->sack);
285 trace_colo_compare_drop_packet(colo_mode[mode],
286 "queue size too big, drop packet");
287 packet_destroy(pkt, NULL);
296 static inline bool after(uint32_t seq1, uint32_t seq2)
298 return (int32_t)(seq1 - seq2) > 0;
301 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
304 ret = compare_chr_send(s,
311 error_report("colo send primary packet failed");
313 trace_colo_compare_main("packet same and release packet");
314 packet_destroy_partial(pkt, NULL);
318 * The IP packets sent by primary and secondary
319 * will be compared in here
320 * TODO support ip fragment, Out-Of-Order
321 * return: 0 means packet same
322 * > 0 || < 0 means packet different
324 static int colo_compare_packet_payload(Packet *ppkt,
331 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
332 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
334 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
335 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
336 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
337 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
339 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
340 pri_ip_dst, spkt->size,
341 sec_ip_src, sec_ip_dst);
344 return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
348 * return true means that the payload is consist and
349 * need to make the next comparison, false means do
352 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
353 int8_t *mark, uint32_t max_ack)
357 if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
358 if (!colo_compare_packet_payload(ppkt, spkt,
359 ppkt->header_size, spkt->header_size,
360 ppkt->payload_size)) {
361 *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
366 /* one part of secondary packet payload still need to be compared */
367 if (!after(ppkt->seq_end, spkt->seq_end)) {
368 if (!colo_compare_packet_payload(ppkt, spkt,
369 ppkt->header_size + ppkt->offset,
370 spkt->header_size + spkt->offset,
371 ppkt->payload_size - ppkt->offset)) {
372 if (!after(ppkt->tcp_ack, max_ack)) {
373 *mark = COLO_COMPARE_FREE_PRIMARY;
374 spkt->offset += ppkt->payload_size - ppkt->offset;
377 /* secondary guest hasn't ack the data, don't send
384 /* primary packet is longer than secondary packet, compare
385 * the same part and mark the primary packet offset
387 if (!colo_compare_packet_payload(ppkt, spkt,
388 ppkt->header_size + ppkt->offset,
389 spkt->header_size + spkt->offset,
390 spkt->payload_size - spkt->offset)) {
391 *mark = COLO_COMPARE_FREE_SECONDARY;
392 ppkt->offset += spkt->payload_size - spkt->offset;
400 static void colo_compare_tcp(CompareState *s, Connection *conn)
402 Packet *ppkt = NULL, *spkt = NULL;
406 * If ppkt and spkt have the same payload, but ppkt's ACK
407 * is greater than spkt's ACK, in this case we can not
408 * send the ppkt because it will cause the secondary guest
409 * to miss sending some data in the next. Therefore, we
410 * record the maximum ACK in the current queue at both
411 * primary side and secondary side. Only when the ack is
412 * less than the smaller of the two maximum ack, then we
413 * can ensure that the packet's payload is acknowledged by
414 * primary and secondary.
416 uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
419 if (g_queue_is_empty(&conn->primary_list)) {
422 ppkt = g_queue_pop_head(&conn->primary_list);
424 if (g_queue_is_empty(&conn->secondary_list)) {
425 g_queue_push_head(&conn->primary_list, ppkt);
428 spkt = g_queue_pop_head(&conn->secondary_list);
430 if (ppkt->tcp_seq == ppkt->seq_end) {
431 colo_release_primary_pkt(s, ppkt);
435 if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
436 trace_colo_compare_main("pri: this packet has compared");
437 colo_release_primary_pkt(s, ppkt);
441 if (spkt->tcp_seq == spkt->seq_end) {
442 packet_destroy(spkt, NULL);
449 if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
450 trace_colo_compare_main("sec: this packet has compared");
451 packet_destroy(spkt, NULL);
459 g_queue_push_head(&conn->secondary_list, spkt);
464 if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
465 trace_colo_compare_tcp_info("pri",
466 ppkt->tcp_seq, ppkt->tcp_ack,
467 ppkt->header_size, ppkt->payload_size,
468 ppkt->offset, ppkt->flags);
470 trace_colo_compare_tcp_info("sec",
471 spkt->tcp_seq, spkt->tcp_ack,
472 spkt->header_size, spkt->payload_size,
473 spkt->offset, spkt->flags);
475 if (mark == COLO_COMPARE_FREE_PRIMARY) {
476 conn->compare_seq = ppkt->seq_end;
477 colo_release_primary_pkt(s, ppkt);
478 g_queue_push_head(&conn->secondary_list, spkt);
481 if (mark == COLO_COMPARE_FREE_SECONDARY) {
482 conn->compare_seq = spkt->seq_end;
483 packet_destroy(spkt, NULL);
486 if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
487 conn->compare_seq = ppkt->seq_end;
488 colo_release_primary_pkt(s, ppkt);
489 packet_destroy(spkt, NULL);
493 g_queue_push_head(&conn->primary_list, ppkt);
494 g_queue_push_head(&conn->secondary_list, spkt);
496 #ifdef DEBUG_COLO_PACKETS
497 qemu_hexdump(stderr, "colo-compare ppkt", ppkt->data, ppkt->size);
498 qemu_hexdump(stderr, "colo-compare spkt", spkt->data, spkt->size);
501 colo_compare_inconsistency_notify(s);
507 * Called from the compare thread on the primary
508 * for compare udp packet
510 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
512 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
513 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
515 trace_colo_compare_main("compare udp");
518 * Because of ppkt and spkt are both in the same connection,
519 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
520 * same with spkt. In addition, IP header's Identification is a random
521 * field, we can handle it in IP fragmentation function later.
522 * COLO just concern the response net packet payload from primary guest
523 * and secondary guest are same or not, So we ignored all IP header include
524 * other field like TOS,TTL,IP Checksum. we only need to compare
525 * the ip payload here.
527 if (ppkt->size != spkt->size) {
528 trace_colo_compare_main("UDP: payload size of packets are different");
531 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
532 ppkt->size - offset)) {
533 trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
534 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
535 #ifdef DEBUG_COLO_PACKETS
536 qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
537 qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
546 * Called from the compare thread on the primary
547 * for compare icmp packet
549 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
551 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
552 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
554 trace_colo_compare_main("compare icmp");
557 * Because of ppkt and spkt are both in the same connection,
558 * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
559 * same with spkt. In addition, IP header's Identification is a random
560 * field, we can handle it in IP fragmentation function later.
561 * COLO just concern the response net packet payload from primary guest
562 * and secondary guest are same or not, So we ignored all IP header include
563 * other field like TOS,TTL,IP Checksum. we only need to compare
564 * the ip payload here.
566 if (ppkt->size != spkt->size) {
567 trace_colo_compare_main("ICMP: payload size of packets are different");
570 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
571 ppkt->size - offset)) {
572 trace_colo_compare_icmp_miscompare("primary pkt size",
574 trace_colo_compare_icmp_miscompare("Secondary pkt size",
576 #ifdef DEBUG_COLO_PACKETS
577 qemu_hexdump(stderr, "colo-compare pri pkt", ppkt->data, ppkt->size);
578 qemu_hexdump(stderr, "colo-compare sec pkt", spkt->data, spkt->size);
587 * Called from the compare thread on the primary
588 * for compare other packet
590 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
592 uint16_t offset = ppkt->vnet_hdr_len;
594 trace_colo_compare_main("compare other");
595 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_IP_INFO)) {
596 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
598 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
599 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
600 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
601 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
603 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
604 pri_ip_dst, spkt->size,
605 sec_ip_src, sec_ip_dst);
608 if (ppkt->size != spkt->size) {
609 trace_colo_compare_main("Other: payload size of packets are different");
612 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
613 ppkt->size - offset);
616 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
618 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
620 if ((now - pkt->creation_ms) > (*check_time)) {
621 trace_colo_old_packet_check_found(pkt->creation_ms);
628 void colo_compare_register_notifier(Notifier *notify)
630 notifier_list_add(&colo_compare_notifiers, notify);
633 void colo_compare_unregister_notifier(Notifier *notify)
635 notifier_remove(notify);
638 static int colo_old_packet_check_one_conn(Connection *conn,
641 GList *result = NULL;
643 result = g_queue_find_custom(&conn->primary_list,
645 (GCompareFunc)colo_old_packet_check_one);
648 /* Do checkpoint will flush old packet */
649 colo_compare_inconsistency_notify(s);
657 * Look for old packets that the secondary hasn't matched,
658 * if we have some then we have to checkpoint to wake
661 static void colo_old_packet_check(void *opaque)
663 CompareState *s = opaque;
666 * If we find one old packet, stop finding job and notify
667 * COLO frame do checkpoint.
669 g_queue_find_custom(&s->conn_list, s,
670 (GCompareFunc)colo_old_packet_check_one_conn);
673 static void colo_compare_packet(CompareState *s, Connection *conn,
674 int (*HandlePacket)(Packet *spkt,
678 GList *result = NULL;
680 while (!g_queue_is_empty(&conn->primary_list) &&
681 !g_queue_is_empty(&conn->secondary_list)) {
682 pkt = g_queue_pop_head(&conn->primary_list);
683 result = g_queue_find_custom(&conn->secondary_list,
684 pkt, (GCompareFunc)HandlePacket);
687 colo_release_primary_pkt(s, pkt);
688 g_queue_remove(&conn->secondary_list, result->data);
691 * If one packet arrive late, the secondary_list or
692 * primary_list will be empty, so we can't compare it
693 * until next comparison. If the packets in the list are
694 * timeout, it will trigger a checkpoint request.
696 trace_colo_compare_main("packet different");
697 g_queue_push_head(&conn->primary_list, pkt);
699 colo_compare_inconsistency_notify(s);
706 * Called from the compare thread on the primary
707 * for compare packet with secondary list of the
708 * specified connection when a new packet was
711 static void colo_compare_connection(void *opaque, void *user_data)
713 CompareState *s = user_data;
714 Connection *conn = opaque;
716 switch (conn->ip_proto) {
718 colo_compare_tcp(s, conn);
721 colo_compare_packet(s, conn, colo_packet_compare_udp);
724 colo_compare_packet(s, conn, colo_packet_compare_icmp);
727 colo_compare_packet(s, conn, colo_packet_compare_other);
732 static void coroutine_fn _compare_chr_send(void *opaque)
734 SendCo *sendco = opaque;
735 CompareState *s = sendco->s;
738 while (!g_queue_is_empty(&sendco->send_list)) {
739 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
740 uint32_t len = htonl(entry->size);
742 ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
744 if (ret != sizeof(len)) {
746 g_slice_free(SendEntry, entry);
750 if (!sendco->notify_remote_frame && s->vnet_hdr) {
752 * We send vnet header len make other module(like filter-redirector)
753 * know how to parse net packet correctly.
755 len = htonl(entry->vnet_hdr_len);
757 ret = qemu_chr_fe_write_all(sendco->chr,
761 if (ret != sizeof(len)) {
763 g_slice_free(SendEntry, entry);
768 ret = qemu_chr_fe_write_all(sendco->chr,
769 (uint8_t *)entry->buf,
772 if (ret != entry->size) {
774 g_slice_free(SendEntry, entry);
779 g_slice_free(SendEntry, entry);
786 while (!g_queue_is_empty(&sendco->send_list)) {
787 SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
789 g_slice_free(SendEntry, entry);
791 sendco->ret = ret < 0 ? ret : -EIO;
798 static int compare_chr_send(CompareState *s,
801 uint32_t vnet_hdr_len,
802 bool notify_remote_frame,
808 if (notify_remote_frame) {
809 sendco = &s->notify_sendco;
811 sendco = &s->out_sendco;
818 entry = g_slice_new(SendEntry);
820 entry->vnet_hdr_len = vnet_hdr_len;
824 entry->buf = g_malloc(size);
825 memcpy(entry->buf, buf, size);
827 g_queue_push_head(&sendco->send_list, entry);
830 sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
831 sendco->done = false;
832 qemu_coroutine_enter(sendco->co);
834 /* report early errors */
843 static int compare_chr_can_read(void *opaque)
845 return COMPARE_READ_LEN_MAX;
849 * Called from the main thread on the primary for packets
850 * arriving over the socket from the primary.
852 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
854 CompareState *s = COLO_COMPARE(opaque);
857 ret = net_fill_rstate(&s->pri_rs, buf, size);
859 qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
861 error_report("colo-compare primary_in error");
866 * Called from the main thread on the primary for packets
867 * arriving over the socket from the secondary.
869 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
871 CompareState *s = COLO_COMPARE(opaque);
874 ret = net_fill_rstate(&s->sec_rs, buf, size);
876 qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
878 error_report("colo-compare secondary_in error");
882 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
884 CompareState *s = COLO_COMPARE(opaque);
887 ret = net_fill_rstate(&s->notify_rs, buf, size);
889 qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
891 error_report("colo-compare notify_dev error");
896 * Check old packet regularly so it can watch for any packets
897 * that the secondary hasn't produced equivalents of.
899 static void check_old_packet_regular(void *opaque)
901 CompareState *s = opaque;
903 /* if have old packet we will notify checkpoint */
904 colo_old_packet_check(s);
905 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
906 s->expired_scan_cycle);
909 /* Public API, Used for COLO frame to notify compare event */
910 void colo_notify_compares_event(void *opaque, int event, Error **errp)
913 qemu_mutex_lock(&colo_compare_mutex);
915 if (!colo_compare_active) {
916 qemu_mutex_unlock(&colo_compare_mutex);
920 qemu_mutex_lock(&event_mtx);
921 QTAILQ_FOREACH(s, &net_compares, next) {
923 qemu_bh_schedule(s->event_bh);
924 event_unhandled_count++;
926 /* Wait all compare threads to finish handling this event */
927 while (event_unhandled_count > 0) {
928 qemu_cond_wait(&event_complete_cond, &event_mtx);
931 qemu_mutex_unlock(&event_mtx);
932 qemu_mutex_unlock(&colo_compare_mutex);
935 static void colo_compare_timer_init(CompareState *s)
937 AioContext *ctx = iothread_get_aio_context(s->iothread);
939 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
940 SCALE_MS, check_old_packet_regular,
942 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
943 s->expired_scan_cycle);
946 static void colo_compare_timer_del(CompareState *s)
948 if (s->packet_check_timer) {
949 timer_del(s->packet_check_timer);
950 timer_free(s->packet_check_timer);
951 s->packet_check_timer = NULL;
955 static void colo_flush_packets(void *opaque, void *user_data);
957 static void colo_compare_handle_event(void *opaque)
959 CompareState *s = opaque;
962 case COLO_EVENT_CHECKPOINT:
963 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
965 case COLO_EVENT_FAILOVER:
971 qemu_mutex_lock(&event_mtx);
972 assert(event_unhandled_count > 0);
973 event_unhandled_count--;
974 qemu_cond_broadcast(&event_complete_cond);
975 qemu_mutex_unlock(&event_mtx);
978 static void colo_compare_iothread(CompareState *s)
980 AioContext *ctx = iothread_get_aio_context(s->iothread);
981 object_ref(OBJECT(s->iothread));
982 s->worker_context = iothread_get_g_main_context(s->iothread);
984 qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
985 compare_pri_chr_in, NULL, NULL,
986 s, s->worker_context, true);
987 qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
988 compare_sec_chr_in, NULL, NULL,
989 s, s->worker_context, true);
991 qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
992 compare_notify_chr, NULL, NULL,
993 s, s->worker_context, true);
996 colo_compare_timer_init(s);
997 s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
1000 static char *compare_get_pri_indev(Object *obj, Error **errp)
1002 CompareState *s = COLO_COMPARE(obj);
1004 return g_strdup(s->pri_indev);
1007 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1009 CompareState *s = COLO_COMPARE(obj);
1011 g_free(s->pri_indev);
1012 s->pri_indev = g_strdup(value);
1015 static char *compare_get_sec_indev(Object *obj, Error **errp)
1017 CompareState *s = COLO_COMPARE(obj);
1019 return g_strdup(s->sec_indev);
1022 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1024 CompareState *s = COLO_COMPARE(obj);
1026 g_free(s->sec_indev);
1027 s->sec_indev = g_strdup(value);
1030 static char *compare_get_outdev(Object *obj, Error **errp)
1032 CompareState *s = COLO_COMPARE(obj);
1034 return g_strdup(s->outdev);
1037 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1039 CompareState *s = COLO_COMPARE(obj);
1042 s->outdev = g_strdup(value);
1045 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1047 CompareState *s = COLO_COMPARE(obj);
1052 static void compare_set_vnet_hdr(Object *obj,
1056 CompareState *s = COLO_COMPARE(obj);
1058 s->vnet_hdr = value;
1061 static char *compare_get_notify_dev(Object *obj, Error **errp)
1063 CompareState *s = COLO_COMPARE(obj);
1065 return g_strdup(s->notify_dev);
1068 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1070 CompareState *s = COLO_COMPARE(obj);
1072 g_free(s->notify_dev);
1073 s->notify_dev = g_strdup(value);
1076 static void compare_get_timeout(Object *obj, Visitor *v,
1077 const char *name, void *opaque,
1080 CompareState *s = COLO_COMPARE(obj);
1081 uint32_t value = s->compare_timeout;
1083 visit_type_uint32(v, name, &value, errp);
1086 static void compare_set_timeout(Object *obj, Visitor *v,
1087 const char *name, void *opaque,
1090 CompareState *s = COLO_COMPARE(obj);
1093 if (!visit_type_uint32(v, name, &value, errp)) {
1097 error_setg(errp, "Property '%s.%s' requires a positive value",
1098 object_get_typename(obj), name);
1101 s->compare_timeout = value;
1104 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1105 const char *name, void *opaque,
1108 CompareState *s = COLO_COMPARE(obj);
1109 uint32_t value = s->expired_scan_cycle;
1111 visit_type_uint32(v, name, &value, errp);
1114 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1115 const char *name, void *opaque,
1118 CompareState *s = COLO_COMPARE(obj);
1121 if (!visit_type_uint32(v, name, &value, errp)) {
1125 error_setg(errp, "Property '%s.%s' requires a positive value",
1126 object_get_typename(obj), name);
1129 s->expired_scan_cycle = value;
1132 static void get_max_queue_size(Object *obj, Visitor *v,
1133 const char *name, void *opaque,
1136 uint32_t value = max_queue_size;
1138 visit_type_uint32(v, name, &value, errp);
1141 static void set_max_queue_size(Object *obj, Visitor *v,
1142 const char *name, void *opaque,
1145 Error *local_err = NULL;
1148 visit_type_uint32(v, name, &value, &local_err);
1153 error_setg(&local_err, "Property '%s.%s' requires a positive value",
1154 object_get_typename(obj), name);
1157 max_queue_size = value;
1160 error_propagate(errp, local_err);
1163 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1165 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1166 Connection *conn = NULL;
1168 if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1169 trace_colo_compare_main("primary: unsupported packet in");
1173 pri_rs->vnet_hdr_len,
1177 /* compare packet in the specified connection */
1178 colo_compare_connection(conn, s);
1182 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1184 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1185 Connection *conn = NULL;
1187 if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1188 trace_colo_compare_main("secondary: unsupported packet in");
1190 /* compare packet in the specified connection */
1191 colo_compare_connection(conn, s);
1195 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1197 CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1199 const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1202 if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1204 notify_rs->packet_len)) {
1205 ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1207 error_report("Notify Xen COLO-frame INIT failed");
1209 } else if (packet_matches_str("COLO_CHECKPOINT",
1211 notify_rs->packet_len)) {
1212 /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1213 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1215 error_report("COLO compare got unsupported instruction");
1220 * Return 0 is success.
1221 * Return 1 is failed.
1223 static int find_and_check_chardev(Chardev **chr,
1227 *chr = qemu_chr_find(chr_name);
1229 error_setg(errp, "Device '%s' not found",
1234 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1235 error_setg(errp, "chardev \"%s\" is not reconnectable",
1240 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1241 error_setg(errp, "chardev \"%s\" cannot switch context",
1250 * Called from the main thread on the primary
1251 * to setup colo-compare.
1253 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1255 CompareState *s = COLO_COMPARE(uc);
1258 if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1259 error_setg(errp, "colo compare needs 'primary_in' ,"
1260 "'secondary_in','outdev','iothread' property set");
1262 } else if (!strcmp(s->pri_indev, s->outdev) ||
1263 !strcmp(s->sec_indev, s->outdev) ||
1264 !strcmp(s->pri_indev, s->sec_indev)) {
1265 error_setg(errp, "'indev' and 'outdev' could not be same "
1266 "for compare module");
1270 if (!s->compare_timeout) {
1271 /* Set default value to 3000 MS */
1272 s->compare_timeout = DEFAULT_TIME_OUT_MS;
1275 if (!s->expired_scan_cycle) {
1276 /* Set default value to 3000 MS */
1277 s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1280 if (!max_queue_size) {
1281 /* Set default queue size to 1024 */
1282 max_queue_size = MAX_QUEUE_SIZE;
1285 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1286 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1290 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1291 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1295 if (find_and_check_chardev(&chr, s->outdev, errp) ||
1296 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1300 net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1301 net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1303 /* Try to enable remote notify chardev, currently just for Xen COLO */
1304 if (s->notify_dev) {
1305 if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1306 !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1310 net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1314 s->out_sendco.s = s;
1315 s->out_sendco.chr = &s->chr_out;
1316 s->out_sendco.notify_remote_frame = false;
1317 s->out_sendco.done = true;
1318 g_queue_init(&s->out_sendco.send_list);
1320 if (s->notify_dev) {
1321 s->notify_sendco.s = s;
1322 s->notify_sendco.chr = &s->chr_notify_dev;
1323 s->notify_sendco.notify_remote_frame = true;
1324 s->notify_sendco.done = true;
1325 g_queue_init(&s->notify_sendco.send_list);
1328 g_queue_init(&s->conn_list);
1330 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1331 connection_key_equal,
1333 connection_destroy);
1335 colo_compare_iothread(s);
1337 qemu_mutex_lock(&colo_compare_mutex);
1338 if (!colo_compare_active) {
1339 qemu_mutex_init(&event_mtx);
1340 qemu_cond_init(&event_complete_cond);
1341 colo_compare_active = true;
1343 QTAILQ_INSERT_TAIL(&net_compares, s, next);
1344 qemu_mutex_unlock(&colo_compare_mutex);
1349 static void colo_flush_packets(void *opaque, void *user_data)
1351 CompareState *s = user_data;
1352 Connection *conn = opaque;
1355 while (!g_queue_is_empty(&conn->primary_list)) {
1356 pkt = g_queue_pop_head(&conn->primary_list);
1363 packet_destroy_partial(pkt, NULL);
1365 while (!g_queue_is_empty(&conn->secondary_list)) {
1366 pkt = g_queue_pop_head(&conn->secondary_list);
1367 packet_destroy(pkt, NULL);
1371 static void colo_compare_class_init(ObjectClass *oc, void *data)
1373 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1375 ucc->complete = colo_compare_complete;
1378 static void colo_compare_init(Object *obj)
1380 CompareState *s = COLO_COMPARE(obj);
1382 object_property_add_str(obj, "primary_in",
1383 compare_get_pri_indev, compare_set_pri_indev);
1384 object_property_add_str(obj, "secondary_in",
1385 compare_get_sec_indev, compare_set_sec_indev);
1386 object_property_add_str(obj, "outdev",
1387 compare_get_outdev, compare_set_outdev);
1388 object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1389 (Object **)&s->iothread,
1390 object_property_allow_set_link,
1391 OBJ_PROP_LINK_STRONG);
1392 /* This parameter just for Xen COLO */
1393 object_property_add_str(obj, "notify_dev",
1394 compare_get_notify_dev, compare_set_notify_dev);
1396 object_property_add(obj, "compare_timeout", "uint32",
1397 compare_get_timeout,
1398 compare_set_timeout, NULL, NULL);
1400 object_property_add(obj, "expired_scan_cycle", "uint32",
1401 compare_get_expired_scan_cycle,
1402 compare_set_expired_scan_cycle, NULL, NULL);
1404 object_property_add(obj, "max_queue_size", "uint32",
1406 set_max_queue_size, NULL, NULL);
1408 s->vnet_hdr = false;
1409 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1410 compare_set_vnet_hdr);
1413 static void colo_compare_finalize(Object *obj)
1415 CompareState *s = COLO_COMPARE(obj);
1416 CompareState *tmp = NULL;
1418 qemu_mutex_lock(&colo_compare_mutex);
1419 QTAILQ_FOREACH(tmp, &net_compares, next) {
1421 QTAILQ_REMOVE(&net_compares, s, next);
1425 if (QTAILQ_EMPTY(&net_compares)) {
1426 colo_compare_active = false;
1427 qemu_mutex_destroy(&event_mtx);
1428 qemu_cond_destroy(&event_complete_cond);
1430 qemu_mutex_unlock(&colo_compare_mutex);
1432 qemu_chr_fe_deinit(&s->chr_pri_in, false);
1433 qemu_chr_fe_deinit(&s->chr_sec_in, false);
1434 qemu_chr_fe_deinit(&s->chr_out, false);
1435 if (s->notify_dev) {
1436 qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1439 colo_compare_timer_del(s);
1441 qemu_bh_delete(s->event_bh);
1443 AioContext *ctx = iothread_get_aio_context(s->iothread);
1444 aio_context_acquire(ctx);
1445 AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1446 if (s->notify_dev) {
1447 AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1449 aio_context_release(ctx);
1451 /* Release all unhandled packets after compare thead exited */
1452 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1453 AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1455 g_queue_clear(&s->conn_list);
1456 g_queue_clear(&s->out_sendco.send_list);
1457 if (s->notify_dev) {
1458 g_queue_clear(&s->notify_sendco.send_list);
1461 if (s->connection_track_table) {
1462 g_hash_table_destroy(s->connection_track_table);
1465 object_unref(OBJECT(s->iothread));
1467 g_free(s->pri_indev);
1468 g_free(s->sec_indev);
1470 g_free(s->notify_dev);
1473 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1475 colo_compare_active = false;
1476 qemu_mutex_init(&colo_compare_mutex);
1479 static const TypeInfo colo_compare_info = {
1480 .name = TYPE_COLO_COMPARE,
1481 .parent = TYPE_OBJECT,
1482 .instance_size = sizeof(CompareState),
1483 .instance_init = colo_compare_init,
1484 .instance_finalize = colo_compare_finalize,
1485 .class_size = sizeof(CompareClass),
1486 .class_init = colo_compare_class_init,
1487 .interfaces = (InterfaceInfo[]) {
1488 { TYPE_USER_CREATABLE },
1493 static void register_types(void)
1495 type_register_static(&colo_compare_info);
1498 type_init(register_types);