Net-Tofu

#pragma once

#include "CommonTypes.h"
#include "SimpleException.h"
#include "Util.h"

namespace Lucid
{
	const unsigned MAX_PACKET_SIZE = 512;
	const unsigned UDP_HEADER_RESERVED = 16;
	const unsigned MAX_SENDABLE_MESSAGE = MAX_PACKET_SIZE - UDP_HEADER_RESERVED - sizeof( uint64 );

	const unsigned CONNECTION_TIMEOUT_MILLIS = 5000;
	const unsigned KEEPALIVE_MILLIS = 500;

	const unsigned INITIAL_ESTIMATED_RTT_MILLIS = 32;

	typedef unsigned SessionID;
	const SessionID INVALID_SESSION_ID = (unsigned) -1;

	//=====================================================================
	//=====================================================================
	class UDPSocket;
	class UDPSocketImpl;

	class UDPAddress
	{
	public:
		UDPAddress();
		/**
		* Constructs an address from (port) and (ip)
		* @param port Port
		* @param ip SZ string IP, or nullptr for any IP
		**/
		UDPAddress( unsigned short port, const char* ip );

		bool operator==( const UDPAddress& ) const;
		bool operator<( const UDPAddress& ) const;

		/**
		* Returns true if the port and IP are valid
		*/
		bool isValid() const;

		friend class UDPSocket;
		friend class UDPSocketImpl;

	private:
		unsigned ip;
		unsigned short port;
	};

	//=====================================================================
	//=====================================================================
	class PortInUseException: public IOException
	{ public: const char* what() const throw() { return "Port already bound"; } };
	
	//=====================================================================
	//=====================================================================
	class InvalidSessionException: public IOException
	{ public: const char* what() const throw() { return "Attempt to reference invalid session"; } };

	//=====================================================================
	//=====================================================================
	class NetworkException: public IOException
	{ public: const char* what() const throw() { return "Error in network subsystem"; } };


#pragma warning( disable : 4100 ) // unreferenced parameters
	//=====================================================================
	// Listener class for receiving notifications from the socket
	//=====================================================================
	class UDPSocketListener
	{
	public:
		virtual void ReceivedMessage(
			const void* data, unsigned dataSize,
			SessionID id, const UDPAddress& remoteAddress ) { }

		virtual void SessionStarted(
			SessionID id, const UDPAddress& remoteAddress ) { }

		virtual void SessionEnded(
			SessionID id, const UDPAddress& remoteAddress,
			bool bUndeliveredMessages ) { }
	};

#pragma warning( default : 4100 )

	//=====================================================================
	//=====================================================================

	class UDPSocket
	{
	public:
		/*
		* Constructs a new UDPSocket.
		* @exception NetworkException
		*/
		UDPSocket();

		~UDPSocket();

		void addListener( UDPSocketListener* );
		void removeListener( UDPSocketListener* );

		/*
		* Begins listening for messages on (port)
		* @param port Port on which to accept incoming messages
		* @exception PortInUseException
		* @exception NetworkException 
		*/
		void bindPort( unsigned short port );

		
		// Send a datagram to (remoteAddr)
		// Send will complete during next update() call
		void send( const void* data, unsigned dataSize, const UDPAddress& remoteAddr );
		void send( const void* data, unsigned dataSize, SessionID sessionID );
			// throws InvalidSessionException()

		// Sends a datagram to (remoteAddr) repeatedly until a confirmation
		// is received.
		void sendGuaranteed( const void* data, unsigned dataSize, const UDPAddress& remoteAddr );
		void sendGuaranteed( const void* data, unsigned dataSize, SessionID sessionID );
			// throws InvalidSessionException()

		// Marks a session for termination.
		// Sessions will terminate when all pending guaranteed messages are
		// acknowledged or the session times out.
		void terminateSession( const UDPAddress& remoteAddr );
			// throws InvalidSessionException()
		void terminateSession( SessionID );
			// throws InvalidSessionException()

		void getRemoteAddressByID( SessionID, UDPAddress& outAddress );
			// throws InvalidSessionException()

		unsigned getRTTMillis( SessionID ) const;
			// throws InvalidSessionException()

		// Process sending and receiving.
		void update();
			// throws NetworkException()
			// TODO - more descriptive exceptions here

	private:
		UDPSocketImpl* m_;

		PREVENT_COPYING( UDPSocket )
	};



}

#include "UDPSocket.h"

#undef SIMULATE_PACKET_LOSS
#define SIMULATED_PACKET_LOSS_PERCENT 75

#define SIMULATE_NETWORK_LATENCY
#define ADDITIONAL_LATENCY_MILLIS 15

