Implement std::condition_variable via pthread_cond_clockwait() where available

[tomcherry]

Summary

std::condition_variable is currently implemented via
pthread_cond_timedwait() on systems that use pthread. This is
problematic, since that function waits by default on CLOCK_REALTIME
and libc++ does not provide any mechanism to change from this
default.

Due to this, regardless of if condition_variable::wait_until() is
called with a chrono::system_clock or chrono::steady_clock parameter,
condition_variable::wait_until() will wait using CLOCK_REALTIME. This
is not accurate to the C++ standard as calling
condition_variable::wait_until() with a chrono::steady_clock parameter
should use CLOCK_MONOTONIC.

This is particularly problematic because CLOCK_REALTIME is a bad
choice as it is subject to discontinuous time adjustments, that may
cause condition_variable::wait_until() to immediately timeout or wait
indefinitely.

This change fixes this issue with a new POSIX function,
pthread_cond_clockwait() proposed on
http://austingroupbugs.net/view.php?id=1216. The new function is
similar to pthread_cond_timedwait() with the addition of a clock
parameter that allows it to wait using either CLOCK_REALTIME or
CLOCK_MONOTONIC, thus allowing condition_variable::wait_until() to
wait using CLOCK_REALTIME for chrono::system_clock and CLOCK_MONOTONIC
for chrono::steady_clock.

pthread_cond_clockwait() is implemented in glibc (2.30 and later) and
Android's bionic (Android API version 30 and later).

This change additionally makes wait_for() and wait_until() with clocks
other than chrono::system_clock use CLOCK_MONOTONIC.

Bug: 35756266
Test: boot, condition variables unaffected by time changes
Test: libc++ tests

Change-Id: I0246d85df106daa7a3e59ed2bb21180c86d6722d

[ldionne]

So, to make sure I understand, this patch uses something that is *not yet* in POSIX, it's only proposed, correct?

Overall, I think this is a good direction since it patches a hole in condition_variable.

include/__mutex_base

423

Can't you just overload on chrono::time_point<chrono::system_clock, _Duration>& and avoid the enable_if dance?

503

Please pardon my naiveness, but what is that number?

[tomcherry]

Clarify that pthread_cond_clockwait() is only proposed to POSIX.
Remove enable_if.

[tomcherry]

So, to make sure I understand, this patch uses something that is *not yet* in POSIX, it's only proposed, correct?

Correct. I updated the commit message to clarify that it's only proposed.

include/__mutex_base

423

Thanks, yes, the new patch simplifies all of this.

503

It's taken from condition_variable.cpp, added in https://reviews.llvm.org/rGaad745a024a3d0fd573ac9649d4c659e6a713702.

I tried to keep all of the safety checks around chrono intact.

[EricWF]

include/__mutex_base

18

Can't we include time.h unconditionally?

308

This deduction is wrong. Either you deduce the entire time point, like we were originally, or you use the timepoint and duration specified by the clock.

You don't specify the clock and deduce the other parameters

Also, fewer overloads is better. We want to declare a _single_ public overload. We can do the magic behind the scenes .

340

This can likely be the same implementation as above.

459

This section seems like it can be generalized to work for both halves of this ifdef.
Having the same code tested in all cases is always better than preprocessor branches.

Forgive me if I'm missing something.

460

using instead of typedef in new code if you don't mind.

461

This assumes the resolution of the steady_clock, no?

495

__lk, __tp, __d, etc.

Everything that isn't a part of the public API needs a reserved name.

503

Lets give it a name so it makes sense.

508

const works the same in all dialects.

512

What does the cast do except silence bugs?

(Same question above).

[ldionne]

include/__mutex_base

308

This deduction is wrong. Either you deduce the entire time point, like we were originally, or you use the timepoint and duration specified by the clock.

Why? I was the one to suggest this in the last diff, because we otherwise needed SFINAE and the resulting code was a lot more complex than it needed to be. What's wrong with this implementation?

[tomcherry]

Responding to review comments

[tomcherry]

I simplified the code a bit more and responded to the comments. I haven't had a good chance to test this yet, but pushing early for feedback.

include/__mutex_base

18

I thought it would be better to not include it if not needed, but I'll make it unconditionally included.

308

The latest patchset has it back to the original public interface.

340

