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

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