#ifdef WIN32
	#include <Winsock2.h>
	#pragma comment ( lib, "ws2_32.lib" )
#endif

#ifdef LINUX
	#include <sys/types.h>
	#include <sys/socket.h>
	#include <sys/fcntl.h>
	#include <sys/ioctl.h>
	#include <netinet/in.h>
	#include <arpa/inet.h>
	#include <netdb.h>
	#include <math.h>
	
	#define SOCKET int
	#define INVALID_SOCKET -1
	#define SOCKET_ERROR -1
	typedef sockaddr SOCKADDR;
#endif

#include "LucidTime.h"
#include "Statistics.h"

#include <string.h>
#include <vector>
#include <map>

#ifdef SIMULATE_NETWORK_LATENCY
	#include <queue>
#endif

namespace Lucid
{
	const unsigned RTT_POLL_FREQUENCY_MILLIS = 10000;
	const unsigned INITIAL_RTT_POLL_FREQUENCY_MILLIS = 1000;
	const unsigned RTT_SAMPLES = 10;

	const unsigned MIN_PUBLISH_RATE_MILLIS = 8;
	const unsigned PUBLISH_RATE_PERCENT_RTT = 50;

	const unsigned MIN_GUARANTEED_SENDRATE_MILLIS = 32;
	const unsigned GUARANTEED_SENDRATE_PERCENT_RTT = 200;
	const unsigned GUARANTEED_MESSAGE_MAX_RETRIES = 100;



	//=====================================================================
	//=====================================================================
	unsigned IPFromSZ( const char* sz )
	{
		if ( sz == nullptr ) return 0;

		unsigned result = 0;
		for ( int curQuad = 0; curQuad < 4; ++ curQuad )
		{
			unsigned char curByte = 0;
			int i;

			for ( i = 0; i < 3; ++i )
			{
				if ( sz[ i ] >= '0' && sz[ i ] <= '9' )
					curByte = curByte * 10 + sz[ i ] - '0';
				else
					break;
			}

			if ( i == 0 ) return (unsigned) -1;
			sz += i;

			if ( curQuad < 3 && *sz != '.' ) return (unsigned) -1;
			if ( curQuad == 3 && *sz != 0 ) return (unsigned) -1;
			++sz;

			result = ( result << 8 ) | curByte;
		}

		return result;
	}

	//=====================================================================
	//=====================================================================
	UDPAddress::UDPAddress() : ip( -1 ), port( 0 ) { }

	//=====================================================================
	//=====================================================================
	UDPAddress::UDPAddress( unsigned short port, const char* ip )
		: port( port ), ip( IPFromSZ( ip ) ) { }

	//=====================================================================
	//=====================================================================
	bool UDPAddress::operator==( const UDPAddress& other ) const
	{
		return ip == other.ip && port == other.port;
	}

	bool UDPAddress::operator<( const UDPAddress& other ) const
	{
		if ( ip == other.ip )
			return port < other.port;
		else
			return ip < other.ip;
	}

	//=====================================================================
	//=====================================================================
	bool UDPAddress::isValid() const
	{
		return ip != (unsigned) -1 && ip != 0 &&
			port != (unsigned short) -1 && port != 0;
	}

	//=====================================================================
	//=====================================================================
	enum UDPMessageType : uint64
	{
		PACKET_NORMAL, PACKET_GUARANTEED, MESSAGE_DELIVERY_ACK,
		PACKET_KEEPALIVE, PACKET_PING, PACKET_PING_RESPONSE
	};

	struct UDPHeader
	{
		UDPMessageType type;

		uint64 sequenceNum;
	};

	//=====================================================================
	//=====================================================================
	struct PendingPacket
	{
		UDPAddress destinationAddr;
		unsigned totalSize;
		
		SystemClocks lastSendTimeClocks;
		SystemClocks guaranteedResendClocks;

		unsigned sendAttempts;

		SessionID session;

		union
		{
			char rawData[ MAX_PACKET_SIZE ];
			UDPHeader header;
		};
	};

	//=====================================================================
	//=====================================================================
	struct SessionInfo
	{
		SessionID id;
		UDPAddress address;

		uint64 nextLocalSeq;
		uint64 nextRemoteSeq;

		SystemClocks lastReceiptTimeClocks;
		SystemClocks lastSendTimeClocks;
		SystemClocks lastPingTimeClocks;

		SystemClocks bufferCreationTimeClocks;

		PendingPacket packetBuffer;

		bool bTerminated;
		bool bTimeout;

		unsigned pendingGuaranteedCount;

		RunningAverage< SystemClocks, RTT_SAMPLES > RTTLog;

