connreset_cpacket_buffer, credits, enlarge window upon resume, unconnected->buffer_wa...

[cor_2_6_31.git] / net / cor / rcv.c

/*
 *      Connection oriented routing
 *      Copyright (C) 2007-2010 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/module.h>
#include <linux/version.h>
#include <linux/kernel.h>
#include <linux/init.h>
#include <linux/in.h>


#include "cor.h"

static atomic_t packets_in_workqueue = ATOMIC_INIT(0);

static atomic_t ooo_packets = ATOMIC_INIT(0);

static struct workqueue_struct *packet_wq;

static struct work_struct outofbufferspace_work;
DEFINE_SPINLOCK(oobss_lock);
static int outofbufferspace_scheduled;

/**
 * buffering space is divided in 4 areas:
 *
 * buffer_init:
 * distributed equally among all conns; shrinks and grows immediately when there
 * are new connections or connections are reset
 *
 * buffer_speed:
 * distributed proportional to speed; shrinks and grows constantly
 *
 * buffer_ata:
 * used to make noise to make traffic analysis harder; may only shrink if data
 * is either send on or "stuck" for a long time
 *
 * buffer_reserve:
 * reserve in case where sudden shrinking causes some connections to contain
 * more old data than allowed by the first 3 buffers. If this area is full, the
 * out-of-memory conn-resetter is triggered
 *
 * Each area commits a certain amount of space to each connection. This is the
 * maximum buffer space a connection is allowed to use. The space of a specific
 * connection is first accounted to buffer_ata. If the buffer space allowed to
 * use is exceeded, the rest is accounted to buffer_speed and then buffer_init.
 * The reserve area will be used last. This should only be the case, if the
 * assigned buffer space of the first 3 areas shrink suddenly. If this area is
 * also used up, connections will be reset.
 */

DEFINE_MUTEX(buffer_conn_list_lock);
LIST_HEAD(buffer_conn_list);

/**
 * used to buffer inserts when main list is locked, moved to main list, after
 * processing of main list finishes
 */
LIST_HEAD(buffer_conn_tmp_list); /* protected by bufferlimits_lock */


DEFINE_MUTEX(bufferlimits_lock);

static __u64 bufferassigned_init;
static __u64 bufferassigned_speed;
static __u64 bufferassigned_ata;

static __u64 bufferusage_init;
static __u64 bufferusage_speed;
static __u64 bufferusage_ata;
static __u64 bufferusage_reserve;

DEFINE_SPINLOCK(st_lock);
static struct speedtracker st;

static __u64 desired_bufferusage(__u64 assigned, __u64 usage, __u64 assignlimit,
                __u64 usagelimit)
{
        __u64 ret;
        __u64 load;

        if (unlikely(assignlimit < usagelimit))
                assignlimit = usagelimit;

        if (multiply_div(usage, 9, 10) > usagelimit)
                return multiply_div(usagelimit, 9, 10);

        load = multiply_div(usage, 192, usagelimit);

        /* slow limit increase, fast decrease */
        if (load == 128) {
                ret = assigned;
        } else if (load < 128) {
                if (load == 0)
                        return multiply_div(usagelimit, 9, 10);

                if (load < 96)
                        load = 96;
                ret = multiply_div(assigned, 128, load);
        } else {
                ret = multiply_div(assigned, 128, load + load - 128);
        }

        if (ret > assignlimit)
                return assignlimit;
        return ret;
}

