first commit

[libutp-c.git] / utp.c

#define _GNU_SOURCE

#include <string.h>
#include <stdio.h>
#include <assert.h>
#include <string.h>
#include <stdlib.h>
#include <errno.h>
#include <limits.h> /* for UINT_MAX */

#include <StdAfx.h>

#include "utp.h"
#include "templates.h"

#ifdef WIN32
#include "win32_inet_ntop.h"

/* newer versions of MSVC define these in errno.h */
#ifndef ECONNRESET
#define ECONNRESET WSAECONNRESET
#define EMSGSIZE WSAEMSGSIZE
#define ECONNREFUSED WSAECONNREFUSED
#define ETIMEDOUT WSAETIMEDOUT
#endif
#endif

#ifdef POSIX
typedef struct sockaddr_storage SOCKADDR_STORAGE;
#endif /* POSIX */

/* number of bytes to increase max window size by, per RTT. This is
   scaled down linearly proportional to off_target. i.e. if all packets
   in one window have 0 delay, window size will increase by this number.
   Typically it's less. TCP increases one MSS per RTT, which is 1500 */
#define MAX_CWND_INCREASE_BYTES_PER_RTT 3000
#define CUR_DELAY_SIZE 3
/* experiments suggest that a clock skew of 10 ms per 325 seconds
   is not impossible. Reset delay_base every 13 minutes. The clock
   skew is dealt with by observing the delay base in the other
   direction, and adjusting our own upwards if the opposite direction
   delay base keeps going down */
#define DELAY_BASE_HISTORY 13
#define MAX_WINDOW_DECAY 100 /* ms */

#define REORDER_BUFFER_SIZE 32
#define REORDER_BUFFER_MAX_SIZE 511
#define OUTGOING_BUFFER_MAX_SIZE 511

#define PACKET_SIZE 350

/* this is the minimum max_window value. It can never drop below this */
#define MIN_WINDOW_SIZE 10

/* when window sizes are smaller than one packet_size, this
   will pace the packets to average at the given window size
   if it's not set, it will simply not send anything until
   there's a timeout */
#define USE_PACKET_PACING 1

/* if we receive 4 or more duplicate acks, we resend the packet
   that hasn't been acked yet */
#define DUPLICATE_ACKS_BEFORE_RESEND 3

#define DELAYED_ACK_BYTE_THRESHOLD 2400 /* bytes */
#define DELAYED_ACK_TIME_THRESHOLD 100 /* milliseconds */

#define RST_INFO_TIMEOUT 10000
#define RST_INFO_LIMIT 1000
/* 29 seconds determined from measuring many home NAT devices */
#define KEEPALIVE_INTERVAL 29000


#define SEQ_NR_MASK 0xFFFF
#define ACK_NR_MASK 0xFFFF

#define DIV_ROUND_UP(num, denom) ((num + denom - 1) / denom)

#include "utp_utils.h"
#include "utp_config.h"

#define LOG_UTP if (g_log_utp) utp_log
#define LOG_UTPV if (g_log_utp_verbose) utp_log

uint32 g_current_ms;

/* The totals are derived from the following data:
   45: IPv6 address including embedded IPv4 address
   11: Scope Id
    2: Brackets around IPv6 address when port is present
    6: Port (including colon)
    1: Terminating null byte */
char addrbuf[65];
char addrbuf2[65];
#define addrfmt(x, s) psa_fmt(x, s, sizeof(s))

#if (defined(__SVR4) && defined(__sun))
#pragma pack(1)
#else
#pragma pack(push,1)
#endif

/*----------------------------------------------------------------------------*/
/* packed socket address */
union PACKED_ATTRIBUTE psa_in {
        /* The values are always stored here in network byte order */
        byte _in6[16];                  /* IPv6 */
        uint16 _in6w[8];                /* IPv6, word based (for convenience) */
        uint32 _in6d[4];                /* Dword access */
        struct in6_addr _in6addr;       /* For convenience */
};

struct PACKED_ATTRIBUTE ALIGNED_ATTRIBUTE(4) psa {
        union psa_in _in;
        /* Host byte order */
        uint16 _port;
};

static byte psa_get_family(struct psa *psa)
{
        return (IN6_IS_ADDR_V4MAPPED(&psa->_in._in6addr) != 0) ? AF_INET : AF_INET6;
}

static bool psa_is_equal(struct psa *lhs, struct psa *rhs)
{
        if (rhs == lhs)
                return true;
        if (lhs->_port != rhs->_port)
                return false;
        return memcmp(&lhs->_in._in6[0], &rhs->_in._in6[0], sizeof(lhs->_in._in6)) == 0;
}

static bool psa_is_not_equal(struct psa *lhs, struct psa *rhs)
{
        return !(psa_is_equal(lhs, rhs));
}

static void psa_init(struct psa *psa, SOCKADDR_STORAGE *sa, socklen_t len)
{
        if (sa->ss_family == AF_INET) {
                struct sockaddr_in *sin;

                assert(len >= sizeof(struct sockaddr_in));

                psa->_in._in6w[0] = 0;
                psa->_in._in6w[1] = 0;
                psa->_in._in6w[2] = 0;
                psa->_in._in6w[3] = 0;
                psa->_in._in6w[4] = 0;
                psa->_in._in6w[5] = 0xffff;

                sin = (struct sockaddr_in*)sa;

                psa->_in._in6d[3] = sin->sin_addr.s_addr;
                psa->_port = ntohs(sin->sin_port);
        } else {
                struct sockaddr_in6 *sin6;

                assert(len >= sizeof(struct sockaddr_in6));

                sin6 = (struct sockaddr_in6*)sa;

                psa->_in._in6addr = sin6->sin6_addr;
                psa->_port = ntohs(sin6->sin6_port);
        }
}

/* structure SOCKADDR_STORAGE is passed by value */
/* len is defaulted to NULL */
static SOCKADDR_STORAGE psa_get_sockaddr_storage(struct psa *psa, socklen_t *len)
{
        SOCKADDR_STORAGE sa;
        byte family;

        family = psa_get_family(psa);
        if (family == AF_INET) {
                struct sockaddr_in *sin;

                sin = (struct sockaddr_in*)&sa;

                if (len) *len = sizeof(struct sockaddr_in);
                memset(sin, 0, sizeof(struct sockaddr_in));
                sin->sin_family = family;
                sin->sin_port = htons(psa->_port);
                sin->sin_addr.s_addr = psa->_in._in6d[3];
        } else {
                struct sockaddr_in6 *sin6;

                sin6 = (struct sockaddr_in6*)&sa;

                memset(sin6, 0, sizeof(struct sockaddr_in6));
                if (len) *len = sizeof(struct sockaddr_in6);
                sin6->sin6_family = family;
                sin6->sin6_addr = psa->_in._in6addr;
                sin6->sin6_port = htons(psa->_port);
        }
        return sa;
}

static str psa_fmt(struct psa *psa, str s, size_t len)
{
        byte family;
        str i;

        memset(s, 0, len);

        family = psa_get_family(psa);
        if (family == AF_INET) {
                inet_ntop(family, (uint32*)&psa->_in._in6d[3], s, len);
                i = s;
                while (*++i) {}
        } else {
                i = s;
                *i++ = '[';
                inet_ntop(family, (struct in6_addr*)&psa->_in._in6addr, i, len-1);
                while (*++i) {}
                *i++ = ']';
        }
        snprintf(i, len - (i-s), ":%u", psa->_port);
        return s;
}
/*----------------------------------------------------------------------------*/

struct PACKED_ATTRIBUTE RST_Info {
        struct psa addr;
        uint32 connid;
        uint32 timestamp;
        uint16 ack_nr;
};

/* these packet sizes are including the uTP header wich
   is either 20 or 23 bytes depending on version */
#define PACKET_SIZE_EMPTY_BUCKET 0
#define PACKET_SIZE_EMPTY 23
#define PACKET_SIZE_SMALL_BUCKET 1
#define PACKET_SIZE_SMALL 373
#define PACKET_SIZE_MID_BUCKET 2
#define PACKET_SIZE_MID 723
#define PACKET_SIZE_BIG_BUCKET 3
#define PACKET_SIZE_BIG 1400
#define PACKET_SIZE_HUGE_BUCKET 4

/* 24 bytes */
struct PACKED_ATTRIBUTE pf { /* Packet Format */
        /* connection ID */
        uint32_big connid;
        uint32_big tv_sec;
        uint32_big tv_usec;
        uint32_big reply_micro;
        /* receive window size in PACKET_SIZE chunks */
        byte windowsize;
        /* Type of the first extension header */
        byte ext;
        /* Flags */
        byte flags;
        /* Sequence number */
        uint16_big seq_nr;
        /* Acknowledgment number */
        uint16_big ack_nr;
};

struct PACKED_ATTRIBUTE pfa { /* Packet Format Ack */
        struct pf pf;
        byte ext_next;
        byte ext_len;
        byte acks[4];
};

struct PACKED_ATTRIBUTE pfe { /* Packet Format Extensions */
        struct pf pf;
        byte ext_next;
        byte ext_len;
        byte extensions[8];
};

/*----------------------------------------------------------------------------*/
/* 20 bytes */
struct PACKED_ATTRIBUTE pf1 { /* Packet Format V1 */
        /* packet_type (4 high bits) */
        /* protocol version (4 low bits) */
        byte ver_type;

        /* Type of the first extension header */
        byte ext;
        /* connection ID */
        uint16_big connid;
        uint32_big tv_usec;
        uint32_big reply_micro;
        /* receive window size in bytes */
        uint32_big windowsize;
        /* Sequence number */
        uint16_big seq_nr;
        /* Acknowledgment number */
        uint16_big ack_nr;
};

static byte pf1_version(struct pf1 *pf1)
{
        return pf1->ver_type & 0xf;
}

static byte pf1_type(struct pf1 *pf1)
{
        return pf1->ver_type >> 4;
}

static void pf1_version_set(struct pf1 *pf1, byte v)
{
        pf1->ver_type = (pf1->ver_type & 0xf0) | (v & 0x0f);
}

static void pf1_type_set(struct pf1 *pf1, byte t)
{
        pf1->ver_type = (pf1->ver_type & 0x0f) | (t << 4);
}
/*----------------------------------------------------------------------------*/

struct PACKED_ATTRIBUTE pfa1 { /* Packet Format Ack V1 */
        struct pf1 pf;
        byte ext_next;
        byte ext_len;
        byte acks[4];
};

struct PACKED_ATTRIBUTE pfe1 { /* Packet Format Extensions V1 */
        struct pf1 pf;
        byte ext_next;
        byte ext_len;
        byte extensions[8];
};

/* XXX:carefull many compilers do support pragma pack */
#if (defined(__SVR4) && defined(__sun))
#pragma pack(0)
#else
#pragma pack(pop)
#endif

/* TODO:should be cpp defines */
enum {
        ST_DATA = 0,    /* Data packet. */
        ST_FIN = 1,     /* Finalize the connection. This is the last packet. */
        ST_STATE = 2,   /* State packet. Used to transmit an ACK with no data. */
        ST_RESET = 3,   /* Terminate connection forcefully. */
        ST_SYN = 4,     /* Connect SYN */
        ST_NUM_STATES   /* used for bounds checking */
};

/* XXX: no lib global initialization function, then must keep it that way or replace it all */
static const cstr flagnames[] = {
        "ST_DATA","ST_FIN","ST_STATE","ST_RESET","ST_SYN"
};

/* TODO:should be cpp defines */
enum CONN_STATE {
        CS_IDLE = 0,
        CS_SYN_SENT = 1,
        CS_CONNECTED = 2,
        CS_CONNECTED_FULL = 3,
        CS_GOT_FIN = 4,
        CS_DESTROY_DELAY = 5,
        CS_FIN_SENT = 6,
        CS_RESET = 7,
        CS_DESTROY = 8
};

/* XXX: no lib global initialization function, then must keep it that way or replace it all */
static const cstr statenames[] = {
        "IDLE","SYN_SENT","CONNECTED","CONNECTED_FULL","GOT_FIN","DESTROY_DELAY","FIN_SENT","RESET","DESTROY"
};

struct op { /* Outgoing Packet */
        size_t length;
        size_t payload;
        uint64 time_sent; /* microseconds */
        uint transmissions;
        bool need_resend;
        byte data[1];
};

static void no_read(void *socket, const byte *bytes, size_t count)
{
        (void)socket;
        (void)bytes;
        (void)count;
}
static void no_write(void *socket, byte *bytes, size_t count)
{
        (void)socket;
        (void)bytes;
        (void)count;
}
static size_t no_rb_size(void *socket)
{
        (void)socket;
        return 0;
}
static void no_state(void *socket, int state)
{
        (void)socket;
        (void)state;
}
static void no_error(void *socket, int errcode)
{
        (void)socket;
        (void)errcode;
}
static void no_overhead(void *socket, bool send, size_t count, int type)
{
        (void)socket;
        (void)send;
        (void)count;
        (void)type;
}

static struct UTPFunctionTable zero_funcs = {
        &no_read,
        &no_write,
        &no_rb_size,
        &no_state,
        &no_error,
        &no_overhead,
};

/*----------------------------------------------------------------------------*/
struct scb {/* Sizable Circular Buffer */
        /* This is the mask. Since it's always a power of 2, adding 1 to this value will return the size. */
        size_t mask;
        /* This is the elements that the circular buffer points to */
        void **elements;
};

static void *scb_get(struct scb *scb, size_t i)
{
        assert(scb->elements);
        return scb->elements ? scb->elements[i & scb->mask] : NULL;
}

static void scb_put(struct scb *scb, size_t i, void *data)
{
        assert(scb->elements);
        scb->elements[i&scb->mask] = data;
}

/* Item contains the element we want to make space for
   index is the index in the list. */
static void scb_grow(struct scb *scb, size_t item, size_t index)
{
        size_t size;
        void **buf;
        size_t i;

        /* Figure out the new size. */
        size = scb->mask + 1;
        do size *= 2; while (index >= size);

        /* Allocate the new buffer */
        buf = calloc(size, sizeof(void*));

        size--;


        /* XXX: hope the resize op is rarely called and with little data */

        /* Copy elements from the old buffer to the new buffer */
        for (i = 0; i <= scb->mask; ++i) {
                buf[(item - index + i) & size] = scb_get(scb, item - index + i);
        }

        /* Swap to the newly allocated buffer */
        scb->mask = size;
        free(scb->elements);
        scb->elements = buf;
}

