e2sarDPReassembler.hpp

#ifndef E2SARDREASSEMBLERPHPP
#define E2SARDREASSEMBLERPHPP
 
#include <boost/asio.hpp>
#include <boost/lockfree/queue.hpp>
#include <boost/pool/pool.hpp>
#include <boost/pool/object_pool.hpp>
#include <boost/unordered_map.hpp>
#include <boost/thread.hpp>
#include <boost/tuple/tuple.hpp>
#include <boost/tuple/tuple_io.hpp>
#include <boost/circular_buffer.hpp>
#include <boost/any.hpp>
#include <boost/asio/ip/udp.hpp>
#include <boost/variant.hpp>
#include <boost/heap/priority_queue.hpp>
#include <boost/container/flat_set.hpp>
 
#include <sys/select.h>
 
#ifdef NETLINK_AVAILABLE
#include <linux/rtnetlink.h>
#endif
 
#include <atomic>
 
#include "e2sarError.hpp"
#include "e2sarUtil.hpp"
#include "e2sarHeaders.hpp"
#include "e2sarNetUtil.hpp"
#include "e2sarCP.hpp"
#include "portable_endian.h"
 
/***
 * Dataplane definitions for E2SAR Reassembler
*/
 
namespace e2sar
{
    const size_t RECV_BUFFER_SIZE{9000};
    /*
        The Reassembler class knows how to reassemble the events back. It relies
        on the RE header structure to reassemble the event, because the LB portion
        of LBRE header is stripped off by the load balancer. 
 
        It runs on or next to the worker performing event processing.
    */
    class Reassembler
    {
 
        friend class Segmenter;
        private:
            EjfatURI dpuri;
            LBManager lbman;
 
            // Segementer queue state - we use lockfree queue
            // and an associated atomic variable that reflects
            // how many event entries are in it.
 
            // Structure to hold each recv-queue item
            struct EventQueueItem {
                boost::chrono::steady_clock::time_point firstSegment; // when first segment arrived
                size_t numFragments; // how many fragments received (in and out of order)
                size_t bytes;  // total length
                size_t curBytes; // current bytes accumulated (could be scattered across fragments)
                EventNum_t eventNum;
                u_int8_t *event;
                u_int16_t dataId;
 
                EventQueueItem(): numFragments{0},  bytes{0}, curBytes{0},  
                    eventNum{0}, event{nullptr}, dataId{0}  {}
 
                ~EventQueueItem() {}
 
                EventQueueItem& operator=(const EventQueueItem &i) = delete;
 
                EventQueueItem(const EventQueueItem &i): firstSegment{i.firstSegment}, 
                    numFragments{i.numFragments},               
                    bytes{i.bytes}, curBytes{i.curBytes}, 
                    eventNum{i.eventNum},   event{i.event},  dataId{i.dataId} {}
                EventQueueItem(REHdr *rehdr): EventQueueItem()
                {
                    this->initFromHeader(rehdr);
                }
 
                inline void initFromHeader(REHdr *rehdr)
                {
                    bytes = rehdr->get_bufferLength();
                    dataId = rehdr->get_dataId();
                    eventNum = rehdr->get_eventNum();
                    // user deallocates this, so we don't use a pool
                    event = new u_int8_t[rehdr->get_bufferLength()];
                    // set the timestamp
                    firstSegment = boost::chrono::steady_clock::now();  
                }
            };
 
            // stats block
            struct AtomicStats {
                std::atomic<EventNum_t> enqueueLoss{0}; // number of events received and lost on enqueue
                std::atomic<EventNum_t> reassemblyLoss{0}; // number of events lost in reassembly (missing segments)
                std::atomic<EventNum_t> eventSuccess{0}; // events successfully processed
                std::atomic<size_t> totalBytesReceived{0};
                std::atomic<size_t> totalPacketsReceived{0};
                std::atomic<size_t> badHeaderDiscards{0}; // number of frames discarded due to failed header check
                // last error code
                std::atomic<int> lastErrno{0};
                // gRPC error count
                std::atomic<int> grpcErrCnt{0};
                // data socket error count
                std::atomic<int> dataErrCnt{0};
                // last e2sar error
                std::atomic<E2SARErrorc> lastE2SARError{E2SARErrorc::NoError};
                // a now unlimited queue to push lost event numbers to
                boost::lockfree::queue<boost::tuple<EventNum_t, u_int16_t, size_t>*, boost::lockfree::fixed_sized<false>> lostEventsQueue{0};
                // this array is accessed by different threads using fd as an index (so no collisions)
                std::vector<size_t> fragmentsPerFd;
                std::vector<u_int16_t> portPerFd; // which port assigned to this FD - initialized at the start
            };
            AtomicStats recvStats;
 
