fast initial bufsize increase

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

/**
 *      Connection oriented routing
 *      Copyright (C) 2007-2020 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 <asm/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

#define SOCKADDRTYPE_PORT 1
#define SOCKADDRTYPE_ADDRPORT 2
struct cor_sockaddr {
        __u16 sin_family;

        __be16 port;
        char addr[64];
};

#define COR_PASS_ON_CLOSE 1
#define COR_PUBLISH_SERVICE 2

#define MAX_CONN_CMD_LEN 64


#define PACKET_TYPE_ANNOUNCE 1
#define PACKET_TYPE_CMSG 2
#define PACKET_TYPE_CONNDATA 4
#define PACKET_TYPE_CONNDATA_LOWBUFDELAYED 5
#define PACKET_TYPE_CONNDATA_FLUSH 6
#define PACKET_TYPE_CONNDATA_LOWBUFDELAYED_FLUSH 7

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

/* KP_PADDING[1] */
#define KP_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_INIT_SESSION 1

#define KP_INIT_SESSION_CMDLEN 5

/*
 * KP_PING[1] cookie[4]
 * KP_PONG[1] cookie[4] respdelay[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_PING 2

#define KP_PING_CMDLEN 5

#define KP_PONG 3

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

/*
 * KP_ACK_CONN[1] conn_id[4] flags[1] seqno[6] window[1] seqno_ooo[6]
 *   length[1-4] priority_seqno[1] priority[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
 *
 * flags defines which of the following fields are sent
 *
 * 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
 *
 * 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 5

#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 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 ooolen(__u8 flags)
{
        int len = ((flags & KP_ACK_CONN_FLAGS_OOO) >> 2);
        if (unlikely(len == 3))
                return 4;
        return len;
}

static inline int ack_conn_len(__u8 flags)
{
        int len = 0;
        if ((flags & KP_ACK_CONN_FLAGS_SEQNO) != 0) {
                len += 6;
                if ((flags & KP_ACK_CONN_FLAGS_WINDOW) != 0)
                        len++;
        }

        if (ooolen(flags) != 0) {
                len += 6;
                len += ooolen(flags);
        }

        if (flags & KP_ACK_CONN_FLAGS_PRIORITY)
                len += 2;

        return len;
}

/*
 * 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[6] seqno2[6] window[1] priority_seqno[1]
 * priority[1]
 */
#define KP_CONNECT 6

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

/* KP_CONN_DATA[1] conn_id[4] seqno[6] length[2] data[length] */
#define KP_CONN_DATA 8
#define KP_CONN_DATA_LOWBUFDELAYED 9
#define KP_CONN_DATA_FLUSH 10
#define KP_CONN_DATA_LOWBUFDELAYED_FLUSH 11

#define KP_CONN_DATA_CMDLEN 13

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

/*
 * KP_SET_MAX_CMSG_DELAY[1] cpacket_ack_delay[4] data_ack_delay[4] cmsg_delay[4]
 * Sent after connecting and at any change
 * delay in specifies in microsecs
 */
#define KP_SET_MAX_CMSG_DELAY 13

#define KP_SET_MAX_CMSG_DELAY_CMDLEN 13


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

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

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

/*
 * CD_LIST_NEIGH sends CDR_BINDATA if the command was 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. Therefore clients must
 * silently discard data at the end they do not expect.
 */

/* list connected neighbors: CD_LIST_NEIGH[2] length[1-4] */
#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] */
#define CD_LIST_SERVICES 4


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

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

/*
 * CDR_EXECFAILED[1] reasoncode[2]
 */
#define CDR_EXECFAILED 2
        #define CDR_EXECFAILED_INVALID_COMMAND 1
        #define CDR_EXECFAILED_TEMPORARILY_OUT_OF_RESSOURCES 2
        #define CDR_EXECFAILED_NB_DOESNTEXIST 3
        #define CDR_EXECFAILED_PORTCLOSED 4

/*
 * must be sent after CDR_EXEC{OK|FAILED}
 * CDR_EXEOK_BINDATA[1] bindatalen[1-4] bindata[bindatalen] */
#define CDR_BINDATA 3

/*
 * 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
/*
 * CRD_UTK_UP[4] length[4] flags[8] addrlen[4] addr[addrlen]
 *
 */


