shrink window size field

[cor.git] / net / cor / cor.h

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

#include <linux/atomic.h>

#include <linux/module.h>
#include <linux/types.h>
#include <linux/interrupt.h>
#include <linux/sched.h>
#include <linux/netdevice.h>
#include <linux/skbuff.h>
#include <linux/spinlock.h>
#include <linux/workqueue.h>
#include <linux/kref.h>
#include <linux/ktime.h>
#include <linux/rbtree.h>

#include <linux/socket.h>
#include <net/sock.h>

#include <linux/math64.h>

#include "settings.h"


#define ETH_P_COR 0x1022
#define AF_COR 99
#define PF_COR AF_COR

#define PROTO_COR_RAW 0
#define PROTO_COR_RDEAMON 1

struct cor_sockaddr {
        __u16 sin_family;

        __be32 port;
        __be64 addr;
};

#define COR_PASS_ON_CLOSE 1

#define COR_PUBLISH_SERVICE 2

#define COR_TOS 3
#define COR_TOS_DEFAULT 0
#define COR_TOS_LOW_LATENCY 1
#define COR_TOS_HIGH_LATENCY 2

#define COR_PRIORITY 4

#define MAX_CONN_CMD_LEN 64

#define PACKET_TYPE_NONE 0
#define PACKET_TYPE_ANNOUNCE 1
#define PACKET_TYPE_CMSG_NOACK 2
#define PACKET_TYPE_CMSG_ACKSLOW 3
#define PACKET_TYPE_CMSG_ACKFAST 4
#define PACKET_TYPE_CONNDATA 64

#define PACKET_TYPE_CONNDATA_FLAGS 63
#define PACKET_TYPE_CONNDATA_FLAGS_FLUSH 32
#define PACKET_TYPE_CONNDATA_FLAGS_WINDOWUSED 31


/**
 * Announce data format:
 * version [2]
 *   is 0, may be increased if the protocol changes
 * min_version [2]
 *   is 0, must be increased if a future version of the protocol is incompatible
 *   to the current version
 * [data]
 *
 * Data format of the announce packet "data" field:
 *{command [2] commandlength [2] commanddata [commandlength]}[...]
 */

/* Commands */

/* ANNCMD_VERSION: version[2] minversion[2] */
#define ANNCMD_VERSION 1

/* ANNCMD_ADDR: addr[8] */
#define ANNCMD_ADDR 2

/* ANNCMD_NOADDR: (no params) */
#define ANNCMD_NOADDR 3


/**
 * Kernel packet data - these commands are sent by the neighbor
 * The end nodes may cause these commands to be sent, but they see them beyond
 * the first hop.
 */
#define KP_MISC 0
#define KP_ACK_CONN 1
#define KP_CONN_DATA 2

/* KP_PADDING[1] */
#define KP_MISC_PADDING 0

/**
 * KP_INIT_SESSION[1] sessionid[4]
 *
 * finishes neighbor discovery and starts a session
 *
 * Before this is received all other commands are ignored. The sessionid is used
 * to prevent usage of old neighbor discovery data (e.g. addresses)
 */
#define KP_MISC_INIT_SESSION 1

#define KP_MISC_INIT_SESSION_CMDLEN 5

/**
 * KP_PING[1] cookie[4]
 * KP_PONG[1] cookie[4] respdelay_full[4] respdelay_netonly[4]
 *
 * This is needed to find out whether the other node is reachable. After a new
 * neighbor is seen, ping requests are sent and the neighbor is only reachable
 * after a few pongs are received. These requests are also used to find out
 * whether a neighber is gone.
 *
 * respdelay:
 * The receiver of a ping may delay the sending of the pong e.g. to create
 * bigger packets. The respdelay is the time in microseconds the packet was
 * delayed.
 */
#define KP_MISC_PING 2

#define KP_MISC_PING_CMDLEN 5

#define KP_MISC_PONG 3

/* KP_ACK[1] seqno[4] */
#define KP_MISC_ACK 4

/**
 * NOTE on connection ids:
 * connection ids we receive with most significant bit 0 have been generated by
 *   us
 * connection ids we receive with most significant bit 1 have been generated by
 *   the other side
 *
 * ATTENTION: the priority seqno are reversed:
 * priority seqnos we send are used when we send updates
 * priority seqnos we received are used when we receive updates
 */

/**
 * incoming connection
 * seqno1... used to ack data sent from the side which initiated the connection
 * seqno2... used to ack data sent to the side which initiated the connection
 * KP_CONNECT[1] conn_id[4] seqno1[4] seqno2[4] window[1] priority_seqno[0.5]
 * priority[1.5] is_highlatency[1]
 */
#define KP_MISC_CONNECT 5

/**
 * incoming connection successful,
 * KP_CONNECT_SUCCESS[1] conn_id[4] window[1]
 */
#define KP_MISC_CONNECT_SUCCESS 6

/**
 * KP_RESET_CONN[1] conn_id[4]
 * We send this, if there is an established connection we want to close.
 */
#define KP_MISC_RESET_CONN 7

/**
 * KP_SET_MAX_CMSG_DELAY[1] cpacket_ack_fast_delay[4] cpacket_ack_slow_delay[4]
 * data_ack_delay[4] cmsg_delay[4]
 * Sent after connecting and at any change
 * delay in specifies in microsecs
 */
#define KP_MISC_SET_MAX_CMSG_DELAY 8

#define KP_MISC_SET_MAX_CMSG_DELAY_CMDLEN 17

/**
 * KP_MISC_SET_RECEIVE_MTU[1] receive_mtu[4]
 * Sent after connecting and at any change
 */
#define KP_MISC_SET_RECEIVE_MTU 9

#define KP_MISC_SET_RECEIVE_MTU_CMDLEN 5


/**
 * KP_ACK_CONN[1] conn_id[4] delay_remaining[1] seqno[4] window[2]
 * bufsize_changerate[1] seqno_ooo[4]
 *   length[1-4] priority_seqno[0.5] priority[1.5] is_highlatency[1]
 *
 * conn_id is the conn_id we use if we sent something through this conn and
 *   *not* the conn_id that the neighbor used to send us the data
 *
 * delay_remaining = time the ack_conn could have remained in the queue
 *   255 means the ack_conn has been sent immediately
 *   0 means it has been delayed by as much the delay set by SET_MAX_CMSG_DELAY
 *
 * seqno = the seqno which is expected in the next non-out-of-order packet
 *
 * window = amount of data which can be sent without receiving the next ack
 *   packets with lower seqno do not overwrite the last window size
 *   The window may also be reduced. However, this only indicates a wish.
 *   Packets must be accepted if they exceed the new window, but not the old
 *   one.
 *
 *   decode:
 *     0 = 0
 *     1...255 = 64*2^((value-1)/7) end result is rounded down to an integer
 *
 * bufsize_changerate = if the next router(s) is increasing or decreasing its
 *   buffer size
 *   0   = for every byte we can send, the end host will receive 2 bytes
 *   64  = for every byte we can send, the end host will receive 1 byte
 *   128 = for every 2 byte we can send, the end host will receive 1 byte
 *   ...
 *
 * seqno_ooo, length = This packet was received out of order. Maybe a previous
 *   packet has been lost. Out of order data should not be retransmitted.
 *   Multiple ooo packets may be merged into a single ack. Ooo packets may be
 *   partially accepted, so that the length does not cover the full packet and/
 *   or the seqno starts in the middle of a packet
 */
#define KP_ACK_CONN_FLAGS_SEQNO 1
#define KP_ACK_CONN_FLAGS_WINDOW 2
#define KP_ACK_CONN_FLAGS_OOO 12 /* 4+8 */
#define KP_ACK_CONN_FLAGS_PRIORITY 16

static inline __u8 cor_ooolen_to_flags(__u32 len)
{
        if (len == 0)
                return 0;
        if (len < 256)
                return 4;
        if (len < 65536)
                return 8;
        return 12;
}

static inline int cor_ooolen(__u8 flags)
{
        int len = ((flags & KP_ACK_CONN_FLAGS_OOO) >> 2);

        if (unlikely(len == 3))
                return 4;
        return len;
}

static inline int cor_ack_conn_len(__u8 flags)
{
        int len = 0;

        if ((flags & KP_ACK_CONN_FLAGS_SEQNO) != 0) {
                len += 4;
                if ((flags & KP_ACK_CONN_FLAGS_WINDOW) != 0)
                        len += 2;
        }

        if (cor_ooolen(flags) != 0) {
                len += 4;
                len += cor_ooolen(flags);
        }

        /* delay_remaining */
        if ((flags & KP_ACK_CONN_FLAGS_SEQNO) != 0 ||
                        cor_ooolen(flags) != 0)
                len++;

        if (flags & KP_ACK_CONN_FLAGS_PRIORITY)
                len += 3;

        return len;
}

/* KP_CONN_DATA[1] conn_id[4] seqno[4] length[1-2] data[length] */
#define KP_CONN_DATA_FLAGS_WINDOWUSED 31
#define KP_CONN_DATA_FLAGS_FLUSH 32

#define KP_CONN_DATA_MAXLEN (128 + 32767)

