[Nscldaq-cvscommits] SF.net SVN: nscldaq:[3573] trunk/nextgen/daq/eventbuilder

[<jr...@us...>]

Revision: 3573
          http://sourceforge.net/p/nscldaq/code/3573
Author:   jrtomps
Date:     2014-06-18 15:22:52 +0000 (Wed, 18 Jun 2014)
Log Message:
-----------
Issue #2064 Adds 2 second startup timeout for 10.2 evbuilder.

Modified Paths:
--------------
    trunk/nextgen/daq/eventbuilder/CFragmentHandler.cpp
    trunk/nextgen/daq/eventbuilder/CFragmentHandler.h

Modified: trunk/nextgen/daq/eventbuilder/CFragmentHandler.cpp
===================================================================
--- trunk/nextgen/daq/eventbuilder/CFragmentHandler.cpp	2014-06-18 13:46:38 UTC (rev 3572)
+++ trunk/nextgen/daq/eventbuilder/CFragmentHandler.cpp	2014-06-18 15:22:52 UTC (rev 3573)
@@ -36,7 +36,8 @@
 
 static const time_t DefaultBuildWindow(20); // default seconds to accumulate data before ordering.
 static const uint32_t IdlePollInterval(2);  // Seconds between idle polls.
-
+static const time_t DefaultStartupTimeout(2); // default seconds to accumulate data before ordering.
+static time_t timeOfFirstSubmission(UINT64_MAX); // 
 /*---------------------------------------------------------------------
  * Debugging
  */
@@ -81,6 +82,7 @@
 
 {
     m_nBuildWindow = DefaultBuildWindow;
+    m_nStartupTimeout = DefaultStartupTimeout;
     m_pInstance = this;
     resetTimestamps();
 
@@ -146,8 +148,10 @@
       m_nMostRecentlyEmptied = m_nNow;
     }
 
+    if (first) {
+      timeOfFirstSubmission = m_nNow;
+    }
 
-
     while (nSize) {
       EVB::pFragmentHeader pHeader = &(pFragments->s_header);
       size_t fragmentSize = totalFragmentSize(pHeader);
@@ -174,9 +178,12 @@
       nSize -= fragmentSize;
     }
     findOldest();		// Probably not needed but pretty quick.
-    flushQueues();		// flush events with received time stamps older than m_nNow - m_nBuildWindow
+    if (m_nNow-timeOfFirstSubmission > m_nStartupTimeout) {
+      // Don't flush until we have allowed time for all data sources to 
+      // establish themselves
+      flushQueues();		// flush events with received time stamps older than m_nNow - m_nBuildWindow
+    }
 
-
 }
 /**
  * setBuildWindow
@@ -205,6 +212,33 @@
 }
 
 /**
+ * setStartupTimeout
+ * 
+ * Set the startup timeout duration.  The startup timeout duration determines the amount
+ * of time after the first fragment is received when calls to flushQueues is not allowed. This is
+ * to ensure that all queues have been created before flushing.
+ * 
+ * @param duration - the number of seconds to use as the startup timeout
+ */
+void
+CFragmentHandler::setStartupTimeout(time_t duration)
+{
+    m_nStartupTimeout = duration;
+}
+/**
+ * getStartupTimeout
+ *
+ * Return the value of the current startup timeout duration
+ *
+ * @return time_t - duration of the startup timeout 
+ */
+time_t
+CFragmentHandler::getStartupTimeout() const
+{
+  return m_nStartupTimeout;
+}
+
+/**
  * addObserver
  *
  * The fragment handler does not really  know how to deal with built
@@ -588,7 +622,11 @@
  loop: 			// avoid recursion with a good old fashioned goto.
   m_nNow = time(NULL);
 
-
+#ifdef DEBUG
+  if (completely) {
+    std::cerr << "flush COMPLETELY!\n";
+  }
+#endif
   // Ensure there's at least one fragment available:
 
 
@@ -627,6 +665,8 @@
       std::cerr << "Empty queues\n";
     } else {
       std::cerr << "Not empty\n";
+      std::cerr << "Time_now - oldestReceived_time = " << m_nNow-m_nOldestReceived;
+      std::cerr << "  buildWindow = " << m_nBuildWindow << "\n"; 
     }
 #endif
   }
@@ -644,12 +684,18 @@
   // tail call to continue building:
   
   if(m_fBarrierPending) {
+#ifdef DEBUG  
+  std::cerr << "Barrier pending\n";
+#endif
     checkBarrier(completely);	// Complete forces barriers out.
   }
   // If we still have non-empty queues and were asked to completely flush
   // tail call:
   
   if (!queuesEmpty() && completely) {
+#ifdef DEBUG  
+  std::cerr << "Goto statement\n";
+#endif
     goto loop;
   }
 }
@@ -1311,7 +1357,13 @@
   if (!pHandler->m_nFragmentsLastPeriod) {
     pHandler->m_nNow = time(NULL);	// Update tod.
     pHandler->findOldest();		// Update oldest fragment time.
-    pHandler->flushQueues();
+    if (pHandler->m_nNow - timeOfFirstSubmission > pHandler->m_nStartupTimeout) { 
+      // Only flush after we have given time for all sources
+      // establish their queues
+      // Also... timeOfFirstSubmission is guaranteed to exist 
+      // because nFragmentLastPeriod is not 0
+      pHandler->flushQueues();
+    }
   } else {
     pHandler->m_nFragmentsLastPeriod = 0;
   }

Modified: trunk/nextgen/daq/eventbuilder/CFragmentHandler.h
===================================================================
--- trunk/nextgen/daq/eventbuilder/CFragmentHandler.h	2014-06-18 13:46:38 UTC (rev 3572)
+++ trunk/nextgen/daq/eventbuilder/CFragmentHandler.h	2014-06-18 15:22:52 UTC (rev 3573)
@@ -214,8 +214,8 @@
   time_t                       m_nNow;
   time_t                       m_nOldestReceived;
   time_t                       m_nMostRecentlyEmptied;
+  time_t                       m_nStartupTimeout;   //!< N seconds to wait before flushing (dflt=2)
 
-
   uint32_t                     m_nFragmentsLastPeriod; //!< # fragments in last flush check interval.
 
   std::list<Observer*>         m_OutputObservers;
@@ -259,6 +259,9 @@
   void setBuildWindow(time_t windowWidth);
   time_t getBuildWindow() const;
 
+  void setStartupTimeout(time_t duration);
+  time_t getStartupTimeout() const;
+
   // Observer management:
 
 public:

This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.