#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_CHECKSUMLEN 4

#define CONN_MNGD_HASDATA (1 << 15)
#define CONN_MNGD_EOF (1 << 0)
#define CONN_MNGD_RCVEND (1 << 1)
#define CONN_MNGD_DATALEN 4095

#define CONN_MNGD_MAX_SEGMENT_SIZE (CONN_MNGD_DATALEN + 1)

#define PRIORITY_MAX 15384774




/* result codes for rcv.c/proc_packet */
#define RC_DROP 0
#define RC_FINISHED 1

#define RC_RCV1_ANNOUNCE 2
#define RC_RCV1_KERNEL 3
#define RC_RCV1_CONN 4


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

struct 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 timer_list qos_resume_timer;
        struct tasklet_struct qos_resume_task;
        int 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;

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

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

#define ANNOUNCE_TYPE_BROADCAST 1
#define ANNOUNCE_TYPE_UNICAST 2

struct 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 resume_block rb;
};

struct 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 neighbor{
        struct list_head nb_list;

        struct kref ref;

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

        __be32 sessionid;

        atomic_t sessionid_rcv_needed;
        atomic_t sessionid_snd_needed;

        char *addr;
        __u16 addrlen;

        struct timer_list cmsg_timer;
        spinlock_t cmsg_lock;
        struct list_head cmsg_queue_pong;
        struct list_head cmsg_queue_ack;
        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;
        __u64 kpacket_seqno; /* not locked, only accessed by single tasklet */

        struct rb_root pending_conn_resets_rb;

        unsigned long timeout;

        __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 tasklet */
        __u8 max_cmsg_delay_sent;

        /* procected by qos_queue->qlock */
        struct resume_block rb_kp;
        struct resume_block rb_cr;
        struct resume_block rb;

        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 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 */
        atomic_t max_remote_ack_delay_us; /* microsecs */
        atomic_t max_remote_ackconn_delay_us; /* microsecs */
        atomic_t max_remote_other_delay_us; /* microsecs */

        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;

        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 rcv_conn_list;

        spinlock_t stalledconn_lock;
        struct work_struct stalledconn_work;
        __u8 stalledconn_work_scheduled;
        struct list_head stalledconn_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_list;
        struct rb_root kp_retransmits_rb;

        spinlock_t retrans_conn_lock;
        struct timer_list retrans_conn_timer;
        struct list_head retrans_conn_list;
};

#define DATABUF_BUF 0
#define DATABUF_SKB 1

struct data_buf_item{
        struct list_head buf_list;

        char *buf;
        __u16 datalen;
        __u16 buflen;

        __u8 type;
};

struct 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 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;

