<commit_before>/*

Copyright (c) 2009, <NAME>
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

    * Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright
      notice, this list of conditions and the following disclaimer in
      the documentation and/or other materials provided with the distribution.
    * Neither the name of the author nor the names of its
      contributors may be used to endorse or promote products derived
      from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.

*/

#include "libtorrent/utp_stream.hpp"
#include "libtorrent/sliding_average.hpp"
#include "libtorrent/utp_socket_manager.hpp"
#include "libtorrent/alloca.hpp"
#include "libtorrent/timestamp_history.hpp"
#include <boost/cstdint.hpp>

#define TORRENT_UTP_LOG 1
#define TORRENT_UT_SEQ 1

#if TORRENT_UTP_LOG
#include <stdarg.h>
#include "libtorrent/socket_io.hpp"
#endif

namespace libtorrent {

#if TORRENT_UTP_LOG

char const* packet_type_names[] = { "ST_DATA", "ST_FIN", "ST_STATE", "ST_RESET", "ST_SYN" };
char const* socket_state_names[] = { "NONE", "SYN_SENT", "CONNECTED", "FIN_SENT", "ERROR", "DELETE" };

static struct utp_logger
{
	FILE* utp_log_file;

	utp_logger() : utp_log_file(0)
	{
		utp_log_file = fopen("utp.log", "w+");
	}
	~utp_logger()
	{
		if (utp_log_file) fclose(utp_log_file);
	}
} log_file_holder;

void utp_log(char const* fmt, ...)
{
	va_list l;
	va_start(l, fmt);
	vfprintf(log_file_holder.utp_log_file, fmt, l);
	va_end(l);
	fflush(log_file_holder.utp_log_file);
}

#define UTP_LOG utp_log
#define UTP_LOGV utp_log

#else

#define UTP_LOG if (false) printf
#define UTP_LOGV if (false) printf

#endif

enum
{
	ACK_MASK = 0xffff,

	// the number of packets that'll fit in the reorder buffer
	max_packets_reorder = 512,
	