		void clearBuffer()
		{
			SystemClocks minResendRate = SecondsToClocks( MIN_GUARANTEED_SENDRATE_MILLIS * 0.001 );

			packetBuffer.session = id;

			packetBuffer.destinationAddr = address;
			packetBuffer.totalSize = UDP_HEADER_RESERVED;

			packetBuffer.lastSendTimeClocks = 0;
			packetBuffer.guaranteedResendClocks = ( RTTLog.getAverage() * GUARANTEED_SENDRATE_PERCENT_RTT ) / 100;
			if ( packetBuffer.guaranteedResendClocks < minResendRate )
				packetBuffer.guaranteedResendClocks = minResendRate;

			packetBuffer.header.type = PACKET_NORMAL;
			packetBuffer.header.sequenceNum = 0;

			packetBuffer.sendAttempts = 0;
			bufferCreationTimeClocks = (SystemClocks) -1;
		}
	};

	//=====================================================================
	//=====================================================================
	struct WinsockInitializer
	{

		bool bInitialized;
		WinsockInitializer() : bInitialized( false ) { }

		void InitWinsock() {
			if ( bInitialized ) return;

#ifdef WIN32
			WSADATA data;
			if ( WSAStartup( WINSOCK_VERSION, &data ) != 0 || data.wVersion != WINSOCK_VERSION )
				throw ResourceException( "Error initializing Winsock" );
#endif
		}

#ifdef WIN32
		~WinsockInitializer()
		{ if ( 0 != WSACleanup() ) { } /* trace( "Error shutting down Winsock" ) */ }
#endif
	};

	//=====================================================================
	//=====================================================================
	class UDPSocketImpl
	{
	public:
		UDPSocketImpl()
		{
			s_winsockInit.InitWinsock();

			sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
			if ( sock == INVALID_SOCKET )
				throw NetworkException();	
			
			// Set socket to non blocking
#ifdef WIN32
			unsigned long setNonBlocking = 1;
			if ( 0 != ioctlsocket( sock, FIONBIO, &setNonBlocking ) )
			{
				if ( 0 != closesocket( sock ) ) trace( "Error shutting down socket" );
				throw NetworkException();
			}
#endif

#ifdef LINUX
			int x;
			x = fcntl( sock, F_GETFL, 0 );
			fcntl( sock, F_SETFL, x | O_NONBLOCK );
#endif

			nextSessionID = 0;
		}

		//=================================================================
		//=================================================================
		~UDPSocketImpl()
		{
#ifdef WIN32
			if ( 0 != closesocket( sock ) ) trace( "Error shutting down socket" );
#endif
#ifdef LINUX
			if ( 0 != close( sock ) ) trace( "Error shutting down socket" );
#endif
		}

		//=================================================================
		//=================================================================
		void queueMessage( SessionInfo& session, const void* data,
			unsigned dataSize, bool bGuaranteed = false )
		{
			if ( dataSize > MAX_SENDABLE_MESSAGE )
				throw std::exception( /* "Invalid argument, message too large to send as one packet" */ );

			// No room for the new message in the packet buffer? -publish the buffer
			if ( session.packetBuffer.totalSize + dataSize + sizeof( uint64 ) > MAX_PACKET_SIZE )
				publishPacket( session );

			uint64* ptMessageSize = reinterpret_cast< uint64* >( &session.packetBuffer.rawData[ session.packetBuffer.totalSize ] );
			char* ptMessage = reinterpret_cast< char* >( ptMessageSize + 1 );

			// Copy the message to the packet
			*ptMessageSize = dataSize;
			memcpy( ptMessage, data, dataSize );

			// Align messages within packet to 8-byte boundary
			unsigned paddingAmt = dataSize % sizeof( uint64 );
			if ( paddingAmt != 0 )
			{
				paddingAmt = sizeof( uint64 ) - paddingAmt;

				memset( ptMessage + dataSize, 0, paddingAmt );
			}

			session.packetBuffer.totalSize += dataSize + sizeof( uint64 ) + paddingAmt;

			if ( bGuaranteed ) session.packetBuffer.header.type = PACKET_GUARANTEED;

			if ( session.bufferCreationTimeClocks == (SystemClocks) -1 )
				session.bufferCreationTimeClocks = GetAbsoluteTimeClocks();
		}

		//=================================================================
		//=================================================================
		void publishPacket( SessionInfo& session )
		{
			if ( session.packetBuffer.header.type == PACKET_GUARANTEED )
			{
				session.packetBuffer.header.sequenceNum = session.nextLocalSeq++;
				pendingGuaranteedPackets.push_back( session.packetBuffer );

				++session.pendingGuaranteedCount;
			}
			else
			{
				pendingPackets.push_back( session.packetBuffer );
			}

			session.clearBuffer();

			session.lastSendTimeClocks = GetAbsoluteTimeClocks();
		}

