make timeout a double, since we'd busy loop otherwise on sub-msec intervals, plus...

diff --git a/pdns/delaypipe.cc b/pdns/delaypipe.cc
index d0798a532f67a34fd6dd66b52f934971847843e7..046a90ef355a2331cd7e73fac04a88ac2b4a9205 100644 (file)
--- a/pdns/delaypipe.cc
+++ b/pdns/delaypipe.cc
@@ -50,16 +50,18 @@ bool ObjectPipe<T>::read(T* t)
 }
 
 template<class T>
-int ObjectPipe<T>::readTimeout(T* t, int msec)
+int ObjectPipe<T>::readTimeout(T* t, double msec)
 {
   T* ptr;
-  int ret = waitForData(d_fds[0], 0, 1000*msec);
-  if(ret <0)
-    unixDie("waiting for data in object pipe");
-  if(ret == 0) 
-    return -1;
+  if(msec != 0) {
+    int ret = waitForData(d_fds[0], 0, 1000*msec);
+    if(ret < 0)
+      unixDie("waiting for data in object pipe");
+    if(ret == 0) 
+      return -1;
+  }
 
-  ret = ::read(d_fds[0], &ptr, sizeof(ptr));
+  int ret = ::read(d_fds[0], &ptr, sizeof(ptr));
 
   if(ret < 0)
     unixDie("read");
@@ -97,31 +99,46 @@ DelayPipe<T>::~DelayPipe()
   d_thread.join();
 }
 
+
+
 template<class T>
 void DelayPipe<T>::worker()
 {
   Combo c;
   for(;;) {
-    int ret = d_pipe.readTimeout(&c, 10); // XXXX NEEDS TO BE DYNAMIC
-    if(ret > 0) {  // we got an object
-      d_work.insert(make_pair(c.when, c.what));
-    }
-    else if(ret==0) { // timeout
-
-      break;
+    double delay=-1;  // infinite
+    struct timespec now;
+    if(!d_work.empty()) {
+      clock_gettime(CLOCK_MONOTONIC, &now);
+      delay=1000*tsdelta(d_work.begin()->first, now);
+      if(delay < 0) {
+       delay=0;   // don't wait
+      }
     }
-    else {
-      //      cout<<"Got a timeout"<<endl;
+    if(delay != 0 ) {
+      int ret = d_pipe.readTimeout(&c, delay); 
+      if(ret > 0) {  // we got an object
+       d_work.insert(make_pair(c.when, c.what));
+      }
+      else if(ret==0) { // EOF
+       break;
+      }
+      else {
+       ;
+      }
+      clock_gettime(CLOCK_MONOTONIC, &now);
     }
-    struct timespec now;
-    clock_gettime(CLOCK_MONOTONIC, &now);
+
     tscomp cmp;
+
     for(auto iter = d_work.begin() ; iter != d_work.end(); ) { // do the needful
       if(cmp(iter->first, now)) {
        iter->second();
        d_work.erase(iter++);
       }
-      else break;
+      else {
+       break;
+      }
     }
   }
 }

diff --git a/pdns/delaypipe.hh b/pdns/delaypipe.hh
index cfc72e74b9c8197aaf4f0014a407de9c7e28a306..80cca590687b1ea77e0338147cb166079331ce4e 100644 (file)
--- a/pdns/delaypipe.hh
+++ b/pdns/delaypipe.hh
@@ -23,7 +23,7 @@ public:
   ~ObjectPipe();
   void write(T& t);
   bool read(T* t); // returns false on EOF
-  int readTimeout(T* t, int msec); // -1 is timeout, 0 is no data, 1 is data
+  int readTimeout(T* t, double msec); // -1 is timeout, 0 is no data, 1 is data
   void close(); 
 private:
   int d_fds[2];
@@ -46,6 +46,11 @@ private:
     struct timespec when;
   };
 
+  double tsdelta(const struct timespec& a, const struct timespec& b) // read as a-b
+  {
+    return 1.0*(a.tv_sec-b.tv_sec)+1.0*(a.tv_nsec-b.tv_nsec)/1000000000.0;
+  }
+
   ObjectPipe<Combo> d_pipe;
   struct tscomp {
     bool operator()(const struct timespec& a, const struct timespec& b) const

diff --git a/pdns/test-delaypipe_hh.cc b/pdns/test-delaypipe_hh.cc
index 94287a935e1fcdfb2e282ca92d64f38ecf33ab45..145adc4cb41f30ae6920c3a2663b15ef5df9902f 100644 (file)
--- a/pdns/test-delaypipe_hh.cc
+++ b/pdns/test-delaypipe_hh.cc
@@ -46,7 +46,7 @@ BOOST_AUTO_TEST_CASE(test_delay_pipe) {
 
   for(; n < 10; ++n) {
     Work w{n};
-    dp.submit(w, 1000);
+    dp.submit(w, 1200);
   }
   sleep(1);
   BOOST_CHECK_EQUAL(done, 5);