	// this is the target buffering delay we're aiming for, in microseconds.
	// i.e. 50 ms
//	target_delay = 50000

};

// 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
TORRENT_EXPORT bool compare_less_wrap(boost::uint32_t lhs, boost::uint32_t rhs, boost::uint32_t mask)
{
	// distance walking from lhs to rhs, downwards
	boost::uint32_t dist_down = (lhs - rhs) & mask;
	// distance walking from lhs to rhs, upwards
	boost::uint32_t dist_up = (rhs - lhs) & mask;

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

// used for out-of-order incoming packets
// as well as sent packets that are waiting to be ACKed
struct packet
{
	// the last time this packet was sent
	ptime send_time;

	// the size of the buffer 'buf' pointst to
	boost::uint16_t size;

	// the number of times this packet has been sent
	boost::uint8_t num_transmissions:7;

	// true if we need to send this packet again. All
	// outstanding packets are marked as needing to be
	// resent on timeouts
	bool need_resend:1;

	// this is the offset to the payload inside the buffer
	// this is also used as a cursor to describe where the
	// next payload that hasn't been consumed yet starts
	boost::uint16_t header_size;
	
	// the actual packet buffer
	char buf[];
};

// since the uTP socket state may be needed after the
// utp_stream is closed, it's kept in a separate struct
// whose lifetime is not tied to the lifetime of utp_stream

// the utp socket is closely modelled after the asio async
// operations and handler model. For writing to the socket,
// the client provides a list of buffers (for gather/writev
// style of I/O) and whenever the socket can write another
// packet to the stream, it picks up data from these buffers.
// When all of the data has been written, or enough time has
// passed since we first started writing, the write handler
// is called and the write buffer is reset. This means that
// we're not writing anything at all while waiting for the
// client to re-issue a write request.

// reading is a little bit more complicated, since we must
// be able to receive data even when the user doesn't have
// an outstanding read operation on the socket. When the user
// does however, we want to receive data directly into the
// user's buffer instead of first copying it into our receive
// buffer. This is why the receive case is more complicated.
// There are two receive buffers. One provided by the user,
// which when present is always used. The other one is used
// when the user doesn't have an outstanding read request,
// and hence hasn't provided any buffer space to receive into.

// the user provided read buffer is called "m_read_buffer" and
// its size is "m_read_buffer_size". The buffer we spill over
// into when the user provided buffer is full or when there
// is none, is "m_receive_buffer" and "m_receive_buffer_size"
// respectively.

// in order to know when to trigger the read and write handlers
// there are two counters, m_read and m_written, which count
// the number of bytes we've stuffed into the user provided
// read buffer or written to the stream from the write buffer.
// These are used to trigger the handlers if we're written a
// large number of bytes. It's also triggered if we're filled
// the whole read buffer, or written the entire write buffer.
// The last way the handlers can be triggered is if we're read
// or written some, and enough time has elapsed since then.

// when we receive data into m_receive_buffer (i.e. the buffer
// used when there's no user provided one) is stored as a
// number of heap allocated packets. This is just because it's
// simple to reuse the data structured and it provides all the
// functionality needed for this buffer.

struct utp_socket_impl
{
	utp_socket_impl(boost::uint16_t recv_id, boost::uint16_t send_id
		, void* userdata, utp_socket_manager* sm)
		: m_sm(sm)
		, m_remote_address()
		, m_port(0)
		, m_last_history_step(time_now_hires())
		, m_send_id(send_id)
		, m_recv_id(recv_id)
		, m_ack_nr(0)
		, m_seq_nr(0)
		, m_acked_seq_nr(0)
		, m_eof_seq_nr(0)
		, m_eof(false)
		, m_fast_resend_seq_nr(0)
		, m_state(UTP_STATE_NONE)
		, m_userdata(userdata)
		, m_read(0)
		, m_write_buffer_size(0)
		, m_written(0)
		, m_receive_buffer_size(0)
		, m_read_buffer_size(0)
		, m_read_handler(0)
		, m_write_handler(0)
		, m_connect_handler(0)
		, m_reply_micro(0)
		, m_cwnd(1500 << 16)
		, m_adv_wnd(1500)
		, m_timeout(time_now_hires() + seconds(3))
		, m_last_cwnd_hit(min_time())
		, m_bytes_in_flight(0)
		, m_mtu(1500 - 20 - 8 - 8)
		, m_buffered_incoming_bytes(0)
		, m_duplicate_acks(0)
		, m_in_buf_size(100 * 1024 * 1024)
		, m_in_packets(0)
		, m_out_packets(0)
	{}

	~utp_socket_impl();

	void init(udp::endpoint const& ep, boost::uint16_t id, void* userdata
		, utp_socket_manager* sm)
	{
		m_remote_address = ep.address();
		m_port = ep.port();
		m_send_id = id + 1;
		m_recv_id = id;
		m_userdata = userdata;
		m_sm = sm;
	}

	void tick(ptime const& now);
	bool incoming_packet(char const* buf, int size, udp::endpoint const& ep);
	bool should_delete() const { return m_state == UTP_STATE_DELETE; }
	tcp::endpoint remote_endpoint(error_code& ec) const
	{
		if (m_state == UTP_STATE_NONE)
			ec = asio::error::not_connected;
		else
			TORRENT_ASSERT(m_remote_address != address_v4::any());
		return tcp::endpoint(m_remote_address, m_port);
	}
	std::size_t available() const;
	void destroy();
	void send_syn();
	void send_fin();

	bool send_pkt(bool ack);
	bool resend_packet(packet* p);
	void send_reset(utp_header* ph);
	void parse_sack(char const* ptr, int size, int* acked_bytes
		, ptime const now, boost::uint32_t& min_rtt);
	void write_payload(char* ptr, int size);
	void ack_packet(packet* p, ptime const& receive_time, boost::uint32_t& min_rtt);
	void write_sack(char* buf, int size) const;
	void incoming(char const* buf, int size, packet* p);
	void do_ledbat(int acked_bytes, boost::uint32_t delay, int in_flight, ptime const now);
	int packet_timeout() const;
	bool test_socket_state();
	void maybe_trigger_receive_callback(ptime now);
	void maybe_trigger_send_callback(ptime now);
	bool cancel_handlers(error_code const& ec, bool kill);
    bool consume_incoming_data(
        utp_header const* ph, char const* ptr, int payload_size, ptime now);

    void check_receive_buffers() const;

	utp_socket_manager* m_sm;

	sliding_average<16> m_rtt;

	address m_remote_address;
	boost::uint16_t m_port;

	// the send and receive buffers
	// maps packet sequence numbers
	packet_buffer m_inbuf;
	packet_buffer m_outbuf;

	ptime m_last_history_step;
	timestamp_history m_delay_hist;
	timestamp_history m_their_delay_hist;

	// this is the error on this socket. If m_state is
	// set to UTP_STATE_ERROR_WAIT, this error should be
	// forwarded to the client as soon as we have a new
	// async operation initiated
	error_code m_error;

	boost::uint16_t m_send_id;
	boost::uint16_t m_recv_id;

	// this is the ack we're sending back. We have
	// received all packets up to this sequence number
	boost::uint16_t m_ack_nr;

	// the sequence number of the next packet
	// we'll send
	boost::uint16_t m_seq_nr;

	// this is the sequence number of the packet that
	// everything has been ACKed up to. Everything we've
	// sent up to this point has been received by the other
	// end.
	boost::uint16_t m_acked_seq_nr;

	// this is the sequence number of the FIN packet
	// we've received. This sequence number is only
	// valid if m_eof is true. We should not accept
	// any packets beyond this sequence number from the
	// other end
	boost::uint16_t m_eof_seq_nr;

	// this is set to true when we receive a fin
	bool m_eof;

	// each packet gets one chance of "fast resend". i.e.
	// if we have multiple duplicate acks, we may send a
	// packet immediately, if m_fast_resend_seq_nr is set
	// to that packet's sequence number
	boost::uint16_t m_fast_resend_seq_nr;

	enum state_t {
		// not yet connected
		UTP_STATE_NONE,
		// sent a syn packet, not received any acks
		UTP_STATE_SYN_SENT,
		// syn-ack received and in normal operation
		// of sending and receiving data
		UTP_STATE_CONNECTED,
		// fin sent, but all packets up to the fin packet
		// have not yet been acked
		UTP_STATE_FIN_SENT,

		// the socket has been gracefully disconnected
		// and is waiting for the client to make a
		// socket call so that we can communicate this
		// fact and actually delete all the state, or
		// there is an error on this socket and we're
		// waiting to communicate this to the client in
		// a callback. The error in either case is stored
		// in m_error. If the socket has gracefully shut
		// down, the error is error::eof.
		UTP_STATE_ERROR_WAIT,

		// there are no more references to this socket
		// and we can delete it
		UTP_STATE_DELETE
	};

	// the state the socket is in
	unsigned char m_state;

	// userdata pointer passed along
	// with any callback. This is initialized to 0
	// then set to point to the utp_stream when
	// hooked up, and then reset to 0 once the utp_stream
	// detaches. This is used to know whether or not
	// the socket impl is still attached to a utp_stream
	// object. When it isn't, we'll never be able to
	// signal anything back to the client, and in case
	// of errors, we just have to delete ourselves
	// i.e. transition to the UTP_STATE_DELETED state
	void* m_userdata;

	// This is a platform-independent replacement
	// for the regular iovec type in posix. Since
	// it's not used in any system call, we might as
	// well define our own type instead of wrapping
	// the system's type.
	struct iovec_t
	{
		iovec_t(void* b, size_t l): buf(b), len(l) {}
		void* buf;
		size_t len;
	};

	// if there's currently an async read or write
	// operation in progress, these buffers are initialized
	// and used, otherwise any bytes received are stuck in
	// m_receive_buffer until another read is made
	// as we flush from the write buffer, individual iovecs
	// are updated to only refer to unflushed portions of the
	// buffers. Buffers that empty are erased from the vector.
	std::vector<iovec_t> m_write_buffer;

	// the number of bytes read into the user provided
	// buffer. If this grows too big, we'll trigger the
	// read handler.
	int m_read;

	// the sum of the lengths of all iovec in m_write_buffer
	int m_write_buffer_size;

	// the number of bytes already written to packets
	// from m_write_buffer
	int m_written;

	// the user provided read buffer. If this has a size greater
	// than 0, we'll always prefer using it over putting received
	// data in the m_receive_buffer. As data is stored in the
	// read buffer, the iovec_t elements are adjusted to only
	// refer to the unwritten portions of the buffers, and the
	// ones that fill up are erased from the vector
	std::vector<iovec_t> m_read_buffer;
	
	// packets we've received without a read operation
	// active. Store them here until the client triggers
	// an async_read_some
	std::vector<packet*> m_receive_buffer;

	// the sum of all packets stored in m_receive_buffer
	int m_receive_buffer_size;
	
	// the sum of all buffers in m_read_buffer
	int m_read_buffer_size;

	// these are the callbacks made into the utp_stream object
	// on read/write/connect events
	utp_stream::handler_t m_read_handler;
	utp_stream::handler_t m_write_handler;
	utp_stream::connect_handler_t m_connect_handler;

	// timers when we should trigger the read and
	// write callbacks (unless the buffers fill up
	// before)
	ptime m_read_timeout;
	ptime m_write_timeout;

	// the timestamp diff in the last packet received
	// this is what we'll send back
	boost::uint32_t m_reply_micro;

	// the max number of bytes in-flight. This is a fixed point
	// value, to get the true number of bytes, shift right 16 bits
	boost::uint32_t m_cwnd;

	// this is the advertized receive window the other end sent
	// we'll never have more un-acked bytes in flight
	// if this ever gets set to zero, we'll try one packet every
	// second until the window opens up again
	boost::uint32_t m_adv_wnd;

	// the time when the last packet we sent times out. Including re-sends.
	// if we ever end up not having sent anything in one second (
	// or one mean rtt + 2 average deviations, whichever is greater)
	// we set our cwnd to 1 MSS. This condition can happen either because
	// a packet has timed out and needs to be resent or because our
	// cwnd is set to less than one MSS during congestion control.
	// it can also happen if the other end sends an advertized window
	// size less than one MSS.
	ptime m_timeout;
	
	// the last time we wanted to send more data, but couldn't because
	// it would bring the number of outstanding bytes above the cwnd.
	// this is used to restrict increasing the cwnd size when we're
	// not sending fast enough to need it bigger
	ptime m_last_cwnd_hit;

	// the number of un-acked bytes we have sent
	int m_bytes_in_flight;

	// the max number of bytes we can send in a packet
	// including the header
	boost::uint32_t m_mtu;

	// the number of bytes we have buffered in m_inbuf
	int m_buffered_incoming_bytes;

	// this is a counter of how many times the current m_acked_seq_nr
	// has been ACKed. If it's ACKed more than 3 times, we assume the
	// packet with the next sequence number has been lost, and we trigger
	// a re-send. Ovbiously an ACK only counts as a duplicate as long as
	// we have outstanding packets following it.
	boost::uint8_t m_duplicate_acks;
	// #error implement nagle

	int m_in_buf_size;

	// counters
	int m_in_packets;
	int m_out_packets;
};

utp_socket_impl* construct_utp_impl(boost::uint16_t recv_id
	, boost::uint16_t send_id, void* userdata
	, utp_socket_manager* sm)
{
	return new utp_socket_impl(recv_id, send_id, userdata, sm);
}

void delete_utp_impl(utp_socket_impl* s)
{
	delete s;
}

bool should_delete(utp_socket_impl* s)
{
	return s->should_delete();
}

void tick_utp_impl(utp_socket_impl* s, ptime const& now)
{
	s->tick(now);
}

bool utp_incoming_packet(utp_socket_impl* s, char const* p, int size, udp::endpoint const& ep)
{
	return s->incoming_packet(p, size, ep);
}

udp::endpoint utp_remote_endpoint(utp_socket_impl* s)
{
	return udp::endpoint(s->m_remote_address, s->m_port);
}

utp_stream::utp_stream(asio::io_service& io_service)
	: m_io_service(io_service)
	, m_impl(0)
	, m_open(false)
{
}

utp_socket_impl* utp_stream::get_impl()
{
	return m_impl;
}

void utp_stream::close()
{
	if (!m_impl) return;
	m_impl->destroy();
}

std::size_t utp_stream::available() const
{
	return m_impl->available();
}

utp_stream::endpoint_type utp_stream::remote_endpoint(error_code& ec) const
{
	if (!m_impl)
    {
        ec = asio::error::not_connected;
        return endpoint_type();
    }
	return m_impl->remote_endpoint(ec);
}

utp_stream::endpoint_type utp_stream::local_endpoint() const
{
	return m_impl->m_sm->local_endpoint();
}

utp_stream::~utp_stream()
{
	if (m_impl) m_impl->destroy();
	m_impl = 0;
}

void utp_stream::set_impl(utp_socket_impl* impl)
{
	TORRENT_ASSERT(m_impl == 0);
	TORRENT_ASSERT(!m_open);
	m_impl = impl;
	m_open = true;
}

int utp_stream::read_buffer_size() const
{
	TORRENT_ASSERT(m_impl);
	return m_impl->m_receive_buffer_size;
}

void utp_stream::on_read(void* self, size_t bytes_transferred, error_code const& ec, bool kill)
{
	utp_stream* s = (utp_stream*)self;

	UTP_LOGV("[%08u] %08p: calling read handler read:%d ec:%s kill:%d\n"
		, int(total_microseconds(time_now_hires() - min_time())), s->m_impl
		, bytes_transferred, ec.message().c_str(), kill);

	TORRENT_ASSERT(s->m_read_handler);
	s->m_io_service.post(boost::bind<void>(s->m_read_handler, ec, bytes_transferred));
	s->m_read_handler.clear();
	if (kill) s->m_impl = 0;
}

void utp_stream::on_write(void* self, size_t bytes_transferred, error_code const& ec, bool kill)
{
	utp_stream* s = (utp_stream*)self;

	UTP_LOGV("[%08u] %08p: calling write handler written:%d ec:%s kill:%d\n"
		, int(total_microseconds(time_now_hires() - min_time())), s->m_impl
		, bytes_transferred, ec.message().c_str(), kill);

	TORRENT_ASSERT(s->m_write_handler);
	s->m_io_service.post(boost::bind<void>(s->m_write_handler, ec, bytes_transferred));
	s->m_write_handler.clear();
	if (kill) s->m_impl = 0;
}

void utp_stream::on_connect(void* self, error_code const& ec, bool kill)
{
	utp_stream* s = (utp_stream*)self;

	UTP_LOGV("[%08u] %08p: calling connect handler ec:%s kill:%d\n"
		, int(total_microseconds(time_now_hires() - min_time())), s->m_impl, ec.message().c_str(), kill);

	TORRENT_ASSERT(s->m_connect_handler);
	s->m_io_service.post(boost::bind<void>(s->m_connect_handler, ec));
	s->m_connect_handler.clear();
	if (kill) s->m_impl = 0;
}

void utp_stream::add_read_buffer(void* buf, size_t len)
{
	TORRENT_ASSERT(m_impl);
	m_impl->m_read_buffer.push_back(utp_socket_impl::iovec_t(buf, len));
	m_impl->m_read_buffer_size += len;

    UTP_LOGV("[%08u] %08p: add_read_buffer %d bytes\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl, len);
}

// this is the wrapper to add a user provided write buffer to the
// utp_socket_impl. It makes sure the m_write_buffer_size is kept
// up to date
void utp_stream::add_write_buffer(void const* buf, size_t len)
{
	TORRENT_ASSERT(m_impl);

#ifndef NDEBUG
	int write_buffer_size = 0;
	for (std::vector<utp_socket_impl::iovec_t>::iterator i = m_impl->m_write_buffer.begin()
		, end(m_impl->m_write_buffer.end()); i != end; ++i)
	{
		write_buffer_size += i->len;
	}
	TORRENT_ASSERT(m_impl->m_write_buffer_size == write_buffer_size);
#endif

	m_impl->m_write_buffer.push_back(utp_socket_impl::iovec_t((void*)buf, len));
	m_impl->m_write_buffer_size += len;

#ifndef NDEBUG
	write_buffer_size = 0;
	for (std::vector<utp_socket_impl::iovec_t>::iterator i = m_impl->m_write_buffer.begin()
		, end(m_impl->m_write_buffer.end()); i != end; ++i)
	{
		write_buffer_size += i->len;
	}
	TORRENT_ASSERT(m_impl->m_write_buffer_size == write_buffer_size);
#endif

    UTP_LOGV("[%08u] %08p: add_write_buffer %d bytes\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl, len);
}

// this is called when all user provided read buffers have been added
// and it's time to execute the async operation. The first thing we
// do is to copy any data stored in m_receive_buffer into the user
// provided buffer. This might be enough to in turn trigger the read
// handler immediately.
void utp_stream::set_read_handler(handler_t h)
{
	m_impl->m_read_handler = h;
	if (m_impl->test_socket_state()) return;

    UTP_LOGV("[%08u] %08p: new read handler. %d bytes in buffer\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl
        , m_impl->m_receive_buffer_size);

	// so, the client wants to read. If we already
	// have some data in the read buffer, move it into the
	// client's buffer right away
	if (m_impl->m_receive_buffer_size == 0) return;

	std::vector<utp_socket_impl::iovec_t>::iterator target = m_impl->m_read_buffer.begin();

	int pop_packets = 0;
	for (std::vector<packet*>::iterator i = m_impl->m_receive_buffer.begin()
		, end(m_impl->m_receive_buffer.end()); i != end;)
	{
		if (target == m_impl->m_read_buffer.end()) 
        {
            UTP_LOGV("  No more target buffers: %d bytes left in buffer\n"
                , m_impl->m_receive_buffer_size);
            TORRENT_ASSERT(m_impl->m_read_buffer.empty());
            break;
        }

        m_impl->check_receive_buffers();

		packet* p = *i;
		int to_copy = (std::min)(p->size - p->header_size, int(target->len));
        TORRENT_ASSERT(to_copy >= 0);
		memcpy(target->buf, p->buf + p->header_size, to_copy);
		m_impl->m_read += to_copy;
		target->buf = ((char*)target->buf) + to_copy;
		target->len -= to_copy;
		m_impl->m_receive_buffer_size -= to_copy;
		TORRENT_ASSERT(m_impl->m_read_buffer_size >= to_copy);
		m_impl->m_read_buffer_size -= to_copy;
		p->header_size += to_copy;
		if (target->len == 0) target = m_impl->m_read_buffer.erase(target);

        m_impl->check_receive_buffers();

        TORRENT_ASSERT(m_impl->m_receive_buffer_size >= 0);

        // Consumed entire packet
        if (p->header_size == p->size)
        {
            free(p);
            ++pop_packets;
            *i = 0;
            ++i;
        }

        if (m_impl->m_receive_buffer_size == 0)
        {
            UTP_LOGV("  Didn't fill entire target: %d bytes left in buffer\n"
              , m_impl->m_receive_buffer_size);
            break;
        }
	}
	// remove the packets from the receive_buffer that we already copied over
	// and freed
	m_impl->m_receive_buffer.erase(m_impl->m_receive_buffer.begin()
		, m_impl->m_receive_buffer.begin() + pop_packets);
	// we exited either because we ran out of bytes to copy
	// or because we ran out of space to copy the bytes to
	TORRENT_ASSERT(m_impl->m_receive_buffer_size == 0
		|| m_impl->m_read_buffer.empty());

    UTP_LOGV("[%08u] %08p: %d packets moved from buffer to user space\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl
        , pop_packets);

	m_impl->maybe_trigger_receive_callback(time_now_hires());
}

// this is called when all user provided write buffers have been
// added. Start trying to send packets with the payload immediately.
void utp_stream::set_write_handler(handler_t h)
{
    UTP_LOGV("[%08u] %08p: new write handler. %d bytes to write\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl
        , m_impl->m_write_buffer_size);

	m_impl->m_write_handler = h;
	m_impl->m_written = 0;
	if (m_impl->test_socket_state()) return;
	// try to write. send_pkt returns false if there's
	// no more payload to send or if the congestion window
	// is full and we can't send more packets right now
	while (m_impl->send_pkt(false));
}

void utp_stream::do_connect(tcp::endpoint const& ep, utp_stream::connect_handler_t handler)
{
	TORRENT_ASSERT(m_impl->m_connect_handler == 0);
	m_impl->m_remote_address = ep.address();
	m_impl->m_port = ep.port();
	m_impl->m_connect_handler = handler;

	if (m_impl->test_socket_state()) return;
	m_impl->send_syn();
}

// =========== utp_socket_impl ============

utp_socket_impl::~utp_socket_impl()
{
    UTP_LOGV("[%08u] %08p: destroying utp socket state\n"
		, int(total_microseconds(time_now_hires() - min_time())), this);

	// free any buffers we're holding
	for (boost::uint16_t i = m_inbuf.cursor(), end((m_inbuf.cursor()
		+ m_inbuf.capacity()) & ACK_MASK);
		i != end; i = (i + 1) & ACK_MASK)
	{
		void* p = m_inbuf.remove(i);
		free(p);
	}
	for (boost::uint16_t i = m_outbuf.cursor(), end((m_outbuf.cursor()
		+ m_outbuf.capacity()) & ACK_MASK);
		i != end; i = (i + 1) & ACK_MASK)
	{
		void* p = m_outbuf.remove(i);
		free(p);
	}
}

void utp_socket_impl::maybe_trigger_receive_callback(ptime now)
{
	// nothing has been read or there's no outstanding read operation
	if (m_read == 0 || m_read_handler == 0) return;

	if (m_read > 10000 || m_read_buffer_size == 0 || now >= m_read_timeout)
	{
		UTP_LOGV("[%08u] %08p: calling read handler read:%d\n"
			, int(total_microseconds(time_now_hires() - min_time())), this
			, m_read);
		m_read_handler(m_userdata, m_read, m_error, false);
		m_read_handler = 0;
		m_read = 0;
		m_read_buffer_size = 0;
		m_read_buffer.clear();
	}
}

void utp_socket_impl::maybe_trigger_send_callback(ptime now)
{
	// nothing has been written or there's no outstanding write operation
	if (m_written == 0 || m_write_handler == 0) return;

	if (m_written > 10000 || m_write_buffer_size == 0 || now >= m_write_timeout)
	{
		UTP_LOGV("[%08u] %08p: calling write handler written:%d\n"
			, int(total_microseconds(time_now_hires() - min_time())), this
			, m_written);

		m_write_handler(m_userdata, m_written, m_error, false);
		m_write_handler = 0;
		m_written = 0;
		m_write_buffer_size = 0;
		m_write_buffer.clear();
	}
}

void utp_socket_impl::destroy()
{
	UTP_LOGV("[%08u] %08p: destroy state:%s\n"
		, int(total_microseconds(time_now_hires() - min_time())), this
		, socket_state_names[m_state]);

	if (m_userdata == 0) return;

	m_error = asio::error::operation_aborted;
	cancel_handlers(m_error, true);

	m_userdata = 0;
    m_read_buffer.clear();
    m_read_buffer_size = 0;

	if (m_state == UTP_STATE_ERROR_WAIT
		|| m_state == UTP_STATE_NONE
		|| m_state == UTP_STATE_SYN_SENT)
	{
		m_state = UTP_STATE_DELETE;
		UTP_LOGV("[%08u] %08p: state:%s\n"
			, int(total_microseconds(time_now_hires() - min_time())), this
			, socket_state_names[m_state]);
		return;
	}

	// you should never close a socket with an outstanding write!
	TORRENT_ASSERT(!m_write_handler);

	send_fin();

	// #error our end is closing. Wait for everything to be acked
}

void utp_socket_impl::send_syn()
{
	m_seq_nr = rand();
	m_acked_seq_nr = (m_seq_nr - 1) & ACK_MASK;
	m_ack_nr = 0;

	packet* p = (packet*)malloc(sizeof(packet) + sizeof(utp_header));
	p->size = sizeof(utp_header);
	p->header_size = sizeof(utp_header);
	p->num_transmissions = 1;
	p->need_resend = false;
	utp_header* h = (utp_header*)p->buf;
	h->ver = 1;
	h->type = ST_SYN;
	h->extension = 0;
	// using recv_id here is intentional! This is an odd
	// thing in uTP. The syn packet is sent with the connection
	// ID that it expects to receive the syn ack on. All
	// subsequent connection IDs will be this plus one.
	h->connection_id = m_recv_id;
	h->timestamp_difference_microseconds = m_reply_micro;
	h->wnd_size = 0;
	h->seq_nr = m_seq_nr;
	h->ack_nr = 0;

	ptime now = time_now_hires();
	p->send_time = now;
	h->timestamp_microseconds = total_microseconds(now - min_time());

	UTP_LOGV("[%08u] %08p: send_syn seq_nr:%d id:%d target:%s\n"
		, int(total_microseconds(now - min_time()))
		, this, int(m_seq_nr), int(m_recv_id)
		, print_endpoint(udp::endpoint(m_remote_address, m_port)).c_str());

	TORRENT_ASSERT(!m_error);
	m_sm->send_packet(udp::endpoint(m_remote_address, m_port), (char const*)h
		, sizeof(utp_header), m_error);
	if (m_error)
	{
		free(p);
		m_state = UTP_STATE_ERROR_WAIT;
		test_socket_state();
		return;
	}

	TORRENT_ASSERT(!m_outbuf.at(m_seq_nr));
	m_outbuf.insert(m_seq_nr, p);

	m_seq_nr = (m_seq_nr + 1) & ACK_MASK;

	m_state = UTP_STATE_SYN_SENT;
	UTP_LOGV("[%08u] %08p: state:%s\n"
		, int(total_microseconds(time_now_hires() - min_time())), this
		, socket_state_names[m_state]);
}

void utp_socket_impl::send_fin()
{
	bool ret = false;

	// we need a heap allocated packet in order to stick it
	// in the send buffer, so that we can resend it
	packet* p = (packet*)malloc(sizeof(packet) + sizeof(utp_header));

	p->size = sizeof(utp_header);
	p->header_size = sizeof(utp_header);
	p->num_transmissions = 1;
	p->need_resend = false;
	utp_header* h = (utp_header*)p->buf;

	h->ver = 1;
	h->type = ST_FIN;
	h->extension = 0;
	h->connection_id = m_send_id;
	h->timestamp_difference_microseconds = m_reply_micro;
	h->wnd_size = m_in_buf_size - m_buffered_incoming_bytes - m_receive_buffer_size;
	h->seq_nr = m_seq_nr;
	h->ack_nr = m_ack_nr;

	ptime now = time_now_hires();
	p->send_time = now;
	h->timestamp_microseconds = total_microseconds(now - min_time());

	m_sm->send_packet(udp::endpoint(m_remote_address, m_port)
		, (char const*)h, sizeof(utp_header), m_error);

	UTP_LOGV("[%08u] %08p: sending FIN seq_nr:%d ack_nr:%d type:%s "
		"id:%d target:%s size:%d error:%s send_buffer_size:%d\n"
		, int(total_microseconds(now - min_time()))
		, this, int(h->seq_nr), int(h->ack_nr), packet_type_names[h->type]
		, m_send_id, print_endpoint(udp::endpoint(m_remote_address, m_port)).c_str()
		, sizeof(utp_header), m_error.message().c_str(), m_write_buffer_size);

	if (m_error)
	{
		m_state = UTP_STATE_ERROR_WAIT;
		test_socket_state();
	}

#if !TORRENT_UT_SEQ
	// if the other end closed the connection immediately
	// our FIN packet will end up having the same sequence
	// number as the SYN, so this assert is invalid
	TORRENT_ASSERT(!m_outbuf.at(m_seq_nr));
#endif

	packet* old = (packet*)m_outbuf.insert(m_seq_nr, p);
	if (old)
	{
		if (!old->need_resend) m_bytes_in_flight -= old->size - old->header_size;
		free(old);
	}
	m_seq_nr = (m_seq_nr + 1) & ACK_MASK;

	m_state = UTP_STATE_FIN_SENT;

	UTP_LOGV("[%08u] %08p: state:%s\n"
		, int(total_microseconds(time_now_hires() - min_time())), this
		, socket_state_names[m_state]);
}

void utp_socket_impl::send_reset(utp_header* ph)
{
	utp_header h;
	h.ver = 1;
	h.type = ST_RESET;
	h.extension = 0;
	h.connection_id = m_send_id;
	h.timestamp_difference_microseconds = m_reply_micro;
	h.wnd_size = 0;
	h.seq_nr = rand();
	h.ack_nr = ph->seq_nr;
	ptime now = time_now_hires();
	h.timestamp_microseconds = total_microseconds(now - min_time());

	UTP_LOGV("[%08u] %08p: send_reset seq_nr:%d id:%d ack_nr:%d\n"
		, int(total_microseconds(now - min_time()))
		, this, int(h.seq_nr), int(m_send_id), int(ph->seq_nr));

	// ignore errors here
	error_code ec;
	m_sm->send_packet(udp::endpoint(m_remote_address, m_port), (char const*)&h, sizeof(h), ec);
}

std::size_t utp_socket_impl::available() const
{
	return m_receive_buffer_size;
}

void utp_socket_impl::parse_sack(char const* ptr, int size, int* acked_bytes
	, ptime const now, boost::uint32_t& min_rtt)
{
	if (size == 0) return;

	// this is the sequence number the current bit represents
	int ack_nr = (m_acked_seq_nr + 2) & ACK_MASK;

	UTP_LOGV("[%08u] %08p: got SACK ack_nr:%d "
		, int(total_microseconds(now - min_time())), this
		, ack_nr);

	// for each byte
	for (char const* end = ptr + size; ptr != end; ++ptr)
	{
		unsigned char bitfield = unsigned(*ptr);
		unsigned char mask = 1;
		// for each bit
		for (int i = 0; i < 8; ++i)
		{
			UTP_LOGV("%d", mask & bitfield ? 1 : 0);
			if (mask & bitfield)
			{
				// this bit was set, ack_nr was received
				packet* p = (packet*)m_outbuf.remove(ack_nr);
				if (!p) continue;
				acked_bytes += p->size - p->header_size;
				UTP_LOGV("[%08u] %08p: acked packet %d (%d bytes)\n"
					, int(total_microseconds(now - min_time())), this
					, ack_nr, p->size - p->header_size);
				ack_packet(p, now, min_rtt);
				// each ACKed packet counts as a duplicate ack
				++m_duplicate_acks;
			}

			mask <<= 1;
			ack_nr = (ack_nr + 1) & ACK_MASK;

			// we haven't sent packets past this point.
			// if there are any more bits set, we have to
			// ignore them anyway
			if (ack_nr == m_seq_nr)
			{
				UTP_LOGV("\n");
				return;
			}
		}
	}
	UTP_LOGV("\n");
}

// copies data from the write buffer into the packet
// pointed to by ptr
void utp_socket_impl::write_payload(char* ptr, int size)
{
#ifndef NDEBUG
	int write_buffer_size = 0;
	for (std::vector<iovec_t>::iterator i = m_write_buffer.begin()
		, end(m_write_buffer.end()); i != end; ++i)
	{<commit_msg>fixed bug in SACK parsing<commit_after>/*

Copyright (c) 2009, <NAME>
All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions
are met:

    * Redistributions of source code must retain the above copyright
      notice, this list of conditions and the following disclaimer.
    * Redistributions in binary form must reproduce the above copyright
      notice, this list of conditions and the following disclaimer in
      the documentation and/or other materials provided with the distribution.
    * Neither the name of the author nor the names of its
      contributors may be used to endorse or promote products derived
      from this software without specific prior written permission.

THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
POSSIBILITY OF SUCH DAMAGE.

*/

#include "libtorrent/utp_stream.hpp"
#include "libtorrent/sliding_average.hpp"
#include "libtorrent/utp_socket_manager.hpp"
#include "libtorrent/alloca.hpp"
#include "libtorrent/timestamp_history.hpp"
#include <boost/cstdint.hpp>

#define TORRENT_UTP_LOG 1
#define TORRENT_UT_SEQ 1

#if TORRENT_UTP_LOG
#include <stdarg.h>
#include "libtorrent/socket_io.hpp"
#endif

namespace libtorrent {

#if TORRENT_UTP_LOG

char const* packet_type_names[] = { "ST_DATA", "ST_FIN", "ST_STATE", "ST_RESET", "ST_SYN" };
char const* socket_state_names[] = { "NONE", "SYN_SENT", "CONNECTED", "FIN_SENT", "ERROR", "DELETE" };

static struct utp_logger
{
	FILE* utp_log_file;

	utp_logger() : utp_log_file(0)
	{
		utp_log_file = fopen("utp.log", "w+");
	}
	~utp_logger()
	{
		if (utp_log_file) fclose(utp_log_file);
	}
} log_file_holder;

void utp_log(char const* fmt, ...)
{
	va_list l;
	va_start(l, fmt);
	vfprintf(log_file_holder.utp_log_file, fmt, l);
	va_end(l);
	fflush(log_file_holder.utp_log_file);
}

#define UTP_LOG utp_log
#define UTP_LOGV utp_log

#else

#define UTP_LOG if (false) printf
#define UTP_LOGV if (false) printf

#endif

enum
{
	ACK_MASK = 0xffff,

	// the number of packets that'll fit in the reorder buffer
	max_packets_reorder = 512,
	
	// this is the target buffering delay we're aiming for, in microseconds.
	// i.e. 50 ms
//	target_delay = 50000

};

// 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
TORRENT_EXPORT bool compare_less_wrap(boost::uint32_t lhs, boost::uint32_t rhs, boost::uint32_t mask)
{
	// distance walking from lhs to rhs, downwards
	boost::uint32_t dist_down = (lhs - rhs) & mask;
	// distance walking from lhs to rhs, upwards
	boost::uint32_t dist_up = (rhs - lhs) & mask;

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

// used for out-of-order incoming packets
// as well as sent packets that are waiting to be ACKed
struct packet
{
	// the last time this packet was sent
	ptime send_time;

	// the size of the buffer 'buf' pointst to
	boost::uint16_t size;

	// the number of times this packet has been sent
	boost::uint8_t num_transmissions:7;

	// true if we need to send this packet again. All
	// outstanding packets are marked as needing to be
	// resent on timeouts
	bool need_resend:1;

	// this is the offset to the payload inside the buffer
	// this is also used as a cursor to describe where the
	// next payload that hasn't been consumed yet starts
	boost::uint16_t header_size;
	
	// the actual packet buffer
	char buf[];
};

// since the uTP socket state may be needed after the
// utp_stream is closed, it's kept in a separate struct
// whose lifetime is not tied to the lifetime of utp_stream

// the utp socket is closely modelled after the asio async
// operations and handler model. For writing to the socket,
// the client provides a list of buffers (for gather/writev
// style of I/O) and whenever the socket can write another
// packet to the stream, it picks up data from these buffers.
// When all of the data has been written, or enough time has
// passed since we first started writing, the write handler
// is called and the write buffer is reset. This means that
// we're not writing anything at all while waiting for the
// client to re-issue a write request.

// reading is a little bit more complicated, since we must
// be able to receive data even when the user doesn't have
// an outstanding read operation on the socket. When the user
// does however, we want to receive data directly into the
// user's buffer instead of first copying it into our receive
// buffer. This is why the receive case is more complicated.
// There are two receive buffers. One provided by the user,
// which when present is always used. The other one is used
// when the user doesn't have an outstanding read request,
// and hence hasn't provided any buffer space to receive into.

// the user provided read buffer is called "m_read_buffer" and
// its size is "m_read_buffer_size". The buffer we spill over
// into when the user provided buffer is full or when there
// is none, is "m_receive_buffer" and "m_receive_buffer_size"
// respectively.

// in order to know when to trigger the read and write handlers
// there are two counters, m_read and m_written, which count
// the number of bytes we've stuffed into the user provided
// read buffer or written to the stream from the write buffer.
// These are used to trigger the handlers if we're written a
// large number of bytes. It's also triggered if we're filled
// the whole read buffer, or written the entire write buffer.
// The last way the handlers can be triggered is if we're read
// or written some, and enough time has elapsed since then.

// when we receive data into m_receive_buffer (i.e. the buffer
// used when there's no user provided one) is stored as a
// number of heap allocated packets. This is just because it's
// simple to reuse the data structured and it provides all the
// functionality needed for this buffer.

struct utp_socket_impl
{
	utp_socket_impl(boost::uint16_t recv_id, boost::uint16_t send_id
		, void* userdata, utp_socket_manager* sm)
		: m_sm(sm)
		, m_remote_address()
		, m_port(0)
		, m_last_history_step(time_now_hires())
		, m_send_id(send_id)
		, m_recv_id(recv_id)
		, m_ack_nr(0)
		, m_seq_nr(0)
		, m_acked_seq_nr(0)
		, m_eof_seq_nr(0)
		, m_eof(false)
		, m_fast_resend_seq_nr(0)
		, m_state(UTP_STATE_NONE)
		, m_userdata(userdata)
		, m_read(0)
		, m_write_buffer_size(0)
		, m_written(0)
		, m_receive_buffer_size(0)
		, m_read_buffer_size(0)
		, m_read_handler(0)
		, m_write_handler(0)
		, m_connect_handler(0)
		, m_reply_micro(0)
		, m_cwnd(1500 << 16)
		, m_adv_wnd(1500)
		, m_timeout(time_now_hires() + seconds(3))
		, m_last_cwnd_hit(min_time())
		, m_bytes_in_flight(0)
		, m_mtu(1500 - 20 - 8 - 8)
		, m_buffered_incoming_bytes(0)
		, m_duplicate_acks(0)
		, m_in_buf_size(100 * 1024 * 1024)
		, m_in_packets(0)
		, m_out_packets(0)
	{}

	~utp_socket_impl();

	void init(udp::endpoint const& ep, boost::uint16_t id, void* userdata
		, utp_socket_manager* sm)
	{
		m_remote_address = ep.address();
		m_port = ep.port();
		m_send_id = id + 1;
		m_recv_id = id;
		m_userdata = userdata;
		m_sm = sm;
	}

	void tick(ptime const& now);
	bool incoming_packet(char const* buf, int size, udp::endpoint const& ep);
	bool should_delete() const { return m_state == UTP_STATE_DELETE; }
	tcp::endpoint remote_endpoint(error_code& ec) const
	{
		if (m_state == UTP_STATE_NONE)
			ec = asio::error::not_connected;
		else
			TORRENT_ASSERT(m_remote_address != address_v4::any());
		return tcp::endpoint(m_remote_address, m_port);
	}
	std::size_t available() const;
	void destroy();
	void send_syn();
	void send_fin();

	bool send_pkt(bool ack);
	bool resend_packet(packet* p);
	void send_reset(utp_header* ph);
	void parse_sack(char const* ptr, int size, int* acked_bytes
		, ptime const now, boost::uint32_t& min_rtt);
	void write_payload(char* ptr, int size);
	void ack_packet(packet* p, ptime const& receive_time, boost::uint32_t& min_rtt);
	void write_sack(char* buf, int size) const;
	void incoming(char const* buf, int size, packet* p);
	void do_ledbat(int acked_bytes, boost::uint32_t delay, int in_flight, ptime const now);
	int packet_timeout() const;
	bool test_socket_state();
	void maybe_trigger_receive_callback(ptime now);
	void maybe_trigger_send_callback(ptime now);
	bool cancel_handlers(error_code const& ec, bool kill);
    bool consume_incoming_data(
        utp_header const* ph, char const* ptr, int payload_size, ptime now);

    void check_receive_buffers() const;

	utp_socket_manager* m_sm;

	sliding_average<16> m_rtt;

	address m_remote_address;
	boost::uint16_t m_port;

	// the send and receive buffers
	// maps packet sequence numbers
	packet_buffer m_inbuf;
	packet_buffer m_outbuf;

	ptime m_last_history_step;
	timestamp_history m_delay_hist;
	timestamp_history m_their_delay_hist;

	// this is the error on this socket. If m_state is
	// set to UTP_STATE_ERROR_WAIT, this error should be
	// forwarded to the client as soon as we have a new
	// async operation initiated
	error_code m_error;

	boost::uint16_t m_send_id;
	boost::uint16_t m_recv_id;

	// this is the ack we're sending back. We have
	// received all packets up to this sequence number
	boost::uint16_t m_ack_nr;

	// the sequence number of the next packet
	// we'll send
	boost::uint16_t m_seq_nr;

	// this is the sequence number of the packet that
	// everything has been ACKed up to. Everything we've
	// sent up to this point has been received by the other
	// end.
	boost::uint16_t m_acked_seq_nr;

	// this is the sequence number of the FIN packet
	// we've received. This sequence number is only
	// valid if m_eof is true. We should not accept
	// any packets beyond this sequence number from the
	// other end
	boost::uint16_t m_eof_seq_nr;

	// this is set to true when we receive a fin
	bool m_eof;

	// each packet gets one chance of "fast resend". i.e.
	// if we have multiple duplicate acks, we may send a
	// packet immediately, if m_fast_resend_seq_nr is set
	// to that packet's sequence number
	boost::uint16_t m_fast_resend_seq_nr;

	enum state_t {
		// not yet connected
		UTP_STATE_NONE,
		// sent a syn packet, not received any acks
		UTP_STATE_SYN_SENT,
		// syn-ack received and in normal operation
		// of sending and receiving data
		UTP_STATE_CONNECTED,
		// fin sent, but all packets up to the fin packet
		// have not yet been acked
		UTP_STATE_FIN_SENT,

		// the socket has been gracefully disconnected
		// and is waiting for the client to make a
		// socket call so that we can communicate this
		// fact and actually delete all the state, or
		// there is an error on this socket and we're
		// waiting to communicate this to the client in
		// a callback. The error in either case is stored
		// in m_error. If the socket has gracefully shut
		// down, the error is error::eof.
		UTP_STATE_ERROR_WAIT,

		// there are no more references to this socket
		// and we can delete it
		UTP_STATE_DELETE
	};

	// the state the socket is in
	unsigned char m_state;

	// userdata pointer passed along
	// with any callback. This is initialized to 0
	// then set to point to the utp_stream when
	// hooked up, and then reset to 0 once the utp_stream
	// detaches. This is used to know whether or not
	// the socket impl is still attached to a utp_stream
	// object. When it isn't, we'll never be able to
	// signal anything back to the client, and in case
	// of errors, we just have to delete ourselves
	// i.e. transition to the UTP_STATE_DELETED state
	void* m_userdata;

	// This is a platform-independent replacement
	// for the regular iovec type in posix. Since
	// it's not used in any system call, we might as
	// well define our own type instead of wrapping
	// the system's type.
	struct iovec_t
	{
		iovec_t(void* b, size_t l): buf(b), len(l) {}
		void* buf;
		size_t len;
	};

	// if there's currently an async read or write
	// operation in progress, these buffers are initialized
	// and used, otherwise any bytes received are stuck in
	// m_receive_buffer until another read is made
	// as we flush from the write buffer, individual iovecs
	// are updated to only refer to unflushed portions of the
	// buffers. Buffers that empty are erased from the vector.
	std::vector<iovec_t> m_write_buffer;

	// the number of bytes read into the user provided
	// buffer. If this grows too big, we'll trigger the
	// read handler.
	int m_read;

	// the sum of the lengths of all iovec in m_write_buffer
	int m_write_buffer_size;

	// the number of bytes already written to packets
	// from m_write_buffer
	int m_written;

	// the user provided read buffer. If this has a size greater
	// than 0, we'll always prefer using it over putting received
	// data in the m_receive_buffer. As data is stored in the
	// read buffer, the iovec_t elements are adjusted to only
	// refer to the unwritten portions of the buffers, and the
	// ones that fill up are erased from the vector
	std::vector<iovec_t> m_read_buffer;
	
	// packets we've received without a read operation
	// active. Store them here until the client triggers
	// an async_read_some
	std::vector<packet*> m_receive_buffer;

	// the sum of all packets stored in m_receive_buffer
	int m_receive_buffer_size;
	
	// the sum of all buffers in m_read_buffer
	int m_read_buffer_size;

	// these are the callbacks made into the utp_stream object
	// on read/write/connect events
	utp_stream::handler_t m_read_handler;
	utp_stream::handler_t m_write_handler;
	utp_stream::connect_handler_t m_connect_handler;

	// timers when we should trigger the read and
	// write callbacks (unless the buffers fill up
	// before)
	ptime m_read_timeout;
	ptime m_write_timeout;

	// the timestamp diff in the last packet received
	// this is what we'll send back
	boost::uint32_t m_reply_micro;

	// the max number of bytes in-flight. This is a fixed point
	// value, to get the true number of bytes, shift right 16 bits
	boost::uint32_t m_cwnd;

	// this is the advertized receive window the other end sent
	// we'll never have more un-acked bytes in flight
	// if this ever gets set to zero, we'll try one packet every
	// second until the window opens up again
	boost::uint32_t m_adv_wnd;

	// the time when the last packet we sent times out. Including re-sends.
	// if we ever end up not having sent anything in one second (
	// or one mean rtt + 2 average deviations, whichever is greater)
	// we set our cwnd to 1 MSS. This condition can happen either because
	// a packet has timed out and needs to be resent or because our
	// cwnd is set to less than one MSS during congestion control.
	// it can also happen if the other end sends an advertized window
	// size less than one MSS.
	ptime m_timeout;
	
	// the last time we wanted to send more data, but couldn't because
	// it would bring the number of outstanding bytes above the cwnd.
	// this is used to restrict increasing the cwnd size when we're
	// not sending fast enough to need it bigger
	ptime m_last_cwnd_hit;

	// the number of un-acked bytes we have sent
	int m_bytes_in_flight;

	// the max number of bytes we can send in a packet
	// including the header
	boost::uint32_t m_mtu;

	// the number of bytes we have buffered in m_inbuf
	int m_buffered_incoming_bytes;

	// this is a counter of how many times the current m_acked_seq_nr
	// has been ACKed. If it's ACKed more than 3 times, we assume the
	// packet with the next sequence number has been lost, and we trigger
	// a re-send. Ovbiously an ACK only counts as a duplicate as long as
	// we have outstanding packets following it.
	boost::uint8_t m_duplicate_acks;
	// #error implement nagle

	int m_in_buf_size;

	// counters
	int m_in_packets;
	int m_out_packets;
};

utp_socket_impl* construct_utp_impl(boost::uint16_t recv_id
	, boost::uint16_t send_id, void* userdata
	, utp_socket_manager* sm)
{
	return new utp_socket_impl(recv_id, send_id, userdata, sm);
}

void delete_utp_impl(utp_socket_impl* s)
{
	delete s;
}

bool should_delete(utp_socket_impl* s)
{
	return s->should_delete();
}

void tick_utp_impl(utp_socket_impl* s, ptime const& now)
{
	s->tick(now);
}

bool utp_incoming_packet(utp_socket_impl* s, char const* p, int size, udp::endpoint const& ep)
{
	return s->incoming_packet(p, size, ep);
}

udp::endpoint utp_remote_endpoint(utp_socket_impl* s)
{
	return udp::endpoint(s->m_remote_address, s->m_port);
}

utp_stream::utp_stream(asio::io_service& io_service)
	: m_io_service(io_service)
	, m_impl(0)
	, m_open(false)
{
}

utp_socket_impl* utp_stream::get_impl()
{
	return m_impl;
}

void utp_stream::close()
{
	if (!m_impl) return;
	m_impl->destroy();
}

std::size_t utp_stream::available() const
{
	return m_impl->available();
}

utp_stream::endpoint_type utp_stream::remote_endpoint(error_code& ec) const
{
	if (!m_impl)
    {
        ec = asio::error::not_connected;
        return endpoint_type();
    }
	return m_impl->remote_endpoint(ec);
}

utp_stream::endpoint_type utp_stream::local_endpoint() const
{
	return m_impl->m_sm->local_endpoint();
}

utp_stream::~utp_stream()
{
	if (m_impl) m_impl->destroy();
	m_impl = 0;
}

void utp_stream::set_impl(utp_socket_impl* impl)
{
	TORRENT_ASSERT(m_impl == 0);
	TORRENT_ASSERT(!m_open);
	m_impl = impl;
	m_open = true;
}

int utp_stream::read_buffer_size() const
{
	TORRENT_ASSERT(m_impl);
	return m_impl->m_receive_buffer_size;
}

void utp_stream::on_read(void* self, size_t bytes_transferred, error_code const& ec, bool kill)
{
	utp_stream* s = (utp_stream*)self;

	UTP_LOGV("[%08u] %08p: calling read handler read:%d ec:%s kill:%d\n"
		, int(total_microseconds(time_now_hires() - min_time())), s->m_impl
		, bytes_transferred, ec.message().c_str(), kill);

	TORRENT_ASSERT(s->m_read_handler);
	s->m_io_service.post(boost::bind<void>(s->m_read_handler, ec, bytes_transferred));
	s->m_read_handler.clear();
	if (kill) s->m_impl = 0;
}

void utp_stream::on_write(void* self, size_t bytes_transferred, error_code const& ec, bool kill)
{
	utp_stream* s = (utp_stream*)self;

	UTP_LOGV("[%08u] %08p: calling write handler written:%d ec:%s kill:%d\n"
		, int(total_microseconds(time_now_hires() - min_time())), s->m_impl
		, bytes_transferred, ec.message().c_str(), kill);

	TORRENT_ASSERT(s->m_write_handler);
	s->m_io_service.post(boost::bind<void>(s->m_write_handler, ec, bytes_transferred));
	s->m_write_handler.clear();
	if (kill) s->m_impl = 0;
}

void utp_stream::on_connect(void* self, error_code const& ec, bool kill)
{
	utp_stream* s = (utp_stream*)self;

	UTP_LOGV("[%08u] %08p: calling connect handler ec:%s kill:%d\n"
		, int(total_microseconds(time_now_hires() - min_time())), s->m_impl, ec.message().c_str(), kill);

	TORRENT_ASSERT(s->m_connect_handler);
	s->m_io_service.post(boost::bind<void>(s->m_connect_handler, ec));
	s->m_connect_handler.clear();
	if (kill) s->m_impl = 0;
}

void utp_stream::add_read_buffer(void* buf, size_t len)
{
	TORRENT_ASSERT(m_impl);
	m_impl->m_read_buffer.push_back(utp_socket_impl::iovec_t(buf, len));
	m_impl->m_read_buffer_size += len;

    UTP_LOGV("[%08u] %08p: add_read_buffer %d bytes\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl, len);
}

// this is the wrapper to add a user provided write buffer to the
// utp_socket_impl. It makes sure the m_write_buffer_size is kept
// up to date
void utp_stream::add_write_buffer(void const* buf, size_t len)
{
	TORRENT_ASSERT(m_impl);

#ifndef NDEBUG
	int write_buffer_size = 0;
	for (std::vector<utp_socket_impl::iovec_t>::iterator i = m_impl->m_write_buffer.begin()
		, end(m_impl->m_write_buffer.end()); i != end; ++i)
	{
		write_buffer_size += i->len;
	}
	TORRENT_ASSERT(m_impl->m_write_buffer_size == write_buffer_size);
#endif

	m_impl->m_write_buffer.push_back(utp_socket_impl::iovec_t((void*)buf, len));
	m_impl->m_write_buffer_size += len;

#ifndef NDEBUG
	write_buffer_size = 0;
	for (std::vector<utp_socket_impl::iovec_t>::iterator i = m_impl->m_write_buffer.begin()
		, end(m_impl->m_write_buffer.end()); i != end; ++i)
	{
		write_buffer_size += i->len;
	}
	TORRENT_ASSERT(m_impl->m_write_buffer_size == write_buffer_size);
#endif

    UTP_LOGV("[%08u] %08p: add_write_buffer %d bytes\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl, len);
}

// this is called when all user provided read buffers have been added
// and it's time to execute the async operation. The first thing we
// do is to copy any data stored in m_receive_buffer into the user
// provided buffer. This might be enough to in turn trigger the read
// handler immediately.
void utp_stream::set_read_handler(handler_t h)
{
	m_impl->m_read_handler = h;
	if (m_impl->test_socket_state()) return;

    UTP_LOGV("[%08u] %08p: new read handler. %d bytes in buffer\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl
        , m_impl->m_receive_buffer_size);

	// so, the client wants to read. If we already
	// have some data in the read buffer, move it into the
	// client's buffer right away
	if (m_impl->m_receive_buffer_size == 0) return;

	std::vector<utp_socket_impl::iovec_t>::iterator target = m_impl->m_read_buffer.begin();

	int pop_packets = 0;
	for (std::vector<packet*>::iterator i = m_impl->m_receive_buffer.begin()
		, end(m_impl->m_receive_buffer.end()); i != end;)
	{
		if (target == m_impl->m_read_buffer.end()) 
        {
            UTP_LOGV("  No more target buffers: %d bytes left in buffer\n"
                , m_impl->m_receive_buffer_size);
            TORRENT_ASSERT(m_impl->m_read_buffer.empty());
            break;
        }

        m_impl->check_receive_buffers();

		packet* p = *i;
		int to_copy = (std::min)(p->size - p->header_size, int(target->len));
        TORRENT_ASSERT(to_copy >= 0);
		memcpy(target->buf, p->buf + p->header_size, to_copy);
		m_impl->m_read += to_copy;
		target->buf = ((char*)target->buf) + to_copy;
		target->len -= to_copy;
		m_impl->m_receive_buffer_size -= to_copy;
		TORRENT_ASSERT(m_impl->m_read_buffer_size >= to_copy);
		m_impl->m_read_buffer_size -= to_copy;
		p->header_size += to_copy;
		if (target->len == 0) target = m_impl->m_read_buffer.erase(target);

        m_impl->check_receive_buffers();

        TORRENT_ASSERT(m_impl->m_receive_buffer_size >= 0);

        // Consumed entire packet
        if (p->header_size == p->size)
        {
            free(p);
            ++pop_packets;
            *i = 0;
            ++i;
        }

        if (m_impl->m_receive_buffer_size == 0)
        {
            UTP_LOGV("  Didn't fill entire target: %d bytes left in buffer\n"
              , m_impl->m_receive_buffer_size);
            break;
        }
	}
	// remove the packets from the receive_buffer that we already copied over
	// and freed
	m_impl->m_receive_buffer.erase(m_impl->m_receive_buffer.begin()
		, m_impl->m_receive_buffer.begin() + pop_packets);
	// we exited either because we ran out of bytes to copy
	// or because we ran out of space to copy the bytes to
	TORRENT_ASSERT(m_impl->m_receive_buffer_size == 0
		|| m_impl->m_read_buffer.empty());

    UTP_LOGV("[%08u] %08p: %d packets moved from buffer to user space\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl
        , pop_packets);

	m_impl->maybe_trigger_receive_callback(time_now_hires());
}

// this is called when all user provided write buffers have been
// added. Start trying to send packets with the payload immediately.
void utp_stream::set_write_handler(handler_t h)
{
    UTP_LOGV("[%08u] %08p: new write handler. %d bytes to write\n"
        , int(total_microseconds(time_now_hires() - min_time())), m_impl
        , m_impl->m_write_buffer_size);

	m_impl->m_write_handler = h;
	m_impl->m_written = 0;
	if (m_impl->test_socket_state()) return;
	// try to write. send_pkt returns false if there's
	// no more payload to send or if the congestion window
	// is full and we can't send more packets right now
	while (m_impl->send_pkt(false));
}

void utp_stream::do_connect(tcp::endpoint const& ep, utp_stream::connect_handler_t handler)
{
	TORRENT_ASSERT(m_impl->m_connect_handler == 0);
	m_impl->m_remote_address = ep.address();
	m_impl->m_port = ep.port();
	m_impl->m_connect_handler = handler;

	if (m_impl->test_socket_state()) return;
	m_impl->send_syn();
}

// =========== utp_socket_impl ============

utp_socket_impl::~utp_socket_impl()
{
    UTP_LOGV("[%08u] %08p: destroying utp socket state\n"
		, int(total_microseconds(time_now_hires() - min_time())), this);

	// free any buffers we're holding
	for (boost::uint16_t i = m_inbuf.cursor(), end((m_inbuf.cursor()
		+ m_inbuf.capacity()) & ACK_MASK);
		i != end; i = (i + 1) & ACK_MASK)
	{
		void* p = m_inbuf.remove(i);
		free(p);
	}
	for (boost::uint16_t i = m_outbuf.cursor(), end((m_outbuf.cursor()
		+ m_outbuf.capacity()) & ACK_MASK);
		i != end; i = (i + 1) & ACK_MASK)
	{
		void* p = m_outbuf.remove(i);
		free(p);
	}
}

void utp_socket_impl::maybe_trigger_receive_callback(ptime now)
{
	// nothing has been read or there's no outstanding read operation
	if (m_read == 0 || m_read_handler == 0) return;

	if (m_read > 10000 || m_read_buffer_size == 0 || now >= m_read_timeout)
	{
		UTP_LOGV("[%08u] %08p: calling read handler read:%d\n"
			, int(total_microseconds(time_now_hires() - min_time())), this
			, m_read);
		m_read_handler(m_userdata, m_read, m_error, false);
		m_read_handler = 0;
		m_read = 0;
		m_read_buffer_size = 0;
		m_read_buffer.clear();
	}
}

void utp_socket_impl::maybe_trigger_send_callback(ptime now)
{
	// nothing has been written or there's no outstanding write operation
	if (m_written == 0 || m_write_handler == 0) return;

	if (m_written > 10000 || m_write_buffer_size == 0 || now >= m_write_timeout)
	{
		UTP_LOGV("[%08u] %08p: calling write handler written:%d\n"
			, int(total_microseconds(time_now_hires() - min_time())), this
			, m_written);

		m_write_handler(m_userdata, m_written, m_error, false);
		m_write_handler = 0;
		m_written = 0;
		m_write_buffer_size = 0;
		m_write_buffer.clear();
	}
}

void utp_socket_impl::destroy()
{
	UTP_LOGV("[%08u] %08p: destroy state:%s\n"
		, int(total_microseconds(time_now_hires() - min_time())), this
		, socket_state_names[m_state]);

	if (m_userdata == 0) return;

	m_error = asio::error::operation_aborted;
	cancel_handlers(m_error, true);

	m_userdata = 0;
    m_read_buffer.clear();
    m_read_buffer_size = 0;

	if (m_state == UTP_STATE_ERROR_WAIT
		|| m_state == UTP_STATE_NONE
		|| m_state == UTP_STATE_SYN_SENT)
	{
		m_state = UTP_STATE_DELETE;
		UTP_LOGV("[%08u] %08p: state:%s\n"
			, int(total_microseconds(time_now_hires() - min_time())), this
			, socket_state_names[m_state]);
		return;
	}

	// you should never close a socket with an outstanding write!
	TORRENT_ASSERT(!m_write_handler);

	send_fin();

	// #error our end is closing. Wait for everything to be acked
}

void utp_socket_impl::send_syn()
{
	m_seq_nr = rand();
	m_acked_seq_nr = (m_seq_nr - 1) & ACK_MASK;
	m_ack_nr = 0;

	packet* p = (packet*)malloc(sizeof(packet) + sizeof(utp_header));
	p->size = sizeof(utp_header);
	p->header_size = sizeof(utp_header);
	p->num_transmissions = 1;
	p->need_resend = false;
	utp_header* h = (utp_header*)p->buf;
	h->ver = 1;
	h->type = ST_SYN;
	h->extension = 0;
	// using recv_id here is intentional! This is an odd
	// thing in uTP. The syn packet is sent with the connection
	// ID that it expects to receive the syn ack on. All
	// subsequent connection IDs will be this plus one.
	h->connection_id = m_recv_id;
	h->timestamp_difference_microseconds = m_reply_micro;
	h->wnd_size = 0;
	h->seq_nr = m_seq_nr;
	h->ack_nr = 0;

	ptime now = time_now_hires();
	p->send_time = now;
	h->timestamp_microseconds = total_microseconds(now - min_time());

	UTP_LOGV("[%08u] %08p: send_syn seq_nr:%d id:%d target:%s\n"
		, int(total_microseconds(now - min_time()))
		, this, int(m_seq_nr), int(m_recv_id)
		, print_endpoint(udp::endpoint(m_remote_address, m_port)).c_str());

	TORRENT_ASSERT(!m_error);
	m_sm->send_packet(udp::endpoint(m_remote_address, m_port), (char const*)h
		, sizeof(utp_header), m_error);
	if (m_error)
	{
		free(p);
		m_state = UTP_STATE_ERROR_WAIT;
		test_socket_state();
		return;
	}

	TORRENT_ASSERT(!m_outbuf.at(m_seq_nr));
	m_outbuf.insert(m_seq_nr, p);

	m_seq_nr = (m_seq_nr + 1) & ACK_MASK;

	m_state = UTP_STATE_SYN_SENT;
	UTP_LOGV("[%08u] %08p: state:%s\n"
		, int(total_microseconds(time_now_hires() - min_time())), this
		, socket_state_names[m_state]);
}

void utp_socket_impl::send_fin()
{
	bool ret = false;

	// we need a heap allocated packet in order to stick it
	// in the send buffer, so that we can resend it
	packet* p = (packet*)malloc(sizeof(packet) + sizeof(utp_header));

	p->size = sizeof(utp_header);
	p->header_size = sizeof(utp_header);
	p->num_transmissions = 1;
	p->need_resend = false;
	utp_header* h = (utp_header*)p->buf;

	h->ver = 1;
	h->type = ST_FIN;
	h->extension = 0;
	h->connection_id = m_send_id;
	h->timestamp_difference_microseconds = m_reply_micro;
	h->wnd_size = m_in_buf_size - m_buffered_incoming_bytes - m_receive_buffer_size;
	h->seq_nr = m_seq_nr;
	h->ack_nr = m_ack_nr;

	ptime now = time_now_hires();
	p->send_time = now;
	h->timestamp_microseconds = total_microseconds(now - min_time());

	m_sm->send_packet(udp::endpoint(m_remote_address, m_port)
		, (char const*)h, sizeof(utp_header), m_error);

	UTP_LOGV("[%08u] %08p: sending FIN seq_nr:%d ack_nr:%d type:%s "
		"id:%d target:%s size:%d error:%s send_buffer_size:%d\n"
		, int(total_microseconds(now - min_time()))
		, this, int(h->seq_nr), int(h->ack_nr), packet_type_names[h->type]
		, m_send_id, print_endpoint(udp::endpoint(m_remote_address, m_port)).c_str()
		, sizeof(utp_header), m_error.message().c_str(), m_write_buffer_size);

	if (m_error)
	{
		m_state = UTP_STATE_ERROR_WAIT;
		test_socket_state();
	}

#if !TORRENT_UT_SEQ
	// if the other end closed the connection immediately
	// our FIN packet will end up having the same sequence
	// number as the SYN, so this assert is invalid
	TORRENT_ASSERT(!m_outbuf.at(m_seq_nr));
#endif

	packet* old = (packet*)m_outbuf.insert(m_seq_nr, p);
	if (old)
	{
		if (!old->need_resend) m_bytes_in_flight -= old->size - old->header_size;
		free(old);
	}
	m_seq_nr = (m_seq_nr + 1) & ACK_MASK;

	m_state = UTP_STATE_FIN_SENT;

	UTP_LOGV("[%08u] %08p: state:%s\n"
		, int(total_microseconds(time_now_hires() - min_time())), this
		, socket_state_names[m_state]);
}

void utp_socket_impl::send_reset(utp_header* ph)
{
	utp_header h;
	h.ver = 1;
	h.type = ST_RESET;
	h.extension = 0;
	h.connection_id = m_send_id;
	h.timestamp_difference_microseconds = m_reply_micro;
	h.wnd_size = 0;
	h.seq_nr = rand();
	h.ack_nr = ph->seq_nr;
	ptime now = time_now_hires();
	h.timestamp_microseconds = total_microseconds(now - min_time());

	UTP_LOGV("[%08u] %08p: send_reset seq_nr:%d id:%d ack_nr:%d\n"
		, int(total_microseconds(now - min_time()))
		, this, int(h.seq_nr), int(m_send_id), int(ph->seq_nr));

	// ignore errors here
	error_code ec;
	m_sm->send_packet(udp::endpoint(m_remote_address, m_port), (char const*)&h, sizeof(h), ec);
}

std::size_t utp_socket_impl::available() const
{
	return m_receive_buffer_size;
}

void utp_socket_impl::parse_sack(char const* ptr, int size, int* acked_bytes
	, ptime const now, boost::uint32_t& min_rtt)
{
	if (size == 0) return;

	// this is the sequence number the current bit represents
	int ack_nr = (m_acked_seq_nr + 2) & ACK_MASK;

	UTP_LOGV("[%08u] %08p: got SACK ack_nr:%d "
		, int(total_microseconds(now - min_time())), this
		, ack_nr);

	// for each byte
	for (char const* end = ptr + size; ptr != end; ++ptr)
	{
		unsigned char bitfield = unsigned(*ptr);
		unsigned char mask = 1;
		// for each bit
		for (int i = 0; i < 8; ++i)
		{
			UTP_LOGV("%d", mask & bitfield ? 1 : 0);
			if (mask & bitfield)
			{
				// this bit was set, ack_nr was received
				packet* p = (packet*)m_outbuf.remove(ack_nr);
				if (p)
				{
					acked_bytes += p->size - p->header_size;
					UTP_LOGV("[%08u] %08p: acked packet %d (%d bytes)\n"
						, int(total_microseconds(now - min_time())), this
						, ack_nr, p->size - p->header_size);
					ack_packet(p, now, min_rtt);
					// each ACKed packet counts as a duplicate ack
					++m_duplicate_acks;
				}
			}

			mask <<= 1;
			ack_nr = (ack_nr + 1) & ACK_MASK;

			// we haven't sent packets past this point.
			// if there are any more bits set, we have to
			// ignore them anyway
			if (ack_nr == m_seq_nr)
			{
				UTP_LOGV("\n");
				return;
			}
		}
	}
	UTP_LOGV("\n");
}

// copies data from the write buffer into the packet
// pointed to by ptr
void utp_socket_impl::write_payload(char* ptr, int size)
{
#ifndef NDEBUG
	int write_buffer_size = 0;
	for (std::vector<iovec_t>::iterator i = m_write_buffer.begin()
		, end(m_write_buffer.end()); i != end; ++i)
	{
{'fraction_non_alphanumeric': 0.11188998926142465, 'fraction_numerical': 0.008799665910989142, 'mean_word_length': 4.056867033712949, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 321, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, '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': '4780014', 'n_tokens_mistral': 25392, 'n_tokens_neox': 23869, 'n_words': 7153}