datablockprotocol.h

/*******************************************************************************
 * Copyright (c) 2023 Allied Vision Technologies GmbH
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *******************************************************************************/
 
#ifndef VISIONTRANSFER_DATABLOCKPROTOCOL_H
#define VISIONTRANSFER_DATABLOCKPROTOCOL_H
 
#include <map>
#include <vector>
#include <memory>
#include <chrono>
#include <deque>
 
#include "visiontransfer/alignedallocator.h"
#include "visiontransfer/exceptions.h"
 
namespace visiontransfer {
namespace internal {
 
class DataBlockProtocol {
public:
    enum ProtocolType {
        PROTOCOL_TCP,
        PROTOCOL_UDP
    };
 
    //
    static const int MAX_DATA_BLOCKS = 8;
 
    // Constants that are also used in other places.
    static const int MAX_TCP_BYTES_TRANSFER = 0xFFFF; //64K - 1
    static const int MAX_UDP_RECEPTION = 0x4000; //16K
    static const int MAX_OUTSTANDING_BYTES = 2*MAX_TCP_BYTES_TRANSFER;
 
#pragma pack(push,1)
    // Extends previous one-channel 6-byte raw header buffer
    //  Legacy transfers can be detected via non-zero netTransferSizeDummy
    struct HeaderPreamble {
        uint16_t netHeaderSize;
        int32_t netTransferSizeDummy; // layout compatibility, legacy detection
        uint32_t netTransferSizes[MAX_DATA_BLOCKS]; // per-block total size
    };
    struct SegmentHeaderUDP {
        uint32_t segmentOffset;
    };
    struct SegmentHeaderTCP {
        uint32_t fragmentSize;
        uint32_t segmentOffset;
    };
#pragma pack(pop)
 
    DataBlockProtocol(bool server, ProtocolType protType, int maxUdpPacketSize);
 
    int getProtocolOverhead() const {
        return protType == PROTOCOL_UDP ? sizeof(int) : 0;
    }
 
    int getMaxReceptionSize() const;
 
    void resetTransfer();
 
    void setTransferHeader(unsigned char* data, int headerSize, int blocks);
 
    void setTransferBytes(int block, long bytes);
 
    void setTransferData(int block, unsigned char* data,  int validBytes = 0x7FFFFFFF);
 
    void setTransferValidBytes(int block, int validBytes);
 
    const unsigned char* getTransferMessage(int& length);
 
    bool transferComplete();
 
    unsigned char* getNextReceiveBuffer(int maxLength);
 
    void resetReception(bool dropped);
 
    void processReceivedMessage(int length, bool& transferComplete);
 
    unsigned char* getReceivedData(int& length);
 
    unsigned char* getReceivedHeader(int& length);
 
    int getDroppedReceptions() const {
        return droppedReceptions;
    }
 
    bool newClientConnected();
 
    bool isConnected() const;
 
    const unsigned char* getNextControlMessage(int& length);
 
    unsigned char* getBlockReceiveBuffer(int block) {
        if (block >= numReceptionBlocks) {
            throw ProtocolException("Tried to get receive buffer beyond initialized block range");
        }
        return &blockReceiveBuffers[block][0];
    }
    int getBlockValidSize(int block) {
        if (block >= numReceptionBlocks) {
            throw ProtocolException("Tried to get valid buffer index beyond initialized block range");
        }
        return blockValidSize[block];
    }
    bool isBlockDone(int block) {
        if (block >= numReceptionBlocks) {
            throw ProtocolException("Tried to get completion status of uninitialized block");
        }
        return blockValidSize[block] >= blockReceiveSize[block];
    }
    bool allBlocksDone() {
        for (int i=0; i<numReceptionBlocks; ++i) {
            if (!isBlockDone(i)) return false;
        }
        return true;
    }
    bool anyPayloadReceived() {
        for (int i=0; i<numReceptionBlocks; ++i) {
            if (blockReceiveOffsets[i] > 0) return true;
        }
        return false;
    }
 
    std::string statusReport();
    
    bool wasHeaderReceived() const {
        return headerReceived;
    }
 
private:
    // The pimpl idiom is not necessary here, as this class is usually not
    // used directly
 
