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

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