#warning todo changing speed too fast
static __u64 get_speed(struct speedtracker *st, unsigned long jiffies_tmp)
{
        if (unlikely(time_after(st->jiffies_last_update, jiffies_tmp) ||
                        time_before(st->jiffies_last_update + HZ*10,
                        jiffies_tmp))) {
                st->jiffies_last_update = jiffies_tmp;
                st->speed = 0;
                st->bytes_curr = 0;
                return 0;
        }

        for (;time_before(st->jiffies_last_update, jiffies_tmp);
                        st->jiffies_last_update++) {
                __u32 bytes_curr = 0;
                if ((st->jiffies_last_update + 1) == jiffies_tmp) {
                        bytes_curr = st->bytes_curr;
                        st->bytes_curr = 0;
                }
                st->speed = (st->speed * (HZ-1) + (((__u64)bytes_curr)<<16))/HZ;
        }

        if ((st->jiffies_last_update + 1) == jiffies_tmp) {
                st->jiffies_last_update++;
        }

        return (st->speed * (HZ*2 - 1) + (((__u64)st->bytes_curr) << 17))/HZ/2;
}

/**
 * val1[0], val2[2], res[0] ... least significant
 * val1[val1len-1], val2[val2len-1], res[reslen-1] ... most significant
 */
static void mul(__u32 *val1, unsigned int val1len, __u32 *val2,
                unsigned int val2len, __u32 *res, int reslen)
{
        int digits = val1len + val2len;
        __u64 overflow = 0;
        int i;

        BUG_ON(val1len > 0 && val2len > 0 && reslen < digits);

        memset(res, 0, reslen);

        if (val1len == 0 || val2len == 0)
                return;

        for(i=0;i<digits;i++) {
                int idx1;
                res[i] = (__u32) overflow;
                overflow = overflow >> 32;
                for(idx1=0;idx1<val1len && idx1<=i;idx1++) {
                        int idx2 = i - idx1;
                        __u64 tmpres;

                        if (idx2 >= val2len)
                                continue;

                        tmpres = ((__u64) (val1[idx1])) *
                                        ((__u64) (val2[idx2]));
                        overflow += tmpres >> 32;
                        tmpres = (tmpres << 32) >> 32;
                        if (res[i] + tmpres < res[i])
                                overflow++;
                        res[i] += tmpres;
                }
        }
        BUG_ON(overflow != 0);
}

/**
 * return values:
 * 1 == usage1/speed1 offends more
 * 0 == both offend the same
 * -1 == usage2/speed2 offends more
 */
static int compare_scores(__u32 usage1, __u64 speed1, __u32 usage2,
                __u64 speed2)
{
        int i;

        __u32 speed1squared[4];
        __u32 speed2squared[4];

        __u32 speed1squared_usage2[5];
        __u32 speed2squared_usage1[5];


        __u32 speed1_tmp[2];
        __u32 speed2_tmp[2];

        speed1_tmp[0] = (speed1 << 32) >> 32;
        speed1_tmp[1] = (speed1 >> 32);
        speed2_tmp[0] = (speed2 << 32) >> 32;
        speed2_tmp[1] = (speed2 << 32);

        mul(speed1_tmp, 2, speed1_tmp, 2, speed1squared,4);
        mul(speed2_tmp, 2, speed2_tmp, 2, speed2squared, 4);

        mul(speed1squared, 4, &usage2, 1, speed1squared_usage2, 5);
        mul(speed2squared, 4, &usage2, 1, speed2squared_usage1, 5);

        for(i=4;i>=0;i++) {
                if (speed1squared_usage2[i] > speed2squared_usage1[i])
                        return -1;
                if (speed1squared_usage2[i] < speed2squared_usage1[i])
                        return 1;
        }

        return 0;
}