		//=================================================================
		//=================================================================
		void keepAlive( SessionInfo& session )
		{
			PendingPacket keepAlivePacket;

			keepAlivePacket.destinationAddr = session.address;
			keepAlivePacket.lastSendTimeClocks = 0;
			keepAlivePacket.sendAttempts = 0;
			keepAlivePacket.totalSize = UDP_HEADER_RESERVED;

			keepAlivePacket.header.type = PACKET_KEEPALIVE;
			keepAlivePacket.header.sequenceNum = 0;

			keepAlivePacket.session = session.id;

			pendingPackets.push_back( keepAlivePacket );

			session.lastSendTimeClocks = GetAbsoluteTimeClocks();
		}

		//=================================================================
		//=================================================================
		void sendPing ( SessionInfo& session )
		{
			PendingPacket keepAlivePacket;

			keepAlivePacket.destinationAddr = session.address;
			keepAlivePacket.lastSendTimeClocks = 0;
			keepAlivePacket.sendAttempts = 0;
			keepAlivePacket.totalSize = UDP_HEADER_RESERVED;

			keepAlivePacket.header.type = PACKET_PING;
			keepAlivePacket.header.sequenceNum = (SystemClocks) GetAbsoluteTimeClocks();

			keepAlivePacket.session = session.id;

			pendingPackets.push_back( keepAlivePacket );

			session.lastSendTimeClocks = GetAbsoluteTimeClocks();
		}

		//=================================================================
		//=================================================================
		void respondToPing( SessionInfo& session, uint64 timestamp )
		{
			PendingPacket keepAlivePacket;

			keepAlivePacket.destinationAddr = session.address;
			keepAlivePacket.lastSendTimeClocks = 0;
			keepAlivePacket.sendAttempts = 0;
			keepAlivePacket.totalSize = UDP_HEADER_RESERVED;

			keepAlivePacket.header.type = PACKET_PING_RESPONSE;
			keepAlivePacket.header.sequenceNum = (SystemClocks) timestamp;

			keepAlivePacket.session = session.id;

			pendingPackets.push_back( keepAlivePacket );

			session.lastSendTimeClocks = GetAbsoluteTimeClocks();
		}

		//=================================================================
		//=================================================================
		void processAcknowledge( SessionInfo& session, uint64 sequenceNum )
		{
			for ( int i = (int) pendingGuaranteedPackets.size() - 1; i >= 0; --i )
			{
				PendingPacket& curPacket = pendingGuaranteedPackets[ i ];

				if ( curPacket.session == session.id && curPacket.header.sequenceNum <= sequenceNum )
				{
					pendingGuaranteedPackets[ i ] = pendingGuaranteedPackets[ pendingGuaranteedPackets.size() - 1 ];
					pendingGuaranteedPackets.pop_back();

					--session.pendingGuaranteedCount;
				}
			}
		}

		//=================================================================
		//=================================================================
		void sendAcknowledge( SessionInfo& session )
		{
			if ( session.nextRemoteSeq == 0 ) return;

			PendingPacket ackPacket;

			ackPacket.destinationAddr = session.address;
			ackPacket.lastSendTimeClocks = 0;
			ackPacket.sendAttempts = 0;
			ackPacket.totalSize = UDP_HEADER_RESERVED;

			ackPacket.header.type = MESSAGE_DELIVERY_ACK;
			ackPacket.header.sequenceNum = session.nextRemoteSeq - 1;

			ackPacket.session = session.id;

			pendingPackets.push_back( ackPacket );

			session.lastSendTimeClocks = GetAbsoluteTimeClocks();
		}

		//=================================================================
		//=================================================================
		SessionInfo& getSession( SessionID id )
		{
			auto itr = activeSessions.find( id );
			if ( itr == activeSessions.end() )
				throw InvalidSessionException();

			return itr->second;
		}

		//=================================================================
		//=================================================================
		SessionInfo& getSession( const UDPAddress& address )
		{
			// TODO - maybe too algorithmically expensive using O( n )
			for ( auto itr = activeSessions.begin(); itr != activeSessions.end(); ++itr )
				if ( itr->second.address == address )
					return itr->second;

			throw InvalidSessionException();
		}

