<commit_before>#include "StdAfx.h"

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

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

#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 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.
#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 300 // packet size should be divisible by 2
// This must be (PACKET_SIZE / (2^n))
// This should be fairly small, it makes it faster to
// start back up again after the window size has been
// set to MIN_WINDOW_SIZE, since we don't need to wait
// as long in between packets the smaller the packet is
#define MIN_PACKET_SIZE (PACKET_SIZE/2)

// This is a limit describing when to increase or decrease the
// packet size. When this limit is exceeded, we double the packet
// size. When the metric drops below one quarter of this limit
// the packet size is halved.
// The unit of this limit is the number of packets we can send
// per target delay. i.e. if the target delay is 100 ms, the packet
// size is 300 bytes, and the rate is 10 kB/s. That means we're
// sending 10 * 1000 / 300 = 33.3 packets per second, or 3.3 packets
// per target delay (100 ms).
#define SERIALIZATIONS_PER_TARGET 4

// 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
#define DELAYED_ACK_TIME_THRESHOLD 5

#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) x.fmt(s, sizeof(s))

#pragma pack(push,1)

struct PackedSockAddr {

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

	// Host byte order
	uint16 _port;

#define _sin4 _in._in6d[3]	// IPv4 is stored where it goes if mapped

#define _sin6 _in._in6
#define _sin6w _in._in6w
#define _sin6d _in._in6d

	byte get_family() const
	{
		return (IN6_IS_ADDR_V4MAPPED((in6_addr*)_sin6) != 0) ? AF_INET : AF_INET6;
	}

	bool operator==(const PackedSockAddr& rhs) const
	{
		if (&rhs == this)
			return true;
		if (_port != rhs._port)
			return false;
		return memcmp(_sin6, rhs._sin6, sizeof(_sin6)) == 0;
	}
	bool operator!=(const PackedSockAddr& rhs) const { return !(*this == rhs); }

	PackedSockAddr(const SOCKADDR_STORAGE* sa, socklen_t len)
	{
		if (sa->ss_family == AF_INET) {
			assert(len >= sizeof(sockaddr_in));
			const sockaddr_in *sin = (sockaddr_in*)sa;
			_sin6w[0] = 0;
			_sin6w[1] = 0;
			_sin6w[2] = 0;
			_sin6w[3] = 0;
			_sin6w[4] = 0;
			_sin6w[5] = 0xffff;
			_sin4 = sin->sin_addr.s_addr;
			_port = ntohs(sin->sin_port);
		} else {
			assert(len >= sizeof(sockaddr_in6));
			const sockaddr_in6 *sin6 = (sockaddr_in6*)sa;
			_in._in6addr = sin6->sin6_addr;
			_port = ntohs(sin6->sin6_port);
		}
		assert(sa->ss_family == get_family());
	}

	SOCKADDR_STORAGE get_sockaddr_storage(socklen_t *len = NULL) const
	{
		SOCKADDR_STORAGE sa;
		const byte family = get_family();
		if (family == AF_INET) {
			sockaddr_in *sin = (sockaddr_in*)&sa;
			if (len) *len = sizeof(sockaddr_in);
			memset(sin, 0, sizeof(sockaddr_in));
			sin->sin_family = family;
			sin->sin_port = htons(_port);
			sin->sin_addr.s_addr = _sin4;
		} else {
			sockaddr_in6 *sin6 = (sockaddr_in6*)&sa;
			memset(sin6, 0, sizeof(sockaddr_in6));
			if (len) *len = sizeof(sockaddr_in6);
			sin6->sin6_family = family;
			sin6->sin6_addr = _in._in6addr;
			sin6->sin6_port = htons(_port);
		}
		return sa;
	}

	cstr fmt(str s, size_t len) const
	{
		memset(s, 0, len);
		const byte family = get_family();
		str i;
		if (family == AF_INET) {
			inet_ntop(family, (uint32*)&_sin4, s, len);
			i = s;
			while (*++i) {}
		} else {
			i = s;
			*i++ = '[';
			inet_ntop(family, (in6_addr*)&_in._in6addr, i, len-1);
			while (*++i) {}
			*i++ = ']';
		}
		snprintf(i, len - (i-s), ":%u", _port);
		return s;
	}
};

struct RST_Info {
	PackedSockAddr addr;
	uint32 connid;
	uint32 timestamp;
	uint16 ack_nr;
};

struct PacketFormat {
	// 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 PacketFormatAck {
	PacketFormat pf;
	byte ext_next;
	byte ext_len;
	byte acks[4];
};

struct PacketFormatExtensions {
	PacketFormat pf;
	byte ext_next;
	byte ext_len;
	byte extensions[8];
};

struct PacketFormatV1 {
	// protocol version
	byte version:4;
	// type (formerly flags)
	byte type:4;
	// 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;
};

struct PacketFormatAckV1 {
	PacketFormatV1 pf;
	byte ext_next;
	byte ext_len;
	byte acks[4];
};

struct PacketFormatExtensionsV1 {
	PacketFormatV1 pf;
	byte ext_next;
	byte ext_len;
	byte extensions[8];
};

#pragma pack(pop)

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

static const cstr flagnames[] = {
	"ST_DATA","ST_FIN","ST_STATE","ST_RESET","ST_SYN"
};

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,
};

static const cstr statenames[] = {
	"IDLE","SYN_SENT","CONNECTED","CONNECTED_FULL","GOT_FIN","DESTROY_DELAY","FIN_SENT","RESET","DESTROY"
};

struct OutgoingPacket {
	uint length;
	uint payload;
	uint64 time_sent; // microseconds
	uint transmissions:31;
	bool need_resend:1;
	byte data[1];
};

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

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

struct SizableCircularBuffer {
	// This is the mask. Since it's always a power of 2, adding 1 to this value will return the size.
	uint mask;
	// This is the elements that the circular buffer points to
	void **elements;

	void *get(uint i) { assert(elements); return elements ? elements[i & mask] : NULL; }
	void put(uint i, void *data) { assert(elements); elements[i&mask] = data; }

	void grow(uint item, uint index);
	void ensure_size(uint item, uint index) { if (index > mask) grow(item, index); }
	size_t size() { return mask + 1; }
};

// Item contains the element we want to make space for
// index is the index in the list.
void SizableCircularBuffer::grow(uint item, uint index)
{
	// Figure out the new size.
	uint size = mask + 1;
	do size *= 2; while (index >= size);

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

	size--;

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

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

// 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
bool wrapping_compare_less(uint32 lhs, uint32 rhs)
{
	// distance walking from lhs to rhs, downwards
	const uint32 dist_down = lhs - rhs;
	// distance walking from lhs to rhs, upwards
	const uint32 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 DelayHist {
	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;

	void clear()
	{
		delay_base_initialized = false;
		delay_base = 0;
		cur_delay_idx = 0;
		delay_base_idx = 0;
		delay_base_time = g_current_ms;
		for (size_t i = 0; i < CUR_DELAY_SIZE; i++) {
			cur_delay_hist[i] = 0;
		}
		for (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
			delay_base_hist[i] = 0;
		}
	}

	void shift(const uint32 offset)
	{
		// 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 (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
			delay_base_hist[i] += offset;
		}
		delay_base += offset;
	}

	void add_sample(const 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

		if (!delay_base_initialized) {
			// 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 (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
				// if we don't have a value, set it to the current sample
				delay_base_hist[i] = sample;
				continue;
			}
			delay_base = sample;
			delay_base_initialized = true;
		}

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

		// is sample lower than delay_base? If so, update delay_base
		if (wrapping_compare_less(sample, delay_base)) {
			// sample is smaller than the current delay_base
			// update it
			delay_base = sample;
		}
		
		// this operation may wrap, and is supposed to
		const uint32 delay = sample - 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);

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

		// once every minute
		if (g_current_ms - delay_base_time > 60 * 1000) {
			delay_base_time = g_current_ms;
			delay_base_idx = (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 
			delay_base_hist[delay_base_idx] = sample;
			delay_base = delay_base_hist[0];
			// Assign the lowest delay in the last 2 minutes to delay_base
			for (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
				if (wrapping_compare_less(delay_base_hist[i], delay_base))
					delay_base = delay_base_hist[i];
			}
		}
	}

	uint32 get_value()
	{
		uint32 value = UINT_MAX;
		for (size_t i = 0; i < CUR_DELAY_SIZE; i++) {
			value = min<uint32>(cur_delay_hist[i], value);
		}
		// value could be UINT_MAX if we have no samples yet...
		return value;
	}
};

