source: trunk/FACT++/src/EventBuilder.cc@ 19026

Last change on this file since 19026 was 18987, checked in by tbretz, 7 years ago
Let the compiler know that the second argument is a format string
File size: 47.4 KB
Line 
1#include <poll.h>
2#include <sys/time.h>
3#include <sys/epoll.h>
4#include <netinet/tcp.h>
5
6#include <cstring>
7#include <cstdarg>
8#include <list>
9#include <queue>
10
11#include <boost/algorithm/string/join.hpp>
12
13#include "../externals/Queue.h"
14
15#include "MessageImp.h"
16#include "EventBuilder.h"
17#include "HeadersFAD.h"
18
19using namespace std;
20
21#define MIN_LEN 32 // min #bytes needed to interpret FADheader
22#define MAX_LEN 81920 // one max evt = 1024*2*36 + 8*36 + 72 + 4 = 74092 (data+boardheader+eventheader+endflag)
23
24//#define COMPLETE_EVENTS
25//#define USE_POLL
26//#define USE_EPOLL
27//#define USE_SELECT
28//#define COMPLETE_EPOLL
29//#define PRIORITY_QUEUE
30
31// Reading only 1024: 13: 77Hz, 87%
32// Reading only 1024: 12: 78Hz, 46%
33// Reading only 300: 4: 250Hz, 92%
34// Reading only 300: 3: 258Hz, 40%
35
36// Reading only four threads 1024: 13: 77Hz, 60%
37// Reading only four threads 1024: 12: 78Hz, 46%
38// Reading only four threads 300: 4: 250Hz, 92%
39// Reading only four threads 300: 3: 258Hz, 40%
40
41// Default 300: 4: 249Hz, 92%
42// Default 300: 3: 261Hz, 40%
43// Default 1024: 13: 76Hz, 93%
44// Default 1024: 12: 79Hz, 46%
45
46// Poll [selected] 1024: 13: 63Hz, 45%
47// Poll [selected] 1024: 14: 63Hz, 63%
48// Poll [selected] 1024: 15: 64Hz, 80%
49// Poll [selected] 300: 4: 230Hz, 47%
50// Poll [selected] 300: 3: 200Hz, 94%
51
52// Poll [all] 1024: 13: 65Hz, 47%
53// Poll [all] 1024: 14: 64Hz, 59%
54// Poll [all] 1024: 15: 62Hz, 67%
55// Poll [all] 300: 4: 230Hz, 47%
56// Poll [all] 300: 3: 230Hz, 35%
57
58// ==========================================================================
59
60bool runOpen(const EVT_CTRL2 &evt);
61bool runWrite(const EVT_CTRL2 &evt);
62void runClose(const EVT_CTRL2 &run);
63void applyCalib(const EVT_CTRL2 &evt, const size_t &size);
64void factOut(int severity, const char *message);
65void factReportIncomplete (uint64_t rep);
66void gotNewRun(RUN_CTRL2 &run);
67void runFinished();
68void factStat(const GUI_STAT &gj);
69bool eventCheck(const EVT_CTRL2 &evt);
70void debugHead(void *buf);
71
72// ==========================================================================
73
74int g_reset;
75
76size_t g_maxMem; //maximum memory allowed for buffer
77
78uint16_t g_evtTimeout; // timeout (sec) for one event
79
80FACT_SOCK g_port[NBOARDS]; // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd"
81
82uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards
83
84GUI_STAT gj;
85
86// ==========================================================================
87
88namespace Memory
89{
90 uint64_t inuse = 0;
91 uint64_t allocated = 0;
92
93 uint64_t max_inuse = 0;
94
95 std::mutex mtx;
96
97 std::forward_list<void*> memory;
98
99 void *malloc()
100 {
101 // No free slot available, next alloc would exceed max memory
102 if (memory.empty() && allocated+MAX_TOT_MEM>g_maxMem)
103 return NULL;
104
105 // We will return this amount of memory
106 // This is not 100% thread safe, but it is not a super accurate measure anyway
107 inuse += MAX_TOT_MEM;
108 if (inuse>max_inuse)
109 max_inuse = inuse;
110
111 if (memory.empty())
112 {
113 // No free slot available, allocate a new one
114 allocated += MAX_TOT_MEM;
115 return new char[MAX_TOT_MEM];
116 }
117
118 // Get the next free slot from the stack and return it
119 const std::lock_guard<std::mutex> lock(mtx);
120
121 void *mem = memory.front();
122 memory.pop_front();
123 return mem;
124 };
125
126 void free(void *mem)
127 {
128 if (!mem)
129 return;
130
131 // Decrease the amont of memory in use accordingly
132 inuse -= MAX_TOT_MEM;
133
134 // If the maximum memory has changed, we might be over the limit.
135 // In this case: free a slot
136 if (allocated>g_maxMem)
137 {
138 delete [] (char*)mem;
139 allocated -= MAX_TOT_MEM;
140 return;
141 }
142
143 const std::lock_guard<std::mutex> lock(mtx);
144 memory.push_front(mem);
145 }
146
147};
148
149// ==========================================================================
150
151__attribute__((__format__ (__printf__, 2, 0)))
152void factPrintf(int severity, const char *fmt, ...)
153{
154 char str[1000];
155
156 va_list ap;
157 va_start(ap, fmt);
158 vsnprintf(str, 1000, fmt, ap);
159 va_end(ap);
160
161 factOut(severity, str);
162}
163
164// ==========================================================================
165
166struct READ_STRUCT
167{
168 enum buftyp_t
169 {
170 kStream,
171 kHeader,
172 kData,
173#ifdef COMPLETE_EVENTS
174 kWait
175#endif
176 };
177
178 // ---------- connection ----------
179
180 static uint activeSockets;
181
182 int sockId; // socket id (board number)
183 int socket; // socket handle
184 bool connected; // is this socket connected?
185
186 struct sockaddr_in SockAddr; // Socket address copied from wrapper during socket creation
187
188 // ------------ epoll -------------
189
190 static int fd_epoll;
191 static epoll_event events[NBOARDS];
192
193 static void init();
194 static void close();
195 static int wait();
196 static READ_STRUCT *get(int i) { return reinterpret_cast<READ_STRUCT*>(events[i].data.ptr); }
197
198 // ------------ buffer ------------
199
200 buftyp_t bufTyp; // what are we reading at the moment: 0=header 1=data -1=skip ...
201
202 uint32_t bufLen; // number of bytes left to read
203 uint8_t *bufPos; // next byte to read to the buffer next
204
205 union
206 {
207 uint8_t B[MAX_LEN];
208 uint16_t S[MAX_LEN / 2];
209 uint32_t I[MAX_LEN / 4];
210 uint64_t L[MAX_LEN / 8];
211 PEVNT_HEADER H;
212 };
213
214 timeval time;
215 uint64_t totBytes; // total received bytes
216 uint64_t relBytes; // total released bytes
217 uint32_t skip; // number of bytes skipped before start of event
218
219 uint32_t len() const { return uint32_t(H.package_length)*2; }
220
221 void swapHeader();
222 void swapData();
223
224 // --------------------------------
225
226 READ_STRUCT() : socket(-1), connected(false), totBytes(0), relBytes(0)
227 {
228 if (fd_epoll<0)
229 init();
230 }
231 ~READ_STRUCT()
232 {
233 destroy();
234 }
235
236 void destroy();
237 bool create(sockaddr_in addr);
238 bool check(int, sockaddr_in addr);
239 bool read();
240
241};
242
243#ifdef PRIORITY_QUEUE
244struct READ_STRUCTcomp
245{
246 bool operator()(const READ_STRUCT *r1, const READ_STRUCT *r2)
247 {
248 const int64_t rel1 = r1->totBytes - r1->relBytes;
249 const int64_t rel2 = r2->totBytes - r2->relBytes;
250 return rel1 > rel2;
251 }
252};
253#endif
254
255int READ_STRUCT::wait()
256{
257 // wait for something to do...
258 const int rc = epoll_wait(fd_epoll, events, NBOARDS, 100); // max, timeout[ms]
259 if (rc>=0)
260 return rc;
261
262 if (errno==EINTR) // timout or signal interruption
263 return 0;
264
265 factPrintf(MessageImp::kError, "epoll_wait failed: %m (rc=%d)", errno);
266 return -1;
267}
268
269uint READ_STRUCT::activeSockets = 0;
270int READ_STRUCT::fd_epoll = -1;
271epoll_event READ_STRUCT::events[NBOARDS];
272
273void READ_STRUCT::init()
274{
275 if (fd_epoll>=0)
276 return;
277
278#ifdef USE_EPOLL
279 fd_epoll = epoll_create(NBOARDS);
280 if (fd_epoll<0)
281 {
282 factPrintf(MessageImp::kError, "Waiting for data failed: %d (epoll_create,rc=%d)", errno);
283 return;
284 }
285#endif
286}
287
288void READ_STRUCT::close()
289{
290#ifdef USE_EPOLL
291 if (fd_epoll>=0 && ::close(fd_epoll)>0)
292 factPrintf(MessageImp::kFatal, "Closing epoll failed: %m (close,rc=%d)", errno);
293#endif
294
295 fd_epoll = -1;
296}
297
298bool READ_STRUCT::create(sockaddr_in sockAddr)
299{
300 if (socket>=0)
301 return false;
302
303 const int port = ntohs(sockAddr.sin_port) + 1;
304
305 SockAddr.sin_family = sockAddr.sin_family;
306 SockAddr.sin_addr = sockAddr.sin_addr;
307 SockAddr.sin_port = htons(port);
308
309 if ((socket = ::socket(PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0)
310 {
311 factPrintf(MessageImp::kFatal, "Generating socket %d failed: %m (socket,rc=%d)", sockId, errno);
312 socket = -1;
313 return false;
314 }
315
316 int optval = 1;
317 if (setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, &optval, sizeof(int)) < 0)
318 factPrintf(MessageImp::kInfo, "Setting TCP_NODELAY for socket %d failed: %m (setsockopt,rc=%d)", sockId, errno);
319
320 optval = 1;
321 if (setsockopt (socket, SOL_SOCKET, SO_KEEPALIVE, &optval, sizeof(int)) < 0)
322 factPrintf(MessageImp::kInfo, "Setting SO_KEEPALIVE for socket %d failed: %m (setsockopt,rc=%d)", sockId, errno);
323
324 optval = 10; //start after 10 seconds
325 if (setsockopt (socket, SOL_TCP, TCP_KEEPIDLE, &optval, sizeof(int)) < 0)
326 factPrintf(MessageImp::kInfo, "Setting TCP_KEEPIDLE for socket %d failed: %m (setsockopt,rc=%d)", sockId, errno);
327
328 optval = 10; //do every 10 seconds
329 if (setsockopt (socket, SOL_TCP, TCP_KEEPINTVL, &optval, sizeof(int)) < 0)
330 factPrintf(MessageImp::kInfo, "Setting TCP_KEEPINTVL for socket %d failed: %m (setsockopt,rc=%d)", sockId, errno);
331
332 optval = 2; //close after 2 unsuccessful tries
333 if (setsockopt (socket, SOL_TCP, TCP_KEEPCNT, &optval, sizeof(int)) < 0)
334 factPrintf(MessageImp::kInfo, "Setting TCP_KEEPCNT for socket %d failed: %m (setsockopt,rc=%d)", sockId, errno);
335
336 factPrintf(MessageImp::kInfo, "Generated socket %d (%d)", sockId, socket);
337
338 //connected = false;
339 activeSockets++;
340
341 return true;
342}
343
344void READ_STRUCT::destroy()
345{
346 if (socket<0)
347 return;
348
349#ifdef USE_EPOLL
350 // strictly speaking this should not be necessary
351 if (fd_epoll>=0 && connected && epoll_ctl(fd_epoll, EPOLL_CTL_DEL, socket, NULL)<0)
352 factPrintf(MessageImp::kError, "epoll_ctrl failed: %m (EPOLL_CTL_DEL,rc=%d)", errno);
353#endif
354
355 if (::close(socket) > 0)
356 factPrintf(MessageImp::kFatal, "Closing socket %d failed: %m (close,rc=%d)", sockId, errno);
357 else
358 factPrintf(MessageImp::kInfo, "Closed socket %d (%d)", sockId, socket);
359
360 // Set the socket to "not connected"
361 socket = -1;
362 connected = false;
363 activeSockets--;
364 bufLen = 0;
365}
366
367bool READ_STRUCT::check(int sockDef, sockaddr_in addr)
368{
369 // Continue in the most most likely case (performance)
370 //if (socket>=0 && sockDef!=0 && connected)
371 // return;
372 const int old = socket;
373
374 // socket open, but should not be open
375 if (socket>=0 && sockDef==0)
376 destroy();
377
378 // Socket closed, but should be open
379 if (socket<0 && sockDef!=0)
380 create(addr); //generate address and socket
381
382 const bool retval = old!=socket;
383
384 // Socket closed
385 if (socket<0)
386 return retval;
387
388 // Socket open and connected: Nothing to do
389 if (connected)
390 return retval;
391
392 //try to connect if not yet done
393 const int rc = connect(socket, (struct sockaddr *) &SockAddr, sizeof(SockAddr));
394 if (rc == -1)
395 return retval;
396
397 connected = true;
398
399 if (sockDef<0)
400 {
401 bufTyp = READ_STRUCT::kStream; // full data to be skipped
402 bufLen = MAX_LEN; // huge for skipping
403 }
404 else
405 {
406 bufTyp = READ_STRUCT::kHeader; // expect a header
407 bufLen = sizeof(PEVNT_HEADER); // max size to read at begining
408 }
409
410 bufPos = B; // no byte read so far
411 skip = 0; // start empty
412 totBytes = 0;
413 relBytes = 0;
414
415 factPrintf(MessageImp::kInfo, "Connected socket %d (%d)", sockId, socket);
416
417#ifdef USE_EPOLL
418 epoll_event ev;
419 ev.events = EPOLLIN;
420 ev.data.ptr = this; // user data (union: ev.ptr)
421 if (epoll_ctl(fd_epoll, EPOLL_CTL_ADD, socket, &ev)<0)
422 factPrintf(MessageImp::kError, "epoll_ctl failed: %m (EPOLL_CTL_ADD,rc=%d)", errno);
423#endif
424
425 return retval;
426}
427
428bool READ_STRUCT::read()
429{
430 if (!connected)
431 return false;
432
433 if (bufLen==0)
434 return true;
435
436 const int32_t jrd = recv(socket, bufPos, bufLen, MSG_DONTWAIT);
437 // recv failed
438 if (jrd<0)
439 {
440 // There was just nothing waiting
441 if (errno==EWOULDBLOCK || errno==EAGAIN)
442 return false;
443
444 factPrintf(MessageImp::kError, "Reading from socket %d failed: %m (recv,rc=%d)", sockId, errno);
445 return false;
446 }
447
448 // connection was closed ...
449 if (jrd==0)
450 {
451 factPrintf(MessageImp::kInfo, "Socket %d closed by FAD", sockId);
452
453 destroy();//DestroySocket(rd[i]); //generate address and socket
454 return false;
455 }
456
457 totBytes += jrd;
458
459 // are we skipping this board ...
460 if (bufTyp==kStream)
461 return false;
462
463 if (bufPos==B)
464 gettimeofday(&time, NULL);
465
466 bufPos += jrd; //==> prepare for continuation
467 bufLen -= jrd;
468
469 // not yet all read
470 return bufLen==0;
471}
472
473void READ_STRUCT::swapHeader()
474{
475 S[1] = ntohs(S[1]); // package_length (bytes not swapped!)
476 S[2] = ntohs(S[2]); // version_no
477 S[3] = ntohs(S[3]); // PLLLCK
478 S[4] = ntohs(S[4]); // trigger_crc
479 S[5] = ntohs(S[5]); // trigger_type
480
481 I[3] = ntohl(I[3]); // trigger_id
482 I[4] = ntohl(I[4]); // fad_evt_counter
483 I[5] = ntohl(I[5]); // REFCLK_frequency
484
485 S[12] = ntohs(S[12]); // board id
486 S[13] = ntohs(S[13]); // adc_clock_phase_shift
487 S[14] = ntohs(S[14]); // number_of_triggers_to_generate
488 S[15] = ntohs(S[15]); // trigger_generator_prescaler
489
490 I[10] = ntohl(I[10]); // runnumber;
491 I[11] = ntohl(I[11]); // time;
492
493 // Use back inserter??
494 for (int s=24; s<24+NTemp+NDAC; s++)
495 S[s] = ntohs(S[s]); // drs_temperature / dac
496}
497
498void READ_STRUCT::swapData()
499{
500 // swapEventHeaderBytes: End of the header. to channels now
501
502 int i = 36;
503 for (int ePatchesCount = 0; ePatchesCount<4*9; ePatchesCount++)
504 {
505 S[i+0] = ntohs(S[i+0]);//id
506 S[i+1] = ntohs(S[i+1]);//start_cell
507 S[i+2] = ntohs(S[i+2]);//roi
508 S[i+3] = ntohs(S[i+3]);//filling
509
510 i += 4+S[i+2];//skip the pixel data
511 }
512}
513
514// ==========================================================================
515
516bool checkRoiConsistency(const READ_STRUCT &rd, uint16_t roi[])
517{
518 int xjr = -1;
519 int xkr = -1;
520
521 //points to the very first roi
522 int roiPtr = sizeof(PEVNT_HEADER)/2 + 2;
523
524 roi[0] = ntohs(rd.S[roiPtr]);
525
526 for (int jr = 0; jr < 9; jr++)
527 {
528 roi[jr] = ntohs(rd.S[roiPtr]);
529
530 if (roi[jr]>1024)
531 {
532 factPrintf(MessageImp::kError, "Illegal roi in channel %d (allowed: roi<=1024)", jr, roi[jr]);
533 return false;
534 }
535
536 // Check that the roi of pixels jr are compatible with the one of pixel 0
537 if (jr!=8 && roi[jr]!=roi[0])
538 {
539 xjr = jr;
540 break;
541 }
542
543 // Check that the roi of all other DRS chips on boards are compatible
544 for (int kr = 1; kr < 4; kr++)
545 {
546 const int kroi = ntohs(rd.S[roiPtr]);
547 if (kroi != roi[jr])
548 {
549 xjr = jr;
550 xkr = kr;
551 break;
552 }
553 roiPtr += kroi+4;
554 }
555 }
556
557 if (xjr>=0)
558 {
559 if (xkr<0)
560 factPrintf(MessageImp::kFatal, "Inconsistent Roi accross chips [DRS=%d], expected %d, got %d", xjr, roi[0], roi[xjr]);
561 else
562 factPrintf(MessageImp::kFatal, "Inconsistent Roi accross channels [DRS=%d Ch=%d], expected %d, got %d", xjr, xkr, roi[xjr], ntohs(rd.S[roiPtr]));
563
564 return false;
565 }
566
567 if (roi[8] < roi[0])
568 {
569 factPrintf(MessageImp::kError, "Mismatch of roi (%d) in channel 8. Should be larger or equal than the roi (%d) in channel 0.", roi[8], roi[0]);
570 return false;
571 }
572
573 return true;
574}
575
576list<shared_ptr<EVT_CTRL2>> evtCtrl;
577
578shared_ptr<EVT_CTRL2> mBufEvt(const READ_STRUCT &rd, shared_ptr<RUN_CTRL2> &actrun)
579{
580 /*
581 checkroi consistence
582 find existing entry
583 if no entry, try to allocate memory
584 if entry and memory, init event structure
585 */
586
587 uint16_t nRoi[9];
588 if (!checkRoiConsistency(rd, nRoi))
589 return shared_ptr<EVT_CTRL2>();
590
591 for (auto it=evtCtrl.rbegin(); it!=evtCtrl.rend(); it++)
592 {
593 // A reference is enough because the evtCtrl holds the shared_ptr anyway
594 const shared_ptr<EVT_CTRL2> &evt = *it;
595
596 // If the run is different, go on searching.
597 // We cannot stop searching if a lower run-id is found as in
598 // the case of the events, because theoretically, there
599 // can be the same run on two different days.
600 if (rd.H.runnumber != evt->runNum)
601 continue;
602
603 // If the ID of the new event if higher than the last one stored
604 // in that run, we have to assign a new slot (leave the loop)
605 if (rd.H.fad_evt_counter > evt->evNum/* && runID == evtCtrl[k].runNum*/)
606 break;
607
608 if (rd.H.fad_evt_counter != evt->evNum/* || runID != evtCtrl[k].runNum*/)
609 continue;
610
611 // We have found an entry with the same runID and evtID
612 // Check if ROI is consistent
613 if (evt->nRoi != nRoi[0] || evt->nRoiTM != nRoi[8])
614 {
615 factPrintf(MessageImp::kError, "Mismatch of roi within event. Expected roi=%d and roi_tm=%d, got %d and %d.",
616 evt->nRoi, evt->nRoiTM, nRoi[0], nRoi[8]);
617 return shared_ptr<EVT_CTRL2>();
618 }
619
620 // It is maybe not likely, but the header of this board might have
621 // arrived earlier. (We could also update the run-info, but
622 // this should not make a difference here)
623 if ((rd.time.tv_sec==evt->time.tv_sec && rd.time.tv_usec<evt->time.tv_usec) ||
624 rd.time.tv_sec<evt->time.tv_sec)
625 evt->time = rd.time;
626
627 //everything seems fine so far ==> use this slot ....
628 return evt;
629 }
630
631 if (actrun->runId==rd.H.runnumber && (actrun->roi0 != nRoi[0] || actrun->roi8 != nRoi[8]))
632 {
633 factPrintf(MessageImp::kError, "Mismatch of roi within run. Expected roi=%d and roi_tm=%d, got %d and %d (runID=%d, evID=%d)",
634 actrun->roi0, actrun->roi8, nRoi[0], nRoi[8], rd.H.runnumber, rd.H.fad_evt_counter);
635 return shared_ptr<EVT_CTRL2>();
636 }
637
638 EVT_CTRL2 *evt = new EVT_CTRL2;
639
640 evt->time = rd.time;
641
642 evt->runNum = rd.H.runnumber;
643 evt->evNum = rd.H.fad_evt_counter;
644
645 evt->trgNum = rd.H.trigger_id;
646 evt->trgTyp = rd.H.trigger_type;
647
648 evt->nRoi = nRoi[0];
649 evt->nRoiTM = nRoi[8];
650
651 //evt->firstBoard = rd.sockId;
652
653 const bool newrun = actrun->runId != rd.H.runnumber;
654 if (newrun)
655 {
656 // Since we have started a new run, we know already when to close the
657 // previous run in terms of number of events
658 actrun->maxEvt = actrun->lastEvt;
659
660 factPrintf(MessageImp::kInfo, "New run %d (evt=%d) registered with roi=%d(%d), prev=%d",
661 rd.H.runnumber, rd.H.fad_evt_counter, nRoi[0], nRoi[8], actrun->runId);
662
663 // The new run is the active run now
664 actrun = make_shared<RUN_CTRL2>();
665
666 const time_t &tsec = evt->time.tv_sec;
667
668 actrun->openTime = tsec;
669 actrun->closeTime = tsec + 3600 * 24; // max time allowed
670 actrun->runId = rd.H.runnumber;
671 actrun->roi0 = nRoi[0]; // FIXME: Make obsolete!
672 actrun->roi8 = nRoi[8]; // FIXME: Make obsolete!
673
674 // Signal the fadctrl that a new run has been started
675 // Note this is the only place at which we can ensure that
676 // gotnewRun is called only once
677 gotNewRun(*actrun);
678 }
679
680 // Keep pointer to run of this event
681 evt->runCtrl = actrun;
682
683 // Increase the number of events we have started to receive in this run
684 actrun->lastTime = evt->time.tv_sec; // Time when the last event was received
685 actrun->lastEvt++;
686
687 // An event can be the first and the last, but not the last and the first.
688 // Therefore gotNewRun is called before runFinished.
689 // runFinished signals that the last event of a run was just received. Processing
690 // might still be ongoing, but we can start a new run.
691 const bool cond1 = actrun->lastEvt < actrun->maxEvt; // max number of events not reached
692 const bool cond2 = actrun->lastTime < actrun->closeTime; // max time not reached
693 if (!cond1 || !cond2)
694 runFinished();
695
696 // We don't mind here that this is not common to all events,
697 // because every coming event will fullfil the condition as well.
698 if (!cond1)
699 evt->closeRequest |= kRequestMaxEvtsReached;
700 if (!cond2)
701 evt->closeRequest |= kRequestMaxTimeReached;
702
703 // Secure access to evtCtrl against access in CloseRunFile
704 // This should be the last... otherwise we can run into threading issues
705 // if the event is accessed before it is fully initialized.
706 evtCtrl.emplace_back(evt);
707 return evtCtrl.back();
708}
709
710
711void copyData(const READ_STRUCT &rBuf, EVT_CTRL2 *evt)
712{
713 const int i = rBuf.sockId;
714
715 memcpy(evt->FADhead+i, &rBuf.H, sizeof(PEVNT_HEADER));
716
717 int src = sizeof(PEVNT_HEADER) / 2; // Header is 72 byte = 36 shorts
718
719 // consistency of ROIs have been checked already (is it all correct?)
720 const uint16_t &roi = rBuf.S[src+2];
721
722 // different sort in FAD board.....
723 EVENT *event = evt->fEvent;
724 for (int px = 0; px < 9; px++)
725 {
726 for (int drs = 0; drs < 4; drs++)
727 {
728 const int16_t pixC = rBuf.S[src+1]; // start-cell
729 const int16_t pixR = rBuf.S[src+2]; // roi
730 //here we should check if pixH is correct ....
731
732 const int pixS = i*36 + drs*9 + px;
733
734 event->StartPix[pixS] = pixC;
735
736 memcpy(event->Adc_Data + pixS*roi, &rBuf.S[src+4], roi * 2);
737
738 src += 4+pixR;
739
740 // Treatment for ch 9 (TM channel)
741 if (px != 8)
742 continue;
743
744 const int tmS = i*4 + drs;
745
746 //and we have additional TM info
747 if (pixR > roi)
748 {
749 event->StartTM[tmS] = (pixC + pixR - roi) % 1024;
750
751 memcpy(event->Adc_Data + tmS*roi + NPIX*roi, &rBuf.S[src - roi], roi * 2);
752 }
753 else
754 {
755 event->StartTM[tmS] = -1;
756 }
757 }
758 }
759}
760
761// ==========================================================================
762
763uint64_t reportIncomplete(const shared_ptr<EVT_CTRL2> &evt, const char *txt)
764{
765 factPrintf(MessageImp::kWarn, "skip incomplete evt (run=%d, evt=%d, n=%d, %s)",
766 evt->runNum, evt->evNum, evtCtrl.size(), txt);
767
768 uint64_t report = 0;
769
770 char str[1000];
771
772 int ik=0;
773 for (int ib=0; ib<NBOARDS; ib++)
774 {
775 if (ib%10==0)
776 str[ik++] = '|';
777
778 const int jb = evt->board[ib];
779 if (jb>=0) // data received from that board
780 {
781 str[ik++] = '0'+(jb%10);
782 continue;
783 }
784
785 // FIXME: This is not synchronous... it reports
786 // accoridng to the current connection status, not w.r.t. to the
787 // one when the event was taken.
788 if (gi_NumConnect[ib]==0) // board not connected
789 {
790 str[ik++] = 'x';
791 continue;
792 }
793
794 // data from this board lost
795 str[ik++] = '.';
796 report |= ((uint64_t)1)<<ib;
797 }
798
799 str[ik++] = '|';
800 str[ik] = 0;
801
802 factOut(MessageImp::kWarn, str);
803
804 return report;
805}
806
807// ==========================================================================
808// ==========================================================================
809
810bool proc1(const shared_ptr<EVT_CTRL2> &);
811
812Queue<shared_ptr<EVT_CTRL2>> processingQueue1(bind(&proc1, placeholders::_1));
813
814bool proc1(const shared_ptr<EVT_CTRL2> &evt)
815{
816 applyCalib(*evt, processingQueue1.size());
817 return true;
818}
819
820// If this is not convenient anymore, it could be replaced by
821// a command queue, to which command+data is posted,
822// (e.g. runOpen+runInfo, runClose+runInfo, evtWrite+evtInfo)
823bool writeEvt(const shared_ptr<EVT_CTRL2> &evt)
824{
825 //const shared_ptr<RUN_CTRL2> &run = evt->runCtrl;
826 RUN_CTRL2 &run = *evt->runCtrl;
827
828 // Is this a valid event or just an empty event to trigger run close?
829 // If this is not an empty event open the new run-file
830 // Empty events are there to trigger run-closing conditions
831 if (evt->valid())
832 {
833 // File not yet open
834 if (run.fileStat==kFileNotYetOpen)
835 {
836 // runOpen will close a previous run, if still open
837 if (!runOpen(*evt))
838 {
839 factPrintf(MessageImp::kError, "Could not open new file for run %d (evt=%d, runOpen failed)", evt->runNum, evt->evNum);
840 run.fileStat = kFileClosed;
841 return true;
842 }
843
844 factPrintf(MessageImp::kInfo, "Opened new file for run %d (evt=%d)", evt->runNum, evt->evNum);
845 run.fileStat = kFileOpen;
846 }
847
848 // Here we have a valid calibration and can go on with that.
849 // It is important that _all_ events are sent for calibration (except broken ones)
850 processingQueue1.post(evt);
851 }
852
853 // File already closed
854 if (run.fileStat==kFileClosed)
855 return true;
856
857 // If we will have a software trigger which prevents single events from writing,
858 // the logic of writing the stop time and the trigger counters need to be adapted.
859 // Currently it is just the values of the last valid event.
860 bool rc1 = true;
861 if (evt->valid())
862 {
863 rc1 = runWrite(*evt);
864 if (!rc1)
865 factPrintf(MessageImp::kError, "Writing event %d for run %d failed (runWrite)", evt->evNum, evt->runNum);
866 }
867
868 // File not open... no need to close or to check for close
869 // ... this is the case if CloseRunFile was called before any file was opened.
870 if (run.fileStat!=kFileOpen)
871 return true;
872
873 // File is not yet to be closed.
874 if (rc1 && evt->closeRequest==kRequestNone)
875 return true;
876
877 runClose(*evt);
878 run.fileStat = kFileClosed;
879
880 vector<string> reason;
881 if (evt->closeRequest&kRequestManual)
882 reason.emplace_back("close requested");
883 if (evt->closeRequest&kRequestTimeout)
884 reason.emplace_back("receive timeout");
885 if (evt->closeRequest&kRequestConnectionChange)
886 reason.emplace_back("connection changed");
887 if (evt->closeRequest&kRequestEventCheckFailed)
888 reason.emplace_back("event check failed");
889 if (evt->closeRequest&kRequestMaxTimeReached)
890 reason.push_back(to_string(run.closeTime-run.openTime)+"s reached");
891 if (evt->closeRequest&kRequestMaxEvtsReached)
892 reason.push_back(to_string(run.maxEvt)+" evts reached");
893 if (!rc1)
894 reason.emplace_back("runWrite failed");
895
896 const string str = boost::algorithm::join(reason, ", ");
897 factPrintf(MessageImp::kInfo, "File closed because %s", str.c_str());
898
899 return true;
900}
901
902Queue<shared_ptr<EVT_CTRL2>> secondaryQueue(bind(&writeEvt, placeholders::_1));
903
904bool procEvt(const shared_ptr<EVT_CTRL2> &evt)
905{
906 RUN_CTRL2 &run = *evt->runCtrl;
907
908 bool check = true;
909 if (evt->valid())
910 {
911 EVENT *event = evt->fEvent;
912
913 // This is already done in initMemory()
914 //event->Roi = evt->runCtrl->roi0;
915 //event->RoiTM = evt->runCtrl->roi8;
916 //event->EventNum = evt->evNum;
917 //event->TriggerNum = evt->trgNum;
918 //event->TriggerType = evt->trgTyp;
919
920 event->NumBoards = evt->nBoard;
921
922 event->PCTime = evt->time.tv_sec;
923 event->PCUsec = evt->time.tv_usec;
924
925 for (int ib=0; ib<NBOARDS; ib++)
926 event->BoardTime[ib] = evt->FADhead[ib].time;
927
928 check = eventCheck(*evt);
929
930 // If the event is valid, increase the trigger counter accordingly
931 if (check)
932 {
933 // Physics trigger
934 if (evt->trgTyp && !(evt->trgTyp & FAD::EventHeader::kAll))
935 run.triggerCounter[0]++;
936 // Pure pedestal trigger
937 else if ((evt->trgTyp&FAD::EventHeader::kPedestal) && !(evt->trgTyp&FAD::EventHeader::kTIM))
938 run.triggerCounter[1]++;
939 // external light pulser trigger
940 else if (evt->trgTyp & FAD::EventHeader::kLPext)
941 run.triggerCounter[2]++;
942 // time calibration triggers
943 else if (evt->trgTyp & (FAD::EventHeader::kTIM|FAD::EventHeader::kPedestal))
944 run.triggerCounter[3]++;
945 // internal light pulser trigger
946 else if (evt->trgTyp & FAD::EventHeader::kLPint)
947 run.triggerCounter[4]++;
948 // external trigger input 1
949 else if (evt->trgTyp & FAD::EventHeader::kExt1)
950 run.triggerCounter[5]++;
951 // external trigger input 2
952 else if (evt->trgTyp & FAD::EventHeader::kExt2)
953 run.triggerCounter[6]++;
954 // other triggers
955 else
956 run.triggerCounter[7]++;
957 }
958 }
959
960 // If this is an invalid event, the current triggerCounter needs to be copied
961 // because runClose will use that one to update the TRIGGER_COUNTER.
962 // When closing the file, the trigger counter of the last successfully
963 // written event is used.
964 evt->triggerCounter = run.triggerCounter;
965
966 // If event check has failed, skip the event and post a close request instead.
967 // Otherwise, if file is open post the event for being written
968 if (!check)
969 secondaryQueue.emplace(new EVT_CTRL2(kRequestEventCheckFailed, evt->runCtrl));
970 else
971 secondaryQueue.post(evt);
972
973 return true;
974}
975
976// ==========================================================================
977// ==========================================================================
978
979/*
980 task 1-4:
981
982 lock1()-lock4();
983 while (1)
984 {
985 wait for signal [lockN]; // unlocked
986
987 while (n!=10)
988 wait sockets;
989 read;
990
991 lockM();
992 finished[n] = true;
993 signal(mainloop);
994 unlockM();
995 }
996
997
998 mainloop:
999
1000 while (1)
1001 {
1002 lockM();
1003 while (!finished[0] || !finished[1] ...)
1004 wait for signal [lockM]; // unlocked... signals can be sent
1005 finished[0-1] = false;
1006 unlockM()
1007
1008 copy data to queue // locked
1009
1010 lockN[0-3];
1011 signalN[0-3];
1012 unlockN[0-3];
1013 }
1014
1015
1016 */
1017
1018/*
1019 while (g_reset)
1020 {
1021 shared_ptr<EVT_CTRL2> evt = new shared_ptr<>;
1022
1023 // Check that all sockets are connected
1024
1025 for (int i=0; i<40; i++)
1026 if (rd[i].connected && epoll_ctl(fd_epoll, EPOLL_CTL_ADD, socket, NULL)<0)
1027 factPrintf(kError, "epoll_ctrl failed: %m (EPOLL_CTL_ADD,rc=%d)", errno);
1028
1029 while (g_reset)
1030 {
1031 if (READ_STRUCT::wait()<0)
1032 break;
1033
1034 if (rc_epoll==0)
1035 break;
1036
1037 for (int jj=0; jj<rc_epoll; jj++)
1038 {
1039 READ_STRUCT *rs = READ_STRUCT::get(jj);
1040 if (!rs->connected)
1041 continue;
1042
1043 const bool rc_read = rs->read();
1044 if (!rc_read)
1045 continue;
1046
1047 if (rs->bufTyp==READ_STRUCT::kHeader)
1048 {
1049 [...]
1050 }
1051
1052 [...]
1053
1054 if (epoll_ctl(fd_epoll, EPOLL_CTL_DEL, socket, NULL)<0)
1055 factPrintf(kError, "epoll_ctrl failed: %m (EPOLL_CTL_DEL,rc=%d)", errno);
1056 }
1057
1058 if (once_a_second)
1059 {
1060 if (evt==timeout)
1061 break;
1062 }
1063 }
1064
1065 if (evt.nBoards==actBoards)
1066 primaryQueue.post(evt);
1067 }
1068*/
1069
1070Queue<shared_ptr<EVT_CTRL2>> primaryQueue(bind(&procEvt, placeholders::_1));
1071
1072// This corresponds more or less to fFile... should we merge both?
1073shared_ptr<RUN_CTRL2> actrun;
1074
1075void CloseRunFile()
1076{
1077 // Currently we need actrun here, to be able to set kFileClosed.
1078 // Apart from that we have to ensure that there is an open file at all
1079 // which we can close.
1080 // Submission to the primary queue ensures that the event
1081 // is placed at the right place in the processing chain.
1082 // (Corresponds to the correct run)
1083 primaryQueue.emplace(new EVT_CTRL2(kRequestManual, actrun));
1084}
1085
1086bool mainloop(READ_STRUCT *rd)
1087{
1088 factPrintf(MessageImp::kInfo, "Starting EventBuilder main loop");
1089
1090 primaryQueue.start();
1091 secondaryQueue.start();
1092 processingQueue1.start();;
1093
1094 actrun = make_shared<RUN_CTRL2>();
1095
1096 //time in seconds
1097 time_t gi_SecTime = time(NULL)-1;
1098
1099 //loop until global variable g_runStat claims stop
1100 g_reset = 0;
1101 while (g_reset == 0)
1102 {
1103#ifdef USE_POLL
1104 int pp[40];
1105 int nn = 0;
1106 pollfd fds[40];
1107 for (int i=0; i<40; i++)
1108 {
1109 if (rd[i].socket>=0 && rd[i].connected && rd[i].bufLen>0)
1110 {
1111 fds[nn].fd = rd[i].socket;
1112 fds[nn].events = POLLIN;
1113 pp[nn] = i;
1114 nn++;
1115 }
1116 }
1117
1118 const int rc_epoll = poll(fds, nn, 100);
1119 if (rc_epoll<0)
1120 break;
1121#endif
1122
1123#ifdef USE_SELECT
1124 fd_set readfs;
1125 FD_ZERO(&readfs);
1126 int nfsd = 0;
1127 for (int i=0; i<NBOARDS; i++)
1128 if (rd[i].socket>=0 && rd[i].connected && rd[i].bufLen>0)
1129 {
1130 FD_SET(rd[i].socket, &readfs);
1131 if (rd[i].socket>nfsd)
1132 nfsd = rd[i].socket;
1133 }
1134
1135 timeval tv;
1136 tv.tv_sec = 0;
1137 tv.tv_usec = 100000;
1138 const int rc_select = select(nfsd+1, &readfs, NULL, NULL, &tv);
1139 // 0: timeout
1140 // -1: error
1141 if (rc_select<0)
1142 {
1143 factPrintf(MessageImp::kError, "Waiting for data failed: %d (select,rc=%d)", errno);
1144 continue;
1145 }
1146#endif
1147
1148#ifdef USE_EPOLL
1149 const int rc_epoll = READ_STRUCT::wait();
1150 if (rc_epoll<0)
1151 break;
1152#endif
1153
1154#ifdef PRIORITY_QUEUE
1155 priority_queue<READ_STRUCT*, vector<READ_STRUCT*>, READ_STRUCTcomp> prio;
1156
1157 for (int i=0; i<NBOARDS; i++)
1158 if (rd[i].connected)
1159 prio.push(rd+i);
1160
1161 if (!prio.empty()) do
1162#endif
1163
1164
1165#ifdef USE_POLL
1166 for (int jj=0; jj<nn; jj++)
1167#endif
1168#ifdef USE_EPOLL
1169 for (int jj=0; jj<rc_epoll; jj++)
1170#endif
1171#if !defined(USE_EPOLL) && !defined(USE_POLL) && !defined(PRIORITY_QUEUE)
1172 for (int jj=0; jj<NBOARDS; jj++)
1173#endif
1174 {
1175#ifdef PRIORITY_QUEUE
1176 READ_STRUCT *rs = prio.top();
1177#endif
1178#ifdef USE_SELECT
1179 if (!FD_ISSET(rs->socket, &readfs))
1180 continue;
1181#endif
1182
1183#ifdef USE_POLL
1184 if ((fds[jj].revents&POLLIN)==0)
1185 continue;
1186#endif
1187
1188#ifdef USE_EPOLL
1189 // FIXME: How to get i?
1190 READ_STRUCT *rs = READ_STRUCT::get(jj);
1191#endif
1192
1193#ifdef USE_POLL
1194 // FIXME: How to get i?
1195 READ_STRUCT *rs = &rd[pp[jj]];
1196#endif
1197
1198#if !defined(USE_POLL) && !defined(USE_EPOLL) && !defined(PRIORITY_QUEUE)
1199 const int i = (jj%4)*10 + (jj/4);
1200 READ_STRUCT *rs = &rd[i];
1201#endif
1202
1203#ifdef COMPLETE_EVENTS
1204 if (rs->bufTyp==READ_STRUCT::kWait)
1205 continue;
1206#endif
1207
1208 // ==================================================================
1209
1210 const bool rc_read = rs->read();
1211
1212 // Connect might have gotten closed during read
1213 gi_NumConnect[rs->sockId] = rs->connected;
1214 gj.numConn[rs->sockId] = rs->connected;
1215
1216 // Read either failed or disconnected, or the buffer is not yet full
1217 if (!rc_read)
1218 continue;
1219
1220 // ==================================================================
1221
1222 if (rs->bufTyp==READ_STRUCT::kHeader)
1223 {
1224 //check if startflag correct; else shift block ....
1225 // FIXME: This is not enough... this combination of
1226 // bytes can be anywhere... at least the end bytes
1227 // must be checked somewhere, too.
1228 uint k;
1229 for (k=0; k<sizeof(PEVNT_HEADER)-1; k++)
1230 {
1231 if (rs->B[k]==0xfb && rs->B[k+1] == 0x01)
1232 break;
1233 }
1234 rs->skip += k;
1235
1236 //no start of header found
1237 if (k==sizeof(PEVNT_HEADER)-1)
1238 {
1239 rs->B[0] = rs->B[sizeof(PEVNT_HEADER)-1];
1240 rs->bufPos = rs->B+1;
1241 rs->bufLen = sizeof(PEVNT_HEADER)-1;
1242 continue;
1243 }
1244
1245 if (k > 0)
1246 {
1247 memmove(rs->B, rs->B+k, sizeof(PEVNT_HEADER)-k);
1248
1249 rs->bufPos -= k;
1250 rs->bufLen += k;
1251
1252 continue; // We need to read more (bufLen>0)
1253 }
1254
1255 if (rs->skip>0)
1256 {
1257 factPrintf(MessageImp::kInfo, "Skipped %d bytes on port %d", rs->skip, rs->sockId);
1258 rs->skip = 0;
1259 }
1260
1261 // Swap the header entries from network to host order
1262 rs->swapHeader();
1263
1264 rs->bufTyp = READ_STRUCT::kData;
1265 rs->bufLen = rs->len() - sizeof(PEVNT_HEADER);
1266
1267 debugHead(rs->B); // i and fadBoard not used
1268
1269 continue;
1270 }
1271
1272 const uint16_t &end = *reinterpret_cast<uint16_t*>(rs->bufPos-2);
1273 if (end != 0xfe04)
1274 {
1275 factPrintf(MessageImp::kError, "End-of-event flag wrong on socket %2d for event %d (len=%d), got %04x",
1276 rs->sockId, rs->H.fad_evt_counter, rs->len(), end);
1277
1278 // ready to read next header
1279 rs->bufTyp = READ_STRUCT::kHeader;
1280 rs->bufLen = sizeof(PEVNT_HEADER);
1281 rs->bufPos = rs->B;
1282 // FIXME: What to do with the validity flag?
1283 continue;
1284 }
1285
1286 // get index into mBuffer for this event (create if needed)
1287 const shared_ptr<EVT_CTRL2> evt = mBufEvt(*rs, actrun);
1288
1289 // We have a valid entry, but no memory has yet been allocated
1290 if (evt && !evt->initMemory())
1291 {
1292 const time_t tm = time(NULL);
1293 if (evt->runCtrl->reportMem==tm)
1294 continue;
1295
1296 factPrintf(MessageImp::kError, "No free memory left for %d (run=%d)", evt->evNum, evt->runNum);
1297 evt->runCtrl->reportMem = tm;
1298 continue;
1299 }
1300
1301 // ready to read next header
1302 rs->bufTyp = READ_STRUCT::kHeader;
1303 rs->bufLen = sizeof(PEVNT_HEADER);
1304 rs->bufPos = rs->B;
1305
1306 // Fatal error occured. Event cannot be processed. Skip it. Start reading next header.
1307 if (!evt)
1308 continue;
1309
1310 // This should never happen
1311 if (evt->board[rs->sockId] != -1)
1312 {
1313 factPrintf(MessageImp::kError, "Got event %5d from board %3d (i=%3d, len=%5d) twice.",
1314 evt->evNum, rs->sockId, jj, rs->len());
1315 // FIXME: What to do with the validity flag?
1316 continue; // Continue reading next header
1317 }
1318
1319 // Swap the data entries (board headers) from network to host order
1320 rs->swapData();
1321
1322 // Copy data from rd[i] to mBuffer[evID]
1323 copyData(*rs, evt.get());
1324
1325#ifdef COMPLETE_EVENTS
1326 // Do not read anmymore from this board until the whole event has been received
1327 rs->bufTyp = READ_STRUCT::kWait;
1328#endif
1329 // now we have stored a new board contents into Event structure
1330 evt->board[rs->sockId] = rs->sockId;
1331 evt->header = evt->FADhead+rs->sockId;
1332 evt->nBoard++;
1333
1334#ifdef COMPLETE_EPOLL
1335 if (epoll_ctl(READ_STRUCT::fd_epoll, EPOLL_CTL_DEL, rs->socket, NULL)<0)
1336 {
1337 factPrintf(MessageImp::kError, "epoll_ctrl failed: %m (EPOLL_CTL_DEL,rc=%d)", errno);
1338 break;
1339 }
1340#endif
1341 // event not yet complete
1342 if (evt->nBoard < READ_STRUCT::activeSockets)
1343 continue;
1344
1345 // All previous events are now flagged as incomplete ("expired")
1346 // and will be removed. (This is a bit tricky, because pop_front()
1347 // would invalidate the current iterator if not done _after_ the increment)
1348 for (auto it=evtCtrl.begin(); it!=evtCtrl.end(); )
1349 {
1350 const bool found = it->get()==evt.get();
1351 if (!found)
1352 reportIncomplete(*it, "expired");
1353 else
1354 primaryQueue.post(evt);
1355
1356 // package_len is 0 if nothing was received.
1357 for (int ib=0; ib<40; ib++)
1358 rd[ib].relBytes += uint32_t((*it)->FADhead[ib].package_length)*2;
1359
1360 // The counter must be increased _before_ the pop_front,
1361 // otherwise the counter is invalidated by the pop_front!
1362 it++;
1363 evtCtrl.pop_front();
1364
1365 // We reached the current event, so we are done
1366 if (found)
1367 break;
1368 }
1369
1370#ifdef COMPLETE_EPOLL
1371 for (int j=0; j<40; j++)
1372 {
1373 epoll_event ev;
1374 ev.events = EPOLLIN;
1375 ev.data.ptr = &rd[j]; // user data (union: ev.ptr)
1376 if (epoll_ctl(READ_STRUCT::fd_epoll, EPOLL_CTL_ADD, rd[j].socket, &ev)<0)
1377 {
1378 factPrintf(MessageImp::kError, "epoll_ctl failed: %m (EPOLL_CTL_ADD,rc=%d)", errno);
1379 return;
1380 }
1381 }
1382#endif
1383
1384#ifdef COMPLETE_EVENTS
1385 for (int j=0; j<40; j++)
1386 {
1387 //if (rs->bufTyp==READ_STRUCT::kWait)
1388 {
1389 rs->bufTyp = READ_STRUCT::kHeader;
1390 rs->bufLen = sizeof(PEVNT_HEADER);
1391 rs->bufPos = rs->B;
1392 }
1393 }
1394#endif
1395 } // end for loop over all sockets
1396#ifdef PRIORITY_QUEUE
1397 while (0); // convert continue into break ;)
1398#endif
1399
1400 // ==================================================================
1401
1402 const time_t actTime = time(NULL);
1403 if (actTime == gi_SecTime)
1404 {
1405#if !defined(USE_SELECT) && !defined(USE_EPOLL) && !defined(USE_POLL)
1406 if (evtCtrl.empty())
1407 usleep(actTime-actrun->lastTime>300 ? 10000 : 1);
1408#endif
1409 continue;
1410 }
1411 gi_SecTime = actTime;
1412
1413 // ==================================================================
1414 //loop over all active events and flag those older than read-timeout
1415 //delete those that are written to disk ....
1416
1417 // This could be improved having the pointer which separates the queue with
1418 // the incomplete events from the queue with the complete events
1419 for (auto it=evtCtrl.begin(); it!=evtCtrl.end(); )
1420 {
1421 // A reference is enough because the shared_ptr is hold by the evtCtrl
1422 const shared_ptr<EVT_CTRL2> &evt = *it;
1423
1424 // The first event is the oldest. If the first event within the
1425 // timeout window was received, we can stop searching further.
1426 if (evt->time.tv_sec+g_evtTimeout>=actTime)
1427 break;
1428
1429 // The counter must be increased _before_ the pop_front,
1430 // otherwise the counter is invalidated by the pop_front!
1431 it++;
1432
1433 // This timeout is caused because complete data from one or more
1434 // boards has been received, but the memory could not be allocated.
1435 // There is no reason why we should not go on waiting for
1436 // memory to become free. However, the FADs will disconnect
1437 // after 60s due to their keep-alive timeout, but the event builder
1438 // will still wait for memory to become available.
1439 // Currently, the only possibility to free the memory from the
1440 // evtCtrl to restart the event builder (STOP/START).
1441 if (!evt->valid())
1442 continue;
1443
1444 // This will result in the emission of a dim service.
1445 // It doesn't matter if that takes comparably long,
1446 // because we have to stop the run anyway.
1447 const uint64_t rep = reportIncomplete(evt, "timeout");
1448 factReportIncomplete(rep);
1449
1450 // At least the data from one boards is complete...
1451 // package_len is 0 when nothing was received from this board
1452 for (int ib=0; ib<40; ib++)
1453 rd[ib].relBytes += uint32_t(evt->FADhead[ib].package_length)*2;
1454
1455 evtCtrl.pop_front();
1456 }
1457
1458 // =================================================================
1459
1460 gj.bufNew = evtCtrl.size(); //# incomplete events in buffer
1461 gj.bufEvt = primaryQueue.size(); //# complete events in buffer
1462 gj.bufWrite = secondaryQueue.size(); //# complete events in buffer
1463 gj.bufProc = processingQueue1.size(); //# complete events in buffer
1464 gj.bufTot = Memory::max_inuse/MAX_TOT_MEM;
1465 gj.usdMem = Memory::max_inuse;
1466 gj.totMem = Memory::allocated;
1467 gj.maxMem = g_maxMem;
1468
1469 gj.deltaT = 1000; // temporary, must be improved
1470
1471 bool changed = false;
1472
1473 static vector<uint64_t> store(NBOARDS);
1474
1475 for (int ib=0; ib<NBOARDS; ib++)
1476 {
1477 gj.rateBytes[ib] = store[ib]>rd[ib].totBytes ? rd[ib].totBytes : rd[ib].totBytes-store[ib];
1478 gj.relBytes[ib] = rd[ib].totBytes-rd[ib].relBytes;
1479
1480 store[ib] = rd[ib].totBytes;
1481
1482 if (rd[ib].check(g_port[ib].sockDef, g_port[ib].sockAddr))
1483 changed = true;
1484
1485 gi_NumConnect[ib] = rd[ib].connected;
1486 gj.numConn[ib] = rd[ib].connected;
1487 }
1488
1489 factStat(gj);
1490
1491 Memory::max_inuse = 0;
1492
1493 // =================================================================
1494
1495 // This is a fake event to trigger possible run-closing conditions once a second
1496 // FIXME: This is not yet ideal because a file would never be closed
1497 // if a new file has been started and no events of the new file
1498 // have been received yet
1499 int request = kRequestNone;
1500
1501 // If nothing was received for more than 5min, close file
1502 if (actTime-actrun->lastTime>300)
1503 request |= kRequestTimeout;
1504
1505 // If connection status has changed
1506 if (changed)
1507 request |= kRequestConnectionChange;
1508
1509 if (request!=kRequestNone)
1510 runFinished();
1511
1512 if (actrun->fileStat==kFileOpen)
1513 primaryQueue.emplace(new EVT_CTRL2(request, actrun));
1514 }
1515
1516 // 1: Stop, wait for event to get processed
1517 // 2: Stop, finish immediately
1518 // 101: Restart, wait for events to get processed
1519 // 101: Restart, finish immediately
1520 //
1521 const int gi_reset = g_reset;
1522
1523 const bool abort = gi_reset%100==2;
1524
1525 factPrintf(MessageImp::kInfo, "Stop reading ... RESET=%d (%s threads)", gi_reset, abort?"abort":"join");
1526
1527 primaryQueue.wait(abort);
1528 secondaryQueue.wait(abort);
1529 processingQueue1.wait(abort);
1530
1531 // Here we also destroy all runCtrl structures and hence close all open files
1532 evtCtrl.clear();
1533 actrun.reset();
1534
1535 factPrintf(MessageImp::kInfo, "Exit read Process...");
1536 factPrintf(MessageImp::kInfo, "%llu Bytes flagged as in-use.", Memory::inuse);
1537
1538 factStat(gj);
1539
1540 return gi_reset>=100;
1541}
1542
1543// ==========================================================================
1544// ==========================================================================
1545
1546void StartEvtBuild()
1547{
1548 factPrintf(MessageImp::kInfo, "Starting EventBuilder++");
1549
1550 memset(gi_NumConnect, 0, NBOARDS*sizeof(*gi_NumConnect));
1551
1552 memset(&gj, 0, sizeof(GUI_STAT));
1553
1554 gj.usdMem = Memory::inuse;
1555 gj.totMem = Memory::allocated;
1556 gj.maxMem = g_maxMem;
1557
1558
1559 READ_STRUCT rd[NBOARDS];
1560
1561 // This is only that every socket knows its id (maybe we replace that by arrays instead of an array of sockets)
1562 for (int i=0; i<NBOARDS; i++)
1563 rd[i].sockId = i;
1564
1565 while (mainloop(rd));
1566
1567 //must close all open sockets ...
1568 factPrintf(MessageImp::kInfo, "Close all sockets...");
1569
1570 READ_STRUCT::close();
1571
1572 // Now all sockets get closed. This is not reflected in gi_NumConnect
1573 // The current workaround is to count all sockets as closed when the thread is not running
1574 factPrintf(MessageImp::kInfo, "EventBuilder++ closed");
1575}
Note: See TracBrowser for help on using the repository browser.