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

[Zhang Chen <zhangchen.fnst@cn.fujitsu.com>]

On 08/02/2016 03:14 PM, Jason Wang wrote:
>
>
> 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?

OK, I will add conn_list as parameter input to this function then clear it.

>
>> +            *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.

After clear conn_list I will fix it.

>
>> +    }
>> +
>> +    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?

Good idea~

>
>> +}
>> +
>>   /*
>>    * 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.

Because filter-rewriter need connection_get() without do this work,
We make connection_get() more common.

>
>> +    }
>> +
>> +    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?

In here we didn't push the packet to primary_list, then we comparing 
queued packet can't
get this packet, so,the packet be dropped.


Thanks
Zhang Chen

>
>> +        }
>> +    } 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);
>
>
>
> .
>

-- 
Thanks
zhangchen