struct UTPSocket {
	PackedSockAddr addr;

	uint 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:1;
	// Timeout procedure
	bool fast_timeout:1;

	// max receive window for other end, in bytes
	size_t max_window_user;
	// 0 = original uTP header, 1 = second revision
	byte version;
	uint16 mtu;
	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;
	UTPFunctionTable func;
	void *userdata;

	// Round trip time
	uint rtt;
	// Round trip time variance
	uint rtt_var;
	// Round trip timeout
	uint rto;
	DelayHist 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;

	// these two members are used for determining the send rate.
	// the time stamp is the time when bytes_sent was reset to
	// 0, and bytes_sent is the global number of uTP bytes sent
	// the last time we did this check. The average send rate is
	// calculated by:
	// (get_global_utp_bytes_sent() - bytes_sent) / time_since_last_check
	// The interval defaults to 5 seconds
	uint64 bytes_sent;
	uint32 bytes_sent_timestamp;

	// used when determining the packet size
	uint64 delay_sum_ms;
	int num_delay_samples;

	// the max number of bytes in a packet that is sent
	size_t packet_size;

	DelayHist our_hist;
	DelayHist their_hist;

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

	SizableCircularBuffer inbuf, outbuf;

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

	// Calculates the current receive window
	size_t get_rcv_window() const
	{
		// If we don't have a connection (such as during connection
		// establishment, always act as if we have an empty buffer).
		if (!userdata) return opt_rcvbuf;

		// Trim window down according to what's already in buffer.
		const size_t numbuf = func.get_rb_size(userdata);
		assert((int)numbuf >= 0);
		return opt_rcvbuf > numbuf ? 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.
	bool can_decay_win(int32 msec) const
	{
		return msec - last_rwin_decay >= MAX_WINDOW_DECAY;
	}

	// If we can, decay max window, returns true if we actually did so
	void maybe_decay_win()
	{
		if (can_decay_win(g_current_ms)) {
			// TCP uses 0.5, but we also back off on large
			// delay, so this is just a safety net
			max_window *= .78;
			last_rwin_decay = g_current_ms;
			if (max_window < MIN_WINDOW_SIZE)
				max_window = MIN_WINDOW_SIZE;
		}
	}

	size_t get_header_size() const
	{
		return (version ? sizeof(PacketFormatV1) : sizeof(PacketFormat));
	}

	size_t get_header_extensions_size() const
	{
		return (version ? sizeof(PacketFormatExtensionsV1) : sizeof(PacketFormatExtensions));
	}

	size_t get_max_payload_size() const
	{
		return mtu - get_header_size();
	}

	size_t get_initial_payload_size() const
	{
		if (!DYNAMIC_PACKET_SIZE_ENABLED) {
			return get_max_payload_size();
		}
		const size_t s = MIN_PACKET_SIZE
			* clamp(DYNAMIC_PACKET_SIZE_FACTOR, 1, 8);
		assert(s <= get_max_payload_size());
		return s;
	}

	void sent_ack()
	{
		ack_time = g_current_ms + 0x70000000;
		bytes_since_ack = 0;
	}

	size_t get_udp_mtu() const
	{
		socklen_t len;
		SOCKADDR_STORAGE sa = addr.get_sockaddr_storage(&len);
		return UTP_GetUDPMTU((const struct sockaddr *)&sa, len);
	}

	size_t get_udp_overhead() const
	{
		socklen_t len;
		SOCKADDR_STORAGE sa = addr.get_sockaddr_storage(&len);
		return UTP_GetUDPOverhead((const struct sockaddr *)&sa, len);
	}

	uint64 get_global_utp_bytes_sent() const
	{
		socklen_t len;
		SOCKADDR_STORAGE sa = addr.get_sockaddr_storage(&len);
		return UTP_GetGlobalUTPBytesSent((const struct sockaddr *)&sa, len);
	}

	size_t get_overhead() const
	{
		return get_udp_overhead() + get_header_size();
	}

	void send_data(PacketFormat* b, size_t length, bandwidth_type_t type);

	void send_ack(bool synack = false);

	void send_keep_alive();

	static void send_rst(SendToProc *send_to_proc, void *send_to_userdata,
						 const PackedSockAddr &addr, uint32 conn_id_send,
						 uint16 ack_nr, uint16 seq_nr, byte version);

	void send_packet(OutgoingPacket *pkt);

	bool is_writable(size_t to_write);

	bool flush_packets();

	void write_outgoing_packet(size_t payload, uint flags);

	void update_send_quota();

#ifdef _DEBUG
	void check_invariant();
#endif

	void check_timeouts();

	int ack_packet(uint seq);

	int selective_ack_bytes(uint base, const byte* mask, uint len, int64& min_rtt);

	void selective_ack(uint base, const byte *mask, uint len);

	void apply_ledbat_ccontrol(uint bytes_acked, uint32 actual_delay, int64 min_rtt);
};

Array<RST_Info> g_rst_info;
Array<UTPSocket*> g_utp_sockets;

void send_to_addr(SendToProc *send_to_proc, void *send_to_userdata, const byte *p, size_t len, const PackedSockAddr &addr)
{
	socklen_t tolen;
	SOCKADDR_STORAGE to = addr.get_sockaddr_storage(&tolen);
	send_to_proc(send_to_userdata, p, len, (const struct sockaddr *)&to, tolen);
}

void UTPSocket::send_data(PacketFormat* b, size_t length, bandwidth_type_t type)
{
	// 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
	uint64 time = UTP_GetMicroseconds();

	PacketFormatV1* b1 = (PacketFormatV1*)b;
	if (version == 0) {
		b->tv_sec = time / 1000000;
		b->tv_usec = time % 1000000;
		b->reply_micro = reply_micro;
	} else {
		b1->tv_usec = time;
		b1->reply_micro = reply_micro;
	}

	last_sent_packet = g_current_ms;

#ifdef _DEBUG
	_stats._nbytes_xmit += length;
	++_stats._nxmit;
#endif
	if (userdata) {
		size_t n;
		if (type == payload_bandwidth) {
			// if this packet carries payload, just
			// count the header as overhead
			type = header_overhead;
			n = get_overhead();
		} else {
			n = length + get_udp_overhead();
		}
		func.on_overhead(userdata, true, n, type);
	}
#if g_log_utp_verbose
	int flags = version == 0 ? b->flags : b1->type;
	int seq_nr = version == 0 ? b->seq_nr : b1->seq_nr;
	int ack_nr = version == 0 ? b->ack_nr : b1->ack_nr;
	LOG_UTPV("0x%08x: send %s len:%u id:%u timestamp:"I64u" reply_micro:%u flags:%s seq_nr:%d, ack_nr:%d",
			 this, addrfmt(addr, addrbuf), length, conn_id_send, time, reply_micro, flagnames[flags],
			 seq_nr, ack_nr);
#endif
	send_to_addr(send_to_proc, send_to_userdata, (const byte*)b, length, addr);
}

void UTPSocket::send_ack(bool synack)
{
	PacketFormatExtensions pfe;
	zeromem(&pfe);
	PacketFormatExtensionsV1& pfe1 = (PacketFormatExtensionsV1&)pfe;
	PacketFormatAck& pfa = (PacketFormatAck&)pfe1;
	PacketFormatAckV1& pfa1 = (PacketFormatAckV1&)pfe1;

	uint len;
	last_rcv_win = get_rcv_window();
	if (version == 0) {
		pfa.pf.connid = conn_id_send;
		pfa.pf.ack_nr = (uint16)ack_nr;
		pfa.pf.seq_nr = (uint16)seq_nr;
		pfa.pf.flags = ST_STATE;
		pfa.pf.ext = 0;
		pfa.pf.windowsize = DIV_ROUND_UP(last_rcv_win, PACKET_SIZE);
		len = sizeof(PacketFormat);
	} else {
		pfa1.pf.version = 1;
		pfa1.pf.type = ST_STATE;
		pfa1.pf.ext = 0;
		pfa1.pf.connid = conn_id_send;
		pfa1.pf.ack_nr = ack_nr;
		pfa1.pf.seq_nr = seq_nr;
		pfa1.pf.windowsize = last_rcv_win;
		len = sizeof(PacketFormatV1);
	}

	// we never need to send EACK for connections
	// that are shutting down
	if (reorder_count != 0 && state < CS_GOT_FIN) {
		// 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 (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;
		}
		uint m = 0;

		// reorder count should only be non-zero
		// if the packet ack_nr + 1 has not yet
		// been received
		assert(inbuf.get(ack_nr + 1) == NULL);
		int window = min<int>(14+16, inbuf.size());
		// Generate bit mask of segments received.
		for (uint i = 0; i < window; i++) {
			if (inbuf.get(ack_nr + i + 2) != NULL) {
				m |= 1 << i;
				LOG_UTPV("0x%08x: EACK packet [%u]", this, ack_nr + i + 2);
			}
		}
		if (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 %d [%u] bits:[%032b]", this, ack_nr, 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 %d [%u] with extension bits", this, ack_nr, conn_id_send);
		if (version == 0) {
			pfe.pf.ext = 2;
			pfe.ext_next = 0;
			pfe.ext_len = 8;
			memset(pfe.extensions, 0, 8);
		} else {
			pfe1.pf.ext = 2;
			pfe1.ext_next = 0;
			pfe1.ext_len = 8;
			memset(pfe1.extensions, 0, 8);
		}
		len += 8 + 2;
	} else {
		LOG_UTPV("0x%08x: Sending ACK %d [%u]", this, ack_nr, conn_id_send);
	}

