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
  63 /*
  64  *  + CompareState ++
  65  *  |               |
  66  *  +---------------+   +---------------+         +---------------+
  67  *  |   conn list   + - >      conn     + ------- >      conn     + -- > ......
  68  *  +---------------+   +---------------+         +---------------+
  69  *  |               |     |           |             |          |
  70  *  +---------------+ +---v----+  +---v----+    +---v----+ +---v----+
  71  *                    |primary |  |secondary    |primary | |secondary
  72  *                    |packet  |  |packet  +    |packet  | |packet  +
  73  *                    +--------+  +--------+    +--------+ +--------+
  74  *                        |           |             |          |
  75  *                    +---v----+  +---v----+    +---v----+ +---v----+
  76  *                    |primary |  |secondary    |primary | |secondary
  77  *                    |packet  |  |packet  +    |packet  | |packet  +
  78  *                    +--------+  +--------+    +--------+ +--------+
  79  *                        |           |             |          |
  80  *                    +---v----+  +---v----+    +---v----+ +---v----+
  81  *                    |primary |  |secondary    |primary | |secondary
  82  *                    |packet  |  |packet  +    |packet  | |packet  +
  83  *                    +--------+  +--------+    +--------+ +--------+
  84  */
  85
  86 typedef struct SendCo {
  87     Coroutine *co;
  88     struct CompareState *s;
  89     CharBackend *chr;
  90     GQueue send_list;
  91     bool notify_remote_frame;
  92     bool done;
  93     int ret;
  94 } SendCo;
  95
  96 typedef struct SendEntry {
  97     uint32_t size;
  98     uint32_t vnet_hdr_len;
  99     uint8_t *buf;
 100 } SendEntry;
 101
 102 typedef struct CompareState {
 103     Object parent;
 104
 105     char *pri_indev;
 106     char *sec_indev;
 107     char *outdev;
 108     char *notify_dev;
 109     CharBackend chr_pri_in;
 110     CharBackend chr_sec_in;
 111     CharBackend chr_out;
 112     CharBackend chr_notify_dev;
 113     SocketReadState pri_rs;
 114     SocketReadState sec_rs;
 115     SocketReadState notify_rs;
 116     SendCo out_sendco;
 117     SendCo notify_sendco;
 118     bool vnet_hdr;
 119     uint32_t compare_timeout;
 120     uint32_t expired_scan_cycle;
 121
 122     /*
 123      * Record the connection that through the NIC
 124      * Element type: Connection
 125      */
 126     GQueue conn_list;
 127     /* Record the connection without repetition */
 128     GHashTable *connection_track_table;
 129
 130     IOThread *iothread;
 131     GMainContext *worker_context;
 132     QEMUTimer *packet_check_timer;
 133
 134     QEMUBH *event_bh;
 135     enum colo_event event;
 136
 137     QTAILQ_ENTRY(CompareState) next;
 138 } CompareState;
 139
 140 typedef struct CompareClass {
 141     ObjectClass parent_class;
 142 } CompareClass;
 143
 144 enum {
 145     PRIMARY_IN = 0,
 146     SECONDARY_IN,
 147 };
 148
 149 static const char *colo_mode[] = {
 150     [PRIMARY_IN] = "primary",
 151     [SECONDARY_IN] = "secondary",
 152 };
 153
 154 static int compare_chr_send(CompareState *s,
 155                             uint8_t *buf,
 156                             uint32_t size,
 157                             uint32_t vnet_hdr_len,
 158                             bool notify_remote_frame,
 159                             bool zero_copy);
 160
 161 static bool packet_matches_str(const char *str,
 162                                const uint8_t *buf,
 163                                uint32_t packet_len)
 164 {
 165     if (packet_len != strlen(str)) {
 166         return false;
 167     }
 168
 169     return !memcmp(str, buf, strlen(str));
 170 }
 171
 172 static void notify_remote_frame(CompareState *s)
 173 {
 174     char msg[] = "DO_CHECKPOINT";
 175     int ret = 0;
 176
 177     ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
 178     if (ret < 0) {
 179         error_report("Notify Xen COLO-frame failed");
 180     }
 181 }
 182
 183 static void colo_compare_inconsistency_notify(CompareState *s)
 184 {
 185     if (s->notify_dev) {
 186         notify_remote_frame(s);
 187     } else {
 188         notifier_list_notify(&colo_compare_notifiers,
 189                              migrate_get_current());
 190     }
 191 }
 192
 193 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
 194 {
 195     struct tcp_hdr *atcp, *btcp;
 196
 197     atcp = (struct tcp_hdr *)(a->transport_header);
 198     btcp = (struct tcp_hdr *)(b->transport_header);
 199     return ntohl(atcp->th_seq) - ntohl(btcp->th_seq);
 200 }
 201
 202 static void fill_pkt_tcp_info(void *data, uint32_t *max_ack)
 203 {
 204     Packet *pkt = data;
 205     struct tcp_hdr *tcphd;
 206
 207     tcphd = (struct tcp_hdr *)pkt->transport_header;
 208
 209     pkt->tcp_seq = ntohl(tcphd->th_seq);
 210     pkt->tcp_ack = ntohl(tcphd->th_ack);
 211     *max_ack = *max_ack > pkt->tcp_ack ? *max_ack : pkt->tcp_ack;
 212     pkt->header_size = pkt->transport_header - (uint8_t *)pkt->data
 213                        + (tcphd->th_off << 2) - pkt->vnet_hdr_len;
 214     pkt->payload_size = pkt->size - pkt->header_size;
 215     pkt->seq_end = pkt->tcp_seq + pkt->payload_size;
 216     pkt->flags = tcphd->th_flags;
 217 }
 218
 219 /*
 220  * Return 1 on success, if return 0 means the
 221  * packet will be dropped
 222  */
 223 static int colo_insert_packet(GQueue *queue, Packet *pkt, uint32_t *max_ack)
 224 {
 225     if (g_queue_get_length(queue) <= MAX_QUEUE_SIZE) {
 226         if (pkt->ip->ip_p == IPPROTO_TCP) {
 227             fill_pkt_tcp_info(pkt, max_ack);
 228             g_queue_insert_sorted(queue,
 229                                   pkt,
 230                                   (GCompareDataFunc)seq_sorter,
 231                                   NULL);
 232         } else {
 233             g_queue_push_tail(queue, pkt);
 234         }
 235         return 1;
 236     }
 237     return 0;
 238 }
 239
 240 /*
 241  * Return 0 on success, if return -1 means the pkt
 242  * is unsupported(arp and ipv6) and will be sent later
 243  */
 244 static int packet_enqueue(CompareState *s, int mode, Connection **con)
 245 {
 246     ConnectionKey key;
 247     Packet *pkt = NULL;
 248     Connection *conn;
 249     int ret;
 250
 251     if (mode == PRIMARY_IN) {
 252         pkt = packet_new(s->pri_rs.buf,
 253                          s->pri_rs.packet_len,
 254                          s->pri_rs.vnet_hdr_len);
 255     } else {
 256         pkt = packet_new(s->sec_rs.buf,
 257                          s->sec_rs.packet_len,
 258                          s->sec_rs.vnet_hdr_len);
 259     }
 260
 261     if (parse_packet_early(pkt)) {
 262         packet_destroy(pkt, NULL);
 263         pkt = NULL;
 264         return -1;
 265     }
 266     fill_connection_key(pkt, &key);
 267
 268     conn = connection_get(s->connection_track_table,
 269                           &key,
 270                           &s->conn_list);
 271
 272     if (!conn->processing) {
 273         g_queue_push_tail(&s->conn_list, conn);
 274         conn->processing = true;
 275     }
 276
 277     if (mode == PRIMARY_IN) {
 278         ret = colo_insert_packet(&conn->primary_list, pkt, &conn->pack);
 279     } else {
 280         ret = colo_insert_packet(&conn->secondary_list, pkt, &conn->sack);
 281     }
 282
 283     if (!ret) {
 284         trace_colo_compare_drop_packet(colo_mode[mode],
 285             "queue size too big, drop packet");
 286         packet_destroy(pkt, NULL);
 287         pkt = NULL;
 288     }
 289
 290     *con = conn;
 291
 292     return 0;
 293 }
 294
 295 static inline bool after(uint32_t seq1, uint32_t seq2)
 296 {
 297         return (int32_t)(seq1 - seq2) > 0;
 298 }
 299
 300 static void colo_release_primary_pkt(CompareState *s, Packet *pkt)
 301 {
 302     int ret;
 303     ret = compare_chr_send(s,
 304                            pkt->data,
 305                            pkt->size,
 306                            pkt->vnet_hdr_len,
 307                            false,
 308                            true);
 309     if (ret < 0) {
 310         error_report("colo send primary packet failed");
 311     }
 312     trace_colo_compare_main("packet same and release packet");
 313     packet_destroy_partial(pkt, NULL);
 314 }
 315
 316 /*
 317  * The IP packets sent by primary and secondary
 318  * will be compared in here
 319  * TODO support ip fragment, Out-Of-Order
 320  * return:    0  means packet same
 321  *            > 0 || < 0 means packet different
 322  */
 323 static int colo_compare_packet_payload(Packet *ppkt,
 324                                        Packet *spkt,
 325                                        uint16_t poffset,
 326                                        uint16_t soffset,
 327                                        uint16_t len)
 328
 329 {
 330     if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 331         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
 332
 333         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
 334         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
 335         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
 336         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
 337
 338         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 339                                    pri_ip_dst, spkt->size,
 340                                    sec_ip_src, sec_ip_dst);
 341     }
 342
 343     return memcmp(ppkt->data + poffset, spkt->data + soffset, len);
 344 }
 345
 346 /*
 347  * return true means that the payload is consist and
 348  * need to make the next comparison, false means do
 349  * the checkpoint
 350 */
 351 static bool colo_mark_tcp_pkt(Packet *ppkt, Packet *spkt,
 352                               int8_t *mark, uint32_t max_ack)
 353 {
 354     *mark = 0;
 355
 356     if (ppkt->tcp_seq == spkt->tcp_seq && ppkt->seq_end == spkt->seq_end) {
 357         if (!colo_compare_packet_payload(ppkt, spkt,
 358                                         ppkt->header_size, spkt->header_size,
 359                                         ppkt->payload_size)) {
 360             *mark = COLO_COMPARE_FREE_SECONDARY | COLO_COMPARE_FREE_PRIMARY;
 361             return true;
 362         }
 363     }
 364
 365     /* one part of secondary packet payload still need to be compared */
 366     if (!after(ppkt->seq_end, spkt->seq_end)) {
 367         if (!colo_compare_packet_payload(ppkt, spkt,
 368                                         ppkt->header_size + ppkt->offset,
 369                                         spkt->header_size + spkt->offset,
 370                                         ppkt->payload_size - ppkt->offset)) {
 371             if (!after(ppkt->tcp_ack, max_ack)) {
 372                 *mark = COLO_COMPARE_FREE_PRIMARY;
 373                 spkt->offset += ppkt->payload_size - ppkt->offset;
 374                 return true;
 375             } else {
 376                 /* secondary guest hasn't ack the data, don't send
 377                  * out this packet
 378                  */
 379                 return false;
 380             }
 381         }
 382     } else {
 383         /* primary packet is longer than secondary packet, compare
 384          * the same part and mark the primary packet offset
 385          */
 386         if (!colo_compare_packet_payload(ppkt, spkt,
 387                                         ppkt->header_size + ppkt->offset,
 388                                         spkt->header_size + spkt->offset,
 389                                         spkt->payload_size - spkt->offset)) {
 390             *mark = COLO_COMPARE_FREE_SECONDARY;
 391             ppkt->offset += spkt->payload_size - spkt->offset;
 392             return true;
 393         }
 394     }
 395
 396     return false;
 397 }
 398
 399 static void colo_compare_tcp(CompareState *s, Connection *conn)
 400 {
 401     Packet *ppkt = NULL, *spkt = NULL;
 402     int8_t mark;
 403
 404     /*
 405      * If ppkt and spkt have the same payload, but ppkt's ACK
 406      * is greater than spkt's ACK, in this case we can not
 407      * send the ppkt because it will cause the secondary guest
 408      * to miss sending some data in the next. Therefore, we
 409      * record the maximum ACK in the current queue at both
 410      * primary side and secondary side. Only when the ack is
 411      * less than the smaller of the two maximum ack, then we
 412      * can ensure that the packet's payload is acknowledged by
 413      * primary and secondary.
 414     */
 415     uint32_t min_ack = conn->pack > conn->sack ? conn->sack : conn->pack;
 416
 417 pri:
 418     if (g_queue_is_empty(&conn->primary_list)) {
 419         return;
 420     }
 421     ppkt = g_queue_pop_head(&conn->primary_list);
 422 sec:
 423     if (g_queue_is_empty(&conn->secondary_list)) {
 424         g_queue_push_head(&conn->primary_list, ppkt);
 425         return;
 426     }
 427     spkt = g_queue_pop_head(&conn->secondary_list);
 428
 429     if (ppkt->tcp_seq == ppkt->seq_end) {
 430         colo_release_primary_pkt(s, ppkt);
 431         ppkt = NULL;
 432     }
 433
 434     if (ppkt && conn->compare_seq && !after(ppkt->seq_end, conn->compare_seq)) {
 435         trace_colo_compare_main("pri: this packet has compared");
 436         colo_release_primary_pkt(s, ppkt);
 437         ppkt = NULL;
 438     }
 439
 440     if (spkt->tcp_seq == spkt->seq_end) {
 441         packet_destroy(spkt, NULL);
 442         if (!ppkt) {
 443             goto pri;
 444         } else {
 445             goto sec;
 446         }
 447     } else {
 448         if (conn->compare_seq && !after(spkt->seq_end, conn->compare_seq)) {
 449             trace_colo_compare_main("sec: this packet has compared");
 450             packet_destroy(spkt, NULL);
 451             if (!ppkt) {
 452                 goto pri;
 453             } else {
 454                 goto sec;
 455             }
 456         }
 457         if (!ppkt) {
 458             g_queue_push_head(&conn->secondary_list, spkt);
 459             goto pri;
 460         }
 461     }
 462
 463     if (colo_mark_tcp_pkt(ppkt, spkt, &mark, min_ack)) {
 464         trace_colo_compare_tcp_info("pri",
 465                                     ppkt->tcp_seq, ppkt->tcp_ack,
 466                                     ppkt->header_size, ppkt->payload_size,
 467                                     ppkt->offset, ppkt->flags);
 468
 469         trace_colo_compare_tcp_info("sec",
 470                                     spkt->tcp_seq, spkt->tcp_ack,
 471                                     spkt->header_size, spkt->payload_size,
 472                                     spkt->offset, spkt->flags);
 473
 474         if (mark == COLO_COMPARE_FREE_PRIMARY) {
 475             conn->compare_seq = ppkt->seq_end;
 476             colo_release_primary_pkt(s, ppkt);
 477             g_queue_push_head(&conn->secondary_list, spkt);
 478             goto pri;
 479         }
 480         if (mark == COLO_COMPARE_FREE_SECONDARY) {
 481             conn->compare_seq = spkt->seq_end;
 482             packet_destroy(spkt, NULL);
 483             goto sec;
 484         }
 485         if (mark == (COLO_COMPARE_FREE_PRIMARY | COLO_COMPARE_FREE_SECONDARY)) {
 486             conn->compare_seq = ppkt->seq_end;
 487             colo_release_primary_pkt(s, ppkt);
 488             packet_destroy(spkt, NULL);
 489             goto pri;
 490         }
 491     } else {
 492         g_queue_push_head(&conn->primary_list, ppkt);
 493         g_queue_push_head(&conn->secondary_list, spkt);
 494
 495         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 496             qemu_hexdump((char *)ppkt->data, stderr,
 497                         "colo-compare ppkt", ppkt->size);
 498             qemu_hexdump((char *)spkt->data, stderr,
 499                         "colo-compare spkt", spkt->size);
 500         }
 501
 502         colo_compare_inconsistency_notify(s);
 503     }
 504 }
 505
 506
 507 /*
 508  * Called from the compare thread on the primary
 509  * for compare udp packet
 510  */
 511 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
 512 {
 513     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
 514     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
 515
 516     trace_colo_compare_main("compare udp");
 517
 518     /*
 519      * Because of ppkt and spkt are both in the same connection,
 520      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
 521      * same with spkt. In addition, IP header's Identification is a random
 522      * field, we can handle it in IP fragmentation function later.
 523      * COLO just concern the response net packet payload from primary guest
 524      * and secondary guest are same or not, So we ignored all IP header include
 525      * other field like TOS,TTL,IP Checksum. we only need to compare
 526      * the ip payload here.
 527      */
 528     if (ppkt->size != spkt->size) {
 529         trace_colo_compare_main("UDP: payload size of packets are different");
 530         return -1;
 531     }
 532     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
 533                                     ppkt->size - offset)) {
 534         trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
 535         trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
 536         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 537             qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
 538                          ppkt->size);
 539             qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
 540                          spkt->size);
 541         }
 542         return -1;
 543     } else {
 544         return 0;
 545     }
 546 }
 547
 548 /*
 549  * Called from the compare thread on the primary
 550  * for compare icmp packet
 551  */
 552 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
 553 {
 554     uint16_t network_header_length = ppkt->ip->ip_hl << 2;
 555     uint16_t offset = network_header_length + ETH_HLEN + ppkt->vnet_hdr_len;
 556
 557     trace_colo_compare_main("compare icmp");
 558
 559     /*
 560      * Because of ppkt and spkt are both in the same connection,
 561      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
 562      * same with spkt. In addition, IP header's Identification is a random
 563      * field, we can handle it in IP fragmentation function later.
 564      * COLO just concern the response net packet payload from primary guest
 565      * and secondary guest are same or not, So we ignored all IP header include
 566      * other field like TOS,TTL,IP Checksum. we only need to compare
 567      * the ip payload here.
 568      */
 569     if (ppkt->size != spkt->size) {
 570         trace_colo_compare_main("ICMP: payload size of packets are different");
 571         return -1;
 572     }
 573     if (colo_compare_packet_payload(ppkt, spkt, offset, offset,
 574                                     ppkt->size - offset)) {
 575         trace_colo_compare_icmp_miscompare("primary pkt size",
 576                                            ppkt->size);
 577         trace_colo_compare_icmp_miscompare("Secondary pkt size",
 578                                            spkt->size);
 579         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 580             qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
 581                          ppkt->size);
 582             qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
 583                          spkt->size);
 584         }
 585         return -1;
 586     } else {
 587         return 0;
 588     }
 589 }
 590
 591 /*
 592  * Called from the compare thread on the primary
 593  * for compare other packet
 594  */
 595 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
 596 {
 597     uint16_t offset = ppkt->vnet_hdr_len;
 598
 599     trace_colo_compare_main("compare other");
 600     if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 601         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
 602
 603         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
 604         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
 605         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
 606         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
 607
 608         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 609                                    pri_ip_dst, spkt->size,
 610                                    sec_ip_src, sec_ip_dst);
 611     }
 612
 613     if (ppkt->size != spkt->size) {
 614         trace_colo_compare_main("Other: payload size of packets are different");
 615         return -1;
 616     }
 617     return colo_compare_packet_payload(ppkt, spkt, offset, offset,
 618                                        ppkt->size - offset);
 619 }
 620
 621 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
 622 {
 623     int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 624
 625     if ((now - pkt->creation_ms) > (*check_time)) {
 626         trace_colo_old_packet_check_found(pkt->creation_ms);
 627         return 0;
 628     } else {
 629         return 1;
 630     }
 631 }
 632
 633 void colo_compare_register_notifier(Notifier *notify)
 634 {
 635     notifier_list_add(&colo_compare_notifiers, notify);
 636 }
 637
 638 void colo_compare_unregister_notifier(Notifier *notify)
 639 {
 640     notifier_remove(notify);
 641 }
 642
 643 static int colo_old_packet_check_one_conn(Connection *conn,
 644                                           CompareState *s)
 645 {
 646     GList *result = NULL;
 647
 648     result = g_queue_find_custom(&conn->primary_list,
 649                                  &s->compare_timeout,
 650                                  (GCompareFunc)colo_old_packet_check_one);
 651
 652     if (result) {
 653         /* Do checkpoint will flush old packet */
 654         colo_compare_inconsistency_notify(s);
 655         return 0;
 656     }
 657
 658     return 1;
 659 }
 660
 661 /*
 662  * Look for old packets that the secondary hasn't matched,
 663  * if we have some then we have to checkpoint to wake
 664  * the secondary up.
 665  */
 666 static void colo_old_packet_check(void *opaque)
 667 {
 668     CompareState *s = opaque;
 669
 670     /*
 671      * If we find one old packet, stop finding job and notify
 672      * COLO frame do checkpoint.
 673      */
 674     g_queue_find_custom(&s->conn_list, s,
 675                         (GCompareFunc)colo_old_packet_check_one_conn);
 676 }
 677
 678 static void colo_compare_packet(CompareState *s, Connection *conn,
 679                                 int (*HandlePacket)(Packet *spkt,
 680                                 Packet *ppkt))
 681 {
 682     Packet *pkt = NULL;
 683     GList *result = NULL;
 684
 685     while (!g_queue_is_empty(&conn->primary_list) &&
 686            !g_queue_is_empty(&conn->secondary_list)) {
 687         pkt = g_queue_pop_head(&conn->primary_list);
 688         result = g_queue_find_custom(&conn->secondary_list,
 689                  pkt, (GCompareFunc)HandlePacket);
 690
 691         if (result) {
 692             colo_release_primary_pkt(s, pkt);
 693             g_queue_remove(&conn->secondary_list, result->data);
 694         } else {
 695             /*
 696              * If one packet arrive late, the secondary_list or
 697              * primary_list will be empty, so we can't compare it
 698              * until next comparison. If the packets in the list are
 699              * timeout, it will trigger a checkpoint request.
 700              */
 701             trace_colo_compare_main("packet different");
 702             g_queue_push_head(&conn->primary_list, pkt);
 703
 704             colo_compare_inconsistency_notify(s);
 705             break;
 706         }
 707     }
 708 }
 709
 710 /*
 711  * Called from the compare thread on the primary
 712  * for compare packet with secondary list of the
 713  * specified connection when a new packet was
 714  * queued to it.
 715  */
 716 static void colo_compare_connection(void *opaque, void *user_data)
 717 {
 718     CompareState *s = user_data;
 719     Connection *conn = opaque;
 720
 721     switch (conn->ip_proto) {
 722     case IPPROTO_TCP:
 723         colo_compare_tcp(s, conn);
 724         break;
 725     case IPPROTO_UDP:
 726         colo_compare_packet(s, conn, colo_packet_compare_udp);
 727         break;
 728     case IPPROTO_ICMP:
 729         colo_compare_packet(s, conn, colo_packet_compare_icmp);
 730         break;
 731     default:
 732         colo_compare_packet(s, conn, colo_packet_compare_other);
 733         break;
 734     }
 735 }
 736
 737 static void coroutine_fn _compare_chr_send(void *opaque)
 738 {
 739     SendCo *sendco = opaque;
 740     CompareState *s = sendco->s;
 741     int ret = 0;
 742
 743     while (!g_queue_is_empty(&sendco->send_list)) {
 744         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
 745         uint32_t len = htonl(entry->size);
 746
 747         ret = qemu_chr_fe_write_all(sendco->chr, (uint8_t *)&len, sizeof(len));
 748
 749         if (ret != sizeof(len)) {
 750             g_free(entry->buf);
 751             g_slice_free(SendEntry, entry);
 752             goto err;
 753         }
 754
 755         if (!sendco->notify_remote_frame && s->vnet_hdr) {
 756             /*
 757              * We send vnet header len make other module(like filter-redirector)
 758              * know how to parse net packet correctly.
 759              */
 760             len = htonl(entry->vnet_hdr_len);
 761
 762             ret = qemu_chr_fe_write_all(sendco->chr,
 763                                         (uint8_t *)&len,
 764                                         sizeof(len));
 765
 766             if (ret != sizeof(len)) {
 767                 g_free(entry->buf);
 768                 g_slice_free(SendEntry, entry);
 769                 goto err;
 770             }
 771         }
 772
 773         ret = qemu_chr_fe_write_all(sendco->chr,
 774                                     (uint8_t *)entry->buf,
 775                                     entry->size);
 776
 777         if (ret != entry->size) {
 778             g_free(entry->buf);
 779             g_slice_free(SendEntry, entry);
 780             goto err;
 781         }
 782
 783         g_free(entry->buf);
 784         g_slice_free(SendEntry, entry);
 785     }
 786
 787     sendco->ret = 0;
 788     goto out;
 789
 790 err:
 791     while (!g_queue_is_empty(&sendco->send_list)) {
 792         SendEntry *entry = g_queue_pop_tail(&sendco->send_list);
 793         g_free(entry->buf);
 794         g_slice_free(SendEntry, entry);
 795     }
 796     sendco->ret = ret < 0 ? ret : -EIO;
 797 out:
 798     sendco->co = NULL;
 799     sendco->done = true;
 800     aio_wait_kick();
 801 }
 802
 803 static int compare_chr_send(CompareState *s,
 804                             uint8_t *buf,
 805                             uint32_t size,
 806                             uint32_t vnet_hdr_len,
 807                             bool notify_remote_frame,
 808                             bool zero_copy)
 809 {
 810     SendCo *sendco;
 811     SendEntry *entry;
 812
 813     if (notify_remote_frame) {
 814         sendco = &s->notify_sendco;
 815     } else {
 816         sendco = &s->out_sendco;
 817     }
 818
 819     if (!size) {
 820         return 0;
 821     }
 822
 823     entry = g_slice_new(SendEntry);
 824     entry->size = size;
 825     entry->vnet_hdr_len = vnet_hdr_len;
 826     if (zero_copy) {
 827         entry->buf = buf;
 828     } else {
 829         entry->buf = g_malloc(size);
 830         memcpy(entry->buf, buf, size);
 831     }
 832     g_queue_push_head(&sendco->send_list, entry);
 833
 834     if (sendco->done) {
 835         sendco->co = qemu_coroutine_create(_compare_chr_send, sendco);
 836         sendco->done = false;
 837         qemu_coroutine_enter(sendco->co);
 838         if (sendco->done) {
 839             /* report early errors */
 840             return sendco->ret;
 841         }
 842     }
 843
 844     /* assume success */
 845     return 0;
 846 }
 847
 848 static int compare_chr_can_read(void *opaque)
 849 {
 850     return COMPARE_READ_LEN_MAX;
 851 }
 852
 853 /*
 854  * Called from the main thread on the primary for packets
 855  * arriving over the socket from the primary.
 856  */
 857 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
 858 {
 859     CompareState *s = COLO_COMPARE(opaque);
 860     int ret;
 861
 862     ret = net_fill_rstate(&s->pri_rs, buf, size);
 863     if (ret == -1) {
 864         qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
 865                                  NULL, NULL, true);
 866         error_report("colo-compare primary_in error");
 867     }
 868 }
 869
 870 /*
 871  * Called from the main thread on the primary for packets
 872  * arriving over the socket from the secondary.
 873  */
 874 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 875 {
 876     CompareState *s = COLO_COMPARE(opaque);
 877     int ret;
 878
 879     ret = net_fill_rstate(&s->sec_rs, buf, size);
 880     if (ret == -1) {
 881         qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
 882                                  NULL, NULL, true);
 883         error_report("colo-compare secondary_in error");
 884     }
 885 }
 886
 887 static void compare_notify_chr(void *opaque, const uint8_t *buf, int size)
 888 {
 889     CompareState *s = COLO_COMPARE(opaque);
 890     int ret;
 891
 892     ret = net_fill_rstate(&s->notify_rs, buf, size);
 893     if (ret == -1) {
 894         qemu_chr_fe_set_handlers(&s->chr_notify_dev, NULL, NULL, NULL, NULL,
 895                                  NULL, NULL, true);
 896         error_report("colo-compare notify_dev error");
 897     }
 898 }
 899
 900 /*
 901  * Check old packet regularly so it can watch for any packets
 902  * that the secondary hasn't produced equivalents of.
 903  */
 904 static void check_old_packet_regular(void *opaque)
 905 {
 906     CompareState *s = opaque;
 907
 908     /* if have old packet we will notify checkpoint */
 909     colo_old_packet_check(s);
 910     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 911               s->expired_scan_cycle);
 912 }
 913
 914 /* Public API, Used for COLO frame to notify compare event */
 915 void colo_notify_compares_event(void *opaque, int event, Error **errp)
 916 {
 917     CompareState *s;
 918     qemu_mutex_lock(&colo_compare_mutex);
 919
 920     if (!colo_compare_active) {
 921         qemu_mutex_unlock(&colo_compare_mutex);
 922         return;
 923     }
 924
 925     qemu_mutex_lock(&event_mtx);
 926     QTAILQ_FOREACH(s, &net_compares, next) {
 927         s->event = event;
 928         qemu_bh_schedule(s->event_bh);
 929         event_unhandled_count++;
 930     }
 931     /* Wait all compare threads to finish handling this event */
 932     while (event_unhandled_count > 0) {
 933         qemu_cond_wait(&event_complete_cond, &event_mtx);
 934     }
 935
 936     qemu_mutex_unlock(&event_mtx);
 937     qemu_mutex_unlock(&colo_compare_mutex);
 938 }
 939
 940 static void colo_compare_timer_init(CompareState *s)
 941 {
 942     AioContext *ctx = iothread_get_aio_context(s->iothread);
 943
 944     s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
 945                                 SCALE_MS, check_old_packet_regular,
 946                                 s);
 947     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 948               s->expired_scan_cycle);
 949 }
 950
 951 static void colo_compare_timer_del(CompareState *s)
 952 {
 953     if (s->packet_check_timer) {
 954         timer_del(s->packet_check_timer);
 955         timer_free(s->packet_check_timer);
 956         s->packet_check_timer = NULL;
 957     }
 958  }
 959
 960 static void colo_flush_packets(void *opaque, void *user_data);
 961
 962 static void colo_compare_handle_event(void *opaque)
 963 {
 964     CompareState *s = opaque;
 965
 966     switch (s->event) {
 967     case COLO_EVENT_CHECKPOINT:
 968         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
 969         break;
 970     case COLO_EVENT_FAILOVER:
 971         break;
 972     default:
 973         break;
 974     }
 975
 976     qemu_mutex_lock(&event_mtx);
 977     assert(event_unhandled_count > 0);
 978     event_unhandled_count--;
 979     qemu_cond_broadcast(&event_complete_cond);
 980     qemu_mutex_unlock(&event_mtx);
 981 }
 982
 983 static void colo_compare_iothread(CompareState *s)
 984 {
 985     AioContext *ctx = iothread_get_aio_context(s->iothread);
 986     object_ref(OBJECT(s->iothread));
 987     s->worker_context = iothread_get_g_main_context(s->iothread);
 988
 989     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
 990                              compare_pri_chr_in, NULL, NULL,
 991                              s, s->worker_context, true);
 992     qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
 993                              compare_sec_chr_in, NULL, NULL,
 994                              s, s->worker_context, true);
 995     if (s->notify_dev) {
 996         qemu_chr_fe_set_handlers(&s->chr_notify_dev, compare_chr_can_read,
 997                                  compare_notify_chr, NULL, NULL,
 998                                  s, s->worker_context, true);
 999     }