static void scb_ensure_size(struct scb *scb, size_t item, size_t index)
{
        if (index > scb->mask)
                scb_grow(scb, item, index);
}

static size_t scb_size(struct scb *scb)
{
        return scb->mask + 1; /* remember: power of 2 magic */
}
/*----------------------------------------------------------------------------*/

static struct UTPGlobalStats _global_stats;

/* compare if lhs is less than rhs, taking wrapping
   into account. if lhs is close to UINT_MAX and rhs
   is close to 0, lhs is assumed to have wrapped and
   considered smaller */
static bool wrapping_compare_less(uint32 lhs, uint32 rhs)
{
        uint32 dist_down;
        uint32 dist_up;

        /* distance walking from lhs to rhs, downwards */
        dist_down = lhs - rhs;
        /* distance walking from lhs to rhs, upwards */
        dist_up = rhs - lhs;

        /* if the distance walking up is shorter, lhs
           is less than rhs. If the distance walking down
           is shorter, then rhs is less than lhs */
        return dist_up < dist_down;
}

/*----------------------------------------------------------------------------*/
struct dh {/* Delay History */
        uint32 delay_base;

        /* this is the history of delay samples,
           normalized by using the delay_base. These
           values are always greater than 0 and measures
           the queuing delay in microseconds */
        uint32 cur_delay_hist[CUR_DELAY_SIZE];
        size_t cur_delay_idx;

        /* this is the history of delay_base. It's
           a number that doesn't have an absolute meaning
           only relative. It doesn't make sense to initialize
           it to anything other than values relative to
           what's been seen in the real world. */
        uint32 delay_base_hist[DELAY_BASE_HISTORY];
        size_t delay_base_idx;
        /* the time when we last stepped the delay_base_idx */
        uint32 delay_base_time;

        bool delay_base_initialized;
};

static void dh_clear(struct dh *dh)
{
        size_t i;

        dh->delay_base_initialized = false;
        dh->delay_base = 0;
        dh->cur_delay_idx = 0;
        dh->delay_base_idx = 0;
        dh->delay_base_time = g_current_ms;

        for (i = 0; i < CUR_DELAY_SIZE; i++) {
                dh->cur_delay_hist[i] = 0;
        }

        for (i = 0; i < DELAY_BASE_HISTORY; i++) {
                dh->delay_base_hist[i] = 0;
        }
}

static void dh_shift(struct dh *dh, uint32 offset)
{
        size_t i;

        /* the offset should never be "negative"
           assert(offset < 0x10000000); */

        /* increase all of our base delays by this amount
           this is used to take clock skew into account
           by observing the other side's changes in its base_delay */
        for (i = 0; i < DELAY_BASE_HISTORY; i++) {
                dh->delay_base_hist[i] += offset;
        }
        dh->delay_base += offset;
}

static void dh_add_sample(struct dh *dh, uint32 sample)
{
        /* The two clocks (in the two peers) are assumed not to
           progress at the exact same rate. They are assumed to be
           drifting, which causes the delay samples to contain
           a systematic error, either they are under-
           estimated or over-estimated. This is why we update the
           delay_base every two minutes, to adjust for this.

           This means the values will keep drifting and eventually wrap.
           We can cross the wrapping boundry in two directions, either
           going up, crossing the highest value, or going down, crossing 0.

           if the delay_base is close to the max value and sample actually
           wrapped on the other end we would see something like this:
           delay_base = 0xffffff00, sample = 0x00000400
           sample - delay_base = 0x500 which is the correct difference

           if the delay_base is instead close to 0, and we got an even lower
           sample (that will eventually update the delay_base), we may see
           something like this:
           delay_base = 0x00000400, sample = 0xffffff00
           sample - delay_base = 0xfffffb00
           this needs to be interpreted as a negative number and the actual
           recorded delay should be 0.

           It is important that all arithmetic that assume wrapping
           is done with unsigned intergers. Signed integers are not guaranteed
           to wrap the way unsigned integers do. At least GCC takes advantage
           of this relaxed rule and won't necessarily wrap signed ints.

           remove the clock offset and propagation delay.
           delay base is min of the sample and the current
           delay base. This min-operation is subject to wrapping
           and care needs to be taken to correctly choose the
           true minimum.

           specifically the problem case is when delay_base is very small
           and sample is very large (because it wrapped past zero), sample
           needs to be considered the smaller */

        uint32 delay;

        if (!dh->delay_base_initialized) {
                size_t i;

                /* delay_base being 0 suggests that we haven't initialized
                   it or its history with any real measurements yet. Initialize
                   everything with this sample. */
                for (i = 0; i < DELAY_BASE_HISTORY; ++i) {
                        /* if we don't have a value, set it to the current sample */
                        dh->delay_base_hist[i] = sample;
                        continue;
                }
                dh->delay_base = sample;
                dh->delay_base_initialized = true;
        }

        if (wrapping_compare_less(sample, dh->delay_base_hist[dh->delay_base_idx])) {
                /* sample is smaller than the current delay_base_hist entry
                   update it */
                dh->delay_base_hist[dh->delay_base_idx] = sample;
        }

        /* is sample lower than delay_base? If so, update delay_base */
        if (wrapping_compare_less(sample, dh->delay_base)) {
                /* sample is smaller than the current delay_base
                   update it */
                dh->delay_base = sample;
        }
        
        /* this operation may wrap, and is supposed to */
        delay = sample - dh->delay_base;
        /* sanity check. If this is triggered, something fishy is going on
           it means the measured sample was greater than 32 seconds! */
        /* assert(delay < 0x2000000); */

        dh->cur_delay_hist[dh->cur_delay_idx] = delay;
        dh->cur_delay_idx = (dh->cur_delay_idx + 1) % CUR_DELAY_SIZE;

        /* once every minute */
        if (g_current_ms - dh->delay_base_time > 60 * 1000) {
                size_t i;

                dh->delay_base_time = g_current_ms;
                dh->delay_base_idx = (dh->delay_base_idx + 1) % DELAY_BASE_HISTORY;
                /* clear up the new delay base history spot by initializing
                   it to the current sample, then update it  */
                dh->delay_base_hist[dh->delay_base_idx] = sample;
                dh->delay_base = dh->delay_base_hist[0];
                /* Assign the lowest delay in the last 2 minutes to delay_base */
                for (i = 0; i < DELAY_BASE_HISTORY; ++i) {
                        if (wrapping_compare_less(dh->delay_base_hist[i], dh->delay_base))
                                dh->delay_base = dh->delay_base_hist[i];
                }
        }
}

static uint32 dh_get_value(struct dh *dh)
{
        uint32 value;
        size_t i;

        value = UINT_MAX;
        for (i = 0; i < CUR_DELAY_SIZE; ++i) {
                value = uint32_min(dh->cur_delay_hist[i], value);
        }
        /* value could be UINT_MAX if we have no samples yet... */
        return value;
}
/*----------------------------------------------------------------------------*/

/*----------------------------------------------------------------------------*/
struct UTPSocket {
        struct psa addr;

        size_t idx;

        uint16 reorder_count;
        byte duplicate_ack;

        /* the number of bytes we've received but not acked yet */
        size_t bytes_since_ack;

        /* the number of packets in the send queue. Packets that haven't
           yet been sent count as well as packets marked as needing resend
           the oldest un-acked packet in the send queue is seq_nr - cur_window_packets */
        uint16 cur_window_packets;

        /* how much of the window is used, number of bytes in-flight
           packets that have not yet been sent do not count, packets
           that are marked as needing to be re-sent (due to a timeout)
           don't count either */
        size_t cur_window;
        /* maximum window size, in bytes */
        size_t max_window;
        /* SO_SNDBUF setting, in bytes */
        size_t opt_sndbuf;
        /* SO_RCVBUF setting, in bytes */
        size_t opt_rcvbuf;

        /* Is a FIN packet in the reassembly buffer? */
        bool got_fin;
        /* Timeout procedure */
        bool fast_timeout;

        /* max receive window for other end, in bytes */
        size_t max_window_user;
        /* 0 = original uTP header, 1 = second revision */
        byte version;
        enum CONN_STATE state;
        /* TickCount when we last decayed window (wraps) */
        int32 last_rwin_decay;

        /* the sequence number of the FIN packet. This field is only set
           when we have received a FIN, and the flag field has the FIN flag set.
           it is used to know when it is safe to destroy the socket, we must have
           received all packets up to this sequence number first. */
        uint16 eof_pkt;

        /* All sequence numbers up to including this have been properly received
           by us */
        uint16 ack_nr;
        /* This is the sequence number for the next packet to be sent. */
        uint16 seq_nr;

        uint16 timeout_seq_nr;

        /* This is the sequence number of the next packet we're allowed to
           do a fast resend with. This makes sure we only do a fast-resend
           once per packet. We can resend the packet with this sequence number
           or any later packet (with a higher sequence number). */
        uint16 fast_resend_seq_nr;

        uint32 reply_micro;

        /* the time when we need to send another ack. If there's
           nothing to ack, this is a very large number */
        uint32 ack_time;

        uint32 last_got_packet;
        uint32 last_sent_packet;
        uint32 last_measured_delay;
        uint32 last_maxed_out_window;

        /* the last time we added send quota to the connection
           when adding send quota, this is subtracted from the
           current time multiplied by max_window / rtt
           which is the current allowed send rate. */
        int32 last_send_quota;

        /* the number of bytes we are allowed to send on
           this connection. If this is more than one packet
           size when we run out of data to send, it is clamped
           to the packet size
           this value is multiplied by 100 in order to get
           higher accuracy when dealing with low rates */
        int32 send_quota;

        SendToProc *send_to_proc;
        void *send_to_userdata;
        struct UTPFunctionTable func;
        void *userdata;

        /* Round trip time */
        uint rtt;
        /* Round trip time variance */
        uint rtt_var;
        /* Round trip timeout */
        uint rto;
        struct dh rtt_hist;
        uint retransmit_timeout;
        /* The RTO timer will timeout here. */
        uint rto_timeout;
        /* When the window size is set to zero, start this timer. It will send a new packet every 30secs. */
        uint32 zerowindow_time;

        uint32 conn_seed;
        /* Connection ID for packets I receive */
        uint32 conn_id_recv;
        /* Connection ID for packets I send */
        uint32 conn_id_send;
        /* Last rcv window we advertised, in bytes */
        size_t last_rcv_win;

        struct dh our_hist;
        struct dh their_hist;

        /* extension bytes from SYN packet */
        byte extensions[8];

        struct scb inbuf;
        struct scb outbuf;

#ifdef _DEBUG
        /* Public stats, returned by UTP_GetStats().  See utp.h */
        struct UTPStats _stats;
#endif /* _DEBUG */
};

static size_t us_get_udp_mtu(struct UTPSocket *us)
{
        socklen_t len;
        SOCKADDR_STORAGE sa;

        sa = psa_get_sockaddr_storage(&us->addr, &len);
        return UTP_GetUDPMTU((struct sockaddr *)&sa, len);
}

/* returns the max number of bytes of payload the uTP
   connection is allowed to send */
static size_t us_get_packet_size(struct UTPSocket *us)
{
        int header_size;
        size_t mtu;

        header_size = us->version == 1 ? sizeof(struct pf1) : sizeof(struct pf);

        mtu = us_get_udp_mtu(us);

        if (DYNAMIC_PACKET_SIZE_ENABLED) {
                SOCKADDR_STORAGE sa;
                size_t max_packet_size;

                sa = psa_get_sockaddr_storage(&us->addr, NULL);
                max_packet_size = UTP_sockaddr_GetPacketSize((struct sockaddr*)&sa);
                return size_t_min(mtu - header_size, max_packet_size);
        }
        else
                return mtu - header_size;
}

/* Calculates the current receive window */
static size_t us_get_rcv_window(struct UTPSocket *us)
{
        size_t numbuf;

        /* If we don't have a connection (such as during connection
           establishment, always act as if we have an empty buffer). */
        if (!us->userdata)
                return us->opt_rcvbuf;

        /* Trim window down according to what's already in buffer. */
        numbuf = us->func.get_rb_size(us->userdata);
        assert((int)numbuf >= 0);
        return us->opt_rcvbuf > numbuf ? us->opt_rcvbuf - numbuf : 0;
}

/* Test if we're ready to decay max_window
   XXX this breaks when spaced by > INT_MAX/2, which is 49
   days; the failure mode in that case is we do an extra decay
   or fail to do one when we really shouldn't. */
static bool us_can_decay_win(struct UTPSocket *us, int32 msec)
{
        return msec - us->last_rwin_decay >= MAX_WINDOW_DECAY;
}

/* If we can, decay max window, returns true if we actually did so */
static void us_maybe_decay_win(struct UTPSocket *us)
{
        if (us_can_decay_win(us, g_current_ms)) {
                /* TCP uses 0.5 */
                us->max_window = (size_t)(us->max_window * .5);
                us->last_rwin_decay = g_current_ms;
                if (us->max_window < MIN_WINDOW_SIZE)
                        us->max_window = MIN_WINDOW_SIZE;
        }
}

static size_t us_get_header_size(struct UTPSocket *us)
{
        return (us->version ? sizeof(struct pf1) : sizeof(struct pf));
}

static size_t us_get_header_extensions_size(struct UTPSocket *us)
{
        return (us->version ? sizeof(struct pfe1) : sizeof(struct pfe));
}

static void us_sent_ack(struct UTPSocket *us)
{
        us->ack_time = g_current_ms + 0x70000000;
        us->bytes_since_ack = 0;
}

static size_t us_get_udp_overhead(struct UTPSocket *us)
{
        socklen_t len;
        SOCKADDR_STORAGE sa;

        sa = psa_get_sockaddr_storage(&us->addr, &len);
        return UTP_GetUDPOverhead((struct sockaddr *)&sa, len);
}

#if 0
/* we keep this function around but it's not used */
static uint64 us_get_global_utp_bytes_sent(struct UTPSocket *us)
{
        socklen_t len;
        SOCKADDR_STORAGE sa;

        sa = psa_get_sockaddr_storage(&us->addr, &len);
        return UTP_GetGlobalUTPBytesSent((struct sockaddr *)&sa, len);
}
#endif

static size_t us_get_overhead(struct UTPSocket *us)
{
        return us_get_udp_overhead(us) + us_get_header_size(us);
}
/*----------------------------------------------------------------------------*/

struct array g_rst_info = {NULL, 0, 0, sizeof(struct RST_Info)};
struct array g_utp_sockets = {NULL, 0, 0, sizeof(struct UTPSocket*)};

