Merge remote-tracking branch 'remotes/cohuck/tags/s390x-20200127' into staging
[qemu.git] / net / colo-compare.c
1 /*
2 * COarse-grain LOck-stepping Virtual Machines for Non-stop Service (COLO)
3 * (a.k.a. Fault Tolerance or Continuous Replication)
4 *
5 * Copyright (c) 2016 HUAWEI TECHNOLOGIES CO., LTD.
6 * Copyright (c) 2016 FUJITSU LIMITED
7 * Copyright (c) 2016 Intel Corporation
8 *
9 * Author: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
10 *
11 * This work is licensed under the terms of the GNU GPL, version 2 or
12 * later. See the COPYING file in the top-level directory.
13 */
14
15 #include "qemu/osdep.h"
16 #include "qemu-common.h"
17 #include "qemu/error-report.h"
18 #include "trace.h"
19 #include "qapi/error.h"
20 #include "net/net.h"
21 #include "net/eth.h"
22 #include "qom/object_interfaces.h"
23 #include "qemu/iov.h"
24 #include "qom/object.h"
25 #include "net/queue.h"
26 #include "chardev/char-fe.h"
27 #include "qemu/sockets.h"
28 #include "colo.h"
29 #include "sysemu/iothread.h"
30 #include "net/colo-compare.h"
31 #include "migration/colo.h"
32 #include "migration/migration.h"
33 #include "util.h"
34
35 #define TYPE_COLO_COMPARE "colo-compare"
36 #define COLO_COMPARE(obj) \
37 OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
38
39 static QTAILQ_HEAD(, CompareState) net_compares =
40 QTAILQ_HEAD_INITIALIZER(net_compares);
41
42 static NotifierList colo_compare_notifiers =
43 NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
44
45 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
46 #define MAX_QUEUE_SIZE 1024
47
48 #define COLO_COMPARE_FREE_PRIMARY 0x01
49 #define COLO_COMPARE_FREE_SECONDARY 0x02
50
51 /* TODO: Should be configurable */
52 #define REGULAR_PACKET_CHECK_MS 3000
53
54 static QemuMutex event_mtx;
55 static QemuCond event_complete_cond;
56 static int event_unhandled_count;
57
58 /*
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 * +--------+ +--------+ +--------+ +--------+
79 */
80 typedef struct CompareState {
81 Object parent;
82
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;
95
96 /*
97 * Record the connection that through the NIC
98 * Element type: Connection
99 */
100 GQueue conn_list;
101 /* Record the connection without repetition */
102 GHashTable *connection_track_table;
103
104 IOThread *iothread;
105 GMainContext *worker_context;
106 QEMUTimer *packet_check_timer;
107
108 QEMUBH *event_bh;
109 enum colo_event event;
110
111 QTAILQ_ENTRY(CompareState) next;
112 } CompareState;
113
114 typedef struct CompareClass {
115 ObjectClass parent_class;
116 } CompareClass;
117
118 enum {
119 PRIMARY_IN = 0,
120 SECONDARY_IN,
121 };
122
123
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);
129
130 static bool packet_matches_str(const char *str,
131 const uint8_t *buf,
132 uint32_t packet_len)
133 {
134 if (packet_len != strlen(str)) {
135 return false;
136 }
137
138 return !memcmp(str, buf, strlen(str));
139 }
140
141 static void notify_remote_frame(CompareState *s)
142 {
143 char msg[] = "DO_CHECKPOINT";
144 int ret = 0;
145
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");
149 }
150 }
151
152 static void colo_compare_inconsistency_notify(CompareState *s)
153 {
154 if (s->notify_dev) {
155 notify_remote_frame(s);
156 } else {
157 notifier_list_notify(&colo_compare_notifiers,
158 migrate_get_current());
159 }
160 }
161
162 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
163 {
164 struct tcp_hdr *atcp, *btcp;
165
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);
169 }
170
171 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
172 {
173 Packet *pkt = data;
174 struct tcp_hdr *tcphd;
175
176 tcphd = (struct tcp_hdr *)pkt->transport_header;
177
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;
186 }
187
188 /*
189 * Return 1 on success, if return 0 means the
190 * packet will be dropped
191 */
192 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
193 {
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);
203 }
204 return 1;
205 }
206 return 0;
207 }
208
209 /*
210 * Return 0 on success, if return -1 means the pkt
211 * is unsupported(arp and ipv6) and will be sent later
212 */
213 static int packet_enqueue(CompareState *s, int mode, Connection **con)
214 {
215 ConnectionKey key;
216 Packet *pkt = NULL;
217 Connection *conn;
218
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);
227 }
228
229 if (parse_packet_early(pkt)) {
230 packet_destroy(pkt, NULL);
231 pkt = NULL;
232 return -1;
233 }
234 fill_connection_key(pkt, &key);
235
236 conn = connection_get(s->connection_track_table,
237 &key,
238 &s->conn_list);
239
240 if (!conn->processing) {
241 g_queue_push_tail(&s->conn_list, conn);
242 conn->processing = true;
243 }
244
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");
249 }
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");
254 }
255 }
256 *con = conn;
257
258 return 0;
259 }
260
261 static inline bool after(uint32_t seq1, uint32_t seq2)
262 {
263 return (int32_t)(seq1 - seq2) > 0;
264 }
265
266 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
267 {
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");
276 }
277 trace_colo_compare_main("packet same and release packet");
278 packet_destroy(pkt, NULL);
279 }
280
281 /*
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
287 */
288 static int colo_compare_packet_payload(Packet *ppkt,
289 Packet *spkt,
290 uint16_t poffset,
291 uint16_t soffset,
292 uint16_t len)
293
294 {
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];
297
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));
302
303 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
304 pri_ip_dst, spkt->size,
305 sec_ip_src, sec_ip_dst);
306 }
307
308 return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
309 }
310
311 /*
312 * return true means that the payload is consist and
313 * need to make the next comparison, false means do
314 * the checkpoint
315 */
316 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
317 int8_t *mark, uint32_t max_ack)
318 {
319 *mark = 0;
320
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;
327 }
328 }
329
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
343 */
344 return false;
345 }
346 }
347 } else {
348 /* primary packet is longer than secondary packet, compare
349 * the same part and mark the primary packet offset
350 */
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;
358 }
359 }
360
361 return false;
362 }
363
364 static void colo_compare_tcp(CompareState *s, Connection *conn)
365 {
366 Packet *ppkt = NULL, *spkt = NULL;
367 int8_t mark;
368
369 /*
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.
379 */
380 uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
381
382 pri:
383 if (g_queue_is_empty(&conn->primary_list)) {
384 return;
385 }
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;
391 }
392 spkt = g_queue_pop_head(&conn->secondary_list);
393
394 if (ppkt->tcp_seq == ppkt->seq_end) {
395 colo_release_primary_pkt(s, ppkt);
396 ppkt = NULL;
397 }
398
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;
403 }
404
405 if (spkt->tcp_seq == spkt->seq_end) {
406 packet_destroy(spkt, NULL);
407 if (!ppkt) {
408 goto pri;
409 } else {
410 goto sec;
411 }
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;
420 }
421 }
422 if (!ppkt) {
423 g_queue_push_head(&conn->secondary_list, spkt);
424 goto pri;
425 }
426 }
427
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);
433
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);
438
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;
444 }
445 if (mark == COLO_COMPARE_FREE_SECONDARY) {
446 conn->compare_seq = spkt->seq_end;
447 packet_destroy(spkt, NULL);
448 goto sec;
449 }
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;
455 }
456 } else {
457 g_queue_push_head(&conn->primary_list, ppkt);
458 g_queue_push_head(&conn->secondary_list, spkt);
459
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);
464
465 colo_compare_inconsistency_notify(s);
466 }
467 }
468
469
470 /*
471 * Called from the compare thread on the primary
472 * for compare udp packet
473 */
474 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
475 {
476 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
477 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
478
479 trace_colo_compare_main("compare udp");
480
481 /*
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.
490 */
491 if (ppkt->size != spkt->size) {
492 trace_colo_compare_main("UDP: payload size of packets are different");
493 return -1;
494 }
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);
504 }
505 return -1;
506 } else {
507 return 0;
508 }
509 }
510
511 /*
512 * Called from the compare thread on the primary
513 * for compare icmp packet
514 */
515 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
516 {
517 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
518 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
519
520 trace_colo_compare_main("compare icmp");
521
522 /*
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.
531 */
532 if (ppkt->size != spkt->size) {
533 trace_colo_compare_main("ICMP: payload size of packets are different");
534 return -1;
535 }
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);
547 }
548 return -1;
549 } else {
550 return 0;
551 }
552 }
553
554 /*
555 * Called from the compare thread on the primary
556 * for compare other packet
557 */
558 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
559 {
560 uint16_t offset = ppkt->vnet_hdr_len;
561
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];
565
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));
570
571 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
572 pri_ip_dst, spkt->size,
573 sec_ip_src, sec_ip_dst);
574 }
575
576 if (ppkt->size != spkt->size) {
577 trace_colo_compare_main("Other: payload size of packets are different");
578 return -1;
579 }
580 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
581 ppkt->size - offset);
582 }
583
584 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
585 {
586 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
587
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;
593 }
594 }
595
596 void colo_compare_register_notifier(Notifier *notify)
597 {
598 notifier_list_add(&colo_compare_notifiers, notify);
599 }
600
601 void colo_compare_unregister_notifier(Notifier *notify)
602 {
603 notifier_remove(notify);
604 }
605
606 static int colo_old_packet_check_one_conn(Connection *conn,
607 CompareState *s)
608 {
609 GList *result = NULL;
610 int64_t check_time = REGULAR_PACKET_CHECK_MS;
611
612 result = g_queue_find_custom(&conn->primary_list,
613 &check_time,
614 (GCompareFunc)colo_old_packet_check_one);
615
616 if (result) {
617 /* Do checkpoint will flush old packet */
618 colo_compare_inconsistency_notify(s);
619 return 0;
620 }
621
622 return 1;
623 }
624
625 /*
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.
629 */
630 static void colo_old_packet_check(void *opaque)
631 {
632 CompareState *s = opaque;
633
634 /*
635 * If we find one old packet, stop finding job and notify
636 * COLO frame do checkpoint.
637 */
638 g_queue_find_custom(&s->conn_list, s,
639 (GCompareFunc)colo_old_packet_check_one_conn);
640 }
641
642 static void colo_compare_packet(CompareState *s, Connection *conn,
643 int (*HandlePacket)(Packet *spkt,
644 Packet *ppkt))
645 {
646 Packet *pkt = NULL;
647 GList *result = NULL;
648
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);
654
655 if (result) {
656 colo_release_primary_pkt(s, pkt);
657 g_queue_remove(&conn->secondary_list, result->data);
658 } else {
659 /*
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.
664 */
665 trace_colo_compare_main("packet different");
666 g_queue_push_head(&conn->primary_list, pkt);
667
668 colo_compare_inconsistency_notify(s);
669 break;
670 }
671 }
672 }
673
674 /*
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.
679 */
680 static void colo_compare_connection(void *opaque, void *user_data)
681 {
682 CompareState *s = user_data;
683 Connection *conn = opaque;
684
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;
698 }
699 }
700
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)
706 {
707 int ret = 0;
708 uint32_t len = htonl(size);
709
710 if (!size) {
711 return 0;
712 }
713
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));
720 }
721
722 if (ret != sizeof(len)) {
723 goto err;
724 }
725
726 if (s->vnet_hdr) {
727 /*
728 * We send vnet header len make other module(like filter-redirector)
729 * know how to parse net packet correctly.
730 */
731 len = htonl(vnet_hdr_len);
732
733 if (!notify_remote_frame) {
734 ret = qemu_chr_fe_write_all(&s->chr_out,
735 (uint8_t *)&len,
736 sizeof(len));
737 }
738
739 if (ret != sizeof(len)) {
740 goto err;
741 }
742 }
743
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);
750 }
751
752 if (ret != size) {
753 goto err;
754 }
755
756 return 0;
757
758 err:
759 return ret < 0 ? ret : -EIO;
760 }
761
762 static int compare_chr_can_read(void *opaque)
763 {
764 return COMPARE_READ_LEN_MAX;
765 }
766
767 /*
768 * Called from the main thread on the primary for packets
769 * arriving over the socket from the primary.
770 */
771 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
772 {
773 CompareState *s = COLO_COMPARE(opaque);
774 int ret;
775
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");
781 }
782 }
783
784 /*
785 * Called from the main thread on the primary for packets
786 * arriving over the socket from the secondary.
787 */
788 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
789 {
790 CompareState *s = COLO_COMPARE(opaque);
791 int ret;
792
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");
798 }
799 }
800
801 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
802 {
803 CompareState *s = COLO_COMPARE(opaque);
804 int ret;
805
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");
811 }
812 }
813
814 /*
815 * Check old packet regularly so it can watch for any packets
816 * that the secondary hasn't produced equivalents of.
817 */
818 static void check_old_packet_regular(void *opaque)
819 {
820 CompareState *s = opaque;
821
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);
826 }
827
828 /* Public API, Used for COLO frame to notify compare event */
829 void colo_notify_compares_event(void *opaque, int event, Error **errp)
830 {
831 CompareState *s;
832
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++;
838 }
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);
842 }
843
844 qemu_mutex_unlock(&event_mtx);
845 }
846
847 static void colo_compare_timer_init(CompareState *s)
848 {
849 AioContext *ctx = iothread_get_aio_context(s->iothread);
850
851 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
852 SCALE_MS, check_old_packet_regular,
853 s);
854 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
855 REGULAR_PACKET_CHECK_MS);
856 }
857
858 static void colo_compare_timer_del(CompareState *s)
859 {
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;
864 }
865 }
866
867 static void colo_flush_packets(void *opaque, void *user_data);
868
869 static void colo_compare_handle_event(void *opaque)
870 {
871 CompareState *s = opaque;
872
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;
881 }
882
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);
888 }
889
890 static void colo_compare_iothread(CompareState *s)
891 {
892 object_ref(OBJECT(s->iothread));
893 s->worker_context = iothread_get_g_main_context(s->iothread);
894
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);
905 }
906
907 colo_compare_timer_init(s);
908 s->event_bh = qemu_bh_new(colo_compare_handle_event, s);
909 }
910
911 static char *compare_get_pri_indev(Object *obj, Error **errp)
912 {
913 CompareState *s = COLO_COMPARE(obj);
914
915 return g_strdup(s->pri_indev);
916 }
917
918 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
919 {
920 CompareState *s = COLO_COMPARE(obj);
921
922 g_free(s->pri_indev);
923 s->pri_indev = g_strdup(value);
924 }
925
926 static char *compare_get_sec_indev(Object *obj, Error **errp)
927 {
928 CompareState *s = COLO_COMPARE(obj);
929
930 return g_strdup(s->sec_indev);
931 }
932
933 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
934 {
935 CompareState *s = COLO_COMPARE(obj);
936
937 g_free(s->sec_indev);
938 s->sec_indev = g_strdup(value);
939 }
940
941 static char *compare_get_outdev(Object *obj, Error **errp)
942 {
943 CompareState *s = COLO_COMPARE(obj);
944
945 return g_strdup(s->outdev);
946 }
947
948 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
949 {
950 CompareState *s = COLO_COMPARE(obj);
951
952 g_free(s->outdev);
953 s->outdev = g_strdup(value);
954 }
955
956 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
957 {
958 CompareState *s = COLO_COMPARE(obj);
959
960 return s->vnet_hdr;
961 }
962
963 static void compare_set_vnet_hdr(Object *obj,
964 bool value,
965 Error **errp)
966 {
967 CompareState *s = COLO_COMPARE(obj);
968
969 s->vnet_hdr = value;
970 }
971
972 static char *compare_get_notify_dev(Object *obj, Error **errp)
973 {
974 CompareState *s = COLO_COMPARE(obj);
975
976 return g_strdup(s->notify_dev);
977 }
978
979 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
980 {
981 CompareState *s = COLO_COMPARE(obj);
982
983 g_free(s->notify_dev);
984 s->notify_dev = g_strdup(value);
985 }
986
987 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
988 {
989 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
990 Connection *conn = NULL;
991
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);
1002 }
1003 }
1004
1005 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1006 {
1007 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1008 Connection *conn = NULL;
1009
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);
1015 }
1016 }
1017
1018 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1019 {
1020 CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1021
1022 const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1023 int ret;
1024
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");
1031 }
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");
1039 }
1040 }
1041
1042 /*
1043 * Return 0 is success.
1044 * Return 1 is failed.
1045 */
1046 static int find_and_check_chardev(Chardev **chr,
1047 char *chr_name,
1048 Error **errp)
1049 {
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;
1055 }
1056
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;
1061 }
1062
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;
1067 }
1068
1069 return 0;
1070 }
1071
1072 /*
1073 * Called from the main thread on the primary
1074 * to setup colo-compare.
1075 */
1076 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1077 {
1078 CompareState *s = COLO_COMPARE(uc);
1079 Chardev *chr;
1080
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;
1091 }
1092
1093 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1094 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1095 return;
1096 }
1097
1098 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1099 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1100 return;
1101 }
1102
1103 if (find_and_check_chardev(&chr, s->outdev, errp) ||
1104 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1105 return;
1106 }
1107
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);
1110
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;
1116 }
1117
1118 net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1119 s->vnet_hdr);
1120 }
1121
1122 QTAILQ_INSERT_TAIL(&net_compares, s, next);
1123
1124 g_queue_init(&s->conn_list);
1125
1126 qemu_mutex_init(&event_mtx);
1127 qemu_cond_init(&event_complete_cond);
1128
1129 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1130 connection_key_equal,
1131 g_free,
1132 connection_destroy);
1133
1134 colo_compare_iothread(s);
1135 return;
1136 }
1137
1138 static void colo_flush_packets(void *opaque, void *user_data)
1139 {
1140 CompareState *s = user_data;
1141 Connection *conn = opaque;
1142 Packet *pkt = NULL;
1143
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);
1152 }
1153 while (!g_queue_is_empty(&conn->secondary_list)) {
1154 pkt = g_queue_pop_head(&conn->secondary_list);
1155 packet_destroy(pkt, NULL);
1156 }
1157 }
1158
1159 static void colo_compare_class_init(ObjectClass *oc, void *data)
1160 {
1161 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1162
1163 ucc->complete = colo_compare_complete;
1164 }
1165
1166 static void colo_compare_init(Object *obj)
1167 {
1168 CompareState *s = COLO_COMPARE(obj);
1169
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);
1187
1188 s->vnet_hdr = false;
1189 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1190 compare_set_vnet_hdr, NULL);
1191 }
1192
1193 static void colo_compare_finalize(Object *obj)
1194 {
1195 CompareState *s = COLO_COMPARE(obj);
1196 CompareState *tmp = NULL;
1197
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);
1203 }
1204
1205 if (s->iothread) {
1206 colo_compare_timer_del(s);
1207 }
1208
1209 qemu_bh_delete(s->event_bh);
1210
1211 QTAILQ_FOREACH(tmp, &net_compares, next) {
1212 if (tmp == s) {
1213 QTAILQ_REMOVE(&net_compares, s, next);
1214 break;
1215 }
1216 }
1217
1218 /* Release all unhandled packets after compare thead exited */
1219 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1220
1221 g_queue_clear(&s->conn_list);
1222
1223 if (s->connection_track_table) {
1224 g_hash_table_destroy(s->connection_track_table);
1225 }
1226
1227 if (s->iothread) {
1228 object_unref(OBJECT(s->iothread));
1229 }
1230
1231 qemu_mutex_destroy(&event_mtx);
1232 qemu_cond_destroy(&event_complete_cond);
1233
1234 g_free(s->pri_indev);
1235 g_free(s->sec_indev);
1236 g_free(s->outdev);
1237 g_free(s->notify_dev);
1238 }
1239
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 },
1250 { }
1251 }
1252 };
1253
1254 static void register_types(void)
1255 {
1256 type_register_static(&colo_compare_info);
1257 }
1258
1259 type_init(register_types);