#define OOBS_SIZE 10
static void _outofbufferspace(void)
{
        int i;

        struct list_head *curr;
        struct conn *offendingconns[OOBS_SIZE];
        __u32 offendingusage[OOBS_SIZE];
        __u64 offendingspeed[OOBS_SIZE];

        memset(&offendingconns, 0, sizeof(offendingconns));

        mutex_lock(&buffer_conn_list_lock);

        curr = buffer_conn_list.next;
        while (curr != &buffer_conn_list) {
                unsigned long iflags;

                struct conn *conn = container_of(curr, struct conn,
                                source.in.buffer_list);

                __u32 usage;
                __u64 speed;

                int i;

                curr = curr->next;

                mutex_lock(&(conn->rcv_lock));

                BUG_ON(conn->sourcetype != SOURCE_IN);

                usage = conn->source.in.usage_reserve;

                spin_lock_irqsave(&st_lock, iflags);
                speed = get_speed(&(conn->source.in.st), jiffies);
                spin_unlock_irqrestore(&st_lock, iflags);

                mutex_unlock(&(conn->rcv_lock));

                if (offendingconns[OOBS_SIZE-1] != 0 &&
                                compare_scores(
                                offendingusage[OOBS_SIZE-1],
                                offendingspeed[OOBS_SIZE-1],
                                usage, speed) >= 0)
                        continue;

                if (offendingconns[OOBS_SIZE-1] != 0)
                        kref_put(&(offendingconns[OOBS_SIZE-1]->ref),
                                        free_conn);

                kref_get(&(conn->ref));
                offendingconns[OOBS_SIZE-1] = conn;
                offendingusage[OOBS_SIZE-1] = usage;
                offendingspeed[OOBS_SIZE-1] = speed;

                for (i=OOBS_SIZE-2;i>=0;i++) {
                        struct conn *tmpconn;
                        __u32 usage_tmp;
                        __u64 speed_tmp;

                        if (offendingconns[i] != 0 && compare_scores(
                                        offendingusage[i], offendingspeed[i],
                                        offendingusage[i+1],
                                        offendingspeed[i+1]) >= 0)
                                break;

                        tmpconn = offendingconns[i];
                        usage_tmp = offendingusage[i];
                        speed_tmp = offendingspeed[i];

                        offendingconns[i] = offendingconns[i+1];
                        offendingusage[i] = offendingusage[i+1];
                        offendingspeed[i] = offendingspeed[i+1];

                        offendingconns[i+1] = tmpconn;
                        offendingusage[i+1] = usage_tmp;
                        offendingspeed[i+1] = speed_tmp;
                }
        }

        for (i=0;i<OOBS_SIZE;i++) {
                kref_get(&(offendingconns[i]->ref));
        }

        mutex_unlock(&buffer_conn_list_lock);

        for (i=0;i<OOBS_SIZE;i++) {
                int resetneeded;

                if (offendingconns[i] == 0)
                        break;

                mutex_lock(&bufferlimits_lock);
                resetneeded = ((bufferusage_reserve*4)/3 > BUFFERSPACE_RESERVE);
                mutex_unlock(&bufferlimits_lock);

                if (resetneeded)
                        reset_conn(offendingconns[i]);
                kref_put(&(offendingconns[i]->ref), free_conn);
        }

        mutex_lock(&bufferlimits_lock);
        mutex_lock(&buffer_conn_list_lock);
        while(list_empty(&buffer_conn_tmp_list) == 0) {
                curr = buffer_conn_tmp_list.next;
                list_del(curr);
                list_add(curr, &buffer_conn_list);
        }
        mutex_unlock(&buffer_conn_list_lock);
        mutex_unlock(&bufferlimits_lock);
}

static void outofbufferspace(struct work_struct *work)
{
        while (1) {
                unsigned long iflags;
                int resetneeded;

                mutex_lock(&bufferlimits_lock);
                spin_lock_irqsave(&oobss_lock, iflags);
                resetneeded = (bufferusage_reserve > BUFFERSPACE_RESERVE);

                if (resetneeded == 0)
                        outofbufferspace_scheduled = 0;

                spin_unlock_irqrestore(&oobss_lock, iflags);
                mutex_unlock(&bufferlimits_lock);

                if (resetneeded == 0)
                        return;

                _outofbufferspace();
        }
}

