ReliabilityLayer.h

Go to the documentation of this file.

#ifndef __RELIABILITY_LAYER_H
#define __RELIABILITY_LAYER_H

#include "RakMemoryOverride.h"
#include "MTUSize.h"
#include "DS_LinkedList.h"
#include "DS_List.h"
#include "SocketLayer.h"
#include "PacketPriority.h"
#include "DS_Queue.h"
#include "BitStream.h"
#include "InternalPacket.h"
#include "DataBlockEncryptor.h"
#include "RakNetStatistics.h"
#include "SHA1.h"
#include "DS_OrderedList.h"
#include "DS_RangeList.h"
#include "DS_BPlusTree.h"
#include "DS_MemoryPool.h"
#include "CCRakNetUDT.h"
#include "DS_Multilist.h"
#include "RakNetDefines.h"
#include "DS_Heap.h"

class PluginInterface2;
class RakNetRandom;
typedef uint64_t reliabilityHeapWeightType;

#define NUMBER_OF_ORDERED_STREAMS 32 // 2^5

#define RESEND_TREE_ORDER 32

#include "BitStream.h"

// int SplitPacketIndexComp( SplitPacketIndexType const &key, InternalPacket* const &data );
struct SplitPacketChannel//<SplitPacketChannel>
{
        CCTimeType lastUpdateTime;

        // This is here for progress notifications, since progress notifications return the first packet data, if available
        InternalPacket *firstPacket;

        // DataStructures::OrderedList<SplitPacketIndexType, InternalPacket*, SplitPacketIndexComp> splitPacketList;
        DataStructures::List<InternalPacket*> splitPacketList;
};
int RAK_DLL_EXPORT SplitPacketChannelComp( SplitPacketIdType const &key, SplitPacketChannel* const &data );

// Helper class
struct BPSTracker
{
        BPSTracker();
        ~BPSTracker();
        void Reset(const char *file, unsigned int line);
        void Push1(CCTimeType time, uint64_t value1);
//      void Push2(RakNetTimeUS time, uint64_t value1, uint64_t value2);
        uint64_t GetBPS1(CCTimeType time);
        uint64_t GetBPS1Threadsafe(CCTimeType time);
//      uint64_t GetBPS2(RakNetTimeUS time);
//      void GetBPS1And2(RakNetTimeUS time, uint64_t &out1, uint64_t &out2);
        uint64_t GetTotal1(void) const;
//      uint64_t GetTotal2(void) const;

        struct TimeAndValue2
        {
                TimeAndValue2();
                ~TimeAndValue2();
                TimeAndValue2(CCTimeType t, uint64_t v1);
        //      TimeAndValue2(RakNetTimeUS t, uint64_t v1, uint64_t v2);
        //      uint64_t value1, value2;
                uint64_t value1;
                CCTimeType time;
        };

        uint64_t total1, lastSec1;
//      uint64_t total2, lastSec2;
        DataStructures::Queue<TimeAndValue2> dataQueue;
        void ClearExpired1(CCTimeType time);
//      void ClearExpired2(RakNetTimeUS time);
};

class ReliabilityLayer//<ReliabilityLayer>
{
public:

        // Constructor
        ReliabilityLayer();

        // Destructor
        ~ReliabilityLayer();

        void Reset( bool resetVariables, int MTUSize );

        void SetEncryptionKey( const unsigned char *key );

        void SetTimeoutTime( RakNetTimeMS time );

        RakNetTimeMS GetTimeoutTime(void);

        bool HandleSocketReceiveFromConnectedPlayer(
                const char *buffer, unsigned int length, SystemAddress systemAddress, DataStructures::List<PluginInterface2*> &messageHandlerList, int MTUSize,
                SOCKET s, RakNetRandom *rnr, unsigned short remotePortRakNetWasStartedOn_PS3, CCTimeType timeRead);

        BitSize_t Receive( unsigned char**data );

        bool Send( char *data, BitSize_t numberOfBitsToSend, PacketPriority priority, PacketReliability reliability, unsigned char orderingChannel, bool makeDataCopy, int MTUSize, CCTimeType currentTime, uint32_t receipt );