		//=================================================================
		//=================================================================
		SessionInfo& openSession( const UDPAddress& address )
		{
			SessionInfo newSession;

			newSession.address = address;
			newSession.id = nextSessionID++;

			newSession.lastReceiptTimeClocks = GetAbsoluteTimeClocks();
			newSession.lastSendTimeClocks = 0;
			newSession.lastPingTimeClocks = GetAbsoluteTimeClocks();

			newSession.nextLocalSeq = 0;
			newSession.nextRemoteSeq = 0;

			newSession.bTerminated = false;
			newSession.bTimeout = false;

			newSession.pendingGuaranteedCount = 0;

			newSession.RTTLog.reset( SecondsToClocks( INITIAL_ESTIMATED_RTT_MILLIS * 0.001 ) );
			newSession.clearBuffer();

			newSessions.push_back( newSession.id );

			return ( activeSessions[ newSession.id ] = newSession );
		}

		//=================================================================
		//=================================================================
		SessionInfo& getOrOpenSession( const UDPAddress& address )
		{
			try { 
				return getSession( address );
			}
			catch ( InvalidSessionException ) {
				return openSession( address );
			}
		}

		//=================================================================
		//=================================================================
		void notifyNewSessions()
		{
			try {
				for ( int sessionIdx = 0; sessionIdx < newSessions.size(); ++sessionIdx )
					for ( int listenerIdx = 0; listenerIdx < listeners.size(); ++listenerIdx )
					{
						SessionID newID = newSessions[ sessionIdx ];

						listeners[ listenerIdx ]->SessionStarted( newID,
							getSession( newID ).address );
					}
			} catch ( InvalidSessionException ) {
				PROMISES( false, "Internal error in socket class" );
			}
			newSessions.clear();
		}

		//=================================================================
		//=================================================================
		void notifyReceived( const SessionInfo& session, int bytesRemaining )
		{
			bytesRemaining -= UDP_HEADER_RESERVED;

			UDPHeader* ptPacketHeader = reinterpret_cast< UDPHeader* >( receiveBuffer );
			uint64* ptMessageSize = reinterpret_cast< uint64* >( &receiveBuffer[ UDP_HEADER_RESERVED ] );
			void* ptMessageBody = ptMessageSize + 1;

			while ( bytesRemaining >= sizeof( uint64 ) )
			{
				uint64 readBytes = *ptMessageSize + sizeof( uint64 );

				// Malformed packet?
				if ( readBytes > bytesRemaining )
					throw NetworkException();

				for ( int listenerIdx = 0; listenerIdx < listeners.size(); ++listenerIdx )
				{
					listeners[ listenerIdx ]->ReceivedMessage( ptMessageBody,
						(unsigned) *ptMessageSize, session.id, session.address );
				}

				ptMessageSize = OffsetByBytes( ptMessageSize, readBytes );
				ptMessageBody = ptMessageSize + 1;

				bytesRemaining -= (int) readBytes;
			}
		}

		//=================================================================
		//=================================================================
		bool GetNextPacket( int& outBytesReceived, sockaddr_in& outAddr )
		{
			unsigned long dataAvail = 0;

#ifdef WIN32
			if ( 0 != ioctlsocket( sock, FIONREAD, &dataAvail ) )
				throw NetworkException();
#endif
#ifdef LINUX
			if ( 0 != ioctl( sock, FIONREAD, &dataAvail ) )
				throw NetworkException();
#endif

			if ( dataAvail == 0 ) return false;

			int addrSize = sizeof( sockaddr_in );

#ifdef WIN32
			outBytesReceived = recvfrom( sock, receiveBuffer,
				MAX_PACKET_SIZE, 0, (SOCKADDR*) &outAddr, &addrSize );
#endif
#ifdef LINUX
			outBytesReceived = recvfrom( sock, receiveBuffer,
				MAX_PACKET_SIZE, 0, (SOCKADDR*) &outAddr, (unsigned*) &addrSize );
#endif

#ifdef SIMULATE_PACKET_LOSS
			if ( rand() % 100 < SIMULATED_PACKET_LOSS_PERCENT )
				return;
#endif

#ifdef WIN32
			if ( SOCKET_ERROR == outBytesReceived ) switch ( WSAGetLastError() )
			{
			case WSAEMSGSIZE: // too big
				outBytesReceived = MAX_PACKET_SIZE;
				break;

			case WSAECONNRESET: // port unreachable from previous send() call
			case WSAETIMEDOUT: // took too long
			case WSAENETRESET: // time to live expired
				outBytesReceived = 0;
				return true;
			default:
				throw NetworkException();
			}
#endif

#ifdef LINUX
			if ( 0 == outBytesReceived ) switch ( errno )
			{
			case EMSGSIZE: // too big
				outBytesReceived = MAX_PACKET_SIZE;
				break;

			case ECONNRESET: // port unreachable from previous send() call
			case ETIMEDOUT: // took too long
			case ENETRESET: // time to live expired
				return true;
			default:
				throw NetworkException();
			}
#endif
			return true;
		}