The new patchset does this all with the one public interface.

459

I'm not sure it can. The first branch does the overflow checking with steady_clock and the second does it with system_clock. If we have steady_clock available, we'll want to use it. If we don't have steady_clock available, then we must use system_clock, since all non supported clocks end up calling wait_for() with the difference between their now() and the input time.

460

Will do.

461

Sort of. It mostly assumes that steady_clock::now() can be converted into nanoseconds safely, which I think is a reasonable assumption to make. The old code assumed the resolution of system_clock too, so it's not a regression from there.

I'm happy to hear a better solution, because both this code and the original have never sit well with me. What we really want is a safe_duration_cast, where we get a ceiling effect if there would be overflow. Since that is not how std::chrono was created, we're left with two options:

1. This function ends up with signed integer overflow if someone passes milliseconds::max() to it and steady/system_clock.
2. Work-arounds like this

495

Got it, done in next patch.

503

In retrospect, I removed it. There's no accompanying comments on why it was added originally and there's no obvious reason why it would be needed.

508

It's constexpr in the original code, but const is sufficient, so I changed it to that.

512

The first cast makes sense, we don't know what size time_t is, but we know that if s.count() < ts_sec_max, that we can convert into it.

The second cast isn't needed, removed.

[mclow.lists]

include/__mutex_base

461

What we really want is a safe_duration_cast, where we get a ceiling effect if there would be overflow. Since that is not how std::chrono was created, we're left with two options:

I suggest a third option: Write __safe_duration_cast

[tomcherry]

Add __safe_nanosecond_cast

[tomcherry]

include/__mutex_base

461

I intentionally didn't suggest that since that full solution would be out of the scope of this change, and I find it strange that we'd implement a private version, when we really do need a public safe_duration_cast.

But I did write something similar today: a safe_nanosecond_cast that only handles the cases needed for this code.

I assume that I cannot use __builtin_mul_overflow or similar in libcxx, so I avoided those, though would happily add them back if it were acceptable.

[tomcherry]

Are there any other comments for this change? I think my latest patchset takes into consideration the previous comments.

Thanks!

[danalbert]

LGTM. @EricWF any further comments before I submit this?

[tomcherry]

Restore original behavior that cv_status::timeout is immediately returned if the absolute time of wait_until() has already passed.
Fix return 0 location in test.

[tomcherry]

Add an extra conversion since clocks might not necessarily return nanoseconds from time_from_epoch().

[danalbert]

Closed by https://reviews.llvm.org/rL372016

include/__config

 #    define _LIBCPP_HAS_THREAD_API_PTHREAD
 #  elif defined(_LIBCPP_WIN32API)
 #    define _LIBCPP_HAS_THREAD_API_WIN32
 #  else
 #    error "No thread API"
 #  endif // _LIBCPP_HAS_THREAD_API
 #endif // _LIBCPP_HAS_NO_THREADS
 
+#if defined(_LIBCPP_HAS_THREAD_API_PTHREAD)
+#  if (defined(__ANDROID__) && __ANDROID_API__ >= 30) || _LIBCPP_GLIBC_PREREQ(2, 30)
+#    define _LIBCPP_HAS_COND_CLOCKWAIT
+#  endif
+#endif
+
 #if defined(_LIBCPP_HAS_NO_THREADS) && defined(_LIBCPP_HAS_THREAD_API_PTHREAD)
 #error _LIBCPP_HAS_THREAD_API_PTHREAD may only be defined when \
        _LIBCPP_HAS_NO_THREADS is not defined.
 #endif
 
 #if defined(_LIBCPP_HAS_NO_THREADS) && defined(_LIBCPP_HAS_THREAD_API_EXTERNAL)
 #error _LIBCPP_HAS_THREAD_API_EXTERNAL may not be defined when \
        _LIBCPP_HAS_NO_THREADS is defined.

include/__mutex_base

 #ifndef _LIBCPP___MUTEX_BASE
 #define _LIBCPP___MUTEX_BASE
 
 #include <__config>
 #include <chrono>
 #include <system_error>
 #include <__threading_support>
 
+#include <time.h>
 
 #if !defined(_LIBCPP_HAS_NO_PRAGMA_SYSTEM_HEADER)
 #pragma GCC system_header
 #endif
 
 _LIBCPP_PUSH_MACROS
 #include <__undef_macros>
 