            // receive event queue definitions
            static const size_t QSIZE{1000};
            boost::lockfree::queue<EventQueueItem*> eventQueue{QSIZE};
            std::atomic<size_t> eventQueueDepth{0};
 
            // push event on the common event queue
            // return 1 if event is lost, 0 on success
            inline int enqueue(const std::shared_ptr<EventQueueItem> &item) noexcept
            {
                int ret = 0;
                // get rid of the shared object here
                // lockfree queue uses atomic operations and cannot use shared_ptr type
                auto newItem = new EventQueueItem(*item.get());
                if (eventQueue.push(newItem))
                    eventQueueDepth++;
                else 
                {
                    delete newItem; // the shared ptr object will be released by caller
                    ret = 1; // event lost, queue was full
                }
                // queue is lock free so we don't lock
                recvThreadCond.notify_all();
                return ret;
            }
 
            // pop event off the event queue
            inline EventQueueItem* dequeue() noexcept
            {
                EventQueueItem* item{nullptr};
                auto a = eventQueue.pop(item);
                if (a) 
                {
                    eventQueueDepth--;
                    return item;
                } else 
                    return nullptr; // queue was empty
            }
 
            // PID-related parameters
            const u_int32_t epochMs; // length of a schedule epoch - 1 sec
            const float setPoint; 
            const float Kp; // PID proportional
            const float Ki; // PID integral
            const float Kd; // PID derivative
            const float weight; // processing power factor
            const float min_factor;  
            const float max_factor;
            struct PIDSample {
                UnixTimeMicro_t sampleTime; // in usec since epoch
                float error;
                float integral;
 
                PIDSample(UnixTimeMicro_t st, float er, float intg): 
                    sampleTime{st}, error{er}, integral{intg} {}
            };
            boost::circular_buffer<PIDSample> pidSampleBuffer;
 
            // have we registered a worker
            bool registeredWorker{false};
 
 
            struct GCThreadState {
                Reassembler &reas;
                boost::thread threadObj;
 
                GCThreadState(Reassembler &r): reas{r} {}
 
                // Go through the list of threads on the recv thread list and
                // do garbage collection on their partially assembled events
                void _threadBody(); 
            };
            GCThreadState gcThreadState;
 
            struct RecvThreadState {
                // owner object
                Reassembler &reas;
                boost::thread threadObj;
 
                // timers
                struct timeval sleep_tv;
 
                // UDP sockets
                std::vector<int> udpPorts;
                std::vector<int> sockets;
                int maxFdPlusOne;
                fd_set fdSet;
 
                // object pool from which receive frames come from
                // template parameter is allocator
                //boost::pool<> recvBufferPool{RECV_BUFFER_SIZE};
                // map from <event number, data id> to event queue item (shared pointer)
                // for those items that are in assembly. Note that event
                // is uniquely identified by <event number, data id> and
                // so long as the entropy doesn't change while the event
                // segments are transmitted, they are guarangeed to go
                // to the same port
                boost::unordered_map<std::pair<EventNum_t, u_int16_t>, std::shared_ptr<EventQueueItem>, pair_hash, pair_equal> eventsInProgress;
                // mutex for guarding access to events in progress (recv thread, gc thread)
                boost::mutex evtsInProgressMutex;
                // thread local instance of events we lost
                boost::container::flat_set<std::pair<EventNum_t, u_int16_t>> lostEvents;
 
                // CPU core ids
                std::vector<int> cpuCoreList;
 
                // this constructor deliberately uses move semantics for uports
                inline RecvThreadState(Reassembler &r, std::vector<int> &&uports, 
                    const std::vector<int> &ccl): 
                    reas{r}, udpPorts{uports}, cpuCoreList{ccl}
                {
                    sleep_tv.tv_sec = 0;
                    sleep_tv.tv_usec = 10000; // 10 msec max
                }
 
                inline ~RecvThreadState()
                {
                    //recvBufferPool.purge_memory();
                }
 
                // open v4/v6 sockets
                result<int> _open();
                // close sockets
                result<int> _close();
                // thread loop
                void _threadBody();
                // process a single packet
                void _processPacket(u_int8_t* , ssize_t, bool);
 
                // log a lost event and add to lost queue for external inspection
                // boolean flag discriminates between enqueue losses (true)
                // and reassembly losses (false)
                inline void logLostEvent(std::shared_ptr<EventQueueItem> item, bool enqueLoss)
                {
                    std::pair<EventNum_t, u_int16_t> evt(item->eventNum, item->dataId);
 
                    if (lostEvents.contains(evt))
                        return;
                    // this is thread-local
                    lostEvents.insert(evt);
                    // lockfree queue (only takes trivial types)
                    boost::tuple<EventNum_t, u_int16_t, size_t> *evtPtr = 
                        new boost::tuple<EventNum_t, u_int16_t, size_t>(evt.first, evt.second, item->numFragments);
                    reas.recvStats.lostEventsQueue.push(evtPtr);
                    // this is atomic
                    if (enqueLoss)
                        reas.recvStats.enqueueLoss++;
                    else
                        reas.recvStats.reassemblyLoss++;                    
                }
            };
            friend struct RecvThreadState;
            std::list<RecvThreadState> recvThreadState;
 
