credit system reclaced with priority system, init session endiness bugfix

[cor.git] / net / cor / common.c

/**
 *      Connection oriented routing
 *      Copyright (C) 2007-2013 Michael Blizek
 *
 *      This program is free software; you can redistribute it and/or
 *      modify it under the terms of the GNU General Public License
 *      as published by the Free Software Foundation; either version 2
 *      of the License, or (at your option) any later version.
 *
 *      This program is distributed in the hope that it will be useful,
 *      but WITHOUT ANY WARRANTY; without even the implied warranty of
 *      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *      GNU General Public License for more details.
 *
 *      You should have received a copy of the GNU General Public License
 *      along with this program; if not, write to the Free Software
 *      Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 *      02110-1301, USA.
 */

#include <linux/mutex.h>

#include "cor.h"

DEFINE_SPINLOCK(cor_bindnodes);
static DEFINE_SPINLOCK(conn_free);
static DEFINE_SPINLOCK(connid_gen);

static LIST_HEAD(openports);

static struct kmem_cache *conn_slab;
static struct kmem_cache *connid_reuse_slab;

atomic_t num_conns;


struct conn *get_conn(struct neighbor *nb, __u32 conn_id)
{
        struct rb_node * n = 0;
        struct conn *ret = 0;

        spin_lock_bh(&(nb->connid_lock));

        n = nb->connid_rb.rb_node;

        while (likely(n != 0) && ret == 0) {
                struct conn *src_in_o = container_of(n, struct conn,
                                source.in.rbn);

                BUG_ON(src_in_o->sourcetype != SOURCE_IN);

                if (conn_id < src_in_o->source.in.conn_id)
                        n = n->rb_left;
                else if (conn_id > src_in_o->source.in.conn_id)
                        n = n->rb_right;
                else
                        ret = src_in_o;
        }

        if (ret != 0)
                kref_get(&(ret->ref));

        spin_unlock_bh(&(nb->connid_lock));

        return ret;
}

static int insert_connid(struct neighbor *nb, struct conn *src_in_ll)
{
        int rc = 0;

        __u32 conn_id = src_in_ll->source.in.conn_id;

        struct rb_node **p = 0;
        struct rb_node *parent = 0;

        BUG_ON(src_in_ll->sourcetype != SOURCE_IN);

        spin_lock_bh(&(nb->connid_lock));

        p = &(nb->connid_rb.rb_node);

        while ((*p) != 0) {
                struct conn *src_in_o = container_of(*p, struct conn,
                                source.in.rbn);

                BUG_ON(src_in_o->sourcetype != SOURCE_IN);

                parent = *p;
                if (unlikely(conn_id == src_in_o->source.in.conn_id)) {
                        goto duplicate;
                } else if (conn_id < src_in_o->source.in.conn_id) {
                        p = &(*p)->rb_left;
                } else if (conn_id > src_in_o->source.in.conn_id) {
                        p = &(*p)->rb_right;
                } else {
                        BUG();
                }
        }

        kref_get(&(src_in_ll->ref));
        rb_link_node(&(src_in_ll->source.in.rbn), parent, p);

        if (0) {
duplicate:
                rc = 1;
        }

        spin_unlock_bh(&(nb->connid_lock));

        return rc;
}

struct connid_reuse_item *get_connid_reuseitem(struct neighbor *nb,
                __u32 conn_id)
{
        struct rb_node *n = 0;
        struct connid_reuse_item *ret = 0;

        spin_lock_bh(&(nb->connid_reuse_lock));

        n = nb->connid_reuse_rb.rb_node;

        while (likely(n != 0) && ret == 0) {
                struct connid_reuse_item *cir = container_of(n,
                                struct connid_reuse_item, rbn);

                BUG_ON(cir->conn_id == 0);

                if (conn_id < cir->conn_id)
                        n = n->rb_left;
                else if (conn_id > cir->conn_id)
                        n = n->rb_right;
                else
                        ret = cir;
        }

        if (ret != 0)
                kref_get(&(ret->ref));

        spin_unlock_bh(&(nb->connid_reuse_lock));

        return ret;
}

