bool mainloop

(

READ_STRUCT *

rd

)

Definition at line 1085 of file EventBuilder.cc.

References READ_STRUCT::activeSockets, actrun, Memory::allocated, READ_STRUCT::B, READ_STRUCT::bufLen, READ_STRUCT::bufPos, READ_STRUCT::bufTyp, READ_STRUCT::connected, copyData(), debugHead(), end, factPrintf(), factReportIncomplete(), factStat(), READ_STRUCT::fd_epoll, g_evtTimeout, g_maxMem, g_reset, READ_STRUCT::get(), gi_NumConnect, gj, READ_STRUCT::H, i, Memory::inuse, READ_STRUCT::kData, MessageImp::kError, kFileOpen, READ_STRUCT::kHeader, MessageImp::kInfo, kRequestConnectionChange, kRequestNone, kRequestTimeout, READ_STRUCT::len(), Memory::max_inuse, MAX_TOT_MEM, mBufEvt(), NBOARDS, Queue< T, List >::post(), READ_STRUCT::read(), READ_STRUCT::relBytes, reportIncomplete(), runFinished(), Queue< T, List >::size(), READ_STRUCT::skip, READ_STRUCT::socket, READ_STRUCT::sockId, Queue< T, List >::start(), READ_STRUCT::swapData(), READ_STRUCT::swapHeader(), time, READ_STRUCT::totBytes, Queue< T, List >::wait(), and READ_STRUCT::wait().

Referenced by StartEvtBuild().

