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

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