        void Update( SOCKET s, SystemAddress systemAddress, int MTUSize, CCTimeType time,
                unsigned bitsPerSecondLimit,
                DataStructures::List<PluginInterface2*> &messageHandlerList,
                RakNetRandom *rnr, unsigned short remotePortRakNetWasStartedOn_PS3);
        

        bool IsCheater( void ) const;

        bool IsDeadConnection( void ) const;

        void KillConnection(void);

        RakNetStatistics * const GetStatistics( RakNetStatistics *rns );

        bool IsOutgoingDataWaiting(void);
        bool AreAcksWaiting(void);

        // Set outgoing lag and packet loss properties
        void ApplyNetworkSimulator( double _maxSendBPS, RakNetTime _minExtraPing, RakNetTime _extraPingVariance );

        bool IsNetworkSimulatorActive( void );

        void SetSplitMessageProgressInterval(int interval);
        void SetUnreliableTimeout(RakNetTimeMS timeoutMS);
        bool AckTimeout(RakNetTimeMS curTime);
        CCTimeType GetNextSendTime(void) const;
        CCTimeType GetTimeBetweenPackets(void) const;
        CCTimeType GetAckPing(void) const;
        RakNetTimeMS GetTimeLastDatagramArrived(void) const {return timeLastDatagramArrived;}

        // If true, will update time between packets quickly based on ping calculations
        //void SetDoFastThroughputReactions(bool fast);

        // Encoded as numMessages[unsigned int], message1BitLength[unsigned int], message1Data (aligned), ...
        //void GetUndeliveredMessages(RakNet::BitStream *messages, int MTUSize);

private:
        void SendBitStream( SOCKET s, SystemAddress systemAddress, RakNet::BitStream *bitStream, RakNetRandom *rnr, unsigned short remotePortRakNetWasStartedOn_PS3, CCTimeType currentTime);

        BitSize_t WriteToBitStreamFromInternalPacket( RakNet::BitStream *bitStream, const InternalPacket *const internalPacket, CCTimeType curTime );


        InternalPacket* CreateInternalPacketFromBitStream( RakNet::BitStream *bitStream, CCTimeType time );

        unsigned RemovePacketFromResendListAndDeleteOlderReliableSequenced( const MessageNumberType messageNumber, CCTimeType time, DataStructures::List<PluginInterface2*> &messageHandlerList, SystemAddress systemAddress );

        void SendAcknowledgementPacket( const DatagramSequenceNumberType messageNumber, CCTimeType time);

        bool IsSendThrottled( int MTUSize );

        void UpdateWindowFromPacketloss( CCTimeType time );

        void UpdateWindowFromAck( CCTimeType time );

        BitSize_t GetMaxMessageHeaderLengthBits( void );
        BitSize_t GetMessageHeaderLengthBits( const InternalPacket *const internalPacket );

        void GetSHA1( unsigned char * const buffer, unsigned int nbytes, char code[ SHA1_LENGTH ] );

        bool CheckSHA1( char code[ SHA1_LENGTH ], unsigned char * const buffer, unsigned int nbytes );

        void DeleteSequencedPacketsInList( unsigned char orderingChannel, DataStructures::List<InternalPacket*>&theList, int splitPacketId = -1 );

        void DeleteSequencedPacketsInList( unsigned char orderingChannel, DataStructures::Queue<InternalPacket*>&theList );

        bool IsOlderOrderedPacket( OrderingIndexType newPacketOrderingIndex, OrderingIndexType waitingForPacketOrderingIndex );

        void SplitPacket( InternalPacket *internalPacket );

        void InsertIntoSplitPacketList( InternalPacket * internalPacket, CCTimeType time );

        InternalPacket * BuildPacketFromSplitPacketList( SplitPacketIdType splitPacketId, CCTimeType time,
                SOCKET s, SystemAddress systemAddress, RakNetRandom *rnr, unsigned short remotePortRakNetWasStartedOn_PS3);
        InternalPacket * BuildPacketFromSplitPacketList( SplitPacketChannel *splitPacketChannel, CCTimeType time );

        //void DeleteOldUnreliableSplitPackets( CCTimeType time );