            // receive related parameters
            const std::vector<int> cpuCoreList;
            ip::address dataIP; 
            const u_int16_t dataPort; 
            const int portRange; // translates into 2^portRange - 1 ports we listen to
            const size_t numRecvThreads;
            const size_t numRecvPorts;
            std::vector<std::list<int>> threadsToPorts;
            const bool withLBHeader;
            const int eventTimeout_ms; // how long we allow events to linger 'in progress' before we give up
            const int recvWaitTimeout_ms{10}; // how long we wait on condition variable before we come up for air
            // recv socket buffer size for setsockop
            const int rcvSocketBufSize;
 
            // lock with recv thread
            boost::mutex recvThreadMtx;
            // condition variable for recv thread queue
            boost::condition_variable recvThreadCond;
            // lock for the mutex
            //thread_local boost::unique_lock<boost::mutex> condLock(recvThreadMtx, boost::defer_lock);
 
            inline void assignPortsToThreads()
            {
                // O(numRecvPorts)
                for(size_t i=0; i<numRecvPorts;)
                {
                    for(size_t j=0; i<numRecvPorts && j<numRecvThreads; i++, j++)
                    {
                        threadsToPorts[j].push_back(dataPort + i);
                    } 
                }
            }
 
            struct SendStateThreadState {
                // owner object
                Reassembler &reas;
                boost::thread threadObj;
 
                const u_int16_t period_ms;
 
                // UDP sockets
                int socketFd{0};
 
                inline SendStateThreadState(Reassembler &r, u_int16_t period_ms): 
                    reas{r}, period_ms{period_ms}
                {}
 
                // thread loop. all important behavior is encapsulated inside LBManager
                void _threadBody();
            };
            friend struct sendStateThreadState;
            SendStateThreadState sendStateThreadState;
            bool useCP; // for debugging we may not want to have CP running
            bool reportStats; // report worker stats in sendState thread (usually false)
            unsigned int rcvIovecSize; // size of iovec vector when recvmmsg is used
            // global thread stop signal
            bool threadsStop{false};
 
            inline void sanityChecks()
            {
                if (numRecvThreads > 128)
                    throw E2SARException("Too many reassembly threads requested, limit 128");
 
                if (numRecvPorts > (2 << 13))
                    throw E2SARException("Too many receive ports reqiuested, limit 2^14");
 
                if (eventTimeout_ms > 10000)
                    throw E2SARException("Event timeout exception unreasonably long, limit 10s");
 
                if (dataPort < 1024)
                    throw E2SARException("Base receive port in the privileged range (<1024)");
 
                if (portRange > 14)
                    throw E2SARException("Port range out of bounds: [0, 14]");
 
                if (!dpuri.has_dataAddr())
                    throw E2SARException("Data address not present in the URI");
            }
        public:
            struct ReportedStats {
                EventNum_t enqueueLoss;  // number of events received and lost on enqueue
                EventNum_t reassemblyLoss; // number of events lost in reassembly due to missing segments
                EventNum_t eventSuccess; // events successfully processed
                int lastErrno; 
                int grpcErrCnt; 
                int dataErrCnt; 
                E2SARErrorc lastE2SARError;
                size_t totalPackets, totalBytes, badHeaderDiscards;
 
                ReportedStats() = delete;
                ReportedStats(const AtomicStats &as): enqueueLoss{as.enqueueLoss}, 
                    reassemblyLoss{as.reassemblyLoss}, eventSuccess{as.eventSuccess},
                    lastErrno{as.lastErrno}, grpcErrCnt{as.grpcErrCnt}, dataErrCnt{as.dataErrCnt},
                    lastE2SARError{as.lastE2SARError}, totalPackets{as.totalPacketsReceived}, 
                    totalBytes{as.totalBytesReceived}, badHeaderDiscards{as.badHeaderDiscards}
                    {}
            };
 
