diff options
Diffstat (limited to 'absl/synchronization/internal/waiter.cc')
| -rw-r--r-- | absl/synchronization/internal/waiter.cc | 394 |
1 files changed, 394 insertions, 0 deletions
diff --git a/absl/synchronization/internal/waiter.cc b/absl/synchronization/internal/waiter.cc new file mode 100644 index 00000000..cd16c788 --- /dev/null +++ b/absl/synchronization/internal/waiter.cc @@ -0,0 +1,394 @@ +// Copyright 2017 The Abseil Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "absl/synchronization/internal/waiter.h" + +#include "absl/base/config.h" + +#ifdef _WIN32 +#include <windows.h> +#else +#include <pthread.h> +#include <sys/time.h> +#include <unistd.h> +#endif + +#ifdef __linux__ +#include <linux/futex.h> +#include <sys/syscall.h> +#endif + +#ifdef ABSL_HAVE_SEMAPHORE_H +#include <semaphore.h> +#endif + +#include <errno.h> +#include <stdio.h> +#include <time.h> + +#include <atomic> +#include <cassert> + +#include "absl/base/internal/malloc_extension.h" +#include "absl/base/internal/raw_logging.h" +#include "absl/base/internal/thread_identity.h" +#include "absl/synchronization/internal/kernel_timeout.h" + +namespace absl { +namespace synchronization_internal { + +static void MaybeBecomeIdle() { + base_internal::ThreadIdentity *identity = + base_internal::CurrentThreadIdentityIfPresent(); + assert(identity != nullptr); + const bool is_idle = identity->is_idle.load(std::memory_order_relaxed); + const int ticker = identity->ticker.load(std::memory_order_relaxed); + const int wait_start = identity->wait_start.load(std::memory_order_relaxed); + if (!is_idle && ticker - wait_start > Waiter::kIdlePeriods) { + identity->is_idle.store(true, std::memory_order_relaxed); + base_internal::MallocExtension::instance()->MarkThreadIdle(); + } +} + +#if ABSL_WAITER_MODE == ABSL_WAITER_MODE_FUTEX + +// Some Android headers are missing these definitions even though they +// support these futex operations. +#ifdef __BIONIC__ +#ifndef SYS_futex +#define SYS_futex __NR_futex +#endif +#ifndef FUTEX_WAIT_BITSET +#define FUTEX_WAIT_BITSET 9 +#endif +#ifndef FUTEX_PRIVATE_FLAG +#define FUTEX_PRIVATE_FLAG 128 +#endif +#ifndef FUTEX_CLOCK_REALTIME +#define FUTEX_CLOCK_REALTIME 256 +#endif +#ifndef FUTEX_BITSET_MATCH_ANY +#define FUTEX_BITSET_MATCH_ANY 0xFFFFFFFF +#endif +#endif + +void Waiter::Init() { + futex_.store(0, std::memory_order_relaxed); +} + +bool Waiter::Wait(KernelTimeout t) { + // Loop until we can atomically decrement futex from a positive + // value, waiting on a futex while we believe it is zero. + while (true) { + int x = futex_.load(std::memory_order_relaxed); + if (x != 0) { + if (!futex_.compare_exchange_weak(x, x - 1, + std::memory_order_acquire, + std::memory_order_relaxed)) { + continue; // Raced with someone, retry. + } + return true; // Consumed a wakeup, we are done. + } + + int err = 0; + if (t.has_timeout()) { + // https://locklessinc.com/articles/futex_cheat_sheet/ + // Unlike FUTEX_WAIT, FUTEX_WAIT_BITSET uses absolute time. + struct timespec abs_timeout = t.MakeAbsTimespec(); + // Atomically check that the futex value is still 0, and if it + // is, sleep until abs_timeout or until woken by FUTEX_WAKE. + err = syscall( + SYS_futex, reinterpret_cast<int *>(&futex_), + FUTEX_WAIT_BITSET | FUTEX_PRIVATE_FLAG | FUTEX_CLOCK_REALTIME, 0, + &abs_timeout, nullptr, FUTEX_BITSET_MATCH_ANY); + } else { + // Atomically check that the futex value is still 0, and if it + // is, sleep until woken by FUTEX_WAKE. + err = syscall(SYS_futex, reinterpret_cast<int *>(&futex_), + FUTEX_WAIT | FUTEX_PRIVATE_FLAG, 0, nullptr); + } + if (err != 0) { + if (errno == EINTR || errno == EWOULDBLOCK) { + // Do nothing, the loop will retry. + } else if (errno == ETIMEDOUT) { + return false; // Timeout. + } else { + ABSL_RAW_LOG(FATAL, "Futex operation failed with errno %d\n", errno); + } + } + + MaybeBecomeIdle(); + } +} + +void Waiter::Post() { + if (futex_.fetch_add(1, std::memory_order_release) == 0) { + // We incremented from 0, need to wake a potential waker. + Poke(); + } +} + +void Waiter::Poke() { + // Wake one thread waiting on the futex. + int err = syscall(SYS_futex, reinterpret_cast<int *>(&futex_), + FUTEX_WAKE | FUTEX_PRIVATE_FLAG, 1); + if (err < 0) { + ABSL_RAW_LOG(FATAL, "FUTEX_WAKE failed with errno %d\n", errno); + } +} + +#elif ABSL_WAITER_MODE == ABSL_WAITER_MODE_CONDVAR + +class PthreadMutexHolder { + public: + explicit PthreadMutexHolder(pthread_mutex_t *mu) : mu_(mu) { + const int err = pthread_mutex_lock(mu_); + if (err != 0) { + ABSL_RAW_LOG(FATAL, "pthread_mutex_lock failed: %d", err); + } + } + + PthreadMutexHolder(const PthreadMutexHolder &rhs) = delete; + PthreadMutexHolder &operator=(const PthreadMutexHolder &rhs) = delete; + + ~PthreadMutexHolder() { + const int err = pthread_mutex_unlock(mu_); + if (err != 0) { + ABSL_RAW_LOG(FATAL, "pthread_mutex_unlock failed: %d", err); + } + } + + private: + pthread_mutex_t *mu_; +}; + +void Waiter::Init() { + const int err = pthread_mutex_init(&mu_, 0); + if (err != 0) { + ABSL_RAW_LOG(FATAL, "pthread_mutex_init failed: %d", err); + } + + const int err2 = pthread_cond_init(&cv_, 0); + if (err2 != 0) { + ABSL_RAW_LOG(FATAL, "pthread_cond_init failed: %d", err2); + } + + waiter_count_.store(0, std::memory_order_relaxed); + wakeup_count_.store(0, std::memory_order_relaxed); +} + +bool Waiter::Wait(KernelTimeout t) { + struct timespec abs_timeout; + if (t.has_timeout()) { + abs_timeout = t.MakeAbsTimespec(); + } + + PthreadMutexHolder h(&mu_); + waiter_count_.fetch_add(1, std::memory_order_relaxed); + // Loop until we find a wakeup to consume or timeout. + while (true) { + int x = wakeup_count_.load(std::memory_order_relaxed); + if (x != 0) { + if (!wakeup_count_.compare_exchange_weak(x, x - 1, + std::memory_order_acquire, + std::memory_order_relaxed)) { + continue; // Raced with someone, retry. + } + // Successfully consumed a wakeup, we're done. + waiter_count_.fetch_sub(1, std::memory_order_relaxed); + return true; + } + + // No wakeups available, time to wait. + if (!t.has_timeout()) { + const int err = pthread_cond_wait(&cv_, &mu_); + if (err != 0) { + ABSL_RAW_LOG(FATAL, "pthread_cond_wait failed: %d", err); + } + } else { + const int err = pthread_cond_timedwait(&cv_, &mu_, &abs_timeout); + if (err == ETIMEDOUT) { + waiter_count_.fetch_sub(1, std::memory_order_relaxed); + return false; + } + if (err != 0) { + ABSL_RAW_LOG(FATAL, "pthread_cond_wait failed: %d", err); + } + } + MaybeBecomeIdle(); + } +} + +void Waiter::Post() { + wakeup_count_.fetch_add(1, std::memory_order_release); + Poke(); +} + +void Waiter::Poke() { + if (waiter_count_.load(std::memory_order_relaxed) == 0) { + return; + } + // Potentially a waker. Take the lock and check again. + PthreadMutexHolder h(&mu_); + if (waiter_count_.load(std::memory_order_relaxed) == 0) { + return; + } + const int err = pthread_cond_signal(&cv_); + if (err != 0) { + ABSL_RAW_LOG(FATAL, "pthread_cond_signal failed: %d", err); + } +} + +#elif ABSL_WAITER_MODE == ABSL_WAITER_MODE_SEM + +void Waiter::Init() { + if (sem_init(&sem_, 0, 0) != 0) { + ABSL_RAW_LOG(FATAL, "sem_init failed with errno %d\n", errno); + } + wakeups_.store(0, std::memory_order_relaxed); +} + +bool Waiter::Wait(KernelTimeout t) { + struct timespec abs_timeout; + if (t.has_timeout()) { + abs_timeout = t.MakeAbsTimespec(); + } + + // Loop until we timeout or consume a wakeup. + while (true) { + int x = wakeups_.load(std::memory_order_relaxed); + if (x != 0) { + if (!wakeups_.compare_exchange_weak(x, x - 1, + std::memory_order_acquire, + std::memory_order_relaxed)) { + continue; // Raced with someone, retry. + } + // Successfully consumed a wakeup, we're done. + return true; + } + + // Nothing to consume, wait (looping on EINTR). + while (true) { + if (!t.has_timeout()) { + if (sem_wait(&sem_) == 0) break; + if (errno == EINTR) continue; + ABSL_RAW_LOG(FATAL, "sem_wait failed: %d", errno); + } else { + if (sem_timedwait(&sem_, &abs_timeout) == 0) break; + if (errno == EINTR) continue; + if (errno == ETIMEDOUT) return false; + ABSL_RAW_LOG(FATAL, "sem_timedwait failed: %d", errno); + } + } + MaybeBecomeIdle(); + } +} + +void Waiter::Post() { + wakeups_.fetch_add(1, std::memory_order_release); // Post a wakeup. + Poke(); +} + +void Waiter::Poke() { + if (sem_post(&sem_) != 0) { // Wake any semaphore waiter. + ABSL_RAW_LOG(FATAL, "sem_post failed with errno %d\n", errno); + } +} + +#elif ABSL_WAITER_MODE == ABSL_WAITER_MODE_WIN32 + +class LockHolder { + public: + explicit LockHolder(SRWLOCK* mu) : mu_(mu) { + AcquireSRWLockExclusive(mu_); + } + + LockHolder(const LockHolder&) = delete; + LockHolder& operator=(const LockHolder&) = delete; + + ~LockHolder() { + ReleaseSRWLockExclusive(mu_); + } + + private: + SRWLOCK* mu_; +}; + +void Waiter::Init() { + InitializeSRWLock(&mu_); + InitializeConditionVariable(&cv_); + waiter_count_.store(0, std::memory_order_relaxed); + wakeup_count_.store(0, std::memory_order_relaxed); +} + +bool Waiter::Wait(KernelTimeout t) { + LockHolder h(&mu_); + waiter_count_.fetch_add(1, std::memory_order_relaxed); + + // Loop until we find a wakeup to consume or timeout. + while (true) { + int x = wakeup_count_.load(std::memory_order_relaxed); + if (x != 0) { + if (!wakeup_count_.compare_exchange_weak(x, x - 1, + std::memory_order_acquire, + std::memory_order_relaxed)) { + continue; // Raced with someone, retry. + } + // Successfully consumed a wakeup, we're done. + waiter_count_.fetch_sub(1, std::memory_order_relaxed); + return true; + } + + // No wakeups available, time to wait. + if (!SleepConditionVariableSRW( + &cv_, &mu_, t.InMillisecondsFromNow(), 0)) { + // GetLastError() returns a Win32 DWORD, but we assign to + // unsigned long to simplify the ABSL_RAW_LOG case below. The uniform + // initialization guarantees this is not a narrowing conversion. + const unsigned long err{GetLastError()}; // NOLINT(runtime/int) + if (err == ERROR_TIMEOUT) { + waiter_count_.fetch_sub(1, std::memory_order_relaxed); + return false; + } else { + ABSL_RAW_LOG(FATAL, "SleepConditionVariableSRW failed: %lu", err); + } + } + + MaybeBecomeIdle(); + } +} + +void Waiter::Post() { + wakeup_count_.fetch_add(1, std::memory_order_release); + Poke(); +} + +void Waiter::Poke() { + if (waiter_count_.load(std::memory_order_relaxed) == 0) { + return; + } + // Potentially a waker. Take the lock and check again. + LockHolder h(&mu_); + if (waiter_count_.load(std::memory_order_relaxed) == 0) { + return; + } + WakeConditionVariable(&cv_); +} + +#else +#error Unknown ABSL_WAITER_MODE +#endif + +} // namespace synchronization_internal +} // namespace absl |