/**
 * There are 2 conn objects per bi-directional connection. They refer to each
 * other with in the reversedir field.
 *
 *
 *
 * Naming:
 *
 * cn: conn we have no clue what is inside
 * src_in, trgt_unconn, trgt_out, ...: A conn with the specified source or
 * targettype. In the unlocked case the types are actually just a guess,
 * because they could 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
 *
 * Exception: they may not change after they are set to source/target sock
 * until the socket is released.
 *
 *
 * 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.
 *
 *
 *
 * Locking:
 * The following fields are immutable after the conn has been allocated:
 * is_client, reversedir
 *
 * 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, htab_entries, ???. In this case the side with is_client == 1
 * needs to be locked first. Changes to conn_id and neighbor also require
 * removing the conn from the htables 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 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

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

        __u8    is_client:1,
                is_highlatency:1;

        /*
         * isreset values:
         * 0... connection active
         * 1... connection is about to be reset, target does not need to be
         *      notified
         * 2... connection is reset
         * 3... connection is reset + no pointers to "struct conn *reversedir"
         *      remaining except from this conn
         */
        __u8    isreset;

        __u8    flush;

        struct kref ref;

        spinlock_t rcv_lock;

        union{
                struct{
                        struct neighbor *nb;
                        /* list of all connections from this neighbor */
                        struct list_head nb_list;

                        struct list_head reorder_queue;
                        __u32 reorder_memused;

                        struct rb_node rbn;
                        struct connid_reuse_item *cir;
                        __u32 conn_id;
                        __u64 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;

                        __u64 window_seqnolimit;
                        __u64 window_seqnolimit_remote;

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

                        unsigned long jiffies_last_act;
                }in;

                struct{
                        struct cor_sock *cs;

                        struct list_head cl_list;

                        __u32 priority;

                        struct snd_speed snd_speed;
                }sock;
        }source;

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

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

                        /* protected by nb->stalledconn_lock */
                        struct list_head nbstalled_lh;

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

                        struct resume_block rb;

                        __u32 rb_priority;
                        __u16 maxsend_extra;

                        __u16 burst_bytes;

                        __u8 priority_last;
                        __u8 priority_seqno;
                        __u8 priority_send_allowed;

                        __u8 windowlimit_reached;

                        __u8 established;

                        /* 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;
        }target;

        struct{
                struct list_head items;
                struct data_buf_item *nextread;
                __u64 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;

        struct conn *reversedir;
};

#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 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 conn *trgt_out_o;
        __u64 seqno;
        __u32 length;

        __u8 snd_delayed_lowbuf;
        __u8 state;
        unsigned long timeout;
};

#define RCVOOO_BUF 0
#define RCVOOO_SKB 1
struct rcvooo{
        struct list_head lh;
        __u64 seqno;
        __u8 type;
        __u8 flush;
};

struct rcvooo_buf{
        struct rcvooo r;
        char *data;
        __u32 len;
};

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

                struct{
                        __u32 offset;
                }announce2;

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

                struct{
                        struct 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

#define CS_SHUTDOWN_SHUTDOWN_RD (1 << 0)
#define CS_SHUTDOWN_SHUTDOWN_WR (1 << 1)
#define CS_SHUTDOWN_SENT_EOF (1 << 2)
#define CS_SHUTDOWN_SENT_RCVEND (1 << 3)
#define CS_SHUTDOWN_RCVD_EOF (1 << 4)
#define CS_SHUTDOWN_RCVD_RCVEND (1 << 5)

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;

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

                struct {
                        struct conn *src_sock;
                        struct conn *trgt_sock;

                        struct data_buf_item *rcvitem;
                        __u16 rcvoffset;

                        __u8 snd_delayed_lowbuf;

                        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 shutdownflags;

                        __u8 snd_delayed_lowbuf;

                        __be64 cookie;
                        struct rb_node rbn;

                        struct conn *src_sock;
                        struct conn *trgt_sock;

                        char snd_hdr[CONN_MNGD_HEADERLEN];
                        char snd_chksum[CONN_MNGD_CHECKSUMLEN];
                        char *snd_buf;
                        __u16 snd_segment_size;
                        __u16 snd_hdr_flags;
                        __u16 snd_data_len;
                        __u16 sent;
                        /* protected by cor_sock->lock */
                        __u8 in_flushtoconn_oom_list;
                        /* protected by flushtoconn_oom_lock */
                        struct list_head flushtoconn_oom_lh;

                        /* receiving */
                        char rcv_hdr[CONN_MNGD_HEADERLEN];
                        char rcv_chksum[CONN_MNGD_CHECKSUMLEN];
                        char *rcv_buf;
                        __u16 rcv_hdr_flags;
                        __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;
};



/* common.c */
extern atomic_t num_conns;

extern spinlock_t cor_bindnodes;

extern struct conn *get_conn(struct neighbor *nb, __u32 conn_id);

extern void delete_connid_reuse_items(struct neighbor *nb);

extern void connid_used_pingsuccess(struct neighbor *nb);

extern void _set_last_act(struct conn *src_in_l);

extern void free_conn(struct kref *ref);

extern int conn_init_out(struct conn *trgt_unconn_ll, struct neighbor *nb,
                __u32 rcvd_connid, int use_rcvd_connid);

extern void conn_init_sock_source(struct conn *cn);

extern void conn_init_sock_target(struct conn *cn);

extern __u32 list_services(char *buf, __u32 buflen);

extern void set_publish_service(struct cor_sock *cs, __u8 value);

extern void close_port(struct cor_sock *cs);

extern int open_port(struct cor_sock *cs_l, __be16 port);

extern int connect_port(struct conn *trgt_unconn_ll, __be16 port);

extern int connect_neigh(struct conn *trgt_unconn_ll, char *addr,
                __u16 addrlen);

extern struct conn* alloc_conn(gfp_t allocflags);