#endif
     void notify_one() _NOEXCEPT;
     void notify_all() _NOEXCEPT;
 
     void wait(unique_lock<mutex>& __lk) _NOEXCEPT;
     template <class _Predicate>
         _LIBCPP_METHOD_TEMPLATE_IMPLICIT_INSTANTIATION_VIS
         void wait(unique_lock<mutex>& __lk, _Predicate __pred);
 
     template <class _Clock, class _Duration>
         _LIBCPP_METHOD_TEMPLATE_IMPLICIT_INSTANTIATION_VIS
         cv_status
         wait_until(unique_lock<mutex>& __lk,
                    const chrono::time_point<_Clock, _Duration>& __t);
 
     template <class _Clock, class _Duration, class _Predicate>
         _LIBCPP_METHOD_TEMPLATE_IMPLICIT_INSTANTIATION_VIS
         bool
template <class _Rep, class _Period, class _Predicate>
                  _Predicate __pred);
 
     typedef __libcpp_condvar_t* native_handle_type;
     _LIBCPP_INLINE_VISIBILITY native_handle_type native_handle() {return &__cv_;}
 
 private:
     void __do_timed_wait(unique_lock<mutex>& __lk,
        chrono::time_point<chrono::system_clock, chrono::nanoseconds>) _NOEXCEPT;
+#if defined(_LIBCPP_HAS_COND_CLOCKWAIT)
+    void __do_timed_wait(unique_lock<mutex>& __lk,
+       chrono::time_point<chrono::steady_clock, chrono::nanoseconds>) _NOEXCEPT;
+#endif
+    template <class _Clock>
+    void __do_timed_wait(unique_lock<mutex>& __lk,
+       chrono::time_point<_Clock, chrono::nanoseconds>) _NOEXCEPT;
 };
 #endif // !_LIBCPP_HAS_NO_THREADS
 
-template <class _To, class _Rep, class _Period>
+template <class _Rep, class _Period>
 inline _LIBCPP_INLINE_VISIBILITY
 typename enable_if
 <
-    chrono::__is_duration<_To>::value,
-    _To
+    is_floating_point<_Rep>::value,
+    chrono::nanoseconds
 >::type
