ConnectionReceiver.cpp

#include "Headers.h"
#include "ConnectionReceiver.h"
#include "PacketConnectionFeedback.h"
#include "Numeric.h"

#include "Debug.h"
#include "Console.h"

#pragma warning(disable : 4996)

double CConnectionReceiver::lossIntervalWeigth[MAX_LOSS_EVENTS] = {1.0, 1.0, 1.0, 1.0, 0.8, 0.6, 0.4, 0.2};

CConnectionReceiver::CConnectionReceiver(
    __uint32                                                            idEntry,
    __uint16                                                            port,
    CSimHandler*                                                        sim,
    IDelegate5<void, __uint16, __uint16, CAddress, __byte, CPacket*>*   delegateSend,
    IDelegate2<void, CConnectionReceiver*, CPacket*>*                   delegateRecv,
    IDelegate1<void, CConnection*>*                                     delegateDispose,
    CAddress                                                            remoteAddress,
    __uint16                                                            remotePort
    ) : CConnection(REQUESTER, idEntry, port, sim, delegateSend, delegateDispose, remoteAddress, remotePort)
{
    // Delegate receive
    this->delegateRecv = delegateRecv;

    // Flow
    this->flowLossRate = 0;
    this->flowRateRecv = 0;
    this->flowSequenceLast = 0;
    this->flowTimeLast = 0;

    // Timers
    this->timerFeedback = new CTimer<CConnectionReceiver>(this->sim, this, &CConnectionReceiver::TimerFeedback);

    // Receive function : set to first
    this->recvData = &CConnectionReceiver::RecvDataFirst;

    // Loss
    this->lossEventFirst = 0;
    this->lossEventLast = 0;
    this->lossEventCount = 0;

    // Receive rate
    this->recvSize = 0;

    // Feedback
    this->timeFeedbackLast = 0;
}

CConnectionReceiver::~CConnectionReceiver()
{
    // Timers
    delete this->timerFeedback;
}

void CConnectionReceiver::Recv(CPacketConnection *packet)
{
    // General processing for all received packets (without considering the connection state)
}

void CConnectionReceiver::RecvData(CPacketConnectionData* packet)
{
    // Check the packet source and destination
    assert(packet->Dst() == this->id);
    assert(packet->DstEntry() == this->idEntry);
    assert(packet->Flow() == this->flow);

    // Check the connection state : must be OPENED or greater
    if(this->state < OPENED) return;

#ifdef LOG_FLOW
    CConsole::SetColor(CConsole::LIGHT_GREEN);
    printf("\n\tRECV T = %8.3lf RecvData() : %s %s @ %u", this->sim->Time(), this->ToString(), packet->Payload()->ToString(), packet->Sequence());
    CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif

    // Global packet processing
    this->flowRtt = packet->Rtt();                                                              // Round-trip-time

    assert(this->flowRtt > 0);

    // Process packet
    (this->*this->recvData)(packet);

    // Send payload to the delegate
    (*this->delegateRecv)(this, packet->Payload());
}

void CConnectionReceiver::RecvDataFirst(CPacketConnectionData* packet)
{
    // Send a feedback packet
    CPacketConnectionFeedback* reply = new CPacketConnectionFeedback(
        this->flow,
        this->id,
        this->remoteId,
        this->idEntry,
        this->remoteIdEntry,
        this->sim->Time(),
        packet->TimeTx(),
        0,                  // Time delay
        this->flowRateRecv,
        this->flowLossRate,
        1,                  // Bitmap of acknowledged packets ( **** 0001)
        packet->Sequence(), // SEQ of first acknowledged packet
        0,                  // Bitmap of loss packets (N/A)
        0                   // SEQ of first loss packet
        );

    (*this->delegateSend)(this->port, this->remotePort, this->remoteAddress, 128, reply);

#ifdef LOG_FLOW
    CConsole::SetColor(CConsole::LIGHT_RED);
    printf("\n\tRECV T = %8.3lf SendFeedback() : %s", this->sim->Time(), this->ToString());
    CConsole::SetColor(CConsole::LIGHT_GRAY);
    printf("\n\t\tACK : %u", packet->Sequence());
#endif

    // Do not add the received packet info to packet history - the packet has already been acknowledged

    // Set the maximum sequence number and time to this packet (as first packet)
    this->flowSequenceLast = packet->Sequence();
    this->flowTimeLast = this->sim->Time();

    // Add the payload size to the received size
    this->recvSize += (packet->Payload())?packet->Payload()->Size():0;

    // Set the feedback timer
    this->timerFeedback->SetAfter(this->flowRtt);

    // Receive function : set to next
    this->recvData = &CConnectionReceiver::RecvDataNext;
}