static void insert_connid_reuse(struct neighbor *nb,
                struct connid_reuse_item *ins)
{
        struct rb_node **p = 0;
        struct rb_node *parent = 0;

        BUG_ON(ins->conn_id == 0);

        spin_lock_bh(&(nb->connid_reuse_lock));

        p = &(nb->connid_reuse_rb.rb_node);

        while ((*p) != 0) {
                struct connid_reuse_item *curr = container_of(*p,
                                struct connid_reuse_item, rbn);

                BUG_ON(curr->conn_id == 0);

                parent = *p;
                if (unlikely(ins->conn_id == curr->conn_id)) {
                        BUG();
                } else if (ins->conn_id < curr->conn_id) {
                        p = &(*p)->rb_left;
                } else if (ins->conn_id > curr->conn_id) {
                        p = &(*p)->rb_right;
                } else {
                        BUG();
                }
        }

        kref_get(&(ins->ref));
        rb_link_node(&(ins->rbn), parent, p);

        spin_unlock_bh(&(nb->connid_reuse_lock));
}

static void free_connid_reuse(struct kref *ref)
{
        struct connid_reuse_item *cir = container_of(ref,
                        struct connid_reuse_item, ref);

        kmem_cache_free(connid_reuse_slab, cir);
}

void connid_used_pingsuccess(struct neighbor *nb)
{
        struct connid_reuse_item *cri;

        spin_lock_bh(&(nb->connid_reuse_lock));

        nb->connid_reuse_pingcnt++;
        while (list_empty(&(nb->connid_reuse_list)) == 0) {
                cri = container_of(nb->connid_reuse_list.next,
                                struct connid_reuse_item, lh);
                if ((cri->pingcnt + CONNID_REUSE_RTTS -
                                nb->connid_reuse_pingcnt) < 32768)
                        break;

                rb_erase(&(cri->rbn), &(nb->connid_reuse_rb));
                kref_put(&(cri->ref), kreffree_bug);

                list_del(&(cri->lh));
                kref_put(&(cri->ref), free_connid_reuse);
        }

        spin_unlock_bh(&(nb->connid_reuse_lock));
}

static int connid_used(struct neighbor *nb, __u32 conn_id)
{
        struct conn *cn;
        struct connid_reuse_item *cir;

        cn = get_conn(nb, conn_id);
        if (unlikely(cn != 0)) {
                kref_put(&(cn->ref), free_conn);
                return 1;
        }

        cir = get_connid_reuseitem(nb, conn_id);
        if (unlikely(cir != 0)) {
                kref_put(&(cir->ref), free_connid_reuse);
                return 1;
        }

        return 0;
}

static int connid_alloc(struct neighbor *nb, struct conn *src_in_ll)
{
        __u32 conn_id;
        int i;

        BUG_ON(src_in_ll->sourcetype != SOURCE_IN);
        BUG_ON(src_in_ll->reversedir->targettype != TARGET_OUT);

        spin_lock_bh(&connid_gen);
        for (i=0;i<16;i++) {
                conn_id = 0;
                get_random_bytes((char *) &conn_id, sizeof(conn_id));
                conn_id = (conn_id & ~(1 << 31));

                if (unlikely(conn_id == 0))
                        continue;

                if (unlikely(connid_used(nb, conn_id)))
                        continue;

                goto found;
        }
        spin_unlock_bh(&connid_gen);

        return 1;

found:
        src_in_ll->source.in.conn_id = conn_id;
        src_in_ll->reversedir->target.out.conn_id = (conn_id | (1 << 31));
        if (insert_connid(nb, src_in_ll) != 0) {
                BUG();
        }
        spin_unlock_bh(&connid_gen);
        return 0;
}