static inline __u32 get_kp_conn_data_length(__u32 datalen)
{
        if (datalen < 128)
                return 10 + datalen;
        else
                return 11 + datalen;
}

static inline __u8 get_kp_code(__u8 maj, __u8 min)
{
        BUILD_BUG_ON(maj > 3);
        BUG_ON(min > 63);
        return (maj << 6) + min;
}

static inline __u8 kp_maj(__u8 code)
{
        return code >> 6;
}

static inline __u8 kp_min(__u8 code)
{
        return code & 63;
}


/**
 * Connection data which in interpreted when connection has no target yet
 * These commands are sent by the end node.
 *
 * Format:
 * cmd[2] length[1-4] parameter[length]
 * unrecogniced commands are ignored
 * parameters which are longer than expected are ignored as well
 */

#define CD_CONTINUE_ON_ERROR_FLAG 32768
#define CD_NOPARAM_FLAG 16384

/* outgoing connection: CD_CONNECT_NB[2] length[1-4] addr[8] */
#define CD_CONNECT_NB 1

/* connection to local open part: CD_CONNECT_PORT[2] length[1-4] port[4] */
#define CD_CONNECT_PORT 2

/**
 * list connected neighbors: CD_LIST_NEIGH[2] length[1-4]
 * responds with CDR_BINDATA if successful
 * The response
 * format is:
 *
 * numneighs[1-4]
 * numfields[1-4] (field[2] fieldlen[1-4])[numfields]
 * rows[responserows]:
 *   fieldlen[1-4], only if fieldlen in the header was "0"
 *   fielddata[fieldlen]
 *
 * Future versions may append data to field definition. Clients must silently
 * discard fields they do not expect.
 */
#define CD_LIST_NEIGH 3

/**
 * addr[fieldlen]
 */
#define LIST_NEIGH_FIELD_ADDR 1

/**
 * latency_in_microsecs[1] (64_11 encoding)
 * Only raw network latency in measured. Delays caused by the priority queues
 * are *not* included.
 */
#define LIST_NEIGH_FIELD_LATENCY 2

/**
 * list services: CD_LIST_SERVICES[2] length[1-4]
 * responds with CDR_BINDATA if successful
 */
#define CD_LIST_SERVICES 4

/**
 * list services: CD_LIST_SERVICES[2] length[1-4]
 * responds with CDR_BINDATA if successful
 */
#define CD_LIST_L4PROTOCOLS 5


/**
 * Connection data response
 * Format is the same as with connection data
 */

/**
 * CDR_EXECOK[1]
 */
#define CDR_EXECOK 1

/**
 * CDR_EXEOK_BINDATA[1] bindatalen[1-4] bindata[bindatalen]
 * CDR_EXECOK_BINDATA_NORESP[1]
 */
#define CDR_EXECOK_BINDATA 2
#define CDR_EXECOK_BINDATA_NORESP 3

/**
 * CDR_EXECFAILED[1] reasoncode[2]
 */
#define CDR_EXECFAILED 4
        #define CDR_EXECFAILED_INVALID_COMMAND 1
        #define CDR_EXECFAILED_COMMAND_PARSE_ERROR 2
        #define CDR_EXECFAILED_TEMPORARILY_OUT_OF_RESOURCES 3
        #define CDR_EXECFAILED_NB_DOESNTEXIST 4
        #define CDR_EXECFAILED_UNKNOWN_L4PROTOCOL 5
        #define CDR_EXECFAILED_PORTCLOSED 6

#define L4PROTO_STREAM 42399


/**
 * routing daemon sock
 * format:
 * cmdcode[4] length[4] cmddata[length]
 */
#define CRD_KTU_SUPPORTEDVERSIONS 1
/**
 * CRD_KTU_SUPPORTEDVERSIONS[4] length[4] min[4] max[4]
 */

#define CRD_KTU_CONNECT 2
/**
 * CRD_KTU_KTOU_CONNECT[4] length[4] cookie[8] targetlen[4] target[targetlen]
 */

#define CRD_UTK_VERSION 1
/**
 * CRD_UTK_VERSION[4] length[4] version[4]
 */

#define CRD_UTK_UP 2
#define CRD_UTK_UP_FLAGS_ADDR 1
#define CRD_UTK_UP_FLAGS_INTERFACES 2
/**
 * CRD_UTK_UP[4] length[4] flags[8]
 * if CRD_UTK_UP_FLAGS_ADDR
 *   addr[8]
 * if CRD_UTK_UP_FLAGS_INTERFACES:
 *   num_interfaces[4] (length[4] interface[length])[num_interfaces]
 */


#define CRD_UTK_CONNECTERROR 3
/**
 * CRD_UTK_CONNECTERROR[4] length[4] cookie[8] error[4]
 */

#define CRD_UTK_CONNECTERROR_ACCES 1
#define CRD_UTK_CONNECTERROR_NETUNREACH 2
#define CRD_UTK_CONNECTERROR_TIMEDOUT 3
#define CRD_UTK_CONNECTERROR_REFUSED 4

#define CONN_MNGD_HEADERLEN 2
#define CONN_MNGD_MAX_CTRL_DATALEN 8
#define CONN_MNGD_CHECKSUMLEN 4

#define CONN_MNGD_HASDATA (1 << 15)
#define CONN_MNGD_EOF (1 << 0)
#define CONN_MNGD_RCVEND (1 << 1)
#define CONN_MNGD_KEEPALIVE_REQ (1 << 2)
#define CONN_MNGD_KEEPALIVE_RESP (1 << 3)
#define CONN_MNGD_DATALEN 4095

#define CONN_MNGD_MAX_SEGMENT_SIZE (CONN_MNGD_DATALEN + 1)


struct cor_interface_config {
        char *name;
        __u32 name_len;
};

#define CONGSTATUS_NONE 0
#define CONGSTATUS_CONNDATA 1
#define CONGSTATUS_ANNOUNCE 2
#define CONGSTATUS_RETRANS 3
#define CONGSTATUS_KPACKETS 4

struct cor_qos_queue {
        spinlock_t qlock;

        struct kref ref;

        struct list_head queue_list;

        struct net_device *dev; /* may not change while queue is in list */

        struct task_struct *qos_resume_thread;
        wait_queue_head_t qos_resume_wq;
        atomic_t qos_resume_scheduled;
        unsigned long jiffies_lastprogress;

        struct list_head kpackets_waiting;
        struct list_head conn_retrans_waiting;
        struct list_head announce_waiting;
        struct list_head neighbors_waiting;
        struct list_head neighbors_waiting_nextpass;

        unsigned long jiffies_nb_pass_start;
        unsigned long jiffies_nb_lastduration;


        unsigned long jiffies_lastdrop;

        __u32 numconns;
        __u64 priority_sum;

        atomic_t cong_status;
};


/**
 * switch to and from RB_INQUEUE_NBCONGWIN is only done with nbcongwin.lock
 * *and* qlock held
 */
#define RB_INQUEUE_FALSE 0
#define RB_INQUEUE_TRUE 1
#define RB_INQUEUE_NBCONGWIN 2 /* only for nb->rb */
#define RB_INQUEUE_NBNOTACTIVE 3 /* only for nb->rb */

struct cor_resume_block {
        struct list_head lh;
        __u8 in_queue;
};

#define ANNOUNCE_TYPE_BROADCAST 1
#define ANNOUNCE_TYPE_UNICAST 2

struct cor_announce_data {
        struct kref ref;

        struct list_head lh;
        __u8 type;
        __u16 sndcnt;
        struct net_device *dev;
        char mac[MAX_ADDR_LEN];
        struct delayed_work announce_work;
        struct cor_resume_block rb;
};

struct cor_neighbor_discdata {
        struct list_head lh;
        unsigned long jiffies_created;

        __be32 sessionid;

        struct net_device *dev;
        char mac[MAX_ADDR_LEN];

        __u8 nb_allocated;

        __u8 rcvd_version;
        __u8 rcvd_addr;

        __u16 version;
        __u16 minversion;

        __u8 has_addr;
        __be64 addr;
};

struct cor_ping_cookie {
        ktime_t time_created;
        ktime_t time_sent;
        unsigned long jiffies_sent;

        __u32 cookie;
        __u8 pongs; /* count of pongs for pings sent after this one */
};

#define NEIGHBOR_STATE_INITIAL 0
#define NEIGHBOR_STATE_ACTIVE 1
#define NEIGHBOR_STATE_STALLED 2
#define NEIGHBOR_STATE_KILLED 3

#define NBCONGWIN_SHIFT 16
#define NBCONGWIN_MUL (1 << NBCONGWIN_SHIFT)

struct cor_neighbor {
        struct list_head nb_list;
        __u8 in_nb_list;

        struct kref ref;

        struct net_device *dev;
        char mac[MAX_ADDR_LEN];
        struct cor_qos_queue *queue;

        __be32 sessionid;

        atomic_t sessionid_rcv_needed;
        atomic_t sessionid_snd_needed;

        __u8 has_addr;
        __be64 addr;