extern void reset_conn_locked(struct conn *cn_ll);

extern void reset_conn(struct conn *cn);

/* util.c */
extern __u8 __attribute__((const)) enc_log_256_16(__u32 value);

extern __u32 __attribute__((const)) dec_log_256_16(__u8 value);

extern __u8 __attribute__((const)) enc_log_64_11(__u32 value);

extern __u32 __attribute__((const)) dec_log_64_11(__u8 value);

extern __u8 __attribute__((const)) enc_log_64_7(__u64 value);

extern __u64 __attribute__((const)) dec_log_64_7(__u8 value);

extern void kreffree_bug(struct kref *ref);

extern int __init cor_util_init(void);

/* credits.c */
extern int newconn_checkpriority(struct neighbor *nb, __u8 priority);

extern __u32 refresh_conn_priority(struct conn *cn, int locked);

extern void set_conn_in_priority(struct neighbor *nb, __u32 conn_id,
                struct conn *src_in, __u8 priority_seqno, __u8 priority);

extern void connreset_priority(struct conn *cn);

extern int __init credits_init(void);

/* neighbor.c */
extern void neighbor_free(struct kref *ref);

extern int is_from_nb(struct sk_buff *skb, struct neighbor *nb);

extern struct neighbor *get_neigh_by_mac(struct sk_buff *skb);

extern struct neighbor *find_neigh(char *addr, __u16 addrlen);

extern __u32 generate_neigh_list(char *buf, __u32 buflen);

extern int get_neigh_state(struct neighbor *nb);

extern void ping_resp(struct neighbor *nb, __u32 cookie, __u32 respdelay);

extern __u32 add_ping_req(struct neighbor *nb, unsigned long *last_ping_time,
                ktime_t now);

extern void ping_sent(struct neighbor *nb, __u32 cookie);

extern void unadd_ping_req(struct neighbor *nb, __u32 cookie,
                unsigned long last_ping_time, int congested);

#define TIMETOSENDPING_NO 0
#define TIMETOSENDPING_YES 1
#define TIMETOSENDPING_FORCE 2
extern int time_to_send_ping(struct neighbor *nb);

extern unsigned long get_next_ping_time(struct neighbor *nb);

extern int force_ping(struct neighbor *nb);

extern int rcv_announce(struct sk_buff *skb);

extern int _send_announce(struct announce_data *ann, int fromqos, int *sent);

extern void announce_data_free(struct kref *ref);

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

extern void cor_neighbor_down(void);

extern int cor_neighbor_up(char *addr2, __u32 addrlen2);

extern int is_clientmode(void);

extern int __init cor_neighbor_init(void);

/* rcv.c */
extern void reset_ooo_queue(struct conn *src_in_lx);

extern void drain_ooo_queue(struct conn *src_in_l);

extern void conn_rcv(struct neighbor *nb, struct sk_buff *skb, char *data, __u32 len,
                __u32 conn_id, __u64 seqno, int rcv_delayed_lowbuf, __u8 flush);

extern void cor_rcv_down(void);

extern void cor_rcv_up(void);

extern int __init cor_rcv_init(void);

/* kpacket_parse.c */
extern void kernel_packet(struct neighbor *nb, struct sk_buff *skb,
                __u64 seqno);

/* kpacket_gen.c */
struct 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 */

extern struct control_msg_out *alloc_control_msg(struct neighbor *nb,
                int priority);

extern void free_control_msg(struct control_msg_out *cm);

extern void retransmit_timerfunc(struct timer_list *retrans_timer);

extern void kern_ack_rcvd(struct neighbor *nb, __u64 seqno);

extern int send_messages(struct neighbor *nb, int *sent);

extern void controlmsg_timerfunc(struct timer_list *cmsg_timer);

extern void schedule_controlmsg_timer(struct neighbor *nb_cmsglocked);

extern void send_pong(struct neighbor *nb, __u32 cookie);

extern int send_reset_conn(struct neighbor *nb, __u32 conn_id, int lowprio);

extern void send_ack(struct neighbor *nb, __u64 seqno);

extern void send_ack_conn_ifneeded(struct conn *src_in_l, __u64 seqno_ooo,
                __u32 ooo_length);

extern void send_priority(struct conn *trgt_out_ll, int force,
                __u8 priority);