{
    factPrintf(MessageImp::kInfo, "Starting EventBuilder main loop");

    primaryQueue.start();
    secondaryQueue.start();
    processingQueue1.start();;

    actrun = make_shared<RUN_CTRL2>();

    //time in seconds
    time_t gi_SecTime = time(NULL)-1;

    //loop until global variable g_runStat claims stop
    g_reset = 0;
    while (g_reset == 0)
    {
#ifdef USE_POLL
        int    pp[40];
        int    nn = 0;
        pollfd fds[40];
        for (int i=0; i<40; i++)
        {
            if (rd[i].socket>=0 && rd[i].connected && rd[i].bufLen>0)
            {
                fds[nn].fd = rd[i].socket;
                fds[nn].events = POLLIN;
                pp[nn] = i;
                nn++;
            }
        }

        const int rc_epoll = poll(fds, nn, 100);
        if (rc_epoll<0)
            break;
#endif

#ifdef USE_SELECT
        fd_set readfs;
        FD_ZERO(&readfs);
        int nfsd = 0;
        for (int i=0; i<NBOARDS; i++)
            if (rd[i].socket>=0 && rd[i].connected && rd[i].bufLen>0)
            {
                FD_SET(rd[i].socket, &readfs);
                if (rd[i].socket>nfsd)
                    nfsd = rd[i].socket;
            }

        timeval tv;
        tv.tv_sec = 0;
        tv.tv_usec = 100000;
        const int rc_select = select(nfsd+1, &readfs, NULL, NULL, &tv);
        // 0: timeout
        // -1: error
        if (rc_select<0)
        {
            factPrintf(MessageImp::kError, "Waiting for data failed: %d (select,rc=%d)", errno);
            continue;
        }
#endif

#ifdef USE_EPOLL
        const int rc_epoll = READ_STRUCT::wait();
        if (rc_epoll<0)
            break;
#endif

#ifdef PRIORITY_QUEUE
        priority_queue<READ_STRUCT*, vector<READ_STRUCT*>, READ_STRUCTcomp> prio;

        for (int i=0; i<NBOARDS; i++)
            if (rd[i].connected)
                prio.push(rd+i);

        if (!prio.empty()) do
#endif


#ifdef USE_POLL
        for (int jj=0; jj<nn; jj++)
#endif
#ifdef USE_EPOLL
        for (int jj=0; jj<rc_epoll; jj++)
#endif
#if !defined(USE_EPOLL) && !defined(USE_POLL) && !defined(PRIORITY_QUEUE)
        for (int jj=0; jj<NBOARDS; jj++)
#endif
        {
#ifdef PRIORITY_QUEUE
            READ_STRUCT *rs = prio.top();
#endif
#ifdef USE_SELECT
            if (!FD_ISSET(rs->socket, &readfs))
                continue;
#endif

#ifdef USE_POLL
            if ((fds[jj].revents&POLLIN)==0)
                continue;
#endif

#ifdef USE_EPOLL
            // FIXME: How to get i?
            READ_STRUCT *rs = READ_STRUCT::get(jj);
#endif

#ifdef USE_POLL
            // FIXME: How to get i?
            READ_STRUCT *rs = &rd[pp[jj]];
#endif

#if !defined(USE_POLL) && !defined(USE_EPOLL) && !defined(PRIORITY_QUEUE)
            const int i = (jj%4)*10 + (jj/4);
            READ_STRUCT *rs = &rd[i];
#endif

#ifdef COMPLETE_EVENTS
            if (rs->bufTyp==READ_STRUCT::kWait)
                continue;
#endif

            // ==================================================================

            const bool rc_read = rs->read();

            // Connect might have gotten closed during read
            gi_NumConnect[rs->sockId] = rs->connected;
            gj.numConn[rs->sockId]    = rs->connected;

            // Read either failed or disconnected, or the buffer is not yet full
            if (!rc_read)
                continue;

            // ==================================================================

            if (rs->bufTyp==READ_STRUCT::kHeader)
            {
                //check if startflag correct; else shift block ....
                // FIXME: This is not enough... this combination of
                //        bytes can be anywhere... at least the end bytes
                //        must be checked somewhere, too.
                uint k;
                for (k=0; k<sizeof(PEVNT_HEADER)-1; k++)
                {
                    if (rs->B[k]==0xfb && rs->B[k+1] == 0x01)
                        break;
                }
                rs->skip += k;

                //no start of header found
                if (k==sizeof(PEVNT_HEADER)-1)
                {
                    rs->B[0]   = rs->B[sizeof(PEVNT_HEADER)-1];
                    rs->bufPos = rs->B+1;
                    rs->bufLen = sizeof(PEVNT_HEADER)-1;
                    continue;
                }

                if (k > 0)
                {
                    memmove(rs->B, rs->B+k, sizeof(PEVNT_HEADER)-k);

                    rs->bufPos -= k;
                    rs->bufLen += k;

                    continue; // We need to read more (bufLen>0)
                }

                if (rs->skip>0)
                {
                    factPrintf(MessageImp::kInfo, "Skipped %d bytes on port %d", rs->skip, rs->sockId);
                    rs->skip = 0;
                }

                // Swap the header entries from network to host order
                rs->swapHeader();

                rs->bufTyp = READ_STRUCT::kData;
                rs->bufLen = rs->len() - sizeof(PEVNT_HEADER);

                debugHead(rs->B);  // i and fadBoard not used

                continue;
            }

            const uint16_t &end = *reinterpret_cast<uint16_t*>(rs->bufPos-2);
            if (end != 0xfe04)
            {
                factPrintf(MessageImp::kError, "End-of-event flag wrong on socket %2d for event %d (len=%d), got %04x",
                           rs->sockId, rs->H.fad_evt_counter, rs->len(), end);

                // ready to read next header
                rs->bufTyp = READ_STRUCT::kHeader;
                rs->bufLen = sizeof(PEVNT_HEADER);
                rs->bufPos = rs->B;
                // FIXME: What to do with the validity flag?
                continue;
            }

            // get index into mBuffer for this event (create if needed)
            const shared_ptr<EVT_CTRL2> evt = mBufEvt(*rs, actrun);

            // We have a valid entry, but no memory has yet been allocated
            if (evt && !evt->initMemory())
            {
                const time_t tm = time(NULL);
                if (evt->runCtrl->reportMem==tm)
                    continue;

                factPrintf(MessageImp::kError, "No free memory left for %d (run=%d)", evt->evNum, evt->runNum);
                evt->runCtrl->reportMem = tm;
                continue;
            }

            // ready to read next header
            rs->bufTyp = READ_STRUCT::kHeader;
            rs->bufLen = sizeof(PEVNT_HEADER);
            rs->bufPos = rs->B;

            // Fatal error occured. Event cannot be processed. Skip it. Start reading next header.
            if (!evt)
                continue;

            // This should never happen
            if (evt->board[rs->sockId] != -1)
            {
                factPrintf(MessageImp::kError, "Got event %5d from board %3d (i=%3d, len=%5d) twice.",
                           evt->evNum, rs->sockId, jj, rs->len());
                // FIXME: What to do with the validity flag?
                continue; // Continue reading next header
            }

            // Swap the data entries (board headers) from network to host order
            rs->swapData();

            // Copy data from rd[i] to mBuffer[evID]
            copyData(*rs, evt.get());

#ifdef COMPLETE_EVENTS
            // Do not read anmymore from this board until the whole event has been received
            rs->bufTyp = READ_STRUCT::kWait;
#endif
            // now we have stored a new board contents into Event structure
            evt->board[rs->sockId] = rs->sockId;
            evt->header = evt->FADhead+rs->sockId;
            evt->nBoard++;

#ifdef COMPLETE_EPOLL
            if (epoll_ctl(READ_STRUCT::fd_epoll, EPOLL_CTL_DEL, rs->socket, NULL)<0)
            {
                factPrintf(MessageImp::kError, "epoll_ctrl failed: %m (EPOLL_CTL_DEL,rc=%d)", errno);
                break;
            }
#endif
            // event not yet complete
            if (evt->nBoard < READ_STRUCT::activeSockets)
                continue;

            // All previous events are now flagged as incomplete ("expired")
            // and will be removed. (This is a bit tricky, because pop_front()
            // would invalidate the current iterator if not done _after_ the increment)
            for (auto it=evtCtrl.begin(); it!=evtCtrl.end(); )
            {
                const bool found = it->get()==evt.get();
                if (!found)
                    reportIncomplete(*it, "expired");
                else
                    primaryQueue.post(evt);

                // package_len is 0 if nothing was received.
                for (int ib=0; ib<40; ib++)
                    rd[ib].relBytes += uint32_t((*it)->FADhead[ib].package_length)*2;

                // The counter must be increased _before_ the pop_front,
                // otherwise the counter is invalidated by the pop_front!
                it++;
                evtCtrl.pop_front();

                // We reached the current event, so we are done
                if (found)
                    break;
            }

#ifdef COMPLETE_EPOLL
            for (int j=0; j<40; j++)
            {
                epoll_event ev;
                ev.events = EPOLLIN;
                ev.data.ptr = &rd[j];  // user data (union: ev.ptr)
                if (epoll_ctl(READ_STRUCT::fd_epoll, EPOLL_CTL_ADD, rd[j].socket, &ev)<0)
                {
                    factPrintf(MessageImp::kError, "epoll_ctl failed: %m (EPOLL_CTL_ADD,rc=%d)", errno);
                    return;
                }
            }
#endif

#ifdef COMPLETE_EVENTS
            for (int j=0; j<40; j++)
            {
                //if (rs->bufTyp==READ_STRUCT::kWait)
                {
                    rs->bufTyp = READ_STRUCT::kHeader;
                    rs->bufLen = sizeof(PEVNT_HEADER);
                    rs->bufPos = rs->B;
                }
            }
#endif
        } // end for loop over all sockets
#ifdef PRIORITY_QUEUE
        while (0); // convert continue into break ;)
#endif

        // ==================================================================

        const time_t actTime = time(NULL);
        if (actTime == gi_SecTime)
        {
#if !defined(USE_SELECT) && !defined(USE_EPOLL) && !defined(USE_POLL)
            if (evtCtrl.empty())
                usleep(actTime-actrun->lastTime>300 ? 10000 : 1);
#endif
            continue;
        }
        gi_SecTime = actTime;

        // ==================================================================
        //loop over all active events and flag those older than read-timeout
        //delete those that are written to disk ....

        // This could be improved having the pointer which separates the queue with
        // the incomplete events from the queue with the complete events
        for (auto it=evtCtrl.begin(); it!=evtCtrl.end(); )
        {
            // A reference is enough because the shared_ptr is hold by the evtCtrl
            const shared_ptr<EVT_CTRL2> &evt = *it;

            // The first event is the oldest. If the first event within the
            // timeout window was received, we can stop searching further.
            if (evt->time.tv_sec+g_evtTimeout>=actTime)
                break;

            // The counter must be increased _before_ the pop_front,
            // otherwise the counter is invalidated by the pop_front!
            it++;

            // This timeout is caused because complete data from one or more
            // boards has been received, but the memory could not be allocated.
            // There is no reason why we should not go on waiting for
            // memory to become free. However, the FADs will disconnect
            // after 60s due to their keep-alive timeout, but the event builder
            // will still wait for memory to become available.
            // Currently, the only possibility to free the memory from the
            // evtCtrl to restart the event builder (STOP/START).
            if (!evt->valid())
                continue;

            // This will result in the emission of a dim service.
            // It doesn't matter if that takes comparably long,
            // because we have to stop the run anyway.
            const uint64_t rep = reportIncomplete(evt, "timeout");
            factReportIncomplete(rep);

            // At least the data from one boards is complete...
            // package_len is 0 when nothing was received from this board
            for (int ib=0; ib<40; ib++)
                rd[ib].relBytes += uint32_t(evt->FADhead[ib].package_length)*2;

            evtCtrl.pop_front();
        }

        // =================================================================

        gj.bufNew   = evtCtrl.size();            //# incomplete events in buffer
        gj.bufEvt   = primaryQueue.size();       //# complete events in buffer
        gj.bufWrite = secondaryQueue.size();     //# complete events in buffer
        gj.bufProc  = processingQueue1.size();   //# complete events in buffer
        gj.bufTot   = Memory::max_inuse/MAX_TOT_MEM;
        gj.usdMem   = Memory::max_inuse;
        gj.totMem   = Memory::allocated;
        gj.maxMem   = g_maxMem;

        gj.deltaT = 1000; // temporary, must be improved

        bool changed = false;

        static vector<uint64_t> store(NBOARDS);

        for (int ib=0; ib<NBOARDS; ib++)
        {
            gj.rateBytes[ib] = store[ib]>rd[ib].totBytes ? rd[ib].totBytes : rd[ib].totBytes-store[ib];
            gj.relBytes[ib]  = rd[ib].totBytes-rd[ib].relBytes;

            store[ib] = rd[ib].totBytes;

            if (rd[ib].check(g_port[ib].sockDef, g_port[ib].sockAddr))
                changed = true;

            gi_NumConnect[ib] = rd[ib].connected;
            gj.numConn[ib]    = rd[ib].connected;
        }

        factStat(gj);

        Memory::max_inuse = 0;

        // =================================================================

        // This is a fake event to trigger possible run-closing conditions once a second
        // FIXME: This is not yet ideal because a file would never be closed
        //        if a new file has been started and no events of the new file
        //        have been received yet
        int request = kRequestNone;

        // If nothing was received for more than 5min, close file
        if (actTime-actrun->lastTime>300)
            request |= kRequestTimeout;

        // If connection status has changed
        if (changed)
            request |= kRequestConnectionChange;

        if (request!=kRequestNone)
            runFinished();

        if (actrun->fileStat==kFileOpen)
            primaryQueue.emplace(new EVT_CTRL2(request, actrun));
    }

    //   1: Stop, wait for event to get processed
    //   2: Stop, finish immediately
    // 101: Restart, wait for events to get processed
    // 101: Restart, finish immediately
    //
    const int gi_reset = g_reset;

    const bool abort = gi_reset%100==2;

    factPrintf(MessageImp::kInfo, "Stop reading ... RESET=%d (%s threads)", gi_reset, abort?"abort":"join");

    primaryQueue.wait(abort);
    secondaryQueue.wait(abort);
    processingQueue1.wait(abort);

    // Here we also destroy all runCtrl structures and hence close all open files
    evtCtrl.clear();
    actrun.reset();

    factPrintf(MessageImp::kInfo, "Exit read Process...");
    factPrintf(MessageImp::kInfo, "%llu Bytes flagged as in-use.", Memory::inuse);

    factStat(gj);

    return gi_reset>=100;
}

Here is the call graph for this function:

Here is the caller graph for this function: