remove nbstalled_list

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

/**
 *      Connection oriented routing
 *      Copyright (C) 2007-2021 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 LIST_HEAD(cor_openports);

static struct kmem_cache *cor_conn_slab;
struct kmem_cache *cor_connid_reuse_slab;

atomic_t cor_num_conns;

int cor_new_incoming_conn_allowed(struct cor_neighbor *nb)
{
        /**
         * MAX_CONNS is only loosly enforced for now
         * (cor_num_conns is not checked and incremented at the same time)
         */
        if (atomic_read(&cor_num_conns) >= MAX_CONNS)
                return 0;
        else
                return 1;
}

static __u64 cor_get_prio_in(__u64 priority)
{
        if (PRIORITY_IN_MULITPLIER_PERCENT >= 100)
                return priority;

        if (unlikely(priority > U64_MAX/100)) {
                return (priority / 100) * PRIORITY_IN_MULITPLIER_PERCENT;
        } else {
                return (priority * PRIORITY_IN_MULITPLIER_PERCENT) / 100;
        }
}

static __u32 cor_conn_prio_sum_limit(__u32 priority, __u64 priority_sum)
{
        __u32 shiftcnt = 0;

        while ((PRIORITY_SUM_IN_MAX >> shiftcnt) > U32_MAX) {
                shiftcnt++;
        }

        return div_u64(((__u64) priority) * (PRIORITY_SUM_IN_MAX >> shiftcnt),
                        (priority_sum >> shiftcnt));
}

__u32 _cor_conn_refresh_priority(struct cor_conn *cn_lx)
{
        BUG_ON(cn_lx->is_client == 0);

        if (cn_lx->sourcetype == SOURCE_IN) {
                __u32 priority = (__u32)
                                cor_get_prio_in(cn_lx->source.in.priority);
                __u64 priority_sum = cor_get_prio_in(atomic64_read(
                                &(cn_lx->source.in.nb->priority_sum)));
                if (PRIORITY_SUM_IN_MAX != U64_MAX &&
                                priority_sum > PRIORITY_SUM_IN_MAX) {
                        return cor_conn_prio_sum_limit(priority, priority_sum);
                } else {
                        return priority;
                }
        } else if (cn_lx->sourcetype == SOURCE_SOCK) {
                return cn_lx->source.sock.priority;
        } else {
                BUG();
                return 0;
        }
}

__u32 cor_conn_refresh_priority(struct cor_conn *cn, int locked)
{
        struct cor_conn *cn_reversedir = cor_get_conn_reversedir(cn);
        __u32 priority = 0;

        if (likely(locked == 0)) {
                if (cn->is_client == 0)
                        return cor_conn_refresh_priority(cn_reversedir, 0);

                spin_lock_bh(&(cn->rcv_lock));
                spin_lock_bh(&(cor_get_conn_reversedir(cn)->rcv_lock));
        } else {
                BUG_ON(cn->is_client == 0);
        }

        if (unlikely(cn->isreset != 0) || cn->targettype != TARGET_OUT)
                goto out;

        priority = _cor_conn_refresh_priority(cn);

        if (cn->target.out.priority_send_allowed != 0) {
                __u16 priority_enc = cor_enc_priority(priority);
                if (priority_enc != cn->target.out.priority_last)
                        cor_send_priority(cn, priority_enc);
        }

out:
        if (likely(locked == 0)) {
                spin_unlock_bh(&(cor_get_conn_reversedir(cn)->rcv_lock));
                spin_unlock_bh(&(cn->rcv_lock));
        }

        return priority;
}

static void _cor_set_conn_in_priority(struct cor_conn *src_in_lx,
                __u32 newpriority)
{
        struct cor_neighbor *nb = src_in_lx->source.in.nb;

        __u32 oldpriority = src_in_lx->source.in.priority;

        cor_update_atomic_sum(&(nb->priority_sum),
                        oldpriority, newpriority);

        src_in_lx->source.in.priority = newpriority;
}

void cor_set_conn_in_priority(struct cor_neighbor *nb, __u32 conn_id,
                struct cor_conn *src_in, __u8 priority_seqno, __u16 priority)
{
        __u32 newpriority;

        if (unlikely(src_in->is_client == 0))
                return;

        spin_lock_bh(&(src_in->rcv_lock));
        spin_lock_bh(&(cor_get_conn_reversedir(src_in)->rcv_lock));

        if (unlikely(cor_is_conn_in(src_in, nb, conn_id) == 0))
                goto out;

