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

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