Re: [Qemu-devel] [PATCH V12 05/10] colo-compare: track connection and enqueue packet

[Jason Wang]

On 2016年08月17日 16:10, 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 <address@hidden>
> Signed-off-by: Li Zhijian <address@hidden>
> Signed-off-by: Wen Congyang <address@hidden>
> ---
>   net/colo-compare.c |  51 ++++++++++++++++++-----
>   net/colo.c         | 117 +++++++++++++++++++++++++++++++++++++++++++++++++++++
>   net/colo.h         |  27 +++++++++++++
>   3 files changed, 185 insertions(+), 10 deletions(-)
>
> diff --git a/net/colo-compare.c b/net/colo-compare.c
> index d9e4459..bab215b 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;
>   } CompareState;
> @@ -100,7 +105,9 @@ static int compare_chr_send(CharDriverState *out,
>    */
>   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);
> @@ -113,17 +120,34 @@ 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->conn_list);
>   
> -    /*
> -     * TODO: insert pkt to it's conn->primary_list
> -     * or conn->secondary_list
> -     */
> +    if (!conn->processing) {
> +        g_queue_push_tail(&s->conn_list, conn);
> +        conn->processing = true;
> +    }
> +
> +    if (mode == PRIMARY_IN) {
> +        if (g_queue_get_length(&conn->primary_list) <
> +                               MAX_QUEUE_SIZE) {

Should be "<=" I believe.

> +            g_queue_push_tail(&conn->primary_list, pkt);
> +        } else {
> +            error_report("colo compare primary queue size too big,"
> +            "drop packet");

indentation here looks odd.

> +        }
> +    } 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;
>   }
> @@ -325,7 +349,12 @@ 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);
>   
> -    /* use g_hash_table_new_full() to new a hashtable */
> +    g_queue_init(&s->conn_list);
> +
> +    s->connection_track_table = g_hash_table_new_full(connection_key_hash,
> +                                                      connection_key_equal,
> +                                                      g_free,
> +                                                      connection_destroy);
>   
>       return;
>   }
> @@ -366,6 +395,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);
> diff --git a/net/colo.c b/net/colo.c
> index 4daedd4..bc86553 100644
> --- a/net/colo.c
> +++ b/net/colo.c
> @@ -16,6 +16,29 @@
>   #include "qemu/error-report.h"
>   #include "net/colo.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;
> @@ -43,6 +66,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_header);
> +        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_header + 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);
> @@ -68,3 +147,41 @@ void connection_hashtable_reset(GHashTable 
> *connection_track_table)
>   {
>       g_hash_table_remove_all(connection_track_table);
>   }
> +
> +static void colo_rm_connection(void *opaque, void *user_data)
> +{

user_data is unused here.

> +    connection_destroy(opaque);
> +}
> +
> +/* if not found, create a new connection and add to hash table */
> +Connection *connection_get(GHashTable *connection_track_table,
> +                           ConnectionKey *key,
> +                           GQueue *conn_list)
> +{
> +    Connection *conn = g_hash_table_lookup(connection_track_table, key);
> +    static uint32_t hashtable_size;
> +
> +    if (conn == NULL) {
> +        ConnectionKey *new_key = g_memdup(key, sizeof(*key));
> +
> +        conn = connection_new(key);
> +
> +        hashtable_size += 1;

Use of uninitialized variable?

> +        if (hashtable_size > HASHTABLE_MAX_SIZE) {

Should we use g_hash_table_size() here.

> +            error_report("colo proxy connection hashtable full, clear it");
> +            connection_hashtable_reset(connection_track_table);
> +            /*
> +             * clear the conn_list
> +             */
> +            if (conn_list) {
> +                g_queue_foreach(conn_list, colo_rm_connection, NULL);
> +            }
> +
> +            hashtable_size = 0;
> +        }
> +
> +        g_hash_table_insert(connection_track_table, new_key, conn);

Then there's no need for hashtable_size.

> +    }
> +
> +    return conn;
> +}
> diff --git a/net/colo.h b/net/colo.h
> index 8559f28..9cbc14e 100644
> --- a/net/colo.h
> +++ b/net/colo.h
> @@ -30,7 +30,34 @@ 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;
> +} 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,
> +                           GQueue *conn_list);
>   void connection_hashtable_reset(GHashTable *connection_track_table);
>   Packet *packet_new(const void *data, int size);
>   void packet_destroy(void *opaque, void *user_data);