static void refresh_bufferusage(struct conn *rconn)
{
        BUG_ON(rconn->sourcetype != SOURCE_IN);

        bufferusage_init -= rconn->source.in.usage_init;
        bufferusage_speed -= rconn->source.in.usage_speed;
        bufferusage_ata -= rconn->source.in.usage_ata;
        bufferusage_reserve -= rconn->source.in.usage_reserve;

        rconn->source.in.usage_ata = rconn->data_buf.totalsize;
        if (rconn->source.in.usage_ata > rconn->source.in.buffer_ata)
                rconn->source.in.usage_ata = rconn->source.in.buffer_ata;


        if (rconn->source.in.usage_ata == rconn->data_buf.totalsize)
                rconn->source.in.usage_speed = 0;
        else
                rconn->source.in.usage_speed = rconn->data_buf.totalsize -
                                rconn->source.in.usage_ata;

        if (rconn->source.in.usage_speed > rconn->source.in.buffer_speed)
                rconn->source.in.usage_speed = rconn->source.in.buffer_speed;


        if ((rconn->source.in.usage_ata + rconn->source.in.usage_speed) ==
                        rconn->data_buf.totalsize)
                rconn->source.in.usage_init = 0;
        else
                rconn->source.in.usage_init = rconn->data_buf.totalsize -
                                rconn->source.in.usage_ata -
                                rconn->source.in.usage_speed;

        if (rconn->source.in.usage_init > rconn->source.in.buffer_init)
                rconn->source.in.usage_init = rconn->source.in.buffer_init;


        if ((rconn->source.in.usage_ata + rconn->source.in.usage_speed +
                        rconn->source.in.usage_init) ==
                        rconn->data_buf.totalsize)
                rconn->source.in.usage_reserve = 0;
        else
                rconn->source.in.usage_reserve = rconn->data_buf.totalsize -
                                rconn->source.in.usage_ata -
                                rconn->source.in.usage_speed -
                                rconn->source.in.usage_init;

        bufferusage_init += rconn->source.in.usage_init;
        bufferusage_speed += rconn->source.in.usage_speed;
        bufferusage_ata += rconn->source.in.usage_ata;
        bufferusage_reserve += rconn->source.in.usage_reserve;

        if (bufferusage_reserve > BUFFERSPACE_RESERVE) {
                unsigned long iflags;
                spin_lock_irqsave(&oobss_lock, iflags);
                if (outofbufferspace_scheduled == 0) {
                        schedule_work(&outofbufferspace_work);
                        outofbufferspace_scheduled = 1;
                }

                spin_unlock_irqrestore(&oobss_lock, iflags);
        }
}

static __u8 __get_window(struct conn *rconn)
{
        __u64 window = 0;

        if (rconn->source.in.usage_reserve != 0)
                return 0;

        BUG_ON(rconn->source.in.usage_init > rconn->source.in.buffer_init);
        BUG_ON(rconn->source.in.usage_speed > rconn->source.in.buffer_speed);
        BUG_ON(rconn->source.in.usage_ata > rconn->source.in.buffer_ata);

        window += rconn->source.in.buffer_init;
        window += rconn->source.in.buffer_speed;
        window += rconn->source.in.buffer_ata;

        window -= rconn->source.in.usage_init;
        window -= rconn->source.in.usage_speed;
        window -= rconn->source.in.usage_ata;

        if (window > MAX_ANNOUNCE_WINDOW)
                window = MAX_ANNOUNCE_WINDOW;

        return enc_window(window);
}