extern void free_ack_conns(struct conn *src_in_lx);

extern void send_connect_success(struct control_msg_out *cm, __u32 conn_id,
                struct conn *src_in);

extern void send_connect_nb(struct control_msg_out *cm, __u32 conn_id,
                __u64 seqno1, __u64 seqno2, struct conn *src_in_ll);

extern void send_conndata(struct control_msg_out *cm, __u32 conn_id,
                __u64 seqno, char *data_orig, char *data, __u32 datalen,
                __u8 snd_delayed_lowbuf, __u8 flush, __u8 highlatency,
                struct conn_retrans *cr);

extern int __init cor_kgen_init(void);

/* cpacket_parse.c */
extern int encode_len(char *buf, int buflen, __u32 len);

extern void proc_cpacket(struct conn *trgt_unconn);

extern int __init cor_cpacket_init(void);

/* snd.c */
extern void qos_set_lastdrop(struct qos_queue *q);

#ifdef DEBUG_QOS_SLOWSEND
extern 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 qos_queue *q,
                int caller)
{
        int rc = _cor_dev_queue_xmit(skb, caller);
        if (unlikely(rc != NET_XMIT_SUCCESS))
                qos_set_lastdrop(q);
        return rc;
}

extern void free_qos(struct kref *ref);

#ifdef COR_NBCONGWIN
extern void nbcongwin_data_acked(struct neighbor *nb, __u64 bytes_acked);
#endif

extern struct qos_queue *get_queue(struct net_device *dev);

extern int destroy_queue(struct net_device *dev);

extern int create_queue(struct net_device *dev);

#define QOS_RESUME_DONE 0
#define QOS_RESUME_CONG 1
#define QOS_RESUME_NEXTNEIGHBOR 2 /* resume_neighbors() internal */

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

extern void qos_enqueue(struct qos_queue *q, struct resume_block *rb,
                int caller);

extern void qos_remove_conn(struct conn *trgt_out_l);

extern int may_send_announce(struct net_device *dev);

extern struct sk_buff *create_packet_cmsg(struct neighbor *nb, int size,
                gfp_t alloc_flags, __u64 seqno);

extern void reschedule_conn_retrans_timer(struct neighbor *nb_retranslocked);

extern void cancel_all_conn_retrans(struct conn *trgt_out_l);

extern void retransmit_conn_taskfunc(unsigned long nb);

extern void retransmit_conn_timerfunc(struct timer_list *retrans_timer_conn);

extern void conn_ack_ooo_rcvd(struct neighbor *nb, __u32 conn_id,
                struct conn *trgt_out, __u64 seqno_ooo, __u32 length,
                __u64 *bytes_acked);

extern void conn_ack_rcvd(struct neighbor *nb, __u32 conn_id,
                struct conn *trgt_out, __u64 seqno, int setwindow, __u8 window,
                __u64 *bytes_acked);

extern void schedule_retransmit_conn(struct conn_retrans *cr, int connlocked,
                int nbretrans_locked);