static void UTP_RegisterSentPacket(size_t length) {
        if (length <= PACKET_SIZE_MID) {
                if (length <= PACKET_SIZE_EMPTY) {
                        _global_stats._nraw_send[PACKET_SIZE_EMPTY_BUCKET]++;
                } else if (length <= PACKET_SIZE_SMALL) {
                        _global_stats._nraw_send[PACKET_SIZE_SMALL_BUCKET]++;
                } else
                        _global_stats._nraw_send[PACKET_SIZE_MID_BUCKET]++;
        } else {
                if (length <= PACKET_SIZE_BIG) {
                        _global_stats._nraw_send[PACKET_SIZE_BIG_BUCKET]++;
                } else
                        _global_stats._nraw_send[PACKET_SIZE_HUGE_BUCKET]++;
        }
}

static void send_to_addr(SendToProc *send_to_proc, void *send_to_userdata, byte *p, size_t len, struct psa *addr)
{
        socklen_t tolen;
        SOCKADDR_STORAGE to;

        to = psa_get_sockaddr_storage(addr, &tolen);
        UTP_RegisterSentPacket(len);
        send_to_proc(send_to_userdata, p, len, (const struct sockaddr *)&to, tolen);
}

static void us_send_data(struct UTPSocket *us, struct pf *b, size_t length, enum bandwidth_type_t type)
{
        uint64 time;
        struct pf1* b1;
#if g_log_utp_verbose
        int flags;
        uint16 seq_nr;
        uint16 ack_nr;
#endif

        /* time stamp this packet with local time, the stamp goes into
           the header of every packet at the 8th byte for 8 bytes :
           two integers, check packet.h for more */
        time = UTP_GetMicroseconds();

        b1 = (struct pf1*)b;
        if (us->version == 0) {
                b->tv_sec = htonl((uint32)(time / 1000000));
                b->tv_usec = htonl(time % 1000000);
                b->reply_micro = htonl(us->reply_micro);
        } else {
                b1->tv_usec = htonl((uint32)time);
                b1->reply_micro = htonl(us->reply_micro);
        }

        us->last_sent_packet = g_current_ms;

#ifdef _DEBUG
        us->_stats._nbytes_xmit += length;
        ++us->_stats._nxmit;
#endif
        if (us->userdata) {
                size_t n;

                if (type == payload_bandwidth) {
                        /* if this packet carries payload, just
                           count the header as overhead */
                        type = header_overhead;
                        n = us_get_overhead(us);
                } else 
                        n = length + us_get_udp_overhead(us);
                us->func.on_overhead(us->userdata, true, n, type);
        }
#if g_log_utp_verbose
        flags = us->version == 0 ? b->flags : pf1_type(b1);
        seq_nr = us->version == 0 ? ntohs(b->seq_nr) : ntohs(b1->seq_nr);
        ack_nr = us->version == 0 ? ntohs(b->ack_nr) : ntohs(b1->ack_nr);
        LOG_UTPV("0x%08x: send %s len:%u id:%u timestamp:"I64u" reply_micro:%u flags:%s seq_nr:%u ack_nr:%u", us, addrfmt(&us->addr, addrbuf), (uint)length, us->conn_id_send, time, us->reply_micro, flagnames[flags], seq_nr, ack_nr);
#endif
        send_to_addr(us->send_to_proc, us->send_to_userdata, (byte*)b, length, &us->addr);
}

/* XXX:carefull, synack defaults to false */
static void us_send_ack(struct UTPSocket *us, bool synack)
{
        /* all following structs fit in PacketFormatExtensions */
        struct pfe pfe;
        struct pfe1 *pfe1;
        struct pfa *pfa;
        struct pfa1 *pfa1;
        size_t len;

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

        pfe1 = (struct pfe1*)&pfe;
        pfa = (struct pfa*)pfe1;
        pfa1 = (struct pfa1*)pfe1;

        us->last_rcv_win = us_get_rcv_window(us);
        if (us->version == 0) {
                pfa->pf.connid = htonl(us->conn_id_send);
                pfa->pf.ack_nr = htons(us->ack_nr);
                pfa->pf.seq_nr = htons(us->seq_nr);
                pfa->pf.flags = ST_STATE;
                pfa->pf.ext = 0;
                pfa->pf.windowsize = (byte)DIV_ROUND_UP(us->last_rcv_win, PACKET_SIZE);
                len = sizeof(struct pf);
        } else {
                pf1_version_set(&pfa1->pf, 1);
                pf1_type_set(&pfa1->pf, ST_STATE);
                pfa1->pf.ext = 0;
                pfa1->pf.connid = htons((uint16)us->conn_id_send);
                pfa1->pf.ack_nr = htons(us->ack_nr);
                pfa1->pf.seq_nr = htons(us->seq_nr);
                pfa1->pf.windowsize = htonl((uint32)us->last_rcv_win);
                len = sizeof(struct pf1);
        }

        /* we never need to send EACK for connections
           that are shutting down */
        if (us->reorder_count != 0 && us->state < CS_GOT_FIN) {
                uint m;
                size_t window;
                size_t i;

                /* if reorder count > 0, send an EACK.
                   reorder count should always be 0
                   for synacks, so this should not be
                   as synack */
                assert(!synack);
                if (us->version == 0) {
                        pfa->pf.ext = 1;
                        pfa->ext_next = 0;
                        pfa->ext_len = 4;
                } else {
                        pfa1->pf.ext = 1;
                        pfa1->ext_next = 0;
                        pfa1->ext_len = 4;
                }

                m = 0;

                /* reorder count should only be non-zero
                   if the packet ack_nr + 1 has not yet
                   been received */
                assert(scb_get(&us->inbuf, us->ack_nr + 1) == NULL);
                window = size_t_min(14 + 16, scb_size(&us->inbuf));
                /* Generate bit mask of segments received. */
                for (i = 0; i < window; ++i) {
                        if (scb_get(&us->inbuf, us->ack_nr + i + 2) != NULL) {
                                m |= 1 << i;
                                LOG_UTPV("0x%08x: EACK packet [%u]", us, us->ack_nr + i + 2);
                        }
                }
                if (us->version == 0) {
                        pfa->acks[0] = (byte)m;
                        pfa->acks[1] = (byte)(m >> 8);
                        pfa->acks[2] = (byte)(m >> 16);
                        pfa->acks[3] = (byte)(m >> 24);
                } else {
                        pfa1->acks[0] = (byte)m;
                        pfa1->acks[1] = (byte)(m >> 8);
                        pfa1->acks[2] = (byte)(m >> 16);
                        pfa1->acks[3] = (byte)(m >> 24);
                }
                len += 4 + 2;
                LOG_UTPV("0x%08x: Sending EACK %u [%u] bits:[%032b]", us, us->ack_nr, us->conn_id_send, m);
        } else if (synack) {
                /* we only send "extensions" in response to SYN
                   and the reorder count is 0 in that state */

                LOG_UTPV("0x%08x: Sending ACK %u [%u] with extension bits", us, us->ack_nr, us->conn_id_send);
                if (us->version == 0) {
                        pfe.pf.ext = 2;
                        pfe.ext_next = 0;
                        pfe.ext_len = 8;
                        memset(&pfe.extensions[0], 0, 8);
                } else {
                        pfe1->pf.ext = 2;
                        pfe1->ext_next = 0;
                        pfe1->ext_len = 8;
                        memset(&pfe1->extensions[0], 0, 8);
                }
                len += 8 + 2;
        } else {
                LOG_UTPV("0x%08x: Sending ACK %u [%u]", us, us->ack_nr, us->conn_id_send);
        }

        us_sent_ack(us);
        us_send_data(us, (struct pf*)&pfe, len, ack_overhead);
}

static void us_send_keep_alive(struct UTPSocket *us)
{
        us->ack_nr--;
        LOG_UTPV("0x%08x: Sending KeepAlive ACK %u [%u]", us, us->ack_nr, us->conn_id_send);
        us_send_ack(us, false);
        us->ack_nr++;
}

/* XXX: static class member function */
static void us_send_rst(SendToProc *send_to_proc, void *send_to_userdata, struct psa *addr, uint32 conn_id_send, uint16 ack_nr, uint16 seq_nr, byte version)
{
        struct pf pf;           /* 23 bytes */
        struct pf1 *pf1;        /* 20 bytes */
        size_t len;

        memset(&pf, 0, sizeof(pf));
        pf1 = (struct pf1*)&pf;

        if (version == 0) {
                pf.connid = htonl(conn_id_send);
                pf.ack_nr = htons(ack_nr);
                pf.seq_nr = htons(seq_nr);
                pf.flags = ST_RESET;
                pf.ext = 0;
                pf.windowsize = 0;
                len = sizeof(pf);
        } else {
                pf1_version_set(pf1, 1);
                pf1_type_set(pf1, ST_RESET);
                pf1->ext = 0;
                pf1->connid = htons((uint16)conn_id_send);
                pf1->ack_nr = htons(ack_nr);
                pf1->seq_nr = htons(seq_nr);
                pf1->windowsize = 0;
                len = sizeof(*pf1);
        }

        LOG_UTPV("%s: Sending RST id:%u seq_nr:%u ack_nr:%u", addrfmt(addr, addrbuf), conn_id_send, seq_nr, ack_nr);
        LOG_UTPV("send %s len:%u id:%u", addrfmt(addr, addrbuf), (uint)len, conn_id_send);
        send_to_addr(send_to_proc, send_to_userdata, (byte*)pf1, len, addr);
}

static void us_send_packet(struct UTPSocket *us, struct op *pkt)
{
        size_t max_send;
        size_t packet_size;
        struct pf1* p1;
        struct pf* p;

        /* only count against the quota the first time we
           send the packet. Don't enforce quota when closing
           a socket. Only enforce the quota when we're sending
           at slow rates (max window < packet size) */
        max_send = size_t_min3(us->max_window, us->opt_sndbuf, us->max_window_user);

        if (pkt->transmissions == 0 || pkt->need_resend)
                us->cur_window += pkt->payload;

        packet_size = us_get_packet_size(us);
        if (pkt->transmissions == 0 && max_send < packet_size) {
                assert(us->state == CS_FIN_SENT || (int32)pkt->payload <= us->send_quota / 100);
                us->send_quota = us->send_quota - (int32)(pkt->payload * 100);
        }

        pkt->need_resend = false;

        p1 = (struct pf1*)pkt->data;
        p = (struct pf*)pkt->data;
        if (us->version == 0) {
                p->ack_nr = htons(us->ack_nr);
        } else {
                p1->ack_nr = htons(us->ack_nr);
        }
        pkt->time_sent = UTP_GetMicroseconds();
        pkt->transmissions++;
        us_sent_ack(us);
        us_send_data(us, (struct pf*)pkt->data, pkt->length, (us->state == CS_SYN_SENT) ? connect_overhead : (pkt->transmissions == 1) ? payload_bandwidth : retransmit_overhead);
}

static bool us_is_writable(struct UTPSocket *us, size_t to_write)
{
        size_t max_send;
        size_t packet_size;

        /* return true if it's OK to stuff another packet into the
           outgoing queue. Since we may be using packet pacing, we
           might not actually send the packet right away to affect the
           cur_window. The only thing that happens when we add another
           packet is that cur_window_packets is increased. */
        max_send = size_t_min3(us->max_window, us->opt_sndbuf, us->max_window_user);

        packet_size = us_get_packet_size(us);

        if (us->cur_window + packet_size >= us->max_window)
                us->last_maxed_out_window = g_current_ms;

        /* if we don't have enough quota, we can't write regardless */
        if (USE_PACKET_PACING)
                if (us->send_quota / 100 < (int32)to_write)
                        return false;

        /* subtract one to save space for the FIN packet */
        if (us->cur_window_packets >= OUTGOING_BUFFER_MAX_SIZE - 1)
                return false;

        /* if sending another packet would not make the window exceed
           the max_window, we can write */
        if (us->cur_window + packet_size <= max_send)
                return true;

        /* if the window size is less than a packet, and we have enough
           quota to send a packet, we can write, even though it would
           make the window exceed the max size
           the last condition is needed to not put too many packets
           in the send buffer. cur_window isn't updated until we flush
           the send buffer, so we need to take the number of packets
           into account */
        if (USE_PACKET_PACING)
                if (us->max_window < to_write && us->cur_window < us->max_window && us->cur_window_packets == 0)
                        return true;
        return false;
}

static bool us_flush_packets(struct UTPSocket *us)
{
        size_t packet_size;
        uint16 i;

        packet_size = us_get_packet_size(us);

        /* send packets that are waiting on the pacer to be sent
           i has to be an unsigned 16 bit counter to wrap correctly
           signed types are not guaranteed to wrap the way you expect */
        for (i = us->seq_nr - us->cur_window_packets; i != us->seq_nr; ++i) {
                struct op *pkt;

                pkt = (struct op*)scb_get(&us->outbuf, i);
                if (pkt == 0 || (pkt->transmissions > 0 && pkt->need_resend == false))
                        continue;
                /* have we run out of quota? */
                if (!us_is_writable(us, pkt->payload))
                        return true;

                /* Nagle check
                  don't send the last packet if we have one packet in-flight
                  and the current packet is still smaller than packet_size. */
                if (i != ((us->seq_nr - 1) & ACK_NR_MASK) || us->cur_window_packets == 1 || pkt->payload >= packet_size) {
                        us_send_packet(us, pkt);

                        /* No need to send another ack if there is nothing to reorder. */
                        if (us->reorder_count == 0)
                                us_sent_ack(us);
                }
        }
        return false;
}