		//=================================================================
		//=================================================================
		void processReceive()
		{
			int bytesReceived;
			sockaddr_in remoteAddrIn;
			UDPAddress remoteAddress;

#ifdef SIMULATE_NETWORK_LATENCY
			while ( GetNextPacket( bytesReceived, remoteAddrIn ) )
			{
				PendingPacket newPacket;

				newPacket.destinationAddr.ip = ntohl( remoteAddrIn.sin_addr.s_addr );
				newPacket.destinationAddr.port = ntohs( remoteAddrIn.sin_port );
				newPacket.lastSendTimeClocks = GetAbsoluteTimeClocks();
				newPacket.totalSize = bytesReceived;

				memcpy( newPacket.rawData, receiveBuffer, MAX_PACKET_SIZE );

				latency_queue.push( newPacket );
			}
#endif
			
			// while there are packets on the socket
			
			for ( ; ; )
			{
				// receive the next packet
#ifdef SIMULATE_NETWORK_LATENCY
				if ( latency_queue.size() == 0 ) return;

				SystemClocks receiveDelayClocks = SecondsToClocks( ADDITIONAL_LATENCY_MILLIS * 0.001 );
				PendingPacket& curPacket = latency_queue.front();
				
				if ( curPacket.lastSendTimeClocks + receiveDelayClocks <= GetAbsoluteTimeClocks() )
				{
					memcpy( receiveBuffer, curPacket.rawData, MAX_PACKET_SIZE );
					remoteAddress = curPacket.destinationAddr;

					bytesReceived = curPacket.totalSize;
					latency_queue.pop();
				}
				else
				{
					return;
				}
#else
				if ( !GetNextPacket( bytesReceived, remoteAddrIn ) ) return;

				remoteAddress.ip = ntohl( remoteAddrIn.sin_addr.s_addr );
				remoteAddress.port = ntohs( remoteAddrIn.sin_port );
#endif
				
				// Got a packet, start processing it

				// Too small to be valid?
				if ( bytesReceived < UDP_HEADER_RESERVED ) continue;

				UDPHeader* header = (UDPHeader*) receiveBuffer;
				
				// Determine the packet's session
				SessionInfo* curSession;

				try {
					curSession = &getSession( remoteAddress );
				}
				catch ( InvalidSessionException ) {
					// No active session for this address

					// Do not start new sessions for OOO guaranteed messages,
					// ACKs, or keepalives
					bool bInvalidPacketForNewSession( false );

					switch ( header->type )
					{
					case PACKET_PING:
					case PACKET_PING_RESPONSE:
					case PACKET_KEEPALIVE:
					case MESSAGE_DELIVERY_ACK:
						bInvalidPacketForNewSession = true;
						break;

					case PACKET_GUARANTEED:
						if ( header->sequenceNum != 0 )
							bInvalidPacketForNewSession = true;
						break;
					}

					if ( bInvalidPacketForNewSession ) continue;

					curSession = &openSession( remoteAddress );

					// To avoid listeners getting a packet before a session started
					// message we need to inform them if there was a new session here
					notifyNewSessions();
				}

				// Update the last activity for the session
				if ( !curSession->bTerminated && !curSession->bTimeout )
					curSession->lastReceiptTimeClocks = GetAbsoluteTimeClocks();

				bool bNotifyListeners( true );

				switch ( header->type )
				{
				case PACKET_PING:
					bNotifyListeners = false;
					respondToPing( *curSession, header->sequenceNum );
					break;

				case PACKET_PING_RESPONSE:
					curSession->RTTLog.logSample(
						GetAbsoluteTimeClocks() - (SystemClocks) header->sequenceNum );
					bNotifyListeners = false;
					break;

				case PACKET_KEEPALIVE:
					bNotifyListeners = false;
					break;

				case MESSAGE_DELIVERY_ACK:
					bNotifyListeners = false;
					processAcknowledge( *curSession, header->sequenceNum );
					break;

				case PACKET_GUARANTEED:
					if ( header->sequenceNum == curSession->nextRemoteSeq )
						++curSession->nextRemoteSeq;
					else
						bNotifyListeners = false;

					sendAcknowledge( *curSession );

					break;

				case PACKET_NORMAL:
					break;

				// Malformed packet
				default: throw NetworkException();
				}
				
				if ( bNotifyListeners )
					notifyReceived( *curSession, bytesReceived );
			}
		}

