Merge pull request #277 from austin-beer/fix_sync_queue_time_jump_issues

Fix sync_timed_queue time jump issues

Author: viboes

include/boost/thread/concurrent_queues/sync_timed_queue.hpp

@@ -16,6 +16,8 @@
 #include <boost/chrono/system_clocks.hpp>
 #include <boost/chrono/chrono_io.hpp>
 
+#include <algorithm> // std::min
+
 #include <boost/config/abi_prefix.hpp>
 
 namespace boost
@@ -59,6 +61,45 @@ namespace detail
     }
   }; //end struct
 
+  template <class Duration>
+  chrono::time_point<chrono::steady_clock,Duration>
+  limit_timepoint(chrono::time_point<chrono::steady_clock,Duration> const& tp)
+  {
+    // Clock == chrono::steady_clock
+    return tp;
+  }
+
+  template <class Clock, class Duration>
+  chrono::time_point<Clock,Duration>
+  limit_timepoint(chrono::time_point<Clock,Duration> const& tp)
+  {
+    // Clock != chrono::steady_clock
+    // The system time may jump while wait_until() is waiting. To compensate for this and time out near
+    // the correct time, we limit how long wait_until() can wait before going around the loop again.
+    const chrono::time_point<Clock,Duration> tpmax(chrono::time_point_cast<Duration>(Clock::now() + chrono::milliseconds(BOOST_THREAD_POLL_INTERVAL_MILLISECONDS)));
+    return (std::min)(tp, tpmax);
+  }
+
+  template <class Duration>
+  chrono::steady_clock::time_point
+  convert_to_steady_clock_timepoint(chrono::time_point<chrono::steady_clock,Duration> const& tp)
+  {
+    // Clock == chrono::steady_clock
+    return chrono::time_point_cast<chrono::steady_clock::duration>(tp);
+  }
+
+  template <class Clock, class Duration>
+  chrono::steady_clock::time_point
+  convert_to_steady_clock_timepoint(chrono::time_point<Clock,Duration> const& tp)
+  {
+    // Clock != chrono::steady_clock
+    // The system time may jump while wait_until() is waiting. To compensate for this and time out near
+    // the correct time, we limit how long wait_until() can wait before going around the loop again.
+    const chrono::steady_clock::duration dura(chrono::duration_cast<chrono::steady_clock::duration>(tp - Clock::now()));
+    const chrono::steady_clock::duration duramax(chrono::milliseconds(BOOST_THREAD_POLL_INTERVAL_MILLISECONDS));
+    return chrono::steady_clock::now() + (std::min)(dura, duramax);
+  }
+
 } //end detail namespace
 
   template <class T, class Clock = chrono::steady_clock, class TimePoint=typename Clock::time_point>
@@ -88,8 +129,8 @@ namespace detail
     T pull();
     void pull(T& elem);
 
-    template <class WClock, class Duration>
-    queue_op_status pull_until(chrono::time_point<WClock,Duration> const& tp, T& elem);
+    template <class Duration>
+    queue_op_status pull_until(chrono::time_point<clock,Duration> const& tp, T& elem);
     template <class Rep, class Period>
     queue_op_status pull_for(chrono::duration<Rep,Period> const& dura, T& elem);
 
@@ -122,8 +163,9 @@ namespace detail
     inline bool not_empty_and_time_reached(lock_guard<mutex>& lk) const;
 
     bool wait_to_pull(unique_lock<mutex>&);
-    template <class WClock, class Duration>
-    queue_op_status wait_to_pull_until(unique_lock<mutex>&, chrono::time_point<WClock, Duration> const& tp);
+    queue_op_status wait_to_pull_until(unique_lock<mutex>&, TimePoint const& tp);
+    template <class Rep, class Period>
+    queue_op_status wait_to_pull_for(unique_lock<mutex>& lk, chrono::duration<Rep,Period> const& dura);
 
     T pull(unique_lock<mutex>&);
     T pull(lock_guard<mutex>&);
@@ -228,14 +270,13 @@ namespace detail
       if (not_empty_and_time_reached(lk)) return false; // success
       if (super::closed(lk)) return true; // closed
 
-      const time_point tp(super::data_.top().time);
-      super::cond_.wait_until(lk, tp);
+      const time_point tpmin(detail::limit_timepoint(super::data_.top().time));
+      super::cond_.wait_until(lk, tpmin);
     }
   }
 
   template <class T, class Clock, class TimePoint>