-__ceil(chrono::duration<_Rep, _Period> __d)
+__safe_nanosecond_cast(chrono::duration<_Rep, _Period> __d)
 {
     using namespace chrono;
-    _To __r = duration_cast<_To>(__d);
-    if (__r < __d)
-        ++__r;
-    return __r;
+    using __ratio = ratio_divide<_Period, nano>;
+    using __ns_rep = nanoseconds::rep;
+    _Rep __result_float = __d.count() * __ratio::num / __ratio::den;
+
+    _Rep __result_max = numeric_limits<__ns_rep>::max();
+    if (__result_float >= __result_max) {
+        return nanoseconds::max();
+    }
+
+    _Rep __result_min = numeric_limits<__ns_rep>::min();
+    if (__result_float <= __result_min) {
+        return nanoseconds::min();
+    }
+
+    return nanoseconds(static_cast<__ns_rep>(__result_float));
+}
+
+template <class _Rep, class _Period>
+inline _LIBCPP_INLINE_VISIBILITY
+typename enable_if
+<
+    !is_floating_point<_Rep>::value,
+    chrono::nanoseconds
+>::type
+__safe_nanosecond_cast(chrono::duration<_Rep, _Period> __d)
+{
+    using namespace chrono;
+    if (__d.count() == 0) {
+        return nanoseconds(0);
+    }
+
+    using __ratio = ratio_divide<_Period, nano>;
+    using __ns_rep = nanoseconds::rep;
+    __ns_rep __result_max = std::numeric_limits<__ns_rep>::max();
+    if (__d.count() > 0 && __d.count() > __result_max / __ratio::num) {
+        return nanoseconds::max();
+    }
+
+    __ns_rep __result_min = std::numeric_limits<__ns_rep>::min();
+    if (__d.count() < 0 && __d.count() < __result_min / __ratio::num) {
+        return nanoseconds::min();
+    }
+
+    __ns_rep __result = __d.count() * __ratio::num / __ratio::den;
+    if (__result == 0) {
+        return nanoseconds(1);
+    }
+
+    return nanoseconds(__result);
 }
 
 #ifndef _LIBCPP_HAS_NO_THREADS
 template <class _Predicate>
 void
 condition_variable::wait(unique_lock<mutex>& __lk, _Predicate __pred)
 {
     while (!__pred())
         wait(__lk);
 }
 
 template <class _Clock, class _Duration>
 cv_status
 condition_variable::wait_until(unique_lock<mutex>& __lk,
                                const chrono::time_point<_Clock, _Duration>& __t)
 {
     using namespace chrono;
-    wait_for(__lk, __t - _Clock::now());
+    using __clock_tp_ns = time_point<_Clock, nanoseconds>;
+
+    typename _Clock::time_point __now = _Clock::now();
+
+    __clock_tp_ns __t_ns = __clock_tp_ns(__safe_nanosecond_cast(__t.time_since_epoch()));
+
+    __do_timed_wait(__lk, __t_ns);
     return _Clock::now() < __t ? cv_status::no_timeout : cv_status::timeout;
 }
 
 template <class _Clock, class _Duration, class _Predicate>
 bool
 condition_variable::wait_until(unique_lock<mutex>& __lk,
                    const chrono::time_point<_Clock, _Duration>& __t,
                    _Predicate __pred)

 template <class _Rep, class _Period>
 cv_status
 condition_variable::wait_for(unique_lock<mutex>& __lk,
                              const chrono::duration<_Rep, _Period>& __d)
 {
     using namespace chrono;
     if (__d <= __d.zero())
         return cv_status::timeout;
-    typedef time_point<system_clock, duration<long double, nano> > __sys_tpf;
-    typedef time_point<system_clock, nanoseconds> __sys_tpi;
-    __sys_tpf _Max = __sys_tpi::max();
+    using __ns_rep = nanoseconds::rep;
     steady_clock::time_point __c_now = steady_clock::now();
-    system_clock::time_point __s_now = system_clock::now();
-    if (_Max - __d > __s_now)
-        __do_timed_wait(__lk, __s_now + __ceil<nanoseconds>(__d));
-    else
-        __do_timed_wait(__lk, __sys_tpi::max());
+
+#if defined(_LIBCPP_HAS_COND_CLOCKWAIT)
+    using __clock_tp_ns = time_point<steady_clock, nanoseconds>;
+    __ns_rep __now_count = __c_now.time_since_epoch().count();
+#else
+    using __clock_tp_ns = time_point<system_clock, nanoseconds>;
+    __ns_rep __now_count = system_clock::now().time_since_epoch().count();
+#endif
+
+    __ns_rep __d_ns_count = __safe_nanosecond_cast(__d).count();
+
+    if (__now_count > numeric_limits<__ns_rep>::max() - __d_ns_count) {
+        __do_timed_wait(__lk, __clock_tp_ns::max());
+    } else {
+        __do_timed_wait(__lk, __clock_tp_ns(nanoseconds(__now_count + __d_ns_count)));
+    }
+
     return steady_clock::now() - __c_now < __d ? cv_status::no_timeout :
                                                  cv_status::timeout;
 }
 
 template <class _Rep, class _Period, class _Predicate>
 inline
 bool
 condition_variable::wait_for(unique_lock<mutex>& __lk,
                              const chrono::duration<_Rep, _Period>& __d,
                              _Predicate __pred)
 {
     return wait_until(__lk, chrono::steady_clock::now() + __d,
                       _VSTD::move(__pred));
 }
 
