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