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 #include "block/aio-wait.h"
  36 #include "qemu/coroutine.h"
  37
  38 #define TYPE_COLO_COMPARE "colo-compare"
  39 typedef struct CompareState CompareState;
  40 DECLARE_INSTANCE_CHECKER(CompareState, COLO_COMPARE,
  41                          TYPE_COLO_COMPARE)
  42
  43 static QTAILQ_HEAD(, CompareState) net_compares =
  44        QTAILQ_HEAD_INITIALIZER(net_compares);
  45
  46 static NotifierList colo_compare_notifiers =
  47     NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
  48
  49 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
  50 #define MAX_QUEUE_SIZE 1024
  51
  52 #define COLO_COMPARE_FREE_PRIMARY     0x01
  53 #define COLO_COMPARE_FREE_SECONDARY   0x02
  54
  55 #define REGULAR_PACKET_CHECK_MS 3000
  56 #define DEFAULT_TIME_OUT_MS 3000
  57
  58 /* #define DEBUG_COLO_PACKETS */
  59
  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;
  66
  67 /*
  68  *  + CompareState ++
  69  *  |               |
  70  *  +---------------+   +---------------+         +---------------+
  71  *  |   conn list   + - >      conn     + ------- >      conn     + -- > ......
  72  *  +---------------+   +---------------+         +---------------+
  73  *  |               |     |           |             |          |
  74  *  +---------------+ +---v----+  +---v----+    +---v----+ +---v----+
  75  *                    |primary |  |secondary    |primary | |secondary
  76  *                    |packet  |  |packet  +    |packet  | |packet  +
  77  *                    +--------+  +--------+    +--------+ +--------+
  78  *                        |           |             |          |
  79  *                    +---v----+  +---v----+    +---v----+ +---v----+
  80  *                    |primary |  |secondary    |primary | |secondary
  81  *                    |packet  |  |packet  +    |packet  | |packet  +
  82  *                    +--------+  +--------+    +--------+ +--------+
  83  *                        |           |             |          |
  84  *                    +---v----+  +---v----+    +---v----+ +---v----+
  85  *                    |primary |  |secondary    |primary | |secondary
  86  *                    |packet  |  |packet  +    |packet  | |packet  +
  87  *                    +--------+  +--------+    +--------+ +--------+
  88  */
  89
  90 typedef struct SendCo {
  91     Coroutine *co;
  92     struct CompareState *s;
  93     CharBackend *chr;
  94     GQueue send_list;
  95     bool notify_remote_frame;
  96     bool done;
  97     int ret;
  98 } SendCo;
  99
 100 typedef struct SendEntry {
 101     uint32_t size;
 102     uint32_t vnet_hdr_len;
 103     uint8_t *buf;
 104 } SendEntry;
 105
 106 struct CompareState {
 107     Object parent;
 108
 109     char *pri_indev;
 110     char *sec_indev;
 111     char *outdev;
 112     char *notify_dev;
 113     CharBackend chr_pri_in;
 114     CharBackend chr_sec_in;
 115     CharBackend chr_out;
 116     CharBackend chr_notify_dev;
 117     SocketReadState pri_rs;
 118     SocketReadState sec_rs;
 119     SocketReadState notify_rs;
 120     SendCo out_sendco;
 121     SendCo notify_sendco;
 122     bool vnet_hdr;
 123     uint32_t compare_timeout;
 124     uint32_t expired_scan_cycle;
 125
 126     /*
 127      * Record the connection that through the NIC
 128      * Element type: Connection
 129      */
 130     GQueue conn_list;
 131     /* Record the connection without repetition */
 132     GHashTable *connection_track_table;
 133
 134     IOThread *iothread;
 135     GMainContext *worker_context;
 136     QEMUTimer *packet_check_timer;
 137
 138     QEMUBH *event_bh;
 139     enum colo_event event;
 140
 141     QTAILQ_ENTRY(CompareState) next;
 142 };
 143
 144 typedef struct CompareClass {
 145     ObjectClass parent_class;
 146 } CompareClass;
 147
 148 enum {
 149     PRIMARY_IN = 0,
 150     SECONDARY_IN,
 151 };
 152
 153 static const char *colo_mode[] = {
 154     [PRIMARY_IN] = "primary",
 155     [SECONDARY_IN] = "secondary",
 156 };
 157
 158 static int compare_chr_send(CompareState *s,
 159                             uint8_t *buf,
 160                             uint32_t size,
 161                             uint32_t vnet_hdr_len,
 162                             bool notify_remote_frame,
 163                             bool zero_copy);
 164
 165 static bool packet_matches_str(const char *str,
 166                                const uint8_t *buf,
 167                                uint32_t packet_len)
 168 {
 169     if (packet_len != strlen(str)) {
 170         return false;
 171     }
 172
 173     return !memcmp(str, buf, strlen(str));
 174 }
 175
 176 static void notify_remote_frame(CompareState *s)
 177 {
 178     char msg[] = "DO_CHECKPOINT";
 179     int ret = 0;
 180
 181     ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
 182     if (ret < 0) {
 183         error_report("Notify Xen COLO-frame failed");
 184     }
 185 }
 186
 187 static void colo_compare_inconsistency_notify(CompareState *s)
 188 {
 189     if (s->notify_dev) {
 190         notify_remote_frame(s);
 191     } else {
 192         notifier_list_notify(&colo_compare_notifiers,
 193                              migrate_get_current());
 194     }
 195 }
 196
 197 /* Use restricted to colo_insert_packet() */
 198 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
 199 {
 200     return a->tcp_seq - b->tcp_seq;
 201 }
 202
 203 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
 204 {
 205     Packet *pkt = data;
 206     struct tcp_hdr *tcphd;
 207
 208     tcphd = (struct tcp_hdr *)pkt->transport_header;
 209
 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;
 218 }
 219
 220 /*
 221  * Return 1 on success, if return 0 means the
 222  * packet will be dropped
 223  */
 224 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
 225 {
 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,
 230                                   pkt,
 231                                   (GCompareDataFunc)seq_sorter,
 232                                   NULL);
 233         } else {
 234             g_queue_push_tail(queue, pkt);
 235         }
 236         return 1;
 237     }
 238     return 0;
 239 }
 240
 241 /*
 242  * Return 0 on success, if return -1 means the pkt
 243  * is unsupported(arp and ipv6) and will be sent later
 244  */
 245 static int packet_enqueue(CompareState *s, int mode, Connection **con)
 246 {
 247     ConnectionKey key;
 248     Packet *pkt = NULL;
 249     Connection *conn;
 250     int ret;
 251
 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);
 256     } else {
 257         pkt = packet_new(s->sec_rs.buf,
 258                          s->sec_rs.packet_len,
 259                          s->sec_rs.vnet_hdr_len);
 260     }
 261
 262     if (parse_packet_early(pkt)) {
 263         packet_destroy(pkt, NULL);
 264         pkt = NULL;
 265         return -1;
 266     }
 267     fill_connection_key(pkt, &key);
 268
 269     conn = connection_get(s->connection_track_table,
 270                           &key,
 271                           &s->conn_list);
 272
 273     if (!conn->processing) {
 274         g_queue_push_tail(&s->conn_list, conn);
 275         conn->processing = true;
 276     }
 277
 278     if (mode == PRIMARY_IN) {
 279         ret = colo_insert_packet(&conn->primary_list, pkt, &conn->pack);
 280     } else {
 281         ret = colo_insert_packet(&conn->secondary_list, pkt, &conn->sack);
 282     }
 283
 284     if (!ret) {
 285         trace_colo_compare_drop_packet(colo_mode[mode],
 286             "queue size too big, drop packet");
 287         packet_destroy(pkt, NULL);
 288         pkt = NULL;
 289     }
 290
 291     *con = conn;
 292
 293     return 0;
 294 }
 295
 296 static inline bool after(uint32_t seq1, uint32_t seq2)
 297 {
 298         return (int32_t)(seq1 - seq2) > 0;
 299 }
 300
 301 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
 302 {
 303     int ret;
 304     ret = compare_chr_send(s,
 305                            pkt->data,
 306                            pkt->size,
 307                            pkt->vnet_hdr_len,
 308                            false,
 309                            true);
 310     if (ret < 0) {
 311         error_report("colo send primary packet failed");
 312     }
 313     trace_colo_compare_main("packet same and release packet");
 314     packet_destroy_partial(pkt, NULL);
 315 }
 316
 317 /*
 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
 323  */
 324 static int colo_compare_packet_payload(Packet *ppkt,
 325                                        Packet *spkt,
 326                                        uint16_t poffset,
 327                                        uint16_t soffset,
 328                                        uint16_t len)
 329
 330 {
 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];
 333
 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));
 338
 339         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 340                                    pri_ip_dst, spkt->size,
 341                                    sec_ip_src, sec_ip_dst);
 342     }
 343
 344     return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
 345 }
 346
 347 /*
 348  * return true means that the payload is consist and
 349  * need to make the next comparison, false means do
 350  * the checkpoint
 351 */
 352 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
 353                               int8_t *mark, uint32_t max_ack)
 354 {
 355     *mark = 0;
 356
 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;
 362             return true;
 363         }
 364     }
 365
 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;
 375                 return true;
 376             } else {
 377                 /* secondary guest hasn't ack the data, don't send
 378                  * out this packet
 379                  */
 380                 return false;
 381             }
 382         }
 383     } else {
 384         /* primary packet is longer than secondary packet, compare
 385          * the same part and mark the primary packet offset
 386          */
 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;
 393             return true;
 394         }
 395     }
 396
 397     return false;
 398 }
 399
 400 static void colo_compare_tcp(CompareState *s, Connection *conn)
 401 {
 402     Packet *ppkt = NULL, *spkt = NULL;
 403     int8_t mark;
 404
 405     /*
 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.
 415     */
 416     uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
 417
 418 pri:
 419     if (g_queue_is_empty(&conn->primary_list)) {
 420         return;
 421     }
 422     ppkt = g_queue_pop_head(&conn->primary_list);
 423 sec:
 424     if (g_queue_is_empty(&conn->secondary_list)) {
 425         g_queue_push_head(&conn->primary_list, ppkt);
 426         return;
 427     }
 428     spkt = g_queue_pop_head(&conn->secondary_list);
 429
 430     if (ppkt->tcp_seq == ppkt->seq_end) {
 431         colo_release_primary_pkt(s, ppkt);
 432         ppkt = NULL;
 433     }
 434
 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);
 438         ppkt = NULL;
 439     }
 440
 441     if (spkt->tcp_seq == spkt->seq_end) {
 442         packet_destroy(spkt, NULL);
 443         if (!ppkt) {
 444             goto pri;
 445         } else {
 446             goto sec;
 447         }
 448     } else {
 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);
 452             if (!ppkt) {
 453                 goto pri;
 454             } else {
 455                 goto sec;
 456             }
 457         }
 458         if (!ppkt) {
 459             g_queue_push_head(&conn->secondary_list, spkt);
 460             goto pri;
 461         }
 462     }
 463
 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);
 469
 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);
 474
 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);
 479             goto pri;
 480         }
 481         if (mark == COLO_COMPARE_FREE_SECONDARY) {
 482             conn->compare_seq = spkt->seq_end;
 483             packet_destroy(spkt, NULL);
 484             goto sec;
 485         }
 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);
 490             goto pri;
 491         }
 492     } else {
 493         g_queue_push_head(&conn->primary_list, ppkt);
 494         g_queue_push_head(&conn->secondary_list, spkt);
 495
 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);
 499 #endif
 500
 501         colo_compare_inconsistency_notify(s);
 502     }
 503 }
 504
 505
 506 /*
 507  * Called from the compare thread on the primary
 508  * for compare udp packet
 509  */
 510 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
 511 {
 512     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
 513     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
 514
 515     trace_colo_compare_main("compare udp");
 516
 517     /*
 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.
 526      */
 527     if (ppkt->size != spkt->size) {
 528         trace_colo_compare_main("UDP: payload size of packets are different");
 529         return -1;
 530     }
 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);
 538 #endif
 539         return -1;
 540     } else {
 541         return 0;
 542     }
 543 }
 544
 545 /*
 546  * Called from the compare thread on the primary
 547  * for compare icmp packet
 548  */
 549 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
 550 {
 551     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
 552     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
 553
 554     trace_colo_compare_main("compare icmp");
 555
 556     /*
 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.
 565      */
 566     if (ppkt->size != spkt->size) {
 567         trace_colo_compare_main("ICMP: payload size of packets are different");
 568         return -1;
 569     }
 570     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
 571                                     ppkt->size - offset)) {
 572         trace_colo_compare_icmp_miscompare("primary pkt size",
 573                                            ppkt->size);
 574         trace_colo_compare_icmp_miscompare("Secondary pkt size",
 575                                            spkt->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);
 579 #endif
 580         return -1;
 581     } else {
 582         return 0;
 583     }
 584 }
 585
 586 /*
 587  * Called from the compare thread on the primary
 588  * for compare other packet
 589  */
 590 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
 591 {
 592     uint16_t offset = ppkt->vnet_hdr_len;
 593
 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];
 597
 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));
 602
 603         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 604                                    pri_ip_dst, spkt->size,
 605                                    sec_ip_src, sec_ip_dst);
 606     }
 607
 608     if (ppkt->size != spkt->size) {
 609         trace_colo_compare_main("Other: payload size of packets are different");
 610         return -1;
 611     }
 612     return colo_compare_packet_payload(ppkt, spkt, offset, offset,
 613                                        ppkt->size - offset);
 614 }
 615
 616 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
 617 {
 618     int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 619
 620     if ((now - pkt->creation_ms) > (*check_time)) {
 621         trace_colo_old_packet_check_found(pkt->creation_ms);
 622         return 0;
 623     } else {
 624         return 1;
 625     }
 626 }
 627
 628 void colo_compare_register_notifier(Notifier *notify)
 629 {
 630     notifier_list_add(&colo_compare_notifiers, notify);
 631 }
 632
 633 void colo_compare_unregister_notifier(Notifier *notify)
 634 {
 635     notifier_remove(notify);
 636 }
 637
 638 static int colo_old_packet_check_one_conn(Connection *conn,
 639                                           CompareState *s)
 640 {
 641     GList *result = NULL;
 642
 643     result = g_queue_find_custom(&conn->primary_list,
 644                                  &s->compare_timeout,
 645                                  (GCompareFunc)colo_old_packet_check_one);
 646
 647     if (result) {
 648         /* Do checkpoint will flush old packet */
 649         colo_compare_inconsistency_notify(s);
 650         return 0;
 651     }
 652
 653     return 1;
 654 }
 655
 656 /*
 657  * Look for old packets that the secondary hasn't matched,
 658  * if we have some then we have to checkpoint to wake
 659  * the secondary up.
 660  */
 661 static void colo_old_packet_check(void *opaque)
 662 {
 663     CompareState *s = opaque;
 664
 665     /*
 666      * If we find one old packet, stop finding job and notify
 667      * COLO frame do checkpoint.
 668      */
 669     g_queue_find_custom(&s->conn_list, s,
 670                         (GCompareFunc)colo_old_packet_check_one_conn);
 671 }
 672
 673 static void colo_compare_packet(CompareState *s, Connection *conn,
 674                                 int (*HandlePacket)(Packet *spkt,
 675                                 Packet *ppkt))
 676 {
 677     Packet *pkt = NULL;
 678     GList *result = NULL;
 679
 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);
 685
 686         if (result) {
 687             colo_release_primary_pkt(s, pkt);
 688             g_queue_remove(&conn->secondary_list, result->data);
 689         } else {
 690             /*
 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.
 695              */
 696             trace_colo_compare_main("packet different");
 697             g_queue_push_head(&conn->primary_list, pkt);
 698
 699             colo_compare_inconsistency_notify(s);
 700             break;
 701         }
 702     }
 703 }
 704
 705 /*
 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
 709  * queued to it.
 710  */
 711 static void colo_compare_connection(void *opaque, void *user_data)
 712 {
 713     CompareState *s = user_data;
 714     Connection *conn = opaque;
 715
 716     switch (conn->ip_proto) {
 717     case IPPROTO_TCP:
 718         colo_compare_tcp(s, conn);
 719         break;
 720     case IPPROTO_UDP:
 721         colo_compare_packet(s, conn, colo_packet_compare_udp);
 722         break;
 723     case IPPROTO_ICMP:
 724         colo_compare_packet(s, conn, colo_packet_compare_icmp);
 725         break;
 726     default:
 727         colo_compare_packet(s, conn, colo_packet_compare_other);
 728         break;
 729     }
 730 }
 731
 732 static void coroutine_fn _compare_chr_send(void *opaque)
 733 {
 734     SendCo *sendco = opaque;
 735     CompareState *s = sendco->s;
 736     int ret = 0;
 737
 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);
 741
 742         ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
 743
 744         if (ret != sizeof(len)) {
 745             g_free(entry->buf);
 746             g_slice_free(SendEntry, entry);
 747             goto err;
 748         }
 749
 750         if (!sendco->notify_remote_frame && s->vnet_hdr) {
 751             /*
 752              * We send vnet header len make other module(like filter-redirector)
 753              * know how to parse net packet correctly.
 754              */
 755             len = htonl(entry->vnet_hdr_len);
 756
 757             ret = qemu_chr_fe_write_all(sendco->chr,
 758                                         (uint8_t *)&len,
 759                                         sizeof(len));
 760
 761             if (ret != sizeof(len)) {
 762                 g_free(entry->buf);
 763                 g_slice_free(SendEntry, entry);
 764                 goto err;
 765             }
 766         }
 767
 768         ret = qemu_chr_fe_write_all(sendco->chr,
 769                                     (uint8_t *)entry->buf,
 770                                     entry->size);
 771
 772         if (ret != entry->size) {
 773             g_free(entry->buf);
 774             g_slice_free(SendEntry, entry);
 775             goto err;
 776         }
 777
 778         g_free(entry->buf);
 779         g_slice_free(SendEntry, entry);
 780     }
 781
 782     sendco->ret = 0;
 783     goto out;
 784
 785 err:
 786     while (!g_queue_is_empty(&sendco->send_list)) {
 787         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
 788         g_free(entry->buf);
 789         g_slice_free(SendEntry, entry);
 790     }
 791     sendco->ret = ret < 0 ? ret : -EIO;
 792 out:
 793     sendco->co = NULL;
 794     sendco->done = true;
 795     aio_wait_kick();
 796 }
 797
 798 static int compare_chr_send(CompareState *s,
 799                             uint8_t *buf,
 800                             uint32_t size,
 801                             uint32_t vnet_hdr_len,
 802                             bool notify_remote_frame,
 803                             bool zero_copy)
 804 {
 805     SendCo *sendco;
 806     SendEntry *entry;
 807
 808     if (notify_remote_frame) {
 809         sendco = &s->notify_sendco;
 810     } else {
 811         sendco = &s->out_sendco;
 812     }
 813
 814     if (!size) {
 815         return 0;
 816     }
 817
 818     entry = g_slice_new(SendEntry);
 819     entry->size = size;
 820     entry->vnet_hdr_len = vnet_hdr_len;
 821     if (zero_copy) {
 822         entry->buf = buf;
 823     } else {
 824         entry->buf = g_malloc(size);
 825         memcpy(entry->buf, buf, size);
 826     }
 827     g_queue_push_head(&sendco->send_list, entry);
 828
 829     if (sendco->done) {
 830         sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
 831         sendco->done = false;
 832         qemu_coroutine_enter(sendco->co);
 833         if (sendco->done) {
 834             /* report early errors */
 835             return sendco->ret;
 836         }
 837     }
 838
 839     /* assume success */
 840     return 0;
 841 }
 842
 843 static int compare_chr_can_read(void *opaque)
 844 {
 845     return COMPARE_READ_LEN_MAX;
 846 }
 847
 848 /*
 849  * Called from the main thread on the primary for packets
 850  * arriving over the socket from the primary.
 851  */
 852 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
 853 {
 854     CompareState *s = COLO_COMPARE(opaque);
 855     int ret;
 856
 857     ret = net_fill_rstate(&s->pri_rs, buf, size);
 858     if (ret == -1) {
 859         qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
 860                                  NULL, NULL, true);
 861         error_report("colo-compare primary_in error");
 862     }
 863 }
 864
 865 /*
 866  * Called from the main thread on the primary for packets
 867  * arriving over the socket from the secondary.
 868  */
 869 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 870 {
 871     CompareState *s = COLO_COMPARE(opaque);
 872     int ret;
 873
 874     ret = net_fill_rstate(&s->sec_rs, buf, size);
 875     if (ret == -1) {
 876         qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
 877                                  NULL, NULL, true);
 878         error_report("colo-compare secondary_in error");
 879     }
 880 }
 881
 882 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
 883 {
 884     CompareState *s = COLO_COMPARE(opaque);
 885     int ret;
 886
 887     ret = net_fill_rstate(&s->notify_rs, buf, size);
 888     if (ret == -1) {
 889         qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
 890                                  NULL, NULL, true);
 891         error_report("colo-compare notify_dev error");
 892     }
 893 }
 894
 895 /*
 896  * Check old packet regularly so it can watch for any packets
 897  * that the secondary hasn't produced equivalents of.
 898  */
 899 static void check_old_packet_regular(void *opaque)
 900 {
 901     CompareState *s = opaque;
 902
 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);
 907 }
 908
 909 /* Public API, Used for COLO frame to notify compare event */
 910 void colo_notify_compares_event(void *opaque, int event, Error **errp)
 911 {
 912     CompareState *s;
 913     qemu_mutex_lock(&colo_compare_mutex);
 914
 915     if (!colo_compare_active) {
 916         qemu_mutex_unlock(&colo_compare_mutex);
 917         return;
 918     }
 919
 920     qemu_mutex_lock(&event_mtx);
 921     QTAILQ_FOREACH(s, &net_compares, next) {
 922         s->event = event;
 923         qemu_bh_schedule(s->event_bh);
 924         event_unhandled_count++;
 925     }
 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);
 929     }
 930
 931     qemu_mutex_unlock(&event_mtx);
 932     qemu_mutex_unlock(&colo_compare_mutex);
 933 }
 934
 935 static void colo_compare_timer_init(CompareState *s)
 936 {
 937     AioContext *ctx = iothread_get_aio_context(s->iothread);
 938
 939     s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
 940                                 SCALE_MS, check_old_packet_regular,
 941                                 s);
 942     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 943               s->expired_scan_cycle);
 944 }
 945
 946 static void colo_compare_timer_del(CompareState *s)
 947 {
 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;
 952     }
 953  }
 954
 955 static void colo_flush_packets(void *opaque, void *user_data);
 956
 957 static void colo_compare_handle_event(void *opaque)
 958 {
 959     CompareState *s = opaque;
 960
 961     switch (s->event) {
 962     case COLO_EVENT_CHECKPOINT:
 963         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
 964         break;
 965     case COLO_EVENT_FAILOVER:
 966         break;
 967     default:
 968         break;
 969     }
 970
 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);
 976 }
 977
 978 static void colo_compare_iothread(CompareState *s)
 979 {
 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);
 983
 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);
 990     if (s->notify_dev) {
 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);
 994     }
 995
 996     colo_compare_timer_init(s);
 997     s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
 998 }
 999
