EB exit condition if no data coming for one of sources. Sergey.

diff --git a/hadaq/netmem.c b/hadaq/netmem.c
index e1949d667dc426f833a1078fd74b5cd69c8bc370..a2b0bda8d3cc793cb929558054e988df0df206f7 100644 (file)
--- a/hadaq/netmem.c
+++ b/hadaq/netmem.c
@@ -1,4 +1,4 @@
-static char *rcsId = "$Header: /misc/hadesprojects/daq/cvsroot/eventbuilder/hadaq/netmem.c,v 6.42 2010-07-23 12:38:23 hadaq Exp $";
+static char *rcsId = "$Header: /misc/hadesprojects/daq/cvsroot/eventbuilder/hadaq/netmem.c,v 6.43 2010-07-30 09:19:31 hadaq Exp $";
 
 #define _POSIX_C_SOURCE 199309L
 #define SYSLOG_NAMES
@@ -42,6 +42,9 @@ typedef struct TheArgsS {
 
        unsigned long portNr[MAXINPATH];
        int portNrCnt;
+
+       /* Maximu delay of incoming data in seconds. */
+       unsigned long dataDelay;
 } TheArgs;
 
 typedef struct TheStatsS {
@@ -68,6 +71,7 @@ static void usage(const char *progName)
        syslog(LOG_ERR, "       [-q queueSize]               Size of the queue (shared memory segments) in MBytes.");
        syslog(LOG_ERR, "       [-Q queueNr:queueSize]       Set different queue sizes in MB for different queue numbers,");
        syslog(LOG_ERR, "                                    example: -Q 2:4 -Q 4:8 -Q 5:12");
+       syslog(LOG_ERR, "       [-d seconds]                 Maximum delay of incoming data in seconds.");
 }
 
 static int makeQueues(TheArgs *my)
@@ -143,17 +147,13 @@ static void argsDefault(TheArgs *my)
 
        my->portNrCnt = 0;
 
+       my->dataDelay = 5;                      /* seconds */
+
        strcpy(my->shmname, "");
 
        sprintf(my->evtstat, "%s_evtstat.txt", getenv("DAQ_SETUP"));
 }
 
-static inPath_short2wide(TheArgs *theArgs, char *optarg)
-{
-
-
-}
-
 static int argsFromCL(TheArgs *my, int argc, char *argv[])
 {
        extern char *optarg;
@@ -164,7 +164,7 @@ static int argsFromCL(TheArgs *my, int argc, char *argv[])
        inPathCntFromParam = my->inPathCnt;
        my->inPathCnt = 0;
 
-       while ((i = getopt(argc, argv, "ai:m:q:Q:p:v:bS:")) != -1) {
+       while ((i = getopt(argc, argv, "ai:m:q:Q:p:v:bS:d:")) != -1) {
                switch (i) {
                case 'm':
                        my->nrOfMsgs = atoi(optarg);
@@ -190,6 +190,9 @@ static int argsFromCL(TheArgs *my, int argc, char *argv[])
                case 'S':
                        strcpy(my->shmname, optarg);
                        break;
+               case 'd':
+                       my->dataDelay = atoi(optarg);
+                       break;
                default:
                        return -1;
                        break;
@@ -265,6 +268,61 @@ static char *unit(unsigned long v)
        return retVal;
 }
 
+static int checkDataSources(TheArgs *theArgs, TheStats *theStats)
+{
+       int ready2check = 0;
+       int i;
+       static unsigned long dataSources[MAXINPATH];
+       static int once = 1;
+
+       static time_t t_0 = 0;
+       time_t t, dT;
+
+       t = time(NULL);
+       dT = t - t_0;
+
+       if (dT >= 1) {
+
+               /* Init to zero */
+               if (once) {
+                       for (i = 0; i < theArgs->nrOfMsgs; i++)
+                               dataSources[i] = 0;
+                       once = 0;
+               }
+
+               for (i = 0; i < theArgs->nrOfMsgs; i++) {
+                       if ((*theStats->recvBytesRate[i]) > 0)
+                               dataSources[i]++;
+
+                       if (dataSources[i] > theArgs->dataDelay)
+                               ready2check = 1;
+               }
+
+               int go2exit = 0;
+
+               if (ready2check) {
+                       for (i = 0; i < theArgs->nrOfMsgs; i++) {
+                               if (dataSources[i] == 0) {
+                                       fprintf(stderr, "<E> daq_netmem: source %d, port %d: no data received.\n", i, theArgs->portNr[i]);
+                                       go2exit = 1;
+                               }
+                       }
+
+                       if (go2exit)
+                               return -1;
+
+                       /* Init to zero */
+                       ready2check = 0;
+                       for (i = 0; i < theArgs->nrOfMsgs; i++)
+                               dataSources[i] = 0;
+               }
+
+               t_0 = t;
+       }
+
+       return 0;
+}
+
 static void printTime()
 {
        struct timeval tv;
@@ -319,9 +377,14 @@ static void add2Stat(TheArgs *theArgs, TheStats *my, float interval, ShmTrans **
                        printf("%3d", fillLevel);
 
                        /* Add more statistic for recv bytes per second */
-                       (*my->recvBytesRate[i]) = *nettr[i]->bytesReceived - lastBytesRecv[i];
+                       if (*nettr[i]->bytesReceived - lastBytesRecv[i] > 0) {
+                               (*my->recvBytesRate[i]) = *nettr[i]->bytesReceived - lastBytesRecv[i];
+                       } else {
+                               (*my->recvBytesRate[i]) = *nettr[i]->bytesReceived;
+                       }
                        lastBytesRecv[i] = *nettr[i]->bytesReceived;
                }
+
                t_0 = t;
 
                printf("\n");
@@ -546,6 +609,10 @@ int main(int argc, char *argv[])
 
                        add2Stat(theArgs, theStats, 1, shmTrans, netTrans);
                        statsDump(theArgs, netTrans, 1);
+                       if (-1 == checkDataSources(theArgs, theStats)) {
+                               fprintf(stderr, "Exit.\n");
+                               goto bailOut1;
+                       }
 
                        for (i = 0; i < theArgs->nrOfMsgs; i++) {
                                if (hadTu[i] == NULL) {
@@ -578,6 +645,8 @@ int main(int argc, char *argv[])
        storeInfoStop(argv[0], time(NULL), worker, theArgs);
        statsDump(theArgs, netTrans, 0);
 
+  bailOut1:
+
        for (i = 0; i < theArgs->nrOfMsgs; i++) {
                ShmTrans_remove(shmTrans[i]);
                NetTrans_remove(netTrans[i]);