        if (src_in->source.in.priority_seqno != priority_seqno)
                goto out;
        src_in->source.in.priority_seqno =
                        (src_in->source.in.priority_seqno + 1) & 15;

        newpriority = (cor_dec_priority(priority)*4)/5;
        _cor_set_conn_in_priority(src_in, newpriority);
        cor_conn_refresh_priority(src_in, 1);

out:
        spin_unlock_bh(&(cor_get_conn_reversedir(src_in)->rcv_lock));
        spin_unlock_bh(&(src_in->rcv_lock));
}

static void cor_connreset_priority(struct cor_conn *cn_lx)
{
        if (cn_lx->is_client == 0)
                return;

        if (cn_lx->sourcetype == SOURCE_IN)
                _cor_set_conn_in_priority(cn_lx, 0);
}

void _cor_set_last_act(struct cor_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);
}

static void _cor_free_conn(struct cor_conn *cn)
{
        BUG_ON(cn->isreset == 0);

        if (cn->sourcetype == SOURCE_IN) {
                WARN_ONCE(list_empty(&(cn->source.in.reorder_queue)) == 0,
                                "cor_free_conn(): cn->source.in.reorder_queue is not empty");
                WARN_ONCE(list_empty(&(cn->source.in.acks_pending)) == 0,
                                "cor_free_conn():cn->source.in.acks_pending is not empty");

                WARN_ONCE(cn->source.in.conn_id != 0,
                                "cor_free_conn(): cn->source.in.conn_id is not 0");
                kref_put(&(cn->source.in.nb->ref), cor_neighbor_free);
                cn->source.in.nb = 0;
        }

        if (cn->targettype == TARGET_OUT) {
                WARN_ONCE(list_empty(&(cn->target.out.retrans_list)) == 0,
                                "cor_free_conn(): cn->target.out.retrans_list is not empty");
                WARN_ONCE(cn->target.out.rb.in_queue != RB_INQUEUE_FALSE,
                                "cor_free_conn(): cn->target.out.rb.in_queue is not RB_INQUEUE_FALSE");
                WARN_ONCE(cn->target.out.conn_id != 0,
                                "cor_free_conn(): cn->target.out.conn_id is not 0");
                kref_put(&(cn->target.out.nb->ref), cor_neighbor_free);
                cn->target.out.nb = 0;
        }

        WARN_ONCE(cn->data_buf.datasize != 0,
                        "cor_free_conn(): cn->data_buf.datasize is not 0");
        WARN_ONCE(cn->data_buf.overhead != 0,
                        "cor_free_conn(): cn->data_buf.overhead is not 0");
        WARN_ONCE(list_empty(&(cn->data_buf.items)) == 0,
                        "cor_free_conn(): cn->data_buf.items is not empty");
        WARN_ONCE(cn->data_buf.nextread != 0,
                        "cor_free_conn(): cn->data_buf.nextread is not 0");
}

void cor_free_conn(struct kref *ref)
{
        struct cor_conn_bidir *cnb = container_of(ref, struct cor_conn_bidir,
                        ref);

        _cor_free_conn(&(cnb->cli));
        _cor_free_conn(&(cnb->srv));

        memset(cnb, 9*16 + 10, sizeof(struct cor_conn_bidir));
        kmem_cache_free(cor_conn_slab, cnb);
}

/**
 * rc == 0 ==> ok
 * rc == 1 ==> connid_reuse or connid allocation failed
 */