        InternalPacket * CreateInternalPacketCopy( InternalPacket *original, int dataByteOffset, int dataByteLength, CCTimeType time );

        DataStructures::LinkedList<InternalPacket*> *GetOrderingListAtOrderingStream( unsigned char orderingChannel );

        void AddToOrderingList( InternalPacket * internalPacket );

        void InsertPacketIntoResendList( InternalPacket *internalPacket, CCTimeType time, bool firstResend, bool modifyUnacknowledgedBytes );

        void FreeMemory( bool freeAllImmediately );

        void FreeThreadedMemory( void );

        void FreeThreadSafeMemory( void );

        // Initialize the variables
        void InitializeVariables( int MTUSize );

        bool IsExpiredTime(unsigned int input, CCTimeType currentTime) const;

        // Make it so we don't do resends within a minimum threshold of time
        void UpdateNextActionTime(void);


        //unsigned int IsReceivedPacketHole(unsigned int input, RakNetTime currentTime) const;

        //unsigned int MakeReceivedPacketHole(unsigned int input) const;

        unsigned int GetResendListDataSize(void) const;

        void UpdateThreadedMemory(void);

        void CalculateHistogramAckSize(void);

        // Used ONLY for RELIABLE_ORDERED
        // RELIABLE_SEQUENCED just returns the newest one
        DataStructures::List<DataStructures::LinkedList<InternalPacket*>*> orderingList;
        DataStructures::Queue<InternalPacket*> outputQueue;
        int splitMessageProgressInterval;
        CCTimeType unreliableTimeout;

        struct MessageNumberNode
        {
                DatagramSequenceNumberType messageNumber;
                MessageNumberNode *next;
        };
        struct DatagramHistoryNode
        {
                DatagramHistoryNode() {}
                DatagramHistoryNode(MessageNumberNode *_head) :
                head(_head) {}
                MessageNumberNode *head;
        };
        // Queue length is programmatically restricted to DATAGRAM_MESSAGE_ID_ARRAY_LENGTH
        // This is essentially an O(1) lookup to get a DatagramHistoryNode given an index
        DataStructures::Queue<DatagramHistoryNode> datagramHistory;
        DataStructures::MemoryPool<MessageNumberNode> datagramHistoryMessagePool;


        void RemoveFromDatagramHistory(DatagramSequenceNumberType index);
        MessageNumberNode* GetMessageNumberNodeByDatagramIndex(DatagramSequenceNumberType index);
        void AddFirstToDatagramHistory(DatagramSequenceNumberType datagramNumber);
        MessageNumberNode* AddFirstToDatagramHistory(DatagramSequenceNumberType datagramNumber, DatagramSequenceNumberType messageNumber);
        MessageNumberNode* AddSubsequentToDatagramHistory(MessageNumberNode *messageNumberNode, DatagramSequenceNumberType messageNumber);
        DatagramSequenceNumberType datagramHistoryPopCount;
        
        DataStructures::MemoryPool<InternalPacket> internalPacketPool;
        // DataStructures::BPlusTree<DatagramSequenceNumberType, InternalPacket*, RESEND_TREE_ORDER> resendTree;
        InternalPacket *resendBuffer[RESEND_BUFFER_ARRAY_LENGTH];
        InternalPacket *resendLinkedListHead;
        InternalPacket *unreliableLinkedListHead;
        void RemoveFromUnreliableLinkedList(InternalPacket *internalPacket);
        void AddToUnreliableLinkedList(InternalPacket *internalPacket);
//      unsigned int numPacketsOnResendBuffer;
        //unsigned int blockWindowIncreaseUntilTime;
        //      DataStructures::RangeList<DatagramSequenceNumberType> acknowlegements;
        // Resend list is a tree of packets we need to resend

        // Set to the current time when the resend queue is no longer empty
        // Set to zero when it becomes empty
        // Set to the current time if it is not zero, and we get incoming data
        // If the current time - timeResendQueueNonEmpty is greater than a threshold, we are disconnected
//      CCTimeType timeResendQueueNonEmpty;
        RakNetTimeMS timeLastDatagramArrived;