void _set_last_act(struct conn *src_in_l)
{
        unsigned long iflags;
        src_in_l->source.in.jiffies_last_act = jiffies;
        spin_lock_irqsave(&(src_in_l->source.in.nb->conn_list_lock), iflags);
        list_del(&(src_in_l->source.in.nb_list));
        list_add_tail(&(src_in_l->source.in.nb_list),
                        &(src_in_l->source.in.nb->rcv_conn_list));
        spin_unlock_irqrestore(&(src_in_l->source.in.nb->conn_list_lock),
                        iflags);
}

void free_conn(struct kref *ref)
{
        unsigned long iflags;
        struct conn *cn = container_of(ref, struct conn, ref);
        struct conn *reversedir = 0;

        spin_lock_irqsave(&conn_free, iflags);

        BUG_ON(cn->isreset == 0);

        if (cn->reversedir != 0)
                cn->reversedir->isreset = 3;

        if (cn->isreset != 3)
                goto out;

        if (cn->reversedir != 0) {
                cn->reversedir->reversedir = 0;
                reversedir = cn->reversedir;
                cn->reversedir = 0;
        }

        if (cn->sourcetype == SOURCE_IN) {
                BUG_ON(cn->source.in.conn_id != 0);
                kref_put(&(cn->source.in.nb->ref), neighbor_free);
                cn->source.in.nb = 0;
        }

        if (cn->targettype == TARGET_OUT) {
                BUG_ON(cn->target.out.conn_id != 0);
                kref_put(&(cn->target.out.nb->ref), neighbor_free);
                cn->target.out.nb = 0;
        }

        BUG_ON(cn->data_buf.datasize != 0);
        BUG_ON(cn->data_buf.overhead != 0);

        memset(cn, 9*16 + 10, sizeof(struct conn));
        kmem_cache_free(conn_slab, cn);

out:
        spin_unlock_irqrestore(&conn_free, iflags);

        if (reversedir != 0)
                free_conn(&(reversedir->ref));
}

/**
 * rc == 0 ==> ok
 * rc == 1 ==> connid_reuse or connid allocation failed
 */