int cor_conn_init_out(struct cor_conn *trgt_unconn_ll, struct cor_neighbor *nb,
                __u32 rcvd_connid, int use_rcvd_connid)
{
        unsigned long iflags;
        int rc = 0;
        struct cor_conn *src_unconn_ll =
                        cor_get_conn_reversedir(trgt_unconn_ll);
        __u8 tmp;

        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(cor_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(
                                cor_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 cor_connid_reuse_item));

                if (unlikely(cor_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));

        cor_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;
                trgt_unconn_ll->target.out.seqno_windowlimit =
                                trgt_unconn_ll->target.out.seqno_nextsend;
                cor_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 *) &tmp, 1);
        trgt_unconn_ll->target.out.priority_seqno = (tmp & 15);

        src_unconn_ll->source.in.priority_seqno = 0;

        src_unconn_ll->source.in.jiffies_last_act = jiffies;

        trgt_unconn_ll->target.out.jiffies_idle_since =
                        jiffies << JIFFIES_LAST_IDLE_SHIFT;

        trgt_unconn_ll->target.out.remote_bufsize_changerate = 64;

        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 */
        cor_conn_kref_get(src_unconn_ll, "neighbor_list");

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

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

        if (0) {
out_freecir:
                kmem_cache_free(cor_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 cor_conn_init_sock_source(struct cor_conn *cn)
{
        BUG_ON(cn == 0);
        cn->sourcetype = SOURCE_SOCK;
        memset(&(cn->source.sock), 0, sizeof(cn->source.sock));
        cn->source.sock.priority = cor_priority_max();
        cn->source.sock.snd_speed.jiffies_last_refresh = jiffies;
        cn->source.sock.snd_speed.flushed = 1;
}

void cor_conn_init_sock_target(struct cor_conn *cn)
{
        BUG_ON(cn == 0);
        cn->targettype = TARGET_SOCK;
        memset(&(cn->target.sock), 0, sizeof(cn->target.sock));
        cor_reset_seqno(cn, 0);
}

static void _cor_alloc_conn(struct cor_conn *cn, __u8 is_highlatency)
{
        cn->sourcetype = SOURCE_UNCONNECTED;
        cn->targettype = TARGET_UNCONNECTED;

        cn->isreset = 0;

        spin_lock_init(&(cn->rcv_lock));

        cor_databuf_init(cn);

        cor_bufsize_init(cn, 0);

        if (is_highlatency == 0) {
                cn->is_highlatency = 0;
                cn->bufsize.bufsize =
                                (BUFSIZE_INITIAL_LOWLAT << BUFSIZE_SHIFT);
        } else {
                cn->is_highlatency = 1;
                cn->bufsize.bufsize =
                                (BUFSIZE_INITIAL_HIGHLAT << BUFSIZE_SHIFT);
        }
}

struct cor_conn_bidir* cor_alloc_conn(gfp_t allocflags, __u8 is_highlatency)
{
        struct cor_conn_bidir *cnb;

        cnb = kmem_cache_alloc(cor_conn_slab, allocflags);
        if (unlikely(cnb == 0))
                return 0;

        memset(cnb, 0, sizeof(struct cor_conn_bidir));

        cnb->cli.is_client = 1;
        kref_init(&(cnb->ref));

        _cor_alloc_conn(&(cnb->cli), is_highlatency);
        _cor_alloc_conn(&(cnb->srv), is_highlatency);

        return cnb;
}

static struct cor_sock *cor_get_corsock_by_port(__be32 port)
{
        struct list_head *curr = cor_openports.next;

        while (curr != &cor_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 cor_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 = cor_openports.next;
        while (curr != &cor_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 + 4 < buf_offset) ||
                                buf_offset + 4 > buflen)
                        break;

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

cont:
                curr = curr->next;
        }

        spin_unlock_bh(&cor_bindnodes);

        rc = cor_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 cor_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 cor_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 cor_conn *src_sock_o = container_of(
                                cs->data.listener.conn_queue.next,
                                struct cor_conn, source.sock.cl_list);
                BUG_ON(src_sock_o->source.sock.in_cl_list == 0);
                list_del(&(src_sock_o->source.sock.cl_list));
                src_sock_o->source.sock.in_cl_list = 0;
                spin_unlock_bh(&cor_bindnodes);

                cor_reset_conn(src_sock_o);

                spin_lock_bh(&cor_bindnodes);
                cor_conn_kref_put(src_sock_o, "conn_queue");
        }

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

int cor_open_port(struct cor_sock *cs_l, __be32 port)
{
        int rc = 0;

        spin_lock_bh(&cor_bindnodes);
        if (cor_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),
                        &cor_openports);

out:
        spin_unlock_bh(&cor_bindnodes);

        return rc;
}

/**
 * rc == 0 connected
 * rc == 2 port not open
 * rc == 3 listener queue full
 */
