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