int conn_init_out(struct conn *trgt_unconn_ll, struct neighbor *nb,
                __u32 rcvd_connid, int use_rcvd_connid)
{
        unsigned long iflags;
        int rc = 0;
        struct conn *src_unconn_ll = trgt_unconn_ll->reversedir;

        BUG_ON(trgt_unconn_ll->targettype != TARGET_UNCONNECTED);
        BUG_ON(src_unconn_ll == 0);
        BUG_ON(src_unconn_ll->sourcetype != SOURCE_UNCONNECTED);

        memset(&(trgt_unconn_ll->target.out), 0,
                        sizeof(trgt_unconn_ll->target.out));
        memset(&(src_unconn_ll->source.in), 0,
                        sizeof(src_unconn_ll->source.in));

        trgt_unconn_ll->targettype = TARGET_OUT;
        src_unconn_ll->sourcetype = SOURCE_IN;

        if (use_rcvd_connid) {
                BUG_ON((rcvd_connid & (1 << 31)) == 0);

                src_unconn_ll->source.in.conn_id = rcvd_connid;
                if (unlikely(insert_connid(nb, src_unconn_ll) != 0)) {
                        src_unconn_ll->source.in.conn_id = 0;
                        rc = 1;
                        goto out_err;
                }
        } else {
                src_unconn_ll->source.in.cir = kmem_cache_alloc(
                                connid_reuse_slab, GFP_ATOMIC);
                if (unlikely(src_unconn_ll->source.in.cir == 0)) {
                        rc = 1;
                        goto out_err;
                }
                memset(src_unconn_ll->source.in.cir, 0,
                                sizeof(struct connid_reuse_item));

                if (unlikely(connid_alloc(nb, src_unconn_ll))) {
                        rc = 1;
                        goto out_freecir;
                }
        }

        trgt_unconn_ll->target.out.nb = nb;
        src_unconn_ll->source.in.nb = nb;

        /* neighbor pointer */
        kref_get(&(nb->ref));
        kref_get(&(nb->ref));

        INIT_LIST_HEAD(&(src_unconn_ll->source.in.reorder_queue));

        INIT_LIST_HEAD(&(src_unconn_ll->source.in.acks_pending));

        INIT_LIST_HEAD(&(trgt_unconn_ll->target.out.retrans_list));

        reset_seqno(trgt_unconn_ll, 0);
        if (use_rcvd_connid == 0) {
                get_random_bytes((char *)
                                &(trgt_unconn_ll->target.out.seqno_nextsend),
                                sizeof(
                                trgt_unconn_ll->target.out.seqno_nextsend));
                trgt_unconn_ll->target.out.seqno_acked =
                                trgt_unconn_ll->target.out.seqno_nextsend;
                reset_seqno(trgt_unconn_ll,
                                trgt_unconn_ll->target.out.seqno_nextsend);

                get_random_bytes((char *)
                                &(src_unconn_ll->source.in.next_seqno),
                                sizeof(src_unconn_ll->source.in.next_seqno));
                src_unconn_ll->source.in.window_seqnolimit =
                                src_unconn_ll->source.in.next_seqno;
                src_unconn_ll->source.in.window_seqnolimit_remote =
                                src_unconn_ll->source.in.next_seqno;
        }

        get_random_bytes((char *) &(trgt_unconn_ll->target.out.priority_seqno),
                        1);
        trgt_unconn_ll->source.in.priority_seqno = 0;

        src_unconn_ll->source.in.jiffies_last_act = jiffies;

        spin_lock_irqsave(&(nb->conn_list_lock), iflags);
        list_add_tail(&(src_unconn_ll->source.in.nb_list),
                        &(nb->rcv_conn_list));
        spin_unlock_irqrestore(&(nb->conn_list_lock), iflags);

        /* neighbor lists */
        kref_get(&(src_unconn_ll->ref));

        if (src_unconn_ll->is_client)
                atomic_inc(&num_conns);

        if (use_rcvd_connid == 0)
                update_windowlimit(src_unconn_ll);

        if (0) {
out_freecir:
                kmem_cache_free(connid_reuse_slab,
                                src_unconn_ll->source.in.cir);
                src_unconn_ll->source.in.cir = 0;
out_err:
                trgt_unconn_ll->targettype = TARGET_UNCONNECTED;
                src_unconn_ll->sourcetype = SOURCE_UNCONNECTED;
        }
        return rc;
}

void conn_init_sock_source(struct conn *cn)
{
        BUG_ON(cn == 0);
        cn->sourcetype = SOURCE_SOCK;
        memset(&(cn->source.sock), 0, sizeof(cn->source.sock));
        cn->source.sock.priority = PRIORITY_MAX;
}

void conn_init_sock_target(struct conn *cn)
{
        BUG_ON(cn == 0);
        cn->targettype = TARGET_SOCK;
        memset(&(cn->target.sock), 0, sizeof(cn->target.sock));
        reset_seqno(cn, 0);
}

struct conn* alloc_conn(gfp_t allocflags)
{
        struct conn *cn1 = 0;
        struct conn *cn2 = 0;

        cn1 = kmem_cache_alloc(conn_slab, allocflags);
        if (unlikely(cn1 == 0))
                goto out_err0;

        cn2 = kmem_cache_alloc(conn_slab, allocflags);
        if (unlikely(cn2 == 0))
                goto out_err1;

        memset(cn1, 0, sizeof(struct conn));
        memset(cn2, 0, sizeof(struct conn));

        cn1->reversedir = cn2;
        cn2->reversedir = cn1;

        kref_init(&(cn1->ref));
        kref_init(&(cn2->ref));