+#if defined(_LIBCPP_HAS_COND_CLOCKWAIT)
+inline
+void
+condition_variable::__do_timed_wait(unique_lock<mutex>& __lk,
+     chrono::time_point<chrono::steady_clock, chrono::nanoseconds> __tp) _NOEXCEPT
+{
+    using namespace chrono;
+    if (!__lk.owns_lock())
+        __throw_system_error(EPERM,
+                            "condition_variable::timed wait: mutex not locked");
+    nanoseconds __d = __tp.time_since_epoch();
+    timespec __ts;
+    seconds __s = duration_cast<seconds>(__d);
+    using __ts_sec = decltype(__ts.tv_sec);
+    const __ts_sec __ts_sec_max = numeric_limits<__ts_sec>::max();
+    if (__s.count() < __ts_sec_max)
+    {
+        __ts.tv_sec = static_cast<__ts_sec>(__s.count());
+        __ts.tv_nsec = (__d - __s).count();
+    }
+    else
+    {
+        __ts.tv_sec = __ts_sec_max;
+        __ts.tv_nsec = giga::num - 1;
+    }
+    int __ec = pthread_cond_clockwait(&__cv_, __lk.mutex()->native_handle(), CLOCK_MONOTONIC, &__ts);
+    if (__ec != 0 && __ec != ETIMEDOUT)
+        __throw_system_error(__ec, "condition_variable timed_wait failed");
+}
+#endif // _LIBCPP_HAS_COND_CLOCKWAIT
+
+template <class _Clock>
+inline
+void
+condition_variable::__do_timed_wait(unique_lock<mutex>& __lk,
+     chrono::time_point<_Clock, chrono::nanoseconds> __tp) _NOEXCEPT
+{
+    wait_for(__lk, __tp - _Clock::now());
+}
+
 #endif // !_LIBCPP_HAS_NO_THREADS
 
 _LIBCPP_END_NAMESPACE_STD
 
 _LIBCPP_POP_MACROS
 
 #endif  // _LIBCPP___MUTEX_BASE

test/std/thread/thread.condition/thread.condition.condvar/wait_until.pass.cpp

 #include <condition_variable>
 #include <mutex>
 #include <thread>
 #include <chrono>
 #include <cassert>
 
 #include "test_macros.h"
 
-struct Clock
+struct TestClock
 {
     typedef std::chrono::milliseconds duration;
     typedef duration::rep             rep;
     typedef duration::period          period;
-    typedef std::chrono::time_point<Clock> time_point;
+    typedef std::chrono::time_point<TestClock> time_point;
     static const bool is_steady =  true;
 
     static time_point now()
     {
         using namespace std::chrono;
         return time_point(duration_cast<duration>(
                 steady_clock::now().time_since_epoch()
                                                  ));
     }
 };
 
 std::condition_variable cv;
 std::mutex mut;
 
 int test1 = 0;
 int test2 = 0;
 
 int runs = 0;
 
+template <typename Clock>
 void f()
 {
     std::unique_lock<std::mutex> lk(mut);
     assert(test2 == 0);
     test1 = 1;
     cv.notify_one();
-    Clock::time_point t0 = Clock::now();
-    Clock::time_point t = t0 + Clock::duration(250);
+    typename Clock::time_point t0 = Clock::now();
+    typename Clock::time_point t = t0 + std::chrono::milliseconds(250);
     while (test2 == 0 && cv.wait_until(lk, t) == std::cv_status::no_timeout)
         ;
-    Clock::time_point t1 = Clock::now();
+    typename Clock::time_point t1 = Clock::now();
     if (runs == 0)
     {
-        assert(t1 - t0 < Clock::duration(250));
+        assert(t1 - t0 < std::chrono::milliseconds(250));
         assert(test2 != 0);
     }
     else
     {
-        assert(t1 - t0 - Clock::duration(250) < Clock::duration(50));
+        assert(t1 - t0 - std::chrono::milliseconds(250) < std::chrono::milliseconds(50));
         assert(test2 == 0);
     }
     ++runs;
 }
 
-int main(int, char**)
+template <typename Clock>
+void run_test()
 {
+    runs = 0;
+    test1 = 0;
+    test2 = 0;
     {
         std::unique_lock<std::mutex>lk(mut);
-        std::thread t(f);
+        std::thread t(f<Clock>);
         assert(test1 == 0);
         while (test1 == 0)
             cv.wait(lk);
         assert(test1 != 0);
         test2 = 1;
         lk.unlock();
         cv.notify_one();
         t.join();
     }
     test1 = 0;
     test2 = 0;
     {
         std::unique_lock<std::mutex>lk(mut);
-        std::thread t(f);
+        std::thread t(f<Clock>);
         assert(test1 == 0);
         while (test1 == 0)
             cv.wait(lk);
         assert(test1 != 0);
         lk.unlock();
         t.join();
     }
 
   return 0;
 }
+
+int main(int, char**)
+{
+    run_test<TestClock>();
+    run_test<std::chrono::steady_clock>();
+    run_test<std::chrono::system_clock>();
+}