static __u32 _get_window(struct conn *rconn, int listlocked)
{
        unsigned long iflags;

        unsigned long jiffies_tmp;

        __u8 window = 0;

        __s32 conns;
        __u64 bufferlimit_init;
        __u64 connlimit_init;

        __u64 totalspeed;
        __u64 bufferlimit_speed;
        __u64 connlimit_speed;

        mutex_lock(&(rconn->rcv_lock));

        BUG_ON(rconn->sourcetype != SOURCE_IN);

        if (atomic_read(&(rconn->isreset)) != 0) {
                if (listlocked && (rconn->source.in.buffer_list.next != 0 ||
                                rconn->source.in.buffer_list.prev != 0)) {
                        list_del(&(rconn->source.in.buffer_list));
                        rconn->source.in.buffer_list.next = 0;
                        rconn->source.in.buffer_list.prev = 0;
                        kref_put(&(rconn->ref), free_conn);
                }
                goto out;
        }

        if (listlocked){
                if (rconn->source.in.buffer_list.next != 0 ||
                                rconn->source.in.buffer_list.prev != 0) {
                        list_del(&(rconn->source.in.buffer_list));
                } else {
                        kref_get(&(rconn->ref));
                }
                list_add_tail(&(rconn->source.in.buffer_list),
                                &buffer_conn_list);
        } else  if (rconn->source.in.buffer_list.next == 0 &&
                        rconn->source.in.buffer_list.prev == 0) {
                kref_get(&(rconn->ref));
                list_add_tail(&(rconn->source.in.buffer_list),
                                &buffer_conn_tmp_list);
        }


        conns = atomic_read(&num_conns);
        BUG_ON(conns < 0);
        bufferlimit_init = desired_bufferusage(bufferassigned_init,
                        bufferusage_init, BUFFERASSIGN_INIT, BUFFERSPACE_INIT);
        connlimit_init = (bufferlimit_init + conns - 1) / conns;

        bufferassigned_init -= rconn->source.in.buffer_init;
        if (((__u32) connlimit_init) != connlimit_init)
                rconn->source.in.buffer_init = -1;
        else
                rconn->source.in.buffer_init = (__u32) connlimit_init;
        bufferassigned_init += rconn->source.in.buffer_init;

        spin_lock_irqsave(&st_lock, iflags);
        jiffies_tmp = jiffies;
        totalspeed = get_speed(&st, jiffies_tmp);
        bufferlimit_speed = desired_bufferusage(bufferassigned_speed,
                        bufferusage_speed, BUFFERASSIGN_SPEED,
                        BUFFERSPACE_SPEED);
        connlimit_speed = multiply_div(bufferlimit_speed,
                        get_speed(&(rconn->source.in.st), jiffies_tmp),
                        totalspeed);
        spin_unlock_irqrestore(&st_lock, iflags);

        bufferassigned_speed -= rconn->source.in.buffer_speed;
        if (((__u32) connlimit_speed) != connlimit_speed)
                rconn->source.in.buffer_speed = -1;
        else
                rconn->source.in.buffer_speed = (__u32) connlimit_speed;
        bufferassigned_speed += rconn->source.in.buffer_speed;

        refresh_bufferusage(rconn);

        window = __get_window(rconn);

        rconn->source.in.window_seqnolimit_last = rconn->source.in.next_seqno +
                        dec_window(window);
        if (((__s32) (rconn->source.in.window_seqnolimit_last -
                        rconn->source.in.window_seqnolimit_max)) > 0)
                rconn->source.in.window_seqnolimit_max =
                                rconn->source.in.window_seqnolimit_last;

out:
        mutex_unlock(&(rconn->rcv_lock));

        return window;
}

/* do not hold rcv_lock while calling this */
__u8 get_window(struct conn *rconn)
{
        struct conn *rconn2;
        int listlocked;

        __u8 window;

        mutex_lock(&bufferlimits_lock);
        listlocked = mutex_trylock(&buffer_conn_list_lock);

        window = _get_window(rconn, listlocked);

        if (listlocked){
                /**
                 * refresh window of idle conns as well to keep global counters
                 * accurate
                 */

                rconn2 = container_of(buffer_conn_list.next, struct conn,
                                source.in.buffer_list);

                if (list_empty(&buffer_conn_list) == 0 && rconn2 != rconn)
                        _get_window(rconn2, listlocked);


                if (list_empty(&buffer_conn_tmp_list) == 0) {
                        rconn2 = container_of(buffer_conn_tmp_list.next,
                                        struct conn, source.in.buffer_list);
                        BUG_ON(rconn2 == rconn);
                        _get_window(rconn2, listlocked);
                }

                mutex_unlock(&buffer_conn_list_lock);
        }

        mutex_unlock(&bufferlimits_lock);

        return window;
}