static void us_write_outgoing_packet(struct UTPSocket *us, size_t payload, uint flags)
{
        size_t packet_size;

        /* Setup initial timeout timer */
        if (us->cur_window_packets == 0) {
                us->retransmit_timeout = us->rto;
                us->rto_timeout = g_current_ms + us->retransmit_timeout;
                assert(us->cur_window == 0);
        }

        packet_size = us_get_packet_size(us);
        do {
                size_t added;
                struct op *pkt;
                size_t header_size;
                bool append;
                struct pf* p;
                struct pf1* p1;

                assert(us->cur_window_packets < OUTGOING_BUFFER_MAX_SIZE);
                assert(flags == ST_DATA || flags == ST_FIN);

                added = 0;
                pkt = NULL;
                
                if (us->cur_window_packets > 0)
                        pkt = (struct op*)scb_get(&us->outbuf, us->seq_nr - 1);

                header_size = us_get_header_size(us);
                append = true;

                /* if there's any room left in the last packet in the window
                   and it hasn't been sent yet, fill that frame first */
                if (payload && pkt && !pkt->transmissions && pkt->payload < packet_size) {
                        /* Use the previous unsent packet */
                        added = size_t_min(payload + pkt->payload, size_t_max(packet_size, pkt->payload)) - pkt->payload;
                        pkt = realloc(pkt, (sizeof(struct op) - 1) + header_size + pkt->payload + added);
                        scb_put(&us->outbuf, us->seq_nr - 1, pkt);
                        append = false;
                        assert(!pkt->need_resend);
                } else {
                        /* Create the packet to send. */
                        added = payload;
                        pkt = malloc((sizeof(struct op) - 1) + header_size + added);
                        pkt->payload = 0;
                        pkt->transmissions = 0;
                        pkt->need_resend = false;
                }

                if (added)
                        /* Fill it with data from the upper layer. */
                        us->func.on_write(us->userdata, pkt->data + header_size + pkt->payload, added);

                pkt->payload += added;
                pkt->length = header_size + pkt->payload;

                us->last_rcv_win = us_get_rcv_window(us);

                p = (struct pf*)pkt->data;
                p1 = (struct pf1*)pkt->data;
                if (us->version == 0) {
                        p->connid = htonl(us->conn_id_send);
                        p->ext = 0;
                        p->windowsize = (byte)DIV_ROUND_UP(us->last_rcv_win, PACKET_SIZE);
                        p->ack_nr = htons(us->ack_nr);
                        p->flags = flags;
                } else {
                        pf1_version_set(p1, 1);
                        pf1_type_set(p1, flags);
                        p1->ext = 0;
                        p1->connid = htons((uint16)us->conn_id_send);
                        p1->windowsize = htonl((uint32)us->last_rcv_win);
                        p1->ack_nr = htons(us->ack_nr);
                }

                if (append) {
                        /* Remember the message in the outgoing queue. */
                        scb_ensure_size(&us->outbuf, us->seq_nr, us->cur_window_packets);
                        scb_put(&us->outbuf, us->seq_nr, pkt);
                        if (us->version == 0)
                                p->seq_nr = htons(us->seq_nr);
                        else
                                p1->seq_nr = htons(us->seq_nr);
                        us->seq_nr++;
                        us->cur_window_packets++;
                }
                payload -= added;

        } while (payload);

        us_flush_packets(us);
}

static void us_update_send_quota(struct UTPSocket *us)
{
        int dt;
        size_t add;

        dt = g_current_ms - us->last_send_quota;

        if (dt == 0)
                return;
        us->last_send_quota = g_current_ms;
        add = us->max_window * dt * 100 / (us->rtt_hist.delay_base?us->rtt_hist.delay_base:50);
        if (add > us->max_window * 100 && add > MAX_CWND_INCREASE_BYTES_PER_RTT * 100)
                add = us->max_window;
        us->send_quota += (int32)add;
/*      LOG_UTPV("0x%08x: UTPSocket::update_send_quota dt:%d rtt:%u max_window:%u quota:%d", this, dt, rtt, (uint)max_window, send_quota / 100);*/
}

#ifdef _DEBUG
static void us_check_invariant(struct UTPSocket *us)
{
        size_t outstanding_bytes;
        int i;

        if (us->reorder_count > 0)
                assert(scb_get(&us->inbuf, us->ack_nr + 1) == NULL);

        outstanding_bytes = 0;
        for (i = 0; i < us->cur_window_packets; ++i) {
                struct op *pkt;

                pkt = (struct op*)scb_get(&us->outbuf, us->seq_nr - i - 1);
                if (pkt == 0 || pkt->transmissions == 0 || pkt->need_resend)
                        continue;
                outstanding_bytes += pkt->payload;
        }
        assert(outstanding_bytes == us->cur_window);
}
#endif

static void us_check_timeouts(struct UTPSocket *us)
{
        int32 limit;
#ifdef _DEBUG
        us_check_invariant(us);
#endif

        /* this invariant should always be true */
        assert(us->cur_window_packets == 0 || scb_get(&us->outbuf, us->seq_nr - us->cur_window_packets));

        LOG_UTPV("0x%08x: CheckTimeouts timeout:%d max_window:%u cur_window:%u quota:%d " "state:%s cur_window_packets:%u bytes_since_ack:%u ack_time:%d", us, (int)(us->rto_timeout - g_current_ms), (uint)us->max_window, (uint)us->cur_window, us->send_quota / 100, statenames[us->state], us->cur_window_packets, (uint)us->bytes_since_ack, (int)(g_current_ms - us->ack_time));

        us_update_send_quota(us);
        us_flush_packets(us);

        if (USE_PACKET_PACING) {
                /* In case the new send quota made it possible to send another packet
                   Mark the socket as writable. If we don't use pacing, the send
                   quota does not affect if the socket is writeable
                   if we don't use packet pacing, the writable event is triggered
                   whenever the cur_window falls below the max_window, so we don't
                   need this check then */
                if (us->state == CS_CONNECTED_FULL && us_is_writable(us, us_get_packet_size(us))) {
                        us->state = CS_CONNECTED;
                        LOG_UTPV("0x%08x: Socket writable. max_window:%u cur_window:%u quota:%d packet_size:%u", us, (uint)us->max_window, (uint)us->cur_window, us->send_quota / 100, (uint)us_get_packet_size(us));
                        us->func.on_state(us->userdata, UTP_STATE_WRITABLE);
                }
        }

        switch (us->state) {
        case CS_SYN_SENT:
        case CS_CONNECTED_FULL:
        case CS_CONNECTED:
        case CS_FIN_SENT: {

                /* Reset max window... */
                if ((int)(g_current_ms - us->zerowindow_time) >= 0 && us->max_window_user == 0)
                        us->max_window_user = PACKET_SIZE;

                if ((int)(g_current_ms - us->rto_timeout) >= 0 && (!(USE_PACKET_PACING) || us->cur_window_packets > 0) && us->rto_timeout > 0) {
                        uint new_timeout;
                        int i;

                        /*
                        OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - cur_window_packets);
                        
                        // If there were a lot of retransmissions, force recomputation of round trip time
                        if (pkt->transmissions >= 4)
                                rtt = 0;
                        */

                        /* Increase RTO */
                        new_timeout = us->retransmit_timeout * 2;
                        if (new_timeout >= 30000 || (us->state == CS_SYN_SENT && new_timeout > 6000)) {
                                /* more than 30 seconds with no reply. kill it.
                                   if we haven't even connected yet, give up sooner. 6 seconds
                                   means 2 tries at the following timeouts: 3, 6 seconds */
                                if (us->state == CS_FIN_SENT)
                                        us->state = CS_DESTROY;
                                else
                                        us->state = CS_RESET;
                                us->func.on_error(us->userdata, ETIMEDOUT);
                                goto getout;
                        }

                        us->retransmit_timeout = new_timeout;
                        us->rto_timeout = g_current_ms + new_timeout;

                        /* On Timeout */
                        us->duplicate_ack = 0;

                        /* rate = min_rate */
                        us->max_window = us_get_packet_size(us);
                        us->send_quota = int32_max((int32)us->max_window * 100, us->send_quota);

                        /* every packet should be considered lost */
                        for (i = 0; i < us->cur_window_packets; ++i) {
                                struct op *pkt;

                                pkt = (struct op*)scb_get(&us->outbuf, us->seq_nr - i - 1);
                                if (pkt == 0 || pkt->transmissions == 0 || pkt->need_resend)
                                        continue;
                                pkt->need_resend = true;
                                assert(us->cur_window >= pkt->payload);
                                us->cur_window -= pkt->payload;
                        }

                        /* used in parse_log.py */
                        LOG_UTP("0x%08x: Packet timeout. Resend. seq_nr:%u. timeout:%u max_window:%u", us, us->seq_nr - us->cur_window_packets, us->retransmit_timeout, (uint)us->max_window);

                        us->fast_timeout = true;
                        us->timeout_seq_nr = us->seq_nr;

                        if (us->cur_window_packets > 0) {
                                struct op *pkt;

                                pkt = (struct op*)scb_get(&us->outbuf, us->seq_nr - us->cur_window_packets);
                                assert(pkt);
                                us->send_quota = int32_max((int32)pkt->length * 100, us->send_quota);

                                /* Re-send the packet. */
                                us_send_packet(us, pkt);
                        }
                }

                /* Mark the socket as writable */
                if (us->state == CS_CONNECTED_FULL && us_is_writable(us, us_get_packet_size(us))) {
                        us->state = CS_CONNECTED;
                        LOG_UTPV("0x%08x: Socket writable. max_window:%u cur_window:%u quota:%d packet_size:%u", us, (uint)us->max_window, (uint)us->cur_window, us->send_quota / 100, (uint)us_get_packet_size(us));
                        us->func.on_state(us->userdata, UTP_STATE_WRITABLE);
                }

                if (us->state >= CS_CONNECTED && us->state <= CS_FIN_SENT) {
                        /* Send acknowledgment packets periodically, or when the threshold is reached */
                        if (us->bytes_since_ack > DELAYED_ACK_BYTE_THRESHOLD || (int)(g_current_ms - us->ack_time) >= 0)
                                us_send_ack(us, false);

                        if ((int)(g_current_ms - us->last_sent_packet) >= KEEPALIVE_INTERVAL)
                                us_send_keep_alive(us);
                }
                break;
        }

        /* Close? */
        case CS_GOT_FIN:
        case CS_DESTROY_DELAY:
                if ((int)(g_current_ms - us->rto_timeout) >= 0) {
                        us->state = (us->state == CS_DESTROY_DELAY) ? CS_DESTROY : CS_RESET;
                        if (us->cur_window_packets > 0 && us->userdata)
                                us->func.on_error(us->userdata, ECONNRESET);
                }
                break;
        /* prevent warning */
        case CS_IDLE:
        case CS_RESET:
        case CS_DESTROY:
                break;
        }

        getout:

        /* make sure we don't accumulate quota when we don't have
           anything to send */
        limit = int32_max((int32)us->max_window / 2, 5 * (int32)us_get_packet_size(us)) * 100;
        if (us->send_quota > limit) us->send_quota = limit;
}

/* returns:
   0: the packet was acked.
   1: it means that the packet had already been acked
   2: the packet has not been sent yet */
static int us_ack_packet(struct UTPSocket *us, uint16 seq)
{
        struct op *pkt;

        pkt = (struct op*)scb_get(&us->outbuf, seq);

        /* the packet has already been acked (or not sent) */
        if (pkt == NULL) {
                LOG_UTPV("0x%08x: got ack for:%u (already acked, or never sent)", us, seq);
                return 1;
        }

        /* can't ack packets that haven't been sent yet! */
        if (pkt->transmissions == 0) {
                LOG_UTPV("0x%08x: got ack for:%u (never sent, pkt_size:%u need_resend:%u)", us, seq, (uint)pkt->payload, pkt->need_resend);
                return 2;
        }

        LOG_UTPV("0x%08x: got ack for:%u (pkt_size:%u need_resend:%u)", us, seq, (uint)pkt->payload, pkt->need_resend);

        scb_put(&us->outbuf, seq, NULL);

        /* if we never re-sent the packet, update the RTT estimate */
        if (pkt->transmissions == 1) {
                uint32 ertt;

                /* Estimate the round trip time. */
                ertt = (uint32)((UTP_GetMicroseconds() - pkt->time_sent) / 1000);

                if (us->rtt == 0) {
                        /* First round trip time sample */
                        us->rtt = ertt;
                        us->rtt_var = ertt / 2;
                        /* sanity check. rtt should never be more than 6 seconds */
                        /* assert(rtt < 6000); */
                } else {
                        int delta;

                        /* Compute new round trip times */
                        delta = (int)us->rtt - ertt;
                        us->rtt_var = us->rtt_var + (int)(abs(delta) - us->rtt_var) / 4;
                        us->rtt = us->rtt - us->rtt/8 + ertt/8;
                        /* sanity check. rtt should never be more than 6 seconds */
                        /* assert(rtt < 6000); */
                        dh_add_sample(&us->rtt_hist, ertt);
                }
                us->rto = uint_max(us->rtt + us->rtt_var * 4, 500);
                LOG_UTPV("0x%08x: rtt:%u avg:%u var:%u rto:%u", us, ertt, us->rtt, us->rtt_var, us->rto);
        }
        us->retransmit_timeout = us->rto;
        us->rto_timeout = g_current_ms + us->rto;
        /* if need_resend is set, this packet has already
           been considered timed-out, and is not included in
           the cur_window anymore */
        if (!pkt->need_resend) {
                assert(us->cur_window >= pkt->payload);
                us->cur_window -= pkt->payload;
        }
        free(pkt);
        return 0;
}

/* count the number of bytes that were acked by the EACK header */
/* XXX: carefull min_rtt was a cerkerk reference */
static size_t us_selective_ack_bytes(struct UTPSocket *us, uint base, byte* mask, byte len, int64 *min_rtt)
{
        size_t acked_bytes;
        int bits;

        if (us->cur_window_packets == 0)
                return 0;

        acked_bytes = 0;
        bits = len * 8;

        do {
                uint v;
                struct op *pkt;

                v = base + bits;

                /* ignore bits that haven't been sent yet
                   see comment in UTPSocket::selective_ack */
                if (((us->seq_nr - v - 1) & ACK_NR_MASK) >= (uint16)(us->cur_window_packets - 1))
                        continue;

                /* ignore bits that represents packets we haven't sent yet
                   or packets that have already been acked */
                pkt = (struct op*)scb_get(&us->outbuf, v);
                if (!pkt || pkt->transmissions == 0)
                        continue;

                /* Count the number of segments that were successfully received past it. */
                if (bits >= 0 && mask[bits>>3] & (1 << (bits & 7))) {
                        assert((int)(pkt->payload) >= 0);
                        acked_bytes += pkt->payload;
                        *min_rtt = int64_min(*min_rtt, UTP_GetMicroseconds() - pkt->time_sent);
                        continue;
                }
        } while (--bits >= -1);
        return acked_bytes;
}

#define MAX_EACK 128