        atomic64_t cmsg_timer_timeout;
        struct timer_list cmsg_timer;
        spinlock_t cmsg_lock;
        struct list_head cmsg_queue_pong;
        struct list_head cmsg_queue_ack_fast;
        struct list_head cmsg_queue_ack_slow;
        struct list_head cmsg_queue_ackconn_urgent;
        struct list_head cmsg_queue_ackconn;
        struct list_head cmsg_queue_conndata_lowlat;
        struct list_head cmsg_queue_conndata_highlat;
        struct list_head cmsg_queue_other;
        __u8 add_retrans_needed;
        __u32 kpacket_seqno; /* not locked, only accessed by single tasklet */

        struct rb_root pending_conn_resets_rb;

        __u32 cmsg_pongslength;
        __u32 cmsg_otherlength;

        __u32 cmsg_pongscnt; /* size of queue only, protected by cmsg_lock */
        atomic_t cmsg_pongs_retrans_cnt; /* number of retransmits only */
        atomic_t cmsg_othercnt; /* size of queue + retransmits */

        atomic_t cmsg_bulk_readds;

        atomic_t cmsg_delay_conndata;

        /* not locked, only accessed by single thread */
        __u8 max_cmsg_delay_sent;

        atomic_t rcvmtu_sendneeded;


        /* procected by cor_qos_queue->qlock */
        struct cor_resume_block rb_kp;
        struct cor_resume_block rb_cr;
        struct cor_resume_block rb;
        unsigned long cmsg_send_start_j;
        ktime_t cmsg_send_start_kt;

        struct{
                spinlock_t lock;
                struct list_head lh;
                struct list_head lh_nextpass;
                __u32 cnt;
                __u64 priority_sum;
        } conns_waiting;

        struct{
                spinlock_t lock;
                atomic64_t data_intransit;
                atomic64_t cwin;
                __u64 cwin_shrinkto;
        } nbcongwin;

        spinlock_t state_lock;
        unsigned long last_ping_time;
        struct cor_ping_cookie cookies[PING_COOKIES_PER_NEIGH];
        __u32 ping_intransit;
        __u32 lastcookie;
        __u32 cookie_unsent;
        __u64 latency_variance_retrans_us; /* microsecs */
        atomic_t latency_retrans_us; /* microsecs */
        atomic_t latency_stddev_retrans_us; /* microsecs */
        atomic_t latency_advertised_us; /* microsecs */
        __u8    rcvmtu_delayed_send_needed:1,
                rcvmtu_allowed_countdown:2;

        atomic_t max_remote_ack_fast_delay_us; /* microsecs */
        atomic_t max_remote_ack_slow_delay_us; /* microsecs */
        atomic_t max_remote_ackconn_delay_us; /* microsecs */
        atomic_t max_remote_pong_delay_us; /* microsecs */

        atomic_t remote_rcvmtu;

        union {
                unsigned long initial_state_since;/* initial state */
                /**
                 * last_roundtrip:
                 * time of the last sent packet which has been acked or
                 * otherwise responded to (e.g. pong)
                 */
                unsigned long last_roundtrip;/* active/stalled state */
        } state_time;
        ktime_t last_roundtrip_end;
        __u16 ping_success;
        __u8 state;

        __u8 str_timer_pending;
        struct delayed_work stalltimeout_timer;

        spinlock_t connid_lock;
        struct rb_root connid_rb;

        spinlock_t connid_reuse_lock;
        struct rb_root connid_reuse_rb;
        struct list_head connid_reuse_list;
        __u16 connid_reuse_pingcnt;
        __u8 connid_reuse_oom_countdown;

        atomic64_t priority_sum;

        /**
         * connecions which receive data from/send data to this node
         * used when terminating all connections of a neighbor and terminating
         * inactive connections
         */
        spinlock_t conn_list_lock;
        struct list_head snd_conn_idle_list;
        struct list_head snd_conn_busy_list;

        /**
         * the timer has to be inited when adding the neighbor
         * timer_setup(...);
         * add_timer(struct timer_list * timer);
         */
        spinlock_t retrans_lock;
        struct timer_list retrans_timer;
        struct list_head retrans_fast_list;
        struct list_head retrans_slow_list;
        struct rb_root kp_retransmits_rb;

        spinlock_t retrans_conn_lock;
        struct timer_list retrans_conn_timer;
        struct list_head retrans_conn_lowlatency_list;
        struct list_head retrans_conn_highlatency_list;

        struct work_struct reset_neigh_work;
};

static inline void cor_nb_kref_get(struct cor_neighbor *nb, char *reason)
{
        /* printk(KERN_ERR "cor_nb_kref_get %p %s\n", nb, reason); */
        kref_get(&nb->ref);
}

void cor_neighbor_free(struct kref *ref); /* neigh.c */

static inline void cor_nb_kref_put(struct cor_neighbor *nb, char *reason)
{
        /* printk(KERN_ERR "cor_nb_kref_put %p %s\n", nb, reason); */
        kref_put(&nb->ref, cor_neighbor_free);
}

void cor_kreffree_bug(struct kref *ref); /* util.c */

static inline void cor_nb_kref_put_bug(struct cor_neighbor *nb, char *reason)
{
        /* printk(KERN_ERR "cor_nb_kref_put_bug %p %s\n", nb, reason); */
        kref_put(&nb->ref, cor_kreffree_bug);
}


#define DATABUF_BUF 0
#define DATABUF_SKB 1

struct cor_data_buf_item {
        struct list_head buf_list;

        char *buf;
        __u16 datalen;
        __u16 buflen;

        __u8 type;
};

struct cor_connid_reuse_item {
        struct rb_node rbn;

        struct list_head lh;

        struct kref ref;
        __u32 conn_id;
        __u16 pingcnt;
};

#define SNDSPEED_INIT 0
#define SNDSPEED_ACTIVE 1
struct cor_snd_speed {
        __u8 state;
        __u8 flushed;
        unsigned long jiffies_last_refresh;
        __u32 bytes_sent;

        /* bytes per second */
        __u32 speed;
        __u32 speed_limited;
};

struct cor_sock;

/* This struct helps keep struct cor_conn small. */
struct cor_conn_src_sock_extradata {
        struct cor_sock *cs;

        __u32 priority;

        struct cor_snd_speed snd_speed;

        __be32 keepalive_req_cookie;
        __be32 keepalive_resp_cookie;
        /**
         * keepalive_intransit == 0... last resp received
         * keepalive_intransit == 1... req sent
         */
        unsigned long jiffies_keepalive_lastact;

        struct{
                char snd_hdr[CONN_MNGD_HEADERLEN];
                char snd_data[CONN_MNGD_MAX_CTRL_DATALEN];
                char snd_chksum[CONN_MNGD_CHECKSUMLEN];
                __u8 snd_data_len;
        } buf_ctrl;

        struct{
                char snd_hdr[CONN_MNGD_HEADERLEN];
                char snd_chksum[CONN_MNGD_CHECKSUMLEN];
                char *snd_data;
                __u16 snd_data_len;
        } buf_data;

        __u16 sent;
};

/**
 * There are 2 conn objects per bi-directional connection. They refer to each
 * other with in the reversedir field.
 *
 *
 *
 * Naming:
 *
 * cn: conn we do not know what is inside
 * src_in, trgt_unconn, trgt_out, ...: A conn with the specified source or
 * targettype. In the unlocked case the types are only a guess,  because they
 * might have changed since the last access. After locking the
 * source/destination parameters have to be checked whether they still are what
 * we expect. This includes source/targettype, neighbor, conn_id
 *
 *
 * Naming suffixes:
 * no suffix: unlocked
 *
 * _l: this direction is locked
 *
 * _ll: both directions are locked
 *
 * _lx: this direction is locked, the other direction may be locked
 *
 * _o: unlocked, but source or target is known for sure, because an outside
 * lock is taken; For variables on the heap this means that an outside lock must
 * be taken before accessing the struct which points to the conn can be
 * accessed.
 *
 *
 * Most fields are protected by rcv_lock. Fields which which control
 * source and destination of the data flow require both directions to
 * to be locked and external references to be cleared before the change can
 * happen. This includes fields like sourcetype, targettype, connid,
 * list_heads, ???. In this case the side with is_client == 1 needs to be locked
 * first.
 *
 * Some other fields are locked outside (e.g. at struct neighbor).
 */
#define SOURCE_UNCONNECTED 0
#define SOURCE_IN 1
#define SOURCE_SOCK 2

#define TARGET_UNCONNECTED 0
#define TARGET_OUT 1
#define TARGET_SOCK 2
#define TARGET_DISCARD 3

#define BUFSIZE_NOACTION 0
#define BUFSIZE_DECR 1
#define BUFSIZE_DECR_FAST 2
#define BUFSIZE_INCR 3
#define BUFSIZE_INCR_FAST 4

#define JIFFIES_LAST_IDLE_SHIFT 8
#define BUFSIZE_SHIFT 5

#define SOCKTYPE_RAW 0
#define SOCKTYPE_MANAGED 1

#define RCV_BUF_STATE_OK 0
#define RCV_BUF_STATE_INCOMPLETE 1
#define RCV_BUF_STATE_RESET 2

#define SND_BUF_STATE_INCOMPLETE 0
#define SND_BUF_STATE_FILLED 1


