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

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