void CConnectionReceiver::RecvDataNext(CPacketConnectionData* packet)
{
    /*
     * Step 0
     */
    // Record the Tx and Rx time for last received packet
    this->timeRxLast = packet->TimeTx();
    this->timeTxLast = this->sim->Time();

    // Add the payload size to the received size
    this->recvSize += (packet->Payload())?packet->Payload()->Size():0;

    /*
     * Step 1 : Add the received packet info to acknowledgement history : packet SEQ and received time
     */
    this->historyRecv.insert(pair<__uint32, __time>(packet->Sequence(), this->sim->Time()));

    /*
     * Step 2 : Check if packet indicates a new loss event
     */

    bool newLossEvent = false;

    // Calculate whether there are missing packets between this packet and the maximum ID number
    for(__uint32 seq = this->flowSequenceLast + 1; seq < packet->Sequence(); seq++)
    {
        // The packets with this SEQ is missing : add the missing SEQ at the end of the loss history
        this->historyLoss.push(seq);

        // Calculate the interpolated time of this lost packet
        __time lossTime = this->flowTimeLast + (this->sim->Time() - this->flowTimeLast) * (seq - this->flowSequenceLast) / (packet->Sequence() - this->flowSequenceLast);

        // If there exists a last loss event
        if(this->lossEventCount > 0)
        {
            // Check if the packet belongs to the last loss event
            
            // If the loss time is not within an RTT from the last loss event
            if(this->lossEvents[this->lossEventLast].firstLossTime + this->flowRtt < lossTime)
            {
                // Add lost packet belongs to a new loss event

                // Increment the last loss event index
                this->lossEventLast = (++this->lossEventLast) % MAX_LOSS_EVENTS;

                // If the maximum number of loss events has been reached
                if(this->lossEventFirst == this->lossEventLast)
                {
                    // Overwrite the first loss event
                    assert(MAX_LOSS_EVENTS == this->lossEventCount);
                    this->lossEventFirst = (++this->lossEventFirst) % MAX_LOSS_EVENTS;
                }
                else
                    // Increment the loss event count
                    this->lossEventCount++;

                // Add the time and sequence number to the last loss event
                this->lossEvents[this->lossEventLast].firstLossSequence = seq;  // The sequence of the lost packet
                this->lossEvents[this->lossEventLast].firstLossTime = lossTime; // The interpolated loss time

                // Set there has been a new loss event
                newLossEvent = true;
            }
            // Else the lost packet belongs to the last loss event
        }
        else
        {
            // Create the first loss event
            assert(this->lossEventFirst == this->lossEventLast);
            
            // Add the time and sequence number to the last loss event
            this->lossEvents[this->lossEventLast].firstLossSequence = seq;
            this->lossEvents[this->lossEventLast].firstLossTime = lossTime;

            // Increment the loss event count
            this->lossEventCount++;

            // Set there has been a new loss event
            newLossEvent = true;
        }
    }

    // Set the maximum sequence number and time to this packet (as last packet)
    this->flowSequenceLast = packet->Sequence();
    this->flowTimeLast = this->sim->Time();

    /*
     * Step 3 : Calculate loss rate (p)
     */

    // If there has been a new loss event
    if(newLossEvent)
    {
        assert(this->lossEventCount);

        // Set the last packet for which the loss rate is calculated
        this->lossSequenceLast = packet->Sequence();

        double lossInterval;

        // Reset previous lost intervals
        this->lossPrevInterval0 = 0;
        this->lossPrevInterval1 = 0;
        this->lossPrevWeigth = 0;

        // Calculate the previous loss intervals is the diff beween the the loss events
        for(__byte index = 1; index < this->lossEventCount; index++)
        {
            __byte begin = ((this->lossEventLast + MAX_LOSS_EVENTS) - index) % MAX_LOSS_EVENTS;
            __byte end = ((this->lossEventLast + MAX_LOSS_EVENTS) - index + 1) % MAX_LOSS_EVENTS;

            lossInterval = this->lossEvents[end].firstLossSequence - this->lossEvents[begin].firstLossSequence;

            this->lossPrevInterval0 += lossInterval * this->lossIntervalWeigth[index];
            this->lossPrevInterval1 += lossInterval * this->lossIntervalWeigth[index-1];
            this->lossPrevWeigth += this->lossIntervalWeigth[index];
        }

        // Calculate the current loss interva
        double lossInterval0 = (this->flowSequenceLast - this->lossEvents[this->lossEventLast].firstLossSequence + 1) * this->lossIntervalWeigth[0] + this->lossPrevInterval0;
        double lossWeight = this->lossIntervalWeigth[0] + this->lossPrevWeigth;
        lossInterval = MAX(lossInterval0, this->lossPrevInterval1);

        // Calculate the average loss interval
        double rateLoss = lossWeight / lossInterval;

        // If current loss rate is greater than previous loss rate (early loss detection)
        if(rateLoss > this->flowLossRate)
        {
            // Set the loss rate
            this->flowLossRate = rateLoss;

            // Expire the feedback timer
            assert(this->timerFeedback->IsSet());
            this->timerFeedback->Cancel();

            // Send feedback
            this->TimerFeedback(NULL);
        }
        else
        {
            // Set the loss rate
            this->flowLossRate = rateLoss;
        }
    }
    // Else, do nothing


    // Check the feedback timer is set
    assert(this->timerFeedback->IsSet());
}

