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

Last change on this file since 11094 was 11094, checked in by tbretz, 13 years ago
Don't count NumConnected down if already 0.
File size: 43.7 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 64*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 factStat(int severity, int err, char* message ) ;
44
45extern int eventCheck( PEVNT_HEADER *fadhd, EVENT *event) ;
46
47extern void debugRead(int isock, int ibyte, int32_t event, int state,
48 uint32_t tsec, uint32_t tusec ) ;
49extern void debugStream(int isock, void *buf, int len) ;
50
51
52
53int g_actTime = 0 ;
54int g_runStat = 40 ;
55size_t g_maxMem ; //maximum memory allowed for buffer
56
57//no longer needed ...
58 int g_maxBoards ; //maximum number of boards to be initialized
59 int g_actBoards ;
60//
61
62FACT_SOCK g_port[NBOARDS] ; // .addr=string of IP-addr in dotted-decimal "ddd.ddd.ddd.ddd"
63
64
65int gi_runStat ;
66int gp_runStat ;
67int gw_runStat ;
68
69
70uint gi_SecRate[MAX_SOCK] ;
71uint gi_S10Rate[MAX_SOCK] ;
72uint gi_MinRate[MAX_SOCK] ;
73uint gi_ErrCnt[MAX_SOCK] ;
74
75uint gi_NumConnect[NBOARDS]; //4 crates * 10 boards
76
77uint gi_SecTime, gi_S10Time, gi_MinTime ;
78uint gi_EvtStart= 0 ;
79uint gi_EvtRead = 0 ;
80uint gi_EvtBad = 0 ;
81uint gi_EvtTot = 0 ;
82size_t gi_usedMem = 0 ;
83
84uint gw_EvtTot = 0 ;
85uint gp_EvtTot = 0 ;
86
87PIX_MAP g_pixMap[NPIX] ;
88
89EVT_CTRL evtCtrl ; //control of events during processing
90int evtIdx[MAX_EVT*MAX_RUN] ; //index from mBuffer to evtCtrl
91
92WRK_DATA mBuffer[MAX_EVT*MAX_RUN]; //local working space
93
94
95
96
97RUN_HEAD actRun ;
98
99RUN_CTRL runCtrl[MAX_RUN] ;
100
101RUN_TAIL runTail[MAX_RUN] ;
102
103
104/*
105*** Definition of rdBuffer to read in IP packets; keep it global !!!!
106 */
107
108
109typedef union {
110 int8_t B[MAX_LEN/8];
111 int16_t S[MAX_LEN/4];
112 int32_t I[MAX_LEN/2];
113 int64_t L[MAX_LEN ];
114} CNV_FACT ;
115
116typedef struct {
117 int bufTyp ; //what are we reading at the moment: 0=header 1=data -1=skip ...
118 int32_t bufPos ; //next byte to read to the buffer next
119 int32_t bufLen ; //number of bytes left to read
120
121 int sockStat ; //-1 if socket not yet connected , 99 if not exist
122 int socket ; //contains the sockets
123 struct sockaddr_in SockAddr ; //IP for each socket
124
125 int evtID ; // event ID of event currently read
126 int runID ; // run "
127 uint fadLen ; // FADlength of event currently read
128 int fadVers ; // Version of FAD
129 int board ; // boardID (softwareID: 0..40 )
130 int Port ;
131
132 CNV_FACT *rBuf ;
133
134} READ_STRUCT ;
135
136
137typedef union {
138 int8_t B[2];
139 int16_t S ;
140} SHORT_BYTE ;
141
142
143
144
145
146#define MXSTR 1000
147char str[MXSTR] ;
148
149SHORT_BYTE start, stop;
150
151READ_STRUCT rd[MAX_SOCK] ; //buffer to read IP and afterwards store in mBuffer
152
153
154
155/*-----------------------------------------------------------------*/
156
157
158/*-----------------------------------------------------------------*/
159
160
161
162int GenSock(int flag, int sid, int port, struct sockaddr_in *sockAddr, READ_STRUCT *rd) {
163/*
164*** generate Address, create sockets and allocates readbuffer for it
165***
166*** if flag==0 generate socket and buffer
167*** <0 destroy socket and buffer
168*** >0 close and redo socket
169***
170*** sid : board*7 + port id
171 */
172
173 int j ;
174
175 if (rd->sockStat ==0 ) { //close socket if open
176 j=close(rd->socket) ;
177 if (j>0) {
178 snprintf(str,MXSTR,"Error closing socket %d | %m",sid);
179 factOut(kFatal,771, str ) ;
180 } else {
181 snprintf(str,MXSTR,"Succesfully closed socket %d",sid);
182 factOut(kInfo,771, str ) ;
183 }
184 }
185
186
187 if (flag < 0) {
188 free(rd->rBuf) ; //and never open again
189 rd->rBuf = NULL ;
190 rd->sockStat = 99 ;
191 return 0 ;
192 }
193
194
195 if (flag == 0) { //generate address and buffer ...
196 rd->Port = port ;
197 rd->SockAddr.sin_family = sockAddr->sin_family;
198 rd->SockAddr.sin_port = htons(port) ;
199 rd->SockAddr.sin_addr = sockAddr->sin_addr ;
200
201 rd->rBuf = malloc(sizeof(CNV_FACT) ) ;
202 if ( rd->rBuf == NULL ) {
203 snprintf(str,MXSTR,"Could not create local buffer %d",sid);
204 factOut(kFatal,774, str ) ;
205 rd->sockStat = 77 ;
206 return -3 ;
207 }
208 }
209
210
211 if ( (rd->socket = socket (PF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0)) <= 0) {
212 snprintf(str,MXSTR,"Could not generate socket %d | %m",sid);
213 factOut(kFatal,773, str ) ;
214 rd->sockStat = 88 ;
215 return -2 ;
216 }
217
218 snprintf(str,MXSTR,"Successfully generated socket %d ",sid);
219 factOut(kInfo,773, str ) ;
220 rd->sockStat = -1 ; //try to (re)open socket
221 return 0 ;
222
223} /*-----------------------------------------------------------------*/
224
225 /*-----------------------------------------------------------------*/
226
227
228
229
230int mBufInit() {
231// initialize mBuffer (mark all entries as unused\empty)
232
233 int i ;
234 uint32_t actime ;
235
236 actime = g_actTime + 50000000 ;
237
238 for (i=0; i<MAX_EVT*MAX_RUN; i++) {
239 mBuffer[i].evNum = mBuffer[i].runNum = -1;
240
241 evtCtrl.evtBuf[ i] = -1 ;
242 evtCtrl.evtStat[ i] = -1 ;
243 evtCtrl.pcTime[ i] = actime ; //initiate to far future
244
245 }
246
247
248 actRun.FADhead = malloc( NBOARDS* sizeof(PEVNT_HEADER) ) ;
249
250 evtCtrl.frstPtr = 0 ;
251 evtCtrl.lastPtr = 0 ;
252
253 return 0 ;
254
255} /*-----------------------------------------------------------------*/
256
257
258
259
260int mBufEvt(uint evID, uint runID, uint nRoi) {
261// generate a new Event into mBuffer:
262// make sure only complete Event are possible, so 'free' will always work
263// returns index into mBuffer[], or negative value in case of error
264
265
266 int i, k, evFree ;
267 int headmem=0 ;
268 size_t needmem = 0 ;
269
270 if (nRoi <=0 || nRoi > 1024) {
271 snprintf(str,MXSTR,"illegal nRoi: %d",nRoi) ;
272 factOut(kError, 1, str ) ;
273 return 99 ;
274 }
275
276 i = evID % MAX_EVT ;
277 evFree = -1 ;
278
279 for ( k=0; k<MAX_RUN; k++) {
280 if ( mBuffer[i].evNum == evID
281 && mBuffer[i].runNum== runID ) {
282 return i ;
283 }
284 if ( evFree < 0 && mBuffer[i].evNum < 0 ) evFree = i ;
285 i += MAX_EVT ;
286 }
287
288
289 //event does not yet exist; create
290 if (evFree < 0 ) { //no space available in ctrl
291 snprintf(str,MXSTR,"no control slot to keep event...") ;
292 factOut(kError,881, str ) ;
293 return -1 ;
294 }
295 i = evFree ; //found free entry; use it ...
296
297
298 needmem = sizeof(EVENT) + NPIX*nRoi*2 + NTMARK*nRoi*2; //
299
300 headmem = NBOARDS* sizeof(PEVNT_HEADER) ;
301
302 if ( gi_usedMem + needmem + headmem > g_maxMem) {
303 snprintf(str,MXSTR,"no memory left to keep event...") ;
304 factOut(kError,882, str ) ;
305 return -1 ;
306 }
307
308 mBuffer[i].FADhead = malloc( headmem ) ;
309 if (mBuffer[i].FADhead == NULL) {
310 return -12;
311 }
312
313 mBuffer[i].fEvent = malloc( needmem ) ;
314 if (mBuffer[i].fEvent == NULL) {
315 free(mBuffer[i].fEvent) ;
316 mBuffer[i].fEvent = NULL ;
317 return -22;
318 }
319
320 //flag all boards as unused
321 mBuffer[i].nBoard = 0 ;
322 for (k=0; k<NBOARDS; k++ ) {
323 mBuffer[i].board[k] = -1;
324 }
325
326 //flag all pixels as unused
327 for (k=0; k<NPIX; k++ ) {
328 mBuffer[i].fEvent->StartPix[k] = -1 ;
329 }
330
331 //flag all TMark as unused
332 for (k=0; k<NTMARK; k++ ) {
333 mBuffer[i].fEvent->StartTM[k] = -1 ;
334 }
335
336 mBuffer[i].pcTime = g_actTime ;
337 mBuffer[i].nRoi = nRoi ;
338 mBuffer[i].evNum = evID ;
339 mBuffer[i].runNum = runID ;
340 mBuffer[i].evtLen = needmem ;
341
342 gi_usedMem += needmem + headmem;
343
344 //register event in 'active list (reading)'
345
346 evtCtrl.evtBuf[ evtCtrl.lastPtr] = i ;
347 evtCtrl.evtStat[ evtCtrl.lastPtr] = 0 ;
348 evtCtrl.pcTime[ evtCtrl.lastPtr] = g_actTime ;
349 evtIdx[i] = evtCtrl.lastPtr ;
350snprintf(str,MXSTR,"%5d start new evt %8d %8d %2d",evID,i,evtCtrl.lastPtr,0);
351factOut(kDebug,-11, str ) ;
352 evtCtrl.lastPtr++ ;
353 if (evtCtrl.lastPtr == MAX_EVT*MAX_RUN ) evtCtrl.lastPtr = 0;
354
355
356
357
358 gi_EvtStart++ ;
359
360 //check if runId already registered in runCtrl
361 evFree = -1 ;
362 for (k=0; k<MAX_RUN; k++) {
363 if (runCtrl[k].runId == runID ) return i ;//run exists already
364 else if (evFree < 0 && runCtrl[k].runId == 0 ) evFree = k ;
365 }
366
367 if (evFree <0 ) {
368 snprintf(str,MXSTR,"not able to register the new run %d",runID);
369 factOut(kError,883, str ) ;
370 } else {
371 runCtrl[evFree].runId = runID ;
372 }
373
374 return i ;
375
376} /*-----------------------------------------------------------------*/
377
378
379int mBufFree(int i) {
380//delete entry [i] from mBuffer:
381//(and make sure multiple calls do no harm ....)
382
383 int headmem=0 ;
384 size_t freemem = 0 ;
385
386 if ( mBuffer[i].nRoi > 0) { //have an fEvent structure generated ...
387 freemem = mBuffer[i].evtLen ;
388 free(mBuffer[i].fEvent ) ;
389 mBuffer[i].fEvent = NULL ;
390
391 free(mBuffer[i].FADhead ) ;
392 mBuffer[i].FADhead = NULL ;
393
394 }
395 headmem = NBOARDS* sizeof(PEVNT_HEADER) ;
396 mBuffer[i].evNum = mBuffer[i].runNum = mBuffer[i].nRoi= -1;
397
398 gi_usedMem = gi_usedMem - freemem - headmem;
399
400
401 return 0 ;
402
403} /*-----------------------------------------------------------------*/
404
405
406 /*-----------------------------------------------------------------*/
407
408
409
410void initReadFAD() {
411return ;
412} /*-----------------------------------------------------------------*/
413
414
415
416void *readFAD( void *ptr ) {
417/* *** main loop reading FAD data and sorting them to complete events */
418 int head_len,frst_len,numok,numok2,dest,evID,i,j,k ;
419 int actBoards = 0;
420 int32_t jrd ;
421 int32_t myRun ;
422 int boardId, roi,drs,px,src,pixS,pixH,pixC,pixR,tmS ;
423 uint qtot = 0, qread = 0, qconn = 0 ;
424 int errcnt0 = 0 ;
425
426 int goodhed=0;
427
428 struct timespec xwait ;
429
430 int nokCnt[MAX_SOCK],loopCnt=0;
431 int sokCnt[MAX_SOCK];
432 int sockDef[NBOARDS];
433
434 struct timeval *tv, atv;
435 tv=&atv;
436 uint32_t tsec, tusec ;
437
438
439 snprintf(str,MXSTR,"start initializing");
440 factOut(kInfo,-1, str ) ;
441
442 int cpu = 7 ; //read thread
443 cpu_set_t mask;
444
445/* CPU_ZERO initializes all the bits in the mask to zero. */
446 CPU_ZERO( &mask );
447/* CPU_SET sets only the bit corresponding to cpu. */
448 cpu = 7 ;
449 CPU_SET( cpu, &mask );
450
451/* sched_setaffinity returns 0 in success */
452 if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
453 snprintf(str,MXSTR,"W ---> can not create affinity to %d",cpu);
454 factOut(kWarn,-1, str ) ;
455 }
456
457
458 //make sure all sockets are preallocated as 'not exist'
459 for (i=0; i<MAX_SOCK; i++) {
460 rd[i].socket = -1 ;
461 rd[i].sockStat = 99 ;
462 }
463 for (i=0; i<NBOARDS; i++) sockDef[i]= 0 ;
464
465
466 g_actTime = time(NULL) ;
467 for (k=0; k<MAX_SOCK; k++)
468 gi_SecRate[k]=gi_S10Rate[k]=gi_MinRate[k]=gi_ErrCnt[k] = 0 ;
469
470 for (k=0; k<NBOARDS; k++)
471 gi_NumConnect[k]=0;
472
473
474 gi_SecTime= gi_S10Time= gi_MinTime= g_actTime ;
475
476 mBufInit() ; //initialize buffers
477
478 snprintf(str,MXSTR,"end initializing");
479 factOut(kInfo,-1, str ) ;
480
481
482 for (k=0; k<MAX_SOCK; k++) sokCnt[k]=nokCnt[k]=0 ;
483
484 head_len = sizeof(PEVNT_HEADER) ;
485//frst_len = head_len + 36 * 12 ; //fad_header plus 36*pix_header
486 frst_len = head_len ; //fad_header only, so each event must be longer, even for roi=0
487
488
489 numok = numok2 = 0 ;
490
491 start.S=0xFB01;
492 stop.S= 0x04FE;
493
494 myRun = g_actTime ;
495
496 gi_runStat = g_runStat ;
497
498
499 while (g_runStat >=0) { //loop until global variable g_runStat claims stop
500
501 gi_runStat = g_runStat ;
502
503 int b,p,p0,s0,nch;
504 nch = 0 ;
505 for (b=0; b<NBOARDS; b++ ) {
506 k = b*7 ;
507 if ( g_port[b].sockDef != sockDef[b] ) { //something has changed ...
508 nch++ ;
509 gi_NumConnect[ b ] = 0 ; //must close all connections
510 if ( sockDef[b] == 0) s0= 0 ; //sockets to be defined and opened
511 else if (g_port[b].sockDef == 0) s0=-1 ; //sockets to be destroyed
512 else s0=+1 ; //sockets to be closed and reopened
513
514 if (s0 == 0) p0=ntohs(g_port[b].sockAddr.sin_port);
515 else p0=0 ;
516
517 for (p=p0+1; p<p0+8; p++) {
518 GenSock(s0, k, p, &g_port[b].sockAddr, &rd[k]) ; //generate address and socket
519 k++ ;
520 }
521 sockDef[b] = g_port[b].sockDef ;
522 }
523 }
524
525 if (nch > 0 ) {
526 actBoards = 0 ;
527 for (b=0; b<NBOARDS; b++ ) {
528 if ( sockDef[b] > 0 ) actBoards++ ;
529 }
530 }
531
532
533 g_actTime = time(NULL) ;
534 nokCnt[numok]++;
535
536 loopCnt++ ;
537
538 numok = 0 ; //count number of succesfull actions
539
540 for (i=0; i<MAX_SOCK; i++) { //check all sockets if something to read
541 b = i / 7 ;
542
543 gettimeofday( tv, NULL);
544 tsec = atv.tv_sec ;
545 tusec= atv.tv_usec ;
546
547 if (rd[i].sockStat <0 ) { //try to connect if not yet done
548 rd[i].sockStat=connect(rd[i].socket,
549 (struct sockaddr*) &rd[i].SockAddr, sizeof(rd[i].SockAddr)) ;
550 if (rd[i].sockStat ==0 ) { //successfull ==>
551 if (sockDef[b] > 0) {
552 rd[i].bufTyp = 0 ; // expect a header
553 rd[i].bufLen = frst_len ; // max size to read at begining
554 } else {
555 rd[i].bufTyp = -1 ; // data to be skipped
556 rd[i].bufLen = sizeof(CNV_FACT) ; //huge for skipping
557 }
558 rd[i].bufPos = 0 ; // no byte read so far
559 gi_NumConnect[ b ]++ ;
560 numok++ ; //make sure next round will execute
561 snprintf(str,MXSTR,"+++connect %d %d",b,gi_NumConnect[ b ]);
562 factOut(kInfo,-1, str ) ;
563 }
564 }
565
566 if (rd[i].sockStat ==0) { //we have a connection ==> try to read
567 numok++ ;
568 sokCnt[i]++;
569 jrd=recv(rd[i].socket,&rd[i].rBuf->B[ rd[i].bufPos], rd[i].bufLen, MSG_DONTWAIT);
570
571 if (jrd >0 ) {
572 qread+=jrd ;
573 debugStream(i,&rd[i].rBuf->B[ rd[i].bufPos],jrd) ;
574 }
575
576 if (jrd == 0) { //connection has closed ...
577 snprintf(str,MXSTR,"Socket %d closed by FAD",i);
578 factOut(kInfo,441, str ) ;
579 GenSock(1, i, 0,NULL, &rd[i]) ;
580 gi_ErrCnt[i]++ ;
581 gi_NumConnect[ b ]-- ;
582
583 } else if ( jrd<0 ) { //did not read anything
584 if (errno != EAGAIN && errno != EWOULDBLOCK ) {
585 snprintf(str,MXSTR,"Error Reading from %d | %m",i);
586 factOut(kError,442, str ) ;
587 gi_ErrCnt[i]++ ;
588 } else numok-- ; //else nothing waiting to be read
589
590 } else if ( rd[i].bufTyp <0 ) { // we are skipping this board ...
591// just do nothing
592
593 } else if ( rd[i].bufTyp >0 ) { // we are reading data ...
594 if ( jrd < rd[i].bufLen ) { //not yet all read
595 rd[i].bufPos += jrd ; //==> prepare for continuation
596 rd[i].bufLen -= jrd ;
597 debugRead(i,jrd,rd[i].evtID, 0,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 0=reading data
598 } else { //full dataset read
599 rd[i].bufLen = rd[i].bufPos + j ;
600 rd[i].bufPos = rd[i].fadLen ;
601 if ( rd[i].rBuf->B[ rd[i].bufPos-1] != stop.B[0]
602 && rd[i].rBuf->B[ rd[i].bufPos ] != stop.B[1]) {
603 gi_ErrCnt[i]++ ;
604 snprintf(str,MXSTR,"wrong end of buffer found %d",rd[i].bufPos);
605 factOut(kError,301, str ) ;
606 goto EndBuf ;
607
608 }
609 debugRead(i,jrd,rd[i].evtID, 1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; 1=finished event
610
611 //we have a complete buffer, copy to WORK area
612 gi_SecRate[i]++ ;
613
614 roi = ntohs(rd[i].rBuf->S[ head_len/2 + 2 ]) ;
615 //get index into mBuffer for this event (create if needed)
616 evID = mBufEvt( rd[i].evtID, rd[i].runID, roi ) ;
617
618 if (evID < 0) {
619 snprintf(str,MXSTR,"no space left ...%d",evID) ;
620 factOut(kError,201, str ) ;
621 goto EndBuf ; //--> skip event (and hope it will improve)
622 }
623
624 //we have a valid entry in mBuffer[]; fill it
625
626 boardId = b ;
627 int fadBoard = ntohs(rd[i].rBuf->S[12] ) ;
628 int fadCrate = fadBoard/256 ;
629 if (boardId != (fadCrate*10 + fadBoard%256) ) {
630 snprintf(str,MXSTR,"wrong Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ;
631 if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ; //print only few times
632// } else {
633// snprintf(str,MXSTR,"correct Board ID %d %d %d",fadCrate,fadBoard%256,boardId) ;
634// if (errcnt0++ < 99 ) factOut(kWarn,301, str ) ; //print only few times
635 }
636 if ( mBuffer[evID].board[ boardId ] != -1) {
637 snprintf(str,MXSTR,"double board %d for event %d",boardId,evID) ;
638 factOut(kWarn,501, str ) ;
639 goto EndBuf ; //--> skip Board
640 }
641
642 int iDx = evtIdx[evID] ; //index into evtCtrl
643
644 memcpy( &mBuffer[evID].FADhead[boardId].start_package_flag,
645 &rd[i].rBuf->S[0], head_len) ;
646 roi = mBuffer[evID].nRoi ;
647
648 pixS = boardId*36 -1 ; //
649 tmS = boardId*4 -1 ; //
650 src = head_len/2 ;
651 for ( drs=0; drs<4; drs++ ) {
652 for ( px=0; px<9; px++ ) {
653 pixH= ntohs(rd[i].rBuf->S[src++]) ;
654 pixC= ntohs(rd[i].rBuf->S[src++]) ;
655 pixR= ntohs(rd[i].rBuf->S[src++]) ;
656
657 src++ ;
658 pixS++ ; //pixS = pixH2S[pixH] ;
659 if ( ( px < PX8 && pixR == roi )
660 || ( px ==PX8 && pixR == 2*roi )
661 || ( px ==PX8 && pixR == roi && roi > 512 ) ) {
662 // correct roi
663 mBuffer[evID].fEvent->StartPix[pixS] =pixC;
664 dest= pixS * roi ;
665 memcpy(
666 &mBuffer[evID].fEvent->Adc_Data[dest],
667 &rd[i].rBuf->S[src], roi * 2) ;
668 src+= roi ;
669 if ( px==PX8 ) {
670 tmS++; // tmS = tmH2S[pixH]
671 dest= tmS * roi + NPIX* roi ;
672 if ( roi <=512 ) {
673 mBuffer[evID].fEvent->StartTM[tmS] =(pixC+roi)%1024 ;
674 memcpy(
675 &mBuffer[evID].fEvent->Adc_Data[dest],
676 &rd[i].rBuf->S[src], roi * 2) ;
677 src+=roi ;
678 } else {
679 mBuffer[evID].fEvent->StartTM[tmS] = -1 ;
680 }
681 }
682 } else {
683 snprintf(str,MXSTR,"wrong roi %d %d %d %d",px,pixR,roi,src-2);
684 factOut(kError,202, str ) ;
685 goto EndBuf ;
686 }
687 }
688 }// now we have stored a new board contents into Event structure
689 mBuffer[evID].board[ boardId ] = boardId ;
690 evtCtrl.evtStat[ iDx ]++ ;
691 evtCtrl.pcTime[ iDx ] = g_actTime ;
692
693 if (++mBuffer[evID].nBoard >= actBoards ) {
694 snprintf(str,MXSTR,"%5d complete event %8d %8d %2d",mBuffer[evID].evNum,evtCtrl.evtBuf[iDx],iDx,evtCtrl.evtStat[ iDx ]);
695 factOut(kDebug,-1, str ) ;
696 //complete event read ---> flag for next processing
697 evtCtrl.evtStat[ iDx ] = 99;
698 gi_EvtRead++ ;
699 gi_EvtTot++ ;
700 }
701
702EndBuf:
703 rd[i].bufTyp = 0 ; //ready to read next header
704 rd[i].bufLen = frst_len ;
705 rd[i].bufPos = 0 ;
706 }
707
708 } else { //we are reading event header
709 rd[i].bufPos += jrd ;
710 rd[i].bufLen -= jrd ;
711 if ( rd[i].bufPos > MIN_LEN ){ //sufficient data to take action
712 //check if startflag correct; else shift block ....
713 for (k=0; k<rd[i].bufPos -1 ; k++) {
714 if (rd[i].rBuf->B[k ] == start.B[1]
715 && rd[i].rBuf->B[k+1] == start.B[0] ) break ;
716 }
717
718
719 for (k=0; k<rd[i].bufPos -1 ; k++) {
720 if (rd[i].rBuf->B[k ] == start.B[1]
721 && rd[i].rBuf->B[k+1] == start.B[0] ) break ;
722 }
723
724 if (k >= rd[i].bufPos-1 ) { //no start of header found
725 snprintf(str,MXSTR,"no start of header on port%d", i ) ;
726 factOut(kWarn,666, str ) ;
727
728 rd[i].bufPos = 0 ;
729 rd[i].bufLen = head_len ;
730 } else if ( k>0 ) {
731 rd[i].bufPos -= k ;
732 rd[i].bufLen += k ;
733 memcpy(&rd[i].rBuf->B[0], &rd[i].rBuf->B[k], rd[i].bufPos ) ;
734 }
735 if ( rd[i].bufPos > MIN_LEN ) {
736 goodhed++;
737 rd[i].fadLen = ntohs(rd[i].rBuf->S[1])*2 ;
738 rd[i].fadVers= ntohs(rd[i].rBuf->S[2]) ;
739 rd[i].evtID = ntohl(rd[i].rBuf->I[4]) ; //(FADevt)
740 rd[i].runID = ntohl(rd[i].rBuf->I[11]) ;
741if (rd[i].runID ==0 ) rd[i].runID = myRun ;
742 rd[i].bufTyp = 1 ; //ready to read full record
743 rd[i].bufLen = rd[i].fadLen - rd[i].bufPos ;
744 if (rd[i].bufLen <=0 ) rd[i].bufLen = 100000 ; //?
745 debugRead(i,jrd,rd[i].evtID,-1,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid;-1=start event
746 } else {
747 debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
748 }
749 } else {
750 debugRead(i,jrd,0,-2,tsec,tusec) ; // i=socket; jrd=#bytes; ievt=eventid; -2=start event, unknown id yet
751 }
752
753 } //end interpreting last read
754 } //end of successful read anything
755 } //finished trying to read all sockets
756
757 int qwait=0, qdel=0, qskip=0 ;
758 g_actTime = time(NULL) ;
759 if ( g_actTime > gi_SecTime ) {
760 gi_SecTime = g_actTime ;
761// PrintRate() ;
762
763
764 //loop over all active events and flag those older than read-timeout
765 //delete those that are written to disk ....
766
767 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
768 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
769
770 int k1=evtCtrl.frstPtr;
771 for ( k=k1; k<(k1+kd); k++ ) {
772 int k0 = k % (MAX_EVT*MAX_RUN) ;
773//would be better to use bitmaps for evtStat (allow '&' instead of multi-if)
774 if (evtCtrl.evtStat[k0] > 0
775 && evtCtrl.evtStat[k0] < 90 ) {
776
777 qwait++;
778
779 if( evtCtrl.pcTime[k0] < g_actTime-10 ) {
780 int id =evtCtrl.evtBuf[k0] ;
781 snprintf(str,MXSTR,"%5d skip short evt %8d %8d %2d",mBuffer[id].evNum,evtCtrl.evtBuf[k0],k0 ,evtCtrl.evtStat[k0]);
782 factOut(kWarn,601, str ) ;
783 evtCtrl.evtStat[k0] = 91 ; //timeout for incomplete events
784 gi_EvtBad++ ;
785 gi_EvtTot++ ;
786 qskip++;
787 }
788
789
790 } else if (evtCtrl.evtStat[k0] >= 900 ) {
791
792 int id =evtCtrl.evtBuf[k0] ;
793 snprintf(str,MXSTR,"%5d free event buffer (written) %3d", mBuffer[id].evNum, mBuffer[id].nBoard ) ;
794 factOut(kDebug,-1, str ) ;
795 mBufFree(id) ; //event written--> free memory
796 evtCtrl.evtStat[k0] = -1;
797 qdel++;
798 qtot++;
799 }
800
801 if ( k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] <0 ) {
802 evtCtrl.frstPtr = (evtCtrl.frstPtr+1) % (MAX_EVT*MAX_RUN) ;
803 }
804 }
805
806qconn=0 ;
807int ib ;
808for (ib=0; ib<NBOARDS; ib++) qconn+=gi_NumConnect[ib] ;
809
810snprintf(str,MXSTR,"bfr%5d skp%4d free%4d (tot%7d) mem%9lu rd%10d %3d",qwait,qskip,qdel,qtot,gi_usedMem,qread,qconn);
811factStat(kInfo,-1, str ) ;
812qread=0 ;
813 }
814
815
816
817
818 if (numok > 0 ) numok2=0;
819 else if (numok2++ > 3) {
820 if (g_runStat == 1) {
821 xwait.tv_sec = 1;
822 xwait.tv_nsec= 0 ; // hibernate for 1 sec
823 } else {
824 xwait.tv_sec = 0;
825 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
826 }
827 nanosleep( &xwait , NULL ) ;
828 }
829
830 } //and do next loop over all sockets ...
831
832 //must quit eventbuilding
833 snprintf(str,MXSTR,"stop reading ...");
834 factOut(kInfo,-1, str ) ;
835
836 //flag all events as 'read finished'
837 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
838 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
839
840 int k1=evtCtrl.frstPtr;
841
842 for ( k=k1; k<(k1+kd); k++ ) {
843 int k0 = k % (MAX_EVT*MAX_RUN) ;
844 if (evtCtrl.evtStat[k0] > 0
845 && evtCtrl.evtStat[k0] < 90 ) {
846 evtCtrl.evtStat[k0] = 91 ;
847 gi_EvtBad++ ;
848 gi_EvtTot++ ;
849 }
850 }
851
852 //must close all open sockets ...
853 snprintf(str,MXSTR,"close all sockets ...");
854 factOut(kInfo,-1, str ) ;
855 for (i=0; i<MAX_SOCK; i++) {
856 GenSock(-1, i, 0, NULL, &rd[i]) ; //close and destroy socket
857 if (gi_NumConnect[ i/7 ]>0)
858 gi_NumConnect[ i/7 ]-- ;
859 }
860
861 xwait.tv_sec = 0;
862 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
863 nanosleep( &xwait , NULL ) ;
864 gi_runStat = -11 ; //inform all that no update to happen any more
865
866
867 int minclear = 900 ; //usually wait until writing finished (stat 900)
868 if (g_runStat <-1 ) minclear = 0 ; //in case of abort clear all
869
870
871 //and clear all buffers (might have to wait until all others are done)
872 snprintf(str,MXSTR,"clear all buffers ...");
873 factOut(kInfo,-1, str ) ;
874 int numclear=1 ;
875 while (numclear > 0 ) {
876 numclear = 0 ;
877 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
878 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
879
880 int k1=evtCtrl.frstPtr;
881 for ( k=k1; k<(k1+kd); k++ ) {
882 int k0 = k % (MAX_EVT*MAX_RUN) ;
883 if (evtCtrl.evtStat[k0] > minclear ) {
884 int id =evtCtrl.evtBuf[k0] ;
885 mBufFree(id) ; //event written--> free memory
886 evtCtrl.evtStat[k0] = -1;
887 } else if (evtCtrl.evtStat[k0] > 0) numclear++ ; //writing is still ongoing...
888
889 if ( k0 == evtCtrl.frstPtr && evtCtrl.evtStat[k0] <0 )
890 evtCtrl.frstPtr = (evtCtrl.frstPtr+1) % (MAX_EVT*MAX_RUN) ;
891 }
892
893 xwait.tv_sec = 0;
894 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
895 nanosleep( &xwait , NULL ) ;
896 }
897
898 snprintf(str,MXSTR,"Exit read Process ...");
899 factOut(kInfo,-1, str ) ;
900 gi_runStat = -99 ;
901 return 0;
902
903} /*-----------------------------------------------------------------*/
904
905
906void *procEvt( void *ptr ) {
907/* *** main loop processing file, including SW-trigger */
908 int numProc, numWait ;
909 int k ;
910 struct timespec xwait ;
911 char str[MXSTR] ;
912
913 cpu_set_t mask;
914 int cpu = 5 ; //process thread (will be several in final version)
915
916 snprintf(str,MXSTR,"Starting process-thread");
917 factOut(kInfo,-1, str ) ;
918
919/* CPU_ZERO initializes all the bits in the mask to zero. */
920 CPU_ZERO( &mask );
921/* CPU_SET sets only the bit corresponding to cpu. */
922 CPU_SET( cpu, &mask );
923/* sched_setaffinity returns 0 in success */
924 if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
925 snprintf(str,MXSTR,"P ---> can not create affinity to %d",cpu);
926 factOut(kWarn,-1, str ) ;
927 }
928
929
930 while (g_runStat > -2) { //in case of 'exit' we still must process pending events
931
932 numWait = numProc = 0 ;
933 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
934 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
935
936 int k1=evtCtrl.frstPtr;
937 for ( k=k1; k<(k1+kd); k++ ) {
938 int k0 = k % (MAX_EVT*MAX_RUN) ;
939//would be better to use bitmaps for evtStat (allow '&' instead of multi-if)
940 if (evtCtrl.evtStat[k0] > 90 && evtCtrl.evtStat[k0] <500) {
941 int id = evtCtrl.evtBuf[k0] ;
942 int ievt = mBuffer[id].evNum ;
943 int roi = mBuffer[id].nRoi ;
944// uint32_t irun = mBuffer[id].runNum ;
945//snprintf(str,MXSTR,"P processing %d %d %d %d",ievt,k,id,evtCtrl.evtStat[k0]) ;
946//factOut(kDebug,-1, str ) ;
947
948//make sure unused pixels/tmarks are cleared to zero
949 int ip,it,dest,ib;
950 for (ip=0; ip<NPIX; ip++) {
951 if (mBuffer[id].fEvent->StartPix[ip] == -1 ) {
952 dest= ip*roi ;
953 bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ;
954 }
955 }
956 for (it=0; it<NTMARK; it++) {
957 if (mBuffer[id].fEvent->StartTM[it] == -1 ) {
958 dest= it*roi + NPIX*roi ;
959 bzero( &mBuffer[id].fEvent->Adc_Data[dest], roi*2) ;
960 }
961 }
962
963
964//and set correct event header ; also check for consistency in event (not yet)
965 mBuffer[id].fEvent->Roi = roi ;
966 mBuffer[id].fEvent->EventNum = ievt ;
967 mBuffer[id].fEvent->TriggerType = 0 ; // TBD
968 mBuffer[id].fEvent->SoftTrig = 0 ;
969 for (ib=0; ib<NBOARDS; ib++) {
970 if (mBuffer[id].board[ib] == -1 ) { //board is not read
971 mBuffer[id].FADhead[ib].start_package_flag = 0 ;
972 mBuffer[id].fEvent->BoardTime[ib] = 0 ;
973 } else {
974 mBuffer[id].fEvent->BoardTime[ib] =
975 ntohl(mBuffer[id].FADhead[ib].time) ;
976 }
977 }
978
979 int i=eventCheck(mBuffer[id].FADhead,mBuffer[id].fEvent) ;
980
981 if (i<0) evtCtrl.evtStat[k0] = 999 ; //flag event to be skipped
982
983 numProc++ ;
984 evtCtrl.evtStat[k0] = 520 ;
985 gp_EvtTot++ ;
986 } else if ( evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] < 90 ) {
987 numWait++ ;
988 }
989 }
990
991 if ( gi_runStat < -10 && numWait == 0) { //nothing left to do
992 snprintf(str,MXSTR,"Exit Processing Process ...");
993 factOut(kInfo,-1, str ) ;
994 gp_runStat = -22 ; //==> we should exit
995 return 0 ;
996 }
997
998 if (numProc == 0) {
999 //seems we have nothing to do, so sleep a little
1000 xwait.tv_sec = 0;
1001 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1002 nanosleep( &xwait , NULL ) ;
1003 }
1004 gp_runStat = gi_runStat ;
1005
1006 }
1007
1008 //we are asked to abort asap ==> must flag all remaining events
1009 // when gi_runStat claims that all events are in the buffer...
1010
1011 snprintf(str,MXSTR,"Abort Processing Process ...");
1012 factOut(kInfo,-1, str ) ;
1013 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1014 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1015
1016 int k1=evtCtrl.frstPtr;
1017 for ( k=k1; k<(k1+kd); k++ ) {
1018 int k0 = k % (MAX_EVT*MAX_RUN) ;
1019 if (evtCtrl.evtStat[k0] >=0 && evtCtrl.evtStat[k0] <500) {
1020 evtCtrl.evtStat[k0] = 555 ; //flag event as 'processed'
1021 }
1022 }
1023
1024 gp_runStat = -99 ;
1025
1026 return 0;
1027
1028} /*-----------------------------------------------------------------*/
1029
1030int CloseRunFile(uint32_t runId, uint32_t closeTime) {
1031/* close run runId (all all runs if runId=0) */
1032/* return: 0=close scheduled / >0 already closed / <0 does not exist */
1033 int i, j ;
1034
1035 if (runId == 0 ) {
1036 for ( j=0; j<MAX_RUN; j++) {
1037 if ( runCtrl[j].fileId == 0 ) { //run is open
1038 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1039 if (i<0) {
1040 snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
1041 factOut(kError,506, str ) ;
1042 runCtrl[j].fileId = 888 ;
1043 } else {
1044 snprintf(str,MXSTR,"closing run %d ok AAA",runCtrl[j].runId);
1045 factOut(kInfo,507, str ) ;
1046 runCtrl[j].fileId = 7777 ;
1047 }
1048 runCtrl[j].closeTime = closeTime ;
1049 }
1050 }
1051 return 0 ;
1052 }
1053
1054
1055 for ( j=0; j<MAX_RUN; j++) {
1056 if ( runCtrl[j].runId == runId ) {
1057 if ( runCtrl[j].fileId == 0 ) { //run is open
1058 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1059 if (i<0) {
1060 snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
1061 factOut(kError,506, str ) ;
1062 runCtrl[j].fileId = 888 ;
1063 } else {
1064 snprintf(str,MXSTR,"closing run %d ok AAA",runCtrl[j].runId);
1065 factOut(kInfo,507, str ) ;
1066 runCtrl[j].fileId = 7777 ;
1067 }
1068 runCtrl[j].closeTime = closeTime ;
1069 return 0;
1070 } else if ( runCtrl[j].fileId <0 ) { //run not yet opened
1071 runCtrl[j].closeTime = closeTime ;
1072 return +1;
1073 } else { // run already closed
1074 return +2;
1075 }
1076 }
1077 } //we only reach here if the run was never created
1078 return -1;
1079
1080} /*-----------------------------------------------------------------*/
1081
1082
1083void *writeEvt( void *ptr ) {
1084/* *** main loop writing event (including opening and closing run-files */
1085
1086 int numWrite, numWait ;
1087 int k,j ;
1088 struct timespec xwait ;
1089 char str[MXSTR] ;
1090
1091 cpu_set_t mask;
1092 int cpu = 3 ; //write thread
1093
1094 snprintf(str,MXSTR,"Starting write-thread");
1095 factOut(kInfo,-1, str ) ;
1096
1097/* CPU_ZERO initializes all the bits in the mask to zero. */
1098 CPU_ZERO( &mask );
1099/* CPU_SET sets only the bit corresponding to cpu. */
1100 CPU_SET( cpu, &mask );
1101/* sched_setaffinity returns 0 in success */
1102 if ( sched_setaffinity( 0, sizeof(mask), &mask ) == -1 ) {
1103 snprintf(str,MXSTR,"W ---> can not create affinity to %d",cpu);
1104 }
1105
1106 int lastRun = 0 ; //usually run from last event still valid
1107
1108 while (g_runStat >-2) {
1109
1110 numWait = numWrite = 0 ;
1111 int kd = evtCtrl.lastPtr - evtCtrl.frstPtr ;
1112 if ( kd < 0 ) kd+= (MAX_EVT*MAX_RUN) ;
1113
1114 int k1=evtCtrl.frstPtr;
1115 for ( k=k1; k<(k1+kd); k++ ) {
1116 int k0 = k % (MAX_EVT*MAX_RUN) ;
1117//would be better to use bitmaps for evtStat (allow '&' instead of multi-if)
1118 if (evtCtrl.evtStat[k0] > 500 && evtCtrl.evtStat[k0] < 900) {
1119 int id = evtCtrl.evtBuf[k0] ;
1120 uint32_t irun = mBuffer[id].runNum ;
1121 int ievt = mBuffer[id].evNum ;
1122
1123 if (runCtrl[lastRun].runId == irun ) {
1124 j = lastRun ;
1125 } else {
1126 //check which fileID to use (or open if needed)
1127 for ( j=0; j<MAX_RUN; j++) {
1128 if ( runCtrl[j].runId == irun ) break ;
1129 }
1130 if ( j >= MAX_RUN ) {
1131 snprintf(str,MXSTR,"W error: can not find run %d for event %d in %d", irun,ievt,id);
1132 factOut(kFatal,901, str ) ;
1133for ( j=0; j<MAX_RUN; j++) printf("j %d run.j %d run %d\n",j,runCtrl[j].runId,irun );
1134exit(111);
1135 }
1136 lastRun = j ;
1137 }
1138
1139 if (runCtrl[j].fileId < 0 ) {
1140 actRun.Version = 1 ;
1141 actRun.RunType = -1 ;
1142 actRun.NBoard = NBOARDS ;
1143 actRun.NPix = NPIX ;
1144 actRun.NTm = NTMARK ;
1145 actRun.Nroi = mBuffer[id].nRoi ;
1146// actRun.FADhead = mBuffer[id].FADhead ; //to be corrected
1147 runCtrl[j].nextEvt= 0;
1148 runCtrl[j].lastTime=g_actTime ;
1149 runCtrl[j].fileHd = runOpen(irun, &actRun, sizeof(actRun) ) ;
1150 if (runCtrl[j].fileHd == NULL ) {
1151 snprintf(str,MXSTR,"W could not open a file for run %d",irun);
1152 factOut(kError,502, str ) ;
1153 runCtrl[j].fileId = 99 ;
1154 } else {
1155 snprintf(str,MXSTR,"W opened new run_file %d",irun) ;
1156 factOut(kInfo,-1, str ) ;
1157 runCtrl[j].fileId = 0 ;
1158 }
1159
1160 }
1161
1162 if (runCtrl[j].fileId > 0 ) {
1163 snprintf(str,MXSTR,"W no open file for this run %d",irun) ;
1164 factOut(kDebug,123,str) ;
1165 evtCtrl.evtStat[k0] = 902 ;
1166 } else {
1167 int i=runWrite(runCtrl[j].fileHd, mBuffer[id].fEvent, sizeof(mBuffer[id]) );
1168 if (i<0) {
1169 snprintf(str,MXSTR,"W error writing event for run %d",irun) ;
1170 factOut(kError,503, str ) ;
1171 evtCtrl.evtStat[k0] = 901 ;
1172 //close run
1173 i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1174 if (i<0) {
1175 snprintf(str,MXSTR,"W error closing run %d",irun) ;
1176 factOut(kError,503, str ) ;
1177 } else {
1178 snprintf(str,MXSTR,"W closed run %d because of write error",irun) ;
1179 factOut(kInfo,503, str ) ;
1180 }
1181 runCtrl[j].fileId = 9999 ;
1182 } else {
1183 runCtrl[j].lastTime = g_actTime;
1184 evtCtrl.evtStat[k0] = 901 ;
1185 snprintf(str,MXSTR,"%5d successfully wrote for run %d id %5d",ievt,irun,k0);
1186 factOut(kDebug,504, str ) ;
1187 }
1188 }
1189 } else if (evtCtrl.evtStat[k0] > 0 ) numWait++ ;
1190 }
1191
1192 //check if we should close a run ...
1193 for ( j=0; j<MAX_RUN; j++) {
1194 if ( runCtrl[j].fileId==0
1195 && ( runCtrl[j].closeTime < g_actTime
1196 ||runCtrl[j].lastTime < g_actTime-120) ) {
1197 int i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1198 if (i<0) {
1199 snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
1200 factOut(kError,506, str ) ;
1201 runCtrl[j].fileId = 888 ;
1202 } else {
1203 snprintf(str,MXSTR,"closing run %d ok BBB",runCtrl[j].runId);
1204 factOut(kInfo,507, str ) ;
1205 runCtrl[j].fileId = 7777 ;
1206 }
1207 }
1208 }
1209
1210 if (numWrite == 0) {
1211 //seems we have nothing to do, so sleep a little
1212 xwait.tv_sec = 0;
1213 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1214 nanosleep( &xwait , NULL ) ;
1215 }
1216
1217 if ( gi_runStat < -10 && numWait == 0) { //nothing left to do
1218 snprintf(str,MXSTR,"Finish Write Process ...");
1219 factOut(kInfo,-1, str ) ;
1220 gw_runStat = -22 ; //==> we should exit
1221 goto closerun ;
1222 }
1223 gw_runStat = gi_runStat ;
1224
1225 }
1226
1227 //must close all open files ....
1228 snprintf(str,MXSTR,"Abort Writing Process ...");
1229 factOut(kInfo,-1, str ) ;
1230closerun:
1231 snprintf(str,MXSTR,"Close all open files ...");
1232 factOut(kInfo,-1, str ) ;
1233 for ( j=0; j<MAX_RUN; j++)
1234 if ( runCtrl[j].fileId ==0 ) {
1235 int i=runClose(runCtrl[j].fileHd, &runTail[j], sizeof(runTail[j]) );
1236 if (i<0) {
1237 snprintf(str,MXSTR,"error closing run %d %d",runCtrl[j].runId,i) ;
1238 factOut(kError,506, str ) ;
1239 runCtrl[j].fileId = 888 ;
1240 } else {
1241 snprintf(str,MXSTR,"closing run %d ok AAA",runCtrl[j].runId);
1242 factOut(kInfo,507, str ) ;
1243 runCtrl[j].fileId = 7777 ;
1244 }
1245 }
1246
1247 gw_runStat = -99;
1248 snprintf(str,MXSTR,"Exit Writing Process ...");
1249 factOut(kInfo,-1, str ) ;
1250 return 0;
1251
1252
1253
1254
1255} /*-----------------------------------------------------------------*/
1256
1257
1258
1259
1260void StartEvtBuild() {
1261
1262 int i,j,imax,status,th_ret[50] ;
1263 pthread_t thread[50] ;
1264 struct timespec xwait ;
1265 uint32_t actime ;
1266
1267 gi_runStat = gp_runStat = gw_runStat = 0 ;
1268
1269 snprintf(str,MXSTR,"Starting EventBuilder");
1270 factOut(kInfo,-1, str ) ;
1271
1272
1273 evtCtrl.frstPtr = 0 ;
1274 evtCtrl.lastPtr = 0 ;
1275
1276 actime = g_actTime + 50000000 ;
1277/* initialize run control logics */
1278 for (i=0; i<MAX_RUN; i++) {
1279 runCtrl[i].runId = 0 ;
1280 runCtrl[i].lastTime = 0 ;
1281 runCtrl[i].closeTime = time(NULL) + 3600*24*7;
1282
1283 runCtrl[i].nextEvt = 0 ;
1284 runCtrl[i].fileId = -2 ;
1285
1286 runTail[i].nEventsOk =
1287 runTail[i].nEventsRej =
1288 runTail[i].nEventsBad =
1289 runTail[i].PCtime0 =
1290 runTail[i].PCtimeX = 0 ;
1291 }
1292
1293//start all threads (more to come) when we are allowed to ....
1294 while (g_runStat == 0 ) {
1295 xwait.tv_sec = 0;
1296 xwait.tv_nsec= 2000000 ; // sleep for ~2 msec
1297 nanosleep( &xwait , NULL ) ;
1298 }
1299
1300 i=0 ;
1301 th_ret[i] = pthread_create( &thread[i], NULL, readFAD, NULL );
1302 i++;
1303 th_ret[i] = pthread_create( &thread[i], NULL, procEvt, NULL );
1304 i++;
1305 th_ret[i] = pthread_create( &thread[i], NULL, writeEvt, NULL );
1306 i++;
1307 imax=i ;
1308
1309
1310
1311
1312
1313/*
1314 xwait.tv_sec = 20;;
1315 xwait.tv_nsec= 0 ; // sleep for ~20sec
1316 nanosleep( &xwait , NULL ) ;
1317
1318
1319 printf("close all runs in 2 seconds\n");
1320
1321
1322 CloseRunFile( 0, time(NULL)+2) ;
1323
1324 xwait.tv_sec = 5;;
1325 xwait.tv_nsec= 0 ; // sleep for ~20sec
1326 nanosleep( &xwait , NULL ) ;
1327
1328 printf("setting g_runstat to -1\n");
1329
1330 g_runStat = -1 ;
1331*/
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343//wait for all threads to finish
1344 for (i=0; i<imax; i++) {
1345 j = pthread_join ( thread[i], (void **)&status) ;
1346 }
1347
1348} /*-----------------------------------------------------------------*/
1349
1350
1351
1352
1353
1354 /*-----------------------------------------------------------------*/
1355 /*-----------------------------------------------------------------*/
1356 /*-----------------------------------------------------------------*/
1357 /*-----------------------------------------------------------------*/
1358 /*-----------------------------------------------------------------*/
1359 /*-----------------------------------------------------------------*/
1360 /*-----------------------------------------------------------------*/
1361 /*-----------------------------------------------------------------*/
1362 /*-----------------------------------------------------------------*/
1363 /*-----------------------------------------------------------------*/
1364
1365
1366
1367/*
1368FileHandle_t runOpen(uint32_t irun, RUN_HEAD *runhd, size_t len )
1369{ return 1; } ;
1370
1371int runWrite(FileHandle_t fileHd , EVENT *event, size_t len )
1372{ return 1; } ;
1373
1374int runClose(FileHandle_t fileHd , RUN_TAIL *runth, size_t len )
1375{ return 1; } ;
1376
1377
1378
1379int eventCheck( PEVNT_HEADER *fadhd, EVENT *event)
1380{
1381 int i=0;
1382
1383 printf("------------%d\n",ntohl(fadhd[7].fad_evt_counter) );
1384 for (i=0; i<NBOARDS; i++) {
1385 printf("b=%2d,=%5d\n",i,fadhd[i].board_id);
1386 }
1387 return 0;
1388}
1389
1390
1391void factStat(int severity, int err, char* message ) {
1392 printf("%3d %3d : %s\n",severity,err,message) ;
1393}
1394
1395
1396void debugRead(int isock, int ibyte, int32_t event, int state, uint32_t tsec, uint32_t tusec ) {
1397// printf("%3d %5d %9d %3d %12d\n",isock, ibyte, event, state, tusec) ;
1398}
1399
1400
1401
1402void debugStream(int isock, void *buf, int len) {
1403}
1404
1405
1406void factOut(int severity, int err, char* message ) {
1407static FILE * fd ;
1408static int file=0 ;
1409
1410 if (file==0) {
1411 printf("open file\n");
1412 fd=fopen("x.out","w+") ;
1413 file=999;
1414 }
1415
1416 fprintf(fd,"%3d %3d | %s \n",severity,err,message) ;
1417
1418 if (severity != kDebug)
1419 printf("%3d %3d | %s\n",severity,err,message) ;
1420}
1421
1422
1423
1424int main() {
1425 int i,b,c,p ;
1426 char ipStr[100] ;
1427 struct in_addr IPaddr ;
1428
1429 g_maxMem = 1024*1024 ; //MBytes
1430 g_maxMem = g_maxMem * 1024 *10 ; //10GBytes
1431
1432
1433 g_runStat = 40 ;
1434
1435 i=0 ;
1436
1437// version for standard crates
1438//for (c=0; c<4,c++) {
1439// for (b=0; b<10; b++) {
1440// sprintf(ipStr,"10.0.%d.%d",128+c,128+b)
1441//
1442// inet_pton(PF_INET, ipStr, &IPaddr) ;
1443//
1444// g_port[i].sockAddr.sin_family = PF_INET;
1445// g_port[i].sockAddr.sin_port = htons(5000) ;
1446// g_port[i].sockAddr.sin_addr = IPaddr ;
1447// g_port[i].sockDef = 1 ;
1448// i++ ;
1449// }
1450//}
1451//
1452//version for PC-test
1453 for (c=0; c<4; c++) {
1454 for (b=0; b<10; b++) {
1455 sprintf(ipStr,"10.0.%d.11",128+c) ;
1456 if (c==0) sprintf(ipStr,"10.0.100.11") ;
1457
1458 inet_pton(PF_INET, ipStr, &IPaddr) ;
1459 p = 31919+100*c+10*b;
1460
1461
1462 g_port[i].sockAddr.sin_family = PF_INET;
1463 g_port[i].sockAddr.sin_port = htons(p) ;
1464 g_port[i].sockAddr.sin_addr = IPaddr ;
1465 g_port[i].sockDef = 1 ;
1466
1467 i++ ;
1468 }
1469 }
1470
1471
1472//g_port[17].sockDef =-1 ;
1473//g_actBoards-- ;
1474
1475 StartEvtBuild() ;
1476
1477 return 0;
1478
1479}
1480*/
Note: See TracBrowser for help on using the repository browser.