From: hadaq <hadaq>
Date: Fri, 25 Nov 2011 09:00:33 +0000 (+0000)
Subject: JAM: improved fast cleanup of eventbuilder queues in case of too many trigger mismatc... 
X-Git-Url: https://jspc29.x-matter.uni-frankfurt.de/git/?a=commitdiff_plain;h=215aee61b9a1959952626a2f025bd6f346ce89d9;p=daqdata.git

JAM: improved fast cleanup of eventbuilder queues in case of too many trigger mismatch. Now complete double buffers are discarded. This works (always?), and is fast.
---

diff --git a/hadaq/evtbuild.c b/hadaq/evtbuild.c
index 0e6f1b6..d6feb25 100644
--- a/hadaq/evtbuild.c
+++ b/hadaq/evtbuild.c
@@ -1,4 +1,4 @@
-static char *rcsId = "$Header: /misc/hadesprojects/daq/cvsroot/eventbuilder/hadaq/evtbuild.c,v 6.155 2011-11-24 18:09:02 hadaq Exp $";
+static char *rcsId = "$Header: /misc/hadesprojects/daq/cvsroot/eventbuilder/hadaq/evtbuild.c,v 6.156 2011-11-25 09:00:33 hadaq Exp $";
 
 #define _POSIX_C_SOURCE 199309L
 #define SYSLOG_NAMES
@@ -54,9 +54,11 @@ static RFILE *fRemote = NULL;
 
 #define RUNID_INFO_DT  600		/* time in s until not available runid info is logged  */
 #define LOST_SUBEVT_LIMIT 10000	/* number of popped subevents per trigger sequence until warning */
-#define LOST_SUBEVT_DROP  4		/* drop all queue contents after this number of lost sub warnings per trigger sequence */
+#define LOST_SUBEVT_DROP  2		/* drop all queue contents after this number of lost sub warnings per trigger sequence */
 #define QUEUE_FLUSH_EVENTS 10000
 #define QUEUE_FLUSH_REPEAT 5
+/* define this to test fast flushing of queues by double buffer switch*/
+#define QUEUE_FLUSH_FAST 1
 
 #define NEVTIDS 64UL			/* must be 2^n */
 #define NEVTIDS_IN_FILE 0UL		/* must be 2^n */
@@ -1303,7 +1305,7 @@ int main(int argc, char *argv[])
 					 * Scale down warning message frequency:*/
 					if ((popCnt % LOST_SUBEVT_LIMIT) == 0) {
 						sprintf(msglog,
-								"<W> Subevent stream %d has popped  nr:%d queue entries to sync with master stream of trig no %d after found stream trig no %d",
+								"<W> Subevent stream %d has popped  nr:%d queue entries to sync with master stream of trig no 0x%x after found stream trig no 0x%x",
 								i, popCnt, currTrigNr, trigNr);
 						storeLogInfo(theArgs, msglog);
 						popTermCnt++;
@@ -1316,6 +1318,35 @@ int main(int argc, char *argv[])
 						popCnt = 0;
 						popTermCnt = 0;	/*reset globally */
 
+#ifdef QUEUE_FLUSH_FAST
+						/* faster: discard  whole double buffer without scanning hadtuque headers */
+						storeLogInfo(theArgs, "<W> too many lost subevents. Discarding whole double buffers...");
+
+						for (q = 0; q < theArgs->nrOfMsgs; q++) {
+							void *storage;
+							if (hadTuQueue[q] != NULL) {
+								desHadTuQueue(hadTuQueue[q]);
+								free(hadTuQueue[q]);
+								hadTuQueue[q] = NULL;
+							}
+							ShmTrans_requestData(shmTrans[q]);
+							/* this function does not "request data", but switches to other double buffer
+							 * todo: rename/cleanup some of the API functions? JAM*/
+							ShmTrans_requestData(shmTrans[q]);	/* second time to switch back, discarding all previous data */
+
+							storage = ShmTrans_recv(shmTrans[q]);
+							hadTuQueue[q] = malloc(HadTuQueue_sizeOf());
+							conHadTuQueue_voidP(hadTuQueue[q], storage);
+
+							sprintf(msglog, "<I>  Cleared double buffers of stream %d.", q);
+							storeLogInfo(theArgs, msglog);
+						}
+						/* here we lose track of number of discarded events.
+						 * todo: find out difference between trigger numbers at next cycle?*/
+
+
+#else
+
 						storeLogInfo(theArgs, "<W> too many lost subevents. Dropping historic buffers...");
 						/* avoid here that other channels run full during clearing */
 						for (jj = 0; jj < QUEUE_FLUSH_REPEAT; jj++) {
@@ -1371,14 +1402,17 @@ int main(int argc, char *argv[])
 							(*theStats->evtsDiscarded) += discEvCnt;	/* account max of discarded subevents as discarded events */
 							discEventSum += discEvCnt;
 						}		/* for jj */
-						i = 0;	/* start again with next trigger id */
-						step = 0;
+
 
 						sprintf(msglog, "<I>  leaving queue cleanup section after discarding %d events", discEventSum);
 						storeLogInfo(theArgs, msglog);
+
+#endif
 						discEvCnt = 0;
 						discSubCnt = 0;
-
+						evtIsBroken = 1;	/* must leave loop, since discarded stream 0 subevent was already copied to event */
+						/*i = 0;     start again with next trigger id
+						   step = 0; */
 					}
 				}
 			} else {