1000
1001     colo_compare_timer_init(s);
1002     s->event_bh = aio_bh_new(ctx, colo_compare_handle_event, s);
1003 }
1004
1005 static char *compare_get_pri_indev(Object *obj, Error **errp)
1006 {
1007     CompareState *s = COLO_COMPARE(obj);
1008
1009     return g_strdup(s->pri_indev);
1010 }
1011
1012 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
1013 {
1014     CompareState *s = COLO_COMPARE(obj);
1015
1016     g_free(s->pri_indev);
1017     s->pri_indev = g_strdup(value);
1018 }
1019
1020 static char *compare_get_sec_indev(Object *obj, Error **errp)
1021 {
1022     CompareState *s = COLO_COMPARE(obj);
1023
1024     return g_strdup(s->sec_indev);
1025 }
1026
1027 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
1028 {
1029     CompareState *s = COLO_COMPARE(obj);
1030
1031     g_free(s->sec_indev);
1032     s->sec_indev = g_strdup(value);
1033 }
1034
1035 static char *compare_get_outdev(Object *obj, Error **errp)
1036 {
1037     CompareState *s = COLO_COMPARE(obj);
1038
1039     return g_strdup(s->outdev);
1040 }
1041
1042 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
1043 {
1044     CompareState *s = COLO_COMPARE(obj);
1045
1046     g_free(s->outdev);
1047     s->outdev = g_strdup(value);
1048 }
1049
1050 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
1051 {
1052     CompareState *s = COLO_COMPARE(obj);
1053
1054     return s->vnet_hdr;
1055 }
1056
1057 static void compare_set_vnet_hdr(Object *obj,
1058                                  bool value,
1059                                  Error **errp)
1060 {
1061     CompareState *s = COLO_COMPARE(obj);
1062
1063     s->vnet_hdr = value;
1064 }
1065
1066 static char *compare_get_notify_dev(Object *obj, Error **errp)
1067 {
1068     CompareState *s = COLO_COMPARE(obj);
1069
1070     return g_strdup(s->notify_dev);
1071 }
1072
1073 static void compare_set_notify_dev(Object *obj, const char *value, Error **errp)
1074 {
1075     CompareState *s = COLO_COMPARE(obj);
1076
1077     g_free(s->notify_dev);
1078     s->notify_dev = g_strdup(value);
1079 }
1080
1081 static void compare_get_timeout(Object *obj, Visitor *v,
1082                                 const char *name, void *opaque,
1083                                 Error **errp)
1084 {
1085     CompareState *s = COLO_COMPARE(obj);
1086     uint32_t value = s->compare_timeout;
1087
1088     visit_type_uint32(v, name, &value, errp);
1089 }
1090
1091 static void compare_set_timeout(Object *obj, Visitor *v,
1092                                 const char *name, void *opaque,
1093                                 Error **errp)
1094 {
1095     CompareState *s = COLO_COMPARE(obj);
1096     uint32_t value;
1097
1098     if (!visit_type_uint32(v, name, &value, errp)) {
1099         return;
1100     }
1101     if (!value) {
1102         error_setg(errp, "Property '%s.%s' requires a positive value",
1103                    object_get_typename(obj), name);
1104         return;
1105     }
1106     s->compare_timeout = value;
1107 }
1108
1109 static void compare_get_expired_scan_cycle(Object *obj, Visitor *v,
1110                                            const char *name, void *opaque,
1111                                            Error **errp)
1112 {
1113     CompareState *s = COLO_COMPARE(obj);
1114     uint32_t value = s->expired_scan_cycle;
1115
1116     visit_type_uint32(v, name, &value, errp);
1117 }
1118
1119 static void compare_set_expired_scan_cycle(Object *obj, Visitor *v,
1120                                            const char *name, void *opaque,
1121                                            Error **errp)
1122 {
1123     CompareState *s = COLO_COMPARE(obj);
1124     uint32_t value;
1125
1126     if (!visit_type_uint32(v, name, &value, errp)) {
1127         return;
1128     }
1129     if (!value) {
1130         error_setg(errp, "Property '%s.%s' requires a positive value",
1131                    object_get_typename(obj), name);
1132         return;
1133     }
1134     s->expired_scan_cycle = value;
1135 }
1136
1137 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
1138 {
1139     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
1140     Connection *conn = NULL;
1141
1142     if (packet_enqueue(s, PRIMARY_IN, &conn)) {
1143         trace_colo_compare_main("primary: unsupported packet in");
1144         compare_chr_send(s,
1145                          pri_rs->buf,
1146                          pri_rs->packet_len,
1147                          pri_rs->vnet_hdr_len,
1148                          false,
1149                          false);
1150     } else {
1151         /* compare packet in the specified connection */
1152         colo_compare_connection(conn, s);
1153     }
1154 }
1155
1156 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
1157 {
1158     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
1159     Connection *conn = NULL;
1160
1161     if (packet_enqueue(s, SECONDARY_IN, &conn)) {
1162         trace_colo_compare_main("secondary: unsupported packet in");
1163     } else {
1164         /* compare packet in the specified connection */
1165         colo_compare_connection(conn, s);
1166     }
1167 }
1168
1169 static void compare_notify_rs_finalize(SocketReadState *notify_rs)
1170 {
1171     CompareState *s = container_of(notify_rs, CompareState, notify_rs);
1172
1173     const char msg[] = "COLO_COMPARE_GET_XEN_INIT";
1174     int ret;
1175
1176     if (packet_matches_str("COLO_USERSPACE_PROXY_INIT",
1177                            notify_rs->buf,
1178                            notify_rs->packet_len)) {
1179         ret = compare_chr_send(s, (uint8_t *)msg, strlen(msg), 0, true, false);
1180         if (ret < 0) {
1181             error_report("Notify Xen COLO-frame INIT failed");
1182         }
1183     } else if (packet_matches_str("COLO_CHECKPOINT",
1184                                   notify_rs->buf,
1185                                   notify_rs->packet_len)) {
1186         /* colo-compare do checkpoint, flush pri packet and remove sec packet */
1187         g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1188     } else {
1189         error_report("COLO compare got unsupported instruction");
1190     }
1191 }
1192
1193 /*
1194  * Return 0 is success.
1195  * Return 1 is failed.
1196  */
1197 static int find_and_check_chardev(Chardev **chr,
1198                                   char *chr_name,
1199                                   Error **errp)
1200 {
1201     *chr = qemu_chr_find(chr_name);
1202     if (*chr == NULL) {
1203         error_setg(errp, "Device '%s' not found",
1204                    chr_name);
1205         return 1;
1206     }
1207
1208     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
1209         error_setg(errp, "chardev \"%s\" is not reconnectable",
1210                    chr_name);
1211         return 1;
1212     }
1213
1214     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_GCONTEXT)) {
1215         error_setg(errp, "chardev \"%s\" cannot switch context",
1216                    chr_name);
1217         return 1;
1218     }
1219
1220     return 0;
1221 }
1222
1223 /*
1224  * Called from the main thread on the primary
1225  * to setup colo-compare.
1226  */
1227 static void colo_compare_complete(UserCreatable *uc, Error **errp)
1228 {
1229     CompareState *s = COLO_COMPARE(uc);
1230     Chardev *chr;
1231
1232     if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
1233         error_setg(errp, "colo compare needs 'primary_in' ,"
1234                    "'secondary_in','outdev','iothread' property set");
1235         return;
1236     } else if (!strcmp(s->pri_indev, s->outdev) ||
1237                !strcmp(s->sec_indev, s->outdev) ||
1238                !strcmp(s->pri_indev, s->sec_indev)) {
1239         error_setg(errp, "'indev' and 'outdev' could not be same "
1240                    "for compare module");
1241         return;
1242     }
1243
1244     if (!s->compare_timeout) {
1245         /* Set default value to 3000 MS */
1246         s->compare_timeout = DEFAULT_TIME_OUT_MS;
1247     }
1248
1249     if (!s->expired_scan_cycle) {
1250         /* Set default value to 3000 MS */
1251         s->expired_scan_cycle = REGULAR_PACKET_CHECK_MS;
1252     }
1253
1254     if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
1255         !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
1256         return;
1257     }
1258
1259     if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
1260         !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
1261         return;
1262     }
1263
1264     if (find_and_check_chardev(&chr, s->outdev, errp) ||
1265         !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
1266         return;
1267     }
1268
1269     net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
1270     net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
1271
1272     /* Try to enable remote notify chardev, currently just for Xen COLO */
1273     if (s->notify_dev) {
1274         if (find_and_check_chardev(&chr, s->notify_dev, errp) ||
1275             !qemu_chr_fe_init(&s->chr_notify_dev, chr, errp)) {
1276             return;
1277         }
1278
1279         net_socket_rs_init(&s->notify_rs, compare_notify_rs_finalize,
1280                            s->vnet_hdr);
1281     }
1282
1283     s->out_sendco.s = s;
1284     s->out_sendco.chr = &s->chr_out;
1285     s->out_sendco.notify_remote_frame = false;
1286     s->out_sendco.done = true;
1287     g_queue_init(&s->out_sendco.send_list);
1288
1289     if (s->notify_dev) {
1290         s->notify_sendco.s = s;
1291         s->notify_sendco.chr = &s->chr_notify_dev;
1292         s->notify_sendco.notify_remote_frame = true;
1293         s->notify_sendco.done = true;
1294         g_queue_init(&s->notify_sendco.send_list);
1295     }
1296
1297     g_queue_init(&s->conn_list);
1298
1299     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
1300                                                       connection_key_equal,
1301                                                       g_free,
1302                                                       connection_destroy);
1303
1304     colo_compare_iothread(s);
1305
1306     qemu_mutex_lock(&colo_compare_mutex);
1307     if (!colo_compare_active) {
1308         qemu_mutex_init(&event_mtx);
1309         qemu_cond_init(&event_complete_cond);
1310         colo_compare_active = true;
1311     }
1312     QTAILQ_INSERT_TAIL(&net_compares, s, next);
1313     qemu_mutex_unlock(&colo_compare_mutex);
1314
1315     return;
1316 }
1317
1318 static void colo_flush_packets(void *opaque, void *user_data)
1319 {
1320     CompareState *s = user_data;
1321     Connection *conn = opaque;
1322     Packet *pkt = NULL;
1323
1324     while (!g_queue_is_empty(&conn->primary_list)) {
1325         pkt = g_queue_pop_head(&conn->primary_list);
1326         compare_chr_send(s,
1327                          pkt->data,
1328                          pkt->size,
1329                          pkt->vnet_hdr_len,
1330                          false,
1331                          true);
1332         packet_destroy_partial(pkt, NULL);
1333     }
1334     while (!g_queue_is_empty(&conn->secondary_list)) {
1335         pkt = g_queue_pop_head(&conn->secondary_list);
1336         packet_destroy(pkt, NULL);
1337     }
1338 }
1339
1340 static void colo_compare_class_init(ObjectClass *oc, void *data)
1341 {
1342     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
1343
1344     ucc->complete = colo_compare_complete;
1345 }
1346
1347 static void colo_compare_init(Object *obj)
1348 {
1349     CompareState *s = COLO_COMPARE(obj);
1350
1351     object_property_add_str(obj, "primary_in",
1352                             compare_get_pri_indev, compare_set_pri_indev);
1353     object_property_add_str(obj, "secondary_in",
1354                             compare_get_sec_indev, compare_set_sec_indev);
1355     object_property_add_str(obj, "outdev",
1356                             compare_get_outdev, compare_set_outdev);
1357     object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
1358                             (Object **)&s->iothread,
1359                             object_property_allow_set_link,
1360                             OBJ_PROP_LINK_STRONG);
1361     /* This parameter just for Xen COLO */
1362     object_property_add_str(obj, "notify_dev",
1363                             compare_get_notify_dev, compare_set_notify_dev);
1364
1365     object_property_add(obj, "compare_timeout", "uint32",
1366                         compare_get_timeout,
1367                         compare_set_timeout, NULL, NULL);
1368
1369     object_property_add(obj, "expired_scan_cycle", "uint32",
1370                         compare_get_expired_scan_cycle,
1371                         compare_set_expired_scan_cycle, NULL, NULL);
1372
1373     s->vnet_hdr = false;
1374     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
1375                              compare_set_vnet_hdr);
1376 }
1377
1378 static void colo_compare_finalize(Object *obj)
1379 {
1380     CompareState *s = COLO_COMPARE(obj);
1381     CompareState *tmp = NULL;
1382
1383     qemu_mutex_lock(&colo_compare_mutex);
1384     QTAILQ_FOREACH(tmp, &net_compares, next) {
1385         if (tmp == s) {
1386             QTAILQ_REMOVE(&net_compares, s, next);
1387             break;
1388         }
1389     }
1390     if (QTAILQ_EMPTY(&net_compares)) {
1391         colo_compare_active = false;
1392         qemu_mutex_destroy(&event_mtx);
1393         qemu_cond_destroy(&event_complete_cond);
1394     }
1395     qemu_mutex_unlock(&colo_compare_mutex);
1396
1397     qemu_chr_fe_deinit(&s->chr_pri_in, false);
1398     qemu_chr_fe_deinit(&s->chr_sec_in, false);
1399     qemu_chr_fe_deinit(&s->chr_out, false);
1400     if (s->notify_dev) {
1401         qemu_chr_fe_deinit(&s->chr_notify_dev, false);
1402     }
1403
1404     if (s->iothread) {
1405         colo_compare_timer_del(s);
1406     }
1407
1408     qemu_bh_delete(s->event_bh);
1409
1410     AioContext *ctx = iothread_get_aio_context(s->iothread);
1411     aio_context_acquire(ctx);
1412     AIO_WAIT_WHILE(ctx, !s->out_sendco.done);
1413     if (s->notify_dev) {
1414         AIO_WAIT_WHILE(ctx, !s->notify_sendco.done);
1415     }
1416     aio_context_release(ctx);
1417
1418     /* Release all unhandled packets after compare thead exited */
1419     g_queue_foreach(&s->conn_list, colo_flush_packets, s);
1420     AIO_WAIT_WHILE(NULL, !s->out_sendco.done);
1421
1422     g_queue_clear(&s->conn_list);
1423     g_queue_clear(&s->out_sendco.send_list);
1424     if (s->notify_dev) {
1425         g_queue_clear(&s->notify_sendco.send_list);
1426     }
1427
1428     if (s->connection_track_table) {
1429         g_hash_table_destroy(s->connection_track_table);
1430     }
1431
1432     if (s->iothread) {
1433         object_unref(OBJECT(s->iothread));
1434     }
1435
1436     g_free(s->pri_indev);
1437     g_free(s->sec_indev);
1438     g_free(s->outdev);
1439     g_free(s->notify_dev);
1440 }
1441
1442 static void __attribute__((__constructor__)) colo_compare_init_globals(void)
1443 {
1444     colo_compare_active = false;
1445     qemu_mutex_init(&colo_compare_mutex);
1446 }
1447
1448 static const TypeInfo colo_compare_info = {
1449     .name = TYPE_COLO_COMPARE,
1450     .parent = TYPE_OBJECT,
1451     .instance_size = sizeof(CompareState),
1452     .instance_init = colo_compare_init,
1453     .instance_finalize = colo_compare_finalize,
1454     .class_size = sizeof(CompareClass),
1455     .class_init = colo_compare_class_init,
1456     .interfaces = (InterfaceInfo[]) {
1457         { TYPE_USER_CREATABLE },
1458         { }
1459     }
1460 };
1461
1462 static void register_types(void)
1463 {
1464     type_register_static(&colo_compare_info);
1465 }
1466
1467 type_init(register_types);