static void us_selective_ack(struct UTPSocket *us, uint base, byte *mask, byte len)
{
        int bits;
        int count;
        int resends[MAX_EACK];
        int nr;
        bool back_off;
        int i;

        if (us->cur_window_packets == 0)
                return;

        /* the range is inclusive [0, 31] bits */
        bits = len * 8 - 1;

        count = 0;

        /* resends is a stack of sequence numbers we need to resend. Since we
           iterate in reverse over the acked packets, at the end, the top packets
           are the ones we want to resend */
        /* resends[MAX_EACK]; */
        nr = 0;

        LOG_UTPV("0x%08x: Got EACK [%032b] base:%u", us, *(uint32*)mask, base);
        do {
                uint v;
                bool bit_set;
                struct op *pkt;

                /* we're iterating over the bits from higher sequence numbers
                   to lower (kind of in reverse order, wich might not be very
                   intuitive) */
                v = base + bits;

                /* ignore bits that haven't been sent yet
                   and bits that fall below the ACKed sequence number
                   this can happen if an EACK message gets
                   reordered and arrives after a packet that ACKs up past
                   the base for thie EACK message

                   this is essentially the same as:
                   if v >= seq_nr || v <= seq_nr - cur_window_packets
                   but it takes wrapping into account

                   if v == seq_nr the -1 will make it wrap. if v > seq_nr
                   it will also wrap (since it will fall further below 0)
                   and be > cur_window_packets.
                   if v == seq_nr - cur_window_packets, the result will be
                   seq_nr - (seq_nr - cur_window_packets) - 1
                   == seq_nr - seq_nr + cur_window_packets - 1
                   == cur_window_packets - 1 which will be caught by the
                   test. If v < seq_nr - cur_window_packets the result will grow
                   fall furhter outside of the cur_window_packets range.

                   sequence number space:
                   
                       rejected <   accepted   > rejected 
                   <============+--------------+============>
                                ^              ^
                                |              |
                          (seq_nr-wnd)         seq_nr */

                if (((us->seq_nr - v - 1) & ACK_NR_MASK) >= (uint16)(us->cur_window_packets - 1))
                        continue;

                /* this counts as a duplicate ack, even though we might have
                   received an ack for this packet previously (in another EACK
                   message for instance) */
                bit_set = bits >= 0 && mask[bits>>3] & (1 << (bits & 7));

                /* if this packet is acked, it counts towards the duplicate ack counter */
                if (bit_set)
                        count++;

                /* ignore bits that represents packets we haven't sent yet
                   or packets that have already been acked */
                pkt = (struct op*)scb_get(&us->outbuf, v);
                if (!pkt || pkt->transmissions == 0) {
                        LOG_UTPV("0x%08x: skipping %u. pkt:%08x transmissions:%u %s", us, v, pkt, pkt?pkt->transmissions:0, pkt?"(not sent yet?)":"(already acked?)");
                        continue;
                }

                /* Count the number of segments that were successfully received past it. */
                if (bit_set) {
                        /* the selective ack should never ACK the packet we're waiting for to decrement cur_window_packets */
                        assert((v & us->outbuf.mask) != ((us->seq_nr - us->cur_window_packets) & us->outbuf.mask));
                        us_ack_packet(us, v);
                        continue;
                }

                /* Resend segments
                   if count is less than our re-send limit, we haven't seen enough
                   acked packets in front of this one to warrant a re-send.
                   if count == 0, we're still going through the tail of zeroes */
                if (((v - us->fast_resend_seq_nr) & ACK_NR_MASK) <= OUTGOING_BUFFER_MAX_SIZE && count >= DUPLICATE_ACKS_BEFORE_RESEND && us->duplicate_ack < DUPLICATE_ACKS_BEFORE_RESEND) {
                        /* resends is a stack, and we're mostly interested in the top of it
                           if we're full, just throw away the lower half */
                        if (nr >= MAX_EACK - 2) {
                                memmove(resends, &resends[MAX_EACK/2], MAX_EACK/2 * sizeof(resends[0]));
                                nr -= MAX_EACK / 2;
                        }
                        resends[nr++] = v;
                        LOG_UTPV("0x%08x: no ack for %u", us, v);
                } else {
                        LOG_UTPV("0x%08x: not resending %u count:%d dup_ack:%u fast_resend_seq_nr:%u", us, v, count, us->duplicate_ack, us->fast_resend_seq_nr);
                }
        } while (--bits >= -1);

        if (((base - 1 - us->fast_resend_seq_nr) & ACK_NR_MASK) <= OUTGOING_BUFFER_MAX_SIZE && count >= DUPLICATE_ACKS_BEFORE_RESEND) {
                /* if we get enough duplicate acks to start
                   resending, the first packet we should resend
                   is base-1 */
                resends[nr++] = (base - 1) & ACK_NR_MASK;
        } else {
                LOG_UTPV("0x%08x: not resending %u count:%d dup_ack:%u fast_resend_seq_nr:%u", us, base - 1, count, us->duplicate_ack, us->fast_resend_seq_nr);
        }

        back_off = false;
        i = 0;
        while (nr > 0) {
                uint v;
                struct op *pkt;

                v = resends[--nr];
                /* don't consider the tail of 0:es to be lost packets
                   only unacked packets with acked packets after should
                   be considered lost */
                pkt = (struct op*)scb_get(&us->outbuf, v);

                /* this may be an old (re-ordered) packet, and some of the
                   packets in here may have been acked already. In which
                   case they will not be in the send queue anymore */
                if (!pkt)
                        continue;

                /* used in parse_log.py */
                LOG_UTP("0x%08x: Packet %u lost. Resending", us, v);

                /* On Loss */
                back_off = true;
#ifdef _DEBUG
                ++us->_stats._rexmit;
#endif
                us_send_packet(us, pkt);
                us->fast_resend_seq_nr = v + 1;

                /* Re-send max 4 packets. */
                if (++i >= 4)
                        break;
        }

        if (back_off)
                us_maybe_decay_win(us);

        us->duplicate_ack = count;
}

static void us_apply_ledbat_ccontrol(struct UTPSocket *us, size_t bytes_acked, uint32 actual_delay, int64 min_rtt)
{
        int32 our_delay;
        SOCKADDR_STORAGE sa;
        int target;
        double off_target;
        double window_factor;
        double delay_factor;
        double scaled_gain;

        /* the delay can never be greater than the rtt. The min_rtt
           variable is the RTT in microseconds */
        
        assert(min_rtt >= 0);
        our_delay = uint32_min(dh_get_value(&us->our_hist), (uint32)min_rtt);
        assert(our_delay != INT_MAX);
        assert(our_delay >= 0);

        sa = psa_get_sockaddr_storage(&us->addr, NULL);
        UTP_DelaySample((struct sockaddr*)&sa, our_delay / 1000);

        /* This test the connection under heavy load from foreground
           traffic. Pretend that our delays are very high to force the
           connection to use sub-packet size window sizes */
        /*our_delay *= 4; */

        /* target is microseconds */
        target = CCONTROL_TARGET;
        if (target <= 0)
                target = 100000;

        off_target = target - our_delay;

        /* this is the same as:
        
           (min(off_target, target) / target) * (bytes_acked / max_window) * MAX_CWND_INCREASE_BYTES_PER_RTT
        
           so, it's scaling the max increase by the fraction of the window this ack represents, and the fraction
           of the target delay the current delay represents.
           The min() around off_target protects against crazy values of our_delay, which may happen when th
           timestamps wraps, or by just having a malicious peer sending garbage. This caps the increase
           of the window size to MAX_CWND_INCREASE_BYTES_PER_RTT per rtt.
           as for large negative numbers, this direction is already capped at the min packet size further down
           the min around the bytes_acked protects against the case where the window size was recently
           shrunk and the number of acked bytes exceeds that. This is considered no more than one full
           window, in order to keep the gain within sane boundries. */

        assert(bytes_acked > 0);
        window_factor = (double)size_t_min(bytes_acked, us->max_window) / (double)size_t_max(us->max_window, bytes_acked);
        delay_factor = off_target / target;
        scaled_gain = MAX_CWND_INCREASE_BYTES_PER_RTT * window_factor * delay_factor;

        /* since MAX_CWND_INCREASE_BYTES_PER_RTT is a cap on how much the window size (max_window)
           may increase per RTT, we may not increase the window size more than that proportional
           to the number of bytes that were acked, so that once one window has been acked (one rtt)
           the increase limit is not exceeded
           the +1. is to allow for floating point imprecision */
        assert(scaled_gain <= 1. + MAX_CWND_INCREASE_BYTES_PER_RTT * (int)size_t_min(bytes_acked, us->max_window) / (double)size_t_max(us->max_window, bytes_acked));

        if (scaled_gain > 0 && g_current_ms - us->last_maxed_out_window > 300)
                /* if it was more than 300 milliseconds since we tried to send a packet
                   and stopped because we hit the max window, we're most likely rate
                   limited (which prevents us from ever hitting the window size)
                   if this is the case, we cannot let the max_window grow indefinitely */
                scaled_gain = 0;

        if (scaled_gain + us->max_window < MIN_WINDOW_SIZE)
                us->max_window = MIN_WINDOW_SIZE;
        else
                us->max_window = (size_t)(us->max_window + scaled_gain);

        /* make sure that the congestion window is below max
           make sure that we don't shrink our window too small */
        us->max_window = size_t_clamp(us->max_window, MIN_WINDOW_SIZE, us->opt_sndbuf);

        /* used in parse_log.py */
        LOG_UTP("0x%08x: actual_delay:%u our_delay:%d their_delay:%u off_target:%d max_window:%u " "delay_base:%u delay_sum:%d target_delay:%d acked_bytes:%u cur_window:%u " "scaled_gain:%f rtt:%u rate:%u quota:%d wnduser:%u rto:%u timeout:%d get_microseconds:"I64u" " "cur_window_packets:%u packet_size:%u their_delay_base:%u their_actual_delay:%u", us, actual_delay, our_delay / 1000, dh_get_value(&us->their_hist) / 1000, (int)off_target / 1000, (uint)(us->max_window), us->our_hist.delay_base, (our_delay + dh_get_value(&us->their_hist)) / 1000, target / 1000, (uint)bytes_acked, (uint)(us->cur_window - bytes_acked), (float)(scaled_gain), us->rtt, (uint)(us->max_window * 1000 / (us->rtt_hist.delay_base?us->rtt_hist.delay_base:50)), us->send_quota / 100, (uint)us->max_window_user, us->rto, (int)(us->rto_timeout - g_current_ms), UTP_GetMicroseconds(), us->cur_window_packets, (uint)us_get_packet_size(us), us->their_hist.delay_base, us->their_hist.delay_base + dh_get_value(&us->their_hist));
}

static void UTP_RegisterRecvPacket(struct UTPSocket *conn, size_t len)
{
#ifdef _DEBUG
        ++conn->_stats._nrecv;
        conn->_stats._nbytes_recv += len;
#else
        (void)conn;
#endif

        if (len <= PACKET_SIZE_MID)
                if (len <= PACKET_SIZE_EMPTY)
                        _global_stats._nraw_recv[PACKET_SIZE_EMPTY_BUCKET]++;
                else if (len <= PACKET_SIZE_SMALL)
                        _global_stats._nraw_recv[PACKET_SIZE_SMALL_BUCKET]++;
                else 
                        _global_stats._nraw_recv[PACKET_SIZE_MID_BUCKET]++;
        else
                if (len <= PACKET_SIZE_BIG)
                        _global_stats._nraw_recv[PACKET_SIZE_BIG_BUCKET]++;
                else 
                        _global_stats._nraw_recv[PACKET_SIZE_HUGE_BUCKET]++;
}

/* Process an incoming packet
   syn is true if this is the first packet received. It will cut off parsing
   as soon as the header is done */