		//=================================================================
		//=================================================================
		int winSend( PendingPacket& curPacket )
		{
			sockaddr_in remoteAddress;
			remoteAddress.sin_family = AF_INET;
			remoteAddress.sin_port = htons( curPacket.destinationAddr.port );
			remoteAddress.sin_addr.s_addr = htonl( curPacket.destinationAddr.ip );

			return sendto( sock, curPacket.rawData, curPacket.totalSize,
				0, (SOCKADDR*) &remoteAddress, sizeof( sockaddr_in ) );
		}

		//=================================================================
		//=================================================================
		void processSend()
		{
			SystemClocks keepAliveClocks = SecondsToClocks( KEEPALIVE_MILLIS * 0.001 );
			SystemClocks initialPingClocks = SecondsToClocks( INITIAL_RTT_POLL_FREQUENCY_MILLIS * 0.001 );
			SystemClocks normalPingClocks = SecondsToClocks( RTT_POLL_FREQUENCY_MILLIS * 0.001 );
			SystemClocks minPublishClocks = SecondsToClocks( MIN_PUBLISH_RATE_MILLIS * 0.001 );

			SystemClocks curTime = GetAbsoluteTimeClocks();

			for ( auto sessionItr = activeSessions.begin(); sessionItr != activeSessions.end(); ++sessionItr )
			{
				SessionInfo& curSession = sessionItr->second;

				SystemClocks publishClocks = ( curSession.RTTLog.getAverage() * PUBLISH_RATE_PERCENT_RTT ) / 100;
				if ( publishClocks < minPublishClocks ) publishClocks = minPublishClocks;

				// Send RTT pings
				if ( ( curSession.RTTLog.validSamples == RTT_SAMPLES && curSession.lastPingTimeClocks + normalPingClocks <= curTime ) ||
					( curSession.RTTLog.validSamples != RTT_SAMPLES &&curSession.lastPingTimeClocks + initialPingClocks <= curTime ) )
				{
					sendPing( curSession );
					curSession.lastPingTimeClocks = curTime;
				}

				// check sessions for packets that need publishing
				if ( curSession.bufferCreationTimeClocks != -1 && 
					( curSession.bTerminated ||
					curSession.bufferCreationTimeClocks + publishClocks <= curTime ) )
				{
					publishPacket( curSession );
				}

				// check sessions to see if keepalives are required
				else if ( !curSession.bTerminated &&
					curSession.lastSendTimeClocks + keepAliveClocks <= curTime )
				{
					if ( curSession.bufferCreationTimeClocks != -1 )
						publishPacket( curSession );
					else
						keepAlive( curSession );
				}

			}

			// send non-guaranteed packets
			while ( pendingPackets.size() > 0 )
			{
				int result = winSend( pendingPackets[ pendingPackets.size() - 1 ] );

				pendingPackets.pop_back();

				if ( result == SOCKET_ERROR )
					throw NetworkException();
			}
			
			// send any guaranteed packets that are due
			for ( int packetItr = (int) pendingGuaranteedPackets.size() - 1; packetItr >= 0; --packetItr )
			{
				PendingPacket& curPacket = pendingGuaranteedPackets[ packetItr ];

				if ( curPacket.lastSendTimeClocks + curPacket.guaranteedResendClocks <= curTime )
				{
					curPacket.lastSendTimeClocks = curTime;

					if ( ++curPacket.sendAttempts <= GUARANTEED_MESSAGE_MAX_RETRIES )
					{
						if ( winSend( curPacket ) == SOCKET_ERROR )
							throw NetworkException();
					}
					else
					{
						// if we are past its max retries, mark the session as terminated
						auto sessionItr = activeSessions.find( curPacket.session );
						if ( sessionItr != activeSessions.end() )
							sessionItr->second.bTimeout = true;

						// remove the packet from consideration
						pendingGuaranteedPackets[ packetItr ] = pendingGuaranteedPackets[ pendingGuaranteedPackets.size() - 1 ];
						pendingGuaranteedPackets.pop_back();
					}

				}
			}
		}