	sent_ack();
	send_data((PacketFormat*)&pfe, len, ack_overhead);
}

void UTPSocket::send_keep_alive()
{
	ack_nr--;
	LOG_UTPV("0x%08x: Sending KeepAlive ACK %d [%u]", this, ack_nr, conn_id_send);
	send_ack();
	ack_nr++;
}

void UTPSocket::send_rst(SendToProc *send_to_proc, void *send_to_userdata,
						 const PackedSockAddr &addr, uint32 conn_id_send, uint16 ack_nr, uint16 seq_nr, byte version)
{
	PacketFormat pf;
	zeromem(&pf);
	PacketFormatV1& pf1 = (PacketFormatV1&)pf;

	size_t len;
	if (version == 0) {
		pf.connid = conn_id_send;
		pf.ack_nr = ack_nr;
		pf.seq_nr = seq_nr;
		pf.flags = ST_RESET;
		pf.ext = 0;
		pf.windowsize = 0;
		len = sizeof(PacketFormat);
	} else {
		pf1.version = 1;
		pf1.type= ST_RESET;
		pf1.ext = 0;
		pf1.connid = conn_id_send;
		pf1.ack_nr = ack_nr;
		pf1.seq_nr = seq_nr;
		pf1.windowsize = 0;
		len = sizeof(PacketFormatV1);
	}

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

void UTPSocket::send_packet(OutgoingPacket *pkt)
{
	// 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)
	size_t max_send = min(max_window, opt_sndbuf, max_window_user);

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

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

	pkt->need_resend = false;

	PacketFormatV1* p1 = (PacketFormatV1*)pkt->data;
	PacketFormat* p = (PacketFormat*)pkt->data;
	if (version == 0) {
		p->ack_nr = ack_nr;
	} else {
		p1->ack_nr = ack_nr;
	}
	pkt->time_sent = UTP_GetMicroseconds();
	pkt->transmissions++;
	send_data((PacketFormat*)pkt->data, pkt->length,
			  pkt->transmissions == 1 ? payload_bandwidth : retransmit_overhead);
}

bool UTPSocket::is_writable(size_t to_write)
{
	// 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.
	size_t max_send = min(max_window, opt_sndbuf, max_window_user);

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

	// if we don't have enough quota, we can't write regardless
	if (USE_PACKET_PACING) {
		if (send_quota / 100 < to_write) return false;
	}

	// subtract one to save space for the FIN packet
	if (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 (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 (max_window < to_write &&
			cur_window < max_window &&
			cur_window_packets == 0) {
			return true;
		}
	}

	return false;
}

bool UTPSocket::flush_packets()
{
	// 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 (uint16 i = seq_nr - cur_window_packets; i != seq_nr; ++i) {
		OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(i);
		if (pkt == 0 || (pkt->transmissions > 0 && pkt->need_resend == false)) continue;
		// have we run out of quota?
		if (!is_writable(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 != ((seq_nr - 1) & ACK_NR_MASK) ||
			cur_window_packets == 1 ||
			pkt->payload >= packet_size) {
			send_packet(pkt);

			// No need to send another ack if there is nothing to reorder.
			if (reorder_count == 0) {
				sent_ack();
			}
		}
	}
	return false;
}

void UTPSocket::write_outgoing_packet(size_t payload, uint flags)
{
	// Setup initial timeout timer
	if (cur_window_packets == 0) {
		retransmit_timeout = rto;
		rto_timeout = g_current_ms + retransmit_timeout;
		assert(cur_window == 0);
	}

	do {
		assert(cur_window_packets < OUTGOING_BUFFER_MAX_SIZE);
		assert(flags == ST_DATA || flags == ST_FIN);

		uint added = 0;

		OutgoingPacket *pkt = NULL;
		
		if (cur_window_packets > 0) {
			pkt = (OutgoingPacket*)outbuf.get(seq_nr - 1);
		}

		const size_t header_size = get_header_size();
		assert(header_size + payload <= mtu);
		bool 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 = min(payload + pkt->payload, max<size_t>(packet_size, pkt->payload)) - pkt->payload;
			pkt = (OutgoingPacket*)realloc(pkt,
										   (sizeof(OutgoingPacket) - 1) +
										   header_size +
										   pkt->payload + added);
			outbuf.put(seq_nr - 1, pkt);
			append = false;
			assert(!pkt->need_resend);
		} else {
			// Create the packet to send.
			added = payload;
			pkt = (OutgoingPacket*)malloc((sizeof(OutgoingPacket) - 1) +
										  header_size +
										  added);
			pkt->payload = 0;
			pkt->transmissions = 0;
			pkt->need_resend = false;
		}

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

		last_rcv_win = get_rcv_window();

		PacketFormat* p = (PacketFormat*)pkt->data;
		PacketFormatV1* p1 = (PacketFormatV1*)pkt->data;
		if (version == 0) {
			p->connid = conn_id_send;
			p->ext = 0;
			p->windowsize = DIV_ROUND_UP(last_rcv_win, PACKET_SIZE);
			p->ack_nr = ack_nr;
			p->flags = flags;
		} else {
			p1->version = 1;
			p1->type = flags;
			p1->ext = 0;
			p1->connid = conn_id_send;
			p1->windowsize = last_rcv_win;
			p1->ack_nr = ack_nr;
		}

		if (append) {
			// Remember the message in the outgoing queue.
			outbuf.ensure_size(seq_nr, cur_window_packets);
			outbuf.put(seq_nr, pkt);
			if (version == 0) p->seq_nr = seq_nr;
			else p1->seq_nr = seq_nr;
			seq_nr++;
			cur_window_packets++;
		}

		payload -= added;

	} while (payload);

	flush_packets();
}

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

#ifdef _DEBUG
void UTPSocket::check_invariant()
{
	if (reorder_count > 0) {
		assert(inbuf.get(ack_nr + 1) == NULL);
	}

	int outstanding_bytes = 0;
	for (int i = 0; i < cur_window_packets; ++i) {
		OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - i - 1);
		if (pkt == 0 || pkt->transmissions == 0 || pkt->need_resend) continue;
		outstanding_bytes += pkt->payload;
	}
	assert(outstanding_bytes == cur_window);
}
#endif

