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 "net/queue.h"
  27 #include "chardev/char-fe.h"
  28 #include "qemu/sockets.h"
  29 #include "qapi-visit.h"
  30 #include "net/colo.h"
  31 #include "sysemu/iothread.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     bool vnet_hdr;
  77
  78     /*
  79      * Record the connection that through the NIC
  80      * Element type: Connection
  81      */
  82     GQueue conn_list;
  83     /* Record the connection without repetition */
  84     GHashTable *connection_track_table;
  85
  86     IOThread *iothread;
  87     GMainContext *worker_context;
  88     QEMUTimer *packet_check_timer;
  89 } CompareState;
  90
  91 typedef struct CompareClass {
  92     ObjectClass parent_class;
  93 } CompareClass;
  94
  95 enum {
  96     PRIMARY_IN = 0,
  97     SECONDARY_IN,
  98 };
  99
 100 static int compare_chr_send(CompareState *s,
 101                             const uint8_t *buf,
 102                             uint32_t size,
 103                             uint32_t vnet_hdr_len);
 104
 105 static gint seq_sorter(Packet *a, Packet *b, gpointer data)
 106 {
 107     struct tcphdr *atcp, *btcp;
 108
 109     atcp = (struct tcphdr *)(a->transport_header);
 110     btcp = (struct tcphdr *)(b->transport_header);
 111     return ntohl(atcp->th_seq) - ntohl(btcp->th_seq);
 112 }
 113
 114 /*
 115  * Return 1 on success, if return 0 means the
 116  * packet will be dropped
 117  */
 118 static int colo_insert_packet(GQueue *queue, Packet *pkt)
 119 {
 120     if (g_queue_get_length(queue) <= MAX_QUEUE_SIZE) {
 121         if (pkt->ip->ip_p == IPPROTO_TCP) {
 122             g_queue_insert_sorted(queue,
 123                                   pkt,
 124                                   (GCompareDataFunc)seq_sorter,
 125                                   NULL);
 126         } else {
 127             g_queue_push_tail(queue, pkt);
 128         }
 129         return 1;
 130     }
 131     return 0;
 132 }
 133
 134 /*
 135  * Return 0 on success, if return -1 means the pkt
 136  * is unsupported(arp and ipv6) and will be sent later
 137  */
 138 static int packet_enqueue(CompareState *s, int mode, Connection **con)
 139 {
 140     ConnectionKey key;
 141     Packet *pkt = NULL;
 142     Connection *conn;
 143
 144     if (mode == PRIMARY_IN) {
 145         pkt = packet_new(s->pri_rs.buf,
 146                          s->pri_rs.packet_len,
 147                          s->pri_rs.vnet_hdr_len);
 148     } else {
 149         pkt = packet_new(s->sec_rs.buf,
 150                          s->sec_rs.packet_len,
 151                          s->sec_rs.vnet_hdr_len);
 152     }
 153
 154     if (parse_packet_early(pkt)) {
 155         packet_destroy(pkt, NULL);
 156         pkt = NULL;
 157         return -1;
 158     }
 159     fill_connection_key(pkt, &key);
 160
 161     conn = connection_get(s->connection_track_table,
 162                           &key,
 163                           &s->conn_list);
 164
 165     if (!conn->processing) {
 166         g_queue_push_tail(&s->conn_list, conn);
 167         conn->processing = true;
 168     }
 169
 170     if (mode == PRIMARY_IN) {
 171         if (!colo_insert_packet(&conn->primary_list, pkt)) {
 172             error_report("colo compare primary queue size too big,"
 173                          "drop packet");
 174         }
 175     } else {
 176         if (!colo_insert_packet(&conn->secondary_list, pkt)) {
 177             error_report("colo compare secondary queue size too big,"
 178                          "drop packet");
 179         }
 180     }
 181     *con = conn;
 182
 183     return 0;
 184 }
 185
 186 /*
 187  * The IP packets sent by primary and secondary
 188  * will be compared in here
 189  * TODO support ip fragment, Out-Of-Order
 190  * return:    0  means packet same
 191  *            > 0 || < 0 means packet different
 192  */
 193 static int colo_packet_compare_common(Packet *ppkt,
 194                                       Packet *spkt,
 195                                       int poffset,
 196                                       int soffset)
 197 {
 198     if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 199         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
 200
 201         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
 202         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
 203         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
 204         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
 205
 206         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 207                                    pri_ip_dst, spkt->size,
 208                                    sec_ip_src, sec_ip_dst);
 209     }
 210
 211     poffset = ppkt->vnet_hdr_len + poffset;
 212     soffset = ppkt->vnet_hdr_len + soffset;
 213
 214     if (ppkt->size - poffset == spkt->size - soffset) {
 215         return memcmp(ppkt->data + poffset,
 216                       spkt->data + soffset,
 217                       spkt->size - soffset);
 218     } else {
 219         trace_colo_compare_main("Net packet size are not the same");
 220         return -1;
 221     }
 222 }
 223
 224 /*
 225  * Called from the compare thread on the primary
 226  * for compare tcp packet
 227  * compare_tcp copied from Dr. David Alan Gilbert's branch
 228  */
 229 static int colo_packet_compare_tcp(Packet *spkt, Packet *ppkt)
 230 {
 231     struct tcphdr *ptcp, *stcp;
 232     int res;
 233
 234     trace_colo_compare_main("compare tcp");
 235
 236     ptcp = (struct tcphdr *)ppkt->transport_header;
 237     stcp = (struct tcphdr *)spkt->transport_header;
 238
 239     /*
 240      * The 'identification' field in the IP header is *very* random
 241      * it almost never matches.  Fudge this by ignoring differences in
 242      * unfragmented packets; they'll normally sort themselves out if different
 243      * anyway, and it should recover at the TCP level.
 244      * An alternative would be to get both the primary and secondary to rewrite
 245      * somehow; but that would need some sync traffic to sync the state
 246      */
 247     if (ntohs(ppkt->ip->ip_off) & IP_DF) {
 248         spkt->ip->ip_id = ppkt->ip->ip_id;
 249         /* and the sum will be different if the IDs were different */
 250         spkt->ip->ip_sum = ppkt->ip->ip_sum;
 251     }
 252
 253     /*
 254      * Check tcp header length for tcp option field.
 255      * th_off > 5 means this tcp packet have options field.
 256      * The tcp options maybe always different.
 257      * for example:
 258      * From RFC 7323.
 259      * TCP Timestamps option (TSopt):
 260      * Kind: 8
 261      *
 262      * Length: 10 bytes
 263      *
 264      *    +-------+-------+---------------------+---------------------+
 265      *    |Kind=8 |  10   |   TS Value (TSval)  |TS Echo Reply (TSecr)|
 266      *    +-------+-------+---------------------+---------------------+
 267      *       1       1              4                     4
 268      *
 269      * In this case the primary guest's timestamp always different with
 270      * the secondary guest's timestamp. COLO just focus on payload,
 271      * so we just need skip this field.
 272      */
 273     if (ptcp->th_off > 5) {
 274         ptrdiff_t ptcp_offset, stcp_offset;
 275
 276         ptcp_offset = ppkt->transport_header - (uint8_t *)ppkt->data
 277                       + (ptcp->th_off * 4) - ppkt->vnet_hdr_len;
 278         stcp_offset = spkt->transport_header - (uint8_t *)spkt->data
 279                       + (stcp->th_off * 4) - spkt->vnet_hdr_len;
 280
 281         /*
 282          * When network is busy, some tcp options(like sack) will unpredictable
 283          * occur in primary side or secondary side. it will make packet size
 284          * not same, but the two packet's payload is identical. colo just
 285          * care about packet payload, so we skip the option field.
 286          */
 287         res = colo_packet_compare_common(ppkt, spkt, ptcp_offset, stcp_offset);
 288     } else if (ptcp->th_sum == stcp->th_sum) {
 289         res = colo_packet_compare_common(ppkt, spkt, ETH_HLEN, ETH_HLEN);
 290     } else {
 291         res = -1;
 292     }
 293
 294     if (res != 0 &&
 295         trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 296         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
 297
 298         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
 299         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
 300         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
 301         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
 302
 303         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 304                                    pri_ip_dst, spkt->size,
 305                                    sec_ip_src, sec_ip_dst);
 306
 307         trace_colo_compare_tcp_info("pri tcp packet",
 308                                     ntohl(ptcp->th_seq),
 309                                     ntohl(ptcp->th_ack),
 310                                     res, ptcp->th_flags,
 311                                     ppkt->size);
 312
 313         trace_colo_compare_tcp_info("sec tcp packet",
 314                                     ntohl(stcp->th_seq),
 315                                     ntohl(stcp->th_ack),
 316                                     res, stcp->th_flags,
 317                                     spkt->size);
 318
 319         qemu_hexdump((char *)ppkt->data, stderr,
 320                      "colo-compare ppkt", ppkt->size);
 321         qemu_hexdump((char *)spkt->data, stderr,
 322                      "colo-compare spkt", spkt->size);
 323     }
 324
 325     return res;
 326 }
 327
 328 /*
 329  * Called from the compare thread on the primary
 330  * for compare udp packet
 331  */
 332 static int colo_packet_compare_udp(Packet *spkt, Packet *ppkt)
 333 {
 334     int ret;
 335     int network_header_length = ppkt->ip->ip_hl * 4;
 336
 337     trace_colo_compare_main("compare udp");
 338
 339     /*
 340      * Because of ppkt and spkt are both in the same connection,
 341      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
 342      * same with spkt. In addition, IP header's Identification is a random
 343      * field, we can handle it in IP fragmentation function later.
 344      * COLO just concern the response net packet payload from primary guest
 345      * and secondary guest are same or not, So we ignored all IP header include
 346      * other field like TOS,TTL,IP Checksum. we only need to compare
 347      * the ip payload here.
 348      */
 349     ret = colo_packet_compare_common(ppkt, spkt,
 350                                      network_header_length + ETH_HLEN,
 351                                      network_header_length + ETH_HLEN);
 352
 353     if (ret) {
 354         trace_colo_compare_udp_miscompare("primary pkt size", ppkt->size);
 355         trace_colo_compare_udp_miscompare("Secondary pkt size", spkt->size);
 356         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 357             qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
 358                          ppkt->size);
 359             qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
 360                          spkt->size);
 361         }
 362     }
 363
 364     return ret;
 365 }
 366
 367 /*
 368  * Called from the compare thread on the primary
 369  * for compare icmp packet
 370  */
 371 static int colo_packet_compare_icmp(Packet *spkt, Packet *ppkt)
 372 {
 373     int network_header_length = ppkt->ip->ip_hl * 4;
 374
 375     trace_colo_compare_main("compare icmp");
 376
 377     /*
 378      * Because of ppkt and spkt are both in the same connection,
 379      * The ppkt's src ip, dst ip, src port, dst port, ip_proto all are
 380      * same with spkt. In addition, IP header's Identification is a random
 381      * field, we can handle it in IP fragmentation function later.
 382      * COLO just concern the response net packet payload from primary guest
 383      * and secondary guest are same or not, So we ignored all IP header include
 384      * other field like TOS,TTL,IP Checksum. we only need to compare
 385      * the ip payload here.
 386      */
 387     if (colo_packet_compare_common(ppkt, spkt,
 388                                    network_header_length + ETH_HLEN,
 389                                    network_header_length + ETH_HLEN)) {
 390         trace_colo_compare_icmp_miscompare("primary pkt size",
 391                                            ppkt->size);
 392         trace_colo_compare_icmp_miscompare("Secondary pkt size",
 393                                            spkt->size);
 394         if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 395             qemu_hexdump((char *)ppkt->data, stderr, "colo-compare pri pkt",
 396                          ppkt->size);
 397             qemu_hexdump((char *)spkt->data, stderr, "colo-compare sec pkt",
 398                          spkt->size);
 399         }
 400         return -1;
 401     } else {
 402         return 0;
 403     }
 404 }
 405
 406 /*
 407  * Called from the compare thread on the primary
 408  * for compare other packet
 409  */
 410 static int colo_packet_compare_other(Packet *spkt, Packet *ppkt)
 411 {
 412     trace_colo_compare_main("compare other");
 413     if (trace_event_get_state_backends(TRACE_COLO_COMPARE_MISCOMPARE)) {
 414         char pri_ip_src[20], pri_ip_dst[20], sec_ip_src[20], sec_ip_dst[20];
 415
 416         strcpy(pri_ip_src, inet_ntoa(ppkt->ip->ip_src));
 417         strcpy(pri_ip_dst, inet_ntoa(ppkt->ip->ip_dst));
 418         strcpy(sec_ip_src, inet_ntoa(spkt->ip->ip_src));
 419         strcpy(sec_ip_dst, inet_ntoa(spkt->ip->ip_dst));
 420
 421         trace_colo_compare_ip_info(ppkt->size, pri_ip_src,
 422                                    pri_ip_dst, spkt->size,
 423                                    sec_ip_src, sec_ip_dst);
 424     }
 425
 426     return colo_packet_compare_common(ppkt, spkt, 0, 0);
 427 }
 428
 429 static int colo_old_packet_check_one(Packet *pkt, int64_t *check_time)
 430 {
 431     int64_t now = qemu_clock_get_ms(QEMU_CLOCK_HOST);
 432
 433     if ((now - pkt->creation_ms) > (*check_time)) {
 434         trace_colo_old_packet_check_found(pkt->creation_ms);
 435         return 0;
 436     } else {
 437         return 1;
 438     }
 439 }
 440
 441 static int colo_old_packet_check_one_conn(Connection *conn,
 442                                           void *user_data)
 443 {
 444     GList *result = NULL;
 445     int64_t check_time = REGULAR_PACKET_CHECK_MS;
 446
 447     result = g_queue_find_custom(&conn->primary_list,
 448                                  &check_time,
 449                                  (GCompareFunc)colo_old_packet_check_one);
 450
 451     if (result) {
 452         /* Do checkpoint will flush old packet */
 453         /*
 454          * TODO: Notify colo frame to do checkpoint.
 455          * colo_compare_inconsistent_notify();
 456          */
 457         return 0;
 458     }
 459
 460     return 1;
 461 }
 462
 463 /*
 464  * Look for old packets that the secondary hasn't matched,
 465  * if we have some then we have to checkpoint to wake
 466  * the secondary up.
 467  */
 468 static void colo_old_packet_check(void *opaque)
 469 {
 470     CompareState *s = opaque;
 471
 472     /*
 473      * If we find one old packet, stop finding job and notify
 474      * COLO frame do checkpoint.
 475      */
 476     g_queue_find_custom(&s->conn_list, NULL,
 477                         (GCompareFunc)colo_old_packet_check_one_conn);
 478 }
 479
 480 /*
 481  * Called from the compare thread on the primary
 482  * for compare packet with secondary list of the
 483  * specified connection when a new packet was
 484  * queued to it.
 485  */
 486 static void colo_compare_connection(void *opaque, void *user_data)
 487 {
 488     CompareState *s = user_data;
 489     Connection *conn = opaque;
 490     Packet *pkt = NULL;
 491     GList *result = NULL;
 492     int ret;
 493
 494     while (!g_queue_is_empty(&conn->primary_list) &&
 495            !g_queue_is_empty(&conn->secondary_list)) {
 496         pkt = g_queue_pop_head(&conn->primary_list);
 497         switch (conn->ip_proto) {
 498         case IPPROTO_TCP:
 499             result = g_queue_find_custom(&conn->secondary_list,
 500                      pkt, (GCompareFunc)colo_packet_compare_tcp);
 501             break;
 502         case IPPROTO_UDP:
 503             result = g_queue_find_custom(&conn->secondary_list,
 504                      pkt, (GCompareFunc)colo_packet_compare_udp);
 505             break;
 506         case IPPROTO_ICMP:
 507             result = g_queue_find_custom(&conn->secondary_list,
 508                      pkt, (GCompareFunc)colo_packet_compare_icmp);
 509             break;
 510         default:
 511             result = g_queue_find_custom(&conn->secondary_list,
 512                      pkt, (GCompareFunc)colo_packet_compare_other);
 513             break;
 514         }
 515
 516         if (result) {
 517             ret = compare_chr_send(s,
 518                                    pkt->data,
 519                                    pkt->size,
 520                                    pkt->vnet_hdr_len);
 521             if (ret < 0) {
 522                 error_report("colo_send_primary_packet failed");
 523             }
 524             trace_colo_compare_main("packet same and release packet");
 525             g_queue_remove(&conn->secondary_list, result->data);
 526             packet_destroy(pkt, NULL);
 527         } else {
 528             /*
 529              * If one packet arrive late, the secondary_list or
 530              * primary_list will be empty, so we can't compare it
 531              * until next comparison.
 532              */
 533             trace_colo_compare_main("packet different");
 534             g_queue_push_head(&conn->primary_list, pkt);
 535             /* TODO: colo_notify_checkpoint();*/
 536             break;
 537         }
 538     }
 539 }
 540
 541 static int compare_chr_send(CompareState *s,
 542                             const uint8_t *buf,
 543                             uint32_t size,
 544                             uint32_t vnet_hdr_len)
 545 {
 546     int ret = 0;
 547     uint32_t len = htonl(size);
 548
 549     if (!size) {
 550         return 0;
 551     }
 552
 553     ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)&len, sizeof(len));
 554     if (ret != sizeof(len)) {
 555         goto err;
 556     }
 557
 558     if (s->vnet_hdr) {
 559         /*
 560          * We send vnet header len make other module(like filter-redirector)
 561          * know how to parse net packet correctly.
 562          */
 563         len = htonl(vnet_hdr_len);
 564         ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)&len, sizeof(len));
 565         if (ret != sizeof(len)) {
 566             goto err;
 567         }
 568     }
 569
 570     ret = qemu_chr_fe_write_all(&s->chr_out, (uint8_t *)buf, size);
 571     if (ret != size) {
 572         goto err;
 573     }
 574
 575     return 0;
 576
 577 err:
 578     return ret < 0 ? ret : -EIO;
 579 }
 580
 581 static int compare_chr_can_read(void *opaque)
 582 {
 583     return COMPARE_READ_LEN_MAX;
 584 }
 585
 586 /*
 587  * Called from the main thread on the primary for packets
 588  * arriving over the socket from the primary.
 589  */
 590 static void compare_pri_chr_in(void *opaque, const uint8_t *buf, int size)
 591 {
 592     CompareState *s = COLO_COMPARE(opaque);
 593     int ret;
 594
 595     ret = net_fill_rstate(&s->pri_rs, buf, size);
 596     if (ret == -1) {
 597         qemu_chr_fe_set_handlers(&s->chr_pri_in, NULL, NULL, NULL, NULL,
 598                                  NULL, NULL, true);
 599         error_report("colo-compare primary_in error");
 600     }
 601 }
 602
 603 /*
 604  * Called from the main thread on the primary for packets
 605  * arriving over the socket from the secondary.
 606  */
 607 static void compare_sec_chr_in(void *opaque, const uint8_t *buf, int size)
 608 {
 609     CompareState *s = COLO_COMPARE(opaque);
 610     int ret;
 611
 612     ret = net_fill_rstate(&s->sec_rs, buf, size);
 613     if (ret == -1) {
 614         qemu_chr_fe_set_handlers(&s->chr_sec_in, NULL, NULL, NULL, NULL,
 615                                  NULL, NULL, true);
 616         error_report("colo-compare secondary_in error");
 617     }
 618 }
 619
 620 /*
 621  * Check old packet regularly so it can watch for any packets
 622  * that the secondary hasn't produced equivalents of.
 623  */
 624 static void check_old_packet_regular(void *opaque)
 625 {
 626     CompareState *s = opaque;
 627
 628     /* if have old packet we will notify checkpoint */
 629     colo_old_packet_check(s);
 630     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 631                 REGULAR_PACKET_CHECK_MS);
 632 }
 633
 634 static void colo_compare_timer_init(CompareState *s)
 635 {
 636     AioContext *ctx = iothread_get_aio_context(s->iothread);
 637
 638     s->packet_check_timer = aio_timer_new(ctx, QEMU_CLOCK_VIRTUAL,
 639                                 SCALE_MS, check_old_packet_regular,
 640                                 s);
 641     timer_mod(s->packet_check_timer, qemu_clock_get_ms(QEMU_CLOCK_VIRTUAL) +
 642                     REGULAR_PACKET_CHECK_MS);
 643 }
 644
 645 static void colo_compare_timer_del(CompareState *s)
 646 {
 647     if (s->packet_check_timer) {
 648         timer_del(s->packet_check_timer);
 649         timer_free(s->packet_check_timer);
 650         s->packet_check_timer = NULL;
 651     }
 652  }
 653
 654 static void colo_compare_iothread(CompareState *s)
 655 {
 656     object_ref(OBJECT(s->iothread));
 657     s->worker_context = iothread_get_g_main_context(s->iothread);
 658
 659     qemu_chr_fe_set_handlers(&s->chr_pri_in, compare_chr_can_read,
 660                              compare_pri_chr_in, NULL, NULL,
 661                              s, s->worker_context, true);
 662     qemu_chr_fe_set_handlers(&s->chr_sec_in, compare_chr_can_read,
 663                              compare_sec_chr_in, NULL, NULL,
 664                              s, s->worker_context, true);
 665
 666     colo_compare_timer_init(s);
 667 }
 668
 669 static char *compare_get_pri_indev(Object *obj, Error **errp)
 670 {
 671     CompareState *s = COLO_COMPARE(obj);
 672
 673     return g_strdup(s->pri_indev);
 674 }
 675
 676 static void compare_set_pri_indev(Object *obj, const char *value, Error **errp)
 677 {
 678     CompareState *s = COLO_COMPARE(obj);
 679
 680     g_free(s->pri_indev);
 681     s->pri_indev = g_strdup(value);
 682 }
 683
 684 static char *compare_get_sec_indev(Object *obj, Error **errp)
 685 {
 686     CompareState *s = COLO_COMPARE(obj);
 687
 688     return g_strdup(s->sec_indev);
 689 }
 690
 691 static void compare_set_sec_indev(Object *obj, const char *value, Error **errp)
 692 {
 693     CompareState *s = COLO_COMPARE(obj);
 694
 695     g_free(s->sec_indev);
 696     s->sec_indev = g_strdup(value);
 697 }
 698
 699 static char *compare_get_outdev(Object *obj, Error **errp)
 700 {
 701     CompareState *s = COLO_COMPARE(obj);
 702
 703     return g_strdup(s->outdev);
 704 }
 705
 706 static void compare_set_outdev(Object *obj, const char *value, Error **errp)
 707 {
 708     CompareState *s = COLO_COMPARE(obj);
 709
 710     g_free(s->outdev);
 711     s->outdev = g_strdup(value);
 712 }
 713
 714 static bool compare_get_vnet_hdr(Object *obj, Error **errp)
 715 {
 716     CompareState *s = COLO_COMPARE(obj);
 717
 718     return s->vnet_hdr;
 719 }
 720
 721 static void compare_set_vnet_hdr(Object *obj,
 722                                  bool value,
 723                                  Error **errp)
 724 {
 725     CompareState *s = COLO_COMPARE(obj);
 726
 727     s->vnet_hdr = value;
 728 }
 729
 730 static void compare_pri_rs_finalize(SocketReadState *pri_rs)
 731 {
 732     CompareState *s = container_of(pri_rs, CompareState, pri_rs);
 733     Connection *conn = NULL;
 734
 735     if (packet_enqueue(s, PRIMARY_IN, &conn)) {
 736         trace_colo_compare_main("primary: unsupported packet in");
 737         compare_chr_send(s,
 738                          pri_rs->buf,
 739                          pri_rs->packet_len,
 740                          pri_rs->vnet_hdr_len);
 741     } else {
 742         /* compare packet in the specified connection */
 743         colo_compare_connection(conn, s);
 744     }
 745 }
 746
 747 static void compare_sec_rs_finalize(SocketReadState *sec_rs)
 748 {
 749     CompareState *s = container_of(sec_rs, CompareState, sec_rs);
 750     Connection *conn = NULL;
 751
 752     if (packet_enqueue(s, SECONDARY_IN, &conn)) {
 753         trace_colo_compare_main("secondary: unsupported packet in");
 754     } else {
 755         /* compare packet in the specified connection */
 756         colo_compare_connection(conn, s);
 757     }
 758 }
 759
 760
 761 /*
 762  * Return 0 is success.
 763  * Return 1 is failed.
 764  */
 765 static int find_and_check_chardev(Chardev **chr,
 766                                   char *chr_name,
 767                                   Error **errp)
 768 {
 769     *chr = qemu_chr_find(chr_name);
 770     if (*chr == NULL) {
 771         error_setg(errp, "Device '%s' not found",
 772                    chr_name);
 773         return 1;
 774     }
 775
 776     if (!qemu_chr_has_feature(*chr, QEMU_CHAR_FEATURE_RECONNECTABLE)) {
 777         error_setg(errp, "chardev \"%s\" is not reconnectable",
 778                    chr_name);
 779         return 1;
 780     }
 781
 782     return 0;
 783 }
 784
 785 /*
 786  * Called from the main thread on the primary
 787  * to setup colo-compare.
 788  */
 789 static void colo_compare_complete(UserCreatable *uc, Error **errp)
 790 {
 791     CompareState *s = COLO_COMPARE(uc);
 792     Chardev *chr;
 793
 794     if (!s->pri_indev || !s->sec_indev || !s->outdev || !s->iothread) {
 795         error_setg(errp, "colo compare needs 'primary_in' ,"
 796                    "'secondary_in','outdev','iothread' property set");
 797         return;
 798     } else if (!strcmp(s->pri_indev, s->outdev) ||
 799                !strcmp(s->sec_indev, s->outdev) ||
 800                !strcmp(s->pri_indev, s->sec_indev)) {
 801         error_setg(errp, "'indev' and 'outdev' could not be same "
 802                    "for compare module");
 803         return;
 804     }
 805
 806     if (find_and_check_chardev(&chr, s->pri_indev, errp) ||
 807         !qemu_chr_fe_init(&s->chr_pri_in, chr, errp)) {
 808         return;
 809     }
 810
 811     if (find_and_check_chardev(&chr, s->sec_indev, errp) ||
 812         !qemu_chr_fe_init(&s->chr_sec_in, chr, errp)) {
 813         return;
 814     }
 815
 816     if (find_and_check_chardev(&chr, s->outdev, errp) ||
 817         !qemu_chr_fe_init(&s->chr_out, chr, errp)) {
 818         return;
 819     }
 820
 821     net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize, s->vnet_hdr);
 822     net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize, s->vnet_hdr);
 823
 824     g_queue_init(&s->conn_list);
 825
 826     s->connection_track_table = g_hash_table_new_full(connection_key_hash,
 827                                                       connection_key_equal,
 828                                                       g_free,
 829                                                       connection_destroy);
 830
 831     colo_compare_iothread(s);
 832     return;
 833 }
 834
 835 static void colo_flush_packets(void *opaque, void *user_data)
 836 {
 837     CompareState *s = user_data;
 838     Connection *conn = opaque;
 839     Packet *pkt = NULL;
 840
 841     while (!g_queue_is_empty(&conn->primary_list)) {
 842         pkt = g_queue_pop_head(&conn->primary_list);
 843         compare_chr_send(s,
 844                          pkt->data,
 845                          pkt->size,
 846                          pkt->vnet_hdr_len);
 847         packet_destroy(pkt, NULL);
 848     }
 849     while (!g_queue_is_empty(&conn->secondary_list)) {
 850         pkt = g_queue_pop_head(&conn->secondary_list);
 851         packet_destroy(pkt, NULL);
 852     }
 853 }
 854
 855 static void colo_compare_class_init(ObjectClass *oc, void *data)
 856 {
 857     UserCreatableClass *ucc = USER_CREATABLE_CLASS(oc);
 858
 859     ucc->complete = colo_compare_complete;
 860 }
 861
 862 static void colo_compare_init(Object *obj)
 863 {
 864     CompareState *s = COLO_COMPARE(obj);
 865
 866     object_property_add_str(obj, "primary_in",
 867                             compare_get_pri_indev, compare_set_pri_indev,
 868                             NULL);
 869     object_property_add_str(obj, "secondary_in",
 870                             compare_get_sec_indev, compare_set_sec_indev,
 871                             NULL);
 872     object_property_add_str(obj, "outdev",
 873                             compare_get_outdev, compare_set_outdev,
 874                             NULL);
 875     object_property_add_link(obj, "iothread", TYPE_IOTHREAD,
 876                             (Object **)&s->iothread,
 877                             object_property_allow_set_link,
 878                             OBJ_PROP_LINK_UNREF_ON_RELEASE, NULL);
 879
 880     s->vnet_hdr = false;
 881     object_property_add_bool(obj, "vnet_hdr_support", compare_get_vnet_hdr,
 882                              compare_set_vnet_hdr, NULL);
 883 }
 884
 885 static void colo_compare_finalize(Object *obj)
 886 {
 887     CompareState *s = COLO_COMPARE(obj);
 888
 889     qemu_chr_fe_deinit(&s->chr_pri_in, false);
 890     qemu_chr_fe_deinit(&s->chr_sec_in, false);
 891     qemu_chr_fe_deinit(&s->chr_out, false);
 892     if (s->iothread) {
 893         colo_compare_timer_del(s);
 894     }
 895     /* Release all unhandled packets after compare thead exited */
 896     g_queue_foreach(&s->conn_list, colo_flush_packets, s);
 897
 898     g_queue_clear(&s->conn_list);
 899
 900     if (s->connection_track_table) {
 901         g_hash_table_destroy(s->connection_track_table);
 902     }
 903
 904     if (s->iothread) {
 905         object_unref(OBJECT(s->iothread));
 906     }
 907     g_free(s->pri_indev);
 908     g_free(s->sec_indev);
 909     g_free(s->outdev);
 910 }
 911
 912 static const TypeInfo colo_compare_info = {
 913     .name = TYPE_COLO_COMPARE,
 914     .parent = TYPE_OBJECT,
 915     .instance_size = sizeof(CompareState),
 916     .instance_init = colo_compare_init,
 917     .instance_finalize = colo_compare_finalize,
 918     .class_size = sizeof(CompareClass),
 919     .class_init = colo_compare_class_init,
 920     .interfaces = (InterfaceInfo[]) {
 921         { TYPE_USER_CREATABLE },
 922         { }
 923     }
 924 };
 925
 926 static void register_types(void)
 927 {
 928     type_register_static(&colo_compare_info);
 929 }
 930
 931 type_init(register_types);