        // If we backoff due to packetloss, don't remeasure until all waiting resends have gone out or else we overcount
//      bool packetlossThisSample;
//      int backoffThisSample;
//      unsigned packetlossThisSampleResendCount;
//      CCTimeType lastPacketlossTime;

        //DataStructures::Queue<InternalPacket*> sendPacketSet[ NUMBER_OF_PRIORITIES ];
        DataStructures::Heap<reliabilityHeapWeightType, InternalPacket*, false> outgoingPacketBuffer;
        reliabilityHeapWeightType outgoingPacketBufferNextWeights[NUMBER_OF_PRIORITIES];
        void InitHeapWeights(void);
        reliabilityHeapWeightType GetNextWeight(int priorityLevel);
//      unsigned int messageInSendBuffer[NUMBER_OF_PRIORITIES];
//      double bytesInSendBuffer[NUMBER_OF_PRIORITIES];


    DataStructures::OrderedList<SplitPacketIdType, SplitPacketChannel*, SplitPacketChannelComp> splitPacketChannelList;

        MessageNumberType sendReliableMessageNumberIndex;
        MessageNumberType internalOrderIndex;
        //unsigned int windowSize;
        RakNet::BitStream updateBitStream;
        OrderingIndexType waitingForOrderedPacketWriteIndex[ NUMBER_OF_ORDERED_STREAMS ], waitingForSequencedPacketWriteIndex[ NUMBER_OF_ORDERED_STREAMS ];
        
        // STUFF TO NOT MUTEX HERE (called from non-conflicting threads, or value is not important)
        OrderingIndexType waitingForOrderedPacketReadIndex[ NUMBER_OF_ORDERED_STREAMS ], waitingForSequencedPacketReadIndex[ NUMBER_OF_ORDERED_STREAMS ];
        bool deadConnection, cheater;
        // unsigned int lastPacketSendTime,retransmittedFrames, sentPackets, sentFrames, receivedPacketsCount, bytesSent, bytesReceived,lastPacketReceivedTime;
        SplitPacketIdType splitPacketId;
        RakNetTimeMS timeoutTime; // How long to wait in MS before timing someone out
        //int MAX_AVERAGE_PACKETS_PER_SECOND; // Name says it all
//      int RECEIVED_PACKET_LOG_LENGTH, requestedReceivedPacketLogLength; // How big the receivedPackets array is
//      unsigned int *receivedPackets;
        RakNetStatistics statistics;

//      CCTimeType histogramStart;
//      unsigned histogramBitsSent;


        // DataStructures::Queue<CCTimeType> hasReceivedPacketQueue;
        DataStructures::Queue<bool> hasReceivedPacketQueue;
        DatagramSequenceNumberType receivedPacketsBaseIndex;
        bool resetReceivedPackets;

        CCTimeType lastUpdateTime;
        CCTimeType timeBetweenPackets, nextSendTime, ackPing;
//      CCTimeType ackPingSamples[ACK_PING_SAMPLES_SIZE]; // Must be range of unsigned char to wrap ackPingIndex properly
        CCTimeType ackPingSum;
        unsigned char ackPingIndex;
        //CCTimeType nextLowestPingReset;
        RemoteSystemTimeType remoteSystemTime;
//      bool continuousSend;
//      CCTimeType lastTimeBetweenPacketsIncrease,lastTimeBetweenPacketsDecrease;
        // Limit changes in throughput to once per ping - otherwise even if lag starts we don't know about it
        // In the meantime the connection is flooded and overrun.
        CCTimeType nextAllowedThroughputSample;
        bool bandwidthExceededStatistic;

        // If Update::maxBitsPerSecond > 0, then throughputCapCountdown is used as a timer to prevent sends for some amount of time after each send, depending on
        // the amount of data sent
        long long throughputCapCountdown;

        DataBlockEncryptor encryptor;
        unsigned receivePacketCount;

        bool freeThreadedMemoryOnNextUpdate;

        //long double timeBetweenPacketsIncreaseMultiplier, timeBetweenPacketsDecreaseMultiplier;

#ifdef _DEBUG
        struct DataAndTime//<InternalPacket>
        {
                SOCKET s;
                char data[ MAXIMUM_MTU_SIZE ];
                unsigned int length;
                RakNetTimeMS sendTime;
        //      SystemAddress systemAddress;
                unsigned short remotePortRakNetWasStartedOn_PS3;
        };
        DataStructures::Queue<DataAndTime*> delayList;