struct cor_conn {
        __u8    sourcetype:4,
                targettype:4;

        __u8    is_client; /* immutable after allocated */

        /**
         * isreset values:
         * 0... connection active
         * 1... connection is about to be reset, target does not need to be
         *      notified
         * 2... connection is reset
         */
        __u8    isreset;

        __u8    flush:1,
                is_highlatency:1,
                is_highlatency_send_needed:1;

        spinlock_t rcv_lock;

        union{
                struct{
                        struct cor_neighbor *nb;

                        struct list_head reorder_queue;
                        __u32 reorder_memused;

                        struct rb_node rbn;
                        __u32 conn_id;
                        __u32 next_seqno;

                        /* number of ack sent, not data seqno */
                        __u32 ack_seqno;

                        __u16 small_ooo_packets;

                        __u32 priority;
                        __u8 priority_seqno;
                        __u8 inorder_ack_needed;

                        __u8 established;

                        __u32 window_seqnolimit;
                        __u32 window_seqnolimit_remote;

                        /* protected by nb->cmsg_lock */
                        struct list_head acks_pending;
                } in;

                struct{
                        struct cor_conn_src_sock_extradata *ed;

                        /**
                         * cl_list and in_cl_list is protected by cor_bindnodes
                         */
                        struct list_head cl_list;
                        __u8 in_cl_list;

                        /**
                         * keepalive_lh and in_keepalive_list is protected by
                         * cor_keepalive_req_lock
                         */
                        struct timer_list keepalive_timer;
                        struct list_head keepalive_lh;

                        __u8 in_keepalive_list;

                        /* protected by flushtoconn_oom_lock */
                        struct list_head flushtoconn_oom_lh;
                        /* protected by conn->rcv_lock */
                        __u8 in_flushtoconn_oom_list;


                        __u8    keepalive_intransit:1,
                                buf_ctrl_filled:1,
                                buf_data_filled:1,
                                send_keepalive_req_needed:1,
                                send_keepalive_resp_needed:1,
                                send_eof_needed:1,
                                send_rcvend_needed:1;

                        __u8 last_windowused;
                        __u8    flush:1,
                                socktype:1;
                } sock;
        } src;

        union{
                struct{
                        __u32 paramlen;
                        __u32 cmdread;
                        __u16 cmd;
                        __u8 paramlen_read;
                        char *cmdparams;
                        char paramlen_buf[4];
                } unconnected;

                struct{
                        struct cor_neighbor *nb;

                        /* list of all connections to this neighbor */
                        struct list_head nb_list;
                        unsigned long jiffies_last_act;
                        __u32 nblist_busy_remaining;

                        __u32 conn_id;
                        __u32 seqno_nextsend;
                        __u32 seqno_acked;
                        __u32 seqno_windowlimit;

                        /* protected by nb->retrans_conn_lock, sorted by seqno
                         */
                        struct list_head retrans_list;

                        struct cor_resume_block rb;

                        /* for bursting */
                        unsigned long jiffies_idle_since;

                        __u32 rb_priority;
                        __u16 maxsend_extra;

                        __u16 burst_bytes;

                        __u8 lastsend_windowused;

                        __u8 remote_bufsize_changerate;

                        __u8    priority_send_allowed:1,
                                established:1,
                                in_nb_busy_list:1;

                        __u16   priority_last:12,
                                priority_seqno;

                        /* protected by nb->retrans_conn_lock */
                        __u16 retrans_lowwindow;
                } out;

                struct{
                        __u8 waiting_for_userspace;
                        unsigned long waiting_for_userspace_since;

                        struct cor_sock *cs;

                        __u8 socktype;

                        __u8 rcv_buf_state;
                        char rcv_hdr[CONN_MNGD_HEADERLEN];
                        char rcv_chksum[CONN_MNGD_CHECKSUMLEN];
                        char *rcv_buf;
                        __u16 rcvd;
                        __u16 rcv_hdr_flags;
                        __u16 rcv_data_len;
                } sock;
        } trgt;

        struct{
                struct list_head items;
                struct cor_data_buf_item *nextread;
                __u32 first_offset;

                __u32 datasize;
                __u32 overhead;
                __u32 read_remaining;

                __u16 next_read_offset;
        } data_buf;

        __u32 bufspace_accounted;

        struct{
                __u32 bufsize; /* 32 ==> 1 byte, see BUFSIZE_SHIFT */
                __u32 ignore_rcv_lowbuf;
                union{
                        struct{
                                __u32 bytesleft;
                        } noact;

                        struct{
                                __u32 size_start;
                        } decr;

                        struct{
                                __u32 size_start;
                                __u32 size_end;
                        } incr;
                } act;

                __u32   state:3,
                        bytes_rcvd:24;
        } bufsize;
};

static inline __u32 cor_get_connid_reverse(__u32 conn_id)
{
        return conn_id ^ (1 << 31);
}

struct cor_conn_bidir {
        struct cor_conn cli;
        struct cor_conn srv;

        struct kref ref;
};

static inline struct cor_conn_bidir *cor_get_conn_bidir(struct cor_conn *cn)
{
        if (cn->is_client)
                return container_of(cn, struct cor_conn_bidir, cli);
        else
                return container_of(cn, struct cor_conn_bidir, srv);
}

static inline struct cor_conn *cor_get_conn_reversedir(struct cor_conn *cn)
{
        if (cn->is_client) {
                struct cor_conn_bidir *cnb = container_of(cn,
                                struct cor_conn_bidir, cli);
                return &cnb->srv;
        } else {
                struct cor_conn_bidir *cnb = container_of(cn,
                                struct cor_conn_bidir, srv);
                return &cnb->cli;
        }
}

static inline void cor_conn_kref_get(struct cor_conn *cn, char *reason)
{
        /* printk(KERN_ERR "cor_conn_kref_get %p %s\n", cn, reason); */
        kref_get(&cor_get_conn_bidir(cn)->ref);
}

void cor_free_conn(struct kref *ref); /* conn.c */

static inline void cor_conn_kref_put(struct cor_conn *cn, char *reason)
{
        /* printk(KERN_ERR "cor_conn_kref_put %p %s\n", cn, reason); */
        kref_put(&cor_get_conn_bidir(cn)->ref, cor_free_conn);
}

static inline void cor_conn_kref_put_bug(struct cor_conn *cn, char *reason)
{
        /* printk(KERN_ERR "cor_conn_kref_put_bug %p %s\n", cn, reason); */
        kref_put(&cor_get_conn_bidir(cn)->ref, cor_kreffree_bug);
}



#define CONN_RETRANS_INITIAL 0
#define CONN_RETRANS_SCHEDULED 1
#define CONN_RETRANS_LOWWINDOW 2
#define CONN_RETRANS_SENDING 3
#define CONN_RETRANS_ACKED 4
struct cor_conn_retrans {
        /* timeout_list and conn_list share a single ref */
        struct kref ref;
        /* only in timeout_list if state == CONN_RETRANS_SCHEDULED */
        struct list_head timeout_list;
        struct list_head conn_list;
        struct cor_conn *trgt_out_o;
        __u32 seqno;
        __u32 length;

        __u8 windowused;
        __u8 state;
        unsigned long timeout;
};

#define RCVOOO_BUF 0
#define RCVOOO_SKB 1
struct cor_rcvooo {
        struct list_head lh;
        __u32 seqno;
        __u8 type;
        __u8 windowused;
        __u8 flush;
};

struct cor_rcvooo_buf {
        struct cor_rcvooo r;
        char *data;
        __u32 len;
};

/* inside skb->cb */
struct cor_skb_procstate {
        union{
                struct{
                        struct work_struct work;
                } announce1;

                struct{
                        __u32 offset;
                } announce2;

                struct{
                        __u32 skb_memused;
                        struct cor_rcvooo r;
                } rcv_ooo;

                struct{
                        struct cor_data_buf_item dbi;
                } rcv;
        } funcstate;
};

#define CS_TYPE_UNCONNECTED 0
#define CS_TYPE_LISTENER 1
#define CS_TYPE_CONN_RAW 2
#define CS_TYPE_CONN_MANAGED 3

#define CS_CONNECTSTATE_UNCONNECTED 0
#define CS_CONNECTSTATE_CONNECTING 1
#define CS_CONNECTSTATE_CONNECTED 2
#define CS_CONNECTSTATE_ERROR 3

struct cor_sock {
        struct sock sk; /* must be first */

        struct mutex lock;
        struct kref ref;

        /* type may not change once it is set to != CS_TYPE_UNCONNECTED */
        __u8 type;
        __u8 isreleased;

        __u8 publish_service;

        __u8 is_highlatency;
        __u8 is_client;
        __u32 priority;

        union {
                struct {
                        /* listener is protected by cor_bindnodes */
                        struct list_head lh;
                        __be32 port;
                        __u8 publish_service;
                        __u32 queue_maxlen;
                        __u32 queue_len;
                        struct list_head conn_queue;
                } listener;

                struct {
                        struct cor_conn *src_sock;
                        struct cor_conn *trgt_sock;