        cn1->sourcetype = SOURCE_UNCONNECTED;
        cn2->sourcetype = SOURCE_UNCONNECTED;
        cn1->targettype = TARGET_UNCONNECTED;
        cn2->targettype = TARGET_UNCONNECTED;

        cn1->isreset = 0;
        cn2->isreset = 0;

        spin_lock_init(&(cn1->rcv_lock));
        spin_lock_init(&(cn2->rcv_lock));

        databuf_init(cn1);
        databuf_init(cn2);

        speedtracker_init(&(cn1->st));
        speedtracker_init(&(cn2->st));

        return cn1;

out_err1:
        kmem_cache_free(conn_slab, cn1);
out_err0:
        return 0;
}

static struct cor_sock *get_corsock_by_port(__be64 port)
{
        struct list_head *curr = openports.next;

        while (curr != &openports) {
                struct cor_sock *cs = container_of(curr, struct cor_sock,
                                data.listener.lh);
                BUG_ON(cs->type != CS_TYPE_LISTENER);
                if (cs->data.listener.port == port)
                        return cs;

                curr = curr->next;
        }

        return 0;
}

__u32 list_services(char *buf, __u32 buflen)
{
        __u32 cnt = 0;

        __u32 buf_offset = 4;

        struct list_head *curr;
        int rc;

        /*
         * The variable length header rowcount need to be generated after the
         * data. This is done by reserving the maximum space they could take. If
         * they end up being smaller, the data is moved so that there is no gap.
         */

        BUG_ON(buf == 0);
        BUG_ON(buflen < buf_offset);

        spin_lock_bh(&cor_bindnodes);

        curr = openports.next;

        while (curr != &openports) {
                struct cor_sock *cs = container_of(curr, struct cor_sock,
                                data.listener.lh);
                BUG_ON(cs->type != CS_TYPE_LISTENER);

                if (cs->data.listener.publish_service == 0)
                        goto cont;

                if (unlikely(buf_offset + 2 < buf_offset) ||
                                buf_offset + 2 > buflen)
                        break;

                buf[buf_offset] = ((char *) &(cs->data.listener.port))[0];
                buf[buf_offset+1] = ((char *) &(cs->data.listener.port))[1];
                buf_offset += 2;
                cnt++;

cont:
                curr = curr->next;
        }

        spin_unlock_bh(&cor_bindnodes);

        rc = encode_len(buf, 4, cnt);
        BUG_ON(rc <= 0);
        BUG_ON(rc > 4);

        if (likely(rc < 4))
                memmove(buf + ((__u32) rc), buf+4, buf_offset);

        return buf_offset - 4 + ((__u32) rc);

}

void set_publish_service(struct cor_sock *cs, __u8 value)
{
        BUG_ON (value != 0 && value != 1);

        mutex_lock(&(cs->lock));

        cs->publish_service = value;

        if (cs->type == CS_TYPE_LISTENER) {
                spin_lock_bh(&cor_bindnodes);
                cs->data.listener.publish_service = value;
                spin_unlock_bh(&cor_bindnodes);
        }

        mutex_unlock(&(cs->lock));
}

void close_port(struct cor_sock *cs)
{
        mutex_lock(&(cs->lock));
        if (unlikely(cs->type != CS_TYPE_LISTENER))
                goto out;

        spin_lock_bh(&cor_bindnodes);

        list_del(&(cs->data.listener.lh));

        while (list_empty(&(cs->data.listener.conn_queue)) == 0) {
                struct conn *src_sock_o = container_of(
                                cs->data.listener.conn_queue.next,
                                struct conn, source.sock.cl_list);
                list_del(&(src_sock_o->source.sock.cl_list));
                reset_conn(src_sock_o);
                kref_get(&(src_sock_o->reversedir->ref));
                kref_put(&(src_sock_o->ref), free_conn);
        }

        spin_unlock_bh(&cor_bindnodes);
out:
        mutex_unlock(&(cs->lock));
}