void reset_bufferusage(struct conn *conn)
{
        int listlocked;

        mutex_lock(&bufferlimits_lock);
        listlocked = mutex_trylock(&buffer_conn_list_lock);
        mutex_lock(&(conn->rcv_lock));

        if (conn->sourcetype != SOURCE_IN)
                goto out;

        bufferusage_init -= conn->source.in.usage_init;
        bufferusage_speed -= conn->source.in.usage_speed;
        bufferusage_ata -= conn->source.in.usage_ata;
        bufferusage_reserve -= conn->source.in.usage_reserve;

        bufferassigned_init -= conn->source.in.buffer_init;
        bufferassigned_speed -= conn->source.in.buffer_speed;
        bufferassigned_ata -= conn->source.in.buffer_ata;

        if (listlocked && (conn->source.in.buffer_list.next != 0 ||
                        conn->source.in.buffer_list.prev != 0)) {
                list_del(&(conn->source.in.buffer_list));
                conn->source.in.buffer_list.next = 0;
                conn->source.in.buffer_list.prev = 0;
                kref_put(&(conn->ref), free_conn);
        }

out:
        mutex_unlock(&(conn->rcv_lock));
        if (listlocked)
                mutex_unlock(&buffer_conn_list_lock);
        mutex_unlock(&bufferlimits_lock);
}

void refresh_speedstat(struct conn *rconn, __u32 written)
{
        unsigned long iflags;
        unsigned long jiffies_tmp;

        spin_lock_irqsave(&st_lock, iflags);

        jiffies_tmp = jiffies;

        if (rconn->source.in.st.jiffies_last_update != jiffies_tmp)
                get_speed(&(rconn->source.in.st), jiffies_tmp);
        if (rconn->source.in.st.bytes_curr + written < written)
                rconn->source.in.st.bytes_curr = -1;
        rconn->source.in.st.bytes_curr += written;

        if (st.jiffies_last_update != jiffies_tmp)
                get_speed(&st, jiffies_tmp);
        if (st.bytes_curr + written < written)
                st.bytes_curr = -1;
        st.bytes_curr += written;

        spin_unlock_irqrestore(&st_lock, iflags);
}

void drain_ooo_queue(struct conn *rconn)
{
        struct sk_buff *skb;

        BUG_ON(SOURCE_IN != rconn->sourcetype);

        skb = rconn->source.in.reorder_queue.next;

        while ((void *) skb != (void *) &(rconn->source.in.reorder_queue)) {
                struct skb_procstate *ps = skb_pstate(skb);
                int drop;

                if (rconn->source.in.next_seqno != ps->funcstate.rcv2.seqno)
                        break;

                drop = receive_skb(rconn, skb);
                if (drop)
                        break;

                skb_unlink(skb, &(rconn->source.in.reorder_queue));
                rconn->source.in.ooo_packets--;
                atomic_dec(&(rconn->source.in.nb->ooo_packets));
                atomic_dec(&ooo_packets);

                rconn->source.in.next_seqno += skb->len;
        }
}