1000 static char *compare_get_pri_indev(Object *obj, Error **errp)
1001 {
1002     CompareState *s = COLO_COMPARE(obj);
1003
1004     return g_strdup(s->pri_indev);
1005 }
1006
1007 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1008 {
1009     CompareState *s = COLO_COMPARE(obj);
1010
1011     g_free(s->pri_indev);
1012     s->pri_indev = g_strdup(value);
1013 }
1014
1015 static char *compare_get_sec_indev(Object *obj, Error **errp)
1016 {
1017     CompareState *s = COLO_COMPARE(obj);
1018
1019     return g_strdup(s->sec_indev);
1020 }
1021
1022 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1023 {
1024     CompareState *s = COLO_COMPARE(obj);
1025
1026     g_free(s->sec_indev);
1027     s->sec_indev = g_strdup(value);
1028 }
1029
1030 static char *compare_get_outdev(Object *obj, Error **errp)
1031 {
1032     CompareState *s = COLO_COMPARE(obj);
1033
1034     return g_strdup(s->outdev);
1035 }
1036
1037 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1038 {
1039     CompareState *s = COLO_COMPARE(obj);
1040
1041     g_free(s->outdev);
1042     s->outdev = g_strdup(value);
1043 }
1044
1045 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1046 {
1047     CompareState *s = COLO_COMPARE(obj);
1048
1049     return s->vnet_hdr;
1050 }
1051
1052 static void compare_set_vnet_hdr(Object *obj,
1053                                  bool value,
1054                                  Error **errp)
1055 {
1056     CompareState *s = COLO_COMPARE(obj);
1057
1058     s->vnet_hdr = value;
1059 }
1060
1061 static char *compare_get_notify_dev(Object *obj, Error **errp)
1062 {
1063     CompareState *s = COLO_COMPARE(obj);
1064
1065     return g_strdup(s->notify_dev);
1066 }
1067
1068 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1069 {
1070     CompareState *s = COLO_COMPARE(obj);
1071
1072     g_free(s->notify_dev);
1073     s->notify_dev = g_strdup(value);
1074 }
1075
1076 static void compare_get_timeout(Object *obj, Visitor *v,
1077                                 const char *name, void *opaque,
1078                                 Error **errp)
1079 {
1080     CompareState *s = COLO_COMPARE(obj);
1081     uint32_t value = s->compare_timeout;
1082
1083     visit_type_uint32(v, name, &value, errp);
1084 }
1085
1086 static void compare_set_timeout(Object *obj, Visitor *v,
1087                                 const char *name, void *opaque,
1088                                 Error **errp)
1089 {
1090     CompareState *s = COLO_COMPARE(obj);
1091     uint32_t value;
1092
1093     if (!visit_type_uint32(v, name, &value, errp)) {
1094         return;
1095     }
1096     if (!value) {
1097         error_setg(errp, "Property '%s.%s' requires a positive value",
1098                    object_get_typename(obj), name);
1099         return;
1100     }
1101     s->compare_timeout = value;
1102 }
1103
1104 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1105                                            const char *name, void *opaque,
1106                                            Error **errp)
1107 {
1108     CompareState *s = COLO_COMPARE(obj);
1109     uint32_t value = s->expired_scan_cycle;
1110
1111     visit_type_uint32(v, name, &value, errp);
1112 }
1113
1114 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1115                                            const char *name, void *opaque,
1116                                            Error **errp)
1117 {
1118     CompareState *s = COLO_COMPARE(obj);
1119     uint32_t value;
1120
1121     if (!visit_type_uint32(v, name, &value, errp)) {
1122         return;
1123     }
1124     if (!value) {
1125         error_setg(errp, "Property '%s.%s' requires a positive value",
1126                    object_get_typename(obj), name);
1127         return;
1128     }
1129     s->expired_scan_cycle = value;
1130 }
1131
1132 static void get_max_queue_size(Object *obj, Visitor *v,
1133                                const char *name, void *opaque,
1134                                Error **errp)
1135 {
1136     uint32_t value = max_queue_size;
1137
1138     visit_type_uint32(v, name, &value, errp);
1139 }
1140
1141 static void set_max_queue_size(Object *obj, Visitor *v,
1142                                const char *name, void *opaque,
1143                                Error **errp)
1144 {
1145     Error *local_err = NULL;
1146     uint32_t value;
1147
1148     visit_type_uint32(v, name, &value, &local_err);
1149     if (local_err) {
1150         goto out;
1151     }
1152     if (!value) {
1153         error_setg(&local_err, "Property '%s.%s' requires a positive value",
1154                    object_get_typename(obj), name);
1155         goto out;
1156     }
1157     max_queue_size = value;
1158
1159 out:
1160     error_propagate(errp, local_err);
1161 }
1162
1163 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1164 {
1165     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1166     Connection *conn = NULL;
1167
1168     if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1169         trace_colo_compare_main("primary: unsupported packet in");
1170         compare_chr_send(s,
1171                          pri_rs->buf,
1172                          pri_rs->packet_len,
1173                          pri_rs->vnet_hdr_len,
1174                          false,
1175                          false);
1176     } else {
1177         /* compare packet in the specified connection */
1178         colo_compare_connection(conn, s);
1179     }
1180 }
1181
1182 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1183 {
1184     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1185     Connection *conn = NULL;
1186
1187     if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1188         trace_colo_compare_main("secondary: unsupported packet in");
1189     } else {
1190         /* compare packet in the specified connection */
1191         colo_compare_connection(conn, s);
1192     }
1193 }
1194
1195 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1196 {
1197     CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1198
1199     const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1200     int ret;
1201
1202     if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1203                            notify_rs->buf,
1204                            notify_rs->packet_len)) {
1205         ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1206         if (ret < 0) {
1207             error_report("Notify Xen COLO-frame INIT failed");
1208         }
1209     } else if (packet_matches_str("COLO_CHECKPOINT",
1210                                   notify_rs->buf,
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);
1214     } else {
1215         error_report("COLO compare got unsupported instruction");
1216     }
1217 }
1218
1219 /*
1220  * Return 0 is success.
1221  * Return 1 is failed.
1222  */
1223 static int find_and_check_chardev(Chardev **chr,
1224                                   char *chr_name,
1225                                   Error **errp)
1226 {
1227     *chr = qemu_chr_find(chr_name);
1228     if (*chr == NULL) {
1229         error_setg(errp, "Device '%s' not found",
1230                    chr_name);
1231         return 1;
1232     }
1233
1234     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1235         error_setg(errp, "chardev \"%s\" is not reconnectable",
1236                    chr_name);
1237         return 1;
1238     }
1239
1240     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1241         error_setg(errp, "chardev \"%s\" cannot switch context",
1242                    chr_name);
1243         return 1;
1244     }
1245
1246     return 0;
1247 }
1248
1249 /*
1250  * Called from the main thread on the primary
1251  * to setup colo-compare.
1252  */
1253 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1254 {
1255     CompareState *s = COLO_COMPARE(uc);
1256     Chardev *chr;
1257
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");
1261         return;
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");
1267         return;
1268     }
1269
1270     if (!s->compare_timeout) {
1271         /* Set default value to 3000 MS */
1272         s->compare_timeout = DEFAULT_TIME_OUT_MS;
1273     }
1274
1275     if (!s->expired_scan_cycle) {
1276         /* Set default value to 3000 MS */
1277         s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1278     }
1279
1280     if (!max_queue_size) {
1281         /* Set default queue size to 1024 */
1282         max_queue_size = MAX_QUEUE_SIZE;
1283     }
1284
1285     if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1286         !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1287         return;
1288     }
1289
1290     if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1291         !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1292         return;
1293     }
1294
1295     if (find_and_check_chardev(&chr, s->outdev, errp) ||
1296         !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1297         return;
1298     }
1299
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);
1302
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)) {
1307             return;
1308         }
1309
1310         net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1311                            s->vnet_hdr);
1312     }
1313
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);
1319
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);
1326     }
1327
1328     g_queue_init(&s->conn_list);
1329
1330     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1331                                                       connection_key_equal,
1332                                                       g_free,
1333                                                       connection_destroy);
1334
1335     colo_compare_iothread(s);
1336
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;
1342     }
1343     QTAILQ_INSERT_TAIL(&net_compares, s, next);
1344     qemu_mutex_unlock(&colo_compare_mutex);
1345
1346     return;
1347 }
1348
1349 static void colo_flush_packets(void *opaque, void *user_data)
1350 {
1351     CompareState *s = user_data;
1352     Connection *conn = opaque;
1353     Packet *pkt = NULL;
1354
1355     while (!g_queue_is_empty(&conn->primary_list)) {
1356         pkt = g_queue_pop_head(&conn->primary_list);
1357         compare_chr_send(s,
1358                          pkt->data,
1359                          pkt->size,
1360                          pkt->vnet_hdr_len,
1361                          false,
1362                          true);
1363         packet_destroy_partial(pkt, NULL);
1364     }
1365     while (!g_queue_is_empty(&conn->secondary_list)) {
1366         pkt = g_queue_pop_head(&conn->secondary_list);
1367         packet_destroy(pkt, NULL);
1368     }
1369 }
1370
1371 static void colo_compare_class_init(ObjectClass *oc, void *data)
1372 {
1373     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1374
1375     ucc->complete = colo_compare_complete;
1376 }
1377
1378 static void colo_compare_init(Object *obj)
1379 {
1380     CompareState *s = COLO_COMPARE(obj);
1381
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);
1395
1396     object_property_add(obj, "compare_timeout", "uint32",
1397                         compare_get_timeout,
1398                         compare_set_timeout, NULL, NULL);
1399
1400     object_property_add(obj, "expired_scan_cycle", "uint32",
1401                         compare_get_expired_scan_cycle,
1402                         compare_set_expired_scan_cycle, NULL, NULL);
1403
1404     object_property_add(obj, "max_queue_size", "uint32",
1405                         get_max_queue_size,
1406                         set_max_queue_size, NULL, NULL);
1407
1408     s->vnet_hdr = false;
1409     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1410                              compare_set_vnet_hdr);
1411 }
1412
1413 static void colo_compare_finalize(Object *obj)
1414 {
1415     CompareState *s = COLO_COMPARE(obj);
1416     CompareState *tmp = NULL;
1417
1418     qemu_mutex_lock(&colo_compare_mutex);
1419     QTAILQ_FOREACH(tmp, &net_compares, next) {
1420         if (tmp == s) {
1421             QTAILQ_REMOVE(&net_compares, s, next);
1422             break;
1423         }
1424     }
1425     if (QTAILQ_EMPTY(&net_compares)) {
1426         colo_compare_active = false;
1427         qemu_mutex_destroy(&event_mtx);
1428         qemu_cond_destroy(&event_complete_cond);
1429     }
1430     qemu_mutex_unlock(&colo_compare_mutex);
1431
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);
1437     }
1438
1439     colo_compare_timer_del(s);
1440
1441     qemu_bh_delete(s->event_bh);
1442
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);
1448     }
1449     aio_context_release(ctx);
1450
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);
1454
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);
1459     }
1460
1461     if (s->connection_track_table) {
1462         g_hash_table_destroy(s->connection_track_table);
1463     }
1464
1465     object_unref(OBJECT(s->iothread));
1466
1467     g_free(s->pri_indev);
1468     g_free(s->sec_indev);
1469     g_free(s->outdev);
1470     g_free(s->notify_dev);
1471 }
1472
1473 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1474 {
1475     colo_compare_active = false;
1476     qemu_mutex_init(&colo_compare_mutex);
1477 }
1478
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 },
1489         { }
1490     }
1491 };
1492
1493 static void register_types(void)
1494 {
1495     type_register_static(&colo_compare_info);
1496 }
1497
1498 type_init(register_types);