Skip to content

Commit ed37a45

Browse files
derekmaurocopybara-github
derekmauro
authored and
copybara-github
committed
Synchronization: Add support for true relative timeouts using
monotonic clocks on Linux when the implementation uses futexes After this change, when synchronization methods that wait are passed an absl::Duration to limit the wait time, these methods will wait for that interval, even if the system clock is changed (subject to any limitations with how CLOCK_MONOTONIC keeps track of time). In other words, an observer measuring the time with a stop watch will now see the correct interval, even if the system clock is changed. Previously, the duration was added to the current time, and methods would wait until that time was reached on the possibly changed realtime system clock. The behavior of the synchronization methods that take an absl::Time is unchanged. These methods always wait until the absolute point in time is reached and respect changes to the system clock. In other words, an observer will always see the timeout occur when a wall clock reaches that time, even if the clock is manipulated externally. Note: ABSL_PREDICT_FALSE was removed from the error case in Futex as timeouts are handled by this case, and timeouts are part of normal operation. PiperOrigin-RevId: 510405347 Change-Id: I0b3ea390de97014cfa353079ae2e0c1c637aca69
1 parent 0372af1 commit ed37a45

File tree

4 files changed

+177
-47
lines changed

4 files changed

+177
-47
lines changed

‎absl/synchronization/internal/futex.h‎

Lines changed: 47 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -16,9 +16,7 @@
1616

1717
#include "absl/base/config.h"
1818