static int _conn_rcv_ooo(struct conn *rconn, struct sk_buff *skb)
{
        struct skb_procstate *ps = skb_pstate(skb);
        struct sk_buff_head *reorder_queue = &(rconn->source.in.reorder_queue);
        struct sk_buff *curr = reorder_queue->next;

        long ooo;

        #warning todo limit amount of data, not packet count
        rconn->source.in.ooo_packets++;
        if (rconn->source.in.ooo_packets > MAX_TOTAL_OOO_PER_CONN)
                goto drop_ooo3;

        ooo = atomic_inc_return(&(rconn->source.in.nb->ooo_packets));
        if (ooo > MAX_TOTAL_OOO_PER_NEIGH)
                goto drop_ooo2;

        ooo = atomic_inc_return(&ooo_packets);
        if (ooo > MAX_TOTAL_OOO_PACKETS)
                goto drop_ooo1;


        while (1) {
                struct skb_procstate *ps2 = skb_pstate(curr);

                if ((void *) curr == (void *) reorder_queue) {
                        skb_queue_tail(reorder_queue, skb);
                        break;
                }

                if (ps->funcstate.rcv2.seqno > ps2->funcstate.rcv2.seqno) {
                        skb_insert(curr, skb, reorder_queue);
                        break;
                }

                curr = curr->next;
        }

        if (0) {
drop_ooo1:
                atomic_dec(&ooo_packets);
drop_ooo2:
                atomic_dec(&(rconn->source.in.nb->ooo_packets));
drop_ooo3:
                rconn->source.in.ooo_packets--;

                return 1;
        }

        return 0;
}

#warning todo check if in window
static void _conn_rcv(struct conn *rconn, struct sk_buff *skb)
{
        struct skb_procstate *ps = skb_pstate(skb);
        struct control_msg_out *cm = alloc_control_msg(rconn->source.in.nb,
                        ACM_PRIORITY_MED);

        int in_order;
        int drop = 1;
        int flush = 0;

        __u32 len = skb->len;

        BUG_ON(rconn->sourcetype != SOURCE_IN);

        if (unlikely(cm == 0)) {
                kfree_skb(skb);
                return;
        }

        mutex_lock(&(rconn->rcv_lock));

        in_order = (rconn->source.in.next_seqno == ps->funcstate.rcv2.seqno);

        if (in_order == 0) {
                drop = _conn_rcv_ooo(rconn, skb);
        } else {
                rconn->source.in.next_seqno += skb->len;
                drop = receive_skb(rconn, skb);
        }

        if (drop) {
                kfree_skb(skb);
                free_control_msg(cm);
                goto out;
        }

        if (in_order == 0) {
                flush = 1;
                send_ack_conn_ooo(cm, rconn,
                                rconn->reversedir->target.out.conn_id,
                                rconn->source.in.next_seqno,
                                ps->funcstate.rcv2.seqno, len);
        } else {
                drain_ooo_queue(rconn);
                send_ack_conn(cm, rconn, rconn->reversedir->target.out.conn_id,
                                rconn->source.in.next_seqno);
        }

out:
        mutex_unlock(&(rconn->rcv_lock));

        if (drop == 0)
                flush_buf(rconn);
}

static void conn_rcv(struct sk_buff *skb, __u32 conn_id, __u32 seqno)
{
        struct conn *rconn;
        struct skb_procstate *ps = skb_pstate(skb);

        ps->funcstate.rcv2.seqno = seqno;

        rconn = get_conn(conn_id);

        if (unlikely(rconn == 0)) {
                printk(KERN_DEBUG "unknown conn_id when receiving: %d",
                                conn_id);
                kfree_skb(skb);
                return;
        }
        _conn_rcv(rconn, skb);
        kref_put(&(rconn->ref), free_conn);
}

void conn_rcv_buildskb(char *data, __u32 datalen, __u32 conn_id, __u32 seqno)
{
        struct sk_buff *skb = alloc_skb(datalen, GFP_KERNEL);
        char *dst = skb_put(skb, datalen);
        memcpy(dst, data, datalen);
        conn_rcv(skb, conn_id, seqno);
}