                        struct cor_data_buf_item *rcvitem;
                        __u16 rcvoffset;

                        struct cor_sock *pass_on_close;
                } conn_raw;

                struct {
                        struct cor_sockaddr remoteaddr;

                        struct list_head rd_msgs;
                        struct list_head crd_lh;
                        __u8 in_crd_list;

                        __u8 connect_state;

                        __u8 is_reset;

                        __u8 flush;

                        __u8    shutdown_rd:1,
                                shutdown_wr:1,
                                sent_eof:1,
                                sent_rcvend:1,
                                rcvd_eof:1,
                                rcvd_rcvend:1;

                        __be64 cookie;
                        struct rb_node rbn;

                        struct cor_conn *src_sock;
                        struct cor_conn *trgt_sock;

                        /* sending */
                        char *snd_buf;
                        __u16 snd_segment_size;
                        __u16 snd_data_len;
                        __u8 send_in_progress;

                        /* receiving */
                        char *rcv_buf;
                        __u16 rcv_data_len;
                        __u16 rcvbuf_consumed;
                        __u8 rcv_buf_state;
                } conn_managed;
        } data;

        struct work_struct readfromconn_work;
        atomic_t readfromconn_work_scheduled;

        atomic_t ready_to_read;
        atomic_t ready_to_write;
        atomic_t ready_to_accept;
};

#define ACK_NEEDED_NO 0
#define ACK_NEEDED_SLOW 1
#define ACK_NEEDED_FAST 2

/* config.c */
extern spinlock_t cor_local_addr_lock;
extern __u8 cor_local_has_addr;
extern __be64 cor_local_addr;
extern __be32 cor_local_addr_sessionid;

int cor_is_device_configurated(struct net_device *dev);

void cor_set_interface_config(struct cor_interface_config *new_config,
                __u32 new_num_interfaces, int new_all_interfaces);

void cor_config_down(void);

int cor_config_up(__u8 has_addr, __be64 addr);

int cor_is_clientmode(void);

/* dev.c */
void cor_qos_set_lastdrop(struct cor_qos_queue *q);

#ifdef DEBUG_QOS_SLOWSEND
int _cor_dev_queue_xmit(struct sk_buff *skb, int caller);
#else
static inline int _cor_dev_queue_xmit(struct sk_buff *skb, int caller)
{
        return dev_queue_xmit(skb);
}
#endif

static inline int cor_dev_queue_xmit(struct sk_buff *skb,
                struct cor_qos_queue *q, int caller)
{
        int rc = _cor_dev_queue_xmit(skb, caller);

        if (unlikely(rc != NET_XMIT_SUCCESS))
                cor_qos_set_lastdrop(q);
        return rc;
}

void cor_free_qos(struct kref *ref);

#ifdef COR_NBCONGWIN
void cor_nbcongwin_data_retransmitted(struct cor_neighbor *nb,
                __u64 bytes_sent);

void cor_nbcongwin_data_acked(struct cor_neighbor *nb, __u64 bytes_acked);

void cor_nbcongwin_data_sent(struct cor_neighbor *nb, __u32 bytes_sent);

int cor_nbcongwin_send_allowed(struct cor_neighbor *nb);

#else

static inline void cor_nbcongwin_data_retransmitted(struct cor_neighbor *nb,
                __u64 bytes_sent)
{
}

static inline void cor_nbcongwin_data_acked(struct cor_neighbor *nb,
                __u64 bytes_acked)
{
}

static inline void cor_nbcongwin_data_sent(struct cor_neighbor *nb,
                __u32 bytes_sent)
{
}

static inline int cor_nbcongwin_send_allowed(struct cor_neighbor *nb)
{
        return 1;
}
#endif

unsigned long cor_get_conn_idletime(struct cor_conn *trgt_out_lx);

struct cor_qos_queue *cor_get_queue(struct net_device *dev);

int cor_destroy_queue(struct net_device *dev);

int cor_create_queue(struct net_device *dev);

#define QOS_RESUME_DONE 0
#define QOS_RESUME_CONG 1
#define QOS_RESUME_NEXTNEIGHBOR 2 /* cor_resume_neighbors() internal */
#define QOS_RESUME_EXIT 3

#define QOS_CALLER_KPACKET 0
#define QOS_CALLER_CONN_RETRANS 1
#define QOS_CALLER_ANNOUNCE 2
#define QOS_CALLER_NEIGHBOR 3

static inline void cor_schedule_qos_resume(struct cor_qos_queue *q)
{
        if (atomic_cmpxchg(&q->qos_resume_scheduled, 0, 1) == 0) {
                barrier();
                wake_up(&q->qos_resume_wq);
        }
}

void cor_qos_enqueue(struct cor_qos_queue *q, struct cor_resume_block *rb,
                unsigned long cmsg_send_start_j, ktime_t cmsg_send_start_kt,
                int caller, int from_nbnotactive_resume);

void cor_qos_remove_conn(struct cor_conn *trgt_out_l);

int cor_may_send_announce(struct net_device *dev);

struct sk_buff *cor_create_packet_cmsg(struct cor_neighbor *nb, int size,
                gfp_t alloc_flags, __u32 seqno);

struct sk_buff *cor_create_packet(struct cor_neighbor *nb, int size,
                gfp_t alloc_flags);

struct sk_buff *cor_create_packet_conndata(struct cor_neighbor *nb, int size,
                gfp_t alloc_flags, __u32 conn_id, __u32 seqno, __u8 windowused,
                __u8 flush);

void cor_qos_enqueue_conn(struct cor_conn *trgt_out_lx);

void cor_dev_down(void);

int cor_dev_up(void);

void __exit cor_dev_exit1(void);

int __init cor_dev_init(void);

/* util.c */
static inline __u16 cor_enc_priority(__u32 value)
{
        __u16 exponent = 0;
        __u16 mantissa;
        __u32 ret;

        while ((value >> exponent) > 255) {
                exponent++;
        }
        BUG_ON(exponent > 15);

        mantissa = (value >> exponent);
        ret = (mantissa << 4) | exponent;
        BUG_ON(ret > 4095);
        return (__u16) ret;
}

static inline __u32 cor_dec_priority(__u16 priority)
{
        __u32 mantissa = (__u32) (priority >> 4);
        __u16 exponent = (priority & 15);

        BUG_ON(priority > 4095);
        return (mantissa << exponent);
}

static inline __u32 cor_priority_max(void)
{
        return cor_dec_priority(4095);
}

__u8 __attribute__((const)) cor_enc_log_64_11(__u32 value);

__u32 __attribute__((const)) cor_dec_log_64_11(__u8 value);

void cor_swap_list_items(struct list_head *lh1, struct list_head *lh2);

int __init cor_util_init(void);



/* neigh.c */
extern atomic_t cor_num_neighs;

int cor_is_from_nb(struct sk_buff *skb, struct cor_neighbor *nb);

struct cor_neighbor *_cor_get_neigh_by_mac(struct net_device *dev,
                char *source_hw);

struct cor_neighbor *cor_get_neigh_by_mac(struct sk_buff *skb);

struct cor_neighbor *cor_find_neigh(__be64 addr);

void cor_resend_rcvmtu(struct net_device *dev);

__u32 cor_generate_neigh_list(char *buf, __u32 buflen);

void cor_reset_neighbors(struct net_device *dev);

int cor_get_neigh_state(struct cor_neighbor *nb);

void cor_ping_resp(struct cor_neighbor *nb, __u32 cookie, __u32 respdelay);

__u32 cor_add_ping_req(struct cor_neighbor *nb, unsigned long *last_ping_time);

void cor_ping_sent(struct cor_neighbor *nb, __u32 cookie);

void cor_unadd_ping_req(struct cor_neighbor *nb, __u32 cookie,
                unsigned long last_ping_time, int congested);

#define TIMETOSENDPING_NO 0
#define TIMETOSENDPING_YES 1
#define TIMETOSENDPING_FORCE 2
int cor_time_to_send_ping(struct cor_neighbor *nb);

unsigned long cor_get_next_ping_time(struct cor_neighbor *nb);

void cor_add_neighbor(struct cor_neighbor_discdata *nb_dd);

struct cor_conn *cor_get_conn(struct cor_neighbor *nb, __u32 conn_id);

int cor_insert_connid(struct cor_neighbor *nb, struct cor_conn *src_in_ll);

void cor_insert_connid_reuse(struct cor_neighbor *nb, __u32 conn_id);

int cor_connid_alloc(struct cor_neighbor *nb, struct cor_conn *src_in_ll);

int __init cor_neighbor_init(void);

void __exit cor_neighbor_exit2(void);

/* neigh_ann_rcv.c */
int cor_rcv_announce(struct sk_buff *skb);

int __init cor_neigh_ann_rcv_init(void);

void __exit cor_neigh_ann_rcv_exit2(void);

/* neigh_ann_snd.c */
int _cor_send_announce(struct cor_announce_data *ann, int fromqos, int *sent);

void cor_announce_data_free(struct kref *ref);

void cor_announce_send_start(struct net_device *dev, char *mac, int type);

void cor_announce_send_stop(struct net_device *dev, char *mac, int type);