int open_port(struct cor_sock *cs_l, __be16 port)
{
        int rc = 0;

        spin_lock_bh(&cor_bindnodes);
        if (get_corsock_by_port(port) != 0) {
                rc = -EADDRINUSE;
                goto out;
        }

        BUG_ON(cs_l->type != CS_TYPE_UNCONNECTED);

        cs_l->type = CS_TYPE_LISTENER;
        cs_l->data.listener.port = port;
        cs_l->data.listener.publish_service = cs_l->publish_service;

        /* kref is not used here */
        INIT_LIST_HEAD(&(cs_l->data.listener.conn_queue));

        list_add_tail((struct list_head *) &(cs_l->data.listener.lh),
                        &openports);

out:
        spin_unlock_bh(&cor_bindnodes);

        return rc;
}

/**
 * rc == 0 connected
 * rc == 2 port not open
 * rc == 3 listener queue full
 */
int connect_port(struct conn *trgt_unconn_l, __be16 port)
{
        struct cor_sock *cs;
        int rc = 0;

        spin_lock_bh(&cor_bindnodes);

        cs = get_corsock_by_port(port);
        if (cs == 0) {
                rc = 2;
                goto out;
        }

        if (unlikely(cs->data.listener.queue_len >=
                        cs->data.listener.queue_maxlen)) {
                if (cs->data.listener.queue_maxlen <= 0)
                        rc = 2;
                else
                        rc = 3;

                goto out;
        }

        kref_get(&(trgt_unconn_l->ref));
        kref_get(&(trgt_unconn_l->reversedir->ref));

        BUG_ON(trgt_unconn_l->is_client != 1);
        spin_lock_bh(&(trgt_unconn_l->reversedir->rcv_lock));
        conn_init_sock_target(trgt_unconn_l);
        conn_init_sock_source(trgt_unconn_l->reversedir);
        spin_unlock_bh(&(trgt_unconn_l->reversedir->rcv_lock));

        list_add_tail(&(trgt_unconn_l->reversedir->source.sock.cl_list),
                        &(cs->data.listener.conn_queue));
        cs->data.listener.queue_len++;
        cs->sk.sk_state_change(&(cs->sk));

out:
        spin_unlock_bh(&cor_bindnodes);
        return rc;
}

/**
 * rc == 0 connected
 * rc == 3 addr not found
 * rc == 4 ==> connid allocation failed
 * rc == 4 ==> control msg alloc failed
 */
int connect_neigh(struct conn *trgt_unconn_l, char *addr, __u16 addrlen)
{
        int rc = 0;
        int ciorc;
        struct control_msg_out *cm;
        struct neighbor *nb = 0;
        __u64 seqno2;

        nb = find_neigh(addr, addrlen);

        if (nb == 0) {
                rc = 3;
                goto discard;
        }

        cm = alloc_control_msg(nb, ACM_PRIORITY_LOW);
        if (unlikely(cm == 0)) {
                rc = 4;
                goto discard;
        }

        spin_lock_bh(&(trgt_unconn_l->reversedir->rcv_lock));
        ciorc = conn_init_out(trgt_unconn_l, nb, 0, 0);
        seqno2 = trgt_unconn_l->reversedir->source.in.next_seqno;
        spin_unlock_bh(&(trgt_unconn_l->reversedir->rcv_lock));
        if (unlikely(ciorc)) {
                rc = 4;
                goto freecm;
        }

        send_connect_nb(cm, trgt_unconn_l->target.out.conn_id,
                        trgt_unconn_l->target.out.seqno_nextsend, seqno2,
                        trgt_unconn_l->reversedir);

        if (0) {
freecm:
                free_control_msg(cm);
discard:
                trgt_unconn_l->targettype = TARGET_DISCARD;
        }

        if (nb != 0)
                kref_put(&(nb->ref), neighbor_free);

        return rc;
}