extern int srcin_buflimit_reached(struct 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 /* 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

extern int flush_out(struct conn *trgt_out_lx, __u32 *sent);

extern void resume_nbstalled_conns(struct work_struct *work);

extern int __init cor_snd_init(void);

/* forward.c */
extern struct kmem_cache *data_buf_item_slab;

extern void databuf_init(struct conn *cn_init);

extern void bufsize_init(struct conn *cn_l, __u32 bufsize);

extern int account_bufspace(struct conn *cn_lx);

extern int cpacket_write_allowed(struct conn *src_unconn_lx);

extern void update_windowlimit(struct conn *src_in_lx);

extern void bufsize_read_to_sock(struct conn *trgt_sock_lx);

extern void databuf_ackdiscard(struct conn *cn_lx);

extern void reset_seqno(struct conn *cn_l, __u64 initseqno);

extern void databuf_pull(struct conn *cn_lx, char *dst, __u32 len);

static inline __u32 databuf_trypull(struct conn *cn_l, char *dst, __u32 len)
{
        if (len > cn_l->data_buf.read_remaining)
                len = cn_l->data_buf.read_remaining;
        databuf_pull(cn_l, dst, len);
        return len;
}

extern void databuf_unpull_dpi(struct conn *trgt_sock, struct cor_sock *cs,
                struct data_buf_item *item, __u16 next_read_offset);

extern void databuf_pull_dbi(struct cor_sock *cs_rl, struct conn *trgt_sock_l);

extern void databuf_unpull(struct conn *trgt_out_l, __u32 bytes);

extern void databuf_pullold(struct conn *trgt_out_l, __u64 startpos, char *dst,
                int len);

extern void databuf_ack(struct conn *trgt_out_l, __u64 pos);

extern void databuf_ackread(struct conn *cn_lx);

extern __u32 receive_buf(struct conn *cn_lx, char *buf, __u32 datalen,
                int rcv_delayed_lowbuf, __u8 flush);

extern __u32 receive_skb(struct conn *src_in_l, struct sk_buff *skb,
                int rcv_delayed_lowbuf, __u8 flush);

extern void wake_sender(struct conn *cn);

extern int __init forward_init(void);

/* sock_rdaemon.c */
extern int cor_create_rdaemon_sock(struct net *net, struct socket *sock,
                int protocol, int kern);

extern int rdreq_connect(struct cor_sock *cs);

extern void cor_usersock_release(struct cor_sock *cs);

extern int __init cor_rd_init1(void);

extern int __init cor_rd_init2(void);

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

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

extern void __set_sock_connecterror(struct cor_sock *cs_m_l, int errorno);

extern void _set_sock_connecterror(struct cor_sock *cs, int errorno);

static inline void set_sock_connecterror(__be64 cookie, int errorno)
{
        struct cor_sock *cs = get_corsock_by_cookie(cookie);
        _set_sock_connecterror(cs, errorno);
}

extern void flush_sock_managed(struct conn *trgt_sock_lx, int from_recvmsg,
                __u8 *do_wake_sender);

extern void cor_mngdsocket_readfromconn_fromatomic(struct cor_sock *cs);

extern void cor_mngdsocket_readfromconn_wq(struct work_struct *work);

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

extern int __init cor_sock_managed_init1(void);

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

extern void flush_sock(struct conn *trgt_sock_lx);

extern void update_src_sock_sndspeed(struct conn *src_sock_l, __u32 bytes_sent);

extern int cor_sock_sndbufavailable(struct conn *src_sock_lx);

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

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

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

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

extern int __init cor_sock_init1(void);

extern int __init cor_sock_init2(void);


static inline struct skb_procstate *skb_pstate(struct sk_buff *skb)
{
        return (struct skb_procstate *) &(skb->cb[0]);
}

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

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

static inline int qos_fastsend_allowed_announce(struct net_device *dev)
{
        int rc;
        struct qos_queue *q = get_queue(dev);

        if (q == 0)
                return 0;

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

        kref_put(&(q->ref), free_qos);

        return rc;
}

static inline int qos_fastsend_allowed_conn(struct conn *trgt_out_lx)
{
        struct qos_queue *q = trgt_out_lx->target.out.nb->queue;
        return atomic_read(&(q->cong_status)) < CONGSTATUS_CONNDATA;
}

static inline __u32 mss(struct neighbor *nb, __u32 l3overhead)
{
        __u32 mtu = (nb->dev->mtu > 4096) ? 4096 : nb->dev->mtu;
        return mtu - LL_RESERVED_SPACE(nb->dev) - l3overhead;
}

static inline __u32 mss_cmsg(struct neighbor *nb)
{
        return mss(nb, 7);
}

static inline __u32 mss_conndata(struct neighbor *nb)
{
        __u32 mss_tmp = mss(nb, 11);
        __u32 i;

        if (mss_tmp < 256)
                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 send_conndata_as_skb(struct neighbor *nb, __u32 size)
{
        return size >= mss_conndata(nb)/2;
}

static inline long 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 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 put_u64(char *dst, __u64 value)
{
        put_be64(dst, cpu_to_be64(value));
}

static inline void put_u48(char *dst, __u64 value)
{
        char *p_value = (char *) &value;

        value = cpu_to_be64(value);

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

static inline void 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 put_u32(char *dst, __u32 value)
{
        put_be32(dst, cpu_to_be32(value));
}

static inline void put_be16(char *dst, __be16 value)
{
        char *p_value = (char *) &value;
        dst[0] = p_value[0];
        dst[1] = p_value[1];
}

static inline void put_u16(char *dst, __u16 value)
{
        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 parse_be64(char *buf)
{
        __u64 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 parse_u64(char *buf)
{
        return be64_to_cpu(parse_be64(buf));
}

static inline __u64 parse_u48(char *ptr)
{
        __u64 ret = 0;

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

        return be64_to_cpu(ret);
}

static inline __be32 parse_be32(char *ptr)
{
        __u32 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 parse_u32(char *ptr)
{
        return be32_to_cpu(parse_be32(ptr));
}

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

        BUG_ON(ptr == 0);

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

        return ret;
}

static inline __u16 parse_u16(char *ptr)
{
        return be16_to_cpu(parse_be16(ptr));
}

static inline __u64 pull_u48(struct sk_buff *skb)
{
        return parse_u48(cor_pull_skb(skb, 6));
}

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

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

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

static inline __u8 pull_u8(struct sk_buff *skb)
{
        char *ptr = cor_pull_skb(skb, 1);
        BUG_ON(ptr == 0);
        return *ptr;
}

static inline int is_conn_in(struct conn *cn_l, struct neighbor *nb,
                __u32 conn_id)
{
        if (unlikely(unlikely(cn_l->sourcetype != SOURCE_IN) ||
                        unlikely(cn_l->source.in.nb != nb) ||
                        unlikely(cn_l->source.in.conn_id != conn_id) ||
                        unlikely(cn_l->isreset != 0)))
                return 0;
        return 1;
}

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

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

static inline void set_last_act(struct conn *src_in_l)
{
        unsigned long jiffies_tmp = jiffies;

        BUG_ON(src_in_l->sourcetype != SOURCE_IN);

        if (unlikely(time_after(jiffies_tmp,
                        src_in_l->source.in.jiffies_last_act +
                        HZ * CONN_ACTIVITY_UPDATEINTERVAL_SEC)))
                _set_last_act(src_in_l);
}

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

static inline __u32 buf_optlen(__u32 datalen)
{
        __u32 optlen = BUFLEN_MIN;
        while (optlen < datalen && optlen < PAGESIZE && optlen < BUFLEN_MAX)
                optlen = (optlen << 1);
        return optlen;
}


inline static void databuf_item_free(struct data_buf_item *item)
{
        if (item->type == DATABUF_BUF) {
                kfree(item->buf);
                kmem_cache_free(data_buf_item_slab, item);
        } else if (item->type == DATABUF_SKB) {
                struct sk_buff *skb = skb_from_pstate(container_of(item,
                                struct skb_procstate, funcstate.rcv.dbi));
                kfree_skb(skb);
        } else {
                BUG();
        }
}

static inline __u64 seqno_clean(__u64 seqno)
{
        return seqno & ((1LL << 48) - 1);
}

static inline int seqno_eq(__u64 seqno1, __u64 seqno2)
{
        seqno1 = seqno1 << 16;
        seqno2 = seqno2 << 16;
        return seqno1 == seqno2;
}

static inline int seqno_before(__u64 seqno1, __u64 seqno2)
{
        seqno1 = seqno1 << 16;
        seqno2 = seqno2 << 16;
        return (seqno1 - seqno2) >= (1LL << 63);
}

static inline int seqno_before_eq(__u64 seqno1, __u64 seqno2)
{
        return seqno_eq(seqno1, seqno2) || seqno_before(seqno1, seqno2);
}

static inline int seqno_after(__u64 seqno1, __u64 seqno2)
{
        return seqno_before_eq(seqno1, seqno2) ? 0 : 1;
}

static inline int seqno_after_eq(__u64 seqno1, __u64 seqno2)
{
        return 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 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 proc_cpacket */
static inline void flush_buf(struct conn *cn_lx)
{
        if (unlikely(cn_lx->targettype == TARGET_UNCONNECTED)) {
                proc_cpacket(cn_lx);
        } else if (cn_lx->targettype == TARGET_SOCK) {
                flush_sock(cn_lx);
        } else if (cn_lx->targettype == TARGET_OUT) {
                __u32 bytessent = 0;
                flush_out(cn_lx, &bytessent);
        } else if (unlikely(cn_lx->targettype == TARGET_DISCARD)) {
                databuf_ackdiscard(cn_lx);
        } else {
                BUG();
        }
}