/* neigh_rcv.c */
void cor_kernel_packet(struct cor_neighbor *nb, struct sk_buff *skb,
                int ackneeded);

/* neigh_snd.c */
struct cor_control_msg_out;

#define ACM_PRIORITY_LOW 1 /* oom recovery easy */
#define ACM_PRIORITY_MED 2 /* oom may cause timeouts */
#define ACM_PRIORITY_HIGH 3 /* cm acks - needed for freeing old cms */

struct cor_control_msg_out *cor_alloc_control_msg(struct cor_neighbor *nb,
                int priority);

void cor_free_control_msg(struct cor_control_msg_out *cm);

void cor_retransmit_timerfunc(struct timer_list *retrans_timer);

void cor_kern_ack_rcvd(struct cor_neighbor *nb, __u32 seqno);

int cor_send_messages(struct cor_neighbor *nb, unsigned long cmsg_send_start_j,
                ktime_t cmsg_send_start_kt, int *sent);

void cor_controlmsg_timerfunc(struct timer_list *cmsg_timer);

void cor_schedule_controlmsg_timer(struct cor_neighbor *nb_cmsglocked);

void cor_send_rcvmtu(struct cor_neighbor *nb);

void cor_send_pong(struct cor_neighbor *nb, __u32 cookie, ktime_t ping_rcvtime);

int cor_send_reset_conn(struct cor_neighbor *nb, __u32 conn_id, int lowprio);

void cor_send_ack(struct cor_neighbor *nb, __u32 seqno, __u8 fast);

void cor_send_ack_conn_ifneeded(struct cor_conn *src_in_l, __u32 seqno_ooo,
                __u32 ooo_length);

void cor_send_priority(struct cor_conn *trgt_out_ll, __u16 priority);

void cor_free_ack_conns(struct cor_conn *src_in_lx);

void cor_send_connect_success(struct cor_control_msg_out *cm, __u32 conn_id,
                struct cor_conn *src_in);

void cor_send_connect_nb(struct cor_control_msg_out *cm, __u32 conn_id,
                __u32 seqno1, __u32 seqno2, struct cor_conn *src_in_ll);

void cor_send_conndata(struct cor_control_msg_out *cm, __u32 conn_id,
                __u32 seqno, char *data_orig, char *data, __u32 datalen,
                __u8 windowused, __u8 flush, __u8 highlatency,
                struct cor_conn_retrans *cr);

int __init cor_kgen_init(void);

void __exit cor_kgen_exit2(void);

/* conn.c */
extern struct kmem_cache *cor_connid_reuse_slab;

extern atomic_t cor_num_conns;

extern spinlock_t cor_bindnodes;

int cor_new_incoming_conn_allowed(struct cor_neighbor *nb);

__u32 _cor_conn_refresh_priority(struct cor_conn *cn_lx);

__u32 cor_conn_refresh_priority(struct cor_conn *cn, int locked);

void cor_set_conn_is_highlatency(struct cor_conn *cn, __u8 is_highlatency,
                int locked, int call_refresh_priority);

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

void cor_conn_set_last_act(struct cor_conn *trgt_out_lx);

int cor_conn_init_out(struct cor_conn *trgt_unconn_ll, struct cor_neighbor *nb,
                __u32 rcvd_connid, int use_rcvd_connid);

int cor_conn_init_sock_source(struct cor_conn *cn);

void cor_conn_init_sock_target(struct cor_conn *cn);

__u32 cor_list_services(char *buf, __u32 buflen);

void cor_set_publish_service(struct cor_sock *cs, __u8 value);

void cor_close_port(struct cor_sock *cs);

int cor_open_port(struct cor_sock *cs_l, __be32 port);

#define CONNECT_PORT_OK 0
#define CONNECT_PORT_PORTCLOSED 1
#define CONNECT_PORT_TEMPORARILY_OUT_OF_RESOURCES 2

int cor_connect_port(struct cor_conn *trgt_unconn_ll, __be32 port);

int cor_connect_neigh(struct cor_conn *trgt_unconn_ll, __be64 addr);

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

void cor_reset_conn_locked(struct cor_conn_bidir *cnb_ll);

void cor_reset_conn(struct cor_conn *cn);

/* conn_src_in.c */
void cor_reset_ooo_queue(struct cor_conn *src_in_lx);

void cor_drain_ooo_queue(struct cor_conn *src_in_l);

void cor_conn_rcv(struct cor_neighbor *nb, struct sk_buff *skb, char *data,
                __u32 len, __u32 conn_id, __u32 seqno, __u8 windowused,
                __u8 flush);

int __init cor_rcv_init(void);

void __exit cor_rcv_exit2(void);

/* conn_src_sock.c */
void cor_update_src_sock_sndspeed(struct cor_conn *src_sock_l,
                __u32 bytes_sent);

int cor_sock_sndbufavailable(struct cor_conn *src_sock_lx, int for_wakeup);

#define RC_FTC_OK 0
#define RC_FTC_OOM 1
#define RC_FTC_ERR 2
int _cor_mngdsocket_flushtoconn(struct cor_conn *src_sock_l);

int cor_mngdsocket_flushtoconn_ctrl(struct cor_sock *cs_m_l, __u8 send_eof,
                __u8 send_rcvend, __u8 send_keepalive_resp,
                __be32 keepalive_resp_cookie);

int cor_mngdsocket_flushtoconn_data(struct cor_sock *cs_m_l);

void cor_keepalive_req_timerfunc(struct timer_list *retrans_conn_timer);

void cor_keepalive_req_sched_timer(struct cor_conn *src_sock_lx);

void cor_keepalive_resp_rcvd(struct cor_sock *cs_m_l, __be32 cookie);

int __init cor_conn_src_sock_init1(void);

void __exit cor_conn_src_sock_exit1(void);

/* conn_trgt_unconn.c */
int cor_encode_len(char *buf, int buflen, __u32 len);

void cor_proc_cpacket(struct cor_conn *trgt_unconn);

/* conn_trgt_out.c */
void cor_free_connretrans(struct kref *ref);

void cor_reschedule_conn_retrans_timer(struct cor_neighbor *nb_retranslocked);

void cor_cancel_all_conn_retrans(struct cor_conn *trgt_out_l);

int cor_send_retrans(struct cor_neighbor *nb, int *sent);

void cor_retransmit_conn_timerfunc(struct timer_list *retrans_timer_conn);

void cor_conn_ack_ooo_rcvd(struct cor_neighbor *nb, __u32 conn_id,
                struct cor_conn *trgt_out, __u32 seqno_ooo, __u32 length,
                __u64 *bytes_acked);

void cor_conn_ack_rcvd(struct cor_neighbor *nb, __u32 conn_id,
                struct cor_conn *trgt_out, __u32 seqno, int setwindow,
                __u8 window, __u8 bufsize_changerate, __u64 *bytes_acked);

void cor_schedule_retransmit_conn(struct cor_conn_retrans *cr, int connlocked,
                int nbretrans_locked);

int cor_srcin_buflimit_reached(struct cor_conn *src_in_lx);

/* RC_FLUSH_CONN_OUT_SENT | RC_FLUSH_CONN_OUT_{^SENT} */
#define RC_FLUSH_CONN_OUT_OK 1
#define RC_FLUSH_CONN_OUT_SENT_CONG 2 /* cor_flush_out internal only */
#define RC_FLUSH_CONN_OUT_NBNOTACTIVE 3
#define RC_FLUSH_CONN_OUT_CONG 4
#define RC_FLUSH_CONN_OUT_MAXSENT 5
#define RC_FLUSH_CONN_OUT_OOM 6

int _cor_flush_out(struct cor_conn *trgt_out_lx, __u32 maxsend, __u32 *sent,
                int from_qos, int maxsend_forcedelay);

static inline int cor_flush_out(struct cor_conn *trgt_out_lx, __u32 *sent)
{
        int rc = _cor_flush_out(trgt_out_lx, 1 << 30, sent, 0, 0);

        if (rc == RC_FLUSH_CONN_OUT_CONG || rc == RC_FLUSH_CONN_OUT_MAXSENT ||
                        rc == RC_FLUSH_CONN_OUT_OOM ||
                        rc == RC_FLUSH_CONN_OUT_NBNOTACTIVE)
                cor_qos_enqueue_conn(trgt_out_lx);

        return rc;
}

int __init cor_snd_init(void);

void __exit cor_snd_exit2(void);

/* conn_trgt_sock.c */
void cor_flush_sock_managed(struct cor_conn *trgt_sock_lx, int from_recvmsg,
                __u8 *do_wake_sender);

void cor_flush_sock(struct cor_conn *trgt_sock_lx);

/* conn_databuf.c */
extern struct kmem_cache *cor_data_buf_item_slab;

void cor_databuf_init(struct cor_conn *cn_init);

void cor_bufsize_init(struct cor_conn *cn_l, __u32 bufsize);

int cor_account_bufspace(struct cor_conn *cn_lx);

int cor_conn_src_unconn_write_allowed(struct cor_conn *src_unconn_lx);

void cor_update_windowlimit(struct cor_conn *src_in_lx);

__u8 _cor_bufsize_update_get_changerate(struct cor_conn *cn_lx);