            struct ReassemblerFlags 
            {
                bool useCP;
                bool useHostAddress;
                u_int16_t period_ms;
                bool validateCert;
                float Ki, Kp, Kd, setPoint;
                u_int32_t epoch_ms;
                int portRange; 
                int eventTimeout_ms;
                int rcvSocketBufSize;
                float weight, min_factor, max_factor;
                bool reportStats;
                unsigned int rcvIovecSize;
                ReassemblerFlags(): useCP{true}, useHostAddress{false},
                    period_ms{100}, validateCert{true}, Ki{0.}, Kp{0.}, Kd{0.}, setPoint{0.},
                    epoch_ms{1000}, portRange{-1}, eventTimeout_ms{500},
                    rcvSocketBufSize{1024*1024*3}, weight{1.0}, min_factor{0.5}, max_factor{2.0},
                    reportStats{true}, rcvIovecSize{100} {}
                static result<Reassembler::ReassemblerFlags> getFromINI(const std::string &iniFile) noexcept;
            };
            Reassembler(const EjfatURI &uri, ip::address data_ip, u_int16_t starting_port,
                        std::vector<int> cpuCoreList, 
                        const ReassemblerFlags &rflags = ReassemblerFlags());
            Reassembler(const EjfatURI &uri, ip::address data_ip, u_int16_t starting_port,
                        size_t numRecvThreads = 1, const ReassemblerFlags &rflags = ReassemblerFlags());
 
            Reassembler(const EjfatURI &uri, u_int16_t starting_port,
                        std::vector<int> cpuCoreList, 
                        const ReassemblerFlags &rflags = ReassemblerFlags(),
                        bool v6 = false);
            Reassembler(const EjfatURI &uri, u_int16_t starting_port,
                        size_t numRecvThreads = 1, 
                        const ReassemblerFlags &rflags = ReassemblerFlags(),
                        bool v6 = false);
 
            Reassembler(const Reassembler &r) = delete;
            Reassembler & operator=(const Reassembler &o) = delete;
            ~Reassembler()
            {
                if (useCP && registeredWorker)
                    auto res = lbman.deregisterWorker();
 
                stopThreads();
 
                // pool memory is implicitly freed when pool goes out of scope
            }
            
            result<int> registerWorker(const std::string &node_name) noexcept;
 
            result<int> deregisterWorker() noexcept;
 
            result<int> openAndStart() noexcept;
 
            result<int> getEvent(uint8_t **event, size_t *bytes, EventNum_t* eventNum, uint16_t *dataId) noexcept;
 
            result<int> recvEvent(uint8_t **event, size_t *bytes, EventNum_t* eventNum, uint16_t *dataId, u_int64_t wait_ms=0) noexcept;
 
            inline const ReportedStats getStats() const noexcept
            {
                return ReportedStats(recvStats);
            }
 
            inline result<boost::tuple<EventNum_t, u_int16_t, size_t>> get_LostEvent() noexcept
            {
                boost::tuple<EventNum_t, u_int16_t, size_t> *res = nullptr;
                if (recvStats.lostEventsQueue.pop(res))
                {
                    auto ret{*res};
                    delete res;
                    return ret;
                }
                else
                    return E2SARErrorInfo{E2SARErrorc::NotFound, "Lost event queue is empty"};
            }
 
            inline result<std::list<std::pair<u_int16_t, size_t>>> get_FDStats() noexcept
            {
                if (not threadsStop)
                    return E2SARErrorInfo{E2SARErrorc::LogicError, "This method should only be called after the threads have been stopped."};
 
                int i{0};
                std::list<std::pair<u_int16_t, size_t>> ret;
                for(auto count: recvStats.fragmentsPerFd)
                {
                    if (recvStats.portPerFd[i] != 0)
                        ret.push_back(std::make_pair<>(recvStats.portPerFd[i], count));
                    ++i;
                }
                return ret;
            }
 
            inline size_t get_numRecvThreads() const noexcept
            {
                return numRecvThreads;
            }
 
            inline const std::pair<int, int> get_recvPorts() const noexcept
            {
                return std::make_pair(dataPort, dataPort + numRecvPorts - 1);
            }
 
            inline int get_portRange() const noexcept
            {
                return portRange;
            }
 
            inline const ip::address get_dataIP() const noexcept
            {
                return dataIP;
            }
            void stopThreads() 
            {
                if (not threadsStop)
                {
                    threadsStop = true;
 
                    recvThreadCond.notify_all();
 
                    // wait to exit
                    if (useCP)
                        sendStateThreadState.threadObj.join();
 
                    for(auto i = recvThreadState.begin(); i != recvThreadState.end(); ++i)
                        i->threadObj.join();
 
                    // drain event queue
                    EventQueueItem* item{nullptr};
                    bool a{false};
                    do {
                        a = eventQueue.pop(item);
                        if (a) 
                        {
                            if (item->event != nullptr)
                                delete[] item->event;
                            delete item;
                        }
                    } while (a);
 
                    gcThreadState.threadObj.join();
 
                    // drain lost events queue - tuples are heap-allocated in logLostEvent
                    // and only freed by get_LostEvent(); if the caller never calls it they leak
                    boost::tuple<EventNum_t, u_int16_t, size_t>* evtPtr{nullptr};
                    while (recvStats.lostEventsQueue.pop(evtPtr))
                        delete evtPtr;
                }
            }
        protected:
        private:
 
 
 
    };
}
#endif