source: trunk/FACT++/src/EventBuilder.c@ 15470

Last change on this file since 15470 was 15470, checked in by tbretz, 12 years ago
Some more renaming; some values can already be initialized as soon as evtCtrl->fEvent is valid, no need to wait until the event is written or processed.
File size: 72.1 KB
Line 
1
2// // // #define EVTDEBUG
3
4#define NUMSOCK 1 //set to 7 for old configuration
5#define MAXREAD 65536 //64kB wiznet buffer
6
7#include <stdlib.h>
8#include <stdint.h>
9#include <stdarg.h>
10#include <unistd.h>
11#include <stdio.h>
12#include <sys/time.h>
13#include <arpa/inet.h>
14#include <string.h>
15#include <math.h>
16#include <error.h>
17#include <errno.h>
18#include <unistd.h>
19#include <sys/types.h>
20#include <sys/socket.h>
21#include <netinet/in.h>
22#include <netinet/tcp.h>
23#include <pthread.h>
24#include <sched.h>
25
26#include "EventBuilder.h"
27
28enum Severity
29{
30 kMessage = 10, ///< Just a message, usually obsolete
31 kInfo = 20, ///< An info telling something which can be interesting to know
32 kWarn = 30, ///< A warning, things that somehow might result in unexpected or unwanted bahaviour
33 kError = 40, ///< Error, something unexpected happened, but can still be handled by the program
34 kFatal = 50, ///< An error which cannot be handled at all happend, the only solution is program termination
35 kDebug = 99, ///< A message used for debugging only
36};
37
38#define MIN_LEN 32 // min #bytes needed to interpret FADheader
39#define MAX_LEN 256*1024 // size of read-buffer per socket
40
41//#define nanosleep(x,y)
42
43extern FileHandle_t runOpen (uint32_t irun, RUN_HEAD * runhd, size_t len);
44extern int runWrite (FileHandle_t fileHd, EVENT * event, size_t len);
45extern int runClose (FileHandle_t fileHd, RUN_TAIL * runth, size_t len);
46//extern int runFinish (uint32_t runnr);
47
48extern void factOut (int severity, int err, char *message);
49extern void factReportIncomplete (uint64_t rep);
50
51extern void gotNewRun (int runnr, PEVNT_HEADER * headers);
52
53
54extern void factStat (GUI_STAT gj);
55
56extern void factStatNew (EVT_STAT gi);
57
58extern int eventCheck (uint32_t runNr, PEVNT_HEADER * fadhd, EVENT * event);
59
60extern int subProcEvt (int threadID, PEVNT_HEADER * fadhd, EVENT * event,
61 int8_t * buffer);
62
63extern void debugHead (int i, int j, void *buf);
64
65extern void debugRead (int isock, int ibyte, int32_t event, int32_t ftmevt,
66 int32_t runnr, int state, uint32_t tsec,
67 uint32_t tusec);
68extern void debugStream (int isock, void *buf, int len);
69
70int CloseRunFile (uint32_t runId, uint32_t closeTime, uint32_t maxEvt);
71
72
73int evtCtrl_frstPtr;
74int evtCtrl_lastPtr;
75
76
77
78int g_maxProc;
79int g_maxSize;
80int gi_maxSize;
81int gi_maxProc;
82
83uint g_actTime;
84uint g_actUsec;
85int g_runStat;
86int g_reset;
87int g_useFTM;
88
89int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX;
90size_t g_maxMem; //maximum memory allowed for buffer
91
92//no longer needed ...
93int g_maxBoards; //maximum number of boards to be initialized
94int g_actBoards;
95//
96
97FACT_SOCK g_port[NBOARDS]; // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd"
98
99
100int gi_runStat;
101int gp_runStat;
102int gw_runStat;
103
104//int gi_memStat = +1;
105
106uint32_t actrun = 0;
107
108
109uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards
110
111//uint gi_EvtStart= 0 ;
112//uint gi_EvtRead = 0 ;
113//uint gi_EvtBad = 0 ;
114//uint gi_EvtTot = 0 ;
115//size_t gi_usedMem = 0 ;
116
117//uint gw_EvtTot = 0 ;
118//uint gp_EvtTot = 0 ;
119
120//PIX_MAP g_pixMap[NPIX];
121
122EVT_STAT gi;
123GUI_STAT gj;
124
125EVT_CTRL evtCtrl[MAX_EVT * MAX_RUN]; //control of events during processing
126
127//#define MXSTR 1000
128//char str[MXSTR];
129
130void factPrintf(int severity, int id, const char *fmt, ...)
131{
132 char str[1000];
133
134 va_list ap;
135 va_start(ap, fmt);
136 vsnprintf(str, 1000, fmt, ap);
137 va_end(ap);
138
139 factOut(severity, id, str);
140}
141
142
143#define MAX_HEAD_MEM (NBOARDS * sizeof(PEVNT_HEADER))
144#define MAX_TOT_MEM (sizeof(EVENT) + (NPIX+NTMARK)*1024*2 + MAX_HEAD_MEM)
145typedef struct TGB_struct
146{
147 struct TGB_struct *prev;
148 void *mem;
149} TGB_entry;
150
151TGB_entry *tgb_last = NULL;
152uint64_t tgb_memory = 0;
153uint64_t tgb_inuse = 0;
154
155void *TGB_Malloc()
156{
157 // No free slot available, next alloc would exceed max memory
158 if (!tgb_last && tgb_memory+MAX_TOT_MEM>g_maxMem)
159 return NULL;
160
161 // We will return this amount of memory
162 tgb_inuse += MAX_TOT_MEM;
163
164 // No free slot available, allocate a new one
165 if (!tgb_last)
166 {
167 tgb_memory += MAX_TOT_MEM;
168 return malloc(MAX_TOT_MEM);
169 }
170
171 // Get the next free slot from the stack and return it
172 TGB_entry *last = tgb_last;
173
174 void *mem = last->mem;
175 tgb_last = last->prev;
176
177 free(last);
178
179 return mem;
180};
181
182void TGB_free(void *mem)
183{
184 // Add the last free slot to the stack
185 TGB_entry *entry = (TGB_entry*)malloc(sizeof(TGB_entry));
186
187 entry->prev = tgb_last;
188 entry->mem = mem;
189
190 tgb_last = entry;
191
192 // Decrease the amont of memory in use accordingly
193 tgb_inuse -= MAX_TOT_MEM;
194}
195
196RUN_CTRL runCtrl[MAX_RUN];
197//RUN_TAIL runTail[MAX_RUN];
198
199
200/*
201*** Definition of rdBuffer to read in IP packets; keep it global !!!!
202 */
203
204
205typedef union
206{
207 int8_t B[MAX_LEN];
208 int16_t S[MAX_LEN / 2];
209 int32_t I[MAX_LEN / 4];
210 int64_t L[MAX_LEN / 8];
211} CNV_FACT;
212
213typedef struct
214{
215 int bufTyp; //what are we reading at the moment: 0=header 1=data -1=skip ...
216 int32_t bufPos; //next byte to read to the buffer next
217 int32_t bufLen; //number of bytes left to read
218// size_t bufLen; //number of bytes left to read size_t might be better
219 int32_t skip; //number of bytes skipped before start of event
220
221 int errCnt; //how often connect failed since last successful
222 int sockStat; //-1 if socket not yet connected , 99 if not exist
223 int socket; //contains the sockets
224 struct sockaddr_in SockAddr; //IP for each socket
225
226 int evtID; // event ID of event currently read
227 int runID; // run "
228 int ftmID; // event ID from FTM
229 uint fadLen; // FADlength of event currently read
230 int fadVers; // Version of FAD
231 int ftmTyp; // trigger type
232 int board; // boardID (softwareID: 0..40 )
233 int Port;
234
235 CNV_FACT *rBuf;
236
237} READ_STRUCT;
238
239
240typedef union
241{
242 int8_t B[2];
243 int16_t S;
244} SHORT_BYTE;
245
246
247
248
249
250SHORT_BYTE start, stop;
251
252READ_STRUCT rd[MAX_SOCK]; //buffer to read IP and afterwards store in mBuffer
253
254
255
256/*-----------------------------------------------------------------*/
257
258
259/*-----------------------------------------------------------------*/
260
261
262
263int
264runFinish1 (uint32_t runnr)
265{
266 factPrintf(kInfo, 173, "Should finish(1) run %d (but not yet possible)", runnr);
267 return 0;
268}
269int
270runFinish (uint32_t runnr)
271{
272 factPrintf(kInfo, 173, "Should finish run %d (but not yet possible)", runnr);
273 return 0;
274}
275
276int
277GenSock (int flag, int sid, int port, struct sockaddr_in *sockAddr,
278 READ_STRUCT * rs)
279{
280/*
281*** generate Address, create sockets and allocates readbuffer for it
282***
283*** if flag==0 generate socket and buffer
284*** <0 destroy socket and buffer
285*** >0 close and redo socket
286***
287*** sid : board*7 + port id
288 */
289
290 int j;
291 int optval = 1; //activate keepalive
292 socklen_t optlen = sizeof (optval);
293
294
295 if (sid % 7 >= NUMSOCK) {
296 //this is a not used socket, so do nothing ...
297 rs->sockStat = 77;
298 rs->rBuf = NULL ;
299 return 0;
300 }
301
302 if (rs->sockStat == 0) { //close socket if open
303 j = close (rs->socket);
304 if (j > 0) {
305 factPrintf(kFatal, 771, "Closing socket %d failed: %m (close,rc=%d)", sid, errno);
306 } else {
307 factPrintf(kInfo, 771, "Succesfully closed socket %d", sid);
308 }
309 }
310
311 rs->sockStat = 99;
312
313 if (flag < 0) {
314 free (rs->rBuf); //and never open again
315 rs->rBuf = NULL;
316 return 0;
317 }
318
319
320 if (flag == 0) { //generate address and buffer ...
321 rs->Port = port;
322 rs->SockAddr.sin_family = sockAddr->sin_family;
323 rs->SockAddr.sin_port = htons (port);
324 rs->SockAddr.sin_addr = sockAddr->sin_addr;
325
326 rs->rBuf = (CNV_FACT*)malloc (sizeof (CNV_FACT));
327 if (rs->rBuf == NULL) {
328 factPrintf(kFatal, 774, "Could not create local buffer %d (malloc failed)", sid);
329 rs->sockStat = 77;
330 return -3;
331 }
332 }
333
334
335 if ((rs->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) {
336 factPrintf(kFatal, 773, "Generating socket %d failed: %m (socket,rc=%d)", sid, errno);
337 rs->sockStat = 88;
338 return -2;
339 }
340 optval = 1;
341 if (setsockopt (rs->socket, SOL_SOCKET, SO_KEEPALIVE, &optval, optlen) < 0) {
342 factPrintf(kInfo, 173, "Setting SO_KEEPALIVE for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
343 }
344 optval = 10; //start after 10 seconds
345 if (setsockopt (rs->socket, SOL_TCP, TCP_KEEPIDLE, &optval, optlen) < 0) {
346 factPrintf(kInfo, 173, "Setting TCP_KEEPIDLE for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
347 }
348 optval = 10; //do every 10 seconds
349 if (setsockopt (rs->socket, SOL_TCP, TCP_KEEPINTVL, &optval, optlen) < 0) {
350 factPrintf(kInfo, 173, "Setting TCP_KEEPINTVL for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
351 }
352 optval = 2; //close after 2 unsuccessful tries
353 if (setsockopt (rs->socket, SOL_TCP, TCP_KEEPCNT, &optval, optlen) < 0) {
354 factPrintf(kInfo, 173, "Setting TCP_KEEPCNT for socket %d failed: %m (setsockopt,rc=%d)", sid, errno);
355 }
356
357 factPrintf(kInfo, 773, "Successfully generated socket %d", sid);
358
359 rs->sockStat = -1; //try to (re)open socket
360 rs->errCnt = 0;
361 return 0;
362
363} /*-----------------------------------------------------------------*/
364
365 /*-----------------------------------------------------------------*/
366
367
368
369
370int
371mBufInit ()
372{
373// initialize mBuffer (mark all entries as unused\empty)
374
375 //uint32_t actime = g_actTime + 50000000;
376
377 for (int i = 0; i < MAX_EVT * MAX_RUN; i++) {
378 evtCtrl[i].evNum = evtCtrl[i].nRoi = -1;
379 evtCtrl[i].runNum = 0;
380
381 //evtCtrl[i].mBuffer_idx = -1;
382 evtCtrl[i].evtStat = -1;
383 }
384
385 //actRun.FADhead = malloc (NBOARDS * sizeof (PEVNT_HEADER));
386
387 evtCtrl_frstPtr = 0;
388 evtCtrl_lastPtr = 0;
389
390 return 0;
391
392} /*-----------------------------------------------------------------*/
393
394int checkRoiConsistency(int i, int roi[]);
395
396int mBufEvt(int sk)
397{
398// generate a new Event into mBuffer:
399// make sure only complete Event are possible, so 'free' will always work
400// returns index into mBuffer[], or negative value in case of error
401// error: <-9000 if roi screwed up (not consistent with run)
402// <-8000 (not consistent with event)
403// <-7000 (not consistent with board)
404// < 0 if no space left
405
406// int evFree;
407// int headmem = 0;
408// size_t needmem = 0;
409
410 int nRoi[9];
411 if (!checkRoiConsistency(sk, nRoi))
412 return -9999;
413
414 //const int b = sk / 7;
415
416 const int evID = rd[sk].evtID;
417 const uint runID = rd[sk].runID;
418 const int trgTyp = rd[sk].ftmTyp;
419 const int trgNum = rd[sk].ftmID;
420 const int fadNum = rd[sk].evtID;
421
422 int beg = (evtCtrl_lastPtr + MAX_EVT*MAX_RUN - 1) % (MAX_EVT*MAX_RUN);
423 int end = (evtCtrl_frstPtr + MAX_EVT*MAX_RUN - 1) % (MAX_EVT*MAX_RUN);
424
425 for (int k=beg; k!=end; k--, k += MAX_EVT*MAX_RUN, k %= MAX_EVT*MAX_RUN)
426 {
427 // If the run is different, go on searching.
428 // We cannot stop searching if a lower run-id is found as in
429 // the case of the events, because theoretically, there
430 // can be the same run on two different days.
431 if (runID != evtCtrl[k].runNum)
432 continue;
433
434 // The event in the control structure has an event id with
435 // a lower event id than the current one. All previous events
436 // should have an even lower event id, so there is no way it
437 // can be found in the structure.
438 if (evtCtrl[k].evNum < evID/* && runID == evtCtrl[k].runNum*/)
439 break;
440
441 if (evID != evtCtrl[k].evNum/* || runID != evtCtrl[k].runNum*/)
442 continue;
443
444 // is it ok ????
445 if (evtCtrl[k].nRoi != nRoi[0] || evtCtrl[k].nRoiTM != nRoi[8])
446 {
447 factPrintf(kError, 821, "Mismatch of roi within event. Expected roi=%d and roi_tm=%d, got %d and %d.",
448 evtCtrl[k].nRoi, evtCtrl[k].nRoiTM, nRoi[0], nRoi[8]);
449 return -8201;
450 }
451
452 // count for inconsistencies
453 if (evtCtrl[k].trgNum != trgNum)
454 evtCtrl[k].Errors[0]++;
455 if (evtCtrl[k].fadNum != fadNum)
456 evtCtrl[k].Errors[1]++;
457 if (evtCtrl[k].trgTyp != trgTyp)
458 evtCtrl[k].Errors[2]++;
459
460 //everything seems fine so far ==> use this slot ....
461 return k;
462 }
463
464 //event does not yet exist; create it
465
466 if (end-beg==1 || (end==0 && beg==MAX_EVT*MAX_RUN-1)) //no space available in ctrl
467 {
468 factPrintf(kError, 881, "No control slot to keep event %d (run %d) %d %d", evID, runID, beg, end);
469 return -1;
470 }
471
472 // FIXME: This should be the time of the first receiped board
473 struct timeval tv;
474 gettimeofday (&tv, NULL);
475
476 const uint32_t tsec = tv.tv_sec;
477 const uint32_t tusec = tv.tv_usec;
478
479 //check if runId already registered in runCtrl
480
481 uint oldest = g_actTime + 1000;
482 int jold = -1;
483
484 int found = 0;
485
486 // fileId==-2: not yet used or run assigned but not open
487 // fileId== 0: file open
488 // fileId>0: run closed
489
490 for (int k=0; k<MAX_RUN; k++)
491 {
492 // Check if run already registered (old entries should have runId==-1)
493 if (runCtrl[k].runId == runID)
494 {
495 // FIXME: Compare to previous event
496 if (runCtrl[k].roi0 != nRoi[0] || runCtrl[k].roi8 != nRoi[8])
497 {
498 factPrintf(kError, 931, "Mismatch of roi within run. Expected roi=%d and roi_tm=%d, got %d and %d (runID=%d, evID=%d)",
499 runCtrl[k].roi0, runCtrl[k].roi8, nRoi[0], nRoi[8], runID, evID);
500 return -9301;
501 }
502
503 found = 1;
504 break;
505 }
506
507 // This is just for sanity. We use the oldest free entry (until
508 // we have understood the concept and can use "just" a free entry
509 if (runCtrl[k].runId==0 && runCtrl[k].closeTime < oldest)
510 {
511 oldest = runCtrl[k].closeTime;
512 jold = k;
513 }
514 }
515
516 if (!found) // Run not yet registered, register run
517 {
518 if (jold < 0)
519 {
520 factPrintf(kFatal, 883, "Not able to register the new run %d", runID);
521 return -1001;
522 }
523
524 int evFree = jold;
525
526 factPrintf(kInfo, 503, "New run %d (evID=%d, evFree=%d) registered with roi=%d and roi_tm=%d",
527 runID, evID, evFree, nRoi[0], nRoi[8]);
528
529 runCtrl[evFree].runId = runID;
530 runCtrl[evFree].roi0 = nRoi[0]; // FIXME: Make obsolete!
531 runCtrl[evFree].roi8 = nRoi[8]; // FIXME: Make obsolete!
532 runCtrl[evFree].fileId = -2;
533 runCtrl[evFree].procId = -2;
534 runCtrl[evFree].lastEvt = 1; // Number of events partially started to read
535 runCtrl[evFree].actEvt = 0; // Number of written events (write)
536 runCtrl[evFree].procEvt = 0; // Number of successfully checked events (checkEvent)
537 runCtrl[evFree].maxEvt = 999999999; // max number events allowed
538 runCtrl[evFree].lastTime = tsec; // Time when the last event was written
539 runCtrl[evFree].closeTime = tsec + 3600 * 24; //max time allowed
540 }
541
542 const int k = evtCtrl_lastPtr;
543
544 //flag all boards as unused
545 evtCtrl[k].nBoard = 0;
546 for (int b=0; b<NBOARDS; b++)
547 evtCtrl[k].board[b] = -1;
548
549 evtCtrl[k].pcTime[0] = tsec;
550 evtCtrl[k].pcTime[1] = tusec;
551 evtCtrl[k].nRoi = nRoi[0];
552 evtCtrl[k].nRoiTM = nRoi[8];
553 evtCtrl[k].evNum = evID;
554 evtCtrl[k].runNum = runID;
555 evtCtrl[k].fadNum = fadNum;
556 evtCtrl[k].trgNum = trgNum;
557 evtCtrl[k].trgTyp = trgTyp;
558 evtCtrl[k].Errors[0] = 0;
559 evtCtrl[k].Errors[1] = 0;
560 evtCtrl[k].Errors[2] = 0;
561 evtCtrl[k].Errors[3] = 0;
562 evtCtrl[k].fEvent = NULL;
563 evtCtrl[k].FADhead = NULL;
564
565 evtCtrl[k].evtStat = 0;
566
567 // This is dangerous, because theoretically, it can result is
568 // acessing invalid memory in another thread if this is split
569 // in two instructions. Must be done only _after_ the contents
570 // have been initialized
571 evtCtrl_lastPtr = (evtCtrl_lastPtr+1) % MAX_EVT * MAX_RUN;
572
573 return k;
574
575} /*-----------------------------------------------------------------*/
576
577
578void initEvent(int i)
579{
580 evtCtrl[i].fEvent = (EVENT*)((char*)evtCtrl[i].FADhead+MAX_HEAD_MEM);
581 memset(evtCtrl[i].fEvent->Adc_Data, 0, (NPIX+NTMARK)*2*evtCtrl[i].nRoi);
582
583 //flag all pixels as unused
584 for (int k = 0; k < NPIX; k++)
585 evtCtrl[i].fEvent->StartPix[k] = -1;
586
587 //flag all TMark as unused
588 for (int k = 0; k < NTMARK; k++)
589 evtCtrl[i].fEvent->StartTM[k] = -1;
590
591 evtCtrl[i].fEvent->NumBoards = 0;
592 evtCtrl[i].fEvent->SoftTrig = 0;
593 evtCtrl[i].fEvent->PCTime = evtCtrl[i].pcTime[0];
594 evtCtrl[i].fEvent->PCUsec = evtCtrl[i].pcTime[1];
595 evtCtrl[i].fEvent->Roi = evtCtrl[i].nRoi;
596 evtCtrl[i].fEvent->RoiTM = evtCtrl[i].nRoiTM;
597 evtCtrl[i].fEvent->EventNum = evtCtrl[i].evNum;
598 evtCtrl[i].fEvent->TriggerNum = evtCtrl[i].trgNum;
599 evtCtrl[i].fEvent->TriggerType = evtCtrl[i].trgTyp;
600}
601
602
603int
604mBufFree (int i)
605{
606//delete entry [i] from mBuffer:
607//(and make sure multiple calls do no harm ....)
608
609 TGB_free(evtCtrl[i].FADhead);
610
611 evtCtrl[i].fEvent = NULL;
612 evtCtrl[i].FADhead = NULL;
613
614 evtCtrl[i].evNum = evtCtrl[i].nRoi = -1;
615 evtCtrl[i].runNum = 0;
616
617 gj.usdMem = tgb_inuse;
618
619 gj.bufTot--;
620
621 /*if (gi_memStat < 0) {
622 if (gj.usdMem <= 0.75 * gj.maxMem)
623 gi_memStat = +1;
624 }*/
625
626 return 0;
627
628} /*-----------------------------------------------------------------*/
629
630/*
631void
632resetEvtStat ()
633{
634 for (int i = 0; i < MAX_SOCK; i++)
635 gi.numRead[i] = 0;
636
637 for (int i = 0; i < NBOARDS; i++) {
638 gi.gotByte[i] = 0;
639 gi.gotErr[i] = 0;
640
641 }
642
643 gi.evtGet = 0; //#new Start of Events read
644 gi.evtTot = 0; //#complete Events read
645 gi.evtErr = 0; //#Events with Errors
646 gi.evtSkp = 0; //#Events incomplete (timeout)
647
648 gi.procTot = 0; //#Events processed
649 gi.procErr = 0; //#Events showed problem in processing
650 gi.procTrg = 0; //#Events accepted by SW trigger
651 gi.procSkp = 0; //#Events rejected by SW trigger
652
653 gi.feedTot = 0; //#Events used for feedBack system
654 gi.feedErr = 0; //#Events rejected by feedBack
655
656 gi.wrtTot = 0; //#Events written to disk
657 gi.wrtErr = 0; //#Events with write-error
658
659 gi.runOpen = 0; //#Runs opened
660 gi.runClose = 0; //#Runs closed
661 gi.runErr = 0; //#Runs with open/close errors
662
663 return;
664}*/ /*-----------------------------------------------------------------*/
665
666void reportIncomplete(int id)
667{
668 factPrintf(kWarn, 601, "%5d skip incomplete evt %8d",
669 evtCtrl[id].evNum, id);
670
671 uint64_t report = 0;
672
673 char str[1000];
674
675 int ik=0;
676 for (int ib=0; ib<NBOARDS; ib++)
677 {
678 if (ib%10==0)
679 str[ik++] = '|';
680
681 const int jb = evtCtrl[id].board[ib];
682 if (jb>=0) // data received from that board
683 {
684 str[ik++] = '0'+(jb%10);
685 continue;
686 }
687
688 // FIXME: Is that really 'b' or should that be 'ib' ?
689 if (gi_NumConnect[ib]<=0) // board not connected
690 {
691 str[ik++] = 'x';
692 continue;
693 }
694
695 // data from this board lost
696 str[ik++] = '.';
697 report |= ((uint64_t)1)<<ib;
698 }
699
700 str[ik++] = '|';
701 str[ik] = 0;
702
703 factOut(kWarn, 601, str);
704
705 factReportIncomplete(report);
706}
707
708int checkRoiConsistency(int i, int roi[])
709{
710 int xjr = -1;
711 int xkr = -1;
712
713 //points to the very first roi
714 int roiPtr = sizeof(PEVNT_HEADER)/2 + 2;
715
716 roi[0] = ntohs(rd[i].rBuf->S[roiPtr]);
717
718 for (int jr = 0; jr < 9; jr++)
719 {
720 roi[jr] = ntohs(rd[i].rBuf->S[roiPtr]);
721
722 if (roi[jr]<0 || roi[jr]>1024)
723 {
724 factPrintf(kError, 999, "Illegal roi in channel %d (allowed: 0<=roi<=1024)", jr, roi[jr]);
725 return 0;
726 }
727
728 // Check that the roi of pixels jr are compatible with the one of pixel 0
729 if (jr!=8 && roi[jr]!=roi[0])
730 {
731 xjr = jr;
732 break;
733 }
734
735 // Check that the roi of all other DRS chips on boards are compatible
736 for (int kr = 1; kr < 4; kr++)
737 {
738 const int kroi = ntohs(rd[i].rBuf->S[roiPtr]);
739 if (kroi != roi[jr])
740 {
741 xjr = jr;
742 xkr = kr;
743 break;
744 }
745 roiPtr += kroi+4;
746 }
747 }
748
749 if (xjr>=0)
750 {
751 if (xkr<0)
752 factPrintf(kFatal, 1, "Inconsistent Roi accross chips [DRS=%d], expected %d, got %d", xjr, roi[0], roi[xjr]);
753 else
754 factPrintf(kFatal, 1, "Inconsistent Roi accross channels [DRS=%d Ch=%d], expected %d, got %d", xjr, xkr, roi[xjr], ntohs(rd[i].rBuf->S[roiPtr]));
755
756 return 0;
757 }
758
759 //const int b = i / 7;
760
761/*
762 if (roi[0]<0 || roi[0] > 1024)
763 {
764 factPrintf(kError, 999, "Illegal roi in channel 0: %d (allowed: 0<=roi<=1024)", roi[0]);
765 gj.badRoiR++;
766 gj.badRoi[b]++;
767 return 0;
768 }
769 */
770 /*
771 for (int jr = 1; jr < 8; jr++)
772 {
773 if (roi[jr] != roi[0])
774 {
775 factPrintf(kError, 711, "Mismatch of roi (%d) in channel %d with roi (%d) in channel 0.", roi[jr], jr, roi[0]);
776 gj.badRoiB++;
777 gj.badRoi[b]++;
778 return 0;
779 }
780 }
781*/
782 if (roi[8] < roi[0])
783 {
784 factPrintf(kError, 712, "Mismatch of roi (%d) in channel 8. Should be larger or equal than the roi (%d) in channel 0.", roi[8], roi[0]);
785 //gj.badRoiB++;
786 //gj.badRoi[b]++;
787 return 0;
788 }
789
790 return 1;
791}
792
793void swapEventHeaderBytes(int i)
794{
795 //End of the header. to channels now
796 int eStart = 36;
797 for (int ePatchesCount = 0; ePatchesCount<4*9;ePatchesCount++)
798 {
799 rd[i].rBuf->S[eStart+0] = ntohs(rd[i].rBuf->S[eStart+0]);//id
800 rd[i].rBuf->S[eStart+1] = ntohs(rd[i].rBuf->S[eStart+1]);//start_cell
801 rd[i].rBuf->S[eStart+2] = ntohs(rd[i].rBuf->S[eStart+2]);//roi
802 rd[i].rBuf->S[eStart+3] = ntohs(rd[i].rBuf->S[eStart+3]);//filling
803
804 eStart += 4+rd[i].rBuf->S[eStart+2];//skip the pixel data
805 }
806}
807
808void copyData(int i, int evID, /*int roi,*/ int boardId)
809{
810 swapEventHeaderBytes(i);
811
812 memcpy(&evtCtrl[evID].FADhead[boardId].start_package_flag,
813 &rd[i].rBuf->S[0], sizeof(PEVNT_HEADER));
814
815 int src = sizeof(PEVNT_HEADER) / 2;
816
817 // consistency of ROIs have been checked already (is it all correct?)
818 const int roi = rd[i].rBuf->S[src+2];
819
820 // different sort in FAD board.....
821 for (int px = 0; px < 9; px++)
822 {
823 for (int drs = 0; drs < 4; drs++)
824 {
825 // pixH = rd[i].rBuf->S[src++]; // ID
826 src++;
827
828 const int pixC = rd[i].rBuf->S[src++]; // start-cell
829 const int pixR = rd[i].rBuf->S[src++]; // roi
830 //here we should check if pixH is correct ....
831
832 const int pixS = boardId * 36 + drs * 9 + px;
833 src++;
834
835 evtCtrl[evID].fEvent->StartPix[pixS] = pixC;
836
837 const int dest1 = pixS * roi;
838 memcpy(&evtCtrl[evID].fEvent->Adc_Data[dest1],
839 &rd[i].rBuf->S[src], roi * 2);
840
841 src += pixR;
842
843 if (px == 8)
844 {
845 const int tmS = boardId * 4 + drs;
846
847 //and we have additional TM info
848 if (pixR > roi)
849 {
850 const int dest2 = tmS * roi + NPIX * roi;
851
852 const int srcT = src - roi;
853 evtCtrl[evID].fEvent->StartTM[tmS] = (pixC + pixR - roi) % 1024;
854
855 memcpy(&evtCtrl[evID].fEvent->Adc_Data[dest2],
856 &rd[i].rBuf->S[srcT], roi * 2);
857 }
858 else
859 {
860 evtCtrl[evID].fEvent->StartTM[tmS] = -1;
861
862 //ETIENNE because the TM channels are always processed during drs calib,
863 //set them to zero if they are not present
864 //I suspect that it may be more efficient to set all the allocated mem to
865 //zero when allocating it
866 // dest = tmS*roi[0] + NPIX*roi[0];
867 // bzero(&mBuffer[evID].fEvent->Adc_Data[dest],roi[0]*2);
868 }
869 }
870 }
871 }
872}
873
874
875void
876initReadFAD ()
877{
878 return;
879} /*-----------------------------------------------------------------*/
880
881//struct rnd
882//{
883// int val;
884// int idx;
885//};
886//
887//struct rnd random_arr[MAX_SOCK];
888//
889//int compare(const void *f1, const void *f2)
890//{
891// struct rnd *r1 = (struct rnd*)f1;
892// struct rnd *r2 = (struct rnd*)f2;
893// return r1->val - r2->val;
894//}
895
896
897void *readFAD (void *ptr)
898{
899/* *** main loop reading FAD data and sorting them to complete events */
900
901 factPrintf(kInfo, -1, "Start initializing (readFAD)");
902
903// int cpu = 7; //read thread
904// cpu_set_t mask;
905
906/* CPU_ZERO initializes all the bits in the mask to zero. */
907// CPU_ZERO (&mask);
908/* CPU_SET sets only the bit corresponding to cpu. */
909// cpu = 7;
910// CPU_SET (cpu, &mask);
911
912/* sched_setaffinity returns 0 in success */
913// if (sched_setaffinity (0, sizeof (mask), &mask) == -1) {
914// snprintf (str, MXSTR, "W ---> can not create affinity to %d", cpu);
915// factOut (kWarn, -1, str);
916// }
917
918
919 const int minLen = sizeof(PEVNT_HEADER); //min #bytes needed to check header: full header for debug
920
921 start.S = 0xFB01;
922 stop.S = 0x04FE;
923
924/* initialize run control logics */
925 for (int i = 0; i < MAX_RUN; i++) {
926 runCtrl[i].runId = 0;
927 runCtrl[i].fileId = -2;
928 runCtrl[i].procId = -2;
929 }
930 gi_resetS = gi_resetR = 9;
931
932 int sockDef[NBOARDS]; //internal state of sockets
933 memset(sockDef, 0, NBOARDS*sizeof(int));
934
935 START:
936 evtCtrl_frstPtr = 0;
937 evtCtrl_lastPtr = 0;
938
939 //time in seconds
940 uint gi_SecTime = time(NULL);;
941
942 const int cntsock = 8 - NUMSOCK ;
943
944 if (gi_resetS > 0) {
945 //make sure all sockets are preallocated as 'not exist'
946 for (int i = 0; i < MAX_SOCK; i++) {
947 rd[i].socket = -1;
948 rd[i].sockStat = 99;
949 }
950
951 for (int k = 0; k < NBOARDS; k++) {
952 gi_NumConnect[k] = 0;
953 //gi.numConn[k] = 0;
954 gj.numConn[k] = 0;
955 //gj.errConn[k] = 0;
956 gj.rateBytes[k] = 0;
957 gj.totBytes[k] = 0;
958 }
959
960 }
961
962
963 if (gi_resetR > 0) {
964 //resetEvtStat ();
965 gj.bufTot = gj.maxEvt = gj.xxxEvt = 0;
966 gj.usdMem = gj.maxMem = gj.xxxMem = 0;
967 gj.totMem = tgb_memory;
968 gj.bufNew = gj.bufEvt = 0;
969 //gj.badRoiE = gj.badRoiR = gj.badRoiB = 0;
970 gj.evtSkip = gj.evtWrite = gj.evtErr = 0;
971
972 //for (int b = 0; b < NBOARDS; b++)
973 // gj.badRoi[b] = 0;
974
975 mBufInit (); //initialize buffers
976
977 factPrintf(kInfo, -1, "End initializing (readFAD)");
978 }
979
980
981 gi_reset = gi_resetR = gi_resetS = gi_resetW = 0;
982
983 //loop until global variable g_runStat claims stop
984 while (g_runStat >= 0 && g_reset == 0)
985 {
986 gi_runStat = g_runStat;
987 gj.readStat = g_runStat;
988
989 struct timeval tv;
990 gettimeofday (&tv, NULL);
991 g_actTime = tv.tv_sec;
992 g_actUsec = tv.tv_usec;
993
994
995 for (int b = 0; b < NBOARDS; b++)
996 {
997 // Nothing has changed
998 if (g_port[b].sockDef == sockDef[b])
999 continue;
1000
1001 gi_NumConnect[b] = 0; //must close all connections
1002 //gi.numConn[b] = 0;
1003 gj.numConn[b] = 0;
1004
1005 // s0 = 0: sockets to be defined and opened
1006 // s0 = -1: sockets to be destroyed
1007 // s0 = +1: sockets to be closed and reopened
1008
1009 int s0 = 0;
1010 if (sockDef[b] != 0)
1011 s0 = g_port[b].sockDef==0 ? -1 : +1;
1012
1013 const int p0 = s0==0 ? ntohs (g_port[b].sockAddr.sin_port) : 0;
1014
1015 int k = b * 7;
1016 for (int p = p0 + 1; p < p0 + 8; p++, k++)
1017 GenSock (s0, k, p, &g_port[b].sockAddr, &rd[k]); //generate address and socket
1018
1019 sockDef[b] = g_port[b].sockDef;
1020 }
1021
1022 // count the number of active boards
1023 int actBoards = 0;
1024 for (int b = 0; b < NBOARDS; b++)
1025 if (sockDef[b] > 0)
1026 actBoards++;
1027
1028 //count number of succesfull actions
1029// int numok = 0;
1030
1031/*
1032 for (i=0; i<MAX_SOCK/7; i++)
1033 {
1034 random_arr[i].val = rand();
1035 random_arr[i].idx = i;
1036 }
1037
1038 qsort(random_arr, MAX_SOCK/7, sizeof(struct rnd), compare);
1039
1040 for (int iii = 0; iii < MAX_SOCK/7; iii++) { //check all sockets if something to read
1041
1042 b = random_arr[iii].idx;
1043 i = b*7;
1044 p = 0;
1045 */
1046
1047 //check all sockets if something to read
1048 for (int i = 0; i < MAX_SOCK; i+=7)
1049 {
1050 // Do not try to connect this socket
1051 if (rd[i].sockStat > 0)
1052 continue;
1053
1054 const int board = i / 7 ;
1055 //const int p = i % 7 ;
1056
1057 if (rd[i].sockStat == -1)
1058 {
1059 //try to connect if not yet done
1060 rd[i].sockStat = connect (rd[i].socket,
1061 (struct sockaddr *) &rd[i].SockAddr,
1062 sizeof (rd[i].SockAddr));
1063 // Failed
1064 if (rd[i].sockStat == -1)
1065 {
1066 rd[i].errCnt++;
1067 usleep(25000);
1068 continue;
1069 }
1070
1071 // Success (rd[i].sockStat == 0)
1072
1073 if (sockDef[board] > 0)
1074 {
1075 rd[i].bufTyp = 0; // expect a header
1076 rd[i].bufLen = sizeof(PEVNT_HEADER); // max size to read at begining
1077 }
1078 else
1079 {
1080 rd[i].bufTyp = -1; // full data to be skipped
1081 rd[i].bufLen = MAX_LEN; // huge for skipping
1082 }
1083
1084 rd[i].bufPos = 0; // no byte read so far
1085 rd[i].skip = 0; // start empty
1086
1087 gi_NumConnect[board] += cntsock;
1088
1089 //gi.numConn[b]++;
1090 gj.numConn[board]++;
1091
1092 factPrintf(kInfo, -1, "New connection %d (number of connections: %d)", board, gj.numConn[board]);
1093 }
1094
1095 // Do not read from this socket
1096 if (rd[i].bufLen<0)
1097 continue;
1098
1099 //numok++;
1100
1101 if (rd[i].bufLen>0)
1102 {
1103 const int32_t jrd =
1104 recv(rd[i].socket, &rd[i].rBuf->B[rd[i].bufPos],
1105 rd[i].bufLen, MSG_DONTWAIT);
1106
1107 // recv failed
1108 if (jrd<0)
1109 {
1110 // There was just nothing waiting
1111 if (errno==EWOULDBLOCK || errno==EAGAIN)
1112 {
1113 //numok--;
1114 continue;
1115 }
1116
1117 factPrintf(kError, 442, "Reading from socket %d failed: %m (recv,rc=%d)", i, errno);
1118 //gi.gotErr[b]++;
1119 continue;
1120 }
1121
1122 // connection was closed ...
1123 if (jrd==0)
1124 {
1125 factPrintf(kInfo, 441, "Socket %d closed by FAD", i);
1126
1127 const int s0 = sockDef[board] > 0 ? +1 : -1;
1128 GenSock(s0, i, 0, NULL, &rd[i]);
1129
1130 //gi.gotErr[b]++;
1131
1132 gi_NumConnect[board]-= cntsock ;
1133 //gi.numConn[b]--;
1134 gj.numConn[board]--;
1135
1136 continue;
1137 }
1138 // Success (jrd > 0)
1139
1140 gj.rateBytes[board] += jrd;
1141
1142 // are we skipping this board ...
1143 if (rd[i].bufTyp < 0)
1144 continue;
1145
1146 rd[i].bufPos += jrd; //==> prepare for continuation
1147 rd[i].bufLen -= jrd;
1148
1149#ifdef EVTDEBUG
1150 debugRead(i, jrd, rd[i].evtID, rd[i].ftmID, rd[i].runID, rd[i].bufTyp, tv.tv_sec, tv.tv_usec); // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
1151#endif
1152 }
1153
1154 //we are reading event header
1155 if (rd[i].bufTyp <= 0)
1156 {
1157 //not yet sufficient data to take action
1158 if (rd[i].bufPos < minLen)
1159 continue;
1160
1161 //check if startflag correct; else shift block ....
1162 // FIXME: This is not enough... this combination of
1163 // bytes can be anywhere... at least the end bytes
1164 // must be checked somewhere, too.
1165 int k;
1166 for (k = 0; k < rd[i].bufPos - 1; k++)
1167 {
1168 if (rd[i].rBuf->B[k] == start.B[1] && rd[i].rBuf->B[k+1] == start.B[0])
1169 break;
1170 }
1171 rd[i].skip += k;
1172
1173 //no start of header found
1174 if (k >= rd[i].bufPos - 1)
1175 {
1176 rd[i].bufPos = 0;
1177 rd[i].bufLen = sizeof(PEVNT_HEADER);
1178 continue;
1179 }
1180
1181 if (k > 0)
1182 {
1183 rd[i].bufPos -= k;
1184 rd[i].bufLen += k;
1185 memmove (&rd[i].rBuf->B[0], &rd[i].rBuf->B[k],
1186 rd[i].bufPos);
1187 }
1188
1189 if (rd[i].bufPos < minLen)
1190 continue;
1191
1192 if (rd[i].skip > 0)
1193 {
1194 factPrintf(kInfo, 666, "Skipped %d bytes on port %d", rd[i].skip, i);
1195 rd[i].skip = 0;
1196 }
1197
1198 // TGB: This needs much more checks than just the first two bytes!
1199
1200 // Swap everything except start_package_flag.
1201 // It is to difficult to find out where it is used how,
1202 // but it doesn't really matter because it is not really
1203 // used anywehere else
1204 // rd[i].rBuf->S[1] = ntohs(rd[i].rBuf->S[1]); // package_length
1205 rd[i].rBuf->S[2] = ntohs(rd[i].rBuf->S[2]); // version_no
1206 rd[i].rBuf->S[3] = ntohs(rd[i].rBuf->S[3]); // PLLLCK
1207 rd[i].rBuf->S[4] = ntohs(rd[i].rBuf->S[4]); // trigger_crc
1208 rd[i].rBuf->S[5] = ntohs(rd[i].rBuf->S[5]); // trigger_type
1209
1210 rd[i].rBuf->S[12] = ntohs(rd[i].rBuf->S[12]); // board id
1211 rd[i].rBuf->S[13] = ntohs(rd[i].rBuf->S[13]); // adc_clock_phase_shift
1212 rd[i].rBuf->S[14] = ntohs(rd[i].rBuf->S[14]); // number_of_triggers_to_generate
1213 rd[i].rBuf->S[15] = ntohs(rd[i].rBuf->S[15]); // trigger_generator_prescaler
1214
1215 rd[i].rBuf->I[3] = ntohl(rd[i].rBuf->I[3]); // trigger_id
1216 rd[i].rBuf->I[4] = ntohl(rd[i].rBuf->I[4]); // fad_evt_counter
1217 rd[i].rBuf->I[5] = ntohl(rd[i].rBuf->I[5]); // REFCLK_frequency
1218
1219 rd[i].rBuf->I[10] = ntohl(rd[i].rBuf->I[10]); // runnumber;
1220 rd[i].rBuf->I[11] = ntohl(rd[i].rBuf->I[11]); // time;
1221
1222 for (int s=24;s<24+NTemp+NDAC;s++)
1223 rd[i].rBuf->S[s] = ntohs(rd[i].rBuf->S[s]); // drs_temperature / dac
1224
1225 rd[i].fadLen = ntohs(rd[i].rBuf->S[1]) * 2;
1226 rd[i].fadVers = rd[i].rBuf->S[2];
1227 rd[i].ftmTyp = rd[i].rBuf->S[5];
1228 rd[i].ftmID = rd[i].rBuf->I[3]; //(FTMevt)
1229 rd[i].evtID = rd[i].rBuf->I[4]; //(FADevt)
1230 rd[i].runID = rd[i].rBuf->I[11]==0 ? (int)g_actTime : rd[i].rBuf->I[11];
1231 rd[i].bufTyp = 1; //ready to read full record
1232 rd[i].bufLen = rd[i].fadLen - rd[i].bufPos;
1233
1234 const int fadBoard = rd[i].rBuf->S[12];
1235 debugHead(i, fadBoard, rd[i].rBuf);
1236
1237 continue;
1238 }
1239
1240 // are we reading data
1241
1242 // not yet all read
1243 if (rd[i].bufLen > 0)
1244 continue;
1245
1246 if (rd[i].rBuf->B[rd[i].fadLen - 1] != stop.B[0] ||
1247 rd[i].rBuf->B[rd[i].fadLen - 2] != stop.B[1])
1248 {
1249 //gi.evtErr++;
1250 factPrintf(kError, 301, "End-of-event flag wrong on socket %3d for event %4d (len=%5d), expected %3d %3d, got %3d %3d",
1251 i, rd[i].evtID, rd[i].fadLen, stop.B[0], stop.B[1],
1252 rd[i].rBuf->B[rd[i].fadLen - 1], rd[i].rBuf->B[rd[i].fadLen - 2]);
1253
1254 // ready to read next header
1255 rd[i].bufTyp = 0;
1256 rd[i].bufLen = sizeof(PEVNT_HEADER);
1257 rd[i].bufPos = 0;
1258
1259 continue;
1260 }
1261
1262 // int actid;
1263 // if (g_useFTM > 0)
1264 // actid = rd[i].evtID;
1265 // else
1266 // actid = rd[i].ftmID;
1267
1268 //get index into mBuffer for this event (create if needed)
1269 const int idx = mBufEvt(i);
1270
1271 // no free entry in mBuffer, retry later
1272 if (idx == -1)
1273 continue;
1274
1275 // We have a valid entry, but no memory has yet been allocated
1276 if (idx >= 0 && evtCtrl[idx].FADhead == NULL)
1277 {
1278 // Try to get memory from the big buffer
1279 evtCtrl[idx].FADhead = (PEVNT_HEADER*)TGB_Malloc();
1280 if (evtCtrl[idx].FADhead == NULL)
1281 {
1282 // If this works properly, this is a hack which can be removed, or
1283 // replaced by a signal or dim message
1284 if (rd[i].bufTyp==2)
1285 factPrintf(kError, 882, "malloc failed for event %d (run=%d)", evtCtrl[idx].evNum, evtCtrl[idx].runNum);
1286 rd[i].bufTyp = 2;
1287 continue;
1288 }
1289
1290 // Initialise contents of mBuffer[evID]->fEvent
1291 initEvent(idx);
1292
1293 // Some statistics
1294 gj.usdMem = tgb_inuse;
1295
1296 if (gj.usdMem > gj.maxMem)
1297 gj.maxMem = gj.usdMem;
1298
1299 gj.rateNew++;
1300 gj.bufTot++;
1301 if (gj.bufTot > gj.maxEvt)
1302 gj.maxEvt = gj.bufTot;
1303 }
1304
1305 // ready to read next header
1306 rd[i].bufTyp = 0;
1307 rd[i].bufLen = sizeof(PEVNT_HEADER);
1308 rd[i].bufPos = 0;
1309
1310 // Fatal error occured. Event cannot be processed. Skip it. Start reading next header.
1311 if (idx < -1000)
1312 continue;
1313
1314 //we have a valid entry in mBuffer[]; fill it
1315 const int fadBoard = rd[i].rBuf->S[12];
1316 const int fadCrate = fadBoard>>8;
1317
1318 if (board != (fadCrate * 10 + (fadBoard&0xff)))
1319 {
1320 factPrintf(kWarn, 301, "Board ID mismatch. Expected %d, got %d (C=%d, B=%d)",
1321 board, fadBoard, fadCrate, fadBoard&0xff);
1322 }
1323
1324 if (evtCtrl[idx].board[board] != -1)
1325 {
1326 factPrintf(kWarn, 501, "Got event %5d from board %3d (i=%3d, len=%5d) twice: Starts with %3d %3d - ends with %3d %3d",
1327 evtCtrl[idx].evNum, board, i, rd[i].fadLen,
1328 rd[i].rBuf->B[0], rd[i].rBuf->B[1],
1329 rd[i].rBuf->B[rd[i].fadLen - 2],
1330 rd[i].rBuf->B[rd[i].fadLen - 1]);
1331 continue; // Continue reading next header
1332 }
1333
1334 // Copy data from rd[i] to mBuffer[evID]
1335 copyData(i, idx, board);
1336
1337 // now we have stored a new board contents into Event structure
1338
1339 evtCtrl[idx].fEvent->NumBoards++;
1340 evtCtrl[idx].board[board] = board;
1341 evtCtrl[idx].nBoard++;
1342 evtCtrl[idx].evtStat = evtCtrl[idx].nBoard;
1343
1344 // have we already reported first (partial) event of this run ???
1345 if (evtCtrl[idx].nBoard==1 && evtCtrl[idx].runNum != actrun)
1346 {
1347 // Signal the fadctrl that a new run has been started
1348 gotNewRun(evtCtrl[idx].runNum, NULL);
1349
1350 factPrintf(kInfo, 1, "gotNewRun called, prev run %d, new run %d, event %d",
1351 actrun, evtCtrl[idx].runNum, evtCtrl[idx].evNum);
1352
1353 for (int j=0; j<MAX_RUN; j++)
1354 {
1355 // Since we have started a new run, we know already when to close the
1356 // previous run in terms of number of events
1357 if (runCtrl[j].runId==actrun)
1358 runCtrl[j].maxEvt = runCtrl[j].lastEvt;
1359
1360 // We got the first part of this event, so this is
1361 // the number of events we expect for this run
1362 if (runCtrl[j].runId==evtCtrl[idx].runNum)
1363 runCtrl[j].lastEvt++;
1364 }
1365
1366 // Change 'actrun' the the new runnumber
1367 actrun = evtCtrl[idx].runNum;
1368 }
1369
1370 // event not yet complete
1371 if (evtCtrl[idx].nBoard < actBoards)
1372 continue;
1373
1374 // This is a non-ideal hack to lower the probability that
1375 // in mBufEvt the search for correct entry in runCtrl
1376 // will not return a super-old entry
1377 for (int ir=0; ir<MAX_RUN; ir++)
1378 {
1379 if (runCtrl[ir].runId != actrun && runCtrl[ir].fileId>0)
1380 runCtrl[ir].runId = 0;
1381 }
1382
1383 // Flag that the event is ready for processing
1384 evtCtrl[idx].evtStat = 99;
1385
1386 } // end for loop over all sockets
1387
1388 g_actTime = time (NULL);
1389 if (g_actTime <= gi_SecTime)
1390 {
1391 usleep(1);
1392 continue;
1393 }
1394
1395 gi_SecTime = g_actTime;
1396
1397 gj.bufNew = gj.bufEvt = 0;
1398
1399 //loop over all active events and flag those older than read-timeout
1400 //delete those that are written to disk ....
1401 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
1402 {
1403 // Check the more likely case first: incomplete events
1404 if (evtCtrl[k0].evtStat>0 && evtCtrl[k0].evtStat<92)
1405 {
1406 gj.bufNew++; //incomplete event in Buffer
1407
1408 // Event has not yet timed out or was reported already
1409 if (evtCtrl[k0].evtStat>=90 || evtCtrl[k0].pcTime[0]/*evtCtrl[k0].lastRecv*/>=g_actTime - 30)
1410 continue;
1411
1412 reportIncomplete(k0);
1413
1414 //timeout for incomplete events
1415 evtCtrl[k0].evtStat = 91;
1416 gj.evtSkip++;
1417
1418 continue;
1419 }
1420
1421 // complete event in Buffer
1422 if (evtCtrl[k0].evtStat >= 95)
1423 gj.bufEvt++;
1424
1425 // Check the less likely case: 'useless' or 'delete'
1426 if (evtCtrl[k0].evtStat==0 || evtCtrl[k0].evtStat >= 9000)
1427 {
1428 mBufFree(k0); //event written--> free memory
1429 evtCtrl[k0].evtStat = -1;
1430
1431 if (k0==evtCtrl_frstPtr)
1432 {
1433 evtCtrl_frstPtr++;
1434 evtCtrl_frstPtr %= MAX_EVT * MAX_RUN;
1435 }
1436 else
1437 factPrintf(kDebug, -1, "Freed a non-first slot");
1438
1439 gj.evtWrite++;
1440 gj.rateWrite++;
1441 }
1442
1443 }
1444
1445 gj.deltaT = 1000; //temporary, must be improved
1446
1447 for (int ib = 0; ib < NBOARDS; ib++)
1448 gj.totBytes[ib] += gj.rateBytes[ib];
1449
1450 gj.totMem = tgb_memory;
1451
1452 if (gj.maxMem > gj.xxxMem)
1453 gj.xxxMem = gj.maxMem;
1454 if (gj.maxEvt > gj.xxxEvt)
1455 gj.xxxEvt = gj.maxEvt;
1456
1457 factStat (gj);
1458 factStatNew (gi);
1459 gj.rateNew = gj.rateWrite = 0;
1460 gj.maxMem = gj.usdMem;
1461 gj.maxEvt = gj.bufTot;
1462 for (int b = 0; b < NBOARDS; b++)
1463 gj.rateBytes[b] = 0;
1464
1465 } // while (g_runStat >= 0 && g_reset == 0)
1466
1467 factPrintf(kInfo, -1, "Stop reading ... RESET=%d", g_reset);
1468
1469 if (g_reset > 0)
1470 {
1471 gi_reset = g_reset;
1472 gi_resetR = gi_reset % 10; //shall we stop reading ?
1473 gi_resetS = (gi_reset / 10) % 10; //shall we close sockets ?
1474 gi_resetW = (gi_reset / 100) % 10; //shall we close files ?
1475 gi_resetX = gi_reset / 1000; //shall we simply wait resetX seconds ?
1476 g_reset = 0;
1477 }
1478 else
1479 {
1480 gi_reset = 0;
1481 gi_resetR = g_runStat == -1 ? 1 : 7;
1482
1483 gi_resetS = 7; //close all sockets
1484 gi_resetW = 7; //close all files
1485 gi_resetX = 0;
1486
1487 //inform others we have to quit ....
1488 gi_runStat = -11; //inform all that no update to happen any more
1489 gj.readStat = -11; //inform all that no update to happen any more
1490 }
1491
1492 if (gi_resetS > 0)
1493 {
1494 //must close all open sockets ...
1495 factPrintf(kInfo, -1, "Close all sockets...");
1496
1497 for (int i = 0; i < MAX_SOCK; i++)
1498 {
1499 if (rd[i].sockStat != 0)
1500 continue;
1501
1502 GenSock(-1, i, 0, NULL, &rd[i]); //close and destroy open socket
1503 if (i%7)
1504 continue;
1505
1506 gi_NumConnect[i / 7]-= cntsock ;
1507 //gi.numConn[i / 7]--;
1508 gj.numConn[i / 7]--;
1509 sockDef[i / 7] = 0; //flag ro recreate the sockets ...
1510 rd[i / 7].sockStat = -1; //and try to open asap
1511 }
1512 }
1513
1514
1515 if (gi_resetR > 0)
1516 {
1517 //flag all events as 'read finished'
1518 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
1519 {
1520 if (evtCtrl[k0].evtStat > 0 && evtCtrl[k0].evtStat < 90)
1521 {
1522 evtCtrl[k0].evtStat = 91;
1523 //gi.evtSkp++;
1524 //gi.evtTot++;
1525 }
1526 }
1527
1528 //and clear all buffers (might have to wait until all others are done)
1529 // If minclear is 0, an event could be deleted while writing is still ongoing
1530 /*
1531 int minclear;
1532 if (gi_resetR == 1) {
1533 minclear = 900;
1534 factPrintf(kInfo, -1, "Drain all buffers ...");
1535 } else {
1536 minclear = 0;
1537 factPrintf(kInfo, -1, "Flush all buffers ...");
1538 }*/
1539 const int minclear = 900;
1540
1541 int numclear = 1;
1542 while (numclear > 0)
1543 {
1544 numclear = 0;
1545
1546 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
1547 {
1548 if (evtCtrl[k0].evtStat > minclear)
1549 {
1550 mBufFree(k0); //event written--> free memory
1551 evtCtrl[k0].evtStat = -1;
1552
1553 if (k0==evtCtrl_frstPtr)
1554 {
1555 evtCtrl_frstPtr++;
1556 evtCtrl_frstPtr %= MAX_EVT * MAX_RUN;
1557 }
1558 else
1559 factPrintf(kDebug, -1, "Freed a non-first slot");
1560 }
1561 else
1562 if (evtCtrl[k0].evtStat > 0)
1563 numclear++; //writing is still ongoing...
1564 }
1565
1566 usleep(1);
1567 }
1568 }
1569
1570 if (gi_reset > 0)
1571 {
1572 if (gi_resetW > 0)
1573 CloseRunFile (0, 0, 0); //ask all Runs to be closed
1574
1575 if (gi_resetX > 0)
1576 {
1577 struct timespec xwait;
1578 xwait.tv_sec = gi_resetX;
1579 xwait.tv_nsec = 0;
1580 nanosleep (&xwait, NULL);
1581 }
1582
1583 factPrintf(kInfo, -1, "Continue read Process ...");
1584 gi_reset = 0;
1585 goto START;
1586 }
1587
1588 factPrintf(kInfo, -1, "Exit read Process...");
1589
1590 factPrintf(kInfo, -1, "%ld Bytes flaged as in-use.", tgb_inuse);
1591
1592 gi_runStat = -99;
1593 gj.readStat = -99;
1594
1595 factStat (gj);
1596 factStatNew (gi);
1597
1598 return 0;
1599
1600} /*-----------------------------------------------------------------*/
1601
1602
1603void *subProc(void *thrid)
1604{
1605 const int64_t threadID = (int64_t)thrid;
1606
1607 factPrintf(kInfo, -1, "Starting sub-process-thread %ld", threadID);
1608
1609 while (g_runStat > -2) //in case of 'exit' we still must process pending events
1610 {
1611 int numWait = 0;
1612 int numProc = 0;
1613
1614 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
1615 {
1616 if (evtCtrl[k0].evtStat != 1000 + threadID)
1617 {
1618 if (evtCtrl[k0].evtStat < 1000 + threadID)
1619 numWait++;
1620
1621 continue;
1622 }
1623
1624 int jret = 9100; // flag to be deleted (gi_resetR>1 : flush buffers asap)
1625
1626 if (gi_resetR<=1)
1627 {
1628 jret = subProcEvt(threadID, evtCtrl[k0].FADhead,
1629 evtCtrl[k0].fEvent, NULL/*mBuffer[id].buffer*/);
1630
1631
1632 if (jret <= threadID) {
1633 factPrintf(kError, -1, "Process %ld wants to send event to process %d... not allowed.", threadID, jret);
1634 jret = 5300;
1635 } else if (jret <= 0)
1636 jret = 9200 + threadID; // flag as 'to be deleted'
1637 else if (jret >= gi_maxProc)
1638 jret = 5200 + threadID; // flag as 'to be written'
1639 else
1640 jret = 1000 + jret; // flag for next proces
1641 }
1642
1643 evtCtrl[k0].evtStat = jret;
1644 numProc++;
1645 }
1646
1647 if (gj.readStat < -10 && numWait == 0) { //nothing left to do
1648 factPrintf(kInfo, -1, "Exit subProcessing in process %ld", threadID);
1649 return 0;
1650 }
1651
1652 //seems we have nothing to do, so sleep a little
1653 if (numProc == 0)
1654 usleep(1);
1655 }
1656
1657 factPrintf(kInfo, -1, "Ending sub-process-thread %ld", threadID);
1658
1659 return 0;
1660}
1661
1662/*-----------------------------------------------------------------*/
1663
1664
1665void *
1666procEvt (void *ptr)
1667{
1668/* *** main loop processing file, including SW-trigger */
1669 int status;
1670
1671 int lastRun = 0; //usually run from last event still valid
1672
1673// cpu_set_t mask;
1674// int cpu = 1; //process thread (will be several in final version)
1675
1676 factPrintf(kInfo, -1, "Starting process-thread with %d subprocesses", gi_maxProc);
1677
1678/* CPU_ZERO initializes all the bits in the mask to zero. */
1679// CPU_ZERO (&mask);
1680/* CPU_SET sets only the bit corresponding to cpu. */
1681// CPU_SET( 0 , &mask ); leave for system
1682// CPU_SET( 1 , &mask ); used by write process
1683// CPU_SET (2, &mask);
1684// CPU_SET (3, &mask);
1685// CPU_SET (4, &mask);
1686// CPU_SET (5, &mask);
1687// CPU_SET (6, &mask);
1688// CPU_SET( 7 , &mask ); used by read process
1689/* sched_setaffinity returns 0 in success */
1690// if (sched_setaffinity (0, sizeof (mask), &mask) == -1) {
1691// snprintf (str, MXSTR, "P ---> can not create affinity to %d", cpu);
1692// factOut (kWarn, -1, str);
1693// }
1694
1695
1696 pthread_t thread[100];
1697// int th_ret[100];
1698
1699 for (long long k = 0; k < gi_maxProc; k++) {
1700 /*th_ret[k] =*/ pthread_create (&thread[k], NULL, subProc, (void *) k);
1701 }
1702
1703 // in case of 'exit' we still must process pending events
1704 while (g_runStat > -2)
1705 {
1706 int numWait = 0;
1707 int numProc = 0;
1708
1709 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
1710 {
1711 if (evtCtrl[k0].evtStat <= 90 || evtCtrl[k0].evtStat >= 1000)
1712 {
1713 if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat< 90)
1714 numWait++;
1715
1716 continue;
1717 }
1718
1719 //we are asked to flush buffers asap
1720 if (gi_resetR > 1)
1721 {
1722 evtCtrl[k0].evtStat = 9991;
1723 continue;
1724 }
1725
1726 //-------- it is better to open the run already here, so call can be used to initialize
1727 //-------- buffers etc. needed to interprete run (e.g. DRS calibration)
1728 const uint32_t irun = evtCtrl[k0].runNum;
1729 const int32_t ievt = evtCtrl[k0].evNum;
1730
1731 // Find entry in runCtrl which belongs to the event mBuffer[id]
1732 // (only check if there is a need to check)
1733 if (runCtrl[lastRun].runId != irun)
1734 {
1735 //check which fileID to use (or open if needed)
1736 int j;
1737 for (j=0;j<MAX_RUN; j++)
1738 if (runCtrl[j].runId == irun)
1739 break;
1740
1741 if (j>=MAX_RUN)
1742 {
1743 factPrintf(kFatal, 901, "writeEvt: Can not find run %d for event %d in %d", irun, ievt, k0);
1744 // FIXME: What is the right action? (Flag event for deletion?)
1745 continue;
1746 }
1747
1748 lastRun = j;
1749 }
1750
1751 // File not yet open
1752 if (runCtrl[lastRun].fileId < 0)
1753 {
1754 //---- we need to open a new run ==> make sure all older runs are
1755 //---- finished and marked to be closed ....
1756 // This loop is unique to procEvt
1757 for (int j=0; j<MAX_RUN; j++)
1758 {
1759 if (runCtrl[j].fileId == 0)
1760 {
1761 runCtrl[j].procId = 2; //--> do no longer accept events for processing
1762
1763 //---- problem: processing still going on ==> must wait for closing ....
1764 factPrintf(kInfo, -1, "procEvt: Finished run since new one opened %d", runCtrl[j].runId);
1765 runFinish1(runCtrl[j].runId);
1766 }
1767 }
1768
1769 RUN_HEAD actRun;
1770 actRun.Version = 1;
1771 actRun.RunType = -1; //to be adapted
1772 actRun.Nroi = evtCtrl[k0].nRoi; //runCtrl[lastRun].roi0;
1773 actRun.NroiTM = evtCtrl[k0].nRoiTM; //runCtrl[lastRun].roi8;
1774 actRun.RunTime = evtCtrl[k0].pcTime[0]; //runCtrl[lastRun].firstTime;
1775 actRun.RunUsec = evtCtrl[k0].pcTime[1]; //runCtrl[lastRun].firstUsec;
1776 actRun.NBoard = NBOARDS;
1777 actRun.NPix = NPIX;
1778 actRun.NTm = NTMARK;
1779
1780 memcpy(actRun.FADhead, evtCtrl[k0].FADhead, NBOARDS*sizeof(PEVNT_HEADER));
1781
1782 runCtrl[lastRun].fileHd = runOpen (irun, &actRun, sizeof (actRun));
1783 if (runCtrl[lastRun].fileHd == NULL)
1784 {
1785 factPrintf(kError, 502, "procEvt: Could not open a file for run %d (runOpen failed)", irun);
1786 runCtrl[lastRun].fileId = 91;
1787 runCtrl[lastRun].procId = 91; // Is not set in writeEvt
1788 continue;
1789 }
1790
1791 runCtrl[lastRun].fileId = 0;
1792 runCtrl[lastRun].procId = 0; // Is not set in writeEvt
1793
1794 factPrintf(kInfo, -1, "procEvt: Opened new file for run %d (evt=%d)", irun, ievt);
1795 }
1796
1797 //-------- also check if run shall be closed (==> skip event, but do not close the file !!! )
1798 if (runCtrl[lastRun].procId == 0)
1799 {
1800 if (runCtrl[lastRun].closeTime < g_actTime ||
1801 runCtrl[lastRun].lastTime < g_actTime - 300 ||
1802 runCtrl[lastRun].maxEvt <= runCtrl[lastRun].procEvt)
1803 {
1804 factPrintf(kInfo, 502, "procEvt: Reached end of run condition for run %d", irun);
1805 runFinish1 (runCtrl[lastRun].runId);
1806 runCtrl[lastRun].procId = 1;
1807 }
1808 }
1809
1810 // Skip event because of no active run
1811 if (runCtrl[lastRun].procId != 0)
1812 {
1813 evtCtrl[k0].evtStat = 9091;
1814 continue;
1815 }
1816
1817 //--------
1818 //--------
1819
1820 //and set correct event header ; also check for consistency in event (not yet)
1821 evtCtrl[k0].fEvent->Errors[0] = evtCtrl[k0].Errors[0];
1822 evtCtrl[k0].fEvent->Errors[1] = evtCtrl[k0].Errors[1];
1823 evtCtrl[k0].fEvent->Errors[2] = evtCtrl[k0].Errors[2];
1824 evtCtrl[k0].fEvent->Errors[3] = evtCtrl[k0].Errors[3];
1825
1826 for (int ib=0; ib<NBOARDS; ib++)
1827 {
1828 // board is not read
1829 if (evtCtrl[k0].board[ib] == -1)
1830 {
1831 evtCtrl[k0].FADhead[ib].start_package_flag = 0;
1832 evtCtrl[k0].fEvent->BoardTime[ib] = 0;
1833 }
1834 else
1835 {
1836 evtCtrl[k0].fEvent->BoardTime[ib] = evtCtrl[k0].FADhead[ib].time;
1837 }
1838 }
1839
1840 const int rc = eventCheck(evtCtrl[k0].runNum, evtCtrl[k0].FADhead,
1841 evtCtrl[k0].fEvent);
1842 //gi.procTot++;
1843 numProc++;
1844
1845 if (rc < 0)
1846 {
1847 evtCtrl[k0].evtStat = 9999; //flag event to be skipped
1848 //gi.procErr++;
1849 }
1850 else
1851 {
1852 evtCtrl[k0].evtStat = 1000;
1853 runCtrl[lastRun].procEvt++;
1854 }
1855 }
1856
1857 if (gj.readStat < -10 && numWait == 0) { //nothing left to do
1858 factPrintf(kInfo, -1, "Exit Processing Process ...");
1859 gp_runStat = -22; //==> we should exit
1860 gj.procStat = -22; //==> we should exit
1861 return 0;
1862 }
1863
1864 //seems we have nothing to do, so sleep a little
1865 if (numProc == 0)
1866 usleep(1);
1867
1868 gp_runStat = gi_runStat;
1869 gj.procStat = gj.readStat;
1870
1871 }
1872
1873 //we are asked to abort asap ==> must flag all remaining events
1874 // when gi_runStat claims that all events are in the buffer...
1875
1876 factPrintf(kInfo, -1, "Abort Processing Process ...");
1877
1878 for (int k = 0; k < gi_maxProc; k++) {
1879 pthread_join (thread[k], (void **) &status);
1880 }
1881
1882 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
1883 {
1884 if (evtCtrl[k0].evtStat >= 0 && evtCtrl[k0].evtStat < 1000)
1885 evtCtrl[k0].evtStat = 9800; //flag event as 'processed'
1886 }
1887
1888 gp_runStat = -99;
1889 gj.procStat = -99;
1890
1891 return 0;
1892
1893} /*-----------------------------------------------------------------*/
1894
1895int
1896CloseRunFile (uint32_t runId, uint32_t closeTime, uint32_t maxEvt)
1897{
1898/* close run runId (all all runs if runId=0) */
1899/* return: 0=close scheduled / >0 already closed / <0 does not exist */
1900 int j;
1901
1902
1903 if (runId == 0) {
1904 for (j = 0; j < MAX_RUN; j++) {
1905 if (runCtrl[j].fileId == 0) { //run is open
1906 runCtrl[j].closeTime = closeTime;
1907 runCtrl[j].maxEvt = maxEvt;
1908 }
1909 }
1910 return 0;
1911 }
1912
1913 for (j = 0; j < MAX_RUN; j++) {
1914 if (runCtrl[j].runId == runId) {
1915 if (runCtrl[j].fileId == 0) { //run is open
1916 runCtrl[j].closeTime = closeTime;
1917 runCtrl[j].maxEvt = maxEvt;
1918 return 0;
1919 } else if (runCtrl[j].fileId < 0) { //run not yet opened
1920 runCtrl[j].closeTime = closeTime;
1921 runCtrl[j].maxEvt = maxEvt;
1922 return +1;
1923 } else { // run already closed
1924 return +2;
1925 }
1926 }
1927 } //we only reach here if the run was never created
1928 return -1;
1929
1930}
1931
1932void checkAndCloseRun(int j, int irun, int cond, int where)
1933{
1934 if (!cond &&
1935 runCtrl[j].closeTime >= g_actTime &&
1936 runCtrl[j].lastTime >= g_actTime - 300 &&
1937 runCtrl[j].maxEvt > runCtrl[j].actEvt)
1938 return;
1939
1940 //close run for whatever reason
1941 int ii = 0;
1942 if (cond)
1943 ii = 1;
1944 if (runCtrl[j].closeTime < g_actTime)
1945 ii |= 2; // = 2;
1946 if (runCtrl[j].lastTime < g_actTime - 300)
1947 ii |= 4; // = 3;
1948 if (runCtrl[j].maxEvt <= runCtrl[j].actEvt)
1949 ii |= 8; // = 4;
1950
1951 if (runCtrl[j].procId == 0)
1952 {
1953 runFinish1(runCtrl[j].runId);
1954 runCtrl[j].procId = 92;
1955 }
1956
1957 runCtrl[j].closeTime = g_actTime - 1;
1958
1959 const int rc = runClose(runCtrl[j].fileHd, NULL, 0);//&runTail[j], sizeof(runTail[j]));
1960 if (rc<0)
1961 {
1962 factPrintf(kError, 503, "writeEvt-%d: Error closing run %d (runClose,rc=%d)",
1963 where, runCtrl[j].runId, rc);
1964 runCtrl[j].fileId = 92+where*2;
1965 }
1966 else
1967 {
1968 factPrintf(kInfo, 503, "writeEvt-%d: Closed run %d (reason=%d)",
1969 where, irun, ii);
1970 runCtrl[j].fileId = 93+where*2;
1971 }
1972}
1973
1974/*-----------------------------------------------------------------*/
1975
1976
1977void *writeEvt (void *ptr)
1978{
1979/* *** main loop writing event (including opening and closing run-files */
1980
1981// cpu_set_t mask;
1982// int cpu = 1; //write thread
1983
1984 factPrintf(kInfo, -1, "Starting write-thread");
1985
1986/* CPU_ZERO initializes all the bits in the mask to zero. */
1987// CPU_ZERO (&mask);
1988/* CPU_SET sets only the bit corresponding to cpu. */
1989// CPU_SET (cpu, &mask);
1990/* sched_setaffinity returns 0 in success */
1991// if (sched_setaffinity (0, sizeof (mask), &mask) == -1) {
1992// snprintf (str, MXSTR, "W ---> can not create affinity to %d", cpu);
1993// }
1994
1995 int lastRun = 0; //usually run from last event still valid
1996
1997 while (g_runStat > -2)
1998 {
1999 int numWrite = 0;
2000 int numWait = 0;
2001
2002 for (int k0=evtCtrl_frstPtr; k0!=evtCtrl_lastPtr; k0++, k0 %= MAX_EVT*MAX_RUN)
2003 {
2004 if (evtCtrl[k0].evtStat <= 5000 || evtCtrl[k0].evtStat >= 9000)
2005 {
2006 if (evtCtrl[k0].evtStat > 0 && evtCtrl[k0].evtStat < 9000)
2007 numWait++;
2008
2009 continue;
2010 }
2011
2012 //we must drain the buffer asap
2013 if (gi_resetR > 1)
2014 {
2015 evtCtrl[k0].evtStat = 9904;
2016 continue;
2017 }
2018
2019 const int id = k0;//evtCtrl[k0].mBuffer_idx;
2020
2021 const uint32_t irun = evtCtrl[id].runNum;
2022 const int32_t ievt = evtCtrl[id].evNum;
2023
2024 // Find entry in runCtrl which belongs to the event mBuffer[id]
2025 // (only check if there is a need to check)
2026 if (runCtrl[lastRun].runId != irun)
2027 {
2028 //check which fileID to use (or open if needed)
2029 int j;
2030 for (j=0;j<MAX_RUN; j++)
2031 if (runCtrl[j].runId == irun)
2032 break;
2033
2034 if (j>=MAX_RUN)
2035 {
2036 factPrintf(kFatal, 901, "writeEvt: Can not find run %d for event %d in %d", irun, ievt, id);
2037 // FIXME: What is the right action?
2038 continue;
2039 }
2040
2041 lastRun = j;
2042 }
2043
2044 // File not yet open
2045 if (runCtrl[lastRun].fileId < 0)
2046 {
2047 RUN_HEAD actRun;
2048 actRun.Version = 1;
2049 actRun.RunType = -1; //to be adapted
2050 actRun.Nroi = evtCtrl[id].nRoi; //runCtrl[lastRun].roi0;
2051 actRun.NroiTM = evtCtrl[id].nRoiTM; //runCtrl[lastRun].roi8;
2052 actRun.RunTime = evtCtrl[id].pcTime[0];//runCtrl[lastRun].firstTime;
2053 actRun.RunUsec = evtCtrl[id].pcTime[1];//runCtrl[lastRun].firstUsec;
2054 actRun.NBoard = NBOARDS;
2055 actRun.NPix = NPIX;
2056 actRun.NTm = NTMARK;
2057
2058 memcpy(actRun.FADhead, evtCtrl[id].FADhead, NBOARDS * sizeof (PEVNT_HEADER));
2059
2060 runCtrl[lastRun].fileHd = runOpen (irun, &actRun, sizeof (actRun));
2061 if (runCtrl[lastRun].fileHd == NULL)
2062 {
2063 factPrintf(kError, 502, "writeEvt: Could not open a file for run %d (runOpen failed)", irun);
2064 runCtrl[lastRun].fileId = 91;
2065 continue;
2066 }
2067
2068 runCtrl[lastRun].fileId = 0;
2069 factPrintf(kInfo, -1, "writeEvt: Opened new file for run %d (evt %d)", irun, ievt);
2070 }
2071
2072 if (runCtrl[lastRun].fileId > 0)
2073 {
2074 // There is an event but file is already closed
2075 /*
2076 if (runCtrl[j].fileId < 100)
2077 {
2078 factPrintf(kWarn, 123, "writeEvt: File for run %d is closed", irun);
2079 runCtrl[j].fileId += 100;
2080 }*/
2081
2082 evtCtrl[k0].evtStat = 9903;
2083 }
2084
2085 // File is open
2086 if (runCtrl[lastRun].fileId==0)
2087 {
2088 const int rc = runWrite(runCtrl[lastRun].fileHd, evtCtrl[id].fEvent,
2089 sizeof (evtCtrl[id]));
2090 if (rc >= 0)
2091 {
2092 // Sucessfully wrote event
2093 runCtrl[lastRun].lastTime = g_actTime;
2094 runCtrl[lastRun].actEvt++;
2095
2096 evtCtrl[k0].evtStat = 9901;
2097 }
2098 else
2099 {
2100 factPrintf(kError, 503, "writeEvt: Writing event for run %d failed (runWrite)", irun);
2101 evtCtrl[k0].evtStat = 9902;
2102 }
2103
2104 checkAndCloseRun(lastRun, irun, rc<0, 1);
2105 }
2106 }
2107/*
2108 //check if we should close a run (mainly when no event pending)
2109 //ETIENNE but first figure out which one is the latest run with a complete event.
2110 //i.e. max run Id and lastEvt >= 0
2111 //this condition is sufficient because all pending events were written already in the loop just above
2112 //actrun
2113 uint32_t lastStartedTime = 0;
2114 uint32_t runIdFound = 1;
2115
2116 //If we have an active run, look for its start time
2117 if (actrun != 0)
2118 {
2119 runIdfound = 0;
2120 for (int j=0;j<MAX_RUN;j++)
2121 {
2122 if (runCtrl[j].runId == actrun)
2123 {
2124 lastStartedTime = runCtrl[j].lastTime;
2125 runIdFound = 1;
2126 }
2127 }
2128 }
2129
2130 if (runIdFound == 0)
2131 {
2132 factPrintf(kInfo, 0, "An Active run (number %u) has been registered, but it could not be found in the runs list", actrun);
2133 }
2134
2135 //Also check if some files will never be opened
2136 //EDIT: this is completely useless, because as run Numbers are taken from FADs board,
2137 //I will never get run numbers for which no file is to be opened
2138 for (int j=0;j<MAX_RUN;j++)
2139 {
2140 if ((runCtrl[j].fileId < 0) &&
2141 (runCtrl[j].lastTime < lastStartedTime) &&
2142 (runCtrl[j].runId != 0))
2143 {
2144 factPrintf(kInfo, 0, "writeEvt: No file will be opened for run %u. Last run: %u (started)", runCtrl[j].runId, actrun);
2145 ;//TODO notify that this run will never be opened
2146 }
2147 }
2148 */
2149
2150 // Although the are no pending events, we have to check if a run should be closed (timeout)
2151 for (int j=0; j<MAX_RUN; j++)
2152 {
2153 if (runCtrl[j].fileId == 0)
2154 {
2155 //ETIENNE added the condition at this line. dunno what to do with run 0: skipping it
2156 const int cond = /*runCtrl[j].lastTime < lastStartedTime &&*/ runCtrl[j].runId == 0;
2157 checkAndCloseRun(j, runCtrl[j].runId, cond, 2);
2158 }
2159 }
2160
2161 //seems we have nothing to do, so sleep a little
2162 if (numWrite == 0)
2163 usleep(1);
2164
2165 //nothing left to do
2166 if (gj.readStat < -10 && numWait == 0)
2167 {
2168 factPrintf(kInfo, -1, "Finish Write Process ...");
2169 gw_runStat = -22; //==> we should exit
2170 gj.writStat = -22; //==> we should exit
2171 break;
2172 }
2173
2174 gw_runStat = gi_runStat;
2175 gj.writStat = gj.readStat;
2176 }
2177
2178 factPrintf(kInfo, -1, "Close all open files ...");
2179 for (int j=0; j<MAX_RUN; j++)
2180 {
2181 if (runCtrl[j].fileId == 0)
2182 checkAndCloseRun(j, runCtrl[j].runId, 1, 3);
2183 }
2184
2185 gw_runStat = -99;
2186 gj.writStat = -99;
2187
2188 factPrintf(kInfo, -1, "Exit Writing Process ...");
2189
2190 return 0;
2191} /*-----------------------------------------------------------------*/
2192
2193
2194
2195
2196void
2197StartEvtBuild ()
2198{
2199
2200 int i, /*j,*/ imax, status/*, th_ret[50]*/;
2201 pthread_t thread[50];
2202 struct timespec xwait;
2203
2204 gi_runStat = gp_runStat = gw_runStat = 0;
2205 gj.readStat = gj.procStat = gj.writStat = 0;
2206
2207 factPrintf(kInfo, -1, "Starting EventBuilder V15.07 A");
2208
2209//initialize run control logics
2210 for (i = 0; i < MAX_RUN; i++) {
2211 runCtrl[i].runId = 0;
2212 runCtrl[i].fileId = -2;
2213 }
2214
2215//prepare for subProcesses
2216 gi_maxSize = g_maxSize;
2217 if (gi_maxSize <= 0)
2218 gi_maxSize = 1;
2219
2220 gi_maxProc = g_maxProc;
2221 if (gi_maxProc <= 0 || gi_maxProc > 90) {
2222 factPrintf(kFatal, 301, "Illegal number of processes %d", gi_maxProc);
2223 gi_maxProc = 1;
2224 }
2225//partially initialize event control logics
2226 evtCtrl_frstPtr = 0;
2227 evtCtrl_lastPtr = 0;
2228
2229//start all threads (more to come) when we are allowed to ....
2230 while (g_runStat == 0) {
2231 xwait.tv_sec = 0;
2232 xwait.tv_nsec = 10000000; // sleep for ~10 msec
2233 nanosleep (&xwait, NULL);
2234 }
2235
2236 i = 0;
2237 /*th_ret[i] =*/ pthread_create (&thread[i], NULL, readFAD, NULL);
2238 i++;
2239 /*th_ret[i] =*/ pthread_create (&thread[i], NULL, procEvt, NULL);
2240 i++;
2241 /*th_ret[i] =*/ pthread_create (&thread[i], NULL, writeEvt, NULL);
2242 i++;
2243 imax = i;
2244
2245
2246#ifdef BILAND
2247 xwait.tv_sec = 30;;
2248 xwait.tv_nsec = 0; // sleep for ~20sec
2249 nanosleep (&xwait, NULL);
2250
2251 printf ("close all runs in 2 seconds\n");
2252
2253 CloseRunFile (0, time (NULL) + 2, 0);
2254
2255 xwait.tv_sec = 1;;
2256 xwait.tv_nsec = 0; // sleep for ~20sec
2257 nanosleep (&xwait, NULL);
2258
2259 printf ("setting g_runstat to -1\n");
2260
2261 g_runStat = -1;
2262#endif
2263
2264
2265//wait for all threads to finish
2266 for (i = 0; i < imax; i++) {
2267 /*j =*/ pthread_join (thread[i], (void **) &status);
2268 }
2269
2270} /*-----------------------------------------------------------------*/
2271
2272
2273
2274
2275
2276
2277
2278
2279
2280
2281
2282
2283
2284
2285
2286 /*-----------------------------------------------------------------*/
2287 /*-----------------------------------------------------------------*/
2288 /*-----------------------------------------------------------------*/
2289 /*-----------------------------------------------------------------*/
2290 /*-----------------------------------------------------------------*/
2291
2292#ifdef BILAND
2293
2294int
2295subProcEvt (int threadID, PEVNT_HEADER * fadhd, EVENT * event,
2296 int8_t * buffer)
2297{
2298 printf ("called subproc %d\n", threadID);
2299 return threadID + 1;
2300}
2301
2302
2303
2304
2305 /*-----------------------------------------------------------------*/
2306 /*-----------------------------------------------------------------*/
2307 /*-----------------------------------------------------------------*/
2308 /*-----------------------------------------------------------------*/
2309 /*-----------------------------------------------------------------*/
2310
2311
2312
2313
2314FileHandle_t
2315runOpen (uint32_t irun, RUN_HEAD * runhd, size_t len)
2316{
2317 return 1;
2318};
2319
2320int
2321runWrite (FileHandle_t fileHd, EVENT * event, size_t len)
2322{
2323 return 1;
2324 usleep (10000);
2325 return 1;
2326}
2327
2328
2329//{ return 1; } ;
2330
2331int
2332runClose (FileHandle_t fileHd, RUN_TAIL * runth, size_t len)
2333{
2334 return 1;
2335};
2336
2337
2338
2339
2340int
2341eventCheck (uint32_t runNr, PEVNT_HEADER * fadhd, EVENT * event)
2342{
2343 int i = 0;
2344
2345// printf("------------%d\n",ntohl(fadhd[7].fad_evt_counter) );
2346// for (i=0; i<NBOARDS; i++) {
2347// printf("b=%2d,=%5d\n",i,fadhd[i].board_id);
2348// }
2349 return 0;
2350}
2351
2352
2353void
2354factStatNew (EVT_STAT gi)
2355{
2356 int i;
2357
2358//for (i=0;i<MAX_SOCK;i++) {
2359// printf("%4d",gi.numRead[i]);
2360// if (i%20 == 0 ) printf("\n");
2361//}
2362}
2363
2364void
2365gotNewRun (int runnr, PEVNT_HEADER * headers)
2366{
2367 printf ("got new run %d\n", runnr);
2368 return;
2369}
2370
2371void
2372factStat (GUI_STAT gj)
2373{
2374// printf("stat: bfr%5lu skp%4lu free%4lu (tot%7lu) mem%12lu rd%12lu %3lu\n",
2375// array[0],array[1],array[2],array[3],array[4],array[5],array[6]);
2376}
2377
2378
2379void
2380debugRead (int isock, int ibyte, int32_t event, int32_t ftmevt, int32_t runnr,
2381 int state, uint32_t tsec, uint32_t tusec)
2382{
2383// printf("%3d %5d %9d %3d %12d\n",isock, ibyte, event, state, tusec) ;
2384}
2385
2386
2387
2388void
2389debugStream (int isock, void *buf, int len)
2390{
2391}
2392
2393void
2394debugHead (int i, int j, void *buf)
2395{
2396}
2397
2398
2399void
2400factOut (int severity, int err, char *message)
2401{
2402 static FILE *fd;
2403 static int file = 0;
2404
2405 if (file == 0) {
2406 printf ("open file\n");
2407 fd = fopen ("x.out", "w+");
2408 file = 999;
2409 }
2410
2411 fprintf (fd, "%3d %3d | %s \n", severity, err, message);
2412
2413 if (severity != kDebug)
2414 printf ("%3d %3d | %s\n", severity, err, message);
2415}
2416
2417
2418
2419int
2420main ()
2421{
2422 int i, b, c, p;
2423 char ipStr[100];
2424 struct in_addr IPaddr;
2425
2426 g_maxMem = 1024 * 1024; //MBytes
2427//g_maxMem = g_maxMem * 1024 *10 ; //10GBytes
2428 g_maxMem = g_maxMem * 200; //100MBytes
2429
2430 g_maxProc = 20;
2431 g_maxSize = 30000;
2432
2433 g_runStat = 40;
2434
2435 i = 0;
2436
2437// version for standard crates
2438//for (c=0; c<4,c++) {
2439// for (b=0; b<10; b++) {
2440// sprintf(ipStr,"10.0.%d.%d",128+c,128+b)
2441//
2442// inet_pton(PF_INET, ipStr, &IPaddr) ;
2443//
2444// g_port[i].sockAddr.sin_family = PF_INET;
2445// g_port[i].sockAddr.sin_port = htons(5000) ;
2446// g_port[i].sockAddr.sin_addr = IPaddr ;
2447// g_port[i].sockDef = 1 ;
2448// i++ ;
2449// }
2450//}
2451//
2452//version for PC-test *
2453 for (c = 0; c < 4; c++) {
2454 for (b = 0; b < 10; b++) {
2455 sprintf (ipStr, "10.0.%d.11", 128 + c);
2456 if (c < 2)
2457 sprintf (ipStr, "10.0.%d.11", 128);
2458 else
2459 sprintf (ipStr, "10.0.%d.11", 131);
2460// if (c==0) sprintf(ipStr,"10.0.100.11") ;
2461
2462 inet_pton (PF_INET, ipStr, &IPaddr);
2463 p = 31919 + 100 * c + 10 * b;
2464
2465
2466 g_port[i].sockAddr.sin_family = PF_INET;
2467 g_port[i].sockAddr.sin_port = htons (p);
2468 g_port[i].sockAddr.sin_addr = IPaddr;
2469 g_port[i].sockDef = 1;
2470
2471 i++;
2472 }
2473 }
2474
2475
2476//g_port[17].sockDef =-1 ;
2477//g_actBoards-- ;
2478
2479 StartEvtBuild ();
2480
2481 return 0;
2482
2483}
2484#endif
Note: See TracBrowser for help on using the repository browser.