-  template <class WClock, class Duration>
-  queue_op_status sync_timed_queue<T, Clock, TimePoint>::wait_to_pull_until(unique_lock<mutex>& lk, chrono::time_point<WClock, Duration> const& tp)
+  queue_op_status sync_timed_queue<T, Clock, TimePoint>::wait_to_pull_until(unique_lock<mutex>& lk, TimePoint const& tp)
   {
     for (;;)
     {
@@ -249,7 +290,29 @@ namespace detail
       if (super::closed(lk)) return queue_op_status::closed;
       if (clock::now() >= tp) return super::empty(lk) ? queue_op_status::timeout : queue_op_status::not_ready;
 
-      const time_point tpmin(tp < super::data_.top().time ? tp : super::data_.top().time);
+      const time_point tpmin((std::min)(tp, detail::limit_timepoint(super::data_.top().time)));
+      super::cond_.wait_until(lk, tpmin);
+    }
+  }
+
+  template <class T, class Clock, class TimePoint>
+  template <class Rep, class Period>
+  queue_op_status sync_timed_queue<T, Clock, TimePoint>::wait_to_pull_for(unique_lock<mutex>& lk, chrono::duration<Rep,Period> const& dura)
+  {
+    const chrono::steady_clock::time_point tp(chrono::steady_clock::now() + chrono::duration_cast<chrono::steady_clock::duration>(dura));
+    for (;;)
+    {
+      if (not_empty_and_time_reached(lk)) return queue_op_status::success;
+      if (super::closed(lk)) return queue_op_status::closed;
+      if (chrono::steady_clock::now() >= tp) return super::empty(lk) ? queue_op_status::timeout : queue_op_status::not_ready;
+
+      super::wait_until_not_empty_or_closed_until(lk, tp);
+
+      if (not_empty_and_time_reached(lk)) return queue_op_status::success;
+      if (super::closed(lk)) return queue_op_status::closed;
+      if (chrono::steady_clock::now() >= tp) return super::empty(lk) ? queue_op_status::timeout : queue_op_status::not_ready;
+
+      const chrono::steady_clock::time_point tpmin((std::min)(tp, detail::convert_to_steady_clock_timepoint(super::data_.top().time)));
       super::cond_.wait_until(lk, tpmin);
     }
   }
@@ -315,12 +378,12 @@ namespace detail
 
   //////////////////////
   template <class T, class Clock, class TimePoint>
-  template <class WClock, class Duration>
+  template <class Duration>
   queue_op_status
-  sync_timed_queue<T, Clock, TimePoint>::pull_until(chrono::time_point<WClock, Duration> const& tp, T& elem)
+  sync_timed_queue<T, Clock, TimePoint>::pull_until(chrono::time_point<clock,Duration> const& tp, T& elem)
   {
     unique_lock<mutex> lk(super::mtx_);
-    const queue_op_status rc = wait_to_pull_until(lk, tp);
+    const queue_op_status rc = wait_to_pull_until(lk, chrono::time_point_cast<typename time_point::duration>(tp));
     if (rc == queue_op_status::success) pull(lk, elem);
     return rc;
   }
@@ -331,7 +394,10 @@ namespace detail
   queue_op_status
   sync_timed_queue<T, Clock, TimePoint>::pull_for(chrono::duration<Rep,Period> const& dura, T& elem)
   {
-    return pull_until(chrono::steady_clock::now() + dura, elem);
+    unique_lock<mutex> lk(super::mtx_);
+    const queue_op_status rc = wait_to_pull_for(lk, dura);
+    if (rc == queue_op_status::success) pull(lk, elem);
+    return rc;
   }
 
   ///////////////////////////

test/sync/mutual_exclusion/sync_pq/tq_multi_thread_pass.cpp

@@ -24,31 +24,58 @@ using namespace boost::chrono;
 
 typedef boost::concurrent::sync_timed_queue<int> sync_tq;
 
-const int count = 5;
+const int cnt = 5;
 
-void call_push(sync_tq* q)
+void call_push(sync_tq* q, const steady_clock::time_point start)
 {
     // push elements onto the queue every 500 milliseconds but with a decreasing delay each time
-    for (int i = 0; i < count; ++i)
+    for (int i = 0; i < cnt; ++i)
     {
-        q->push(i, sync_tq::clock::now() + seconds(count - i));
-        boost::this_thread::sleep_for(milliseconds(500));
+        boost::this_thread::sleep_until(start + milliseconds(i * 500));
+        const steady_clock::time_point expected = start + milliseconds(i * 500) + seconds(cnt - i);
+        q->push(i, expected);
     }
 }
 