int cor_connect_port(struct cor_conn *trgt_unconn_ll, __be32 port)
{
        struct cor_conn *src_unconn_ll =
                        cor_get_conn_reversedir(trgt_unconn_ll);
        struct cor_sock *cs;
        int rc = 0;

        spin_lock_bh(&cor_bindnodes);

        cs = cor_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;
        }

        BUG_ON(trgt_unconn_ll->is_client != 1);
        cor_conn_init_sock_target(trgt_unconn_ll);
        cor_conn_init_sock_source(src_unconn_ll);

        list_add_tail(&(src_unconn_ll->source.sock.cl_list),
                        &(cs->data.listener.conn_queue));
        src_unconn_ll->source.sock.in_cl_list = 1;
        cor_conn_kref_get(src_unconn_ll, "conn_queue");

        cs->data.listener.queue_len++;
        atomic_set(&(cs->ready_to_accept), 1);
        barrier();
        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 cor_connect_neigh(struct cor_conn *trgt_unconn_ll, char *addr,
                __u16 addrlen)
{
        struct cor_control_msg_out *cm;
        struct cor_neighbor *nb = 0;

        nb = cor_find_neigh(addr, addrlen);
        if (nb == 0)
                return 3;

        cm = cor_alloc_control_msg(nb, ACM_PRIORITY_MED);
        if (unlikely(cm == 0)) {
                kref_put(&(nb->ref), cor_neighbor_free);
                return 4;
        }

        if (unlikely(cor_conn_init_out(trgt_unconn_ll, nb, 0, 0))) {
                cor_free_control_msg(cm);
                kref_put(&(nb->ref), cor_neighbor_free);
                return 4;
        }

        trgt_unconn_ll->target.out.priority_last =
                        _cor_conn_refresh_priority(trgt_unconn_ll);

        cor_send_connect_nb(cm, trgt_unconn_ll->target.out.conn_id,
                        trgt_unconn_ll->target.out.seqno_nextsend,
                        cor_get_conn_reversedir(trgt_unconn_ll)->
                        source.in.next_seqno,
                        cor_get_conn_reversedir(trgt_unconn_ll));

        kref_put(&(nb->ref), cor_neighbor_free);

        return 0;
}

static void _cor_reset_conn(struct cor_conn *cn_ll, int trgt_out_resetneeded)
{
        if (cn_ll->sourcetype == SOURCE_IN) {
                unsigned long iflags;
                struct cor_neighbor *nb = cn_ll->source.in.nb;

                spin_lock_irqsave(&(nb->conn_list_lock), iflags);
                list_del(&(cn_ll->source.in.nb_list));
                spin_unlock_irqrestore(&(nb->conn_list_lock), iflags);
                cor_conn_kref_put_bug(cn_ll, "neighbor_list");

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

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

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

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

                if (cn_ll->source.in.conn_id != 0) {
                        spin_lock_bh(&(nb->connid_lock));
                        rb_erase(&(cn_ll->source.in.rbn), &(nb->connid_rb));
                        spin_unlock_bh(&(nb->connid_lock));
                        cor_conn_kref_put_bug(cn_ll, "connid_table");

                        cn_ll->source.in.conn_id = 0;

                        cor_free_ack_conns(cn_ll);
                }

                if (cn_ll->is_client)
                        atomic_dec(&cor_num_conns);

                cor_reset_ooo_queue(cn_ll);
        } else if (cn_ll->sourcetype == SOURCE_SOCK) {
                if (likely(cn_ll->source.sock.cs != 0)) {
                        cor_sk_write_space(cn_ll->source.sock.cs);
                        kref_put(&(cn_ll->source.sock.cs->ref), cor_free_sock);
                        cn_ll->source.sock.cs = 0;
                }
                if (unlikely(cn_ll->source.sock.in_cl_list != 0)) {
                        list_del(&(cn_ll->source.sock.cl_list));
                        cn_ll->source.sock.in_cl_list = 0;
                        cor_conn_kref_put_bug(cn_ll, "conn_queue");
                }
        }

        if (cn_ll->targettype == TARGET_UNCONNECTED) {
                if (cn_ll->target.unconnected.cmdparams != 0) {
                        kfree(cn_ll->target.unconnected.cmdparams);
                        cn_ll->target.unconnected.cmdparams = 0;
                }
        } else if (cn_ll->targettype == TARGET_OUT) {
                if (trgt_out_resetneeded && cn_ll->target.out.conn_id != 0) {
                        cor_send_reset_conn(cn_ll->target.out.nb,
                                        cn_ll->target.out.conn_id, 0);
                }

                cn_ll->target.out.conn_id = 0;

                cor_cancel_all_conn_retrans(cn_ll);

                cor_qos_remove_conn(cn_ll);
        } else if (cn_ll->targettype == TARGET_SOCK) {
                if (likely(cn_ll->target.sock.cs != 0)) {
                        if (cn_ll->target.sock.socktype == SOCKTYPE_RAW) {
                                cor_sk_data_ready(cn_ll->target.sock.cs);
                        } else {
                                cor_mngdsocket_readfromconn_fromatomic(
                                                cn_ll->target.sock.cs);
                        }
                        kref_put(&(cn_ll->target.sock.cs->ref), cor_free_sock);
                        cn_ll->target.sock.cs = 0;
                        cn_ll->target.sock.rcv_buf = 0;
                }
        }

        cor_databuf_ackdiscard(cn_ll);

        cor_account_bufspace(cn_ll);

        cor_connreset_priority(cn_ll);
}

