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/error-report.h"
  17 #include "trace.h"
  18 #include "qemu-common.h"
  19 #include "qapi/qmp/qerror.h"
  20 #include "qapi/error.h"
  21 #include "net/net.h"
  22 #include "net/eth.h"
  23 #include "qom/object_interfaces.h"
  24 #include "qemu/iov.h"
  25 #include "qom/object.h"
  26 #include "qemu/typedefs.h"
  27 #include "net/queue.h"
  28 #include "sysemu/char.h"
  29 #include "qemu/sockets.h"
  30 #include "qapi-visit.h"
  31 #include "net/colo.h"
  32
  33 #define TYPE_COLO_COMPARE "colo-compare"
  34 #define COLO_COMPARE(obj) \
  35     OBJECT_CHECK(CompareState, (obj), TYPE_COLO_COMPARE)
  36
  37 #define COMPARE_READ_LEN_MAX NET_BUFSIZE
  38 #define MAX_QUEUE_SIZE 1024
  39
  40 /* TODO: Should be configurable */
  41 #define REGULAR_PACKET_CHECK_MS 3000
  42
  43 /*
  44   + CompareState ++
  45   |               |
  46   +---------------+   +---------------+         +---------------+
  47   |conn list      +--->conn           +--------->conn           |
  48   +---------------+   +---------------+         +---------------+
  49   |               |     |           |             |          |
  50   +---------------+ +---v----+  +---v----+    +---v----+ +---v----+
  51                     |primary |  |secondary    |primary | |secondary
  52                     |packet  |  |packet  +    |packet  | |packet  +
  53                     +--------+  +--------+    +--------+ +--------+
  54                         |           |             |          |
  55                     +---v----+  +---v----+    +---v----+ +---v----+
  56                     |primary |  |secondary    |primary | |secondary
  57                     |packet  |  |packet  +    |packet  | |packet  +
  58                     +--------+  +--------+    +--------+ +--------+
  59                         |           |             |          |
  60                     +---v----+  +---v----+    +---v----+ +---v----+
  61                     |primary |  |secondary    |primary | |secondary
  62                     |packet  |  |packet  +    |packet  | |packet  +
  63                     +--------+  +--------+    +--------+ +--------+
  64 */
  65 typedef struct CompareState {
  66     Object parent;
  67
  68     char *pri_indev;
  69     char *sec_indev;
  70     char *outdev;
  71     CharBackend chr_pri_in;
  72     CharBackend chr_sec_in;
  73     CharBackend chr_out;
  74     SocketReadState pri_rs;
  75     SocketReadState sec_rs;
  76
  77     /* connection list: the connections belonged to this NIC could be found
  78      * in this list.
  79      * element type: Connection
  80      */
  81     GQueue conn_list;
  82     /* hashtable to save connection */
  83     GHashTable *connection_track_table;
  84     /* compare thread, a thread for each NIC */
  85     QemuThread thread;
  86     /* Timer used on the primary to find packets that are never matched */
  87     QEMUTimer *timer;
  88     QemuMutex timer_check_lock;
  89 } CompareState;
  90
  91 typedef struct CompareClass {
  92     ObjectClass parent_class;
  93 } CompareClass;
  94
  95 typedef struct CompareChardevProps {
  96     bool is_socket;
  97 } CompareChardevProps;
  98
  99 enum {
 100     PRIMARY_IN = 0,
 101     SECONDARY_IN,
 102 };
 103
 104 static int compare_chr_send(CharBackend *out,
 105                             const uint8_t *buf,
 106                             uint32_t size);
 107
 108 /*
 109  * Return 0 on success, if return -1 means the pkt
 110  * is unsupported(arp and ipv6) and will be sent later
 111  */
 112 static int packet_enqueue(CompareState *s, int mode)
 113 {
 114     ConnectionKey key;
 115     Packet *pkt = NULL;
 116     Connection *conn;
 117
 118     if (mode == PRIMARY_IN) {
 119         pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
 120     } else {
 121         pkt = packet_new(s->sec_rs.buf, s->sec_rs.packet_len);
 122     }
 123
 124     if (parse_packet_early(pkt)) {
 125         packet_destroy(pkt, NULL);
 126         pkt = NULL;
 127         return -1;
 128     }
 129     fill_connection_key(pkt, &key);
 130
 131     conn = connection_get(s->connection_track_table,
 132                           &key,
 133                           &s->conn_list);
 134
 135     if (!conn->processing) {
 136         g_queue_push_tail(&s->conn_list, conn);
 137         conn->processing = true;
 138     }
 139
 140     if (mode == PRIMARY_IN) {
 141         if (g_queue_get_length(&conn->primary_list) <=
 142                                MAX_QUEUE_SIZE) {
 143             g_queue_push_tail(&conn->primary_list, pkt);
 144         } else {
 145             error_report("colo compare primary queue size too big,"
 146                          "drop packet");
 147         }
 148     } else {
 149         if (g_queue_get_length(&conn->secondary_list) <=
 150                                MAX_QUEUE_SIZE) {
 151             g_queue_push_tail(&conn->secondary_list, pkt);
 152         } else {
 153             error_report("colo compare secondary queue size too big,"
 154                          "drop packet");
 155         }
 156     }
 157
 158     return 0;
 159 }
 160
 161 /*
 162  * The IP packets sent by primary and secondary
 163  * will be compared in here
 164  * TODO support ip fragment, Out-Of-Order
 165  * return:    0  means packet same
 166  *            > 0 || < 0 means packet different
 167  */
 168 static int colo_packet_compare(Packet *ppkt, Packet *spkt)
 169 {
 170     trace_colo_compare_ip_info(ppkt->size, inet_ntoa(ppkt->ip->ip_src),
 171                                inet_ntoa(ppkt->ip->ip_dst), spkt->size,
 172                                inet_ntoa(spkt->ip->ip_src),
 173                                inet_ntoa(spkt->ip->ip_dst));
 174
 175     if (ppkt->size == spkt->size) {
 176         return memcmp(ppkt->data, spkt->data, spkt->size);
 177     } else {
 178         return -1;
 179     }
 180 }
 181
 182 /*
 183  * Called from the compare thread on the primary
 184  * for compare tcp packet
 185  * compare_tcp copied from Dr. David Alan Gilbert's branch
 186  */
 187 static int colo_packet_compare_tcp(Packet *spkt, Packet *ppkt)
 188 {
 189     struct tcphdr *ptcp, *stcp;
 190     int res;
 191     char *sdebug, *ddebug;
 192
 193     trace_colo_compare_main("compare tcp");
 194     if (ppkt->size != spkt->size) {
 195         if (trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
 196             trace_colo_compare_main("pkt size not same");
 197         }
 198         return -1;
 199     }
 200
 201     ptcp = (struct tcphdr *)ppkt->transport_header;
 202     stcp = (struct tcphdr *)spkt->transport_header;
 203
 204     /*
 205      * The 'identification' field in the IP header is *very* random
 206      * it almost never matches.  Fudge this by ignoring differences in
 207      * unfragmented packets; they'll normally sort themselves out if different
 208      * anyway, and it should recover at the TCP level.
 209      * An alternative would be to get both the primary and secondary to rewrite
 210      * somehow; but that would need some sync traffic to sync the state
 211      */
 212     if (ntohs(ppkt->ip->ip_off) & IP_DF) {
 213         spkt->ip->ip_id = ppkt->ip->ip_id;
 214         /* and the sum will be different if the IDs were different */
 215         spkt->ip->ip_sum = ppkt->ip->ip_sum;
 216     }
 217
 218     res = memcmp(ppkt->data + ETH_HLEN, spkt->data + ETH_HLEN,
 219                 (spkt->size - ETH_HLEN));
 220
 221     if (res != 0 && trace_event_get_state(TRACE_COLO_COMPARE_MISCOMPARE)) {
 222         sdebug = strdup(inet_ntoa(ppkt->ip->ip_src));
 223         ddebug = strdup(inet_ntoa(ppkt->ip->ip_dst));
 224         fprintf(stderr, "%s: src/dst: %s/%s p: seq/ack=%u/%u"
 225                 " s: seq/ack=%u/%u res=%d flags=%x/%x\n",
 226                 __func__, sdebug, ddebug,
 227                 (unsigned int)ntohl(ptcp->th_seq),
 228                 (unsigned int)ntohl(ptcp->th_ack),
 229                 (unsigned int)ntohl(stcp->th_seq),
 230                 (unsigned int)ntohl(stcp->th_ack),
 231                 res, ptcp->th_flags, stcp->th_flags);
 232
 233         fprintf(stderr, "Primary len = %d\n", ppkt->size);
 234         qemu_hexdump((char *)ppkt->data, stderr, "colo-compare", ppkt->size);
 235         fprintf(stderr, "Secondary len = %d\n", spkt->size);
 236         qemu_hexdump((char *)spkt->data, stderr, "colo-compare", spkt->size);
 237
 238         g_free(sdebug);
 239         g_free(ddebug);
 240     }
 241
 242     return res;
 243 }
 244
 245 /*
 246  * Called from the compare thread on the primary
 247  * for compare udp packet
 248  */
 249 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
 250 {
 251     int ret;
 252
 253     trace_colo_compare_main("compare udp");
 254     ret = colo_packet_compare(ppkt, spkt);
 255
 256     if (ret) {
 257         trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
 258         qemu_hexdump((char *)ppkt->data, stderr, "colo-compare", ppkt->size);
 259         trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
 260         qemu_hexdump((char *)spkt->data, stderr, "colo-compare", spkt->size);
 261     }
 262
 263     return ret;
 264 }
 265
 266 /*
 267  * Called from the compare thread on the primary
 268  * for compare icmp packet
 269  */
 270 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
 271 {
 272     int network_length;
 273
 274     trace_colo_compare_main("compare icmp");
 275     network_length = ppkt->ip->ip_hl * 4;
 276     if (ppkt->size != spkt->size ||
 277         ppkt->size < network_length + ETH_HLEN) {
 278         return -1;
 279     }
 280
 281     if (colo_packet_compare(ppkt, spkt)) {
 282         trace_colo_compare_icmp_miscompare("primary pkt size",
 283                                            ppkt->size);
 284         qemu_hexdump((char *)ppkt->data, stderr, "colo-compare",
 285                      ppkt->size);
 286         trace_colo_compare_icmp_miscompare("Secondary pkt size",
 287                                            spkt->size);
 288         qemu_hexdump((char *)spkt->data, stderr, "colo-compare",
 289                      spkt->size);
 290         return -1;
 291     } else {
 292         return 0;
 293     }
 294 }
 295
 296 /*
 297  * Called from the compare thread on the primary
 298  * for compare other packet
 299  */
 300 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
 301 {
 302     trace_colo_compare_main("compare other");
 303     trace_colo_compare_ip_info(ppkt->size, inet_ntoa(ppkt->ip->ip_src),
 304                                inet_ntoa(ppkt->ip->ip_dst), spkt->size,
 305                                inet_ntoa(spkt->ip->ip_src),
 306                                inet_ntoa(spkt->ip->ip_dst));
 307     return colo_packet_compare(ppkt, spkt);
 308 }
 309
 310 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
 311 {
 312     int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 313
 314     if ((now - pkt->creation_ms) > (*check_time)) {
 315         trace_colo_old_packet_check_found(pkt->creation_ms);
 316         return 0;
 317     } else {
 318         return 1;
 319     }
 320 }
 321
 322 static void colo_old_packet_check_one_conn(void *opaque,
 323                                            void *user_data)
 324 {
 325     Connection *conn = opaque;
 326     GList *result = NULL;
 327     int64_t check_time = REGULAR_PACKET_CHECK_MS;
 328
 329     result = g_queue_find_custom(&conn->primary_list,
 330                                  &check_time,
 331                                  (GCompareFunc)colo_old_packet_check_one);
 332
 333     if (result) {
 334         /* do checkpoint will flush old packet */
 335         /* TODO: colo_notify_checkpoint();*/
 336     }
 337 }
 338
 339 /*
 340  * Look for old packets that the secondary hasn't matched,
 341  * if we have some then we have to checkpoint to wake
 342  * the secondary up.
 343  */
 344 static void colo_old_packet_check(void *opaque)
 345 {
 346     CompareState *s = opaque;
 347
 348     g_queue_foreach(&s->conn_list, colo_old_packet_check_one_conn, NULL);
 349 }
 350
 351 /*
 352  * Called from the compare thread on the primary
 353  * for compare connection
 354  */
 355 static void colo_compare_connection(void *opaque, void *user_data)
 356 {
 357     CompareState *s = user_data;
 358     Connection *conn = opaque;
 359     Packet *pkt = NULL;
 360     GList *result = NULL;
 361     int ret;
 362
 363     while (!g_queue_is_empty(&conn->primary_list) &&
 364            !g_queue_is_empty(&conn->secondary_list)) {
 365         qemu_mutex_lock(&s->timer_check_lock);
 366         pkt = g_queue_pop_tail(&conn->primary_list);
 367         qemu_mutex_unlock(&s->timer_check_lock);
 368         switch (conn->ip_proto) {
 369         case IPPROTO_TCP:
 370             result = g_queue_find_custom(&conn->secondary_list,
 371                      pkt, (GCompareFunc)colo_packet_compare_tcp);
 372             break;
 373         case IPPROTO_UDP:
 374             result = g_queue_find_custom(&conn->secondary_list,
 375                      pkt, (GCompareFunc)colo_packet_compare_udp);
 376             break;
 377         case IPPROTO_ICMP:
 378             result = g_queue_find_custom(&conn->secondary_list,
 379                      pkt, (GCompareFunc)colo_packet_compare_icmp);
 380             break;
 381         default:
 382             result = g_queue_find_custom(&conn->secondary_list,
 383                      pkt, (GCompareFunc)colo_packet_compare_other);
 384             break;
 385         }
 386
 387         if (result) {
 388             ret = compare_chr_send(&s->chr_out, pkt->data, pkt->size);
 389             if (ret < 0) {
 390                 error_report("colo_send_primary_packet failed");
 391             }
 392             trace_colo_compare_main("packet same and release packet");
 393             g_queue_remove(&conn->secondary_list, result->data);
 394             packet_destroy(pkt, NULL);
 395         } else {
 396             /*
 397              * If one packet arrive late, the secondary_list or
 398              * primary_list will be empty, so we can't compare it
 399              * until next comparison.
 400              */
 401             trace_colo_compare_main("packet different");
 402             qemu_mutex_lock(&s->timer_check_lock);
 403             g_queue_push_tail(&conn->primary_list, pkt);
 404             qemu_mutex_unlock(&s->timer_check_lock);
 405             /* TODO: colo_notify_checkpoint();*/
 406             break;
 407         }
 408     }
 409 }
 410
 411 static int compare_chr_send(CharBackend *out,
 412                             const uint8_t *buf,
 413                             uint32_t size)
 414 {
 415     int ret = 0;
 416     uint32_t len = htonl(size);
 417
 418     if (!size) {
 419         return 0;
 420     }
 421
 422     ret = qemu_chr_fe_write_all(out, (uint8_t *)&len, sizeof(len));
 423     if (ret != sizeof(len)) {
 424         goto err;
 425     }
 426
 427     ret = qemu_chr_fe_write_all(out, (uint8_t *)buf, size);
 428     if (ret != size) {
 429         goto err;
 430     }
 431
 432     return 0;
 433
 434 err:
 435     return ret < 0 ? ret : -EIO;
 436 }
 437
 438 static int compare_chr_can_read(void *opaque)
 439 {
 440     return COMPARE_READ_LEN_MAX;
 441 }
 442
 443 /*
 444  * Called from the main thread on the primary for packets
 445  * arriving over the socket from the primary.
 446  */
 447 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
 448 {
 449     CompareState *s = COLO_COMPARE(opaque);
 450     int ret;
 451
 452     ret = net_fill_rstate(&s->pri_rs, buf, size);
 453     if (ret == -1) {
 454         qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL,
 455                                  NULL, NULL, true);
 456         error_report("colo-compare primary_in error");
 457     }
 458 }
 459
 460 /*
 461  * Called from the main thread on the primary for packets
 462  * arriving over the socket from the secondary.
 463  */
 464 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 465 {
 466     CompareState *s = COLO_COMPARE(opaque);
 467     int ret;
 468
 469     ret = net_fill_rstate(&s->sec_rs, buf, size);
 470     if (ret == -1) {
 471         qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL,
 472                                  NULL, NULL, true);
 473         error_report("colo-compare secondary_in error");
 474     }
 475 }
 476
 477 static void *colo_compare_thread(void *opaque)
 478 {
 479     GMainContext *worker_context;
 480     GMainLoop *compare_loop;
 481     CompareState *s = opaque;
 482
 483     worker_context = g_main_context_new();
 484
 485     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
 486                              compare_pri_chr_in, NULL, s, worker_context, true);
 487     qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
 488                              compare_sec_chr_in, NULL, s, worker_context, true);
 489
 490     compare_loop = g_main_loop_new(worker_context, FALSE);
 491
 492     g_main_loop_run(compare_loop);
 493
 494     g_main_loop_unref(compare_loop);
 495     g_main_context_unref(worker_context);
 496     return NULL;
 497 }
 498
 499 static char *compare_get_pri_indev(Object *obj, Error **errp)
 500 {
 501     CompareState *s = COLO_COMPARE(obj);
 502
 503     return g_strdup(s->pri_indev);
 504 }
 505
 506 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
 507 {
 508     CompareState *s = COLO_COMPARE(obj);
 509
 510     g_free(s->pri_indev);
 511     s->pri_indev = g_strdup(value);
 512 }
 513
 514 static char *compare_get_sec_indev(Object *obj, Error **errp)
 515 {
 516     CompareState *s = COLO_COMPARE(obj);
 517
 518     return g_strdup(s->sec_indev);
 519 }
 520
 521 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
 522 {
 523     CompareState *s = COLO_COMPARE(obj);
 524
 525     g_free(s->sec_indev);
 526     s->sec_indev = g_strdup(value);
 527 }
 528
 529 static char *compare_get_outdev(Object *obj, Error **errp)
 530 {
 531     CompareState *s = COLO_COMPARE(obj);
 532
 533     return g_strdup(s->outdev);
 534 }
 535
 536 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
 537 {
 538     CompareState *s = COLO_COMPARE(obj);
 539
 540     g_free(s->outdev);
 541     s->outdev = g_strdup(value);
 542 }
 543
 544 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
 545 {
 546     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
 547
 548     if (packet_enqueue(s, PRIMARY_IN)) {
 549         trace_colo_compare_main("primary: unsupported packet in");
 550         compare_chr_send(&s->chr_out, pri_rs->buf, pri_rs->packet_len);
 551     } else {
 552         /* compare connection */
 553         g_queue_foreach(&s->conn_list, colo_compare_connection, s);
 554     }
 555 }
 556
 557 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
 558 {
 559     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
 560
 561     if (packet_enqueue(s, SECONDARY_IN)) {
 562         trace_colo_compare_main("secondary: unsupported packet in");
 563     } else {
 564         /* compare connection */
 565         g_queue_foreach(&s->conn_list, colo_compare_connection, s);
 566     }
 567 }
 568
 569
 570 /*
 571  * Return 0 is success.
 572  * Return 1 is failed.
 573  */
 574 static int find_and_check_chardev(CharDriverState **chr,
 575                                   char *chr_name,
 576                                   Error **errp)
 577 {
 578     CompareChardevProps props;
 579
 580     *chr = qemu_chr_find(chr_name);
 581     if (*chr == NULL) {
 582         error_setg(errp, "Device '%s' not found",
 583                    chr_name);
 584         return 1;
 585     }
 586
 587     memset(&props, 0, sizeof(props));
 588
 589     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
 590         error_setg(errp, "chardev \"%s\" is not reconnectable",
 591                    chr_name);
 592         return 1;
 593     }
 594
 595     return 0;
 596 }
 597
 598 /*
 599  * Check old packet regularly so it can watch for any packets
 600  * that the secondary hasn't produced equivalents of.
 601  */
 602 static void check_old_packet_regular(void *opaque)
 603 {
 604     CompareState *s = opaque;
 605
 606     timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 607               REGULAR_PACKET_CHECK_MS);
 608     /* if have old packet we will notify checkpoint */
 609     /*
 610      * TODO: Make timer handler run in compare thread
 611      * like qemu_chr_add_handlers_full.
 612      */
 613     qemu_mutex_lock(&s->timer_check_lock);
 614     colo_old_packet_check(s);
 615     qemu_mutex_unlock(&s->timer_check_lock);
 616 }
 617
 618 /*
 619  * Called from the main thread on the primary
 620  * to setup colo-compare.
 621  */
 622 static void colo_compare_complete(UserCreatable *uc, Error **errp)
 623 {
 624     CompareState *s = COLO_COMPARE(uc);
 625     CharDriverState *chr;
 626     char thread_name[64];
 627     static int compare_id;
 628
 629     if (!s->pri_indev || !s->sec_indev || !s->outdev) {
 630         error_setg(errp, "colo compare needs 'primary_in' ,"
 631                    "'secondary_in','outdev' property set");
 632         return;
 633     } else if (!strcmp(s->pri_indev, s->outdev) ||
 634                !strcmp(s->sec_indev, s->outdev) ||
 635                !strcmp(s->pri_indev, s->sec_indev)) {
 636         error_setg(errp, "'indev' and 'outdev' could not be same "
 637                    "for compare module");
 638         return;
 639     }
 640
 641     if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
 642         !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
 643         return;
 644     }
 645
 646     if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
 647         !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
 648         return;
 649     }
 650
 651     if (find_and_check_chardev(&chr, s->outdev, errp) ||
 652         !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
 653         return;
 654     }
 655
 656     net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
 657     net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
 658
 659     g_queue_init(&s->conn_list);
 660     qemu_mutex_init(&s->timer_check_lock);
 661
 662     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
 663                                                       connection_key_equal,
 664                                                       g_free,
 665                                                       connection_destroy);
 666
 667     sprintf(thread_name, "colo-compare %d", compare_id);
 668     qemu_thread_create(&s->thread, thread_name,
 669                        colo_compare_thread, s,
 670                        QEMU_THREAD_JOINABLE);
 671     compare_id++;
 672
 673     /* A regular timer to kick any packets that the secondary doesn't match */
 674     s->timer = timer_new_ms(QEMU_CLOCK_VIRTUAL, /* Only when guest runs */
 675                             check_old_packet_regular, s);
 676     timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 677                         REGULAR_PACKET_CHECK_MS);
 678
 679     return;
 680 }
 681
 682 static void colo_compare_class_init(ObjectClass *oc, void *data)
 683 {
 684     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
 685
 686     ucc->complete = colo_compare_complete;
 687 }
 688
 689 static void colo_compare_init(Object *obj)
 690 {
 691     object_property_add_str(obj, "primary_in",
 692                             compare_get_pri_indev, compare_set_pri_indev,
 693                             NULL);
 694     object_property_add_str(obj, "secondary_in",
 695                             compare_get_sec_indev, compare_set_sec_indev,
 696                             NULL);
 697     object_property_add_str(obj, "outdev",
 698                             compare_get_outdev, compare_set_outdev,
 699                             NULL);
 700 }
 701
 702 static void colo_compare_finalize(Object *obj)
 703 {
 704     CompareState *s = COLO_COMPARE(obj);
 705
 706     qemu_chr_fe_deinit(&s->chr_pri_in);
 707     qemu_chr_fe_deinit(&s->chr_sec_in);
 708     qemu_chr_fe_deinit(&s->chr_out);
 709
 710     g_queue_free(&s->conn_list);
 711
 712     if (qemu_thread_is_self(&s->thread)) {
 713         /* compare connection */
 714         g_queue_foreach(&s->conn_list, colo_compare_connection, s);
 715         qemu_thread_join(&s->thread);
 716     }
 717
 718     if (s->timer) {
 719         timer_del(s->timer);
 720     }
 721
 722     qemu_mutex_destroy(&s->timer_check_lock);
 723
 724     g_free(s->pri_indev);
 725     g_free(s->sec_indev);
 726     g_free(s->outdev);
 727 }
 728
 729 static const TypeInfo colo_compare_info = {
 730     .name = TYPE_COLO_COMPARE,
 731     .parent = TYPE_OBJECT,
 732     .instance_size = sizeof(CompareState),
 733     .instance_init = colo_compare_init,
 734     .instance_finalize = colo_compare_finalize,
 735     .class_size = sizeof(CompareClass),
 736     .class_init = colo_compare_class_init,
 737     .interfaces = (InterfaceInfo[]) {
 738         { TYPE_USER_CREATABLE },
 739         { }
 740     }
 741 };
 742
 743 static void register_types(void)
 744 {
 745     type_register_static(&colo_compare_info);
 746 }
 747
 748 type_init(register_types);