Re: [Qemu-devel] [RFC PATCH V10 4/7] colo-compare: track connection and enqueue packet

[Jason Wang <jasowang@redhat.com>]

On 2016年07月26日 09:49, Zhang Chen wrote:
> In this patch we use kernel jhash table to track
> connection, and then enqueue net packet like this:
>
> + CompareState ++
> |               |
> +---------------+   +---------------+         +---------------+
> |conn list      +--->conn           +--------->conn           |
> +---------------+   +---------------+         +---------------+
> |               |     |           |             |          |
> +---------------+ +---v----+  +---v----+    +---v----+ +---v----+
>                    |primary |  |secondary    |primary | |secondary
>                    |packet  |  |packet  +    |packet  | |packet  +
>                    +--------+  +--------+    +--------+ +--------+
>                        |           |             |          |
>                    +---v----+  +---v----+    +---v----+ +---v----+
>                    |primary |  |secondary    |primary | |secondary
>                    |packet  |  |packet  +    |packet  | |packet  +
>                    +--------+  +--------+    +--------+ +--------+
>                        |           |             |          |
>                    +---v----+  +---v----+    +---v----+ +---v----+
>                    |primary |  |secondary    |primary | |secondary
>                    |packet  |  |packet  +    |packet  | |packet  +
>                    +--------+  +--------+    +--------+ +--------+
>
> We use conn_list to record connection info.
> When we want to enqueue a packet, firstly get the
> connection from connection_track_table. then push
> the packet to g_queue(pri/sec) in it's own conn.
>
> Signed-off-by: Zhang Chen <zhangchen.fnst@cn.fujitsu.com>
> Signed-off-by: Li Zhijian <lizhijian@cn.fujitsu.com>
> Signed-off-by: Wen Congyang <wency@cn.fujitsu.com>
> ---
>   net/colo-base.c    | 108 +++++++++++++++++++++++++++++++++++++++++++++++++++++
>   net/colo-base.h    |  30 +++++++++++++++
>   net/colo-compare.c |  70 +++++++++++++++++++++++++++++-----
>   3 files changed, 198 insertions(+), 10 deletions(-)
>
> diff --git a/net/colo-base.c b/net/colo-base.c
> index f5d5de9..7e91dec 100644
> --- a/net/colo-base.c
> +++ b/net/colo-base.c
> @@ -16,6 +16,29 @@
>   #include "qemu/error-report.h"
>   #include "net/colo-base.h"
>   
> +uint32_t connection_key_hash(const void *opaque)
> +{
> +    const ConnectionKey *key = opaque;
> +    uint32_t a, b, c;
> +
> +    /* Jenkins hash */
> +    a = b = c = JHASH_INITVAL + sizeof(*key);
> +    a += key->src.s_addr;
> +    b += key->dst.s_addr;
> +    c += (key->src_port | key->dst_port << 16);
> +    __jhash_mix(a, b, c);
> +
> +    a += key->ip_proto;
> +    __jhash_final(a, b, c);
> +
> +    return c;
> +}
> +
> +int connection_key_equal(const void *key1, const void *key2)
> +{
> +    return memcmp(key1, key2, sizeof(ConnectionKey)) == 0;
> +}
> +
>   int parse_packet_early(Packet *pkt)
>   {
>       int network_length;
> @@ -47,6 +70,62 @@ int parse_packet_early(Packet *pkt)
>       return 0;
>   }
>   
> +void fill_connection_key(Packet *pkt, ConnectionKey *key)
> +{
> +    uint32_t tmp_ports;
> +
> +    key->ip_proto = pkt->ip->ip_p;
> +
> +    switch (key->ip_proto) {
> +    case IPPROTO_TCP:
> +    case IPPROTO_UDP:
> +    case IPPROTO_DCCP:
> +    case IPPROTO_ESP:
> +    case IPPROTO_SCTP:
> +    case IPPROTO_UDPLITE:
> +        tmp_ports = *(uint32_t *)(pkt->transport_layer);
> +        key->src = pkt->ip->ip_src;
> +        key->dst = pkt->ip->ip_dst;
> +        key->src_port = ntohs(tmp_ports & 0xffff);
> +        key->dst_port = ntohs(tmp_ports >> 16);
> +        break;
> +    case IPPROTO_AH:
> +        tmp_ports = *(uint32_t *)(pkt->transport_layer + 4);
> +        key->src = pkt->ip->ip_src;
> +        key->dst = pkt->ip->ip_dst;
> +        key->src_port = ntohs(tmp_ports & 0xffff);
> +        key->dst_port = ntohs(tmp_ports >> 16);
> +        break;
> +    default:
> +        key->src_port = 0;
> +        key->dst_port = 0;
> +        break;
> +    }
> +}
> +
> +Connection *connection_new(ConnectionKey *key)
> +{
> +    Connection *conn = g_slice_new(Connection);
> +
> +    conn->ip_proto = key->ip_proto;
> +    conn->processing = false;
> +    g_queue_init(&conn->primary_list);
> +    g_queue_init(&conn->secondary_list);
> +
> +    return conn;
> +}
> +
> +void connection_destroy(void *opaque)
> +{
> +    Connection *conn = opaque;
> +
> +    g_queue_foreach(&conn->primary_list, packet_destroy, NULL);
> +    g_queue_free(&conn->primary_list);
> +    g_queue_foreach(&conn->secondary_list, packet_destroy, NULL);
> +    g_queue_free(&conn->secondary_list);
> +    g_slice_free(Connection, conn);
> +}
> +
>   Packet *packet_new(const void *data, int size)
>   {
>       Packet *pkt = g_slice_new(Packet);
> @@ -72,3 +151,32 @@ void connection_hashtable_reset(GHashTable *connection_track_table)
>   {
>       g_hash_table_remove_all(connection_track_table);
>   }
> +
> +/* if not found, create a new connection and add to hash table */
> +Connection *connection_get(GHashTable *connection_track_table,
> +                           ConnectionKey *key,
> +                           uint32_t *hashtable_size)
> +{
> +    Connection *conn = g_hash_table_lookup(connection_track_table, key);
> +
> +    if (conn == NULL) {
> +        ConnectionKey *new_key = g_memdup(key, sizeof(*key));
> +
> +        conn = connection_new(key);
> +
> +        (*hashtable_size) += 1;
> +        if (*hashtable_size > HASHTABLE_MAX_SIZE) {
> +            error_report("colo proxy connection hashtable full, clear it");
> +            connection_hashtable_reset(connection_track_table);
> +            /*
> +             * when hashtable_size == 0, clear the conn_list
> +             * in place where be called.
> +             */

Does this mean it requires the caller to do this? If yes, seems not 
good, why not simply do things here?

> +            *hashtable_size = 0;
> +        }
> +
> +        g_hash_table_insert(connection_track_table, new_key, conn);

Then we lose the track of *hashtable_size here. It should be 1 but we 
set it to zero if we are out of space.

> +    }
> +
> +    return conn;
> +}
> diff --git a/net/colo-base.h b/net/colo-base.h
> index 48835e7..0505608 100644
> --- a/net/colo-base.h
> +++ b/net/colo-base.h
> @@ -30,7 +30,37 @@ typedef struct Packet {
>       int size;
>   } Packet;
>   
> +typedef struct ConnectionKey {
> +    /* (src, dst) must be grouped, in the same way than in IP header */
> +    struct in_addr src;
> +    struct in_addr dst;
> +    uint16_t src_port;
> +    uint16_t dst_port;
> +    uint8_t ip_proto;
> +} QEMU_PACKED ConnectionKey;
> +
> +typedef struct Connection {
> +    /* connection primary send queue: element type: Packet */
> +    GQueue primary_list;
> +    /* connection secondary send queue: element type: Packet */
> +    GQueue secondary_list;
> +    /* flag to enqueue unprocessed_connections */
> +    bool processing;
> +    uint8_t ip_proto;
> +    /* be used by filter-rewriter */
> +    tcp_seq  primary_seq;
> +    tcp_seq  secondary_seq;
> +} Connection;
> +
> +uint32_t connection_key_hash(const void *opaque);
> +int connection_key_equal(const void *opaque1, const void *opaque2);
>   int parse_packet_early(Packet *pkt);
> +void fill_connection_key(Packet *pkt, ConnectionKey *key);
> +Connection *connection_new(ConnectionKey *key);
> +void connection_destroy(void *opaque);
> +Connection *connection_get(GHashTable *connection_track_table,
> +                           ConnectionKey *key,
> +                           uint32_t *hashtable_size);
>   void connection_hashtable_reset(GHashTable *connection_track_table);
>   Packet *packet_new(const void *data, int size);
>   void packet_destroy(void *opaque, void *user_data);
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index 7c52cc8..5f87710 100644
> --- a/net/colo-compare.c
> +++ b/net/colo-compare.c
> @@ -72,6 +72,11 @@ typedef struct CompareState {
>       SocketReadState pri_rs;
>       SocketReadState sec_rs;
>   
> +    /* connection list: the connections belonged to this NIC could be found
> +     * in this list.
> +     * element type: Connection
> +     */
> +    GQueue conn_list;
>       /* hashtable to save connection */
>       GHashTable *connection_track_table;
>       /* to save unprocessed_connections */
> @@ -93,13 +98,30 @@ static int compare_chr_send(CharDriverState *out,
>                               const uint8_t *buf,
>                               uint32_t size);
>   
> +static void colo_rm_connection(void *opaque, void *user_data)
> +{
> +    Connection *conn = opaque;
> +    Packet *pkt = NULL;
> +
> +    while (!g_queue_is_empty(&conn->primary_list)) {
> +        pkt = g_queue_pop_head(&conn->primary_list);
> +        packet_destroy(pkt, NULL);
> +    }
> +    while (!g_queue_is_empty(&conn->secondary_list)) {
> +        pkt = g_queue_pop_head(&conn->secondary_list);
> +        packet_destroy(pkt, NULL);
> +    }

This looks rather similar to connection_destroy(), can we share some codes?

> +}
> +
>   /*
>    * Return 0 on success, if return -1 means the pkt
>    * is unsupported(arp and ipv6) and will be sent later
>    */
>   static int packet_enqueue(CompareState *s, int mode)
>   {
> +    ConnectionKey key = {{ 0 } };
>       Packet *pkt = NULL;
> +    Connection *conn;
>   
>       if (mode == PRIMARY_IN) {
>           pkt = packet_new(s->pri_rs.buf, s->pri_rs.packet_len);
> @@ -112,17 +134,38 @@ static int packet_enqueue(CompareState *s, int mode)
>           pkt = NULL;
>           return -1;
>       }
> -    /* TODO: get connection key from pkt */
> +    fill_connection_key(pkt, &key);
>   
> -    /*
> -     * TODO: use connection key get conn from
> -     * connection_track_table
> -     */
> +    conn = connection_get(s->connection_track_table,
> +                          &key,
> +                          &s->hashtable_size);
>   
> -    /*
> -     * TODO: insert pkt to it's conn->primary_list
> -     * or conn->secondary_list
> -     */
> +    if (!s->hashtable_size) {
> +        g_queue_foreach(&s->conn_list, colo_rm_connection, NULL);
> +    }
> +
> +    if (!conn->processing) {
> +        g_queue_push_tail(&s->conn_list, conn);
> +        conn->processing = true;

Why not simply do this in connection_get(), you can save a 
conn->processing flag.

> +    }
> +
> +    if (mode == PRIMARY_IN) {
> +        if (g_queue_get_length(&conn->primary_list) <
> +                               MAX_QUEUE_SIZE) {
> +            g_queue_push_tail(&conn->primary_list, pkt);
> +        } else {
> +            error_report("colo compare primary queue size too big,"
> +            "drop packet");

I don't see how packet were dropped in this case?

> +        }
> +    } else {
> +        if (g_queue_get_length(&conn->secondary_list) <
> +                               MAX_QUEUE_SIZE) {
> +            g_queue_push_tail(&conn->secondary_list, pkt);
> +        } else {
> +            error_report("colo compare secondary queue size too big,"
> +            "drop packet");
> +        }
> +    }
>   
>       return 0;
>   }
> @@ -267,9 +310,14 @@ static void colo_compare_complete(UserCreatable *uc, Error **errp)
>       net_socket_rs_init(&s->pri_rs, compare_pri_rs_finalize);
>       net_socket_rs_init(&s->sec_rs, compare_sec_rs_finalize);
>   
> +    g_queue_init(&s->conn_list);
> +
>       s->hashtable_size = 0;
>   
> -    /* use g_hash_table_new_full() to new a hashtable */
> +    s->connection_track_table = g_hash_table_new_full(connection_key_hash,
> +                                                      connection_key_equal,
> +                                                      g_free,
> +                                                      connection_destroy);
>   
>       return;
>   }
> @@ -310,6 +358,8 @@ static void colo_compare_finalize(Object *obj)
>           qemu_chr_fe_release(s->chr_out);
>       }
>   
> +    g_queue_free(&s->conn_list);
> +
>       g_free(s->pri_indev);
>       g_free(s->sec_indev);
>       g_free(s->outdev);