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

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