static inline int cor_bufsize_initial_phase(struct cor_conn *cn_lx)
{
        return unlikely(cn_lx->bufsize.bytes_rcvd != (1 << 24) - 1 &&
                        cn_lx->bufsize.bytes_rcvd < cn_lx->bufsize.bufsize);
}

static inline int cor_ackconn_urgent(struct cor_conn *cn_lx)
{
        return cor_bufsize_initial_phase(cn_lx) ||
                        cn_lx->bufsize.state == BUFSIZE_INCR_FAST;
}

void cor_bufsize_read_to_sock(struct cor_conn *trgt_sock_lx);

void cor_databuf_ackdiscard(struct cor_conn *cn_lx);

void cor_reset_seqno(struct cor_conn *cn_l, __u32 initseqno);

void cor_databuf_pull(struct cor_conn *cn_lx, char *dst, __u32 len);

static inline __u32 cor_databuf_trypull(struct cor_conn *cn_l, char *dst,
                __u32 len)
{
        if (len > cn_l->data_buf.read_remaining)
                len = cn_l->data_buf.read_remaining;
        cor_databuf_pull(cn_l, dst, len);
        return len;
}

void cor_databuf_unpull_dpi(struct cor_conn *trgt_sock, struct cor_sock *cs,
                struct cor_data_buf_item *item, __u16 next_read_offset);

void cor_databuf_pull_dbi(struct cor_sock *cs_rl, struct cor_conn *trgt_sock_l);

void cor_databuf_unpull(struct cor_conn *trgt_out_l, __u32 bytes);

void cor_databuf_pullold(struct cor_conn *trgt_out_l, __u32 startpos, char *dst,
                int len);

void cor_databuf_ack(struct cor_conn *trgt_out_l, __u32 pos);

void cor_databuf_ackread(struct cor_conn *cn_lx);

__u32 _cor_receive_buf(struct cor_conn *cn_lx, char *buf, __u32 datalen,
                int from_sock, __u8 windowused, __u8 flush);

static inline __u32 cor_receive_buf(struct cor_conn *cn_lx, char *buf,
                 __u32 datalen, __u8 windowused, __u8 flush)
{
        return _cor_receive_buf(cn_lx, buf, datalen, 0, windowused, flush);
}

static inline __u32 cor_receive_sock(struct cor_conn *src_sock_l, char *buf,
                __u32 datalen, __u8 flush)
{
        __u32 ret;

        BUG_ON(src_sock_l->sourcetype != SOURCE_SOCK);

        ret = _cor_receive_buf(src_sock_l, buf, datalen, 1,
                        src_sock_l->src.sock.last_windowused, flush);

        if (likely(ret > 0)) {
                __u32 bufsize = src_sock_l->bufsize.bufsize >> BUFSIZE_SHIFT;
                __u32 bufused = src_sock_l->data_buf.read_remaining;

                if (bufused >= bufsize)
                        src_sock_l->src.sock.last_windowused = 31;
                else if (unlikely(bufused * 31 > U32_MAX))
                        src_sock_l->src.sock.last_windowused =
                                        bufused / ((bufsize + 30) / 31);
                else
                        src_sock_l->src.sock.last_windowused =
                                        (bufused * 31) / bufsize;
        }

        return ret;
}

__u32 cor_receive_skb(struct cor_conn *src_in_l, struct sk_buff *skb,
                __u8 windowused, __u8 flush);

void cor_wake_sender(struct cor_conn *cn);

int __init cor_forward_init(void);

void __exit cor_forward_exit2(void);

/* sock.c */
void cor_free_sock(struct kref *ref);

int cor_socket_setsockopt_tos(struct socket *sock, char __user *optval,
                unsigned int optlen);

int cor_socket_setsockopt_priority(struct socket *sock, char __user *optval,
                unsigned int optlen);

int cor_socket_socketpair(struct socket *sock1, struct socket *sock2);

int cor_socket_getname(struct socket *sock, struct sockaddr *addr, int peer);

int cor_socket_mmap(struct file *file, struct socket *sock,
                struct vm_area_struct *vma);

int _cor_createsock(struct net *net, struct socket *sock, int protocol,
                int kern, __u8 is_client);

int __init cor_sock_init1(void);

int __init cor_sock_init2(void);

void __exit cor_sock_exit1(void);

/* sock_rdaemon.c */
int cor_is_device_configurated(struct net_device *dev);

int cor_create_rdaemon_sock(struct net *net, struct socket *sock, int protocol,
                int kern);

int cor_rdreq_connect(struct cor_sock *cs);

void cor_usersock_release(struct cor_sock *cs);

int __init cor_rd_init1(void);

int __init cor_rd_init2(void);

void __exit cor_rd_exit1(void);

void __exit cor_rd_exit2(void);

/* sock_raw.c */
int cor_create_raw_sock(struct net *net, struct socket *sock, int protocol,
                int kern);

/* sock_managed.c */
struct cor_sock *cor_get_sock_by_cookie(__be64 cookie);

void __cor_set_sock_connecterror(struct cor_sock *cs_m_l, int errorno);

void _cor_set_sock_connecterror(struct cor_sock *cs, int errorno);

void cor_mngdsocket_chksum(char *hdr, __u32 hdrlen, char *data, __u32 datalen,
                char *chksum, __u32 chksum_len);

static inline void cor_set_sock_connecterror(__be64 cookie, int errorno)
{
        struct cor_sock *cs = cor_get_sock_by_cookie(cookie);

        if (cs != 0) {
                _cor_set_sock_connecterror(cs, errorno);
                kref_put(&cs->ref, cor_free_sock);
        }
}

void cor_mngdsocket_readfromconn_fromatomic(struct cor_sock *cs);

void cor_mngdsocket_readfromconn_wq(struct work_struct *work);

int cor_create_managed_sock(struct net *net, struct socket *sock, int protocol,
                int kern);

int __init cor_sock_managed_init1(void);


static inline struct cor_skb_procstate *cor_skb_pstate(struct sk_buff *skb)
{
        BUILD_BUG_ON(sizeof(struct cor_skb_procstate) > sizeof(skb->cb));
        return (struct cor_skb_procstate *) &skb->cb[0];
}

static inline struct sk_buff *cor_skb_from_pstate(struct cor_skb_procstate *ps)
{
        return (struct sk_buff *) (((char *)ps) - offsetof(struct sk_buff, cb));
}

static inline int cor_qos_fastsend_allowed_conn_retrans(struct cor_neighbor *nb)
{
        return atomic_read(&nb->queue->cong_status) < CONGSTATUS_RETRANS;
}

static inline int cor_qos_fastsend_allowed_announce(struct net_device *dev)
{
        int rc;
        struct cor_qos_queue *q = cor_get_queue(dev);

        if (q == 0)
                return 0;

        rc = atomic_read(&q->cong_status) < CONGSTATUS_ANNOUNCE;

        kref_put(&q->ref, cor_free_qos);

        return rc;
}

static inline int cor_qos_fastsend_allowed_conn(struct cor_conn *trgt_out_lx)
{
        struct cor_qos_queue *q = trgt_out_lx->trgt.out.nb->queue;

        return atomic_read(&q->cong_status) < CONGSTATUS_CONNDATA;
}

static inline __u32 cor_rcv_mtu(struct cor_neighbor *nb)
{
        return nb->dev->mtu;
}

static inline __u32 cor_snd_mtu(struct cor_neighbor *nb)
{
        return min((__u32) nb->dev->mtu,
                        (__u32) atomic_read(&nb->remote_rcvmtu));
}

static inline __u32 cor_mss(struct cor_neighbor *nb, __u32 l3overhead)
{
        return cor_snd_mtu(nb) - LL_RESERVED_SPACE(nb->dev) - l3overhead;
}

static inline __u32 cor_mss_cmsg(struct cor_neighbor *nb)
{
        return cor_mss(nb, 5);
}

static inline __u32 cor_mss_conndata(struct cor_neighbor *nb, int highlatency)
{
        __u32 mss_tmp = cor_mss(nb, 9);
        __u32 i;

        if (mss_tmp < 256 || highlatency || LOWLATENCY_LOWERMTU == 0)
                return mss_tmp;

        for (i = 256; i < 4096; i *= 2) {
                if (i * 2 > mss_tmp)
                        return i;
        }

        return mss_tmp - mss_tmp % 4096;
}

static inline __u32 cor_send_conndata_as_skb(struct cor_neighbor *nb,
                __u32 size)
{
        return size >= cor_mss_conndata(nb, 0) / 2 ||
                        size > KP_CONN_DATA_MAXLEN;
}

static inline long cor_calc_timeout(__u32 latency_us, __u32 latency_stddev_us,
                __u32 max_remote_ack_delay_us)
{
        unsigned long addto;

        if (unlikely(unlikely(latency_us > 1000000000) ||
                        unlikely(latency_stddev_us > 500000000) ||
                        unlikely(max_remote_ack_delay_us > 1000000000))) {
                addto = msecs_to_jiffies(latency_us / 1000 + latency_us / 4000 +
                                latency_stddev_us / 333 +
                                max_remote_ack_delay_us / 1000);
        } else {
                addto = usecs_to_jiffies(latency_us + latency_us / 4 +
                                latency_stddev_us * 3 +
                                max_remote_ack_delay_us);
        }

        /**
         * 2 is added because
         * 1) _to_jiffies rounds down, but should round up, so add 1 to
         *    compensate
         * 2) even if latency is 0, we never want to schedule the retransmit
         *    to run right now, so add 1 more
         */
        return jiffies + 2 + addto;
}

