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     CharDriverState *chr_pri_in;
  72     CharDriverState *chr_sec_in;
  73     CharDriverState *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(CharDriverState *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(CharDriverState *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_add_handlers(s->chr_pri_in, NULL, NULL, NULL, NULL);
 455         error_report("colo-compare primary_in error");
 456     }
 457 }
 458
 459 /*
 460  * Called from the main thread on the primary for packets
 461  * arriving over the socket from the secondary.
 462  */
 463 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 464 {
 465     CompareState *s = COLO_COMPARE(opaque);
 466     int ret;
 467
 468     ret = net_fill_rstate(&s->sec_rs, buf, size);
 469     if (ret == -1) {
 470         qemu_chr_add_handlers(s->chr_sec_in, NULL, NULL, NULL, NULL);
 471         error_report("colo-compare secondary_in error");
 472     }
 473 }
 474
 475 static void *colo_compare_thread(void *opaque)
 476 {
 477     GMainContext *worker_context;
 478     GMainLoop *compare_loop;
 479     CompareState *s = opaque;
 480
 481     worker_context = g_main_context_new();
 482
 483     qemu_chr_add_handlers_full(s->chr_pri_in, compare_chr_can_read,
 484                           compare_pri_chr_in, NULL, s, worker_context);
 485     qemu_chr_add_handlers_full(s->chr_sec_in, compare_chr_can_read,
 486                           compare_sec_chr_in, NULL, s, worker_context);
 487
 488     compare_loop = g_main_loop_new(worker_context, FALSE);
 489
 490     g_main_loop_run(compare_loop);
 491
 492     g_main_loop_unref(compare_loop);
 493     g_main_context_unref(worker_context);
 494     return NULL;
 495 }
 496
 497 static char *compare_get_pri_indev(Object *obj, Error **errp)
 498 {
 499     CompareState *s = COLO_COMPARE(obj);
 500
 501     return g_strdup(s->pri_indev);
 502 }
 503
 504 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
 505 {
 506     CompareState *s = COLO_COMPARE(obj);
 507
 508     g_free(s->pri_indev);
 509     s->pri_indev = g_strdup(value);
 510 }
 511
 512 static char *compare_get_sec_indev(Object *obj, Error **errp)
 513 {
 514     CompareState *s = COLO_COMPARE(obj);
 515
 516     return g_strdup(s->sec_indev);
 517 }
 518
 519 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
 520 {
 521     CompareState *s = COLO_COMPARE(obj);
 522
 523     g_free(s->sec_indev);
 524     s->sec_indev = g_strdup(value);
 525 }
 526
 527 static char *compare_get_outdev(Object *obj, Error **errp)
 528 {
 529     CompareState *s = COLO_COMPARE(obj);
 530
 531     return g_strdup(s->outdev);
 532 }
 533
 534 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
 535 {
 536     CompareState *s = COLO_COMPARE(obj);
 537
 538     g_free(s->outdev);
 539     s->outdev = g_strdup(value);
 540 }
 541
 542 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
 543 {
 544     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
 545
 546     if (packet_enqueue(s, PRIMARY_IN)) {
 547         trace_colo_compare_main("primary: unsupported packet in");
 548         compare_chr_send(s->chr_out, pri_rs->buf, pri_rs->packet_len);
 549     } else {
 550         /* compare connection */
 551         g_queue_foreach(&s->conn_list, colo_compare_connection, s);
 552     }
 553 }
 554
 555 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
 556 {
 557     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
 558
 559     if (packet_enqueue(s, SECONDARY_IN)) {
 560         trace_colo_compare_main("secondary: unsupported packet in");
 561     } else {
 562         /* compare connection */
 563         g_queue_foreach(&s->conn_list, colo_compare_connection, s);
 564     }
 565 }
 566
 567 static int compare_chardev_opts(void *opaque,
 568                                 const char *name, const char *value,
 569                                 Error **errp)
 570 {
 571     CompareChardevProps *props = opaque;
 572
 573     if (strcmp(name, "backend") == 0 &&
 574         strcmp(value, "socket") == 0) {
 575         props->is_socket = true;
 576         return 0;
 577     } else if (strcmp(name, "host") == 0 ||
 578               (strcmp(name, "port") == 0) ||
 579               (strcmp(name, "server") == 0) ||
 580               (strcmp(name, "wait") == 0) ||
 581               (strcmp(name, "path") == 0)) {
 582         return 0;
 583     } else {
 584         error_setg(errp,
 585                    "COLO-compare does not support a chardev with option %s=%s",
 586                    name, value);
 587         return -1;
 588     }
 589 }
 590
 591 /*
 592  * Return 0 is success.
 593  * Return 1 is failed.
 594  */
 595 static int find_and_check_chardev(CharDriverState **chr,
 596                                   char *chr_name,
 597                                   Error **errp)
 598 {
 599     CompareChardevProps props;
 600
 601     *chr = qemu_chr_find(chr_name);
 602     if (*chr == NULL) {
 603         error_setg(errp, "Device '%s' not found",
 604                    chr_name);
 605         return 1;
 606     }
 607
 608     memset(&props, 0, sizeof(props));
 609     if (qemu_opt_foreach((*chr)->opts, compare_chardev_opts, &props, errp)) {
 610         return 1;
 611     }
 612
 613     if (!props.is_socket) {
 614         error_setg(errp, "chardev \"%s\" is not a tcp socket",
 615                    chr_name);
 616         return 1;
 617     }
 618     return 0;
 619 }
 620
 621 /*
 622  * Check old packet regularly so it can watch for any packets
 623  * that the secondary hasn't produced equivalents of.
 624  */
 625 static void check_old_packet_regular(void *opaque)
 626 {
 627     CompareState *s = opaque;
 628
 629     timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 630               REGULAR_PACKET_CHECK_MS);
 631     /* if have old packet we will notify checkpoint */
 632     /*
 633      * TODO: Make timer handler run in compare thread
 634      * like qemu_chr_add_handlers_full.
 635      */
 636     qemu_mutex_lock(&s->timer_check_lock);
 637     colo_old_packet_check(s);
 638     qemu_mutex_unlock(&s->timer_check_lock);
 639 }
 640
 641 /*
 642  * Called from the main thread on the primary
 643  * to setup colo-compare.
 644  */
 645 static void colo_compare_complete(UserCreatable *uc, Error **errp)
 646 {
 647     CompareState *s = COLO_COMPARE(uc);
 648     char thread_name[64];
 649     static int compare_id;
 650
 651     if (!s->pri_indev || !s->sec_indev || !s->outdev) {
 652         error_setg(errp, "colo compare needs 'primary_in' ,"
 653                    "'secondary_in','outdev' property set");
 654         return;
 655     } else if (!strcmp(s->pri_indev, s->outdev) ||
 656                !strcmp(s->sec_indev, s->outdev) ||
 657                !strcmp(s->pri_indev, s->sec_indev)) {
 658         error_setg(errp, "'indev' and 'outdev' could not be same "
 659                    "for compare module");
 660         return;
 661     }
 662
 663     if (find_and_check_chardev(&s->chr_pri_in, s->pri_indev, errp)) {
 664         return;
 665     }
 666
 667     if (find_and_check_chardev(&s->chr_sec_in, s->sec_indev, errp)) {
 668         return;
 669     }
 670
 671     if (find_and_check_chardev(&s->chr_out, s->outdev, errp)) {
 672         return;
 673     }
 674
 675     qemu_chr_fe_claim_no_fail(s->chr_pri_in);
 676
 677     qemu_chr_fe_claim_no_fail(s->chr_sec_in);
 678
 679     qemu_chr_fe_claim_no_fail(s->chr_out);
 680
 681     net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
 682     net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
 683
 684     g_queue_init(&s->conn_list);
 685     qemu_mutex_init(&s->timer_check_lock);
 686
 687     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
 688                                                       connection_key_equal,
 689                                                       g_free,
 690                                                       connection_destroy);
 691
 692     sprintf(thread_name, "colo-compare %d", compare_id);
 693     qemu_thread_create(&s->thread, thread_name,
 694                        colo_compare_thread, s,
 695                        QEMU_THREAD_JOINABLE);
 696     compare_id++;
 697
 698     /* A regular timer to kick any packets that the secondary doesn't match */
 699     s->timer = timer_new_ms(QEMU_CLOCK_VIRTUAL, /* Only when guest runs */
 700                             check_old_packet_regular, s);
 701     timer_mod(s->timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 702                         REGULAR_PACKET_CHECK_MS);
 703
 704     return;
 705 }
 706
 707 static void colo_compare_class_init(ObjectClass *oc, void *data)
 708 {
 709     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
 710
 711     ucc->complete = colo_compare_complete;
 712 }
 713
 714 static void colo_compare_init(Object *obj)
 715 {
 716     object_property_add_str(obj, "primary_in",
 717                             compare_get_pri_indev, compare_set_pri_indev,
 718                             NULL);
 719     object_property_add_str(obj, "secondary_in",
 720                             compare_get_sec_indev, compare_set_sec_indev,
 721                             NULL);
 722     object_property_add_str(obj, "outdev",
 723                             compare_get_outdev, compare_set_outdev,
 724                             NULL);
 725 }
 726
 727 static void colo_compare_finalize(Object *obj)
 728 {
 729     CompareState *s = COLO_COMPARE(obj);
 730
 731     if (s->chr_pri_in) {
 732         qemu_chr_add_handlers(s->chr_pri_in, NULL, NULL, NULL, NULL);
 733         qemu_chr_fe_release(s->chr_pri_in);
 734     }
 735     if (s->chr_sec_in) {
 736         qemu_chr_add_handlers(s->chr_sec_in, NULL, NULL, NULL, NULL);
 737         qemu_chr_fe_release(s->chr_sec_in);
 738     }
 739     if (s->chr_out) {
 740         qemu_chr_fe_release(s->chr_out);
 741     }
 742
 743     g_queue_free(&s->conn_list);
 744
 745     if (qemu_thread_is_self(&s->thread)) {
 746         /* compare connection */
 747         g_queue_foreach(&s->conn_list, colo_compare_connection, s);
 748         qemu_thread_join(&s->thread);
 749     }
 750
 751     if (s->timer) {
 752         timer_del(s->timer);
 753     }
 754
 755     qemu_mutex_destroy(&s->timer_check_lock);
 756
 757     g_free(s->pri_indev);
 758     g_free(s->sec_indev);
 759     g_free(s->outdev);
 760 }
 761
 762 static const TypeInfo colo_compare_info = {
 763     .name = TYPE_COLO_COMPARE,
 764     .parent = TYPE_OBJECT,
 765     .instance_size = sizeof(CompareState),
 766     .instance_init = colo_compare_init,
 767     .instance_finalize = colo_compare_finalize,
 768     .class_size = sizeof(CompareClass),
 769     .class_init = colo_compare_class_init,
 770     .interfaces = (InterfaceInfo[]) {
 771         { TYPE_USER_CREATABLE },
 772         { }
 773     }
 774 };
 775
 776 static void register_types(void)
 777 {
 778     type_register_static(&colo_compare_info);
 779 }
 780
 781 type_init(register_types);