    struct MissingReceiveSegment {
        int offset;
        int length;
        bool isEof;
        unsigned char subsequentData[4];
    };
 
    static constexpr int HEARTBEAT_INTERVAL_MS = 1000;
    static constexpr int RECONNECT_TIMEOUT_MS = 2000;
 
    static constexpr unsigned char CONNECTION_MESSAGE = 0x01;
    static constexpr unsigned char CONFIRM_MESSAGE = 0x02;
    static constexpr unsigned char HEADER_MESSAGE = 0x03;
    static constexpr unsigned char RESEND_MESSAGE = 0x04;
    static constexpr unsigned char EOF_MESSAGE = 0x05;
    static constexpr unsigned char HEARTBEAT_MESSAGE = 0x06;
 
    bool isServer;
    ProtocolType protType;
    int maxPayloadSize;
    int minPayloadSize;
 
    // Transfer related variables
    bool transferDone;
    unsigned char* rawDataArr[MAX_DATA_BLOCKS];
    int rawDataArrStrideHackOrig[MAX_DATA_BLOCKS];
    int rawDataArrStrideHackRepl[MAX_DATA_BLOCKS];
    int rawValidBytes[MAX_DATA_BLOCKS];
    int transferOffset[MAX_DATA_BLOCKS];
    int transferSize[MAX_DATA_BLOCKS];
    char overwrittenTransferData[sizeof(SegmentHeaderTCP)];
    int overwrittenTransferIndex;
    int overwrittenTransferBlock;
    unsigned char* transferHeaderData;
    int transferHeaderSize;
    int totalBytesCompleted;
    int totalTransferSize;
    int numTransferBlocks;
    int lastTransmittedBlock;
 
    // Reliability related variables
    std::deque<MissingReceiveSegment> missingReceiveSegments[MAX_DATA_BLOCKS];
    std::deque<std::pair<int, int> > missingTransferSegments;
    bool waitingForMissingSegments;
    int totalReceiveSize;
 
    unsigned char controlMessageBuffer[1024 * 16];
 
    // Connection related variables
    bool connectionConfirmed;
    bool confirmationMessagePending;
    bool eofMessagePending;
    bool clientConnectionPending;
    bool resendMessagePending;
    std::chrono::steady_clock::time_point lastRemoteHostActivity;
    std::chrono::steady_clock::time_point lastSentHeartbeat;
    std::chrono::steady_clock::time_point lastReceivedHeartbeat;
 
    // Reception related variables
    std::vector<unsigned char, AlignedAllocator<unsigned char> > receiveBuffer;
    std::vector<unsigned char, AlignedAllocator<unsigned char> > blockReceiveBuffers[MAX_DATA_BLOCKS];
    int blockReceiveOffsets[MAX_DATA_BLOCKS];
    int blockReceiveSize[MAX_DATA_BLOCKS];
    int blockValidSize[MAX_DATA_BLOCKS];
    std::vector<unsigned char> receivedHeader;
    bool finishedReception;
    int droppedReceptions;
    int completedReceptions;
    double lostSegmentRate;
    int lostSegmentBytes;
    unsigned char unprocessedMsgPart[MAX_OUTSTANDING_BYTES];
    int unprocessedMsgLength;
    bool headerReceived;
    bool legacyTransfer;
    int numReceptionBlocks;
    int receiveOffset;
 
    const unsigned char* extractPayload(const unsigned char* data, int& length, bool& error);
    bool processControlMessage(int length);
    void restoreTransferBuffer();
    bool generateResendRequest(int& length);
    void getNextTransferSegment(int& block, int& offset, int& length);
    void parseResendMessage(int length);
    void parseEofMessage(int length);
    void integrateMissingUdpSegments(int block, int lastSegmentOffset, int lastSegmentSize);
    void processReceivedUdpMessage(int length, bool& transferComplete);
    void processReceivedTcpMessage(int length, bool& transferComplete);
    void resizeReceiveBuffer();
    int parseReceivedHeader(int length, int offset);
    void zeroStructures();
    void splitRawOffset(int rawSegmentOffset, int& dataBlockID, int& segmentOffset);
    int mergeRawOffset(int dataBlockID, int segmentOffset, int reserved=0);
 
};
 
}} // namespace
 
#endif