static int _reset_conn(struct conn *cn, int trgt_out_resetneeded)
{
        /**
         * active conns have an additional ref to make sure that they are not
         * freed when only one direction is referenced by the connid hashtable
         */
        int krefput = 1;

        /* lock sourcetype/targettype */
        spin_lock_bh(&(cn->rcv_lock));

        if (cn->sourcetype == SOURCE_IN) {
                unsigned long iflags;

                spin_lock_irqsave(&(cn->source.in.nb->conn_list_lock), iflags);
                list_del(&(cn->source.in.nb_list));
                spin_unlock_irqrestore(&(cn->source.in.nb->conn_list_lock),
                                iflags);

                krefput++;

                if (cn->source.in.conn_id != 0 &&
                                (cn->source.in.conn_id & (1 << 31)) != 0) {
                        BUG_ON(cn->source.in.cir != 0);
                } else if (cn->source.in.conn_id != 0 &&
                                (cn->source.in.conn_id & (1 << 31)) == 0) {
                        BUG_ON(cn->source.in.cir == 0);

                        kref_init(&(cn->source.in.cir->ref));
                        cn->source.in.cir->conn_id = cn->source.in.conn_id;
                        cn->source.in.cir->pingcnt =
                                        cn->source.in.nb->connid_reuse_pingcnt;

                        spin_lock_bh(&(cn->source.in.nb->connid_reuse_lock));
                        insert_connid_reuse(cn->source.in.nb,
                                        cn->source.in.cir);
                        list_add_tail(&(cn->source.in.cir->lh),
                                        &(cn->source.in.nb->connid_reuse_list));
                        spin_unlock_bh(&(cn->source.in.nb->connid_reuse_lock));

                        cn->source.in.cir = 0;
                }

                if (cn->source.in.conn_id != 0) {
                        spin_lock_bh(&(cn->source.in.nb->connid_lock));
                        rb_erase(&(cn->source.in.rbn),
                                        &(cn->source.in.nb->connid_rb));
                        spin_unlock_bh(&(cn->source.in.nb->connid_lock));
                        krefput++;

                        cn->source.in.conn_id = 0;

                        free_ack_conns(cn);
                }

                if (cn->is_client)
                        atomic_dec(&num_conns);

                reset_ooo_queue(cn);
        } else if (cn->sourcetype == SOURCE_SOCK) {
                if (likely(cn->source.sock.cs != 0)) {
                        cor_sock_flushtoconn(cn->source.sock.cs);
                        kref_put(&(cn->source.sock.cs->ref), free_sock);
                        cn->source.sock.cs = 0;
                }
        }

        if (cn->targettype == TARGET_UNCONNECTED) {
                connreset_cpacket_buffer(cn);
        } else if (cn->targettype == TARGET_OUT) {
                if (trgt_out_resetneeded && cn->target.out.conn_id != 0) {
                        send_reset_conn(cn->target.out.nb,
                                        cn->target.out.conn_id, 0);
                }

                cn->target.out.conn_id = 0;

                cancel_conn_all_retrans(cn);

                qos_remove_conn(cn);

                spin_lock_bh(&(cn->target.out.nb->stalledconn_lock));
                if (cn->target.out.nbstalled_lh.prev != 0) {
                        list_del(&(cn->target.out.nbstalled_lh));
                        cn->target.out.nbstalled_lh.prev = 0;
                        cn->target.out.nbstalled_lh.next = 0;
                        krefput++;
                }
                spin_unlock_bh(&(cn->target.out.nb->stalledconn_lock));
        } else if (cn->targettype == TARGET_SOCK) {
                if (likely(cn->target.sock.cs != 0)) {
                        cor_sock_readfromconn(cn->target.sock.cs);
                        kref_put(&(cn->target.sock.cs->ref), free_sock);
                        cn->target.sock.cs = 0;
                }
        }

        databuf_ackdiscard(cn);

        spin_unlock_bh(&(cn->rcv_lock));

        reset_bufferusage(cn); /* source in only */
        connreset_priority(cn);

        return krefput;
}