		//=================================================================
		//=================================================================
		void postUpdate()
		{
			SystemClocks timeoutClocks = SecondsToClocks( CONNECTION_TIMEOUT_MILLIS * 0.001 );
			SystemClocks curTime = GetAbsoluteTimeClocks();

			// check sessions for timeouts and terminations
			for ( auto sessionItr = activeSessions.begin(); sessionItr != activeSessions.end(); )
			{
				SessionInfo& curSession = sessionItr->second;

				if ( curSession.lastReceiptTimeClocks + timeoutClocks <= curTime ||
					curSession.bTimeout ||
					( curSession.bTerminated && curSession.pendingGuaranteedCount == 0 ) )
				{
					// remove any guaranteed messages posted by a terminated session
					processAcknowledge( curSession, (uint64) -1 );

					// notify listeners of terminated sessions
					for ( int listenItr = 0; listenItr < (int) listeners.size(); ++listenItr )
					{
						listeners[ listenItr ]->SessionEnded( curSession.id,
							curSession.address, curSession.bTimeout );
					}

					auto temp = sessionItr;
					++sessionItr;
					activeSessions.erase( temp );
				}
				else
				{
					++sessionItr;
				}
			}

		}

		//=================================================================
		//=================================================================
		static WinsockInitializer s_winsockInit;
		SOCKET sock;
		
		char receiveBuffer[ MAX_PACKET_SIZE ];
		
		std::vector< PendingPacket > pendingPackets;
		std::vector< PendingPacket > pendingGuaranteedPackets;

		std::map< SessionID, SessionInfo > activeSessions;
		unsigned nextSessionID;

		std::vector< UDPSocketListener* > listeners;

		std::vector< SessionID > newSessions;

#ifdef SIMULATE_NETWORK_LATENCY
		std::queue< PendingPacket > latency_queue;
#endif
	};

	WinsockInitializer UDPSocketImpl::s_winsockInit;

	//=====================================================================
	//=====================================================================
	UDPSocket::UDPSocket() : m_( new UDPSocketImpl() ) { }
	UDPSocket::~UDPSocket() { delete m_; }

	//=====================================================================
	//=====================================================================
	void UDPSocket::addListener( UDPSocketListener* newListener )
	{
		removeListener( newListener );
		m_->listeners.push_back( newListener );
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::removeListener( UDPSocketListener* oldListener )
	{
		for ( int i = 0; i < m_->listeners.size(); ++i )
			if ( m_->listeners[ i ] == oldListener )
			{
				m_->listeners[ i ] = m_->listeners[ m_->listeners.size() - 1 ];
				m_->listeners.pop_back();
				--i;
			}
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::bindPort( unsigned short port )
	{
		sockaddr_in bindAddr;
		bindAddr.sin_family = AF_INET;
		bindAddr.sin_port = htons( port );
		bindAddr.sin_addr.s_addr = htonl( INADDR_ANY );
			
		int result = bind( m_->sock, (SOCKADDR *) &bindAddr, sizeof( sockaddr_in ) );
		
#ifdef WIN32
		if ( result == SOCKET_ERROR ) switch ( WSAGetLastError() )
		{
		case WSAEADDRINUSE: throw PortInUseException();
		// TODO - more detail on types of errors
		default: throw NetworkException();
		}
#endif
#ifdef LINUX
		if ( result == SOCKET_ERROR ) switch ( errno )
		{
		case EADDRINUSE: throw PortInUseException();
			// TODO - more detail on types of errors
		default: throw NetworkException();
		}
#endif
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::send( const void* data, unsigned dataSize, const UDPAddress& remoteAddr )
	{
		m_->queueMessage( m_->getOrOpenSession( remoteAddr ), data, dataSize );
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::send( const void* data, unsigned dataSize, SessionID sessionID )
	{
		m_->queueMessage( m_->getSession( sessionID ), data, dataSize );
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::sendGuaranteed( const void* data, unsigned dataSize, const UDPAddress& remoteAddr )
	{
		m_->queueMessage( m_->getOrOpenSession( remoteAddr ), data, dataSize, true );
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::sendGuaranteed( const void* data, unsigned dataSize, SessionID sessionID )
	{
		m_->queueMessage( m_->getSession( sessionID ), data, dataSize, true );
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::terminateSession( const UDPAddress& remoteAddr )
	{
		m_->getSession( remoteAddr ).bTerminated = true;
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::terminateSession( SessionID id )
	{
		m_->getSession( id ).bTerminated = true;
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::getRemoteAddressByID( SessionID sessionID, UDPAddress& outAddress )
	{
		outAddress = m_->getSession( sessionID ).address;
	}


	//=====================================================================
	//=====================================================================
	unsigned UDPSocket::getRTTMillis( SessionID id ) const
	{
		return (unsigned) ceil( ClocksToSeconds(
			m_->getSession( id ).RTTLog.getAverage() ) * 1000.0 );
	}

	//=====================================================================
	//=====================================================================
	void UDPSocket::update()
	{
		m_->notifyNewSessions();

		m_->processReceive();
		m_->processSend();

		m_->postUpdate();
	}
}