        // Internet simulator
        double packetloss;
        RakNetTimeMS minExtraPing, extraPingVariance;
#endif

        CCTimeType elapsedTimeSinceLastUpdate;

        CCTimeType nextAckTimeToSend;

        RakNet::CCRakNetUDT congestionManager;
        uint32_t unacknowledgedBytes;
        
        bool ResendBufferOverflow(void) const;
        void ValidateResendList(void) const;
        void ResetPacketsAndDatagrams(void);
        void PushPacket(CCTimeType time, InternalPacket *internalPacket, bool isReliable);
        void PushDatagram(void);
        bool TagMostRecentPushAsSecondOfPacketPair(void);
        void ClearPacketsAndDatagrams(bool releaseToInternalPacketPoolIfNeedsAck);
        void MoveToListHead(InternalPacket *internalPacket);
        void RemoveFromList(InternalPacket *internalPacket, bool modifyUnacknowledgedBytes);
        void AddToListTail(InternalPacket *internalPacket, bool modifyUnacknowledgedBytes);
        void PopListHead(bool modifyUnacknowledgedBytes);
        bool IsResendQueueEmpty(void) const;
        void SortSplitPacketList(DataStructures::List<InternalPacket*> &data, unsigned int leftEdge, unsigned int rightEdge) const;
        void SendACKs(SOCKET s, SystemAddress systemAddress, CCTimeType time, RakNetRandom *rnr, unsigned short remotePortRakNetWasStartedOn_PS3);

        DataStructures::List<InternalPacket*> packetsToSendThisUpdate;
        DataStructures::List<bool> packetsToDeallocThisUpdate;
        // boundary is in packetsToSendThisUpdate, inclusive
        DataStructures::List<unsigned int> packetsToSendThisUpdateDatagramBoundaries;
        DataStructures::List<bool> datagramsToSendThisUpdateIsPair;
        DataStructures::List<unsigned int> datagramSizesInBytes;
        BitSize_t datagramSizeSoFar;
        BitSize_t allDatagramSizesSoFar;
        double totalUserDataBytesAcked;
        CCTimeType timeOfLastContinualSend;
        CCTimeType timeToNextUnreliableCull;

        // This doesn't need to be a member, but I do it to avoid reallocations
        DataStructures::RangeList<DatagramSequenceNumberType> incomingAcks;

        // Every 16 datagrams, we make sure the 17th datagram goes out the same update tick, and is the same size as the 16th
        int countdownToNextPacketPair;
        InternalPacket* AllocateFromInternalPacketPool(void);
        void ReleaseToInternalPacketPool(InternalPacket *ip);

        DataStructures::RangeList<DatagramSequenceNumberType> acknowlegements;
        DataStructures::RangeList<DatagramSequenceNumberType> NAKs;
        bool remoteSystemNeedsBAndAS;

        unsigned int GetMaxDatagramSizeExcludingMessageHeaderBytes(void);
        BitSize_t GetMaxDatagramSizeExcludingMessageHeaderBits(void);

        // ourOffset refers to a section within externallyAllocatedPtr. Do not deallocate externallyAllocatedPtr until all references are lost
        void AllocInternalPacketData(InternalPacket *internalPacket, InternalPacketRefCountedData **refCounter, unsigned char *externallyAllocatedPtr, unsigned char *ourOffset);
        // Set the data pointer to externallyAllocatedPtr, do not allocate
        void AllocInternalPacketData(InternalPacket *internalPacket, unsigned char *externallyAllocatedPtr);
        // Allocate new
        void AllocInternalPacketData(InternalPacket *internalPacket, unsigned int numBytes, const char *file, unsigned int line);
        void FreeInternalPacketData(InternalPacket *internalPacket, const char *file, unsigned int line);
        DataStructures::MemoryPool<InternalPacketRefCountedData> refCountedDataPool;

        BPSTracker bpsMetrics[RNS_PER_SECOND_METRICS_COUNT];
};

#endif