-void call_pull(sync_tq* q)
+void call_pull(sync_tq* q, const steady_clock::time_point start)
 {
     // pull elements off of the queue (earliest element first)
-    steady_clock::time_point start = steady_clock::now();
-    for (int i = count - 1; i >= 0; --i)
+    for (int i = cnt - 1; i >= 0; --i)
     {
         int j;
         q->pull(j);
         BOOST_TEST_EQ(i, j);
-        milliseconds elapsed = duration_cast<milliseconds>(steady_clock::now() - start);
-        milliseconds expected = milliseconds(i * 500) + seconds(count - i);
-        BOOST_TEST_GE(elapsed, expected - milliseconds(BOOST_THREAD_TEST_TIME_MS));
-        BOOST_TEST_LE(elapsed, expected + milliseconds(BOOST_THREAD_TEST_TIME_MS));
+        const steady_clock::time_point expected = start + milliseconds(i * 500) + seconds(cnt - i);
+        BOOST_TEST_GE(steady_clock::now(), expected - milliseconds(BOOST_THREAD_TEST_TIME_MS));
+        BOOST_TEST_LE(steady_clock::now(), expected + milliseconds(BOOST_THREAD_TEST_TIME_MS));
+    }
+}
+
+void call_pull_until(sync_tq* q, const steady_clock::time_point start)
+{
+    // pull elements off of the queue (earliest element first)
+    for (int i = cnt - 1; i >= 0; --i)
+    {
+        int j;
+        q->pull_until(steady_clock::now() + hours(1), j);
+        BOOST_TEST_EQ(i, j);
+        const steady_clock::time_point expected = start + milliseconds(i * 500) + seconds(cnt - i);
+        BOOST_TEST_GE(steady_clock::now(), expected - milliseconds(BOOST_THREAD_TEST_TIME_MS));
+        BOOST_TEST_LE(steady_clock::now(), expected + milliseconds(BOOST_THREAD_TEST_TIME_MS));
+    }
+}
+
+void call_pull_for(sync_tq* q, const steady_clock::time_point start)
+{
+    // pull elements off of the queue (earliest element first)
+    for (int i = cnt - 1; i >= 0; --i)
+    {
+        int j;
+        q->pull_for(hours(1), j);
+        BOOST_TEST_EQ(i, j);
+        const steady_clock::time_point expected = start + milliseconds(i * 500) + seconds(cnt - i);
+        BOOST_TEST_GE(steady_clock::now(), expected - milliseconds(BOOST_THREAD_TEST_TIME_MS));
+        BOOST_TEST_LE(steady_clock::now(), expected + milliseconds(BOOST_THREAD_TEST_TIME_MS));
     }
 }
 
@@ -57,14 +84,41 @@ void test_push_while_pull()
     sync_tq tq;
     BOOST_TEST(tq.empty());
     boost::thread_group tg;
-    tg.create_thread(boost::bind(call_push, &tq));
-    tg.create_thread(boost::bind(call_pull, &tq));
+    const steady_clock::time_point start = steady_clock::now();
+    tg.create_thread(boost::bind(call_push, &tq, start));
+    tg.create_thread(boost::bind(call_pull, &tq, start));
+    tg.join_all();
+    BOOST_TEST(tq.empty());
+}
+
+void test_push_while_pull_until()
+{
+    sync_tq tq;
+    BOOST_TEST(tq.empty());
+    boost::thread_group tg;
+    const steady_clock::time_point start = steady_clock::now();
+    tg.create_thread(boost::bind(call_push, &tq, start));
+    tg.create_thread(boost::bind(call_pull_until, &tq, start));
+    tg.join_all();
+    BOOST_TEST(tq.empty());
+}
+
+void test_push_while_pull_for()
+{
+    sync_tq tq;
+    BOOST_TEST(tq.empty());
+    boost::thread_group tg;
+    const steady_clock::time_point start = steady_clock::now();
+    tg.create_thread(boost::bind(call_push, &tq, start));
+    tg.create_thread(boost::bind(call_pull_for, &tq, start));
     tg.join_all();
     BOOST_TEST(tq.empty());
 }
 
 int main()
 {
     test_push_while_pull();
+    test_push_while_pull_until();
+    test_push_while_pull_for();
     return boost::report_errors();
 }