/* XXX: syn default to false */
static size_t UTP_ProcessIncoming(struct UTPSocket *conn, byte *packet, size_t len, bool syn)
{
        struct pf *pf;
        struct pf1 *pf1;
        byte *packet_end;
        uint16 pk_seq_nr;
        uint16 pk_ack_nr;
        uint8 pk_flags;
        uint64 time;
        byte *selack_ptr;
        byte *data;
        uint extension;
        uint seqnr;
        int acks;
        size_t acked_bytes;
        int64 min_rtt;
        int i;
        uint64 p;
        uint32 their_delay;
        uint32 prev_delay_base;
        uint32 actual_delay;
        byte *buf;
        byte *mem;

        UTP_RegisterRecvPacket(conn, len);

        g_current_ms = UTP_GetMilliseconds();

        us_update_send_quota(conn);

        pf = (struct pf*)packet;
        pf1 = (struct pf1*)packet;
        packet_end = packet + len;

        if (conn->version == 0) {
                pk_seq_nr = ntohs(pf->seq_nr);
                pk_ack_nr = ntohs(pf->ack_nr);
                pk_flags = pf->flags;
        } else {
                pk_seq_nr = ntohs(pf1->seq_nr);
                pk_ack_nr = ntohs(pf1->ack_nr);
                pk_flags = pf1_type(pf1);
        }

        if (pk_flags >= ST_NUM_STATES)
                return 0;

        LOG_UTPV("0x%08x: Got %s. seq_nr:%u ack_nr:%u state:%s version:%u timestamp:"I64u" reply_micro:%u", conn, flagnames[pk_flags], pk_seq_nr, pk_ack_nr, statenames[conn->state], conn->version, conn->version == 0?(uint64)(ntohl(pf->tv_sec)) * 1000000 + ntohl(pf->tv_usec):(uint64)(ntohl(pf1->tv_usec)), conn->version == 0?(uint32)(ntohl(pf->reply_micro)):(uint32)(ntohl(pf1->reply_micro)));

        /* mark receipt time */
        time = UTP_GetMicroseconds();

        /* RSTs are handled earlier, since the connid matches the send id not the recv id */
        assert(pk_flags != ST_RESET);

        /* TODO: maybe send a ST_RESET if we're in CS_RESET? */

        selack_ptr = NULL;

        /* Unpack UTP packet options
           Data pointer */
        data = (byte*)pf + us_get_header_size(conn);
        if (us_get_header_size(conn) > len) {
                LOG_UTPV("0x%08x: Invalid packet size (less than header size)", conn);
                return 0;
        }
        /* Skip the extension headers */
        extension = conn->version == 0 ? pf->ext : pf1->ext;
        if (extension != 0) {
                do {
                        /* Verify that the packet is valid. */
                        data += 2;

                        if ((int)(packet_end - data) < 0 || (int)(packet_end - data) < data[-1]) {
                                LOG_UTPV("0x%08x: Invalid len of extensions", conn);
                                return 0;
                        }

                        switch(extension) {
                        case 1: /* Selective Acknowledgment */
                                selack_ptr = data;
                                break;
                        case 2: /* extension bits */
                                if (data[-1] != 8) {
                                        LOG_UTPV("0x%08x: Invalid len of extension bits header", conn);
                                        return 0;
                                }
                                memcpy(conn->extensions, data, 8);
                                LOG_UTPV("0x%08x: got extension bits:%02x%02x%02x%02x%02x%02x%02x%02x", conn, conn->extensions[0], conn->extensions[1], conn->extensions[2], conn->extensions[3], conn->extensions[4], conn->extensions[5], conn->extensions[6], conn->extensions[7]);
                        }
                        extension = data[-2];
                        data += data[-1];
                } while (extension);
        }

        if (conn->state == CS_SYN_SENT)
                /* if this is a syn-ack, initialize our ack_nr
                   to match the sequence number we got from
                   the other end */
                conn->ack_nr = (pk_seq_nr - 1) & SEQ_NR_MASK;

        g_current_ms = UTP_GetMilliseconds();
        conn->last_got_packet = g_current_ms;

        if (syn)
                return 0;

        /* seqnr is the number of packets past the expected
           packet this is. ack_nr is the last acked, seq_nr is the
           current. Subtracring 1 makes 0 mean "this is the next
           expected packet". */
        seqnr = (pk_seq_nr - conn->ack_nr - 1) & SEQ_NR_MASK;

        /* Getting an invalid sequence number? */
        if (seqnr >= REORDER_BUFFER_MAX_SIZE) {
                if (seqnr >= (SEQ_NR_MASK + 1) - REORDER_BUFFER_MAX_SIZE && pk_flags != ST_STATE)
                        conn->ack_time = g_current_ms + uint_min(conn->ack_time - g_current_ms, DELAYED_ACK_TIME_THRESHOLD);
                LOG_UTPV("    Got old Packet/Ack (%u/%u)=%u!", pk_seq_nr, conn->ack_nr, seqnr);
                return 0;
        }

        /* Process acknowledgment
           acks is the number of packets that was acked */
        acks = (pk_ack_nr - (conn->seq_nr - 1 - conn->cur_window_packets)) & ACK_NR_MASK;

        /* this happens when we receive an old ack nr */
        if (acks > conn->cur_window_packets)
                acks = 0;

        /* if we get the same ack_nr as in the last packet
           increase the duplicate_ack counter, otherwise reset
           it to 0 */
        if (conn->cur_window_packets > 0) {
                if (pk_ack_nr == ((conn->seq_nr - conn->cur_window_packets - 1) & ACK_NR_MASK) && conn->cur_window_packets > 0) {
                        /*++conn->duplicate_ack; */
                } else
                        conn->duplicate_ack = 0;

                /* TODO: if duplicate_ack == DUPLICATE_ACK_BEFORE_RESEND
                   and fast_resend_seq_nr <= ack_nr + 1 
                     resend ack_nr + 1 */
        }

        /* figure out how many bytes were acked */
        acked_bytes = 0;

        /* the minimum rtt of all acks
           this is the upper limit on the delay we get back
           from the other peer. Our delay cannot exceed
           the rtt of the packet. If it does, clamp it.
           this is done in apply_ledbat_ccontrol() */
        min_rtt = INT64_MAX;

        for (i = 0; i < acks; ++i) {
                int seq;
                struct op *pkt;

                seq = conn->seq_nr - conn->cur_window_packets + i;
                pkt = (struct op*)scb_get(&conn->outbuf, seq);
                if (pkt == 0 || pkt->transmissions == 0)
                        continue;
                assert((int)(pkt->payload) >= 0);
                acked_bytes += pkt->payload;
                min_rtt = int64_min(min_rtt, UTP_GetMicroseconds() - pkt->time_sent);
        }
        
        /* count bytes acked by EACK */
        if (selack_ptr != NULL)
                acked_bytes += us_selective_ack_bytes(conn, (pk_ack_nr + 2) & ACK_NR_MASK, selack_ptr, selack_ptr[-1], &min_rtt);

        LOG_UTPV("0x%08x: acks:%d acked_bytes:%u seq_nr:%d cur_window:%u cur_window_packets:%u relative_seqnr:%u max_window:%u min_rtt:%u rtt:%u", conn, acks, (uint)acked_bytes, conn->seq_nr, (uint)conn->cur_window, conn->cur_window_packets, seqnr, (uint)conn->max_window, (uint)(min_rtt / 1000), conn->rtt);

        if (conn->version == 0)
                p = (uint64)ntohl(pf->tv_sec) * 1000000 + ntohl(pf->tv_usec);
        else
                p = ntohl(pf1->tv_usec);

        conn->last_measured_delay = g_current_ms;

        /* get delay in both directions
           record the delay to report back */
        their_delay = (uint32)(p == 0 ? 0 : time - p);
        conn->reply_micro = their_delay;
        prev_delay_base = conn->their_hist.delay_base;
        if (their_delay != 0)
                dh_add_sample(&conn->their_hist, their_delay);

        /* if their new delay base is less than their previous one
           we should shift our delay base in the other direction in order
           to take the clock skew into account */
        if (prev_delay_base != 0 && wrapping_compare_less(conn->their_hist.delay_base, prev_delay_base))
                /* never adjust more than 10 milliseconds */
                if (prev_delay_base - conn->their_hist.delay_base <= 10000)
                        dh_shift(&conn->our_hist, prev_delay_base - conn->their_hist.delay_base);

        actual_delay = conn->version==0 ? (ntohl(pf->reply_micro)==INT_MAX?0:(uint32)(ntohl(pf->reply_micro))) : ((uint32)(ntohl(pf1->reply_micro))==INT_MAX?0:(uint32)(ntohl(pf1->reply_micro)));

        /* if the actual delay is 0, it means the other end
           hasn't received a sample from us yet, and doesn't
           know what it is. We can't update out history unless
           we have a true measured sample */
        prev_delay_base = conn->our_hist.delay_base;
        if (actual_delay != 0)
                dh_add_sample(&conn->our_hist, actual_delay);

        /* if our new delay base is less than our previous one
           we should shift the other end's delay base in the other
           direction in order to take the clock skew into account
           This is commented out because it creates bad interactions
           with our adjustment in the other direction. We don't really
           need our estimates of the other peer to be very accurate
           anyway. The problem with shifting here is that we're more
           likely shift it back later because of a low latency. This
           second shift back would cause us to shift our delay base
           which then get's into a death spiral of shifting delay bases */
/*      if (prev_delay_base != 0 &&
                wrapping_compare_less(conn->our_hist.delay_base, prev_delay_base)) {
                // never adjust more than 10 milliseconds
                if (prev_delay_base - conn->our_hist.delay_base <= 10000) {
                        conn->their_hist.Shift(prev_delay_base - conn->our_hist.delay_base);
                }
        }
*/

        /* if the delay estimate exceeds the RTT, adjust the base_delay to
           compensate */
        if (dh_get_value(&conn->our_hist) > (uint32)(min_rtt)) {
                dh_shift(&conn->our_hist, dh_get_value(&conn->our_hist) - min_rtt);
        }

        /* only apply the congestion controller on acks
           if we don't have a delay measurement, there's
           no point in invoking the congestion control */
        if (actual_delay != 0 && acked_bytes >= 1)
                us_apply_ledbat_ccontrol(conn, acked_bytes, actual_delay, min_rtt);

        /* sanity check, the other end should never ack packets
           past the point we've sent */
        if (acks <= conn->cur_window_packets) {
                conn->max_window_user = conn->version == 0 ? pf->windowsize * PACKET_SIZE : ntohl(pf1->windowsize);

                /* If max user window is set to 0, then we startup a timer
                   That will reset it to 1 after 15 seconds. */
                if (conn->max_window_user == 0)
                        /* Reset max_window_user to 1 every 15 seconds. */
                        conn->zerowindow_time = g_current_ms + 15000;

                /* Respond to connect message
                   Switch to CONNECTED state. */
                if (conn->state == CS_SYN_SENT) {
                        conn->state = CS_CONNECTED;
                        conn->func.on_state(conn->userdata, UTP_STATE_CONNECT);

                /* We've sent a fin, and everything was ACKed (including the FIN),
                   it's safe to destroy the socket. cur_window_packets == acks
                   means that this packet acked all the remaining packets that
                   were in-flight. */
                } else if (conn->state == CS_FIN_SENT && conn->cur_window_packets == acks)
                        conn->state = CS_DESTROY;

                /* Update fast resend counter */
                if (wrapping_compare_less(conn->fast_resend_seq_nr, (pk_ack_nr + 1) & ACK_NR_MASK))
                        conn->fast_resend_seq_nr = pk_ack_nr + 1;

                LOG_UTPV("0x%08x: fast_resend_seq_nr:%u", conn, conn->fast_resend_seq_nr);

                for (i = 0; i < acks; ++i) {
                        int ack_status;

                        ack_status = us_ack_packet(conn, conn->seq_nr - conn->cur_window_packets);
                        /* if ack_status is 0, the packet was acked.
                           if acl_stauts is 1, it means that the packet had already been acked
                           if it's 2, the packet has not been sent yet
                           We need to break this loop in the latter case. This could potentially
                           happen if we get an ack_nr that does not exceed what we have stuffed
                           into the outgoing buffer, but does exceed what we have sent */
                        if (ack_status == 2) {
#ifdef _DEBUG
                                struct op* pkt;

                                pkt = (struct op*)scb_get(&conn->outbuf, conn->seq_nr - conn->cur_window_packets);
                                assert(pkt->transmissions == 0);
#endif
                                break;
                        }
                        conn->cur_window_packets--;
                }
#ifdef _DEBUG
                if (conn->cur_window_packets == 0)
                        assert(conn->cur_window == 0);
#endif

                /* packets in front of this may have been acked by a
                   selective ack (EACK). Keep decreasing the window packet size
                   until we hit a packet that is still waiting to be acked
                   in the send queue
                   this is especially likely to happen when the other end
                   has the EACK send bug older versions of uTP had */
                while (conn->cur_window_packets > 0 && !scb_get(&conn->outbuf, conn->seq_nr - conn->cur_window_packets))
                        conn->cur_window_packets--;

#ifdef _DEBUG
                if (conn->cur_window_packets == 0)
                        assert(conn->cur_window == 0);
#endif

                /* this invariant should always be true */
                assert(conn->cur_window_packets == 0 || scb_get(&conn->outbuf, conn->seq_nr - conn->cur_window_packets));

                /*  flush Nagle */
                if (conn->cur_window_packets == 1) {
                        struct op *pkt;

                        pkt = (struct op*)scb_get(&conn->outbuf, conn->seq_nr - 1);
                        /* do we still have quota? */
                        if (pkt->transmissions == 0 && (!(USE_PACKET_PACING) || conn->send_quota / 100 >= (int32)(pkt->length))) {
                                us_send_packet(conn, pkt);

                                /* No need to send another ack if there is nothing to reorder. */
                                if (conn->reorder_count == 0)
                                        us_sent_ack(conn);
                        }
                }

                /* Fast timeout-retry */
                if (conn->fast_timeout) {
                        LOG_UTPV("Fast timeout %u,%u,%u?", (uint)conn->cur_window, conn->seq_nr - conn->timeout_seq_nr, conn->timeout_seq_nr);
                        /* if the fast_resend_seq_nr is not pointing to the oldest outstanding packet, it suggests that we've already
                           resent the packet that timed out, and we should leave the fast-timeout mode. */
                        if (((conn->seq_nr - conn->cur_window_packets) & ACK_NR_MASK) != conn->fast_resend_seq_nr)
                                conn->fast_timeout = false;
                        else {
                                struct op *pkt;
                                /* resend the oldest packet and increment fast_resend_seq_nr
                                   to not allow another fast resend on it again */
                                pkt = (struct op*)scb_get(&conn->outbuf, conn->seq_nr - conn->cur_window_packets);
                                if (pkt && pkt->transmissions > 0) {
                                        LOG_UTPV("0x%08x: Packet %u fast timeout-retry.", conn, conn->seq_nr - conn->cur_window_packets);
#ifdef _DEBUG
                                        ++conn->_stats._fastrexmit;
#endif
                                        conn->fast_resend_seq_nr++;
                                        us_send_packet(conn, pkt);
                                }
                        }
                }
        }

        /* Process selective acknowledgent */
        if (selack_ptr != NULL)
                us_selective_ack(conn, pk_ack_nr + 2, selack_ptr, selack_ptr[-1]);

        /* this invariant should always be true */
        assert(conn->cur_window_packets == 0 || scb_get(&conn->outbuf, conn->seq_nr - conn->cur_window_packets));

        LOG_UTPV("0x%08x: acks:%d acked_bytes:%u seq_nr:%u cur_window:%u cur_window_packets:%u quota:%d", conn, acks, (uint)acked_bytes, conn->seq_nr, (uint)conn->cur_window, conn->cur_window_packets, conn->send_quota / 100);

        /* In case the ack dropped the current window below
           the max_window size, Mark the socket as writable */
        if (conn->state == CS_CONNECTED_FULL && us_is_writable(conn, us_get_packet_size(conn))) {
                conn->state = CS_CONNECTED;
                LOG_UTPV("0x%08x: Socket writable. max_window:%u cur_window:%u quota:%d packet_size:%u", conn, (uint)conn->max_window, (uint)conn->cur_window, conn->send_quota / 100, (uint)us_get_packet_size(conn));
                conn->func.on_state(conn->userdata, UTP_STATE_WRITABLE);
        }

        if (pk_flags == ST_STATE)
                /* This is a state packet only. */
                return 0;

        /* The connection is not in a state that can accept data? */
        if (conn->state != CS_CONNECTED && conn->state != CS_CONNECTED_FULL && conn->state != CS_FIN_SENT)
                return 0;

        /* Is this a finalize packet? */
        if (pk_flags == ST_FIN && !conn->got_fin) {
                LOG_UTPV("Got FIN eof_pkt:%u", pk_seq_nr);
                conn->got_fin = true;
                conn->eof_pkt = pk_seq_nr;
                /* at this point, it is possible for the
                   other end to have sent packets with
                   sequence numbers higher than seq_nr.
                   if this is the case, our reorder_count
                   is out of sync. This case is dealt with
                   when we re-order and hit the eof_pkt.
                   we'll just ignore any packets with
                   sequence numbers past this */
        }

        /* Getting an in-order packet? */
        if (seqnr == 0) {
                size_t count;

                count = packet_end - data;
                if (count > 0 && conn->state != CS_FIN_SENT) {
                        LOG_UTPV("0x%08x: Got Data len:%u (rb:%u)", conn, (uint)count, (uint)conn->func.get_rb_size(conn->userdata));
                        /* Post bytes to the upper layer */
                        conn->func.on_read(conn->userdata, data, count);
                }
                conn->ack_nr++;
                conn->bytes_since_ack += count;

                /* Check if the next packet has been received too, but waiting
                   in the reorder buffer. */
                for (;;) {
                        if (conn->got_fin && conn->eof_pkt == conn->ack_nr) {
                                if (conn->state != CS_FIN_SENT) {
                                        conn->state = CS_GOT_FIN;
                                        conn->rto_timeout = g_current_ms + uint_min(conn->rto * 3, 60);

                                        LOG_UTPV("0x%08x: Posting EOF", conn);
                                        conn->func.on_state(conn->userdata, UTP_STATE_EOF);
                                }

                                /* if the other end wants to close, ack immediately */
                                us_send_ack(conn, false);

                                /* reorder_count is not necessarily 0 at this point.
                                   even though it is most of the time, the other end
                                   may have sent packets with higher sequence numbers
                                   than what later end up being eof_pkt
                                   since we have received all packets up to eof_pkt
                                   just ignore the ones after it. */
                                conn->reorder_count = 0;
                        }

                        /* Quick get-out in case there is nothing to reorder */
                        if (conn->reorder_count == 0)
                                break;

                        /* Check if there are additional buffers in the reorder buffers
                           that need delivery. */
                        buf = (byte*)scb_get(&conn->inbuf, conn->ack_nr+1);
                        if (buf == NULL)
                                break;
                        scb_put(&conn->inbuf, conn->ack_nr+1, NULL);
                        count = *(uint*)buf;
                        if (count > 0 && conn->state != CS_FIN_SENT)
                                /* Pass the bytes to the upper layer */
                                conn->func.on_read(conn->userdata, buf + sizeof(uint), count);
                        conn->ack_nr++;
                        conn->bytes_since_ack += count;

                        /* Free the element from the reorder buffer */
                        free(buf);
                        assert(conn->reorder_count > 0);
                        conn->reorder_count--;
                }

                /* start the delayed ACK timer */
                conn->ack_time = g_current_ms + uint_min(conn->ack_time - g_current_ms, DELAYED_ACK_TIME_THRESHOLD);
        } else {
                /* Getting an out of order packet.
                   The packet needs to be remembered and rearranged later. */

                /* if we have received a FIN packet, and the EOF-sequence number
                   is lower than the sequence number of the packet we just received
                   something is wrong. */
                if (conn->got_fin && pk_seq_nr > conn->eof_pkt) {
                        LOG_UTPV("0x%08x: Got an invalid packet sequence number, past EOF " "reorder_count:%u len:%u (rb:%u)", conn, conn->reorder_count, (uint)(packet_end - data), (uint)conn->func.get_rb_size(conn->userdata));
                        return 0;
                }

                /* if the sequence number is entirely off the expected
                   one, just drop it. We can't allocate buffer space in
                   the inbuf entirely based on untrusted input */
                if (seqnr > 0x3ff) {
                        LOG_UTPV("0x%08x: Got an invalid packet sequence number, too far off " "reorder_count:%u len:%u (rb:%u)", conn, conn->reorder_count, (uint)(packet_end - data), (uint)conn->func.get_rb_size(conn->userdata));
                        return 0;
                }

                /* we need to grow the circle buffer before we
                   check if the packet is already in here, so that
                   we don't end up looking at an older packet (since
                   the indices wraps around). */
                scb_ensure_size(&conn->inbuf, pk_seq_nr + 1, seqnr + 1);

                /* Has this packet already been received? (i.e. a duplicate)
                   If that is the case, just discard it. */
                if (scb_get(&conn->inbuf, pk_seq_nr) != NULL) {
#ifdef _DEBUG
                        ++conn->_stats._nduprecv;
#endif
                        return 0;
                }

                /* Allocate memory to fit the packet that needs to re-ordered */
                mem = malloc((packet_end - data) + sizeof(uint));
                *(uint*)mem = (uint)(packet_end - data);
                memcpy(mem + sizeof(uint), data, packet_end - data);

                /* Insert into reorder buffer and increment the count
                   of # of packets to be reordered.
                   we add one to seqnr in order to leave the last
                   entry empty, that way the assert in send_ack
                   is valid. we have to add one to seqnr too, in order
                   to make the circular buffer grow around the correct
                   point (which is conn->ack_nr + 1). */
                assert(scb_get(&conn->inbuf, pk_seq_nr) == NULL);
                assert((pk_seq_nr & conn->inbuf.mask) != ((conn->ack_nr+1) & conn->inbuf.mask));
                scb_put(&conn->inbuf, pk_seq_nr, mem);
                conn->reorder_count++;

                LOG_UTPV("0x%08x: Got out of order data reorder_count:%u len:%u (rb:%u)", conn, conn->reorder_count, (uint)(packet_end - data), (uint)conn->func.get_rb_size(conn->userdata));

                /* Setup so the partial ACK message will get sent immediately. */
                conn->ack_time = g_current_ms + uint_min(conn->ack_time - g_current_ms, 1);
        }

        /* If ack_time or ack_bytes indicate that we need to send and ack, send one
           here instead of waiting for the timer to trigger */
        LOG_UTPV("bytes_since_ack:%u ack_time:%d", (uint)conn->bytes_since_ack, (int)(g_current_ms - conn->ack_time));
        if (conn->state == CS_CONNECTED || conn->state == CS_CONNECTED_FULL)
                if (conn->bytes_since_ack > DELAYED_ACK_BYTE_THRESHOLD || (int)(g_current_ms - conn->ack_time) >= 0)
                        us_send_ack(conn, false);
        return (size_t)(packet_end - data);
}