void UTPSocket::check_timeouts()
{
#ifdef _DEBUG
	check_invariant();
#endif

	// this invariant should always be true
	assert(cur_window_packets == 0 || outbuf.get(seq_nr - cur_window_packets));

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

	update_send_quota();
	flush_packets();


	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 (state == CS_CONNECTED_FULL && is_writable(packet_size)) {
			state = CS_CONNECTED;
			LOG_UTPV("0x%08x: Socket writable. max_window:%d cur_window:%d quota:%d packet_size:%d",
					 this, max_window, cur_window, send_quota / 100, packet_size);
			func.on_state(userdata, UTP_STATE_WRITABLE);
		}
	}

	int ms_since_rate_check = g_current_ms - bytes_sent_timestamp;
	switch (state) {
	case CS_SYN_SENT:
	case CS_CONNECTED_FULL:
	case CS_CONNECTED:
	case CS_FIN_SENT: {

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

		if (int(g_current_ms - rto_timeout) >= 0 &&
			(!(USE_PACKET_PACING) || cur_window_packets > 0) &&
			rto_timeout > 0) {

			/*
			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
			const uint new_timeout = retransmit_timeout * 2;
			if (new_timeout >= 30000) {
				// more than 30 seconds with no reply. kill it.
				if (state == CS_FIN_SENT)
					state = CS_DESTROY;
				else
					state = CS_RESET;
				func.on_error(userdata, ETIMEDOUT);
				goto getout;
			}

			retransmit_timeout = new_timeout;
			rto_timeout = g_current_ms + new_timeout;

			// On Timeout
			duplicate_ack = 0;

			// rate = min_rate
			max_window = packet_size;
			send_quota = max<int>(packet_size * 100, send_quota);

			// every packet should be considered lost
			for (int i = 0; i < cur_window_packets; ++i) {
				OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - i - 1);
				if (pkt == 0 || pkt->transmissions == 0 || pkt->need_resend) continue;
				pkt->need_resend = true;
				assert(cur_window >= pkt->payload);
				cur_window -= pkt->payload;
			}

			// used in parse_log.py
			LOG_UTP("0x%08x: Packet timeout. Resend. seq_nr:%d. timeout:%d max_window:%d",
					this, seq_nr - cur_window_packets, retransmit_timeout, max_window);

			fast_timeout = true;
			timeout_seq_nr = seq_nr;

			if (cur_window_packets > 0) {
				OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - cur_window_packets);
				assert(pkt);
				send_quota = max<int>(pkt->length * 100, send_quota);

				// Re-send the packet.
				send_packet(pkt);
			}
		}

		// Mark the socket as writable
		if (state == CS_CONNECTED_FULL && is_writable(packet_size)) {
			state = CS_CONNECTED;
			LOG_UTPV("0x%08x: Socket writable. max_window:%d cur_window:%d quota:%d packet_size:%d",
					 this, max_window, cur_window, send_quota / 100, packet_size);
			func.on_state(userdata, UTP_STATE_WRITABLE);
		}

		const size_t max_payload = get_max_payload_size();
		if (!DYNAMIC_PACKET_SIZE_ENABLED) {
			packet_size = max_payload;
		} else if (ms_since_rate_check > RATE_CHECK_INTERVAL) {
			// this is the average send rate of this socket for the
			// last RATE_CHECK_INTERVAL milliseconds
			const uint64 sent_total = get_global_utp_bytes_sent();
			const int rate = (sent_total - bytes_sent) * 1000 / ms_since_rate_check;
			assert(rate >= 0);

			// average delay during these last 10 seconds
			const int average_delay = num_delay_samples ? delay_sum_ms / num_delay_samples : -1;
			num_delay_samples = 0;
			delay_sum_ms = 0;

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

			LOG_UTPV("0x%08x: rate check. time:%d bytes:%d rate:%d packet_size:%d target_delay:%d",
					 this, ms_since_rate_check, int(sent_total - bytes_sent), rate,
					 packet_size, target / 1000);

			// use the rate to figure out if we should change the packet size
			// one window should have at least 3 packets, so don't double
			// the packet size if the current window can't fit 6 packets

			bool should_increase_packet_size
				= rate > SERIALIZATIONS_PER_TARGET * packet_size * 1000 / target
//				&& max_window / packet_size > 5
				;

			// if the average delay has been very low
			// we can afford to increase the packet size
			should_increase_packet_size = should_increase_packet_size
				|| (average_delay != -1 && average_delay < target / 3);

			// if we're already at max, we can't increase the packet size
			should_increase_packet_size = should_increase_packet_size && (packet_size < max_payload);


			bool should_decrease_packet_size
				= rate < SERIALIZATIONS_PER_TARGET * packet_size * 1000 / 4 / target
//				|| max_window / packet_size < 3
				;

			// don't decrease if we're already at minimum
			should_decrease_packet_size = should_decrease_packet_size && packet_size > MIN_PACKET_SIZE;
			if (should_increase_packet_size) {
				packet_size *= 2;
				if (packet_size > max_payload) {
					packet_size = max_payload;
				}
			} else if (should_decrease_packet_size) {
				if (packet_size == max_payload) {
					packet_size = PACKET_SIZE * 4;
					assert(packet_size < max_payload);
				} else {
					packet_size /= 2;
				}
			}

			bytes_sent = get_global_utp_bytes_sent();

			bytes_sent_timestamp = g_current_ms;
		}

		if (state >= CS_CONNECTED && state <= CS_FIN_SENT) {
			// Send acknowledgment packets periodically, or when the threshold is reached
			if (bytes_since_ack > DELAYED_ACK_BYTE_THRESHOLD ||
				(int)(g_current_ms - ack_time) >= 0) {<commit_msg>fixed broken packet size logic to work it was meant to be (with division by zero fix)<commit_after>#include "StdAfx.h"

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

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

#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 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.
#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 300 // packet size should be divisible by 2
// This must be (PACKET_SIZE / (2^n))
// This should be fairly small, it makes it faster to
// start back up again after the window size has been
// set to MIN_WINDOW_SIZE, since we don't need to wait
// as long in between packets the smaller the packet is
#define MIN_PACKET_SIZE (PACKET_SIZE/2)

// This is a limit describing when to increase or decrease the
// packet size. When this limit is exceeded, we double the packet
// size. When the metric drops below one quarter of this limit
// the packet size is halved.
// The unit of this limit is the number of packets we can send
// per target delay. i.e. if the target delay is 100 ms, the packet
// size is 300 bytes, and the rate is 10 kB/s. That means we're
// sending 10 * 1000 / 300 = 33.3 packets per second, or 3.3 packets
// per target delay (100 ms).
#define SERIALIZATIONS_PER_TARGET 4

// 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
#define DELAYED_ACK_TIME_THRESHOLD 5

#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) x.fmt(s, sizeof(s))

#pragma pack(push,1)

struct PackedSockAddr {

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

	// Host byte order
	uint16 _port;

#define _sin4 _in._in6d[3]	// IPv4 is stored where it goes if mapped

#define _sin6 _in._in6
#define _sin6w _in._in6w
#define _sin6d _in._in6d

	byte get_family() const
	{
		return (IN6_IS_ADDR_V4MAPPED((in6_addr*)_sin6) != 0) ? AF_INET : AF_INET6;
	}

	bool operator==(const PackedSockAddr& rhs) const
	{
		if (&rhs == this)
			return true;
		if (_port != rhs._port)
			return false;
		return memcmp(_sin6, rhs._sin6, sizeof(_sin6)) == 0;
	}
	bool operator!=(const PackedSockAddr& rhs) const { return !(*this == rhs); }

	PackedSockAddr(const SOCKADDR_STORAGE* sa, socklen_t len)
	{
		if (sa->ss_family == AF_INET) {
			assert(len >= sizeof(sockaddr_in));
			const sockaddr_in *sin = (sockaddr_in*)sa;
			_sin6w[0] = 0;
			_sin6w[1] = 0;
			_sin6w[2] = 0;
			_sin6w[3] = 0;
			_sin6w[4] = 0;
			_sin6w[5] = 0xffff;
			_sin4 = sin->sin_addr.s_addr;
			_port = ntohs(sin->sin_port);
		} else {
			assert(len >= sizeof(sockaddr_in6));
			const sockaddr_in6 *sin6 = (sockaddr_in6*)sa;
			_in._in6addr = sin6->sin6_addr;
			_port = ntohs(sin6->sin6_port);
		}
		assert(sa->ss_family == get_family());
	}

	SOCKADDR_STORAGE get_sockaddr_storage(socklen_t *len = NULL) const
	{
		SOCKADDR_STORAGE sa;
		const byte family = get_family();
		if (family == AF_INET) {
			sockaddr_in *sin = (sockaddr_in*)&sa;
			if (len) *len = sizeof(sockaddr_in);
			memset(sin, 0, sizeof(sockaddr_in));
			sin->sin_family = family;
			sin->sin_port = htons(_port);
			sin->sin_addr.s_addr = _sin4;
		} else {
			sockaddr_in6 *sin6 = (sockaddr_in6*)&sa;
			memset(sin6, 0, sizeof(sockaddr_in6));
			if (len) *len = sizeof(sockaddr_in6);
			sin6->sin6_family = family;
			sin6->sin6_addr = _in._in6addr;
			sin6->sin6_port = htons(_port);
		}
		return sa;
	}

	cstr fmt(str s, size_t len) const
	{
		memset(s, 0, len);
		const byte family = get_family();
		str i;
		if (family == AF_INET) {
			inet_ntop(family, (uint32*)&_sin4, s, len);
			i = s;
			while (*++i) {}
		} else {
			i = s;
			*i++ = '[';
			inet_ntop(family, (in6_addr*)&_in._in6addr, i, len-1);
			while (*++i) {}
			*i++ = ']';
		}
		snprintf(i, len - (i-s), ":%u", _port);
		return s;
	}
};

struct RST_Info {
	PackedSockAddr addr;
	uint32 connid;
	uint32 timestamp;
	uint16 ack_nr;
};

struct PacketFormat {
	// 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 PacketFormatAck {
	PacketFormat pf;
	byte ext_next;
	byte ext_len;
	byte acks[4];
};

struct PacketFormatExtensions {
	PacketFormat pf;
	byte ext_next;
	byte ext_len;
	byte extensions[8];
};

struct PacketFormatV1 {
	// protocol version
	byte version:4;
	// type (formerly flags)
	byte type:4;
	// 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;
};

struct PacketFormatAckV1 {
	PacketFormatV1 pf;
	byte ext_next;
	byte ext_len;
	byte acks[4];
};

struct PacketFormatExtensionsV1 {
	PacketFormatV1 pf;
	byte ext_next;
	byte ext_len;
	byte extensions[8];
};

#pragma pack(pop)

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

static const cstr flagnames[] = {
	"ST_DATA","ST_FIN","ST_STATE","ST_RESET","ST_SYN"
};

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,
};

static const cstr statenames[] = {
	"IDLE","SYN_SENT","CONNECTED","CONNECTED_FULL","GOT_FIN","DESTROY_DELAY","FIN_SENT","RESET","DESTROY"
};

struct OutgoingPacket {
	uint length;
	uint payload;
	uint64 time_sent; // microseconds
	uint transmissions:31;
	bool need_resend:1;
	byte data[1];
};

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

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

struct SizableCircularBuffer {
	// This is the mask. Since it's always a power of 2, adding 1 to this value will return the size.
	uint mask;
	// This is the elements that the circular buffer points to
	void **elements;

	void *get(uint i) { assert(elements); return elements ? elements[i & mask] : NULL; }
	void put(uint i, void *data) { assert(elements); elements[i&mask] = data; }

	void grow(uint item, uint index);
	void ensure_size(uint item, uint index) { if (index > mask) grow(item, index); }
	size_t size() { return mask + 1; }
};

// Item contains the element we want to make space for
// index is the index in the list.
void SizableCircularBuffer::grow(uint item, uint index)
{
	// Figure out the new size.
	uint size = mask + 1;
	do size *= 2; while (index >= size);

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

	size--;

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

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

// 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
bool wrapping_compare_less(uint32 lhs, uint32 rhs)
{
	// distance walking from lhs to rhs, downwards
	const uint32 dist_down = lhs - rhs;
	// distance walking from lhs to rhs, upwards
	const uint32 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 DelayHist {
	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;

	void clear()
	{
		delay_base_initialized = false;
		delay_base = 0;
		cur_delay_idx = 0;
		delay_base_idx = 0;
		delay_base_time = g_current_ms;
		for (size_t i = 0; i < CUR_DELAY_SIZE; i++) {
			cur_delay_hist[i] = 0;
		}
		for (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
			delay_base_hist[i] = 0;
		}
	}

	void shift(const uint32 offset)
	{
		// 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 (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
			delay_base_hist[i] += offset;
		}
		delay_base += offset;
	}

	void add_sample(const 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

		if (!delay_base_initialized) {
			// 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 (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
				// if we don't have a value, set it to the current sample
				delay_base_hist[i] = sample;
				continue;
			}
			delay_base = sample;
			delay_base_initialized = true;
		}

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

		// is sample lower than delay_base? If so, update delay_base
		if (wrapping_compare_less(sample, delay_base)) {
			// sample is smaller than the current delay_base
			// update it
			delay_base = sample;
		}
		
		// this operation may wrap, and is supposed to
		const uint32 delay = sample - 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);

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

		// once every minute
		if (g_current_ms - delay_base_time > 60 * 1000) {
			delay_base_time = g_current_ms;
			delay_base_idx = (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 
			delay_base_hist[delay_base_idx] = sample;
			delay_base = delay_base_hist[0];
			// Assign the lowest delay in the last 2 minutes to delay_base
			for (size_t i = 0; i < DELAY_BASE_HISTORY; i++) {
				if (wrapping_compare_less(delay_base_hist[i], delay_base))
					delay_base = delay_base_hist[i];
			}
		}
	}

	uint32 get_value()
	{
		uint32 value = UINT_MAX;
		for (size_t i = 0; i < CUR_DELAY_SIZE; i++) {
			value = min<uint32>(cur_delay_hist[i], value);
		}
		// value could be UINT_MAX if we have no samples yet...
		return value;
	}
};

struct UTPSocket {
	PackedSockAddr addr;

	uint 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:1;
	// Timeout procedure
	bool fast_timeout:1;

	// max receive window for other end, in bytes
	size_t max_window_user;
	// 0 = original uTP header, 1 = second revision
	byte version;
	uint16 mtu;
	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;
	UTPFunctionTable func;
	void *userdata;

	// Round trip time
	uint rtt;
	// Round trip time variance
	uint rtt_var;
	// Round trip timeout
	uint rto;
	DelayHist 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;

	// these two members are used for determining the send rate.
	// the time stamp is the time when bytes_sent was reset to
	// 0, and bytes_sent is the global number of uTP bytes sent
	// the last time we did this check. The average send rate is
	// calculated by:
	// (get_global_utp_bytes_sent() - bytes_sent) / time_since_last_check
	// The interval defaults to 5 seconds
	uint64 bytes_sent;
	uint32 bytes_sent_timestamp;

	// used when determining the packet size
	uint64 delay_sum_ms;
	int num_delay_samples;

	// the max number of bytes in a packet that is sent
	size_t packet_size;

	DelayHist our_hist;
	DelayHist their_hist;

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

	SizableCircularBuffer inbuf, outbuf;

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

	// Calculates the current receive window
	size_t get_rcv_window() const
	{
		// If we don't have a connection (such as during connection
		// establishment, always act as if we have an empty buffer).
		if (!userdata) return opt_rcvbuf;

		// Trim window down according to what's already in buffer.
		const size_t numbuf = func.get_rb_size(userdata);
		assert((int)numbuf >= 0);
		return opt_rcvbuf > numbuf ? 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.
	bool can_decay_win(int32 msec) const
	{
		return msec - last_rwin_decay >= MAX_WINDOW_DECAY;
	}

	// If we can, decay max window, returns true if we actually did so
	void maybe_decay_win()
	{
		if (can_decay_win(g_current_ms)) {
			// TCP uses 0.5, but we also back off on large
			// delay, so this is just a safety net
			max_window *= .78;
			last_rwin_decay = g_current_ms;
			if (max_window < MIN_WINDOW_SIZE)
				max_window = MIN_WINDOW_SIZE;
		}
	}

	size_t get_header_size() const
	{
		return (version ? sizeof(PacketFormatV1) : sizeof(PacketFormat));
	}

	size_t get_header_extensions_size() const
	{
		return (version ? sizeof(PacketFormatExtensionsV1) : sizeof(PacketFormatExtensions));
	}

	size_t get_max_payload_size() const
	{
		return mtu - get_header_size();
	}

	size_t get_initial_payload_size() const
	{
		if (!DYNAMIC_PACKET_SIZE_ENABLED) {
			return get_max_payload_size();
		}
		const size_t s = MIN_PACKET_SIZE
			* clamp(DYNAMIC_PACKET_SIZE_FACTOR, 1, 8);
		assert(s <= get_max_payload_size());
		return s;
	}

	void sent_ack()
	{
		ack_time = g_current_ms + 0x70000000;
		bytes_since_ack = 0;
	}

	size_t get_udp_mtu() const
	{
		socklen_t len;
		SOCKADDR_STORAGE sa = addr.get_sockaddr_storage(&len);
		return UTP_GetUDPMTU((const struct sockaddr *)&sa, len);
	}

	size_t get_udp_overhead() const
	{
		socklen_t len;
		SOCKADDR_STORAGE sa = addr.get_sockaddr_storage(&len);
		return UTP_GetUDPOverhead((const struct sockaddr *)&sa, len);
	}

	uint64 get_global_utp_bytes_sent() const
	{
		socklen_t len;
		SOCKADDR_STORAGE sa = addr.get_sockaddr_storage(&len);
		return UTP_GetGlobalUTPBytesSent((const struct sockaddr *)&sa, len);
	}

	size_t get_overhead() const
	{
		return get_udp_overhead() + get_header_size();
	}

	void send_data(PacketFormat* b, size_t length, bandwidth_type_t type);

	void send_ack(bool synack = false);

	void send_keep_alive();

	static void send_rst(SendToProc *send_to_proc, void *send_to_userdata,
						 const PackedSockAddr &addr, uint32 conn_id_send,
						 uint16 ack_nr, uint16 seq_nr, byte version);

	void send_packet(OutgoingPacket *pkt);

	bool is_writable(size_t to_write);

	bool flush_packets();

	void write_outgoing_packet(size_t payload, uint flags);

	void update_send_quota();

#ifdef _DEBUG
	void check_invariant();
#endif

	void check_timeouts();

	int ack_packet(uint seq);

	int selective_ack_bytes(uint base, const byte* mask, uint len, int64& min_rtt);

	void selective_ack(uint base, const byte *mask, uint len);

	void apply_ledbat_ccontrol(uint bytes_acked, uint32 actual_delay, int64 min_rtt);
};

Array<RST_Info> g_rst_info;
Array<UTPSocket*> g_utp_sockets;

void send_to_addr(SendToProc *send_to_proc, void *send_to_userdata, const byte *p, size_t len, const PackedSockAddr &addr)
{
	socklen_t tolen;
	SOCKADDR_STORAGE to = addr.get_sockaddr_storage(&tolen);
	send_to_proc(send_to_userdata, p, len, (const struct sockaddr *)&to, tolen);
}

void UTPSocket::send_data(PacketFormat* b, size_t length, bandwidth_type_t type)
{
	// 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
	uint64 time = UTP_GetMicroseconds();

	PacketFormatV1* b1 = (PacketFormatV1*)b;
	if (version == 0) {
		b->tv_sec = time / 1000000;
		b->tv_usec = time % 1000000;
		b->reply_micro = reply_micro;
	} else {
		b1->tv_usec = time;
		b1->reply_micro = reply_micro;
	}

	last_sent_packet = g_current_ms;

#ifdef _DEBUG
	_stats._nbytes_xmit += length;
	++_stats._nxmit;
#endif
	if (userdata) {
		size_t n;
		if (type == payload_bandwidth) {
			// if this packet carries payload, just
			// count the header as overhead
			type = header_overhead;
			n = get_overhead();
		} else {
			n = length + get_udp_overhead();
		}
		func.on_overhead(userdata, true, n, type);
	}
#if g_log_utp_verbose
	int flags = version == 0 ? b->flags : b1->type;
	int seq_nr = version == 0 ? b->seq_nr : b1->seq_nr;
	int ack_nr = version == 0 ? b->ack_nr : b1->ack_nr;
	LOG_UTPV("0x%08x: send %s len:%u id:%u timestamp:"I64u" reply_micro:%u flags:%s seq_nr:%d, ack_nr:%d",
			 this, addrfmt(addr, addrbuf), length, conn_id_send, time, reply_micro, flagnames[flags],
			 seq_nr, ack_nr);
#endif
	send_to_addr(send_to_proc, send_to_userdata, (const byte*)b, length, addr);
}

void UTPSocket::send_ack(bool synack)
{
	PacketFormatExtensions pfe;
	zeromem(&pfe);
	PacketFormatExtensionsV1& pfe1 = (PacketFormatExtensionsV1&)pfe;
	PacketFormatAck& pfa = (PacketFormatAck&)pfe1;
	PacketFormatAckV1& pfa1 = (PacketFormatAckV1&)pfe1;

	uint len;
	last_rcv_win = get_rcv_window();
	if (version == 0) {
		pfa.pf.connid = conn_id_send;
		pfa.pf.ack_nr = (uint16)ack_nr;
		pfa.pf.seq_nr = (uint16)seq_nr;
		pfa.pf.flags = ST_STATE;
		pfa.pf.ext = 0;
		pfa.pf.windowsize = DIV_ROUND_UP(last_rcv_win, PACKET_SIZE);
		len = sizeof(PacketFormat);
	} else {
		pfa1.pf.version = 1;
		pfa1.pf.type = ST_STATE;
		pfa1.pf.ext = 0;
		pfa1.pf.connid = conn_id_send;
		pfa1.pf.ack_nr = ack_nr;
		pfa1.pf.seq_nr = seq_nr;
		pfa1.pf.windowsize = last_rcv_win;
		len = sizeof(PacketFormatV1);
	}

	// we never need to send EACK for connections
	// that are shutting down
	if (reorder_count != 0 && state < CS_GOT_FIN) {
		// 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 (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;
		}
		uint m = 0;

		// reorder count should only be non-zero
		// if the packet ack_nr + 1 has not yet
		// been received
		assert(inbuf.get(ack_nr + 1) == NULL);
		int window = min<int>(14+16, inbuf.size());
		// Generate bit mask of segments received.
		for (uint i = 0; i < window; i++) {
			if (inbuf.get(ack_nr + i + 2) != NULL) {
				m |= 1 << i;
				LOG_UTPV("0x%08x: EACK packet [%u]", this, ack_nr + i + 2);
			}
		}
		if (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 %d [%u] bits:[%032b]", this, ack_nr, 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 %d [%u] with extension bits", this, ack_nr, conn_id_send);
		if (version == 0) {
			pfe.pf.ext = 2;
			pfe.ext_next = 0;
			pfe.ext_len = 8;
			memset(pfe.extensions, 0, 8);
		} else {
			pfe1.pf.ext = 2;
			pfe1.ext_next = 0;
			pfe1.ext_len = 8;
			memset(pfe1.extensions, 0, 8);
		}
		len += 8 + 2;
	} else {
		LOG_UTPV("0x%08x: Sending ACK %d [%u]", this, ack_nr, conn_id_send);
	}

	sent_ack();
	send_data((PacketFormat*)&pfe, len, ack_overhead);
}

void UTPSocket::send_keep_alive()
{
	ack_nr--;
	LOG_UTPV("0x%08x: Sending KeepAlive ACK %d [%u]", this, ack_nr, conn_id_send);
	send_ack();
	ack_nr++;
}

void UTPSocket::send_rst(SendToProc *send_to_proc, void *send_to_userdata,
						 const PackedSockAddr &addr, uint32 conn_id_send, uint16 ack_nr, uint16 seq_nr, byte version)
{
	PacketFormat pf;
	zeromem(&pf);
	PacketFormatV1& pf1 = (PacketFormatV1&)pf;

	size_t len;
	if (version == 0) {
		pf.connid = conn_id_send;
		pf.ack_nr = ack_nr;
		pf.seq_nr = seq_nr;
		pf.flags = ST_RESET;
		pf.ext = 0;
		pf.windowsize = 0;
		len = sizeof(PacketFormat);
	} else {
		pf1.version = 1;
		pf1.type= ST_RESET;
		pf1.ext = 0;
		pf1.connid = conn_id_send;
		pf1.ack_nr = ack_nr;
		pf1.seq_nr = seq_nr;
		pf1.windowsize = 0;
		len = sizeof(PacketFormatV1);
	}

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

void UTPSocket::send_packet(OutgoingPacket *pkt)
{
	// 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)
	size_t max_send = min(max_window, opt_sndbuf, max_window_user);

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

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

	pkt->need_resend = false;

	PacketFormatV1* p1 = (PacketFormatV1*)pkt->data;
	PacketFormat* p = (PacketFormat*)pkt->data;
	if (version == 0) {
		p->ack_nr = ack_nr;
	} else {
		p1->ack_nr = ack_nr;
	}
	pkt->time_sent = UTP_GetMicroseconds();
	pkt->transmissions++;
	send_data((PacketFormat*)pkt->data, pkt->length,
			  pkt->transmissions == 1 ? payload_bandwidth : retransmit_overhead);
}

bool UTPSocket::is_writable(size_t to_write)
{
	// 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.
	size_t max_send = min(max_window, opt_sndbuf, max_window_user);

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

	// if we don't have enough quota, we can't write regardless
	if (USE_PACKET_PACING) {
		if (send_quota / 100 < to_write) return false;
	}

	// subtract one to save space for the FIN packet
	if (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 (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 (max_window < to_write &&
			cur_window < max_window &&
			cur_window_packets == 0) {
			return true;
		}
	}

	return false;
}

bool UTPSocket::flush_packets()
{
	// 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 (uint16 i = seq_nr - cur_window_packets; i != seq_nr; ++i) {
		OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(i);
		if (pkt == 0 || (pkt->transmissions > 0 && pkt->need_resend == false)) continue;
		// have we run out of quota?
		if (!is_writable(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 != ((seq_nr - 1) & ACK_NR_MASK) ||
			cur_window_packets == 1 ||
			pkt->payload >= packet_size) {
			send_packet(pkt);

			// No need to send another ack if there is nothing to reorder.
			if (reorder_count == 0) {
				sent_ack();
			}
		}
	}
	return false;
}

void UTPSocket::write_outgoing_packet(size_t payload, uint flags)
{
	// Setup initial timeout timer
	if (cur_window_packets == 0) {
		retransmit_timeout = rto;
		rto_timeout = g_current_ms + retransmit_timeout;
		assert(cur_window == 0);
	}

	do {
		assert(cur_window_packets < OUTGOING_BUFFER_MAX_SIZE);
		assert(flags == ST_DATA || flags == ST_FIN);

		uint added = 0;

		OutgoingPacket *pkt = NULL;
		
		if (cur_window_packets > 0) {
			pkt = (OutgoingPacket*)outbuf.get(seq_nr - 1);
		}

		const size_t header_size = get_header_size();
		assert(header_size + payload <= mtu);
		bool 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 = min(payload + pkt->payload, max<size_t>(packet_size, pkt->payload)) - pkt->payload;
			pkt = (OutgoingPacket*)realloc(pkt,
										   (sizeof(OutgoingPacket) - 1) +
										   header_size +
										   pkt->payload + added);
			outbuf.put(seq_nr - 1, pkt);
			append = false;
			assert(!pkt->need_resend);
		} else {
			// Create the packet to send.
			added = payload;
			pkt = (OutgoingPacket*)malloc((sizeof(OutgoingPacket) - 1) +
										  header_size +
										  added);
			pkt->payload = 0;
			pkt->transmissions = 0;
			pkt->need_resend = false;
		}

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

		last_rcv_win = get_rcv_window();

		PacketFormat* p = (PacketFormat*)pkt->data;
		PacketFormatV1* p1 = (PacketFormatV1*)pkt->data;
		if (version == 0) {
			p->connid = conn_id_send;
			p->ext = 0;
			p->windowsize = DIV_ROUND_UP(last_rcv_win, PACKET_SIZE);
			p->ack_nr = ack_nr;
			p->flags = flags;
		} else {
			p1->version = 1;
			p1->type = flags;
			p1->ext = 0;
			p1->connid = conn_id_send;
			p1->windowsize = last_rcv_win;
			p1->ack_nr = ack_nr;
		}

		if (append) {
			// Remember the message in the outgoing queue.
			outbuf.ensure_size(seq_nr, cur_window_packets);
			outbuf.put(seq_nr, pkt);
			if (version == 0) p->seq_nr = seq_nr;
			else p1->seq_nr = seq_nr;
			seq_nr++;
			cur_window_packets++;
		}

		payload -= added;

	} while (payload);

	flush_packets();
}

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

#ifdef _DEBUG
void UTPSocket::check_invariant()
{
	if (reorder_count > 0) {
		assert(inbuf.get(ack_nr + 1) == NULL);
	}

	int outstanding_bytes = 0;
	for (int i = 0; i < cur_window_packets; ++i) {
		OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - i - 1);
		if (pkt == 0 || pkt->transmissions == 0 || pkt->need_resend) continue;
		outstanding_bytes += pkt->payload;
	}
	assert(outstanding_bytes == cur_window);
}
#endif

void UTPSocket::check_timeouts()
{
#ifdef _DEBUG
	check_invariant();
#endif

	// this invariant should always be true
	assert(cur_window_packets == 0 || outbuf.get(seq_nr - cur_window_packets));

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

	update_send_quota();
	flush_packets();


	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 (state == CS_CONNECTED_FULL && is_writable(packet_size)) {
			state = CS_CONNECTED;
			LOG_UTPV("0x%08x: Socket writable. max_window:%d cur_window:%d quota:%d packet_size:%d",
					 this, max_window, cur_window, send_quota / 100, packet_size);
			func.on_state(userdata, UTP_STATE_WRITABLE);
		}
	}

	int ms_since_rate_check = g_current_ms - bytes_sent_timestamp;
	switch (state) {
	case CS_SYN_SENT:
	case CS_CONNECTED_FULL:
	case CS_CONNECTED:
	case CS_FIN_SENT: {

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

		if (int(g_current_ms - rto_timeout) >= 0 &&
			(!(USE_PACKET_PACING) || cur_window_packets > 0) &&
			rto_timeout > 0) {

			/*
			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
			const uint new_timeout = retransmit_timeout * 2;
			if (new_timeout >= 30000) {
				// more than 30 seconds with no reply. kill it.
				if (state == CS_FIN_SENT)
					state = CS_DESTROY;
				else
					state = CS_RESET;
				func.on_error(userdata, ETIMEDOUT);
				goto getout;
			}

			retransmit_timeout = new_timeout;
			rto_timeout = g_current_ms + new_timeout;

			// On Timeout
			duplicate_ack = 0;

			// rate = min_rate
			max_window = packet_size;
			send_quota = max<int>(packet_size * 100, send_quota);

			// every packet should be considered lost
			for (int i = 0; i < cur_window_packets; ++i) {
				OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - i - 1);
				if (pkt == 0 || pkt->transmissions == 0 || pkt->need_resend) continue;
				pkt->need_resend = true;
				assert(cur_window >= pkt->payload);
				cur_window -= pkt->payload;
			}

			// used in parse_log.py
			LOG_UTP("0x%08x: Packet timeout. Resend. seq_nr:%d. timeout:%d max_window:%d",
					this, seq_nr - cur_window_packets, retransmit_timeout, max_window);

			fast_timeout = true;
			timeout_seq_nr = seq_nr;

			if (cur_window_packets > 0) {
				OutgoingPacket *pkt = (OutgoingPacket*)outbuf.get(seq_nr - cur_window_packets);
				assert(pkt);
				send_quota = max<int>(pkt->length * 100, send_quota);

				// Re-send the packet.
				send_packet(pkt);
			}
		}

		// Mark the socket as writable
		if (state == CS_CONNECTED_FULL && is_writable(packet_size)) {
			state = CS_CONNECTED;
			LOG_UTPV("0x%08x: Socket writable. max_window:%d cur_window:%d quota:%d packet_size:%d",
					 this, max_window, cur_window, send_quota / 100, packet_size);
			func.on_state(userdata, UTP_STATE_WRITABLE);
		}

		const size_t max_payload = get_max_payload_size();
		if (!DYNAMIC_PACKET_SIZE_ENABLED) {
			packet_size = max_payload;
		} else if (ms_since_rate_check > RATE_CHECK_INTERVAL) {
			// this is the average send rate of this socket for the
			// last RATE_CHECK_INTERVAL milliseconds
			const uint64 sent_total = get_global_utp_bytes_sent();
			const int rate = (sent_total - bytes_sent) * 1000 / ms_since_rate_check;
			assert(rate >= 0);

			// average delay during these last 10 seconds
			const int average_delay = num_delay_samples ? delay_sum_ms / num_delay_samples : -1;
			num_delay_samples = 0;
			delay_sum_ms = 0;

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

			LOG_UTPV("0x%08x: rate check. time:%d bytes:%d rate:%d packet_size:%d target_delay:%d",
					 this, ms_since_rate_check, int(sent_total - bytes_sent), rate,
					 packet_size, target / 1000);

			// use the rate to figure out if we should change the packet size
			// one window should have at least 3 packets, so don't double
			// the packet size if the current window can't fit 6 packets

			// each packet takes this many microseconds to send
			int microseconds_per_packet = uint64(packet_size) * 10000000 / (rate > 0 ? rate : 1);

			bool should_increase_packet_size = SERIALIZATIONS_PER_TARGET * microseconds_per_packet < target;

			// if the average delay has been very low
			// we can afford to increase the packet size
			should_increase_packet_size = should_increase_packet_size
				|| (average_delay != -1 && average_delay < target / 3);

			// if we're already at max, we can't increase the packet size
			should_increase_packet_size = should_increase_packet_size && (packet_size < max_payload);

			bool should_decrease_packet_size = SERIALIZATIONS_PER_TARGET * microseconds_per_packet > target * 4;

			// don't decrease if we're already at minimum
			should_decrease_packet_size = should_decrease_packet_size && packet_size > MIN_PACKET_SIZE;
			if (should_increase_packet_size) {
				packet_size *= 2;
				if (packet_size > max_payload) {
					packet_size = max_payload;
				}
			} else if (should_decrease_packet_size) {
				if (packet_size == max_payload) {
					packet_size = PACKET_SIZE * 4;
					assert(packet_size < max_payload);
				} else {
					packet_size /= 2;
				}
			}

			bytes_sent = get_global_utp_bytes_sent();

			bytes_sent_timestamp = g_current_ms;
		}

		if (state >= CS_CONNECTED && state <= CS_FIN_SENT) {
			// Send acknowledgment packets periodically, or when the threshold is reached
			if (bytes_since_ack > DELAYED_ACK_BYTE_THRESHOLD ||
				(int)(g_current_ms - ack_time) >= 0) {
{'fraction_non_alphanumeric': 0.10581353274635147, 'fraction_numerical': 0.01953925943794476, 'mean_word_length': 3.620026746907389, 'pattern_counts': {'":': 2, '<': 100, '<?xml version=': 0, '>': 291, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 5, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7157970', 'n_tokens_mistral': 33107, 'n_tokens_neox': 29632, 'n_words': 9204}