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 #define COLO_COMPARE(obj) \
  40     OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
  41
  42 static QTAILQ_HEAD(, CompareState) net_compares =
  43        QTAILQ_HEAD_INITIALIZER(net_compares);
  44
  45 static NotifierList colo_compare_notifiers =
  46     NOTIFIER_LIST_INITIALIZER(colo_compare_notifiers);
  47
  48 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
  49 #define MAX_QUEUE_SIZE 1024
  50
  51 #define COLO_COMPARE_FREE_PRIMARY     0x01
  52 #define COLO_COMPARE_FREE_SECONDARY   0x02
  53
  54 #define REGULAR_PACKET_CHECK_MS 3000
  55 #define DEFAULT_TIME_OUT_MS 3000
  56
  57 static QemuMutex colo_compare_mutex;
  58 static bool colo_compare_active;
  59 static QemuMutex event_mtx;
  60 static QemuCond event_complete_cond;
  61 static int event_unhandled_count;
  62 static uint32_t max_queue_size;
  63
  64 /*
  65  *  + CompareState ++
  66  *  |               |
  67  *  +---------------+   +---------------+         +---------------+
  68  *  |   conn list   + - >      conn     + ------- >      conn     + -- > ......
  69  *  +---------------+   +---------------+         +---------------+
  70  *  |               |     |           |             |          |
  71  *  +---------------+ +---v----+  +---v----+    +---v----+ +---v----+
  72  *                    |primary |  |secondary    |primary | |secondary
  73  *                    |packet  |  |packet  +    |packet  | |packet  +
  74  *                    +--------+  +--------+    +--------+ +--------+
  75  *                        |           |             |          |
  76  *                    +---v----+  +---v----+    +---v----+ +---v----+
  77  *                    |primary |  |secondary    |primary | |secondary
  78  *                    |packet  |  |packet  +    |packet  | |packet  +
  79  *                    +--------+  +--------+    +--------+ +--------+
  80  *                        |           |             |          |
  81  *                    +---v----+  +---v----+    +---v----+ +---v----+
  82  *                    |primary |  |secondary    |primary | |secondary
  83  *                    |packet  |  |packet  +    |packet  | |packet  +
  84  *                    +--------+  +--------+    +--------+ +--------+
  85  */
  86
  87 typedef struct SendCo {
  88     Coroutine *co;
  89     struct CompareState *s;
  90     CharBackend *chr;
  91     GQueue send_list;
  92     bool notify_remote_frame;
  93     bool done;
  94     int ret;
  95 } SendCo;
  96
  97 typedef struct SendEntry {
  98     uint32_t size;
  99     uint32_t vnet_hdr_len;
 100     uint8_t *buf;
 101 } SendEntry;
 102
 103 typedef struct CompareState {
 104     Object parent;
 105
 106     char *pri_indev;
 107     char *sec_indev;
 108     char *outdev;
 109     char *notify_dev;
 110     CharBackend chr_pri_in;
 111     CharBackend chr_sec_in;
 112     CharBackend chr_out;
 113     CharBackend chr_notify_dev;
 114     SocketReadState pri_rs;
 115     SocketReadState sec_rs;
 116     SocketReadState notify_rs;
 117     SendCo out_sendco;
 118     SendCo notify_sendco;
 119     bool vnet_hdr;
 120     uint32_t compare_timeout;
 121     uint32_t expired_scan_cycle;
 122
 123     /*
 124      * Record the connection that through the NIC
 125      * Element type: Connection
 126      */
 127     GQueue conn_list;
 128     /* Record the connection without repetition */
 129     GHashTable *connection_track_table;
 130
 131     IOThread *iothread;
 132     GMainContext *worker_context;
 133     QEMUTimer *packet_check_timer;
 134
 135     QEMUBH *event_bh;
 136     enum colo_event event;
 137
 138     QTAILQ_ENTRY(CompareState) next;
 139 } CompareState;
 140
 141 typedef struct CompareClass {
 142     ObjectClass parent_class;
 143 } CompareClass;
 144
 145 enum {
 146     PRIMARY_IN = 0,
 147     SECONDARY_IN,
 148 };
 149
 150 static const char *colo_mode[] = {
 151     [PRIMARY_IN] = "primary",
 152     [SECONDARY_IN] = "secondary",
 153 };
 154
 155 static int compare_chr_send(CompareState *s,
 156                             uint8_t *buf,
 157                             uint32_t size,
 158                             uint32_t vnet_hdr_len,
 159                             bool notify_remote_frame,
 160                             bool zero_copy);
 161
 162 static bool packet_matches_str(const char *str,
 163                                const uint8_t *buf,
 164                                uint32_t packet_len)
 165 {
 166     if (packet_len != strlen(str)) {
 167         return false;
 168     }
 169
 170     return !memcmp(str, buf, strlen(str));
 171 }
 172
 173 static void notify_remote_frame(CompareState *s)
 174 {
 175     char msg[] = "DO_CHECKPOINT";
 176     int ret = 0;
 177
 178     ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
 179     if (ret < 0) {
 180         error_report("Notify Xen COLO-frame failed");
 181     }
 182 }
 183
 184 static void colo_compare_inconsistency_notify(CompareState *s)
 185 {
 186     if (s->notify_dev) {
 187         notify_remote_frame(s);
 188     } else {
 189         notifier_list_notify(&colo_compare_notifiers,
 190                              migrate_get_current());
 191     }
 192 }
 193
 194 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
 195 {
 196     struct tcp_hdr *atcp, *btcp;
 197
 198     atcp = (struct tcp_hdr *)(a->transport_header);
 199     btcp = (struct tcp_hdr *)(b->transport_header);
 200     return ntohl(atcp->th_seq) - ntohl(btcp->th_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_MISCOMPARE)) {
 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         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 497             qemu_hexdump((char *)ppkt->data, stderr,
 498                         "colo-compare ppkt", ppkt->size);
 499             qemu_hexdump((char *)spkt->data, stderr,
 500                         "colo-compare spkt", spkt->size);
 501         }
 502
 503         colo_compare_inconsistency_notify(s);
 504     }
 505 }
 506
 507
 508 /*
 509  * Called from the compare thread on the primary
 510  * for compare udp packet
 511  */
 512 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
 513 {
 514     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
 515     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
 516
 517     trace_colo_compare_main("compare udp");
 518
 519     /*
 520      * Because of ppkt and spkt are both in the same connection,
 521      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
 522      * same with spkt. In addition, IP header's Identification is a random
 523      * field, we can handle it in IP fragmentation function later.
 524      * COLO just concern the response net packet payload from primary guest
 525      * and secondary guest are same or not, So we ignored all IP header include
 526      * other field like TOS,TTL,IP Checksum. we only need to compare
 527      * the ip payload here.
 528      */
 529     if (ppkt->size != spkt->size) {
 530         trace_colo_compare_main("UDP: payload size of packets are different");
 531         return -1;
 532     }
 533     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
 534                                     ppkt->size - offset)) {
 535         trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
 536         trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
 537         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 538             qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
 539                          ppkt->size);
 540             qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
 541                          spkt->size);
 542         }
 543         return -1;
 544     } else {
 545         return 0;
 546     }
 547 }
 548
 549 /*
 550  * Called from the compare thread on the primary
 551  * for compare icmp packet
 552  */
 553 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
 554 {
 555     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
 556     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
 557
 558     trace_colo_compare_main("compare icmp");
 559
 560     /*
 561      * Because of ppkt and spkt are both in the same connection,
 562      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
 563      * same with spkt. In addition, IP header's Identification is a random
 564      * field, we can handle it in IP fragmentation function later.
 565      * COLO just concern the response net packet payload from primary guest
 566      * and secondary guest are same or not, So we ignored all IP header include
 567      * other field like TOS,TTL,IP Checksum. we only need to compare
 568      * the ip payload here.
 569      */
 570     if (ppkt->size != spkt->size) {
 571         trace_colo_compare_main("ICMP: payload size of packets are different");
 572         return -1;
 573     }
 574     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
 575                                     ppkt->size - offset)) {
 576         trace_colo_compare_icmp_miscompare("primary pkt size",
 577                                            ppkt->size);
 578         trace_colo_compare_icmp_miscompare("Secondary pkt size",
 579                                            spkt->size);
 580         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 581             qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
 582                          ppkt->size);
 583             qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
 584                          spkt->size);
 585         }
 586         return -1;
 587     } else {
 588         return 0;
 589     }
 590 }
 591
 592 /*
 593  * Called from the compare thread on the primary
 594  * for compare other packet
 595  */
 596 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
 597 {
 598     uint16_t offset = ppkt->vnet_hdr_len;
 599
 600     trace_colo_compare_main("compare other");
 601     if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 602         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
 603
 604         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
 605         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
 606         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
 607         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
 608
 609         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 610                                    pri_ip_dst, spkt->size,
 611                                    sec_ip_src, sec_ip_dst);
 612     }
 613
 614     if (ppkt->size != spkt->size) {
 615         trace_colo_compare_main("Other: payload size of packets are different");
 616         return -1;
 617     }
 618     return colo_compare_packet_payload(ppkt, spkt, offset, offset,
 619                                        ppkt->size - offset);
 620 }
 621
 622 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
 623 {
 624     int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 625
 626     if ((now - pkt->creation_ms) > (*check_time)) {
 627         trace_colo_old_packet_check_found(pkt->creation_ms);
 628         return 0;
 629     } else {
 630         return 1;
 631     }
 632 }
 633
 634 void colo_compare_register_notifier(Notifier *notify)
 635 {
 636     notifier_list_add(&colo_compare_notifiers, notify);
 637 }
 638
 639 void colo_compare_unregister_notifier(Notifier *notify)
 640 {
 641     notifier_remove(notify);
 642 }
 643
 644 static int colo_old_packet_check_one_conn(Connection *conn,
 645                                           CompareState *s)
 646 {
 647     GList *result = NULL;
 648
 649     result = g_queue_find_custom(&conn->primary_list,
 650                                  &s->compare_timeout,
 651                                  (GCompareFunc)colo_old_packet_check_one);
 652
 653     if (result) {
 654         /* Do checkpoint will flush old packet */
 655         colo_compare_inconsistency_notify(s);
 656         return 0;
 657     }
 658
 659     return 1;
 660 }
 661
 662 /*
 663  * Look for old packets that the secondary hasn't matched,
 664  * if we have some then we have to checkpoint to wake
 665  * the secondary up.
 666  */
 667 static void colo_old_packet_check(void *opaque)
 668 {
 669     CompareState *s = opaque;
 670
 671     /*
 672      * If we find one old packet, stop finding job and notify
 673      * COLO frame do checkpoint.
 674      */
 675     g_queue_find_custom(&s->conn_list, s,
 676                         (GCompareFunc)colo_old_packet_check_one_conn);
 677 }
 678
 679 static void colo_compare_packet(CompareState *s, Connection *conn,
 680                                 int (*HandlePacket)(Packet *spkt,
 681                                 Packet *ppkt))
 682 {
 683     Packet *pkt = NULL;
 684     GList *result = NULL;
 685
 686     while (!g_queue_is_empty(&conn->primary_list) &&
 687            !g_queue_is_empty(&conn->secondary_list)) {
 688         pkt = g_queue_pop_head(&conn->primary_list);
 689         result = g_queue_find_custom(&conn->secondary_list,
 690                  pkt, (GCompareFunc)HandlePacket);
 691
 692         if (result) {
 693             colo_release_primary_pkt(s, pkt);
 694             g_queue_remove(&conn->secondary_list, result->data);
 695         } else {
 696             /*
 697              * If one packet arrive late, the secondary_list or
 698              * primary_list will be empty, so we can't compare it
 699              * until next comparison. If the packets in the list are
 700              * timeout, it will trigger a checkpoint request.
 701              */
 702             trace_colo_compare_main("packet different");
 703             g_queue_push_head(&conn->primary_list, pkt);
 704
 705             colo_compare_inconsistency_notify(s);
 706             break;
 707         }
 708     }
 709 }
 710
 711 /*
 712  * Called from the compare thread on the primary
 713  * for compare packet with secondary list of the
 714  * specified connection when a new packet was
 715  * queued to it.
 716  */
 717 static void colo_compare_connection(void *opaque, void *user_data)
 718 {
 719     CompareState *s = user_data;
 720     Connection *conn = opaque;
 721
 722     switch (conn->ip_proto) {
 723     case IPPROTO_TCP:
 724         colo_compare_tcp(s, conn);
 725         break;
 726     case IPPROTO_UDP:
 727         colo_compare_packet(s, conn, colo_packet_compare_udp);
 728         break;
 729     case IPPROTO_ICMP:
 730         colo_compare_packet(s, conn, colo_packet_compare_icmp);
 731         break;
 732     default:
 733         colo_compare_packet(s, conn, colo_packet_compare_other);
 734         break;
 735     }
 736 }
 737
 738 static void coroutine_fn _compare_chr_send(void *opaque)
 739 {
 740     SendCo *sendco = opaque;
 741     CompareState *s = sendco->s;
 742     int ret = 0;
 743
 744     while (!g_queue_is_empty(&sendco->send_list)) {
 745         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
 746         uint32_t len = htonl(entry->size);
 747
 748         ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
 749
 750         if (ret != sizeof(len)) {
 751             g_free(entry->buf);
 752             g_slice_free(SendEntry, entry);
 753             goto err;
 754         }
 755
 756         if (!sendco->notify_remote_frame && s->vnet_hdr) {
 757             /*
 758              * We send vnet header len make other module(like filter-redirector)
 759              * know how to parse net packet correctly.
 760              */
 761             len = htonl(entry->vnet_hdr_len);
 762
 763             ret = qemu_chr_fe_write_all(sendco->chr,
 764                                         (uint8_t *)&len,
 765                                         sizeof(len));
 766
 767             if (ret != sizeof(len)) {
 768                 g_free(entry->buf);
 769                 g_slice_free(SendEntry, entry);
 770                 goto err;
 771             }
 772         }
 773
 774         ret = qemu_chr_fe_write_all(sendco->chr,
 775                                     (uint8_t *)entry->buf,
 776                                     entry->size);
 777
 778         if (ret != entry->size) {
 779             g_free(entry->buf);
 780             g_slice_free(SendEntry, entry);
 781             goto err;
 782         }
 783
 784         g_free(entry->buf);
 785         g_slice_free(SendEntry, entry);
 786     }
 787
 788     sendco->ret = 0;
 789     goto out;
 790
 791 err:
 792     while (!g_queue_is_empty(&sendco->send_list)) {
 793         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
 794         g_free(entry->buf);
 795         g_slice_free(SendEntry, entry);
 796     }
 797     sendco->ret = ret < 0 ? ret : -EIO;
 798 out:
 799     sendco->co = NULL;
 800     sendco->done = true;
 801     aio_wait_kick();
 802 }
 803
 804 static int compare_chr_send(CompareState *s,
 805                             uint8_t *buf,
 806                             uint32_t size,
 807                             uint32_t vnet_hdr_len,
 808                             bool notify_remote_frame,
 809                             bool zero_copy)
 810 {
 811     SendCo *sendco;
 812     SendEntry *entry;
 813
 814     if (notify_remote_frame) {
 815         sendco = &s->notify_sendco;
 816     } else {
 817         sendco = &s->out_sendco;
 818     }
 819
 820     if (!size) {
 821         return 0;
 822     }
 823
 824     entry = g_slice_new(SendEntry);
 825     entry->size = size;
 826     entry->vnet_hdr_len = vnet_hdr_len;
 827     if (zero_copy) {
 828         entry->buf = buf;
 829     } else {
 830         entry->buf = g_malloc(size);
 831         memcpy(entry->buf, buf, size);
 832     }
 833     g_queue_push_head(&sendco->send_list, entry);
 834
 835     if (sendco->done) {
 836         sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
 837         sendco->done = false;
 838         qemu_coroutine_enter(sendco->co);
 839         if (sendco->done) {
 840             /* report early errors */
 841             return sendco->ret;
 842         }
 843     }
 844
 845     /* assume success */
 846     return 0;
 847 }
 848
 849 static int compare_chr_can_read(void *opaque)
 850 {
 851     return COMPARE_READ_LEN_MAX;
 852 }
 853
 854 /*
 855  * Called from the main thread on the primary for packets
 856  * arriving over the socket from the primary.
 857  */
 858 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
 859 {
 860     CompareState *s = COLO_COMPARE(opaque);
 861     int ret;
 862
 863     ret = net_fill_rstate(&s->pri_rs, buf, size);
 864     if (ret == -1) {
 865         qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
 866                                  NULL, NULL, true);
 867         error_report("colo-compare primary_in error");
 868     }
 869 }
 870
 871 /*
 872  * Called from the main thread on the primary for packets
 873  * arriving over the socket from the secondary.
 874  */
 875 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 876 {
 877     CompareState *s = COLO_COMPARE(opaque);
 878     int ret;
 879
 880     ret = net_fill_rstate(&s->sec_rs, buf, size);
 881     if (ret == -1) {
 882         qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
 883                                  NULL, NULL, true);
 884         error_report("colo-compare secondary_in error");
 885     }
 886 }
 887
 888 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
 889 {
 890     CompareState *s = COLO_COMPARE(opaque);
 891     int ret;
 892
 893     ret = net_fill_rstate(&s->notify_rs, buf, size);
 894     if (ret == -1) {
 895         qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
 896                                  NULL, NULL, true);
 897         error_report("colo-compare notify_dev error");
 898     }
 899 }
 900
 901 /*
 902  * Check old packet regularly so it can watch for any packets
 903  * that the secondary hasn't produced equivalents of.
 904  */
 905 static void check_old_packet_regular(void *opaque)
 906 {
 907     CompareState *s = opaque;
 908
 909     /* if have old packet we will notify checkpoint */
 910     colo_old_packet_check(s);
 911     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 912               s->expired_scan_cycle);
 913 }
 914
 915 /* Public API, Used for COLO frame to notify compare event */
 916 void colo_notify_compares_event(void *opaque, int event, Error **errp)
 917 {
 918     CompareState *s;
 919     qemu_mutex_lock(&colo_compare_mutex);
 920
 921     if (!colo_compare_active) {
 922         qemu_mutex_unlock(&colo_compare_mutex);
 923         return;
 924     }
 925
 926     qemu_mutex_lock(&event_mtx);
 927     QTAILQ_FOREACH(s, &net_compares, next) {
 928         s->event = event;
 929         qemu_bh_schedule(s->event_bh);
 930         event_unhandled_count++;
 931     }
 932     /* Wait all compare threads to finish handling this event */
 933     while (event_unhandled_count > 0) {
 934         qemu_cond_wait(&event_complete_cond, &event_mtx);
 935     }
 936
 937     qemu_mutex_unlock(&event_mtx);
 938     qemu_mutex_unlock(&colo_compare_mutex);
 939 }
 940
 941 static void colo_compare_timer_init(CompareState *s)
 942 {
 943     AioContext *ctx = iothread_get_aio_context(s->iothread);
 944
 945     s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
 946                                 SCALE_MS, check_old_packet_regular,
 947                                 s);
 948     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 949               s->expired_scan_cycle);
 950 }
 951
 952 static void colo_compare_timer_del(CompareState *s)
 953 {
 954     if (s->packet_check_timer) {
 955         timer_del(s->packet_check_timer);
 956         timer_free(s->packet_check_timer);
 957         s->packet_check_timer = NULL;
 958     }
 959  }
 960
 961 static void colo_flush_packets(void *opaque, void *user_data);
 962
 963 static void colo_compare_handle_event(void *opaque)
 964 {
 965     CompareState *s = opaque;
 966
 967     switch (s->event) {
 968     case COLO_EVENT_CHECKPOINT:
 969         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
 970         break;
 971     case COLO_EVENT_FAILOVER:
 972         break;
 973     default:
 974         break;
 975     }
 976
 977     qemu_mutex_lock(&event_mtx);
 978     assert(event_unhandled_count > 0);
 979     event_unhandled_count--;
 980     qemu_cond_broadcast(&event_complete_cond);
 981     qemu_mutex_unlock(&event_mtx);
 982 }
 983
 984 static void colo_compare_iothread(CompareState *s)
 985 {
 986     AioContext *ctx = iothread_get_aio_context(s->iothread);
 987     object_ref(OBJECT(s->iothread));
 988     s->worker_context = iothread_get_g_main_context(s->iothread);
 989
 990     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
 991                              compare_pri_chr_in, NULL, NULL,
 992                              s, s->worker_context, true);
 993     qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
 994                              compare_sec_chr_in, NULL, NULL,
 995                              s, s->worker_context, true);
 996     if (s->notify_dev) {
 997         qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
 998                                  compare_notify_chr, NULL, NULL,
 999                                  s, s->worker_context, true);