19-
#ifdef _WIN32
20-
#include <windows.h>
21-
#else
19+
#ifndef _WIN32
2220
#include <sys/time.h>
2321
#include <unistd.h>
2422
#endif
@@ -85,34 +83,60 @@ namespace synchronization_internal {
8583

8684
class FutexImpl {
8785
public:
88-
static int WaitUntil(std::atomic<int32_t> *v, int32_t val,
86+
// Atomically check that `*v == val`, and if it is, then sleep until the
87+
// timeout `t` has been reached, or until woken by `Wake()`.
88+
static int WaitUntil(std::atomic<int32_t>* v, int32_t val,
8989
KernelTimeout t) {
90-
long err = 0; // NOLINT(runtime/int)
91-
if (t.has_timeout()) {
92-
// https://locklessinc.com/articles/futex_cheat_sheet/
93-
// Unlike FUTEX_WAIT, FUTEX_WAIT_BITSET uses absolute time.
94-
struct timespec abs_timeout = t.MakeAbsTimespec();
95-
// Atomically check that the futex value is still 0, and if it
96-
// is, sleep until abs_timeout or until woken by FUTEX_WAKE.
97-
err = syscall(
98-
SYS_futex, reinterpret_cast<int32_t *>(v),
99-
FUTEX_WAIT_BITSET | FUTEX_PRIVATE_FLAG | FUTEX_CLOCK_REALTIME, val,
100-
&abs_timeout, nullptr, FUTEX_BITSET_MATCH_ANY);
90+
if (!t.has_timeout()) {
91+
return Wait(v, val);
92+
} else if (t.is_absolute_timeout()) {
93+
auto abs_timespec = t.MakeAbsTimespec();
94+
return WaitAbsoluteTimeout(v, val, &abs_timespec);
10195
} else {
102-
// Atomically check that the futex value is still 0, and if it
103-
// is, sleep until woken by FUTEX_WAKE.
104-
err = syscall(SYS_futex, reinterpret_cast<int32_t *>(v),
105-
FUTEX_WAIT | FUTEX_PRIVATE_FLAG, val, nullptr);
96+
auto rel_timespec = t.MakeRelativeTimespec();
97+
return WaitRelativeTimeout(v, val, &rel_timespec);
10698
}
107-
if (ABSL_PREDICT_FALSE(err != 0)) {
99+
}
100+
101+
// Atomically check that `*v == val`, and if it is, then sleep until the until
102+
// woken by `Wake()`.
103+
static int Wait(std::atomic<int32_t>* v, int32_t val) {
104+
return WaitAbsoluteTimeout(v, val, nullptr);
105+
}
106+
107+
// Atomically check that `*v == val`, and if it is, then sleep until
108+
// CLOCK_REALTIME reaches `*abs_timeout`, or until woken by `Wake()`.
109+
static int WaitAbsoluteTimeout(std::atomic<int32_t>* v, int32_t val,
110+
const struct timespec* abs_timeout) {
111+
// https://locklessinc.com/articles/futex_cheat_sheet/
112+
// Unlike FUTEX_WAIT, FUTEX_WAIT_BITSET uses absolute time.
113+
auto err =
114+
syscall(SYS_futex, reinterpret_cast<int32_t*>(v),
115+
FUTEX_WAIT_BITSET | FUTEX_PRIVATE_FLAG | FUTEX_CLOCK_REALTIME,
116+
val, abs_timeout, nullptr, FUTEX_BITSET_MATCH_ANY);
117+
if (err != 0) {
118+
return -errno;
119+
}
120+
return 0;
121+
}
122+
123+
// Atomically check that `*v == val`, and if it is, then sleep until
124+
// `*rel_timeout` has elapsed, or until woken by `Wake()`.
125+
static int WaitRelativeTimeout(std::atomic<int32_t>* v, int32_t val,
126+
const struct timespec* rel_timeout) {
127+
// Atomically check that the futex value is still 0, and if it
128+
// is, sleep until abs_timeout or until woken by FUTEX_WAKE.
129+
auto err = syscall(SYS_futex, reinterpret_cast<int32_t*>(v),
130+
FUTEX_PRIVATE_FLAG, val, rel_timeout);
131+
if (err != 0) {
108132
return -errno;
109133
}
110134
return 0;
111135
}
112136

113-
static int Wake(std::atomic<int32_t> *v, int32_t count) {
114-
// NOLINTNEXTLINE(runtime/int)
115-
long err = syscall(SYS_futex, reinterpret_cast<int32_t*>(v),
137+
// Wakes at most `count` waiters that have entered the sleep state on `v`.
138+
static int Wake(std::atomic<int32_t>* v, int32_t count) {
139+
auto err = syscall(SYS_futex, reinterpret_cast<int32_t*>(v),
116140
FUTEX_WAKE | FUTEX_PRIVATE_FLAG, count);
117141
if (ABSL_PREDICT_FALSE(err < 0)) {
118142
return -errno;

‎absl/synchronization/internal/waiter.cc‎

Lines changed: 101 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -67,11 +67,9 @@ static void MaybeBecomeIdle() {
6767

6868
#if ABSL_WAITER_MODE == ABSL_WAITER_MODE_FUTEX
6969

70-
Waiter::Waiter() {
71-
futex_.store(0, std::memory_order_relaxed);
72-
}
70+
Waiter::Waiter() : futex_(0) {}
7371

74-
bool Waiter::Wait(KernelTimeout t) {
72+
bool Waiter::WaitAbsoluteTimeout(KernelTimeout t) {
7573
// Loop until we can atomically decrement futex from a positive
7674
// value, waiting on a futex while we believe it is zero.
7775
// Note that, since the thread ticker is just reset, we don't need to check
@@ -90,7 +88,88 @@ bool Waiter::Wait(KernelTimeout t) {
9088
}
9189

9290
if (!first_pass) MaybeBecomeIdle();
93-
const int err = Futex::WaitUntil(&futex_, 0, t);
91+
auto abs_timeout = t.MakeAbsTimespec();
92+
const int err = Futex::WaitAbsoluteTimeout(&futex_, 0, &abs_timeout);
93+
if (err != 0) {
94+
if (err == -EINTR || err == -EWOULDBLOCK) {
95+
// Do nothing, the loop will retry.
96+
} else if (err == -ETIMEDOUT) {
97+
return false;
98+
} else {
99+
ABSL_RAW_LOG(FATAL, "Futex operation failed with error %d\n", err);
100+
}
101+
}
102+
first_pass = false;
103+
}
104+
}
105+
106+
#ifdef CLOCK_MONOTONIC
107+
108+
// Subtracts the timespec `sub` from `in` if the result would not be negative,
109+
// and returns true. Returns false if the result would be negative, and leaves
110+
// `in` unchanged.
111+
static bool TimespecSubtract(struct timespec& in, const struct timespec& sub) {
112+
if (in.tv_sec < sub.tv_sec) {
113+
return false;
114+
}
115+
if (in.tv_nsec < sub.tv_nsec) {
116+
if (in.tv_sec == sub.tv_sec) {
117+
return false;
118+
}
119+
// Borrow from tv_sec.
120+
in.tv_sec -= 1;
121+
in.tv_nsec += 1'000'000'000;
122+
}
123+
in.tv_sec -= sub.tv_sec;
124+
in.tv_nsec -= sub.tv_nsec;
125+
return true;
126+
}
127+
128+
// On some platforms a background thread periodically calls `Poke()` to briefly
129+
// wake waiter threads so that they may call `MaybeBecomeIdle()`. This means
130+
// that `WaitRelativeTimeout()` differs slightly from `WaitAbsoluteTimeout()`
131+
// because it must adjust the timeout by the amount of time that it has already
132+
// slept.
133+
bool Waiter::WaitRelativeTimeout(KernelTimeout t) {
134+
struct timespec start;
135+
ABSL_RAW_CHECK(clock_gettime(CLOCK_MONOTONIC, &start) == 0,
136+
"clock_gettime() failed");
137+
138+
// Loop until we can atomically decrement futex from a positive
139+
// value, waiting on a futex while we believe it is zero.
140+
// Note that, since the thread ticker is just reset, we don't need to check
141+
// whether the thread is idle on the very first pass of the loop.
142+
bool first_pass = true;
143+
144+
while (true) {
145+
int32_t x = futex_.load(std::memory_order_relaxed);
146+
while (x != 0) {
147+
if (!futex_.compare_exchange_weak(x, x - 1,
148+
std::memory_order_acquire,
149+
std::memory_order_relaxed)) {
150+
continue; // Raced with someone, retry.
151+
}
152+
return true; // Consumed a wakeup, we are done.
153+
}
154+
155+
auto relative_timeout = t.MakeRelativeTimespec();
156+
if (!first_pass) {
157+
MaybeBecomeIdle();
158+
159+
// Adjust relative_timeout for `Poke()`s.
160+
struct timespec now;
161+
ABSL_RAW_CHECK(clock_gettime(CLOCK_MONOTONIC, &now) == 0,
162+
"clock_gettime() failed");
163+
// If TimespecSubstract(now, start) returns false, then the clock isn't
164+
// truly monotonic.
165+
if (TimespecSubtract(now, start)) {
166+
if (!TimespecSubtract(relative_timeout, now)) {
167+
return false; // Timeout.
168+
}
169+
}
170+
}
171+
172+
const int err = Futex::WaitRelativeTimeout(&futex_, 0, &relative_timeout);
94173
if (err != 0) {
95174
if (err == -EINTR || err == -EWOULDBLOCK) {
96175
// Do nothing, the loop will retry.
@@ -104,6 +183,23 @@ bool Waiter::Wait(KernelTimeout t) {
104183
}
105184
}
106185

186+
#else // CLOCK_MONOTONIC
187+
188+
// No support for CLOCK_MONOTONIC.
189+
// KernelTimeout will automatically convert to an absolute timeout.
190+
bool Waiter::WaitRelativeTimeout(KernelTimeout t) {
191+
return WaitAbsoluteTimeout(t);
192+
}
193+
194+
#endif // CLOCK_MONOTONIC
195+
196+
bool Waiter::Wait(KernelTimeout t) {
197+
if (t.is_absolute_timeout()) {
198+
return WaitAbsoluteTimeout(t);
199+
}
200+
return WaitRelativeTimeout(t);
201+
}
202+
107203
void Waiter::Post() {
108204
if (futex_.fetch_add(1, std::memory_order_release) == 0) {
109205
// We incremented from 0, need to wake a potential waiter.

‎absl/synchronization/internal/waiter.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,6 +110,9 @@ class Waiter {
110110
~Waiter() = delete;
111111

112112
#if ABSL_WAITER_MODE == ABSL_WAITER_MODE_FUTEX
113+
bool WaitAbsoluteTimeout(KernelTimeout t);
114+
bool WaitRelativeTimeout(KernelTimeout t);
115+
113116
// Futexes are defined by specification to be 32-bits.
114117
// Thus std::atomic<int32_t> must be just an int32_t with lockfree methods.
115118
std::atomic<int32_t> futex_;

‎absl/synchronization/mutex.cc‎

Lines changed: 26 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -635,21 +635,6 @@ void Mutex::InternalAttemptToUseMutexInFatalSignalHandler() {
635635
std::memory_order_release);
636636
}
637637

638-
// --------------------------time support
639-
640-
// Return the current time plus the timeout. Use the same clock as
641-
// PerThreadSem::Wait() for consistency. Unfortunately, we don't have
642-
// such a choice when a deadline is given directly.
643-
static absl::Time DeadlineFromTimeout(absl::Duration timeout) {
644-
#ifndef _WIN32
645-
struct timeval tv;
646-
gettimeofday(&tv, nullptr);
647-
return absl::TimeFromTimeval(tv) + timeout;
648-
#else
649-
return absl::Now() + timeout;
650-
#endif
651-
}
652-
653638
// --------------------------Mutexes
654639

655640
// In the layout below, the msb of the bottom byte is currently unused. Also,
@@ -1546,7 +1531,13 @@ void Mutex::LockWhen(const Condition &cond) {
15461531
}
15471532

15481533
bool Mutex::LockWhenWithTimeout(const Condition &cond, absl::Duration timeout) {
1549-
return LockWhenWithDeadline(cond, DeadlineFromTimeout(timeout));
1534+
ABSL_TSAN_MUTEX_PRE_LOCK(this, 0);
1535+
GraphId id = DebugOnlyDeadlockCheck(this);
1536+
bool res = LockSlowWithDeadline(kExclusive, &cond,
1537+
KernelTimeout(timeout), 0);
1538+
DebugOnlyLockEnter(this, id);
1539+
ABSL_TSAN_MUTEX_POST_LOCK(this, 0, 0);
1540+
return res;
15501541
}
15511542

15521543
bool Mutex::LockWhenWithDeadline(const Condition &cond, absl::Time deadline) {
@@ -1569,7 +1560,12 @@ void Mutex::ReaderLockWhen(const Condition &cond) {
15691560

15701561
bool Mutex::ReaderLockWhenWithTimeout(const Condition &cond,
15711562
absl::Duration timeout) {
1572-
return ReaderLockWhenWithDeadline(cond, DeadlineFromTimeout(timeout));
1563+
ABSL_TSAN_MUTEX_PRE_LOCK(this, __tsan_mutex_read_lock);
1564+
GraphId id = DebugOnlyDeadlockCheck(this);
1565+
bool res = LockSlowWithDeadline(kShared, &cond, KernelTimeout(timeout), 0);
1566+
DebugOnlyLockEnter(this, id);
1567+
ABSL_TSAN_MUTEX_POST_LOCK(this, __tsan_mutex_read_lock, 0);
1568+
return res;
15731569
}
15741570

15751571
bool Mutex::ReaderLockWhenWithDeadline(const Condition &cond,
@@ -1594,7 +1590,18 @@ void Mutex::Await(const Condition &cond) {
15941590
}
15951591

15961592
bool Mutex::AwaitWithTimeout(const Condition &cond, absl::Duration timeout) {
1597-
return AwaitWithDeadline(cond, DeadlineFromTimeout(timeout));
1593+
if (cond.Eval()) { // condition already true; nothing to do
1594+
if (kDebugMode) {
1595+
this->AssertReaderHeld();
1596+
}
1597+
return true;
1598+
}
1599+
1600+
KernelTimeout t{timeout};
1601+
bool res = this->AwaitCommon(cond, t);
1602+
ABSL_RAW_CHECK(res || t.has_timeout(),
1603+
"condition untrue on return from Await");
1604+
return res;
15981605
}
15991606

16001607
bool Mutex::AwaitWithDeadline(const Condition &cond, absl::Time deadline) {
@@ -2660,7 +2667,7 @@ bool CondVar::WaitCommon(Mutex *mutex, KernelTimeout t) {
26602667
}
26612668

26622669
bool CondVar::WaitWithTimeout(Mutex *mu, absl::Duration timeout) {
2663-
return WaitWithDeadline(mu, DeadlineFromTimeout(timeout));
2670+
return WaitCommon(mu, KernelTimeout(timeout));
26642671
}
26652672

26662673
bool CondVar::WaitWithDeadline(Mutex *mu, absl::Time deadline) {

0 commit comments

Comments
 (0)