/* warning: do not hold the rcv_lock while calling this! */
void reset_conn(struct conn *cn)
{
        int put1;
        int put2;

        int isreset1;
        int isreset2;

        if (cn->is_client) {
                spin_lock_bh(&(cn->rcv_lock));
                spin_lock_bh(&(cn->reversedir->rcv_lock));
        } else {
                spin_lock_bh(&(cn->reversedir->rcv_lock));
                spin_lock_bh(&(cn->rcv_lock));
        }

        BUG_ON(cn->isreset <= 1 && cn->reversedir->isreset >= 2);
        BUG_ON(cn->isreset >= 2 && cn->reversedir->isreset <= 1);

        isreset1 = cn->isreset;
        if (cn->isreset <= 1)
                cn->isreset = 2;

        isreset2 = cn->reversedir->isreset;
        if (cn->reversedir->isreset <= 1)
                cn->reversedir->isreset = 2;

        if (cn->is_client) {
                spin_unlock_bh(&(cn->rcv_lock));
                spin_unlock_bh(&(cn->reversedir->rcv_lock));
        } else {
                spin_unlock_bh(&(cn->reversedir->rcv_lock));
                spin_unlock_bh(&(cn->rcv_lock));
        }

        if (isreset1 >= 2)
                return;

        put1 = _reset_conn(cn, isreset1 == 0);
        put2 = _reset_conn(cn->reversedir, isreset2 == 0);

        /* free_conn may not be called, before both _reset_conn have finished */
        while (put1 > 0) {
                kref_put(&(cn->ref), free_conn);
                put1--;
        }

        while (put2 > 0) {
                kref_put(&(cn->reversedir->ref), free_conn);
                put2--;
        }
}

static int __init cor_common_init(void)
{
        int rc;

        struct conn c;

        printk(KERN_ERR "sizeof conn: %d", (__u32) sizeof(c));
        printk(KERN_ERR "  conn.source: %d", (__u32) sizeof(c.source));
        printk(KERN_ERR "  conn.target: %d", (__u32) sizeof(c.target));
        printk(KERN_ERR "  conn.target.out: %d", (__u32) sizeof(c.target.out));
        printk(KERN_ERR "  conn.buf: %d", (__u32) sizeof(c.data_buf));

        printk(KERN_ERR "  mutex: %d", (__u32) sizeof(struct mutex));
        printk(KERN_ERR "  spinlock: %d", (__u32) sizeof(spinlock_t));
        printk(KERN_ERR "  kref: %d", (__u32) sizeof(struct kref));


        rc = cor_util_init();
        if (unlikely(rc != 0))
                return rc;

        conn_slab = kmem_cache_create("cor_conn", sizeof(struct conn), 8, 0, 0);
        if (unlikely(conn_slab == 0))
                return -ENOMEM;

        connid_reuse_slab = kmem_cache_create("cor_connid_reuse",
                        sizeof(struct connid_reuse_item), 8, 0, 0);
        if (unlikely(connid_reuse_slab == 0))
                return -ENOMEM;


        atomic_set(&num_conns, 0);
        barrier();

        rc = credits_init();
        if (unlikely(rc != 0))
                return rc;

        rc = forward_init();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_kgen_init();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_cpacket_init();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_rd_init1();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_snd_init();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_neighbor_init();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_rcv_init();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_sock_init1();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_sock_init2();
        if (unlikely(rc != 0))
                return rc;

        rc = cor_rd_init2();
        if (unlikely(rc != 0))
                return rc;

        #warning todo add_random_ready_callback

        return 0;
}

module_init(cor_common_init);
MODULE_LICENSE("GPL");