void cor_reset_conn_locked(struct cor_conn_bidir *cnb_ll)
{
        BUG_ON(cnb_ll->cli.isreset <= 1 && cnb_ll->srv.isreset == 2);
        BUG_ON(cnb_ll->cli.isreset == 2 && cnb_ll->srv.isreset <= 1);

        if (cnb_ll->cli.isreset <= 1) {
                __u8 old_isreset_cli = cnb_ll->cli.isreset;
                __u8 old_isreset_srv = cnb_ll->srv.isreset;

                cnb_ll->cli.isreset = 2;
                cnb_ll->srv.isreset = 2;

                _cor_reset_conn(&(cnb_ll->cli), old_isreset_cli == 0);
                _cor_reset_conn(&(cnb_ll->srv), old_isreset_srv == 0);
        }
}

void cor_reset_conn(struct cor_conn *cn)
{
        struct cor_conn_bidir *cnb = cor_get_conn_bidir(cn);

        cor_conn_kref_get(&(cnb->cli), "stack");
        cor_conn_kref_get(&(cnb->srv), "stack");

        spin_lock_bh(&(cnb->cli.rcv_lock));
        spin_lock_bh(&(cnb->srv.rcv_lock));

        cor_reset_conn_locked(cnb);

        spin_unlock_bh(&(cnb->srv.rcv_lock));
        spin_unlock_bh(&(cnb->cli.rcv_lock));

        cor_conn_kref_put_bug(&(cnb->cli), "stack");
        cor_conn_kref_put(&(cnb->srv), "stack");
}

static int __init cor_init(void)
{
        int rc;

        struct cor_conn_bidir cb;
        struct cor_conn c;


        printk(KERN_ERR "sizeof cor_conn_bidir: %u", (__u32) sizeof(cb));
        printk(KERN_ERR "sizeof conn: %u", (__u32) sizeof(c));
        printk(KERN_ERR "  conn.source: %u", (__u32) sizeof(c.source));
        printk(KERN_ERR "  conn.source.in: %u", (__u32) sizeof(c.source.in));
        printk(KERN_ERR "  conn.target: %u", (__u32) sizeof(c.target));
        printk(KERN_ERR "  conn.target.out: %u", (__u32) sizeof(c.target.out));
        printk(KERN_ERR "  conn.data_buf: %u", (__u32) sizeof(c.data_buf));
        printk(KERN_ERR "  conn.bufsize: %u", (__u32) sizeof(c.bufsize));

        printk(KERN_ERR "sizeof cor_neighbor: %u",
                        (__u32) sizeof(struct cor_neighbor));

        printk(KERN_ERR "sizeof mutex: %u", (__u32) sizeof(struct mutex));
        printk(KERN_ERR "sizeof spinlock: %u", (__u32) sizeof(spinlock_t));
        printk(KERN_ERR "sizeof kref: %u", (__u32) sizeof(struct kref));
        printk(KERN_ERR "sizeof list_head: %u",
                        (__u32) sizeof(struct list_head));
        printk(KERN_ERR "sizeof rb_root: %u", (__u32) sizeof(struct rb_root));
        printk(KERN_ERR "sizeof rb_node: %u", (__u32) sizeof(struct rb_node));


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

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

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


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

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

        rc = cor_kgen_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_neigh_ann_rcv_init();
        if (unlikely(rc != 0))
                return rc;

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

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

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

        rc = cor_conn_src_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;

        return 0;
}

static void __exit cor_exit(void)
{
        cor_rd_exit1();
        cor_sock_exit1();
        cor_conn_src_sock_exit1();
        cor_dev_exit1();

        flush_scheduled_work();

        cor_rcv_exit2();
        cor_neighbor_exit2();
        cor_neigh_ann_rcv_exit2();
        cor_snd_exit2();
        cor_rd_exit2();
        cor_kgen_exit2();
        cor_forward_exit2();

        BUG_ON(atomic_read(&cor_num_conns) != 0);

        kmem_cache_destroy(cor_conn_slab);
        cor_conn_slab = 0;

        kmem_cache_destroy(cor_connid_reuse_slab);
        cor_connid_reuse_slab = 0;
}

module_init(cor_init);
module_exit(cor_exit);
MODULE_LICENSE("GPL");