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