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