1000     }
1001
1002     colo_compare_timer_init(s);
1003     s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
1004 }
1005
1006 static char *compare_get_pri_indev(Object *obj, Error **errp)
1007 {
1008     CompareState *s = COLO_COMPARE(obj);
1009
1010     return g_strdup(s->pri_indev);
1011 }
1012
1013 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1014 {
1015     CompareState *s = COLO_COMPARE(obj);
1016
1017     g_free(s->pri_indev);
1018     s->pri_indev = g_strdup(value);
1019 }
1020
1021 static char *compare_get_sec_indev(Object *obj, Error **errp)
1022 {
1023     CompareState *s = COLO_COMPARE(obj);
1024
1025     return g_strdup(s->sec_indev);
1026 }
1027
1028 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1029 {
1030     CompareState *s = COLO_COMPARE(obj);
1031
1032     g_free(s->sec_indev);
1033     s->sec_indev = g_strdup(value);
1034 }
1035
1036 static char *compare_get_outdev(Object *obj, Error **errp)
1037 {
1038     CompareState *s = COLO_COMPARE(obj);
1039
1040     return g_strdup(s->outdev);
1041 }
1042
1043 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1044 {
1045     CompareState *s = COLO_COMPARE(obj);
1046
1047     g_free(s->outdev);
1048     s->outdev = g_strdup(value);
1049 }
1050
1051 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1052 {
1053     CompareState *s = COLO_COMPARE(obj);
1054
1055     return s->vnet_hdr;
1056 }
1057
1058 static void compare_set_vnet_hdr(Object *obj,
1059                                  bool value,
1060                                  Error **errp)
1061 {
1062     CompareState *s = COLO_COMPARE(obj);
1063
1064     s->vnet_hdr = value;
1065 }
1066
1067 static char *compare_get_notify_dev(Object *obj, Error **errp)
1068 {
1069     CompareState *s = COLO_COMPARE(obj);
1070
1071     return g_strdup(s->notify_dev);
1072 }
1073
1074 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1075 {
1076     CompareState *s = COLO_COMPARE(obj);
1077
1078     g_free(s->notify_dev);
1079     s->notify_dev = g_strdup(value);
1080 }
1081
1082 static void compare_get_timeout(Object *obj, Visitor *v,
1083                                 const char *name, void *opaque,
1084                                 Error **errp)
1085 {
1086     CompareState *s = COLO_COMPARE(obj);
1087     uint32_t value = s->compare_timeout;
1088
1089     visit_type_uint32(v, name, &value, errp);
1090 }
1091
1092 static void compare_set_timeout(Object *obj, Visitor *v,
1093                                 const char *name, void *opaque,
1094                                 Error **errp)
1095 {
1096     CompareState *s = COLO_COMPARE(obj);
1097     uint32_t value;
1098
1099     if (!visit_type_uint32(v, name, &value, errp)) {
1100         return;
1101     }
1102     if (!value) {
1103         error_setg(errp, "Property '%s.%s' requires a positive value",
1104                    object_get_typename(obj), name);
1105         return;
1106     }
1107     s->compare_timeout = value;
1108 }
1109
1110 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1111                                            const char *name, void *opaque,
1112                                            Error **errp)
1113 {
1114     CompareState *s = COLO_COMPARE(obj);
1115     uint32_t value = s->expired_scan_cycle;
1116
1117     visit_type_uint32(v, name, &value, errp);
1118 }
1119
1120 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1121                                            const char *name, void *opaque,
1122                                            Error **errp)
1123 {
1124     CompareState *s = COLO_COMPARE(obj);
1125     uint32_t value;
1126
1127     if (!visit_type_uint32(v, name, &value, errp)) {
1128         return;
1129     }
1130     if (!value) {
1131         error_setg(errp, "Property '%s.%s' requires a positive value",
1132                    object_get_typename(obj), name);
1133         return;
1134     }
1135     s->expired_scan_cycle = value;
1136 }
1137
1138 static void get_max_queue_size(Object *obj, Visitor *v,
1139                                const char *name, void *opaque,
1140                                Error **errp)
1141 {
1142     uint32_t value = max_queue_size;
1143
1144     visit_type_uint32(v, name, &value, errp);
1145 }
1146
1147 static void set_max_queue_size(Object *obj, Visitor *v,
1148                                const char *name, void *opaque,
1149                                Error **errp)
1150 {
1151     Error *local_err = NULL;
1152     uint32_t value;
1153
1154     visit_type_uint32(v, name, &value, &local_err);
1155     if (local_err) {
1156         goto out;
1157     }
1158     if (!value) {
1159         error_setg(&local_err, "Property '%s.%s' requires a positive value",
1160                    object_get_typename(obj), name);
1161         goto out;
1162     }
1163     max_queue_size = value;
1164
1165 out:
1166     error_propagate(errp, local_err);
1167 }
1168
1169 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1170 {
1171     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1172     Connection *conn = NULL;
1173
1174     if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1175         trace_colo_compare_main("primary: unsupported packet in");
1176         compare_chr_send(s,
1177                          pri_rs->buf,
1178                          pri_rs->packet_len,
1179                          pri_rs->vnet_hdr_len,
1180                          false,
1181                          false);
1182     } else {
1183         /* compare packet in the specified connection */
1184         colo_compare_connection(conn, s);
1185     }
1186 }
1187
1188 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1189 {
1190     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1191     Connection *conn = NULL;
1192
1193     if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1194         trace_colo_compare_main("secondary: unsupported packet in");
1195     } else {
1196         /* compare packet in the specified connection */
1197         colo_compare_connection(conn, s);
1198     }
1199 }
1200
1201 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1202 {
1203     CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1204
1205     const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1206     int ret;
1207
1208     if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1209                            notify_rs->buf,
1210                            notify_rs->packet_len)) {
1211         ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1212         if (ret < 0) {
1213             error_report("Notify Xen COLO-frame INIT failed");
1214         }
1215     } else if (packet_matches_str("COLO_CHECKPOINT",
1216                                   notify_rs->buf,
1217                                   notify_rs->packet_len)) {
1218         /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1219         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1220     } else {
1221         error_report("COLO compare got unsupported instruction");
1222     }
1223 }
1224
1225 /*
1226  * Return 0 is success.
1227  * Return 1 is failed.
1228  */
1229 static int find_and_check_chardev(Chardev **chr,
1230                                   char *chr_name,
1231                                   Error **errp)
1232 {
1233     *chr = qemu_chr_find(chr_name);
1234     if (*chr == NULL) {
1235         error_setg(errp, "Device '%s' not found",
1236                    chr_name);
1237         return 1;
1238     }
1239
1240     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1241         error_setg(errp, "chardev \"%s\" is not reconnectable",
1242                    chr_name);
1243         return 1;
1244     }
1245
1246     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1247         error_setg(errp, "chardev \"%s\" cannot switch context",
1248                    chr_name);
1249         return 1;
1250     }
1251
1252     return 0;
1253 }
1254
1255 /*
1256  * Called from the main thread on the primary
1257  * to setup colo-compare.
1258  */
1259 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1260 {
1261     CompareState *s = COLO_COMPARE(uc);
1262     Chardev *chr;
1263
1264     if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1265         error_setg(errp, "colo compare needs 'primary_in' ,"
1266                    "'secondary_in','outdev','iothread' property set");
1267         return;
1268     } else if (!strcmp(s->pri_indev, s->outdev) ||
1269                !strcmp(s->sec_indev, s->outdev) ||
1270                !strcmp(s->pri_indev, s->sec_indev)) {
1271         error_setg(errp, "'indev' and 'outdev' could not be same "
1272                    "for compare module");
1273         return;
1274     }
1275
1276     if (!s->compare_timeout) {
1277         /* Set default value to 3000 MS */
1278         s->compare_timeout = DEFAULT_TIME_OUT_MS;
1279     }
1280
1281     if (!s->expired_scan_cycle) {
1282         /* Set default value to 3000 MS */
1283         s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1284     }
1285
1286     if (!max_queue_size) {
1287         /* Set default queue size to 1024 */
1288         max_queue_size = MAX_QUEUE_SIZE;
1289     }
1290
1291     if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1292         !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1293         return;
1294     }
1295
1296     if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1297         !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1298         return;
1299     }
1300
1301     if (find_and_check_chardev(&chr, s->outdev, errp) ||
1302         !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1303         return;
1304     }
1305
1306     net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1307     net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1308
1309     /* Try to enable remote notify chardev, currently just for Xen COLO */
1310     if (s->notify_dev) {
1311         if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1312             !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1313             return;
1314         }
1315
1316         net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1317                            s->vnet_hdr);
1318     }
1319
1320     s->out_sendco.s = s;
1321     s->out_sendco.chr = &s->chr_out;
1322     s->out_sendco.notify_remote_frame = false;
1323     s->out_sendco.done = true;
1324     g_queue_init(&s->out_sendco.send_list);
1325
1326     if (s->notify_dev) {
1327         s->notify_sendco.s = s;
1328         s->notify_sendco.chr = &s->chr_notify_dev;
1329         s->notify_sendco.notify_remote_frame = true;
1330         s->notify_sendco.done = true;
1331         g_queue_init(&s->notify_sendco.send_list);
1332     }
1333
1334     g_queue_init(&s->conn_list);
1335
1336     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1337                                                       connection_key_equal,
1338                                                       g_free,
1339                                                       connection_destroy);
1340
1341     colo_compare_iothread(s);
1342
1343     qemu_mutex_lock(&colo_compare_mutex);
1344     if (!colo_compare_active) {
1345         qemu_mutex_init(&event_mtx);
1346         qemu_cond_init(&event_complete_cond);
1347         colo_compare_active = true;
1348     }
1349     QTAILQ_INSERT_TAIL(&net_compares, s, next);
1350     qemu_mutex_unlock(&colo_compare_mutex);
1351
1352     return;
1353 }
1354
1355 static void colo_flush_packets(void *opaque, void *user_data)
1356 {
1357     CompareState *s = user_data;
1358     Connection *conn = opaque;
1359     Packet *pkt = NULL;
1360
1361     while (!g_queue_is_empty(&conn->primary_list)) {
1362         pkt = g_queue_pop_head(&conn->primary_list);
1363         compare_chr_send(s,
1364                          pkt->data,
1365                          pkt->size,
1366                          pkt->vnet_hdr_len,
1367                          false,
1368                          true);
1369         packet_destroy_partial(pkt, NULL);
1370     }
1371     while (!g_queue_is_empty(&conn->secondary_list)) {
1372         pkt = g_queue_pop_head(&conn->secondary_list);
1373         packet_destroy(pkt, NULL);
1374     }
1375 }
1376
1377 static void colo_compare_class_init(ObjectClass *oc, void *data)
1378 {
1379     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1380
1381     ucc->complete = colo_compare_complete;
1382 }
1383
1384 static void colo_compare_init(Object *obj)
1385 {
1386     CompareState *s = COLO_COMPARE(obj);
1387
1388     object_property_add_str(obj, "primary_in",
1389                             compare_get_pri_indev, compare_set_pri_indev);
1390     object_property_add_str(obj, "secondary_in",
1391                             compare_get_sec_indev, compare_set_sec_indev);
1392     object_property_add_str(obj, "outdev",
1393                             compare_get_outdev, compare_set_outdev);
1394     object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1395                             (Object **)&s->iothread,
1396                             object_property_allow_set_link,
1397                             OBJ_PROP_LINK_STRONG);
1398     /* This parameter just for Xen COLO */
1399     object_property_add_str(obj, "notify_dev",
1400                             compare_get_notify_dev, compare_set_notify_dev);
1401
1402     object_property_add(obj, "compare_timeout", "uint32",
1403                         compare_get_timeout,
1404                         compare_set_timeout, NULL, NULL);
1405
1406     object_property_add(obj, "expired_scan_cycle", "uint32",
1407                         compare_get_expired_scan_cycle,
1408                         compare_set_expired_scan_cycle, NULL, NULL);
1409
1410     object_property_add(obj, "max_queue_size", "uint32",
1411                         get_max_queue_size,
1412                         set_max_queue_size, NULL, NULL);
1413
1414     s->vnet_hdr = false;
1415     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1416                              compare_set_vnet_hdr);
1417 }
1418
1419 static void colo_compare_finalize(Object *obj)
1420 {
1421     CompareState *s = COLO_COMPARE(obj);
1422     CompareState *tmp = NULL;
1423
1424     qemu_mutex_lock(&colo_compare_mutex);
1425     QTAILQ_FOREACH(tmp, &net_compares, next) {
1426         if (tmp == s) {
1427             QTAILQ_REMOVE(&net_compares, s, next);
1428             break;
1429         }
1430     }
1431     if (QTAILQ_EMPTY(&net_compares)) {
1432         colo_compare_active = false;
1433         qemu_mutex_destroy(&event_mtx);
1434         qemu_cond_destroy(&event_complete_cond);
1435     }
1436     qemu_mutex_unlock(&colo_compare_mutex);
1437
1438     qemu_chr_fe_deinit(&s->chr_pri_in, false);
1439     qemu_chr_fe_deinit(&s->chr_sec_in, false);
1440     qemu_chr_fe_deinit(&s->chr_out, false);
1441     if (s->notify_dev) {
1442         qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1443     }
1444
1445     colo_compare_timer_del(s);
1446
1447     qemu_bh_delete(s->event_bh);
1448
1449     AioContext *ctx = iothread_get_aio_context(s->iothread);
1450     aio_context_acquire(ctx);
1451     AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1452     if (s->notify_dev) {
1453         AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1454     }
1455     aio_context_release(ctx);
1456
1457     /* Release all unhandled packets after compare thead exited */
1458     g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1459     AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1460
1461     g_queue_clear(&s->conn_list);
1462     g_queue_clear(&s->out_sendco.send_list);
1463     if (s->notify_dev) {
1464         g_queue_clear(&s->notify_sendco.send_list);
1465     }
1466
1467     if (s->connection_track_table) {
1468         g_hash_table_destroy(s->connection_track_table);
1469     }
1470
1471     object_unref(OBJECT(s->iothread));
1472
1473     g_free(s->pri_indev);
1474     g_free(s->sec_indev);
1475     g_free(s->outdev);
1476     g_free(s->notify_dev);
1477 }
1478
1479 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1480 {
1481     colo_compare_active = false;
1482     qemu_mutex_init(&colo_compare_mutex);
1483 }
1484
1485 static const TypeInfo colo_compare_info = {
1486     .name = TYPE_COLO_COMPARE,
1487     .parent = TYPE_OBJECT,
1488     .instance_size = sizeof(CompareState),
1489     .instance_init = colo_compare_init,
1490     .instance_finalize = colo_compare_finalize,
1491     .class_size = sizeof(CompareClass),
1492     .class_init = colo_compare_class_init,
1493     .interfaces = (InterfaceInfo[]) {
1494         { TYPE_USER_CREATABLE },
1495         { }
1496     }
1497 };
1498
1499 static void register_types(void)
1500 {
1501     type_register_static(&colo_compare_info);
1502 }
1503
1504 type_init(register_types);