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

Last change on this file since 11495 was 11443, checked in by tbretz, 13 years ago
Latest updates.
File size: 55.1 KB
Line 
1
2
3#define PX8 99 //simulator does not create double-length roi for pixel 8
4 //for real data, set PX8 = 8 ==> ask for double roi=TM
5
6
7
8#include <stdlib.h>
9#include <stdint.h>
10#include <unistd.h>
11#include <stdio.h>
12#include <sys/time.h>
13#include <arpa/inet.h>
14#include <string.h>
15#include <math.h>
16#include <error.h>
17#include <errno.h>
18#include <unistd.h>
19#include <sys/types.h>
20#include <sys/socket.h>
21#include <pthread.h>
22#include <sched.h>
23
24#include "EventBuilder.h"
25
26enum Severity
27{
28 kMessage = 10, ///< Just a message, usually obsolete
29 kInfo = 20, ///< An info telling something which can be interesting to know
30 kWarn = 30, ///< A warning, things that somehow might result in unexpected or unwanted bahaviour
31 kError = 40, ///< Error, something unexpected happened, but can still be handled by the program
32 kFatal = 50, ///< An error which cannot be handled at all happend, the only solution is program termination
33 kDebug = 99, ///< A message used for debugging only
34};
35
36#define MIN_LEN 32 // min #bytes needed to interpret FADheader
37#define MAX_LEN 256*1024 // size of read-buffer per socket
38
39extern FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len ) ;
40extern int runWrite(FileHandle_t fileHd , EVENT *event, size_t len ) ;
41extern int runClose(FileHandle_t fileHd , RUN_TAIL *runth, size_t len ) ;
42extern void factOut(int severity, int err, char* message ) ;
43extern void gotNewRun( int runnr, PEVNT_HEADER *headers );
44
45
46extern void factStat(GUI_STAT gj) ;
47
48extern void factStatNew(EVT_STAT gi) ;
49
50extern int eventCheck( PEVNT_HEADER *fadhd, EVENT *event) ;
51
52extern void debugHead(int i, int j, void *buf);
53
54extern void debugRead(int isock, int ibyte, int32_t event,int32_t ftmevt,
55 int32_t runnr, int state, uint32_t tsec, uint32_t tusec ) ;
56extern void debugStream(int isock, void *buf, int len) ;
57
58int CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt);
59
60
61uint g_actTime ;
62uint g_actUsec ;
63int g_runStat ;
64int g_reset ;
65int gi_reset, gi_resetR, gi_resetS, gi_resetW, gi_resetX ;
66size_t g_maxMem ; //maximum memory allowed for buffer
67
68//no longer needed ...
69 int g_maxBoards ; //maximum number of boards to be initialized
70 int g_actBoards ;
71//
72
73FACT_SOCK g_port[NBOARDS] ; // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd"
74
75
76 int gi_runStat ;
77 int gp_runStat ;
78 int gw_runStat ;
79
80 int gi_memStat = +1 ;
81
82 uint32_t gi_myRun = 9876789 ;
83 uint32_t actrun ;
84
85
86uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards
87
88//uint gi_EvtStart= 0 ;
89//uint gi_EvtRead = 0 ;
90//uint gi_EvtBad = 0 ;
91//uint gi_EvtTot = 0 ;
92//size_t gi_usedMem = 0 ;
93
94//uint gw_EvtTot = 0 ;
95//uint gp_EvtTot = 0 ;
96
97PIX_MAP g_pixMap[NPIX] ;
98
99EVT_STAT gi ;
100GUI_STAT gj ;
101
102EVT_CTRL evtCtrl ; //control of events during processing
103int evtIdx[MAX_EVT*MAX_RUN] ; //index from mBuffer to evtCtrl
104
105WRK_DATA mBuffer[MAX_EVT*MAX_RUN]; //local working space
106
107
108
109
110RUN_HEAD actRun ;
111
112RUN_CTRL runCtrl[MAX_RUN] ;
113
114RUN_TAIL runTail[MAX_RUN] ;
115
116
117/*
118*** Definition of rdBuffer to read in IP packets; keep it global !!!!
119 */
120
121
122typedef union {
123 int8_t B[MAX_LEN ];
124 int16_t S[MAX_LEN/2];
125 int32_t I[MAX_LEN/4];
126 int64_t L[MAX_LEN/8];
127} CNV_FACT ;
128
129typedef struct {
130 int bufTyp ; //what are we reading at the moment: 0=header 1=data -1=skip ...
131 int32_t bufPos ; //next byte to read to the buffer next
132 int32_t bufLen ; //number of bytes left to read
133 int32_t skip ; //number of bytes skipped before start of event
134
135 int sockStat ; //-1 if socket not yet connected , 99 if not exist
136 int socket ; //contains the sockets
137 struct sockaddr_in SockAddr ; //IP for each socket
138
139 int evtID ; // event ID of event currently read
140 int runID ; // run "
141 int ftmID ; // event ID from FTM
142 uint fadLen ; // FADlength of event currently read
143 int fadVers ; // Version of FAD
144 int board ; // boardID (softwareID: 0..40 )
145 int Port ;
146
147 CNV_FACT *xBuf ; //a copy of rBuf (temporary for debuging)
148 CNV_FACT *rBuf ;
149
150
151} READ_STRUCT ;
152
153
154typedef union {
155 int8_t B[2];
156 int16_t S ;
157} SHORT_BYTE ;
158
159
160
161
162
163#define MXSTR 1000
164char str[MXSTR] ;
165
166SHORT_BYTE start, stop;
167
168READ_STRUCT rd[MAX_SOCK] ; //buffer to read IP and afterwards store in mBuffer
169
170
171
172/*-----------------------------------------------------------------*/
173
174
175/*-----------------------------------------------------------------*/
176
177
178
179int GenSock(int flag, int sid, int port, struct sockaddr_in *sockAddr, READ_STRUCT *rd) {
180/*
181*** generate Address, create sockets and allocates readbuffer for it
182***
183*** if flag==0 generate socket and buffer
184*** <0 destroy socket and buffer
185*** >0 close and redo socket
186***
187*** sid : board*7 + port id
188 */
189
190 int j ;
191
192 if (rd->sockStat ==0 ) { //close socket if open
193 j=close(rd->socket) ;
194 if (j>0) {
195 snprintf(str,MXSTR,"Error closing socket %d | %m",sid);
196 factOut(kFatal,771, str ) ;
197 } else {
198 snprintf(str,MXSTR,"Succesfully closed socket %d",sid);
199 factOut(kInfo,771, str ) ;
200 }
201 }
202
203
204 if (flag < 0) {
205 free(rd->rBuf) ; //and never open again
206 free(rd->xBuf) ; //and never open again
207 rd->rBuf = NULL ;
208 rd->sockStat = 99 ;
209 return 0 ;
210 }
211
212
213 if (flag == 0) { //generate address and buffer ...
214 rd->Port = port ;
215 rd->SockAddr.sin_family = sockAddr->sin_family;
216 rd->SockAddr.sin_port = htons(port) ;
217 rd->SockAddr.sin_addr = sockAddr->sin_addr ;
218
219 rd->xBuf = malloc(sizeof(CNV_FACT) ) ;
220 rd->rBuf = malloc(sizeof(CNV_FACT) ) ;
221 if ( rd->rBuf == NULL ) {
222 snprintf(str,MXSTR,"Could not create local buffer %d",sid);
223 factOut(kFatal,774, str ) ;
224 rd->sockStat = 77 ;
225 return -3 ;
226 }
227 }
228
229
230 if ( (rd->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) {
231 snprintf(str,MXSTR,"Could not generate socket %d | %m",sid);
232 factOut(kFatal,773, str ) ;
233 rd->sockStat = 88 ;
234 return -2 ;
235 }
236
237 snprintf(str,MXSTR,"Successfully generated socket %d ",sid);
238 factOut(kInfo,773, str ) ;
239 rd->sockStat = -1 ; //try to (re)open socket
240 return 0 ;
241
242} /*-----------------------------------------------------------------*/
243
244 /*-----------------------------------------------------------------*/
245
246
247
248
249int mBufInit() {
250// initialize mBuffer (mark all entries as unused\empty)
251
252 int i ;
253 uint32_t actime ;
254
255 actime = g_actTime + 50000000 ;
256
257 for (i=0; i<MAX_EVT*MAX_RUN; i++) {
258 mBuffer[i].evNum = mBuffer[i].nRoi = -1;
259 mBuffer[i].runNum= 0 ;
260
261 evtCtrl.evtBuf[ i] = -1 ;
262 evtCtrl.evtStat[ i] = -1 ;
263 evtCtrl.pcTime[ i] = actime ; //initiate to far future
264 }
265
266
267 actRun.FADhead = malloc( NBOARDS* sizeof(PEVNT_HEADER) ) ;
268
269 evtCtrl.frstPtr = 0 ;
270 evtCtrl.lastPtr = 0 ;
271
272 return 0 ;
273
274} /*-----------------------------------------------------------------*/
275
276
277
278
279int mBufEvt( int evID, uint runID, int nRoi, int sk, int fadlen) {
280// generate a new Event into mBuffer:
281// make sure only complete Event are possible, so 'free' will always work
282// returns index into mBuffer[], or negative value in case of error
283
284
285 int i, k, evFree ;
286 int headmem=0 ;
287 size_t needmem = 0 ;
288
289 if (nRoi <0 || nRoi > 1024) {
290 snprintf(str,MXSTR,"illegal nRoi: %d",nRoi) ;
291 factOut(kError, 1, str ) ;
292 return -9999 ;
293 }
294
295 i = evID % MAX_EVT ;
296 evFree = -1 ;
297
298 for ( k=0; k<MAX_RUN; k++) {
299 if ( mBuffer[i].evNum == evID
300 && mBuffer[i].runNum== runID ) {
301 return i ;
302 }
303 if ( evFree < 0 && mBuffer[i].evNum < 0 ) evFree = i ;
304 i += MAX_EVT ;
305 }
306
307
308 //event does not yet exist; create
309 if (evFree < 0 ) { //no space available in ctrl
310 snprintf(str,MXSTR,"no control slot to keep event %d",evID) ;
311 factOut(kError,881, str ) ;
312 return -1 ;
313 }
314 i = evFree ; //found free entry; use it ...
315
316
317 needmem = sizeof(EVENT) + NPIX*nRoi*2 + NTMARK*nRoi*2; //
318
319 headmem = NBOARDS* sizeof(PEVNT_HEADER) ;
320
321 if ( gj.usdMem + needmem + headmem > g_maxMem) {
322 gj.maxMem = gj.usdMem + needmem + headmem ;
323 if (gi_memStat>0 ) {
324 gi_memStat = -99 ;
325 snprintf(str,MXSTR,"no memory left to keep event %6d sock %3d",evID,sk) ;
326 factOut(kError,882, str ) ;
327 } else {
328 snprintf(str,MXSTR,"no memory left to keep event %6d sock %3d",evID,sk) ;
329 factOut(kDebug,882, str ) ;
330 }
331 return -11 ;
332 }
333
334 mBuffer[i].FADhead = malloc( headmem ) ;
335 if (mBuffer[i].FADhead == NULL) {
336 snprintf(str,MXSTR,"malloc header failed for event %d",evID) ;
337 factOut(kError,882, str ) ;
338 return -12;
339 }
340
341 mBuffer[i].fEvent = malloc( needmem ) ;
342 if (mBuffer[i].fEvent == NULL) {
343 snprintf(str,MXSTR,"malloc data failed for event %d",evID) ;
344 factOut(kError,882, str ) ;
345 free(mBuffer[i].fEvent) ;
346 mBuffer[i].fEvent = NULL ;
347 return -22;
348 }
349
350 //flag all boards as unused
351 mBuffer[i].nBoard = 0 ;
352 for (k=0; k<NBOARDS; k++ ) {
353 mBuffer[i].board[k] = -1;
354 }
355
356 //flag all pixels as unused
357 for (k=0; k<NPIX; k++ ) {
358 mBuffer[i].fEvent->StartPix[k] = -1 ;
359 }
360
361 //flag all TMark as unused
362 for (k=0; k<NTMARK; k++ ) {
363 mBuffer[i].fEvent->StartTM[k] = -1 ;
364 }
365
366 mBuffer[i].fEvent->PCTime[0] = g_actTime ;
367 mBuffer[i].fEvent->PCTime[1] = g_actUsec ;
368 mBuffer[i].pcTime = g_actTime ;
369 mBuffer[i].nRoi = nRoi ;
370 mBuffer[i].evNum = evID ;
371 mBuffer[i].runNum = runID ;
372 mBuffer[i].evtLen = needmem ;
373
374 gj.usdMem += needmem + headmem;
375 if (gj.usdMem > gj.maxMem ) gj.maxMem = gj.usdMem ;
376
377 gj.rateNew++ ;
378
379 //register event in 'active list (reading)'
380
381 evtCtrl.evtBuf[ evtCtrl.lastPtr] = i ;
382 evtCtrl.evtStat[ evtCtrl.lastPtr] = 0 ;
383 evtCtrl.pcTime[ evtCtrl.lastPtr] = g_actTime ;
384 evtIdx[i] = evtCtrl.lastPtr ;
385
386
387 snprintf(str,MXSTR,"%5d start new evt %8d %8d sock %3d len %5d t %10d",
388 evID,i,evtCtrl.lastPtr,sk,fadlen,mBuffer[i].pcTime);
389 factOut(kDebug,-11, str ) ;
390 evtCtrl.lastPtr++ ;
391 if (evtCtrl.lastPtr == MAX_EVT*MAX_RUN ) evtCtrl.lastPtr = 0;
392
393 gi.evtGet++ ;
394
395 //check if runId already registered in runCtrl
396 evFree = -1 ;
397 for (k=0; k<MAX_RUN; k++) {
398 if (runCtrl[k].runId == runID ) return i ;//run exists already
399 else if (evFree < 0 && runCtrl[k].runId == 0 ) evFree = k ;
400 }
401
402 if (evFree <0 ) {
403 snprintf(str,MXSTR,"not able to register the new run %d",runID);
404 factOut(kFatal,883, str ) ;
405 } else {
406 runCtrl[evFree].runId = runID ;
407 runCtrl[evFree].fileId = -2 ;
408 runCtrl[evFree].lastEvt= -1 ;
409 runCtrl[evFree].nextEvt= 0;
410 runCtrl[evFree].actEvt = 0;
411 runCtrl[evFree].maxEvt = 999999999 ; //max number events allowed
412 runCtrl[evFree].lastTime=g_actTime ;
413 runCtrl[evFree].closeTime=g_actTime + 3600*24 ; //max time allowed
414 runCtrl[evFree].lastTime = 0 ;
415
416 runTail[evFree].nEventsOk =
417 runTail[evFree].nEventsRej =
418 runTail[evFree].nEventsBad =
419 runTail[evFree].PCtime0 =
420 runTail[evFree].PCtimeX = 0 ;
421 }
422
423 return i ;
424
425} /*-----------------------------------------------------------------*/
426
427
428
429
430int mBufFree(int i) {
431//delete entry [i] from mBuffer:
432//(and make sure multiple calls do no harm ....)
433
434 int headmem=0 ;
435 int evid ;
436 size_t freemem = 0 ;
437
438 evid = mBuffer[i].evNum ;
439 freemem = mBuffer[i].evtLen ;
440
441 free(mBuffer[i].fEvent ) ;
442 mBuffer[i].fEvent = NULL ;
443
444 free(mBuffer[i].FADhead ) ;
445 mBuffer[i].FADhead = NULL ;
446
447 headmem = NBOARDS* sizeof(PEVNT_HEADER) ;
448 mBuffer[i].evNum = mBuffer[i].nRoi= -1;
449 mBuffer[i].runNum = 0;
450
451 gj.usdMem = gj.usdMem - freemem - headmem;
452
453 if (gi_memStat < 0 ) {
454 if (gj.usdMem <= 0.75 * gj.maxMem ) gi_memStat = +1 ;
455 }
456
457
458 return 0 ;
459
460} /*-----------------------------------------------------------------*/
461
462
463void resetEvtStat() {
464 int i ;
465
466 for (i=0; i<MAX_SOCK; i++) gi.numRead[i] = 0 ;
467
468 for (i=0; i<NBOARDS; i++ ) {
469 gi.gotByte[i] = 0 ;
470 gi.gotErr[i] = 0 ;
471 }
472
473 gi.evtGet = 0 ; //#new Start of Events read
474 gi.evtTot = 0 ; //#complete Events read
475 gi.evtErr = 0 ; //#Events with Errors
476 gi.evtSkp = 0 ; //#Events incomplete (timeout)
477
478 gi.procTot = 0 ; //#Events processed
479 gi.procErr = 0 ; //#Events showed problem in processing
480 gi.procTrg = 0 ; //#Events accepted by SW trigger
481 gi.procSkp = 0 ; //#Events rejected by SW trigger
482
483 gi.feedTot = 0 ; //#Events used for feedBack system
484 gi.feedErr = 0 ; //#Events rejected by feedBack
485
486 gi.wrtTot = 0 ; //#Events written to disk
487 gi.wrtErr = 0 ; //#Events with write-error
488
489 gi.runOpen = 0 ; //#Runs opened
490 gi.runClose= 0 ; //#Runs closed
491 gi.runErr = 0 ; //#Runs with open/close errors
492
493return ;
494} /*-----------------------------------------------------------------*/
495
496
497
498void initReadFAD() {
499return ;
500} /*-----------------------------------------------------------------*/
501
502
503
504void *readFAD( void *ptr ) {
505/* *** main loop reading FAD data and sorting them to complete events */
506 int head_len,frst_len,numok,numok2,numokx,dest,evID,i,k ;
507 int actBoards = 0, minLen ;
508 int32_t jrd ;
509 uint gi_SecTime ; //time in seconds
510 int boardId, roi,drs,px,src,pixS,pixH,pixC,pixR,tmS ;
511
512 int goodhed = 0 ;
513 int errcnt0 = 0 ;
514
515 int sockDef[NBOARDS]; //internal state of sockets
516 int jrdx ;
517
518
519 struct timespec xwait ;
520
521
522 struct timeval *tv, atv;
523 tv=&atv;
524 uint32_t tsec, tusec ;
525
526
527 gi_myRun = g_actTime ;
528 snprintf(str,MXSTR,"start initializing");
529 factOut(kInfo,-1, str ) ;
530
531 int cpu = 7 ; //read thread
532 cpu_set_t mask;
533
534/* CPU_ZERO initializes all the bits in the mask to zero. */
535 CPU_ZERO( &mask );
536/* CPU_SET sets only the bit corresponding to cpu. */
537 cpu = 7 ;
538 CPU_SET( cpu, &mask );
539
540/* sched_setaffinity returns 0 in success */
541 if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
542 snprintf(str,MXSTR,"W ---> can not create affinity to %d",cpu);
543 factOut(kWarn,-1, str ) ;
544 }
545
546 head_len = sizeof(PEVNT_HEADER) ;
547 frst_len = head_len ; //max #bytes to read first: fad_header only, so each event must be longer, even for roi=0
548 minLen = head_len ; //min #bytes needed to check header: full header for debug
549
550 start.S=0xFB01;
551 stop.S= 0x04FE;
552
553/* initialize run control logics */
554 for (i=0; i<MAX_RUN; i++) {
555 runCtrl[i].runId = 0 ;
556 runCtrl[i].fileId = -2 ;
557 }
558 gi_resetS = gi_resetR = 9;
559 for (i=0; i<NBOARDS; i++) sockDef[i]= 0 ;
560
561START:
562 evtCtrl.frstPtr = 0 ;
563 evtCtrl.lastPtr = 0 ;
564
565 gi_SecTime = g_actTime ;
566 gi_runStat = g_runStat ;
567 gj.readStat= g_runStat ;
568 numok = numok2 = 0 ;
569
570 if ( gi_resetS > 0) {
571 //make sure all sockets are preallocated as 'not exist'
572 for (i=0; i<MAX_SOCK; i++) {
573 rd[i].socket = -1 ;
574 rd[i].sockStat = 99 ;
575 }
576
577 for (k=0; k<NBOARDS; k++) {
578 gi_NumConnect[k]=0;
579 gi.numConn[k] =0;
580 gj.numConn[k] =0;
581 gj.errConn[k] =0;
582 gj.rateBytes[k] =0;
583 gj.totBytes[k] =0;
584 }
585
586 }
587
588
589 if ( gi_resetR > 0) {
590 resetEvtStat();
591 gj.usdMem = gj.maxMem = gj.xxxMem = 0 ;
592 gj.totMem = g_maxMem ;
593 gj.bufNew = gj.bufEvt = 0 ;
594 gj.evtSkip= gj.evtWrite = gj.evtErr = 0 ;
595
596
597 mBufInit() ; //initialize buffers
598
599 snprintf(str,MXSTR,"end initializing");
600 factOut(kInfo,-1, str ) ;
601 }
602
603
604 gi_reset = gi_resetR = gi_resetS = gi_resetW = 0 ;
605
606 while (g_runStat >=0 && g_reset ==0 ) { //loop until global variable g_runStat claims stop
607
608 gi_runStat = g_runStat;
609 gj.readStat= g_runStat;
610
611 int b,p,p0,s0,nch;
612 nch = 0 ;
613 for (b=0; b<NBOARDS; b++ ) {
614 k = b*7 ;
615 if ( g_port[b].sockDef != sockDef[b] ) { //something has changed ...
616 nch++ ;
617 gi_NumConnect[ b ] = 0 ; //must close all connections
618 gi.numConn[ b ] = 0;
619 gj.numConn[ b ] = 0;
620 if ( sockDef[b] == 0) s0= 0 ; //sockets to be defined and opened
621 else if (g_port[b].sockDef == 0) s0= -1 ; //sockets to be destroyed
622 else s0= +1 ; //sockets to be closed and reopened
623
624 if (s0 == 0) p0=ntohs(g_port[b].sockAddr.sin_port);
625 else p0=0 ;
626
627 for (p=p0+1; p<p0+8; p++) {
628 GenSock(s0, k, p, &g_port[b].sockAddr, &rd[k]) ; //generate address and socket
629 k++ ;
630 }
631 sockDef[b] = g_port[b].sockDef ;
632 }
633 }
634
635 if (nch > 0 ) {
636 actBoards = 0 ;
637 for (b=0; b<NBOARDS; b++ ) {
638 if ( sockDef[b] > 0 ) actBoards++ ;
639 }
640 }
641
642
643 jrdx = 0;
644 numokx= 0;
645 numok = 0 ; //count number of succesfull actions
646
647 for (i=0; i<MAX_SOCK; i++) { //check all sockets if something to read
648 b = i / 7 ;
649 if (sockDef[b] > 0) s0= +1 ;
650 else s0= -1 ;
651
652 gettimeofday( tv, NULL);
653 g_actTime = tsec = atv.tv_sec ;
654 g_actUsec = tusec= atv.tv_usec ;
655
656 if (rd[i].sockStat <0 ) { //try to connect if not yet done
657 rd[i].sockStat=connect(rd[i].socket,
658 (struct sockaddr*) &rd[i].SockAddr, sizeof(rd[i].SockAddr)) ;
659 if (rd[i].sockStat ==0 ) { //successfull ==>
660 if (sockDef[b] > 0) {
661 rd[i].bufTyp = 0 ; // expect a header
662 rd[i].bufLen = frst_len ; // max size to read at begining
663 } else {
664 rd[i].bufTyp = -1 ; // full data to be skipped
665 rd[i].bufLen = MAX_LEN ; //huge for skipping
666 }
667 rd[i].bufPos = 0 ; // no byte read so far
668 rd[i].skip = 0 ; // start empty
669 gi_NumConnect[ b ]++ ;
670 gi.numConn[ b ]++ ;
671 gj.numConn[ b ]++ ;
672 snprintf(str,MXSTR,"+++connect %d %d",b,gi.numConn[ b ]);
673 factOut(kInfo,-1, str ) ;
674 }
675 }
676
677 if (rd[i].sockStat ==0 ) { //we have a connection ==> try to read
678 if (rd[i].bufLen > 0) { //might be nothing to read [buffer full]
679 numok++ ;
680 jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT);
681
682 if (jrd >0 ) {
683 debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ;
684 memcpy(&rd[i].xBuf->B[ rd[i].bufPos], &rd[i].rBuf->B[ rd[i].bufPos], jrd) ;
685 snprintf(str,MXSTR,"read sock %3d bytes %5d , %5d",i,jrd,rd[i].bufLen);
686 factOut(kDebug,301, str ) ;
687 }
688
689 if (jrd == 0) { //connection has closed ...
690 snprintf(str,MXSTR,"Socket %d closed by FAD",i);
691 factOut(kInfo,441, str ) ;
692 GenSock(s0, i, 0,NULL, &rd[i]) ;
693 gi.gotErr[ b ]++ ;
694 gi_NumConnect[ b ]-- ;
695 gi.numConn[ b ]-- ;
696 gj.numConn[ b ]-- ;
697
698 } else if ( jrd<0 ) { //did not read anything
699 if (errno != EAGAIN && errno != EWOULDBLOCK ) {
700 snprintf(str,MXSTR,"Error Reading from %d | %m",i);
701 factOut(kError,442, str ) ;
702 gi.gotErr[ b ]++ ;
703 } else numok-- ; //else nothing waiting to be read
704 jrd = 0 ;
705 }
706 } else {
707 jrd = 0 ; //did read nothing as requested
708 snprintf(str,MXSTR,"do not read from socket %d %d",i,rd[i].bufLen ) ;
709 factOut(kDebug,301, str ) ;
710 }
711
712 gi.gotByte[ b ] += jrd ;
713 gj.rateBytes[b] += jrd ;
714
715 if (jrd>0) {numokx++ ; jrdx+= jrd; }
716
717
718 if ( rd[i].bufTyp <0 ) { // we are skipping this board ...
719// just do nothing
720
721 } else if ( rd[i].bufTyp >0 ) { // we are reading data ...
722 if ( jrd < rd[i].bufLen ) { //not yet all read
723 rd[i].bufPos += jrd ; //==> prepare for continuation
724 rd[i].bufLen -= jrd ;
725 debugRead(i,jrd,rd[i].evtID,rd[i].ftmID,rd[i].runID, 0,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 0=reading data
726 } else { //full dataset read
727 rd[i].bufLen = 0 ;
728 rd[i].bufPos = rd[i].fadLen ;
729 if ( rd[i].rBuf->B[ rd[i].bufPos-1] != stop.B[0]
730 || rd[i].rBuf->B[ rd[i].bufPos-2] != stop.B[1]) {
731 gi.evtErr++ ;
732 snprintf(str,MXSTR,"wrong end of buffer found sock %3d ev %4d len %5d %3d %3d - %3d %3d ",
733 i,rd[i].fadLen, rd[i].evtID, rd[i].rBuf->B[0], rd[i].rBuf->B[1], rd[i].rBuf->B[ rd[i].bufPos-2],
734 rd[i].rBuf->B[ rd[i].bufPos-1]);
735 factOut(kError,301, str ) ;
736 goto EndBuf ;
737
738// } else {
739// snprintf(str,MXSTR,"good end of buffer found sock %3d len %5d %d %d : %d %d - %d %d : %d %d",
740// i,rd[i].fadLen,
741// rd[i].rBuf->B[ 0 ], rd[i].rBuf->B[ 1 ], start.B[1],start.B[0],
742// rd[i].rBuf->B[ rd[i].bufPos-2], rd[i].rBuf->B[ rd[i].bufPos-1], stop.B[1], stop.B[0]);
743// factOut(kDebug,301, str ) ;
744 }
745
746 if (jrd>0) debugRead(i,jrd,rd[i].evtID,rd[i].ftmID,rd[i].runID, 1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
747
748 //we have a complete buffer, copy to WORK area
749 roi = ntohs(rd[i].rBuf->S[ head_len/2 + 2 ]) ;
750 //get index into mBuffer for this event (create if needed)
751 evID = mBufEvt( rd[i].evtID, rd[i].runID, roi, i, rd[i].fadLen ) ;
752
753 if (evID <-9000) goto EndBuf ; //illegal event, skip it ...
754 if (evID < 0) {
755
756 if ( rd[i].bufLen != 0) {
757 snprintf(str,MXSTR,"something screwed up");
758 factOut(kFatal, 1, str ) ;
759 }
760 xwait.tv_sec = 0;
761 xwait.tv_nsec= 10000000 ; // sleep for ~10 msec
762 nanosleep( &xwait , NULL ) ;
763 goto EndBuf1 ; //hope there is free space next round
764 }
765
766
767 //we have a valid entry in mBuffer[]; fill it
768
769int xchk = memcmp(&rd[i].xBuf->B[0], &rd[i].rBuf->B[0], rd[i].fadLen ) ;
770if (xchk != 0) {
771 snprintf(str,MXSTR,"ERROR OVERWRITE %d %d on port %d",xchk,rd[i].fadLen,i) ;
772 factOut(kFatal, 1, str ) ;
773
774 uint iq;
775 for (iq=0; iq < rd[i].fadLen ; iq++) {
776 if (rd[i].rBuf->B[iq] != rd[i].xBuf->B[iq] ) {
777 snprintf(str,MXSTR,"ERROR %4d %4d %x %x",i,iq,rd[i].rBuf->B[iq], rd[i].xBuf->B[iq]);
778 factOut(kFatal, 1, str ) ;
779 }
780 }
781}
782
783
784 boardId = b ;
785 int fadBoard = ntohs(rd[i].rBuf->S[12] ) ;
786 int fadCrate = fadBoard/256 ;
787 if (boardId != (fadCrate*10 + fadBoard%256) ) {
788 snprintf(str,MXSTR,"wrong Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ;
789 if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ; //print only few times
790 }
791 if ( mBuffer[evID].board[ boardId ] != -1) {
792 snprintf(str,MXSTR,"double board: ev %5d, b %3d, %3d; len %5d %3d %3d - %3d %3d ",
793 evID,boardId,i,rd[i].fadLen, rd[i].rBuf->B[0],
794 rd[i].rBuf->B[1], rd[i].rBuf->B[ rd[i].bufPos-2], rd[i].rBuf->B[ rd[i].bufPos-1]);
795 factOut(kWarn,501, str ) ;
796 goto EndBuf ; //--> skip Board
797 }
798
799 int iDx = evtIdx[evID] ; //index into evtCtrl
800
801 memcpy( &mBuffer[evID].FADhead[boardId].start_package_flag,
802 &rd[i].rBuf->S[0], head_len) ;
803 roi = mBuffer[evID].nRoi ;
804
805 pixS = boardId*36 -1 ; //
806 tmS = boardId*4 -1 ; //
807 src = head_len/2 ;
808 for ( px=0; px<9; px++ ) { //different sort in FAD board.....
809 for ( drs=0; drs<4; drs++ ) {
810 pixH= ntohs(rd[i].rBuf->S[src++]) ;
811 pixC= ntohs(rd[i].rBuf->S[src++]) ;
812 pixR= ntohs(rd[i].rBuf->S[src++]) ;
813
814 src++ ;
815 pixS++ ; //pixS = pixH2S[pixH] ;
816 if ( ( px < PX8 && pixR == roi )
817 || ( px ==PX8 && pixR == 2*roi )
818 || ( px ==PX8 && pixR == roi && roi > 512 ) ) {
819 // correct roi
820 mBuffer[evID].fEvent->StartPix[pixS] =pixC;
821 dest= pixS * roi ;
822 memcpy(
823 &mBuffer[evID].fEvent->Adc_Data[dest],
824 &rd[i].rBuf->S[src], roi * 2) ;
825 src+= roi ;
826 if ( px==PX8 ) {
827 tmS++; // tmS = tmH2S[pixH]
828 dest= tmS * roi + NPIX* roi ;
829 if ( roi <=512 ) {
830 mBuffer[evID].fEvent->StartTM[tmS] =(pixC+roi)%1024 ;
831 memcpy(
832 &mBuffer[evID].fEvent->Adc_Data[dest],
833 &rd[i].rBuf->S[src], roi * 2) ;
834 src+=roi ;
835 } else {
836 mBuffer[evID].fEvent->StartTM[tmS] = -1 ;
837 }
838 }
839 } else {
840 snprintf(str,MXSTR,"wrong roi %d %d %d %d",px,pixR,roi,src-2);
841 gi.evtErr++ ;
842 factOut(kError,202, str ) ;
843 goto EndBuf ;
844 }
845 }
846 }// now we have stored a new board contents into Event structure
847
848 mBuffer[evID].board[ boardId ] = boardId ;
849 evtCtrl.evtStat[ iDx ]++ ;
850 evtCtrl.pcTime[ iDx ] = g_actTime ;
851
852 if (++mBuffer[evID].nBoard >= actBoards ) {
853 snprintf(str,MXSTR,"%5d complete event %8d %8d %2d",mBuffer[evID].evNum,evtCtrl.evtBuf[iDx],iDx,evtCtrl.evtStat[ iDx ]);
854 factOut(kDebug,-1, str ) ;
855
856 if (mBuffer[evID].runNum != actrun ) { // have we already reported first event of this run ???
857 actrun = mBuffer[i].runNum ;
858 int ir ;
859 for ( ir=0; ir<MAX_RUN; ir++) {
860 if ( runCtrl[ir].runId == actrun) {
861 if ( runCtrl[ir].lastEvt <0 ) {
862 runCtrl[ir].lastEvt = +1 ;
863 gotNewRun( actrun, mBuffer[evID].FADhead );
864 snprintf(str,MXSTR,"gotNewRun %d (ev %d)",mBuffer[evID].runNum,mBuffer[evID].evNum);
865 factOut(kInfo,1, str ) ;
866 break ;
867 }
868 }
869 }
870 }
871
872 //complete event read ---> flag for next processing
873 evtCtrl.evtStat[ iDx ] = 99;
874 gi.evtTot++ ;
875 }
876
877EndBuf:
878 rd[i].bufTyp = 0 ; //ready to read next header
879 rd[i].bufLen = frst_len ;
880 rd[i].bufPos = 0 ;
881EndBuf1:
882 ;
883 }
884
885 } else { //we are reading event header
886 rd[i].bufPos += jrd ;
887 rd[i].bufLen -= jrd ;
888 if ( rd[i].bufPos >= minLen ){ //sufficient data to take action
889 //check if startflag correct; else shift block ....
890 for (k=0; k<rd[i].bufPos -1 ; k++) {
891 if (rd[i].rBuf->B[k ] == start.B[1]
892 && rd[i].rBuf->B[k+1] == start.B[0] ) break ;
893 }
894 rd[i].skip += k ;
895
896 if (k >= rd[i].bufPos-1 ) { //no start of header found
897 rd[i].bufPos = 0 ;
898 rd[i].bufLen = head_len ;
899 } else if ( k>0 ) {
900 rd[i].bufPos -= k ;
901 rd[i].bufLen += k ;
902 memcpy(&rd[i].rBuf->B[0], &rd[i].rBuf->B[k], rd[i].bufPos ) ;
903 memcpy(&rd[i].xBuf->B[0], &rd[i].xBuf->B[k], rd[i].bufPos ) ;
904 }
905 if ( rd[i].bufPos >= minLen ) {
906 if ( rd[i].skip >0 ) {
907 snprintf(str,MXSTR,"skipped %d bytes on port%d", rd[i].skip, i ) ;
908 factOut(kInfo,666, str ) ;
909 rd[i].skip = 0 ;
910 }
911 goodhed++;
912 rd[i].fadLen = ntohs(rd[i].rBuf->S[1])*2 ;
913 rd[i].fadVers= ntohs(rd[i].rBuf->S[2]) ;
914 rd[i].evtID = ntohl(rd[i].rBuf->I[4]) ; //(FADevt)
915 rd[i].ftmID = ntohl(rd[i].rBuf->I[5]) ; //(FTMevt)
916 rd[i].runID = ntohl(rd[i].rBuf->I[11]) ;
917 rd[i].bufTyp = 1 ; //ready to read full record
918 rd[i].bufLen = rd[i].fadLen - rd[i].bufPos ;
919
920 int fadboard = ntohs(rd[i].rBuf->S[12] ) ;
921 int fadcrate = fadboard/256 ;
922 fadboard = (fadcrate*10 + fadboard%256) ;
923snprintf(str,MXSTR,"sk %3d head: %5d %5d %5d %10d %4d %6d",i,rd[i].fadLen,rd[i].evtID,rd[i].ftmID,rd[i].runID,fadboard,jrd) ;
924factOut(kDebug,1, str ) ;
925
926 if (rd[i].runID ==0 ) rd[i].runID = gi_myRun ;
927
928 if (rd[i].bufLen <=head_len || rd[i].bufLen > MAX_LEN ) {
929 snprintf(str,MXSTR,"illegal event-length on port %d",i) ;
930 factOut(kFatal,881, str ) ;
931 rd[i].bufLen = 100000 ; //?
932 }
933 int fadBoard = ntohs(rd[i].rBuf->S[12] ) ;
934 debugHead(i,fadBoard,rd[i].rBuf);
935 debugRead(i,jrd,rd[i].evtID,rd[i].ftmID,rd[i].runID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event
936 } else {
937 debugRead(i,jrd,0,0,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
938 }
939 } else {
940 debugRead(i,jrd,0,0,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
941 }
942
943 } //end interpreting last read
944 } //end of successful read anything
945 } //finished trying to read all sockets
946
947snprintf(str,MXSTR,"Loop ---- %3d --- %8d",numokx,jrdx);
948factOut(kDebug,-1, str ) ;
949 gi.numRead[ numok ] ++ ;
950
951 g_actTime = time(NULL) ;
952 if ( g_actTime > gi_SecTime ) {
953 gi_SecTime = g_actTime ;
954
955
956 //loop over all active events and flag those older than read-timeout
957 //delete those that are written to disk ....
958
959 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
960 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
961
962 gj.bufNew = gj.bufEvt = 0 ;
963 int k1=evtCtrl.frstPtr;
964 for ( k=k1; k<(k1+kd); k++ ) {
965 int k0 = k % (MAX_EVT*MAX_RUN) ;
966//would be better to use bitmaps for evtStat (allow '&' instead of multi-if)
967
968 if (evtCtrl.evtStat[k0] > 0
969 && evtCtrl.evtStat[k0] < 92 ) {
970 gj.bufNew++ ; //incomplete event in Buffer
971 if ( evtCtrl.evtStat[k0] < 90
972 && evtCtrl.pcTime[k0] < g_actTime-10 ) {
973 int id =evtCtrl.evtBuf[k0] ;
974 snprintf(str,MXSTR,"%5d skip short evt %8d %8d %2d",mBuffer[id].evNum,evtCtrl.evtBuf[k0],k0 ,evtCtrl.evtStat[k0]);
975 factOut(kWarn,601, str ) ;
976 evtCtrl.evtStat[k0] = 91 ; //timeout for incomplete events
977 gi.evtSkp++ ;
978 gi.evtTot++ ;
979 gj.evtSkip++;
980 }
981 } else if (evtCtrl.evtStat[k0] >= 900 //'delete'
982 || evtCtrl.evtStat[k0] == 0 ) { //'useless'
983
984 int id =evtCtrl.evtBuf[k0] ;
985 snprintf(str,MXSTR,"%5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard ) ;
986 factOut(kDebug,-1, str ) ;
987 mBufFree(id) ; //event written--> free memory
988 evtCtrl.evtStat[k0] = -1;
989 gj.evtWrite++ ;
990 gj.rateWrite++ ;
991 } else if (evtCtrl.evtStat[k0] >= 95 ) {
992 gj.bufEvt++ ; //complete event in Buffer
993 }
994
995 if ( k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] <0 ) {
996 evtCtrl.frstPtr = (evtCtrl.frstPtr+1) % (MAX_EVT*MAX_RUN) ;
997 }
998 }
999
1000
1001 gj.deltaT = 1000 ; //temporary, must be improved
1002
1003 int b;
1004 for ( b=0; b<NBOARDS; b++) gj.totBytes[b] +=gj.rateBytes[b] ;
1005 gj.totMem = g_maxMem ;
1006 if (gj.maxMem >= gj.xxxMem) gj.xxxMem = gj.maxMem ;
1007
1008 factStat(gj);
1009 factStatNew(gi) ;
1010
1011 gj.rateNew = gj.rateWrite = 0 ;
1012 for ( b=0; b<NBOARDS; b++) gj.rateBytes[b] =0 ;
1013 }
1014
1015 if (numok > 0 ) numok2=0;
1016 else if (numok2++ > 3) {
1017 if (g_runStat == 1) {
1018 xwait.tv_sec = 1;
1019 xwait.tv_nsec= 0 ; // hibernate for 1 sec
1020 } else {
1021 xwait.tv_sec = 0;
1022 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1023 }
1024 nanosleep( &xwait , NULL ) ;
1025 }
1026
1027 } //and do next loop over all sockets ...
1028
1029
1030 snprintf(str,MXSTR,"stop reading ... RESET=%d",g_reset);
1031 factOut(kInfo,-1, str ) ;
1032
1033 if (g_reset >0 ) {
1034 gi_reset = g_reset ;
1035 gi_resetR = gi_reset%10 ; //shall we stop reading ?
1036 gi_resetS = (gi_reset/10)%10 ; //shall we close sockets ?
1037 gi_resetW = (gi_reset/100)%10 ; //shall we close files ?
1038 gi_resetX = gi_reset/1000 ; //shall we simply wait resetX seconds ?
1039 g_reset= 0;
1040 } else {
1041 gi_reset = 0;
1042 if ( g_runStat== -1 ) gi_resetR = 1 ;
1043 else gi_resetR = 7 ;
1044 gi_resetS = 7 ; //close all sockets
1045 gi_resetW = 7 ; //close all files
1046 gi_resetX = 0 ;
1047
1048 //inform others we have to quit ....
1049 gi_runStat = -11 ; //inform all that no update to happen any more
1050 gj.readStat= -11 ; //inform all that no update to happen any more
1051 }
1052
1053 if (gi_resetS > 0) {
1054 //must close all open sockets ...
1055 snprintf(str,MXSTR,"close all sockets ...");
1056 factOut(kInfo,-1, str ) ;
1057 for (i=0; i<MAX_SOCK; i++) {
1058 if (rd[i].sockStat ==0 ) {
1059 GenSock(-1, i, 0, NULL, &rd[i]) ; //close and destroy open socket
1060 if ( i%7 == 0 ) {
1061 gi_NumConnect[ i/7 ]-- ;
1062 gi.numConn[ i/7 ]-- ;
1063 gj.numConn[ i/7 ]-- ;
1064 sockDef[i/7] = 0 ; //flag ro recreate the sockets ...
1065 rd[i/7].sockStat = -1; //and try to open asap
1066 }
1067 }
1068 }
1069 }
1070
1071
1072 if (gi_resetR > 0) {
1073 //flag all events as 'read finished'
1074 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1075 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1076
1077 int k1=evtCtrl.frstPtr;
1078 for ( k=k1; k<(k1+kd); k++ ) {
1079 int k0 = k % (MAX_EVT*MAX_RUN) ;
1080 if (evtCtrl.evtStat[k0] > 0
1081 && evtCtrl.evtStat[k0] < 90 ) {
1082 evtCtrl.evtStat[k0] = 91 ;
1083 gi.evtSkp++ ;
1084 gi.evtTot++ ;
1085 }
1086 }
1087
1088 xwait.tv_sec = 0;
1089 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1090 nanosleep( &xwait , NULL ) ;
1091
1092 //and clear all buffers (might have to wait until all others are done)
1093 int minclear ;
1094 if (gi_resetR == 1) {
1095 minclear = 900 ;
1096 snprintf(str,MXSTR,"drain all buffers ...");
1097 } else {
1098 minclear = 0 ;
1099 snprintf(str,MXSTR,"flush all buffers ...");
1100 }
1101 factOut(kInfo,-1, str ) ;
1102
1103 int numclear=1 ;
1104 while (numclear > 0 ) {
1105 numclear = 0 ;
1106 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1107 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1108
1109 int k1=evtCtrl.frstPtr;
1110 for ( k=k1; k<(k1+kd); k++ ) {
1111 int k0 = k % (MAX_EVT*MAX_RUN) ;
1112 if (evtCtrl.evtStat[k0] > minclear ) {
1113 int id =evtCtrl.evtBuf[k0] ;
1114 snprintf(str,MXSTR,"ev %5d free event buffer, nb=%3d", mBuffer[id].evNum, mBuffer[id].nBoard ) ;
1115 factOut(kDebug,-1, str ) ;
1116 mBufFree(id) ; //event written--> free memory
1117 evtCtrl.evtStat[k0] = -1;
1118 } else if (evtCtrl.evtStat[k0] > 0) numclear++ ; //writing is still ongoing...
1119
1120 if ( k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] <0 )
1121 evtCtrl.frstPtr = (evtCtrl.frstPtr+1) % (MAX_EVT*MAX_RUN) ;
1122 }
1123
1124 xwait.tv_sec = 0;
1125 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1126 nanosleep( &xwait , NULL ) ;
1127 }
1128 }
1129
1130 if (gi_reset > 0) {
1131 if (gi_resetW > 0) {
1132 CloseRunFile(0,0,0) ; //ask all Runs to be closed
1133 }
1134 if (gi_resetX > 0) {
1135 xwait.tv_sec = gi_resetX;
1136 xwait.tv_nsec= 0 ;
1137 nanosleep( &xwait , NULL ) ;
1138 }
1139
1140 snprintf(str,MXSTR,"Continue read Process ...");
1141 factOut(kInfo,-1, str ) ;
1142 gi_reset = 0 ;
1143 goto START ;
1144 }
1145
1146
1147
1148 snprintf(str,MXSTR,"Exit read Process ...");
1149 factOut(kInfo,-1, str ) ;
1150 gi_runStat = -99 ;
1151 gj.readStat= -99 ;
1152 factStat(gj);
1153 factStatNew(gi) ;
1154 return 0;
1155
1156} /*-----------------------------------------------------------------*/
1157
1158
1159void *procEvt( void *ptr ) {
1160/* *** main loop processing file, including SW-trigger */
1161 int numProc, numWait ;
1162 int k ;
1163 struct timespec xwait ;
1164 char str[MXSTR] ;
1165
1166 cpu_set_t mask;
1167 int cpu = 5 ; //process thread (will be several in final version)
1168
1169 snprintf(str,MXSTR,"Starting process-thread");
1170 factOut(kInfo,-1, str ) ;
1171
1172/* CPU_ZERO initializes all the bits in the mask to zero. */
1173 CPU_ZERO( &mask );
1174/* CPU_SET sets only the bit corresponding to cpu. */
1175 CPU_SET( cpu, &mask );
1176/* sched_setaffinity returns 0 in success */
1177 if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
1178 snprintf(str,MXSTR,"P ---> can not create affinity to %d",cpu);
1179 factOut(kWarn,-1, str ) ;
1180 }
1181
1182
1183 while (g_runStat > -2) { //in case of 'exit' we still must process pending events
1184
1185 numWait = numProc = 0 ;
1186 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1187 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1188
1189 int k1=evtCtrl.frstPtr;
1190 for ( k=k1; k<(k1+kd); k++ ) {
1191 int k0 = k % (MAX_EVT*MAX_RUN) ;
1192//would be better to use bitmaps for evtStat (allow '&' instead of multi-if)
1193 if (evtCtrl.evtStat[k0] > 90 && evtCtrl.evtStat[k0] <500) {
1194
1195 if ( gi_resetR > 1 ) { //we are asked to flush buffers asap
1196 evtCtrl.evtStat[k0] = 991 ;
1197 } else {
1198
1199 int id = evtCtrl.evtBuf[k0] ;
1200 int ievt = mBuffer[id].evNum ;
1201 int roi = mBuffer[id].nRoi ;
1202// uint32_t irun = mBuffer[id].runNum ;
1203//snprintf(str,MXSTR,"P processing %d %d %d %d",ievt,k,id,evtCtrl.evtStat[k0]) ;
1204//factOut(kDebug,-1, str ) ;
1205
1206//make sure unused pixels/tmarks are cleared to zero
1207 int ip,it,dest,ib;
1208 for (ip=0; ip<NPIX; ip++) {
1209 if (mBuffer[id].fEvent->StartPix[ip] == -1 ) {
1210 dest= ip*roi ;
1211 bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ;
1212 }
1213 }
1214 for (it=0; it<NTMARK; it++) {
1215 if (mBuffer[id].fEvent->StartTM[it] == -1 ) {
1216 dest= it*roi + NPIX*roi ;
1217 bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ;
1218 }
1219 }
1220
1221
1222//and set correct event header ; also check for consistency in event (not yet)
1223 mBuffer[id].fEvent->Roi = roi ;
1224 mBuffer[id].fEvent->EventNum = ievt ;
1225 mBuffer[id].fEvent->TriggerType = 0 ; // TBD
1226 mBuffer[id].fEvent->SoftTrig = 0 ;
1227 for (ib=0; ib<NBOARDS; ib++) {
1228 if (mBuffer[id].board[ib] == -1 ) { //board is not read
1229 mBuffer[id].FADhead[ib].start_package_flag = 0 ;
1230 mBuffer[id].fEvent->BoardTime[ib] = 0 ;
1231 } else {
1232 mBuffer[id].fEvent->BoardTime[ib] =
1233 ntohl(mBuffer[id].FADhead[ib].time) ;
1234 }
1235 }
1236
1237 int i=eventCheck(mBuffer[id].FADhead,mBuffer[id].fEvent) ;
1238// gj.procEvt++ ;
1239 gi.procTot++ ;
1240 numProc++ ;
1241
1242 if (i<0) {
1243 evtCtrl.evtStat[k0] = 999 ; //flag event to be skipped
1244 gi.procErr++ ;
1245 } else {
1246 evtCtrl.evtStat[k0] = 520 ;
1247 }
1248 }
1249 } else if ( evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] < 90 ) {
1250 numWait++ ;
1251 }
1252 }
1253
1254 if ( gj.readStat < -10 && numWait == 0) { //nothing left to do
1255 snprintf(str,MXSTR,"Exit Processing Process ...");
1256 factOut(kInfo,-1, str ) ;
1257 gp_runStat = -22 ; //==> we should exit
1258 gj.procStat= -22 ; //==> we should exit
1259 return 0 ;
1260 }
1261
1262 if (numProc == 0) {
1263 //seems we have nothing to do, so sleep a little
1264 xwait.tv_sec = 0;
1265 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1266 nanosleep( &xwait , NULL ) ;
1267 }
1268 gp_runStat = gi_runStat ;
1269 gj.procStat= gj.readStat ;
1270
1271 }
1272
1273 //we are asked to abort asap ==> must flag all remaining events
1274 // when gi_runStat claims that all events are in the buffer...
1275
1276 snprintf(str,MXSTR,"Abort Processing Process ...");
1277 factOut(kInfo,-1, str ) ;
1278 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1279 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1280
1281 int k1=evtCtrl.frstPtr;
1282 for ( k=k1; k<(k1+kd); k++ ) {
1283 int k0 = k % (MAX_EVT*MAX_RUN) ;
1284 if (evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] <500) {
1285 evtCtrl.evtStat[k0] = 555 ; //flag event as 'processed'
1286 }
1287 }
1288
1289 gp_runStat = -99 ;
1290 gj.procStat= -99 ;
1291
1292 return 0;
1293
1294} /*-----------------------------------------------------------------*/
1295
1296int CloseRunFile(uint32_t runId, uint32_t closeTime, uint32_t maxEvt) {
1297/* close run runId (all all runs if runId=0) */
1298/* return: 0=close scheduled / >0 already closed / <0 does not exist */
1299 int j ;
1300
1301
1302 if ( runId == 0 ) {
1303 for ( j=0; j<MAX_RUN; j++) {
1304 if ( runCtrl[j].fileId == 0 ) { //run is open
1305 runCtrl[j].closeTime = closeTime ;
1306 runCtrl[j].maxEvt = maxEvt ;
1307 }
1308 }
1309 return 0;
1310 }
1311
1312 for ( j=0; j<MAX_RUN; j++) {
1313 if ( runCtrl[j].runId == runId ) {
1314 if ( runCtrl[j].fileId == 0 ) { //run is open
1315 runCtrl[j].closeTime = closeTime ;
1316 runCtrl[j].maxEvt = maxEvt ;
1317 return 0;
1318 } else if ( runCtrl[j].fileId <0 ) { //run not yet opened
1319 runCtrl[j].closeTime = closeTime ;
1320 runCtrl[j].maxEvt = maxEvt ;
1321 return +1;
1322 } else { // run already closed
1323 return +2;
1324 }
1325 }
1326 } //we only reach here if the run was never created
1327 return -1;
1328
1329} /*-----------------------------------------------------------------*/
1330
1331
1332void *writeEvt( void *ptr ) {
1333/* *** main loop writing event (including opening and closing run-files */
1334
1335 int numWrite, numWait ;
1336 int k,j,i ;
1337 struct timespec xwait ;
1338 char str[MXSTR] ;
1339
1340 cpu_set_t mask;
1341 int cpu = 3 ; //write thread
1342
1343 snprintf(str,MXSTR,"Starting write-thread");
1344 factOut(kInfo,-1, str ) ;
1345
1346/* CPU_ZERO initializes all the bits in the mask to zero. */
1347 CPU_ZERO( &mask );
1348/* CPU_SET sets only the bit corresponding to cpu. */
1349 CPU_SET( cpu, &mask );
1350/* sched_setaffinity returns 0 in success */
1351 if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
1352 snprintf(str,MXSTR,"W ---> can not create affinity to %d",cpu);
1353 }
1354
1355 int lastRun = 0 ; //usually run from last event still valid
1356
1357 while (g_runStat >-2) {
1358
1359 numWait = numWrite = 0 ;
1360 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1361 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1362
1363 int k1=evtCtrl.frstPtr;
1364 for ( k=k1; k<(k1+kd); k++ ) {
1365 int k0 = k % (MAX_EVT*MAX_RUN) ;
1366//would be better to use bitmaps for evtStat (allow '&' instead of multi-if)
1367 if (evtCtrl.evtStat[k0] > 500 && evtCtrl.evtStat[k0] < 900) {
1368
1369 if ( gi_resetR > 1 ) { //we must drain the buffer asap
1370 evtCtrl.evtStat[k0] = 904 ;
1371 } else {
1372
1373
1374 int id = evtCtrl.evtBuf[k0] ;
1375 uint32_t irun = mBuffer[id].runNum ;
1376 int ievt = mBuffer[id].evNum ;
1377
1378 gi.wrtTot++ ;
1379 if (runCtrl[lastRun].runId == irun ) {
1380 j = lastRun ;
1381 } else {
1382 //check which fileID to use (or open if needed)
1383 for ( j=0; j<MAX_RUN; j++) {
1384 if ( runCtrl[j].runId == irun ) break ;
1385 }
1386 if ( j >= MAX_RUN ) {
1387 snprintf(str,MXSTR,"W error: can not find run %d for event %d in %d", irun,ievt,id);
1388 factOut(kFatal,901, str ) ;
1389 gi.wrtErr++ ;
1390for ( j=0; j<MAX_RUN; j++) printf("j %d run.j %d run %d\n",j,runCtrl[j].runId,irun );
1391exit(111);
1392 }
1393 lastRun = j ;
1394 }
1395
1396 if (runCtrl[j].fileId < 0 ) {
1397 actRun.Version = 1 ;
1398 actRun.RunType = -1 ;
1399 actRun.NBoard = NBOARDS ;
1400 actRun.NPix = NPIX ;
1401 actRun.NTm = NTMARK ;
1402 actRun.Nroi = mBuffer[id].nRoi ;
1403 memcpy(actRun.FADhead, mBuffer[id].FADhead, NBOARDS* sizeof(PEVNT_HEADER) ) ;
1404
1405 runCtrl[j].fileHd = runOpen(irun, &actRun, sizeof(actRun) ) ;
1406 if (runCtrl[j].fileHd == NULL ) {
1407 snprintf(str,MXSTR,"W could not open a file for run %d",irun);
1408 factOut(kError,502, str ) ;
1409 runCtrl[j].fileId = 91 ;
1410 } else {
1411 snprintf(str,MXSTR,"W opened new run_file %d",irun) ;
1412 factOut(kInfo,-1, str ) ;
1413 runCtrl[j].fileId = 0 ;
1414 }
1415
1416 }
1417
1418 if (runCtrl[j].fileId != 0 ) {
1419 if (runCtrl[j].fileId < 0 ) {
1420 snprintf(str,MXSTR,"W never opened file for this run %d",irun) ;
1421 factOut(kError,123,str) ;
1422 } else if (runCtrl[j].fileId < 100 ) {
1423 snprintf(str,MXSTR,"W.file for this run is closed %d",irun) ;
1424 factOut(kWarn,123,str) ;
1425 runCtrl[j].fileId += 100 ;
1426 } else {
1427 snprintf(str,MXSTR,"W:file for this run is closed %d",irun) ;
1428 factOut(kDebug,123,str) ;
1429 }
1430 evtCtrl.evtStat[k0] = 903 ;
1431 gi.wrtErr++ ;
1432 } else {
1433 i=runWrite(runCtrl[j].fileHd, mBuffer[id].fEvent, sizeof(mBuffer[id]) );
1434 if ( i>=0 ) {
1435 runCtrl[j].lastTime = g_actTime;
1436 runCtrl[j].actEvt++ ;
1437 evtCtrl.evtStat[k0] = 901 ;
1438 snprintf(str,MXSTR,"%5d successfully wrote for run %d id %5d",ievt,irun,k0);
1439 factOut(kDebug,504, str ) ;
1440// gj.writEvt++ ;
1441 } else {
1442 snprintf(str,MXSTR,"W error writing event for run %d",irun) ;
1443 factOut(kError,503, str ) ;
1444 evtCtrl.evtStat[k0] = 902 ;
1445 gi.wrtErr++ ;
1446 }
1447
1448 if ( i < 0
1449 || runCtrl[j].lastTime < g_actTime-300
1450 || runCtrl[j].closeTime < g_actTime
1451 || runCtrl[j].maxEvt < runCtrl[j].actEvt ) {
1452int ii =0 ;
1453if ( i < 0 ) ii=1 ;
1454else if (runCtrl[j].closeTime < g_actTime ) ii=2 ;
1455else if (runCtrl[j].lastTime < g_actTime-300 ) ii=3 ;
1456else if (runCtrl[j].maxEvt < runCtrl[j].actEvt ) ii=4 ;
1457
1458
1459
1460 //close run for whatever reason
1461 if (runCtrl[j].runId == gi_myRun) gi_myRun = g_actTime ;
1462 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1463 if (i<0) {
1464 snprintf(str,MXSTR,"error closing run %d %d AAA",runCtrl[j].runId,i) ;
1465 factOut(kError,503, str ) ;
1466 runCtrl[j].fileId = 92 ;
1467 } else {
1468 snprintf(str,MXSTR,"W closed run %d AAA %d",irun,ii) ;
1469 factOut(kInfo,503, str ) ;
1470 runCtrl[j].fileId = 93 ;
1471 }
1472 }
1473 }
1474 }
1475 } else if (evtCtrl.evtStat[k0] > 0
1476 && evtCtrl.evtStat[k0] < 900 ) numWait++ ;
1477 }
1478
1479 //check if we should close a run (mainly when no event pending)
1480 for ( j=0; j<MAX_RUN; j++) {
1481 if ( runCtrl[j].fileId==0
1482 && ( runCtrl[j].closeTime < g_actTime
1483 ||runCtrl[j].lastTime < g_actTime-300
1484 ||runCtrl[j].maxEvt < runCtrl[j].actEvt ) ) {
1485 if (runCtrl[j].runId == gi_myRun) gi_myRun = g_actTime ;
1486int ii =0 ;
1487 if (runCtrl[j].closeTime < g_actTime ) ii=2 ;
1488else if (runCtrl[j].lastTime < g_actTime-300 ) ii=3 ;
1489else if (runCtrl[j].maxEvt < runCtrl[j].actEvt ) ii=4 ;
1490
1491
1492 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1493 if (i<0) {
1494 snprintf(str,MXSTR,"error closing run %d %d BBB",runCtrl[j].runId,i) ;
1495 factOut(kError,506, str ) ;
1496 runCtrl[j].fileId = 94 ;
1497 } else {
1498 snprintf(str,MXSTR,"W closed run %d BBB %d",runCtrl[j].runId,ii) ;
1499 factOut(kInfo,507, str ) ;
1500 runCtrl[j].fileId = 95 ;
1501 }
1502 }
1503 }
1504
1505 if (numWrite == 0) {
1506 //seems we have nothing to do, so sleep a little
1507 xwait.tv_sec = 0;
1508 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1509 nanosleep( &xwait , NULL ) ;
1510 }
1511
1512 if ( gj.readStat < -10 && numWait == 0) { //nothing left to do
1513 snprintf(str,MXSTR,"Finish Write Process ...");
1514 factOut(kInfo,-1, str ) ;
1515 gw_runStat = -22 ; //==> we should exit
1516 gj.writStat= -22 ; //==> we should exit
1517 goto closerun ;
1518 }
1519 gw_runStat = gi_runStat ;
1520 gj.writStat= gj.readStat ;
1521
1522 }
1523
1524 //must close all open files ....
1525 snprintf(str,MXSTR,"Abort Writing Process ...");
1526 factOut(kInfo,-1, str ) ;
1527
1528closerun:
1529 snprintf(str,MXSTR,"Close all open files ...");
1530 factOut(kInfo,-1, str ) ;
1531 for ( j=0; j<MAX_RUN; j++)
1532 if ( runCtrl[j].fileId ==0 ) {
1533 if (runCtrl[j].runId == gi_myRun) gi_myRun = g_actTime ;
1534 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1535int ii =0 ;
1536 if (runCtrl[j].closeTime < g_actTime ) ii=2 ;
1537else if (runCtrl[j].lastTime < g_actTime-300 ) ii=3 ;
1538else if (runCtrl[j].maxEvt < runCtrl[j].actEvt ) ii=4 ;
1539 if (i<0) {
1540 snprintf(str,MXSTR,"error closing run %d %d CCC",runCtrl[j].runId,i) ;
1541 factOut(kError,506, str ) ;
1542 runCtrl[j].fileId = 96 ;
1543 } else {
1544 snprintf(str,MXSTR,"W closed run %d CCC %d",runCtrl[j].runId,ii) ;
1545 factOut(kInfo,507, str ) ;
1546 runCtrl[j].fileId = 97 ;
1547 }
1548 }
1549
1550 gw_runStat = -99;
1551 gj.writStat= -99;
1552 snprintf(str,MXSTR,"Exit Writing Process ...");
1553 factOut(kInfo,-1, str ) ;
1554 return 0;
1555
1556
1557
1558
1559} /*-----------------------------------------------------------------*/
1560
1561
1562
1563
1564void StartEvtBuild() {
1565
1566 int i,j,imax,status,th_ret[50] ;
1567 pthread_t thread[50] ;
1568 struct timespec xwait ;
1569
1570 gi_runStat = gp_runStat = gw_runStat = 0 ;
1571 gj.readStat= gj.procStat= gj.writStat= 0 ;
1572
1573 snprintf(str,MXSTR,"Starting EventBuilder V15.07 A");
1574 factOut(kInfo,-1, str ) ;
1575
1576//initialize run control logics
1577 for (i=0; i<MAX_RUN; i++) {
1578 runCtrl[i].runId = 0 ;
1579 runCtrl[i].fileId = -2 ;
1580 }
1581
1582//partially initialize event control logics
1583 evtCtrl.frstPtr = 0 ;
1584 evtCtrl.lastPtr = 0 ;
1585
1586//start all threads (more to come) when we are allowed to ....
1587 while (g_runStat == 0 ) {
1588 xwait.tv_sec = 0;
1589 xwait.tv_nsec= 10000000 ; // sleep for ~10 msec
1590 nanosleep( &xwait , NULL ) ;
1591 }
1592
1593 i=0 ;
1594 th_ret[i] = pthread_create( &thread[i], NULL, readFAD, NULL );
1595 i++;
1596 th_ret[i] = pthread_create( &thread[i], NULL, procEvt, NULL );
1597 i++;
1598 th_ret[i] = pthread_create( &thread[i], NULL, writeEvt, NULL );
1599 i++;
1600 imax=i ;
1601
1602
1603#ifdef BILAND
1604 xwait.tv_sec = 30;;
1605 xwait.tv_nsec= 0 ; // sleep for ~20sec
1606 nanosleep( &xwait , NULL ) ;
1607
1608 printf("close all runs in 2 seconds\n");
1609
1610 CloseRunFile( 0, time(NULL)+2, 0) ;
1611
1612 xwait.tv_sec = 1;;
1613 xwait.tv_nsec= 0 ; // sleep for ~20sec
1614 nanosleep( &xwait , NULL ) ;
1615
1616 printf("setting g_runstat to -1\n");
1617
1618 g_runStat = -1 ;
1619#endif
1620
1621
1622//wait for all threads to finish
1623 for (i=0; i<imax; i++) {
1624 j = pthread_join ( thread[i], (void **)&status) ;
1625 }
1626
1627} /*-----------------------------------------------------------------*/
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643 /*-----------------------------------------------------------------*/
1644 /*-----------------------------------------------------------------*/
1645 /*-----------------------------------------------------------------*/
1646 /*-----------------------------------------------------------------*/
1647 /*-----------------------------------------------------------------*/
1648
1649
1650#ifdef BILAND
1651
1652
1653 /*-----------------------------------------------------------------*/
1654 /*-----------------------------------------------------------------*/
1655 /*-----------------------------------------------------------------*/
1656 /*-----------------------------------------------------------------*/
1657 /*-----------------------------------------------------------------*/
1658
1659
1660
1661
1662FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len )
1663{ return 1; } ;
1664
1665int runWrite(FileHandle_t fileHd , EVENT *event, size_t len )
1666{ return 1; usleep(10000); return 1; }
1667
1668
1669//{ return 1; } ;
1670
1671int runClose(FileHandle_t fileHd , RUN_TAIL *runth, size_t len )
1672{ return 1; } ;
1673
1674
1675
1676int eventCheck( PEVNT_HEADER *fadhd, EVENT *event)
1677{
1678 int i=0;
1679
1680// printf("------------%d\n",ntohl(fadhd[7].fad_evt_counter) );
1681// for (i=0; i<NBOARDS; i++) {
1682// printf("b=%2d,=%5d\n",i,fadhd[i].board_id);
1683// }
1684 return 0;
1685}
1686
1687
1688void factStatNew(EVT_STAT gi) {
1689 int i ;
1690
1691//for (i=0;i<MAX_SOCK;i++) {
1692// printf("%4d",gi.numRead[i]);
1693// if (i%20 == 0 ) printf("\n");
1694//}
1695}
1696
1697void gotNewRun( int runnr, PEVNT_HEADER *headers )
1698{ printf("got new run %d\n",runnr); return; }
1699
1700void factStat(GUI_STAT gj) {
1701// printf("stat: bfr%5lu skp%4lu free%4lu (tot%7lu) mem%12lu rd%12lu %3lu\n",
1702// array[0],array[1],array[2],array[3],array[4],array[5],array[6]);
1703}
1704
1705
1706void debugRead(int isock, int ibyte, int32_t event, int32_t ftmevt, int32_t runnr, int state, uint32_t tsec, uint32_t tusec ) {
1707// printf("%3d %5d %9d %3d %12d\n",isock, ibyte, event, state, tusec) ;
1708}
1709
1710
1711
1712void debugStream(int isock, void *buf, int len) {
1713}
1714
1715void debugHead(int i, int j, void *buf) {
1716}
1717
1718
1719void factOut(int severity, int err, char* message ) {
1720static FILE * fd ;
1721static int file=0 ;
1722
1723 if (file==0) {
1724 printf("open file\n");
1725 fd=fopen("x.out","w+") ;
1726 file=999;
1727 }
1728
1729 fprintf(fd,"%3d %3d | %s \n",severity,err,message) ;
1730
1731 if (severity != kDebug)
1732 printf("%3d %3d | %s\n",severity,err,message) ;
1733}
1734
1735
1736
1737int main() {
1738 int i,b,c,p ;
1739 char ipStr[100] ;
1740 struct in_addr IPaddr ;
1741
1742 g_maxMem = 1024*1024 ; //MBytes
1743//g_maxMem = g_maxMem * 1024 *10 ; //10GBytes
1744 g_maxMem = g_maxMem * 200; //100MBytes
1745
1746
1747 g_runStat = 40 ;
1748
1749 i=0 ;
1750
1751// version for standard crates
1752//for (c=0; c<4,c++) {
1753// for (b=0; b<10; b++) {
1754// sprintf(ipStr,"10.0.%d.%d",128+c,128+b)
1755//
1756// inet_pton(PF_INET, ipStr, &IPaddr) ;
1757//
1758// g_port[i].sockAddr.sin_family = PF_INET;
1759// g_port[i].sockAddr.sin_port = htons(5000) ;
1760// g_port[i].sockAddr.sin_addr = IPaddr ;
1761// g_port[i].sockDef = 1 ;
1762// i++ ;
1763// }
1764//}
1765//
1766//version for PC-test *
1767 for (c=0; c<4; c++) {
1768 for (b=0; b<10; b++) {
1769 sprintf(ipStr,"10.0.%d.11",128+c) ;
1770 if (c<2) sprintf(ipStr,"10.0.%d.11",128) ;
1771 else sprintf(ipStr,"10.0.%d.11",131) ;
1772// if (c==0) sprintf(ipStr,"10.0.100.11") ;
1773
1774 inet_pton(PF_INET, ipStr, &IPaddr) ;
1775 p = 31919+100*c+10*b;
1776
1777
1778 g_port[i].sockAddr.sin_family = PF_INET;
1779 g_port[i].sockAddr.sin_port = htons(p) ;
1780 g_port[i].sockAddr.sin_addr = IPaddr ;
1781 g_port[i].sockDef = 1 ;
1782
1783 i++ ;
1784 }
1785 }
1786
1787
1788//g_port[17].sockDef =-1 ;
1789//g_actBoards-- ;
1790
1791 StartEvtBuild() ;
1792
1793 return 0;
1794
1795}
1796#endif
Note: See TracBrowser for help on using the repository browser.