static void rcv_data(struct sk_buff *skb)
{
        __u32 conn_id;
        __u32 seqno;

        char *connid_p = cor_pull_skb(skb, 4);
        char *seqno_p = cor_pull_skb(skb, 4);

        /* __u8 rand; */

        ((char *)&conn_id)[0] = connid_p[0];
        ((char *)&conn_id)[1] = connid_p[1];
        ((char *)&conn_id)[2] = connid_p[2];
        ((char *)&conn_id)[3] = connid_p[3];

        ((char *)&seqno)[0] = seqno_p[0];
        ((char *)&seqno)[1] = seqno_p[1];
        ((char *)&seqno)[2] = seqno_p[2];
        ((char *)&seqno)[3] = seqno_p[3];

        conn_id = be32_to_cpu(conn_id);
        seqno = be32_to_cpu(seqno);

        /* get_random_bytes(&rand, 1);

        if (rand < 64) {
                printk(KERN_ERR "drop %d %d %d %d %d", conn_id, seqno_p[0],
                                seqno_p[1], seqno_p[2], seqno_p[3]);
                goto drop;
        } */

        if (conn_id == 0) {
                struct neighbor *nb = get_neigh_by_mac(skb);
                if (unlikely(nb == 0))
                        goto drop;
                kernel_packet(nb, skb, seqno);
                kref_put(&(nb->ref), neighbor_free);
        } else {
                conn_rcv(skb, conn_id, seqno);
        }

        if (0) {
drop:
                kfree_skb(skb);
        }
}

static void rcv(struct work_struct *work)
{
        struct sk_buff *skb = skb_from_pstate(container_of(work,
                        struct skb_procstate, funcstate.rcv.work));

        __u8 packet_type;
        char *packet_type_p;

        atomic_dec(&packets_in_workqueue);

        packet_type_p = cor_pull_skb(skb, 1);

        if (unlikely(packet_type_p == 0))
                goto drop;

        packet_type = *packet_type_p;

        if (packet_type == PACKET_TYPE_ANNOUNCE) {
                rcv_announce(skb);
                return;
        }

        if (unlikely(packet_type != PACKET_TYPE_DATA))
                goto drop;

        rcv_data(skb);

        if (0) {
drop:
                kfree_skb(skb);
        }
}

static int queue_rcv_processing(struct sk_buff *skb, struct net_device *dev,
               struct packet_type *pt, struct net_device *orig_dev)
{
        struct skb_procstate *ps = skb_pstate(skb);
        long queuelen;

        if (skb->pkt_type == PACKET_OTHERHOST)
                goto drop;

        BUG_ON(skb->next != 0);

        queuelen = atomic_inc_return(&packets_in_workqueue);

        BUG_ON(queuelen <= 0);

        #warning todo limit per interface, inbound credits
        if (queuelen > MAX_PACKETS_IN_RCVQUEUE) {
                atomic_dec(&packets_in_workqueue);
                goto drop;
        }

        INIT_WORK(&(ps->funcstate.rcv.work), rcv);
        queue_work(packet_wq, &(ps->funcstate.rcv.work));
        return NET_RX_SUCCESS;

drop:
        kfree_skb(skb);
        return NET_RX_DROP;
}

static struct packet_type ptype_cor = {
        .type = htons(ETH_P_COR),
        .dev = 0,
        .func = queue_rcv_processing
};

int __init cor_rcv_init(void)
{
        bufferassigned_init = 0;
        bufferassigned_speed = 0;
        bufferassigned_ata = 0;

        bufferusage_init = 0;
        bufferusage_speed = 0;
        bufferusage_ata = 0;
        bufferusage_reserve = 0;

        memset(&st, 0, sizeof(struct speedtracker));

        BUG_ON(sizeof(struct skb_procstate) > 48);
        packet_wq = create_workqueue("cor_packet");
        INIT_WORK(&outofbufferspace_work, outofbufferspace);
        outofbufferspace_scheduled = 0;

        dev_add_pack(&ptype_cor);
        return 0;
}

MODULE_LICENSE("GPL");