void CConnectionReceiver::TimerFeedback(CTimerInfo* info)
{
    // The feedback timer expired

    // Only if at least one packet has been received since the last feedback

    if(!this->historyRecv.empty())
    {
#ifdef LOG_FLOW
        CConsole::SetColor(CConsole::LIGHT_RED);
        printf("\n\tRECV T = %8.3lf TimerFeedback() : %s (RTT=%.3lf)", this->sim->Time(), this->ToString(), this->flowRtt);
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
        /*
         * Step 1 : Calculate the loss rate
         */
        
        // If the last sequence number is different from the one for which the loss was previously calculated; 
        // and there are loss events
        if((this->flowSequenceLast != this->lossSequenceLast) && (this->lossEventCount))
        {
            // Use the partial data calculated at the last loss event

            double lossInterval0 = (this->flowSequenceLast - this->lossEvents[this->lossEventLast].firstLossSequence + 1) * this->lossIntervalWeigth[0];
            double lossWeigth = this->lossIntervalWeigth[0] + this->lossPrevWeigth;
            double lossInterval = MAX(lossInterval0, this->lossPrevInterval1);

            this->flowLossRate = lossWeigth / lossInterval;
        }

        /*
         * Step 2 : Calculate the received rate since the last expiration of the feedback timer
         */
        this->flowRateRecv = this->recvSize / (this->sim->Time() - this->timeFeedbackLast);

        // Reset the received size
        this->recvSize = 0;

        /*
         * Step 3 : Calculate the packets to acknowledge or report as loss
         */
        
        // Acknowledge ALL packets in the received history
        __uint32 ackSeq = 0;
        __uint64 ack = 0;

        if(this->historyRecv.size() > 0)    // If there are packets to acknowledge
        {
#ifdef LOG_FLOW
            printf("\n\t\tACK : ");
#endif
            // Select the first packet
            THistoryRecv::iterator iter = this->historyRecv.begin();
            ackSeq = iter->first;

            // Select the next packets
            for(; iter != this->historyRecv.end(); iter++)
            {
                // Calculate the bitmap index as the difference between the current ID and first ID
                __byte bit = iter->first - ackSeq;

                // Current implementation only alows acknowledgment of a 64 range of packets at a time
                assert(bit < 64);

                // Set the corresponing bit to 1
                ack |= (((__uint64)1) << bit);

#ifdef LOG_FLOW
                printf("%u ", iter->first);
#endif
            }

            // Clear the received history : all packets have been acknowledged
            this->historyRecv.clear();
        }

        // Report packets in the loss history with
        __uint32 lossSeq = 0;
        __uint64 loss = 0;

        if(this->historyLoss.size() > 0)    // If there are packets loss
        {
#ifdef LOG_FLOW
            printf("\n\t\tLOST : ");
#endif

            // Select the first packet
            lossSeq = this->historyLoss.front();

            // Select the next packets
            while(!this->historyLoss.empty())
            {
                // Get the front SEQ
                __uint32 seq = this->historyLoss.front();

                // Remove the front SEQ
                this->historyLoss.pop();

                // Calculate the bitmap index as the difference between the current ID and the first ID
                __byte bit = seq - lossSeq;
                
                // Current implementation only alows loss reporting of a 64 range of packets at a time
                assert(bit < 64);

                // Set the corresponding bit to 1
                loss |= (((__uint64)1) << bit);

#ifdef LOG_FLOW
                printf("%u ", seq);
#endif
            }
        }

        // Send a feedback packet
        CPacketConnectionFeedback* reply = new CPacketConnectionFeedback(
            this->flow,
            this->id,
            this->remoteId,
            this->idEntry,
            this->remoteIdEntry,
            this->timeTxLast,
            this->timeRxLast,
            this->sim->Time() - this->timeTxLast, // Time delay : current time minus last Tx time
            this->flowRateRecv,
            this->flowLossRate,
            ack,                // Bitmap of acknowledged packets ( **** 0001)
            ackSeq,             // SEQ of first acknowledged packet
            loss,               // Bitmap of loss packets (N/A)
            lossSeq             // SEQ of first loss packet
            );
        (*this->delegateSend)(this->port, this->remotePort, this->remoteAddress, 128, reply);
    }
    else
    {
        // Do nothing
#ifdef LOG_FLOW
        CConsole::SetColor(CConsole::LIGHT_YELLOW);
        printf("\n\tRECV T = %8.3lf TimerFeedback() : %s (RTT=%.3lf)", this->sim->Time(), this->ToString(), this->flowRtt);
        CConsole::SetColor(CConsole::LIGHT_GRAY);
#endif
    }
    
    // Set the current time of the expiration of the feedback timer
    this->timeFeedbackLast = this->sim->Time();

    // Set the feedback timer after one RTT
    this->timerFeedback->SetAfter(this->flowRtt);
}


char* CConnectionReceiver::ToString() const
{
    sprintf((char*)this->str, "[ l=*:%u:%u:%u r=%u:%u:%u:%u t=%s s=%s recv=%u loss=%u ]",
        this->port, this->id, this->idEntry, ((CAddress)this->remoteAddress).Address(), this->remotePort, this->remoteId, this->remoteIdEntry,
        CConnection::strConnectionType[this->type], CConnection::strState[this->state],
        this->historyRecv.size(), this->historyLoss.size()
        );

    return (char*)this->str;
}