1
2
3
4
5
6
7
8
9
10
11
12
13
14
15#include "qemu/osdep.h"
16#include "qemu/error-report.h"
17#include "trace.h"
18#include "qemu-common.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
31#define TYPE_COLO_COMPARE "colo-compare"
32#define COLO_COMPARE(obj) \
33 OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
34
35#define COMPARE_READ_LEN_MAX NET_BUFSIZE
36#define MAX_QUEUE_SIZE 1024
37
38#define COLO_COMPARE_FREE_PRIMARY 0x01
39#define COLO_COMPARE_FREE_SECONDARY 0x02
40
41
42#define REGULAR_PACKET_CHECK_MS 3000
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66typedef struct CompareState {
67 Object parent;
68
69 char *pri_indev;
70 char *sec_indev;
71 char *outdev;
72 CharBackend chr_pri_in;
73 CharBackend chr_sec_in;
74 CharBackend chr_out;
75 SocketReadState pri_rs;
76 SocketReadState sec_rs;
77 bool vnet_hdr;
78
79
80
81
82
83 GQueue conn_list;
84
85 GHashTable *connection_track_table;
86
87 IOThread *iothread;
88 GMainContext *worker_context;
89 QEMUTimer *packet_check_timer;
90} CompareState;
91
92typedef struct CompareClass {
93 ObjectClass parent_class;
94} CompareClass;
95
96enum {
97 PRIMARY_IN = 0,
98 SECONDARY_IN,
99};
100
101static int compare_chr_send(CompareState *s,
102 const uint8_t *buf,
103 uint32_t size,
104 uint32_t vnet_hdr_len);
105
106static gint seq_sorter(Packet *a, Packet *b, gpointer data)
107{
108 struct tcphdr *atcp, *btcp;
109
110 atcp = (struct tcphdr *)(a->transport_header);
111 btcp = (struct tcphdr *)(b->transport_header);
112 return ntohl(atcp->th_seq) - ntohl(btcp->th_seq);
113}
114
115static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
116{
117 Packet *pkt = data;
118 struct tcphdr *tcphd;
119
120 tcphd = (struct tcphdr *)pkt->transport_header;
121
122 pkt->tcp_seq = ntohl(tcphd->th_seq);
123 pkt->tcp_ack = ntohl(tcphd->th_ack);
124 *max_ack = *max_ack > pkt->tcp_ack ? *max_ack : pkt->tcp_ack;
125 pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
126 + (tcphd->th_off << 2) - pkt->vnet_hdr_len;
127 pkt->payload_size = pkt->size - pkt->header_size;
128 pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
129 pkt->flags = tcphd->th_flags;
130}
131
132
133
134
135
136static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
137{
138 if (g_queue_get_length(queue) <= MAX_QUEUE_SIZE) {
139 if (pkt->ip->ip_p == IPPROTO_TCP) {
140 fill_pkt_tcp_info(pkt, max_ack);
141 g_queue_insert_sorted(queue,
142 pkt,
143 (GCompareDataFunc)seq_sorter,
144 NULL);
145 } else {
146 g_queue_push_tail(queue, pkt);
147 }
148 return 1;
149 }
150 return 0;
151}
152
153
154
155
156
157static int packet_enqueue(CompareState *s, int mode, Connection **con)
158{
159 ConnectionKey key;
160 Packet *pkt = NULL;
161 Connection *conn;
162
163 if (mode == PRIMARY_IN) {
164 pkt = packet_new(s->pri_rs.buf,
165 s->pri_rs.packet_len,
166 s->pri_rs.vnet_hdr_len);
167 } else {
168 pkt = packet_new(s->sec_rs.buf,
169 s->sec_rs.packet_len,
170 s->sec_rs.vnet_hdr_len);
171 }
172
173 if (parse_packet_early(pkt)) {
174 packet_destroy(pkt, NULL);
175 pkt = NULL;
176 return -1;
177 }
178 fill_connection_key(pkt, &key);
179
180 conn = connection_get(s->connection_track_table,
181 &key,
182 &s->conn_list);
183
184 if (!conn->processing) {
185 g_queue_push_tail(&s->conn_list, conn);
186 conn->processing = true;
187 }
188
189 if (mode == PRIMARY_IN) {
190 if (!colo_insert_packet(&conn->primary_list, pkt, &conn->pack)) {
191 error_report("colo compare primary queue size too big,"
192 "drop packet");
193 }
194 } else {
195 if (!colo_insert_packet(&conn->secondary_list, pkt, &conn->sack)) {
196 error_report("colo compare secondary queue size too big,"
197 "drop packet");
198 }
199 }
200 *con = conn;
201
202 return 0;
203}
204
205static inline bool after(uint32_t seq1, uint32_t seq2)
206{
207 return (int32_t)(seq1 - seq2) > 0;
208}
209
210static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
211{
212 int ret;
213 ret = compare_chr_send(s,
214 pkt->data,
215 pkt->size,
216 pkt->vnet_hdr_len);
217 if (ret < 0) {
218 error_report("colo send primary packet failed");
219 }
220 trace_colo_compare_main("packet same and release packet");
221 packet_destroy(pkt, NULL);
222}
223
224
225
226
227
228
229
230
231static int colo_compare_packet_payload(Packet *ppkt,
232 Packet *spkt,
233 uint16_t poffset,
234 uint16_t soffset,
235 uint16_t len)
236
237{
238 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
239 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
240
241 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
242 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
243 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
244 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
245
246 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
247 pri_ip_dst, spkt->size,
248 sec_ip_src, sec_ip_dst);
249 }
250
251 return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
252}
253
254
255
256
257
258
259static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
260 int8_t *mark, uint32_t max_ack)
261{
262 *mark = 0;
263
264 if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
265 if (colo_compare_packet_payload(ppkt, spkt,
266 ppkt->header_size, spkt->header_size,
267 ppkt->payload_size)) {
268 *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
269 return true;
270 }
271 }
272 if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
273 if (colo_compare_packet_payload(ppkt, spkt,
274 ppkt->header_size, spkt->header_size,
275 ppkt->payload_size)) {
276 *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
277 return true;
278 }
279 }
280
281
282 if (!after(ppkt->seq_end, spkt->seq_end)) {
283 if (colo_compare_packet_payload(ppkt, spkt,
284 ppkt->header_size + ppkt->offset,
285 spkt->header_size + spkt->offset,
286 ppkt->payload_size - ppkt->offset)) {
287 if (!after(ppkt->tcp_ack, max_ack)) {
288 *mark = COLO_COMPARE_FREE_PRIMARY;
289 spkt->offset += ppkt->payload_size - ppkt->offset;
290 return true;
291 } else {
292
293
294
295 return false;
296 }
297 }
298 } else {
299
300
301
302 if (colo_compare_packet_payload(ppkt, spkt,
303 ppkt->header_size + ppkt->offset,
304 spkt->header_size + spkt->offset,
305 spkt->payload_size - spkt->offset)) {
306 *mark = COLO_COMPARE_FREE_SECONDARY;
307 ppkt->offset += spkt->payload_size - spkt->offset;
308 return true;
309 }
310 }
311
312 return false;
313}
314
315static void colo_compare_tcp(CompareState *s, Connection *conn)
316{
317 Packet *ppkt = NULL, *spkt = NULL;
318 int8_t mark;
319
320
321
322
323
324
325
326
327
328
329
330
331 uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
332
333pri:
334 if (g_queue_is_empty(&conn->primary_list)) {
335 return;
336 }
337 ppkt = g_queue_pop_head(&conn->primary_list);
338sec:
339 if (g_queue_is_empty(&conn->secondary_list)) {
340 g_queue_push_head(&conn->primary_list, ppkt);
341 return;
342 }
343 spkt = g_queue_pop_head(&conn->secondary_list);
344
345 if (ppkt->tcp_seq == ppkt->seq_end) {
346 colo_release_primary_pkt(s, ppkt);
347 ppkt = NULL;
348 }
349
350 if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
351 trace_colo_compare_main("pri: this packet has compared");
352 colo_release_primary_pkt(s, ppkt);
353 ppkt = NULL;
354 }
355
356 if (spkt->tcp_seq == spkt->seq_end) {
357 packet_destroy(spkt, NULL);
358 if (!ppkt) {
359 goto pri;
360 } else {
361 goto sec;
362 }
363 } else {
364 if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
365 trace_colo_compare_main("sec: this packet has compared");
366 packet_destroy(spkt, NULL);
367 if (!ppkt) {
368 goto pri;
369 } else {
370 goto sec;
371 }
372 }
373 if (!ppkt) {
374 g_queue_push_head(&conn->secondary_list, spkt);
375 goto pri;
376 }
377 }
378
379 if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
380 trace_colo_compare_tcp_info("pri",
381 ppkt->tcp_seq, ppkt->tcp_ack,
382 ppkt->header_size, ppkt->payload_size,
383 ppkt->offset, ppkt->flags);
384
385 trace_colo_compare_tcp_info("sec",
386 spkt->tcp_seq, spkt->tcp_ack,
387 spkt->header_size, spkt->payload_size,
388 spkt->offset, spkt->flags);
389
390 if (mark == COLO_COMPARE_FREE_PRIMARY) {
391 conn->compare_seq = ppkt->seq_end;
392 colo_release_primary_pkt(s, ppkt);
393 g_queue_push_head(&conn->secondary_list, spkt);
394 goto pri;
395 }
396 if (mark == COLO_COMPARE_FREE_SECONDARY) {
397 conn->compare_seq = spkt->seq_end;
398 packet_destroy(spkt, NULL);
399 goto sec;
400 }
401 if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
402 conn->compare_seq = ppkt->seq_end;
403 colo_release_primary_pkt(s, ppkt);
404 packet_destroy(spkt, NULL);
405 goto pri;
406 }
407 } else {
408 g_queue_push_head(&conn->primary_list, ppkt);
409 g_queue_push_head(&conn->secondary_list, spkt);
410
411 qemu_hexdump((char *)ppkt->data, stderr,
412 "colo-compare ppkt", ppkt->size);
413 qemu_hexdump((char *)spkt->data, stderr,
414 "colo-compare spkt", spkt->size);
415
416
417
418
419
420 }
421}
422
423
424
425
426
427
428static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
429{
430 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
431 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
432
433 trace_colo_compare_main("compare udp");
434
435
436
437
438
439
440
441
442
443
444
445 if (ppkt->size != spkt->size) {
446 trace_colo_compare_main("UDP: payload size of packets are different");
447 return -1;
448 }
449 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
450 ppkt->size - offset)) {
451 trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
452 trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
453 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
454 qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
455 ppkt->size);
456 qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
457 spkt->size);
458 }
459 return -1;
460 } else {
461 return 0;
462 }
463}
464
465
466
467
468
469static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
470{
471 uint16_t network_header_length = ppkt->ip->ip_hl << 2;
472 uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
473
474 trace_colo_compare_main("compare icmp");
475
476
477
478
479
480
481
482
483
484
485
486 if (ppkt->size != spkt->size) {
487 trace_colo_compare_main("ICMP: payload size of packets are different");
488 return -1;
489 }
490 if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
491 ppkt->size - offset)) {
492 trace_colo_compare_icmp_miscompare("primary pkt size",
493 ppkt->size);
494 trace_colo_compare_icmp_miscompare("Secondary pkt size",
495 spkt->size);
496 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
497 qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
498 ppkt->size);
499 qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
500 spkt->size);
501 }
502 return -1;
503 } else {
504 return 0;
505 }
506}
507
508
509
510
511
512static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
513{
514 uint16_t offset = ppkt->vnet_hdr_len;
515
516 trace_colo_compare_main("compare other");
517 if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
518 char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
519
520 strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
521 strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
522 strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
523 strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
524
525 trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
526 pri_ip_dst, spkt->size,
527 sec_ip_src, sec_ip_dst);
528 }
529
530 if (ppkt->size != spkt->size) {
531 trace_colo_compare_main("Other: payload size of packets are different");
532 return -1;
533 }
534 return colo_compare_packet_payload(ppkt, spkt, offset, offset,
535 ppkt->size - offset);
536}
537
538static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
539{
540 int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
541
542 if ((now - pkt->creation_ms) > (*check_time)) {
543 trace_colo_old_packet_check_found(pkt->creation_ms);
544 return 0;
545 } else {
546 return 1;
547 }
548}
549
550static int colo_old_packet_check_one_conn(Connection *conn,
551 void *user_data)
552{
553 GList *result = NULL;
554 int64_t check_time = REGULAR_PACKET_CHECK_MS;
555
556 result = g_queue_find_custom(&conn->primary_list,
557 &check_time,
558 (GCompareFunc)colo_old_packet_check_one);
559
560 if (result) {
561
562
563
564
565
566 return 0;
567 }
568
569 return 1;
570}
571
572
573
574
575
576
577static void colo_old_packet_check(void *opaque)
578{
579 CompareState *s = opaque;
580
581
582
583
584
585 g_queue_find_custom(&s->conn_list, NULL,
586 (GCompareFunc)colo_old_packet_check_one_conn);
587}
588
589static void colo_compare_packet(CompareState *s, Connection *conn,
590 int (*HandlePacket)(Packet *spkt,
591 Packet *ppkt))
592{
593 Packet *pkt = NULL;
594 GList *result = NULL;
595
596 while (!g_queue_is_empty(&conn->primary_list) &&
597 !g_queue_is_empty(&conn->secondary_list)) {
598 pkt = g_queue_pop_head(&conn->primary_list);
599 result = g_queue_find_custom(&conn->secondary_list,
600 pkt, (GCompareFunc)HandlePacket);
601
602 if (result) {
603 colo_release_primary_pkt(s, pkt);
604 g_queue_remove(&conn->secondary_list, result->data);
605 } else {
606
607
608
609
610
611 trace_colo_compare_main("packet different");
612 g_queue_push_head(&conn->primary_list, pkt);
613
614 break;
615 }
616 }
617}
618
619
620
621
622
623
624
625static void colo_compare_connection(void *opaque, void *user_data)
626{
627 CompareState *s = user_data;
628 Connection *conn = opaque;
629
630 switch (conn->ip_proto) {
631 case IPPROTO_TCP:
632 colo_compare_tcp(s, conn);
633 break;
634 case IPPROTO_UDP:
635 colo_compare_packet(s, conn, colo_packet_compare_udp);
636 break;
637 case IPPROTO_ICMP:
638 colo_compare_packet(s, conn, colo_packet_compare_icmp);
639 break;
640 default:
641 colo_compare_packet(s, conn, colo_packet_compare_other);
642 break;
643 }
644}
645
646static int compare_chr_send(CompareState *s,
647 const uint8_t *buf,
648 uint32_t size,
649 uint32_t vnet_hdr_len)
650{
651 int ret = 0;
652 uint32_t len = htonl(size);
653
654 if (!size) {
655 return 0;
656 }
657
658 ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)&len, sizeof(len));
659 if (ret != sizeof(len)) {
660 goto err;
661 }
662
663 if (s->vnet_hdr) {
664
665
666
667
668 len = htonl(vnet_hdr_len);
669 ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)&len, sizeof(len));
670 if (ret != sizeof(len)) {
671 goto err;
672 }
673 }
674
675 ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)buf, size);
676 if (ret != size) {
677 goto err;
678 }
679
680 return 0;
681
682err:
683 return ret < 0 ? ret : -EIO;
684}
685
686static int compare_chr_can_read(void *opaque)
687{
688 return COMPARE_READ_LEN_MAX;
689}
690
691
692
693
694
695static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
696{
697 CompareState *s = COLO_COMPARE(opaque);
698 int ret;
699
700 ret = net_fill_rstate(&s->pri_rs, buf, size);
701 if (ret == -1) {
702 qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
703 NULL, NULL, true);
704 error_report("colo-compare primary_in error");
705 }
706}
707
708
709
710
711
712static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
713{
714 CompareState *s = COLO_COMPARE(opaque);
715 int ret;
716
717 ret = net_fill_rstate(&s->sec_rs, buf, size);
718 if (ret == -1) {
719 qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
720 NULL, NULL, true);
721 error_report("colo-compare secondary_in error");
722 }
723}
724
725
726
727
728
729static void check_old_packet_regular(void *opaque)
730{
731 CompareState *s = opaque;
732
733
734 colo_old_packet_check(s);
735 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
736 REGULAR_PACKET_CHECK_MS);
737}
738
739static void colo_compare_timer_init(CompareState *s)
740{
741 AioContext *ctx = iothread_get_aio_context(s->iothread);
742
743 s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
744 SCALE_MS, check_old_packet_regular,
745 s);
746 timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
747 REGULAR_PACKET_CHECK_MS);
748}
749
750static void colo_compare_timer_del(CompareState *s)
751{
752 if (s->packet_check_timer) {
753 timer_del(s->packet_check_timer);
754 timer_free(s->packet_check_timer);
755 s->packet_check_timer = NULL;
756 }
757 }
758
759static void colo_compare_iothread(CompareState *s)
760{
761 object_ref(OBJECT(s->iothread));
762 s->worker_context = iothread_get_g_main_context(s->iothread);
763
764 qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
765 compare_pri_chr_in, NULL, NULL,
766 s, s->worker_context, true);
767 qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
768 compare_sec_chr_in, NULL, NULL,
769 s, s->worker_context, true);
770
771 colo_compare_timer_init(s);
772}
773
774static char *compare_get_pri_indev(Object *obj, Error **errp)
775{
776 CompareState *s = COLO_COMPARE(obj);
777
778 return g_strdup(s->pri_indev);
779}
780
781static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
782{
783 CompareState *s = COLO_COMPARE(obj);
784
785 g_free(s->pri_indev);
786 s->pri_indev = g_strdup(value);
787}
788
789static char *compare_get_sec_indev(Object *obj, Error **errp)
790{
791 CompareState *s = COLO_COMPARE(obj);
792
793 return g_strdup(s->sec_indev);
794}
795
796static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
797{
798 CompareState *s = COLO_COMPARE(obj);
799
800 g_free(s->sec_indev);
801 s->sec_indev = g_strdup(value);
802}
803
804static char *compare_get_outdev(Object *obj, Error **errp)
805{
806 CompareState *s = COLO_COMPARE(obj);
807
808 return g_strdup(s->outdev);
809}
810
811static void compare_set_outdev(Object *obj, const char *value, Error **errp)
812{
813 CompareState *s = COLO_COMPARE(obj);
814
815 g_free(s->outdev);
816 s->outdev = g_strdup(value);
817}
818
819static bool compare_get_vnet_hdr(Object *obj, Error **errp)
820{
821 CompareState *s = COLO_COMPARE(obj);
822
823 return s->vnet_hdr;
824}
825
826static void compare_set_vnet_hdr(Object *obj,
827 bool value,
828 Error **errp)
829{
830 CompareState *s = COLO_COMPARE(obj);
831
832 s->vnet_hdr = value;
833}
834
835static void compare_pri_rs_finalize(SocketReadState *pri_rs)
836{
837 CompareState *s = container_of(pri_rs, CompareState, pri_rs);
838 Connection *conn = NULL;
839
840 if (packet_enqueue(s, PRIMARY_IN, &conn)) {
841 trace_colo_compare_main("primary: unsupported packet in");
842 compare_chr_send(s,
843 pri_rs->buf,
844 pri_rs->packet_len,
845 pri_rs->vnet_hdr_len);
846 } else {
847
848 colo_compare_connection(conn, s);
849 }
850}
851
852static void compare_sec_rs_finalize(SocketReadState *sec_rs)
853{
854 CompareState *s = container_of(sec_rs, CompareState, sec_rs);
855 Connection *conn = NULL;
856
857 if (packet_enqueue(s, SECONDARY_IN, &conn)) {
858 trace_colo_compare_main("secondary: unsupported packet in");
859 } else {
860
861 colo_compare_connection(conn, s);
862 }
863}
864
865
866
867
868
869
870static int find_and_check_chardev(Chardev **chr,
871 char *chr_name,
872 Error **errp)
873{
874 *chr = qemu_chr_find(chr_name);
875 if (*chr == NULL) {
876 error_setg(errp, "Device '%s' not found",
877 chr_name);
878 return 1;
879 }
880
881 if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
882 error_setg(errp, "chardev \"%s\" is not reconnectable",
883 chr_name);
884 return 1;
885 }
886
887 return 0;
888}
889
890
891
892
893
894static void colo_compare_complete(UserCreatable *uc, Error **errp)
895{
896 CompareState *s = COLO_COMPARE(uc);
897 Chardev *chr;
898
899 if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
900 error_setg(errp, "colo compare needs 'primary_in' ,"
901 "'secondary_in','outdev','iothread' property set");
902 return;
903 } else if (!strcmp(s->pri_indev, s->outdev) ||
904 !strcmp(s->sec_indev, s->outdev) ||
905 !strcmp(s->pri_indev, s->sec_indev)) {
906 error_setg(errp, "'indev' and 'outdev' could not be same "
907 "for compare module");
908 return;
909 }
910
911 if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
912 !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
913 return;
914 }
915
916 if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
917 !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
918 return;
919 }
920
921 if (find_and_check_chardev(&chr, s->outdev, errp) ||
922 !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
923 return;
924 }
925
926 net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
927 net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
928
929 g_queue_init(&s->conn_list);
930
931 s->connection_track_table = g_hash_table_new_full(connection_key_hash,
932 connection_key_equal,
933 g_free,
934 connection_destroy);
935
936 colo_compare_iothread(s);
937 return;
938}
939
940static void colo_flush_packets(void *opaque, void *user_data)
941{
942 CompareState *s = user_data;
943 Connection *conn = opaque;
944 Packet *pkt = NULL;
945
946 while (!g_queue_is_empty(&conn->primary_list)) {
947 pkt = g_queue_pop_head(&conn->primary_list);
948 compare_chr_send(s,
949 pkt->data,
950 pkt->size,
951 pkt->vnet_hdr_len);
952 packet_destroy(pkt, NULL);
953 }
954 while (!g_queue_is_empty(&conn->secondary_list)) {
955 pkt = g_queue_pop_head(&conn->secondary_list);
956 packet_destroy(pkt, NULL);
957 }
958}
959
960static void colo_compare_class_init(ObjectClass *oc, void *data)
961{
962 UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
963
964 ucc->complete = colo_compare_complete;
965}
966
967static void colo_compare_init(Object *obj)
968{
969 CompareState *s = COLO_COMPARE(obj);
970
971 object_property_add_str(obj, "primary_in",
972 compare_get_pri_indev, compare_set_pri_indev,
973 NULL);
974 object_property_add_str(obj, "secondary_in",
975 compare_get_sec_indev, compare_set_sec_indev,
976 NULL);
977 object_property_add_str(obj, "outdev",
978 compare_get_outdev, compare_set_outdev,
979 NULL);
980 object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
981 (Object **)&s->iothread,
982 object_property_allow_set_link,
983 OBJ_PROP_LINK_STRONG, NULL);
984
985 s->vnet_hdr = false;
986 object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
987 compare_set_vnet_hdr, NULL);
988}
989
990static void colo_compare_finalize(Object *obj)
991{
992 CompareState *s = COLO_COMPARE(obj);
993
994 qemu_chr_fe_deinit(&s->chr_pri_in, false);
995 qemu_chr_fe_deinit(&s->chr_sec_in, false);
996 qemu_chr_fe_deinit(&s->chr_out, false);
997 if (s->iothread) {
998 colo_compare_timer_del(s);
999 }
1000
1001 g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1002
1003 g_queue_clear(&s->conn_list);
1004
1005 if (s->connection_track_table) {
1006 g_hash_table_destroy(s->connection_track_table);
1007 }
1008
1009 if (s->iothread) {
1010 object_unref(OBJECT(s->iothread));
1011 }
1012 g_free(s->pri_indev);
1013 g_free(s->sec_indev);
1014 g_free(s->outdev);
1015}
1016
1017static const TypeInfo colo_compare_info = {
1018 .name = TYPE_COLO_COMPARE,
1019 .parent = TYPE_OBJECT,
1020 .instance_size = sizeof(CompareState),
1021 .instance_init = colo_compare_init,
1022 .instance_finalize = colo_compare_finalize,
1023 .class_size = sizeof(CompareClass),
1024 .class_init = colo_compare_class_init,
1025 .interfaces = (InterfaceInfo[]) {
1026 { TYPE_USER_CREATABLE },
1027 { }
1028 }
1029};
1030
1031static void register_types(void)
1032{
1033 type_register_static(&colo_compare_info);
1034}
1035
1036type_init(register_types);
1037