static __inline__ bool UTP_IsV1(struct pf1 *pf1)
{
        return pf1_version(pf1) == 1 && pf1_type(pf1) < ST_NUM_STATES && pf1->ext < 3;
}

static void UTP_Free(struct UTPSocket *conn)
{
        struct UTPSocket **last_ptr;
        struct UTPSocket *last;
        size_t i;

        LOG_UTPV("0x%08x: Killing socket", conn);

        conn->func.on_state(conn->userdata, UTP_STATE_DESTROYING);
        UTP_SetCallbacks(conn, NULL, NULL);

        assert(conn->idx < ar_GetCount(&g_utp_sockets));
        assert({last_ptr = ar_get(&g_utp_sockets, conn->idx); *last_ptr == conn;});

        /* Unlink object from the global list */
        assert(ar_GetCount(&g_utp_sockets) > 0);

        last_ptr = ar_get(&g_utp_sockets, ar_GetCount(&g_utp_sockets) - 1);
        last = *last_ptr;

        assert(last->idx < ar_GetCount(&g_utp_sockets));
        assert({last_ptr = ar_get(&g_utp_sockets, last->idx); *last_ptr == last;});

        last->idx = conn->idx;
        
        memcpy(ar_get(&g_utp_sockets, conn->idx), &last, sizeof(last));

        /* Decrease the count */
        ar_SetCount(&g_utp_sockets, ar_GetCount(&g_utp_sockets) - 1);

        /* Free all memory occupied by the socket object. */
        for (i = 0; i <= conn->inbuf.mask; i++) {
                free(conn->inbuf.elements[i]);
        }
        for (i = 0; i <= conn->outbuf.mask; i++) {
                free(conn->outbuf.elements[i]);
        }
        free(conn->inbuf.elements);
        free(conn->outbuf.elements);

        /* Finally free the socket object */
        free(conn);
}

/******************************************************************************/
/******************************************************************************/
/******************************************************************************/
/******************************************************************************/
/* public functions */
/******************************************************************************/
/******************************************************************************/
/******************************************************************************/
/******************************************************************************/

/* Create a UTP socket */
struct UTPSocket *UTP_Create(SendToProc *send_to_proc, void *send_to_userdata, const struct sockaddr *addr, socklen_t addrlen)
{
        struct UTPSocket *conn;

        conn = calloc(1, sizeof(*conn));

        g_current_ms = UTP_GetMilliseconds();

        UTP_SetCallbacks(conn, NULL, NULL);
        dh_clear(&conn->our_hist);
        dh_clear(&conn->their_hist);
        conn->rto = 3000;
        conn->rtt_var = 800;
        conn->seq_nr = 1;
        conn->ack_nr = 0;
        conn->max_window_user = 255 * PACKET_SIZE;
        psa_init(&conn->addr, (SOCKADDR_STORAGE*)addr, addrlen);
        conn->send_to_proc = send_to_proc;
        conn->send_to_userdata = send_to_userdata;
        conn->ack_time = g_current_ms + 0x70000000;
        conn->last_got_packet = g_current_ms;
        conn->last_sent_packet = g_current_ms;
        conn->last_measured_delay = g_current_ms + 0x70000000;
        conn->last_rwin_decay = (int32)(g_current_ms) - MAX_WINDOW_DECAY;
        conn->last_send_quota = g_current_ms;
        conn->send_quota = PACKET_SIZE * 100;
        conn->cur_window_packets = 0;
        conn->fast_resend_seq_nr = conn->seq_nr;

        /* default to version 1 */
        UTP_SetSockopt(conn, SO_UTPVERSION, 1);

        /* we need to fit one packet in the window
           when we start the connection */
        conn->max_window = us_get_packet_size(conn);
        conn->state = CS_IDLE;

        conn->outbuf.mask = 15;
        conn->inbuf.mask = 15;

        conn->outbuf.elements = calloc(16, sizeof(void*));
        conn->inbuf.elements = calloc(16, sizeof(void*));

        conn->idx = ar_Append(&g_utp_sockets, &conn);

        LOG_UTPV("0x%08x: UTP_Create", conn);
        return conn;
}

void UTP_SetCallbacks(struct UTPSocket *conn, struct UTPFunctionTable *funcs, void *userdata)
{
        assert(conn);

        if (funcs == NULL)
                funcs = &zero_funcs;
        conn->func = *funcs;
        conn->userdata = userdata;
}

bool UTP_SetSockopt(struct UTPSocket* conn, int opt, int val)
{
        assert(conn);

        switch (opt) {
        case SO_SNDBUF:
                assert(val >= 1);
                conn->opt_sndbuf = val;
                return true;
        case SO_RCVBUF:
                conn->opt_rcvbuf = val;
                return true;
        case SO_UTPVERSION:
                assert(conn->state == CS_IDLE);
                if (conn->state != CS_IDLE)
                        /* too late */
                        return false;
                if (conn->version == 1 && val == 0) {
                        conn->reply_micro = INT_MAX;
                        conn->opt_rcvbuf = 200 * 1024;
                        conn->opt_sndbuf = OUTGOING_BUFFER_MAX_SIZE * PACKET_SIZE;
                } else if (conn->version == 0 && val == 1) {
                        conn->reply_micro = 0;
                        conn->opt_rcvbuf = 3 * 1024 * 1024 + 512 * 1024;
                        conn->opt_sndbuf = conn->opt_rcvbuf;
                }
                conn->version = val;
                return true;
        }
        return false;
}

/* Try to connect to a specified host.
   'initial' is the number of data bytes to send in the connect packet. */
void UTP_Connect(struct UTPSocket *conn)
{
        uint32 conn_seed;
        size_t header_ext_size;
        struct op *pkt;
        struct pfe* p;
        struct pfe1* p1;

        assert(conn);

        assert(conn->state == CS_IDLE);
        assert(conn->cur_window_packets == 0);
        assert(scb_get(&conn->outbuf, conn->seq_nr) == NULL);
        assert(sizeof(struct pf1) == 20);

        conn->state = CS_SYN_SENT;

        g_current_ms = UTP_GetMilliseconds();

        /* Create and send a connect message */
        conn_seed = UTP_Random();

        /* we identify newer versions by setting the
           first two bytes to 0x0001 */
        if (conn->version > 0)
                conn_seed &= 0xffff;

        /* used in parse_log.py */
        LOG_UTP("0x%08x: UTP_Connect conn_seed:%u packet_size:%u (B) " "target_delay:%u (ms) delay_history:%u " "delay_base_history:%u (minutes)", conn, conn_seed, PACKET_SIZE, CCONTROL_TARGET / 1000, CUR_DELAY_SIZE, DELAY_BASE_HISTORY);

        /* Setup initial timeout timer. */
        conn->retransmit_timeout = 3000;
        conn->rto_timeout = g_current_ms + conn->retransmit_timeout;
        conn->last_rcv_win = us_get_rcv_window(conn);

        conn->conn_seed = conn_seed;
        conn->conn_id_recv = conn_seed;
        conn->conn_id_send = conn_seed+1;
        /* if you need compatibiltiy with 1.8.1, use this. it increases attackability though.
           conn->seq_nr = 1; */
        conn->seq_nr = UTP_Random();

        /* Create the connect packet. */
        header_ext_size = us_get_header_extensions_size(conn);

        pkt = malloc(sizeof(struct op) - 1 + header_ext_size);

        p = (struct pfe*)pkt->data;
        p1 = (struct pfe1*)pkt->data;

        memset(p, 0, header_ext_size);
        /* SYN packets are special, and have the receive ID in the connid field,
           instead of conn_id_send. */
        if (conn->version == 0) {
                p->pf.connid = htonl(conn->conn_id_recv);
                p->pf.ext = 2;
                p->pf.windowsize = (byte)DIV_ROUND_UP(conn->last_rcv_win, PACKET_SIZE);
                p->pf.seq_nr = htons(conn->seq_nr);
                p->pf.flags = ST_SYN;
                p->ext_next = 0;
                p->ext_len = 8;
                memset(p->extensions, 0, 8);
        } else {
                pf1_version_set(&p1->pf, 1);
                pf1_type_set(&p1->pf, ST_SYN);
                p1->pf.ext = 2;
                p1->pf.connid = htons((uint16)conn->conn_id_recv);
                p1->pf.windowsize = htonl((uint32)conn->last_rcv_win);
                p1->pf.seq_nr = htons(conn->seq_nr);
                p1->ext_next = 0;
                p1->ext_len = 8;
                memset(p1->extensions, 0, 8);
        }
        pkt->transmissions = 0;
        pkt->length = header_ext_size;
        pkt->payload = 0;

        /*LOG_UTPV("0x%08x: Sending connect %s [%u].",
                         conn, addrfmt(conn->addr, addrbuf), conn_seed); */

        /* Remember the message in the outgoing queue. */
        scb_ensure_size(&conn->outbuf, conn->seq_nr, conn->cur_window_packets);
        scb_put(&conn->outbuf, conn->seq_nr, pkt);
        conn->seq_nr++;
        conn->cur_window_packets++;

        us_send_packet(conn, pkt);
}

