MessageReliableUDP.h

Go to the documentation of this file.

/*  Copyright (C) 2003-2021 Free Electron Organization
    Any use of this software requires a license.  If a valid license
    was not distributed with this file, visit freeelectron.org. */

/** @file */

#ifndef __MessageReliableUDP_h__
#define __MessageReliableUDP_h__

// Future improvements:
//
// 1. Track the percentage of failed packets and adjust the number of
//    burst packets based on that percentage
//
// 2. Multiple transmit threads could be used for different IP addresses
//
// 3. Add the ability to add a max timeout of sending a message
//
// 4. Better unit tests - add the ability drop percentage of random packets

#pragma pack(push, 1)

namespace fe
{
namespace ext
{

// Max number of bytes per packet
#define RELIABLE_UDP_PACKET_SIZE (1024)

// Max number of packets of RELIABLE_UDP_PACKET_SIZE
#define RELIABLE_UDP_MAX_PACKETS (128)

// Minimum time after last transmition to receiver before
// checking for missed packets
#define RELIABLE_UDP_MIN_RESPONSE_TIME (100)

// Default max packets transmitted at one time before delaying to transmit more
#define RELIABLE_UDP_BURST_PACKET_LIMIT (10)

#define RELIABLE_UDP_RECEIVED_MAX_LIST (64)

//------------------------------------------------------------------------------
struct MultiPacketMsg
{
    // Numbers re-cycled but any message in that hasn't been received before
    // being re-cycled
    uint16_t    messageID;
    // Port of sending verification messages to the sender
    uint16_t    verificationPort;
    // Full message size
    uint32_t    messageSize;
    // byte offset into message = offset * RELIABLE_UDP_PACKET_SIZE
    uint16_t    offset;
    // Message size is assumed to be RELIABLE_UDP_PACKET_SIZE bytes unless
    // it is the last packet and then it is calculated based on
    // the full message size the full message size is sent
    // with each message in case the first message is missed so
    // the memory can be allocated on the receiving side
    uint8_t     data[RELIABLE_UDP_PACKET_SIZE];
};

//------------------------------------------------------------------------------
// Single packet response of the state of all the multi-packet messages
//currently received by the client
struct MultiPacketResponseMsg
{
    uint16_t    messageID;
    std::bitset<RELIABLE_UDP_MAX_PACKETS> packetsReceived;
};

//------------------------------------------------------------------------------
struct MultiPacketMsgInProgress
{
    uint16_t    messageID;                  // Message ID
    uint32_t    messageSize;                // Size of message

    // NOTE will be deleted when transmitted successfully or timed out
    uint8_t*    message;                    // Message buffer

    char        ipAddress[INET_ADDRSTRLEN]; // ipAddress xxx.xxx.xxx.xxx\0
    uint16_t    port;                       // port
    int64_t     lastTransmitTime;           // Last time a packet was sent
    int64_t     lastReceivedTime;           // Last time a response was received

    // Bit array of packets sent
    std::bitset<RELIABLE_UDP_MAX_PACKETS> packetsSent;
    // Bit array of packets that have be verified received by the receiver
    std::bitset<RELIABLE_UDP_MAX_PACKETS> packetsReceived;

    bool        doInitialTransmition;       // Initial transmition
    bool        updated;                    // Flag to msg was just updated

    bool partOfThisMessage(const MultiPacketMsg &msg, const char *msgIP) const
    {
        return(messageID == msg.messageID &&
                port == msg.verificationPort &&
                strncmp(ipAddress, msgIP, sizeof(ipAddress)) == 0);
    }

    bool receivedFullMessage() const
    {
        return ((packetsReceived.count() * RELIABLE_UDP_PACKET_SIZE) >=
                messageSize);
    }
};

//------------------------------------------------------------------------------
struct MultiPacketMsgReceived
{
    uint16_t    messageID;                  // Message ID
    char        ipAddress[INET_ADDRSTRLEN]; // received from xxx.xxx.xxx.xxx\0
    uint16_t    verificationPort;           // verification port used by sender

    bool operator==(const MultiPacketMsgInProgress &msg) const
    {
        return(messageID == msg.messageID &&
                verificationPort == msg.port &&
                strncmp(ipAddress, msg.ipAddress, sizeof(ipAddress)) == 0);
    }
};


/**************************************************************************//**
    @brief Message Sender/Receiver multi-packet reliable over UDP

    @ingroup message
*//***************************************************************************/
class FE_DL_PUBLIC MessageReliableUDP : virtual public MessageReliableUDPI
{
public:
    MessageReliableUDP();
    ~MessageReliableUDP();

    virtual bool start(uint16_t receivePort) override;
    virtual uint16_t getReceivePort() override;
    virtual void shutdown() override;
    virtual bool recvReliableFrom(uint8_t **msg, uint32_t &msgSize,
        char *fromIPaddress) override;
    virtual bool sendReliableTo(const uint8_t *msg, const uint32_t msgSize,
        const char *ipAddress, uint16_t port) override;

private:
    bool sendPacket(MultiPacketMsgInProgress &msg, uint32_t offset);
    void checkVerifications();
    void transmitThread();
    void receiveThread();
    int64_t now();
    bool sendResponseMsg(MultiPacketMsgInProgress &receivedMsg);
    bool alreadyReceived(const MultiPacketMsg &msg, const char *msgIP);
    bool alreadyReceived(const MultiPacketMsgInProgress &testMsg);

    bool m_initialized;
    bool m_running;
    Messagegram m_packetMsg;
    Messagegram m_verificationMsg;
    uint16_t m_msgID;           // Next message ID to be assigned
    uint16_t m_verificationPort;// Verification port
    uint16_t m_receivePort;     // Receive port
    int m_burstPacketLimit;     // Limits how many packets can be sent at once
                                // later improvements should very this based on
                                // the number of packets by the receiver

    fe::sp<fe::SingleMaster> m_spSingleMaster;

    bool m_transmitThreadDone;
    fe::sp<MessageI> m_transmitMsgSystem;
    std::thread* m_transmitThread;
    std::mutex m_transmitListLock;
    I32 m_transmitListLockCount;

    // List of messages being transmitted
    std::list<MultiPacketMsgInProgress> m_transmitList;

    bool m_receiveThreadDone;
    fe::sp<MessageI> m_receiveMsgSystem;
    std::thread* m_receiveThread;
    std::mutex m_receiveListLock;
    I32 m_receiveListLockCount;

    // List of messages being received
    std::list<MultiPacketMsgInProgress> m_receiveList;

    // List of messages that have already been received
    //  this is used to ignore any late re-transmit of old message in case the
    //  sender wasn't notified in time or lost the response message reply
    std::list<MultiPacketMsgReceived> m_fullyreceivedMsgList;
};

} // namespace ext
} // namespace fe

#pragma pack(pop)

#endif // __MessageReliableUDP_h__