static inline void cor_put_be64(char *dst, __be64 value)
{
        char *p_value = (char *) &value;

        dst[0] = p_value[0];
        dst[1] = p_value[1];
        dst[2] = p_value[2];
        dst[3] = p_value[3];
        dst[4] = p_value[4];
        dst[5] = p_value[5];
        dst[6] = p_value[6];
        dst[7] = p_value[7];
}

static inline void cor_put_u64(char *dst, __u64 value)
{
        cor_put_be64(dst, cpu_to_be64(value));
}

static inline void cor_put_be32(char *dst, __be32 value)
{
        char *p_value = (char *) &value;

        dst[0] = p_value[0];
        dst[1] = p_value[1];
        dst[2] = p_value[2];
        dst[3] = p_value[3];
}

static inline void cor_put_u32(char *dst, __u32 value)
{
        cor_put_be32(dst, cpu_to_be32(value));
}

static inline void cor_put_be16(char *dst, __be16 value)
{
        char *p_value = (char *) &value;

        dst[0] = p_value[0];
        dst[1] = p_value[1];
}

static inline void cor_put_u16(char *dst, __u16 value)
{
        cor_put_be16(dst, cpu_to_be16(value));
}

static inline char *cor_pull_skb(struct sk_buff *skb, unsigned int len)
{
        char *ptr = skb_pull(skb, len);

        if (unlikely(ptr == 0))
                return 0;

        return ptr - len;
}

static inline __be64 cor_parse_be64(char *buf)
{
        __be64 ret = 0;

        BUG_ON(buf == 0);

        ((char *)&ret)[0] = buf[0];
        ((char *)&ret)[1] = buf[1];
        ((char *)&ret)[2] = buf[2];
        ((char *)&ret)[3] = buf[3];
        ((char *)&ret)[4] = buf[4];
        ((char *)&ret)[5] = buf[5];
        ((char *)&ret)[6] = buf[6];
        ((char *)&ret)[7] = buf[7];

        return ret;
}

static inline __u64 cor_parse_u64(char *buf)
{
        return be64_to_cpu(cor_parse_be64(buf));
}

static inline __be32 cor_parse_be32(char *ptr)
{
        __be32 ret = 0;

        BUG_ON(ptr == 0);

        ((char *)&ret)[0] = ptr[0];
        ((char *)&ret)[1] = ptr[1];
        ((char *)&ret)[2] = ptr[2];
        ((char *)&ret)[3] = ptr[3];

        return ret;
}

static inline __u32 cor_parse_u32(char *ptr)
{
        return be32_to_cpu(cor_parse_be32(ptr));
}

static inline __be16 cor_parse_be16(char *ptr)
{
        __be16 ret = 0;

        BUG_ON(ptr == 0);

        ((char *)&ret)[0] = ptr[0];
        ((char *)&ret)[1] = ptr[1];

        return ret;
}

static inline __u16 cor_parse_u16(char *ptr)
{
        return be16_to_cpu(cor_parse_be16(ptr));
}

static inline __u8 cor_parse_u8(char *ptr)
{
        BUG_ON(ptr == 0);
        return (__u8) ptr[0];
}

static inline __be32 cor_pull_be32(struct sk_buff *skb)
{
        return cor_parse_be32(cor_pull_skb(skb, 4));
}

static inline __u32 cor_pull_u32(struct sk_buff *skb)
{
        return cor_parse_u32(cor_pull_skb(skb, 4));
}

static inline __u16 cor_pull_u16(struct sk_buff *skb)
{
        return cor_parse_u16(cor_pull_skb(skb, 2));
}

static inline __u8 cor_pull_u8(struct sk_buff *skb)
{
        char *ptr = cor_pull_skb(skb, 1);

        BUG_ON(ptr == 0);
        return *ptr;
}

static inline int cor_is_conn_in(struct cor_conn *cn_l, struct cor_neighbor *nb,
                __u32 conn_id)
{
        if (unlikely(unlikely(cn_l->sourcetype != SOURCE_IN) ||
                        unlikely(cn_l->src.in.nb != nb) ||
                        unlikely(cn_l->src.in.conn_id != conn_id) ||
                        unlikely(cn_l->isreset != 0)))
                return 0;
        return 1;
}

static inline int cor_is_src_sock(struct cor_conn *cn_l, struct cor_sock *cs)
{
        if (unlikely(unlikely(cn_l->sourcetype != SOURCE_SOCK) ||
                        unlikely(cn_l->src.sock.ed->cs != cs)))
                return 0;
        return 1;
}

static inline int cor_is_trgt_sock(struct cor_conn *cn_l, struct cor_sock *cs)
{
        if (unlikely(unlikely(cn_l->targettype != TARGET_SOCK) ||
                        unlikely(cn_l->trgt.sock.cs != cs)))
                return 0;
        return 1;
}

#define BUFLEN_MIN 128
#define BUFLEN_MAX 4096
#define PAGESIZE (1 << PAGE_SHIFT)

static inline __u32 cor_buf_optlen(__u32 datalen, int from_sock)
{
        __u32 optlen = BUFLEN_MIN;

        if (from_sock)
                return BUFLEN_MAX;

        while (optlen < datalen && optlen < PAGESIZE && optlen < BUFLEN_MAX)
                optlen = (optlen << 1);
        return optlen;
}


static inline void cor_databuf_item_free(struct cor_data_buf_item *item)
{
        if (item->type == DATABUF_BUF) {
                kfree(item->buf);
                kmem_cache_free(cor_data_buf_item_slab, item);
        } else if (item->type == DATABUF_SKB) {
                struct sk_buff *skb = cor_skb_from_pstate(container_of(item,
                                struct cor_skb_procstate, funcstate.rcv.dbi));
                kfree_skb(skb);
        } else {
                BUG();
        }
}

static inline int cor_seqno_eq(__u32 seqno1, __u32 seqno2)
{
        return seqno1 == seqno2;
}

static inline int cor_seqno_before(__u32 seqno1, __u32 seqno2)
{
        return (seqno1 - seqno2) >= (1LL << 31);
}

static inline int cor_seqno_before_eq(__u32 seqno1, __u32 seqno2)
{
        return cor_seqno_eq(seqno1, seqno2) || cor_seqno_before(seqno1, seqno2);
}

static inline int cor_seqno_after(__u32 seqno1, __u32 seqno2)
{
        return cor_seqno_before_eq(seqno1, seqno2) ? 0 : 1;
}

static inline int cor_seqno_after_eq(__u32 seqno1, __u32 seqno2)
{
        return cor_seqno_before(seqno1, seqno2) ? 0 : 1;
}

static inline int ktime_before_eq(ktime_t time1, ktime_t time2)
{
        return ktime_after(time1, time2) ? 0 : 1;
}

static inline int ktime_after_eq(ktime_t time1, ktime_t time2)
{
        return ktime_before(time1, time2) ? 0 : 1;
}

static inline __u64 cor_update_atomic_sum(atomic64_t *atomic_sum,
                __u32 oldvalue, __u32 newvalue)
{
        __u64 sum_old = atomic64_read(atomic_sum);
        __u64 sum;

        while (1) {
                __u64 cmpxchg_ret;

                sum = sum_old;

                BUG_ON(sum < oldvalue);
                sum -= oldvalue;

                BUG_ON(sum + newvalue < sum);
                sum += newvalue;

                cmpxchg_ret = atomic64_cmpxchg(atomic_sum, sum_old, sum);

                if (likely(cmpxchg_ret == sum_old))
                        break;

                sum_old = cmpxchg_ret;
        }

        return sum;
}

static inline void cor_sk_write_space(struct cor_sock *cs)
{
        atomic_set(&cs->ready_to_write, 1);
        barrier();
        cs->sk.sk_write_space(&cs->sk);
}

static inline void cor_sk_data_ready(struct cor_sock *cs)
{
        atomic_set(&cs->ready_to_read, 1);
        barrier();
        cs->sk.sk_data_ready(&cs->sk);
}

/* the other direction may be locked only if called from cor_proc_cpacket */
static inline void cor_flush_buf(struct cor_conn *cn_lx)
{
        if (unlikely(cn_lx->targettype == TARGET_UNCONNECTED)) {
                cor_proc_cpacket(cn_lx);
        } else if (cn_lx->targettype == TARGET_SOCK) {
                cor_flush_sock(cn_lx);
        } else if (cn_lx->targettype == TARGET_OUT) {
                __u32 bytessent = 0;

                cor_flush_out(cn_lx, &bytessent);
        } else if (unlikely(cn_lx->targettype == TARGET_DISCARD)) {
                cor_databuf_ackdiscard(cn_lx);
        } else {
                BUG();
        }
}