bool UTP_IsIncomingUTP(UTPGotIncomingConnection *incoming_proc, SendToProc *send_to_proc, void *send_to_userdata, const byte *buffer, size_t len, const struct sockaddr *to, socklen_t tolen)
{
        struct psa addr;
        struct pf* p;
        struct pf1* p1;
        byte version;
        uint32 id;
        struct pf *pf;
        struct pf1 *pf1;
        byte flags;
        size_t i;
        uint32 seq_nr;

        psa_init(&addr, (SOCKADDR_STORAGE*)to, tolen);

        if (len < sizeof(struct pf) && len < sizeof(struct pf1)) {
                LOG_UTPV("recv %s len:%u too small", addrfmt(&addr, addrbuf), (uint)len);
                return false;
        }

        p = (struct pf*)buffer;
        p1 = (struct pf1*)buffer;

        version = UTP_IsV1(p1);
        id = (version == 0) ? ntohl(p->connid) : (uint32)(ntohs(p1->connid));

        if (version == 0 && len < sizeof(struct pf)) {
                LOG_UTPV("recv %s len:%u version:%u too small", addrfmt(&addr, addrbuf), (uint)len, version);
                return false;
        }

        if (version == 1 && len < sizeof(struct pf1)) {
                LOG_UTPV("recv %s len:%u version:%u too small", addrfmt(&addr, addrbuf), (uint)len, version);
                return false;
        }

        LOG_UTPV("recv %s len:%u id:%u", addrfmt(&addr, addrbuf), (uint)len, id);

        pf = (struct pf*)p;
        pf1 = (struct pf1*)p;

        if (version == 0) {
                LOG_UTPV("recv id:%u seq_nr:%u ack_nr:%u", id, (uint)ntohs(pf->seq_nr), (uint)ntohs(pf->ack_nr));
        } else {
                LOG_UTPV("recv id:%u seq_nr:%u ack_nr:%u", id, (uint)ntohs(pf1->seq_nr), (uint)ntohs(pf1->ack_nr));
        }

        flags = version == 0 ? pf->flags : pf1_type(pf1);

        for (i = 0; i < ar_GetCount(&g_utp_sockets); i++) {
                struct UTPSocket **conn_ptr;
                struct UTPSocket *conn;

                conn_ptr = ar_get(&g_utp_sockets, i);
                conn = *conn_ptr;
                /*LOG_UTPV("Examining UTPSocket %s for %s and (seed:%u s:%u r:%u) for %u",
                        addrfmt(conn->addr, addrbuf), addrfmt(addr, addrbuf2), conn->conn_seed, conn->conn_id_send, conn->conn_id_recv, id); */
                if (psa_is_not_equal(&conn->addr, &addr))
                        continue;

                if (flags == ST_RESET && (conn->conn_id_send == id || conn->conn_id_recv == id)) {
                        LOG_UTPV("0x%08x: recv RST for existing connection", conn);
                        if (!conn->userdata || conn->state == CS_FIN_SENT)
                                conn->state = CS_DESTROY;
                        else
                                conn->state = CS_RESET;
                        if (conn->userdata) {
                                int err;

                                conn->func.on_overhead(conn->userdata, false, len + us_get_udp_overhead(conn), close_overhead);
                                err = conn->state == CS_SYN_SENT ? ECONNREFUSED : ECONNRESET;
                                conn->func.on_error(conn->userdata, err);
                        }
                        return true;
                } else if (flags != ST_SYN && conn->conn_id_recv == id) {
                        size_t read;

                        LOG_UTPV("0x%08x: recv processing", conn);
                        read = UTP_ProcessIncoming(conn, (byte*)buffer, len, false);
                        if (conn->userdata)
                                conn->func.on_overhead(conn->userdata, false, (len - read) + us_get_udp_overhead(conn), header_overhead);
                        return true;
                }
        }

        if (flags == ST_RESET) {
                LOG_UTPV("recv RST for unknown connection");
                return true;
        }

        seq_nr = version == 0 ? ntohs(pf->seq_nr) : ntohs(pf1->seq_nr);
        if (flags != ST_SYN) {
                size_t i;
                struct RST_Info *r;

                for (i = 0; i < ar_GetCount(&g_rst_info); i++) {
                        struct RST_Info *cur;

                        cur = ar_get(&g_rst_info,i);

                        if (cur->connid != id)
                                continue;
                        if (psa_is_not_equal(&cur->addr, &addr))
                                continue;
                        if (seq_nr != cur->ack_nr)
                                continue;
                        cur->timestamp = UTP_GetMilliseconds();
                        LOG_UTPV("recv not sending RST to non-SYN (stored)");
                        return true;
                }
                if (ar_GetCount(&g_rst_info) > RST_INFO_LIMIT) {
                        LOG_UTPV("recv not sending RST to non-SYN (limit at %u stored)", (uint)ar_GetCount(&g_rst_info));
                        return true;
                }
                LOG_UTPV("recv send RST to non-SYN (%u stored)", (uint)ar_GetCount(&g_rst_info));
                r = ar_Append_new(&g_rst_info);
                r->addr = addr;
                r->connid = id;
                r->ack_nr = seq_nr;
                r->timestamp = UTP_GetMilliseconds();

                us_send_rst(send_to_proc, send_to_userdata, &addr, id, seq_nr, UTP_Random(), version);
                return true;
        }

        if (incoming_proc) {
                struct UTPSocket *conn;
                size_t read;

                LOG_UTPV("Incoming connection from %s uTP version:%u", addrfmt(&addr, addrbuf), version);

                /* Create a new UTP socket to handle this new connection */
                conn = UTP_Create(send_to_proc, send_to_userdata, to, tolen);
                /* Need to track this value to be able to detect duplicate CONNECTs */
                conn->conn_seed = id;
                /* This is value that identifies this connection for them. */
                conn->conn_id_send = id;
                /* This is value that identifies this connection for us. */
                conn->conn_id_recv = id+1;
                conn->ack_nr = seq_nr;
                conn->seq_nr = UTP_Random();
                conn->fast_resend_seq_nr = conn->seq_nr;

                UTP_SetSockopt(conn, SO_UTPVERSION, version);
                conn->state = CS_CONNECTED;

                read = UTP_ProcessIncoming(conn, (byte*)buffer, len, true);

                LOG_UTPV("0x%08x: recv send connect ACK", conn);
                us_send_ack(conn, true);

                incoming_proc(send_to_userdata, conn);

                /* we report overhead after incoming_proc, because the callbacks are setup now */
                if (conn->userdata) {
                        /* SYN */
                        conn->func.on_overhead(conn->userdata, false, (len - read) + us_get_udp_overhead(conn), header_overhead);
                        /* SYNACK */
                        conn->func.on_overhead(conn->userdata, true, us_get_overhead(conn), ack_overhead);
                }
        }
        return true;
}

bool UTP_HandleICMP(const byte* buffer, size_t len, const struct sockaddr *to, socklen_t tolen)
{
        struct psa addr;
        struct pf* p;
        struct pf1* p1;
        byte version;
        uint32 id;
        size_t i;

        psa_init(&addr, (SOCKADDR_STORAGE*)to, tolen);

        /* Want the whole packet so we have connection ID */
        if (len < sizeof(struct pf))
                return false;

        p = (struct pf*)buffer;
        p1 = (struct pf1*)buffer;

        version = UTP_IsV1(p1);
        id = (version == 0) ? ntohl(p->connid) : (uint32)(ntohs(p1->connid));

        for (i = 0; i < ar_GetCount(&g_utp_sockets); ++i) {
                struct UTPSocket **conn_ptr;
                struct UTPSocket *conn;

                conn_ptr = ar_get(&g_utp_sockets,i);
                conn = *conn_ptr;
                if (psa_is_equal(&conn->addr,&addr) && conn->conn_id_recv == id) {
                        /* Don't pass on errors for idle/closed connections */
                        if (conn->state != CS_IDLE) {
                                if (!conn->userdata || conn->state == CS_FIN_SENT) {
                                        LOG_UTPV("0x%08x: icmp packet causing socket destruction", conn);
                                        conn->state = CS_DESTROY;
                                } else
                                        conn->state = CS_RESET;
                                if (conn->userdata) {
                                        int err;

                                        err = conn->state == CS_SYN_SENT ?  ECONNREFUSED : ECONNRESET;
                                        LOG_UTPV("0x%08x: icmp packet causing error on socket:%d", conn, err);
                                        conn->func.on_error(conn->userdata, err);
                                }
                        }
                        return true;
                }
        }
        return false;
}

/* Write bytes to the UTP socket.
   Returns true if the socket is still writable. */
bool UTP_Write(struct UTPSocket *conn, size_t bytes)
{
        size_t packet_size;
        size_t num_to_send;
#ifdef g_log_utp_verbose
        size_t param;
#endif
        assert(conn);

#ifdef g_log_utp_verbose
        param = bytes;
#endif

        if (conn->state != CS_CONNECTED) {
                LOG_UTPV("0x%08x: UTP_Write %u bytes = false (not CS_CONNECTED)", conn, (uint)bytes);
                return false;
        }

        g_current_ms = UTP_GetMilliseconds();

        us_update_send_quota(conn);

        /* don't send unless it will all fit in the window */
        packet_size = us_get_packet_size(conn);
        num_to_send = size_t_min(bytes, packet_size);
        while (us_is_writable(conn, num_to_send)) {
                /* Send an outgoing packet.
                   Also add it to the outgoing of packets that have been sent but not ACKed. */

                if (num_to_send == 0) {
                        LOG_UTPV("0x%08x: UTP_Write %u bytes = true", conn, (uint)param);
                        return true;
                }
                bytes -= num_to_send;

                LOG_UTPV("0x%08x: Sending packet. seq_nr:%u ack_nr:%u wnd:%u/%u/%u rcv_win:%u size:%u quota:%d cur_window_packets:%u", conn, conn->seq_nr, conn->ack_nr, (uint)(conn->cur_window + num_to_send), (uint)conn->max_window, (uint)conn->max_window_user, (uint)conn->last_rcv_win, num_to_send, conn->send_quota / 100, conn->cur_window_packets);
                us_write_outgoing_packet(conn, num_to_send, ST_DATA);
                num_to_send = size_t_min(bytes, packet_size);
        }

        /* mark the socket as not being writable. */
        conn->state = CS_CONNECTED_FULL;
        LOG_UTPV("0x%08x: UTP_Write %u bytes = false", conn, (uint)bytes);
        return false;
}

void UTP_RBDrained(struct UTPSocket *conn)
{
        size_t rcvwin;

        assert(conn);

        rcvwin = us_get_rcv_window(conn);

        if (rcvwin > conn->last_rcv_win) {
                /* If last window was 0 send ACK immediately, otherwise should set timer */
                if (conn->last_rcv_win == 0)
                        us_send_ack(conn, false);
                else
                        conn->ack_time = g_current_ms + uint_min(conn->ack_time - g_current_ms, DELAYED_ACK_TIME_THRESHOLD);
        }
}

void UTP_CheckTimeouts(void)
{
        size_t i;

        g_current_ms = UTP_GetMilliseconds();

        for (i = 0; i < ar_GetCount(&g_rst_info); i++) {
                struct RST_Info *cur;

                cur = ar_get(&g_rst_info, i);
                if ((int)(g_current_ms - cur->timestamp) >= RST_INFO_TIMEOUT) {
                        ar_MoveUpLast(&g_rst_info, i);
                        i--;
                }
        }
        if (ar_GetCount(&g_rst_info) != ar_GetAlloc(&g_rst_info))
                ar_Compact(&g_rst_info);

        for (i = 0; i != ar_GetCount(&g_utp_sockets); i++) {
                struct UTPSocket **conn_ptr;
                struct UTPSocket *conn;

                conn_ptr = ar_get(&g_utp_sockets, i);
                conn = *conn_ptr;
                us_check_timeouts(conn);

                /* Check if the object was deleted */
                if (conn->state == CS_DESTROY) {
                        LOG_UTPV("0x%08x: Destroying", conn);
                        UTP_Free(conn);
                        i--;
                }
        }
}

size_t UTP_GetPacketSize(struct UTPSocket *socket)
{
        return us_get_packet_size(socket);
}

void UTP_GetPeerName(struct UTPSocket *conn, struct sockaddr *addr, socklen_t *addrlen)
{
        socklen_t len;
        SOCKADDR_STORAGE sa;

        assert(conn);

        sa = psa_get_sockaddr_storage(&conn->addr, &len);
        *addrlen = socklen_t_min(len, *addrlen);
        memcpy(addr, &sa, *addrlen);
}

void UTP_GetDelays(struct UTPSocket *conn, int32 *ours, int32 *theirs, uint32 *age)
{
        assert(conn);

        if (ours)
                *ours = dh_get_value(&conn->our_hist);
        if (theirs)
                *theirs = dh_get_value(&conn->their_hist);
        if (age)
                *age = g_current_ms - conn->last_measured_delay;
}

#ifdef _DEBUG
void UTP_GetStats(struct UTPSocket *conn, struct UTPStats *stats)
{
        assert(conn);

        *stats = conn->_stats;
}
#endif /* _DEBUG */

void UTP_GetGlobalStats(struct UTPGlobalStats *stats)
{
        *stats = _global_stats;
}

/* Close the UTP socket.
   It is not valid for the upper layer to refer to socket after it is closed.
   Data will keep to try being delivered after the close. */
void UTP_Close(struct UTPSocket *conn)
{
        assert(conn);

        assert(conn->state != CS_DESTROY_DELAY && conn->state != CS_FIN_SENT && conn->state != CS_DESTROY);

        LOG_UTPV("0x%08x: UTP_Close in state:%s", conn, statenames[conn->state]);

        switch(conn->state) {
        case CS_CONNECTED:
        case CS_CONNECTED_FULL:
                conn->state = CS_FIN_SENT;
                us_write_outgoing_packet(conn, 0, ST_FIN);
                break;

        case CS_SYN_SENT:
                conn->rto_timeout = UTP_GetMilliseconds() + uint_min(conn->rto * 2, 60);
        case CS_GOT_FIN:
                conn->state = CS_DESTROY_DELAY;
                break;

        default:
                conn->state = CS_DESTROY;
                break;
        }
}