Squashed 'third_party/marl/' changes from 16e1dc37c..539094011
539094011 CMake: Export MARL_THREAD_SAFETY_ANALYSIS_SUPPORTED
c7f70ba7a CMake: Bump min version + include(CheckCXXSourceCompiles)
658a204fc Replace SAL annotations with clang's TSA annotations
9630bec2f Fix CMake warning: "Policy CMP0023 is not set"
9f369ad5d Update yarn:: to marl:: in an example
git-subtree-dir: third_party/marl
git-subtree-split: 53909401165022553ed9d1f0c572178559dc25ec
diff --git a/.clang-format b/.clang-format
index b0823b4..08178fe 100644
--- a/.clang-format
+++ b/.clang-format
@@ -3,8 +3,3 @@
---
Language: Cpp
-StatementMacros:
- - _Acquires_lock_
- - _Releases_lock_
- - _Requires_lock_held_
- - _When_
\ No newline at end of file
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 2a8a695..005b855 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -12,12 +12,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-cmake_minimum_required(VERSION 2.8)
+cmake_minimum_required(VERSION 3.0)
-set (CMAKE_CXX_STANDARD 11)
+set(CMAKE_CXX_STANDARD 11)
project(Marl C CXX ASM)
+include(CheckCXXSourceCompiles)
+
###########################################################
# Options
###########################################################
@@ -76,6 +78,32 @@
endif(MARL_BUILD_BENCHMARKS)
###########################################################
+# Compiler feature tests
+###########################################################
+# Check that the Clang Thread Safety Analysis' try_acquire_capability behaves
+# correctly. This is broken on some earlier versions of clang.
+# See: https://bugs.llvm.org/show_bug.cgi?id=32954
+set(SAVE_CMAKE_REQUIRED_FLAGS ${CMAKE_REQUIRED_FLAGS})
+set(CMAKE_REQUIRED_FLAGS "-Wthread-safety -Werror")
+check_cxx_source_compiles(
+ "int main() {
+ struct __attribute__((capability(\"mutex\"))) Mutex {
+ void Unlock() __attribute__((release_capability)) {};
+ bool TryLock() __attribute__((try_acquire_capability(true))) { return true; };
+ };
+ Mutex m;
+ if (m.TryLock()) {
+ m.Unlock(); // Should not warn.
+ }
+ return 0;
+ }"
+ MARL_THREAD_SAFETY_ANALYSIS_SUPPORTED)
+set(CMAKE_REQUIRED_FLAGS ${SAVE_CMAKE_REQUIRED_FLAGS})
+
+# Export MARL_THREAD_SAFETY_ANALYSIS_SUPPORTED as this may be useful to parent projects
+set(MARL_THREAD_SAFETY_ANALYSIS_SUPPORTED PARENT_SCOPE ${MARL_THREAD_SAFETY_ANALYSIS_SUPPORTED})
+
+###########################################################
# File lists
###########################################################
set(MARL_LIST
@@ -111,6 +139,10 @@
# Functions
###########################################################
function(marl_set_target_options target)
+ if (MARL_THREAD_SAFETY_ANALYSIS_SUPPORTED)
+ target_compile_options(${target} PRIVATE "-Wthread-safety")
+ endif()
+
# Enable all warnings
if(MSVC)
target_compile_options(${target} PRIVATE
@@ -137,13 +169,13 @@
if(MARL_ASAN)
target_compile_options(${target} PUBLIC "-fsanitize=address")
- target_link_libraries(${target} "-fsanitize=address")
+ target_link_libraries(${target} PUBLIC "-fsanitize=address")
elseif(MARL_MSAN)
target_compile_options(${target} PUBLIC "-fsanitize=memory")
- target_link_libraries(${target} "-fsanitize=memory")
+ target_link_libraries(${target} PUBLIC "-fsanitize=memory")
elseif(MARL_TSAN)
target_compile_options(${target} PUBLIC "-fsanitize=thread")
- target_link_libraries(${target} "-fsanitize=thread")
+ target_link_libraries(${target} PUBLIC "-fsanitize=thread")
endif()
target_include_directories(${target} PUBLIC $<BUILD_INTERFACE:${MARL_INCLUDE_DIR}>)
@@ -232,7 +264,7 @@
marl_set_target_options(marl-unittests)
- target_link_libraries(marl-unittests marl)
+ target_link_libraries(marl-unittests PRIVATE marl)
endif(MARL_BUILD_TESTS)
# benchmarks
@@ -252,7 +284,7 @@
marl_set_target_options(marl-benchmarks)
- target_link_libraries(marl-benchmarks benchmark::benchmark marl)
+ target_link_libraries(marl-benchmarks PRIVATE benchmark::benchmark marl)
endif(MARL_BUILD_BENCHMARKS)
# examples
@@ -263,7 +295,7 @@
FOLDER "Examples"
)
marl_set_target_options(${target})
- target_link_libraries(${target} marl)
+ target_link_libraries(${target} PRIVATE marl)
endfunction(build_example)
build_example(fractal)
diff --git a/include/marl/blockingcall.h b/include/marl/blockingcall.h
index 983ee82..1ef0277 100644
--- a/include/marl/blockingcall.h
+++ b/include/marl/blockingcall.h
@@ -85,10 +85,10 @@
// void runABlockingFunctionOnATask()
// {
// // Schedule a task that calls a blocking, non-yielding function.
-// yarn::schedule([=] {
+// marl::schedule([=] {
// // call_blocking_function() may block indefinitely.
// // Ensure this call does not block other tasks from running.
-// auto result = yarn::blocking_call(call_blocking_function);
+// auto result = marl::blocking_call(call_blocking_function);
// // call_blocking_function() has now returned.
// // result holds the return value of the blocking function call.
// });
diff --git a/include/marl/conditionvariable.h b/include/marl/conditionvariable.h
index 8dfb9c7..5a57db2 100644
--- a/include/marl/conditionvariable.h
+++ b/include/marl/conditionvariable.h
@@ -19,11 +19,12 @@
#include "debug.h"
#include "defer.h"
#include "memory.h"
+#include "mutex.h"
#include "scheduler.h"
+#include "tsa.h"
#include <atomic>
#include <condition_variable>
-#include <mutex>
namespace marl {
@@ -47,14 +48,14 @@
// wait() blocks the current fiber or thread until the predicate is satisfied
// and the ConditionVariable is notified.
template <typename Predicate>
- inline void wait(std::unique_lock<std::mutex>& lock, Predicate&& pred);
+ inline void wait(marl::lock& lock, Predicate&& pred);
// wait_for() blocks the current fiber or thread until the predicate is
// satisfied, and the ConditionVariable is notified, or the timeout has been
// reached. Returns false if pred still evaluates to false after the timeout
// has been reached, otherwise true.
template <typename Rep, typename Period, typename Predicate>
- bool wait_for(std::unique_lock<std::mutex>& lock,
+ bool wait_for(marl::lock& lock,
const std::chrono::duration<Rep, Period>& duration,
Predicate&& pred);
@@ -63,7 +64,7 @@
// reached. Returns false if pred still evaluates to false after the timeout
// has been reached, otherwise true.
template <typename Clock, typename Duration, typename Predicate>
- bool wait_until(std::unique_lock<std::mutex>& lock,
+ bool wait_until(marl::lock& lock,
const std::chrono::time_point<Clock, Duration>& timeout,
Predicate&& pred);
@@ -73,7 +74,7 @@
ConditionVariable& operator=(const ConditionVariable&) = delete;
ConditionVariable& operator=(ConditionVariable&&) = delete;
- std::mutex mutex;
+ marl::mutex mutex;
containers::list<Scheduler::Fiber*> waiting;
std::condition_variable condition;
std::atomic<int> numWaiting = {0};
@@ -89,7 +90,7 @@
return;
}
{
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
for (auto fiber : waiting) {
fiber->notify();
}
@@ -104,7 +105,7 @@
return;
}
{
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
for (auto fiber : waiting) {
fiber->notify();
}
@@ -115,8 +116,7 @@
}
template <typename Predicate>
-void ConditionVariable::wait(std::unique_lock<std::mutex>& lock,
- Predicate&& pred) {
+void ConditionVariable::wait(marl::lock& lock, Predicate&& pred) {
if (pred()) {
return;
}
@@ -137,7 +137,7 @@
// Currently running outside of the scheduler.
// Delegate to the std::condition_variable.
numWaitingOnCondition++;
- condition.wait(lock, pred);
+ lock.wait(condition, pred);
numWaitingOnCondition--;
}
numWaiting--;
@@ -145,7 +145,7 @@
template <typename Rep, typename Period, typename Predicate>
bool ConditionVariable::wait_for(
- std::unique_lock<std::mutex>& lock,
+ marl::lock& lock,
const std::chrono::duration<Rep, Period>& duration,
Predicate&& pred) {
return wait_until(lock, std::chrono::system_clock::now() + duration, pred);
@@ -153,7 +153,7 @@
template <typename Clock, typename Duration, typename Predicate>
bool ConditionVariable::wait_until(
- std::unique_lock<std::mutex>& lock,
+ marl::lock& lock,
const std::chrono::time_point<Clock, Duration>& timeout,
Predicate&& pred) {
if (pred()) {
@@ -181,7 +181,7 @@
// Delegate to the std::condition_variable.
numWaitingOnCondition++;
defer(numWaitingOnCondition--);
- return condition.wait_until(lock, timeout, pred);
+ return lock.wait_until(condition, timeout, pred);
}
}
diff --git a/include/marl/event.h b/include/marl/event.h
index f7b3023..bac6078 100644
--- a/include/marl/event.h
+++ b/include/marl/event.h
@@ -113,7 +113,7 @@
inline bool wait_until(
const std::chrono::time_point<Clock, Duration>& timeout);
- std::mutex mutex;
+ marl::mutex mutex;
ConditionVariable cv;
const Mode mode;
bool signalled;
@@ -127,7 +127,7 @@
: cv(allocator), mode(mode), signalled(initialState) {}
void Event::Shared::signal() {
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
if (signalled) {
return;
}
@@ -143,7 +143,7 @@
}
void Event::Shared::wait() {
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
cv.wait(lock, [&] { return signalled; });
if (mode == Mode::Auto) {
signalled = false;
@@ -153,7 +153,7 @@
template <typename Rep, typename Period>
bool Event::Shared::wait_for(
const std::chrono::duration<Rep, Period>& duration) {
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
if (!cv.wait_for(lock, duration, [&] { return signalled; })) {
return false;
}
@@ -166,7 +166,7 @@
template <typename Clock, typename Duration>
bool Event::Shared::wait_until(
const std::chrono::time_point<Clock, Duration>& timeout) {
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
if (!cv.wait_until(lock, timeout, [&] { return signalled; })) {
return false;
}
@@ -186,7 +186,7 @@
}
void Event::clear() const {
- std::unique_lock<std::mutex> lock(shared->mutex);
+ marl::lock lock(shared->mutex);
shared->signalled = false;
}
@@ -206,7 +206,7 @@
}
bool Event::test() const {
- std::unique_lock<std::mutex> lock(shared->mutex);
+ marl::lock lock(shared->mutex);
if (!shared->signalled) {
return false;
}
@@ -217,7 +217,7 @@
}
bool Event::isSignalled() const {
- std::unique_lock<std::mutex> lock(shared->mutex);
+ marl::lock lock(shared->mutex);
return shared->signalled;
}
@@ -226,7 +226,7 @@
Event any(mode, false);
for (auto it = begin; it != end; it++) {
auto s = it->shared;
- std::unique_lock<std::mutex> lock(s->mutex);
+ marl::lock lock(s->mutex);
if (s->signalled) {
any.signal();
}
diff --git a/include/marl/mutex.h b/include/marl/mutex.h
new file mode 100644
index 0000000..72ecaf7
--- /dev/null
+++ b/include/marl/mutex.h
@@ -0,0 +1,106 @@
+// Copyright 2020 The Marl 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
+//
+// https://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.
+
+// Wrappers around std::mutex and std::unique_lock that provide clang's
+// Thread Safety Analysis annotations.
+// See: https://clang.llvm.org/docs/ThreadSafetyAnalysis.html
+
+#ifndef marl_mutex_h
+#define marl_mutex_h
+
+#include "tsa.h"
+
+#include <condition_variable>
+#include <mutex>
+
+namespace marl {
+
+// mutex is a wrapper around std::mutex that offers Thread Safety Analysis
+// annotations.
+// mutex also holds methods for performing std::condition_variable::wait() calls
+// as these require a std::unique_lock<> which are unsupported by the TSA.
+class CAPABILITY("mutex") mutex {
+ public:
+ inline void lock() ACQUIRE() { _.lock(); }
+
+ inline void unlock() RELEASE() { _.unlock(); }
+
+ inline bool try_lock() TRY_ACQUIRE(true) { return _.try_lock(); }
+
+ // wait_locked calls cv.wait() on this already locked mutex.
+ template <typename Predicate>
+ inline void wait_locked(std::condition_variable& cv, Predicate&& p)
+ REQUIRES(this) {
+ std::unique_lock<std::mutex> lock(_, std::adopt_lock);
+ cv.wait(lock, std::forward<Predicate>(p));
+ lock.release(); // Keep lock held.
+ }
+
+ // wait_until_locked calls cv.wait() on this already locked mutex.
+ template <typename Predicate, typename Time>
+ inline bool wait_until_locked(std::condition_variable& cv,
+ Time&& time,
+ Predicate&& p) REQUIRES(this) {
+ std::unique_lock<std::mutex> lock(_, std::adopt_lock);
+ auto res = cv.wait_until(lock, std::forward<Time>(time),
+ std::forward<Predicate>(p));
+ lock.release(); // Keep lock held.
+ return res;
+ }
+
+ private:
+ friend class lock;
+ std::mutex _;
+};
+
+// lock is a RAII lock helper that offers Thread Safety Analysis annotations.
+// lock also holds methods for performing std::condition_variable::wait()
+// calls as these require a std::unique_lock<> which are unsupported by the TSA.
+class SCOPED_CAPABILITY lock {
+ public:
+ inline lock(mutex& m) ACQUIRE(m) : _(m._) {}
+ inline ~lock() RELEASE() {}
+
+ // wait calls cv.wait() on this lock.
+ template <typename Predicate>
+ inline void wait(std::condition_variable& cv, Predicate&& p) {
+ cv.wait(_, std::forward<Predicate>(p));
+ }
+
+ // wait_until calls cv.wait() on this lock.
+ template <typename Predicate, typename Time>
+ inline bool wait_until(std::condition_variable& cv,
+ Time&& time,
+ Predicate&& p) {
+ return cv.wait_until(_, std::forward<Time>(time),
+ std::forward<Predicate>(p));
+ }
+
+ inline bool owns_lock() const { return _.owns_lock(); }
+
+ // lock_no_tsa locks the mutex outside of the visiblity of the thread
+ // safety analysis. Use with caution.
+ inline void lock_no_tsa() { _.lock(); }
+
+ // unlock_no_tsa unlocks the mutex outside of the visiblity of the thread
+ // safety analysis. Use with caution.
+ inline void unlock_no_tsa() { _.unlock(); }
+
+ private:
+ std::unique_lock<std::mutex> _;
+};
+
+} // namespace marl
+
+#endif // marl_mutex_h
diff --git a/include/marl/pool.h b/include/marl/pool.h
index 934fa40..393b457 100644
--- a/include/marl/pool.h
+++ b/include/marl/pool.h
@@ -17,9 +17,9 @@
#include "conditionvariable.h"
#include "memory.h"
+#include "mutex.h"
#include <atomic>
-#include <mutex>
namespace marl {
@@ -237,7 +237,7 @@
inline void return_(Item*) override;
Item items[N];
- std::mutex mutex;
+ marl::mutex mutex;
ConditionVariable returned;
Item* free = nullptr;
};
@@ -281,7 +281,7 @@
template <typename T, int N, PoolPolicy POLICY>
template <typename F>
void BoundedPool<T, N, POLICY>::borrow(size_t n, const F& f) const {
- std::unique_lock<std::mutex> lock(storage->mutex);
+ marl::lock lock(storage->mutex);
for (size_t i = 0; i < n; i++) {
storage->returned.wait(lock, [&] { return storage->free != nullptr; });
auto item = storage->free;
@@ -296,14 +296,16 @@
template <typename T, int N, PoolPolicy POLICY>
std::pair<typename BoundedPool<T, N, POLICY>::Loan, bool>
BoundedPool<T, N, POLICY>::tryBorrow() const {
- std::unique_lock<std::mutex> lock(storage->mutex);
- if (storage->free == nullptr) {
- return std::make_pair(Loan(), false);
+ Item* item = nullptr;
+ {
+ marl::lock lock(storage->mutex);
+ if (storage->free == nullptr) {
+ return std::make_pair(Loan(), false);
+ }
+ item = storage->free;
+ storage->free = storage->free->next;
+ item->pool = this;
}
- auto item = storage->free;
- storage->free = storage->free->next;
- item->pool = this;
- lock.unlock();
if (POLICY == PoolPolicy::Reconstruct) {
item->construct();
}
@@ -315,10 +317,11 @@
if (POLICY == PoolPolicy::Reconstruct) {
item->destruct();
}
- std::unique_lock<std::mutex> lock(mutex);
- item->next = free;
- free = item;
- lock.unlock();
+ {
+ marl::lock lock(mutex);
+ item->next = free;
+ free = item;
+ }
returned.notify_one();
}
@@ -359,7 +362,7 @@
inline void return_(Item*) override;
Allocator* allocator;
- std::mutex mutex;
+ marl::mutex mutex;
std::vector<Item*> items;
Item* free = nullptr;
};
@@ -398,7 +401,7 @@
template <typename T, PoolPolicy POLICY>
template <typename F>
inline void UnboundedPool<T, POLICY>::borrow(size_t n, const F& f) const {
- std::unique_lock<std::mutex> lock(storage->mutex);
+ marl::lock lock(storage->mutex);
for (size_t i = 0; i < n; i++) {
if (storage->free == nullptr) {
auto count = std::max<size_t>(storage->items.size(), 32);
@@ -427,10 +430,9 @@
if (POLICY == PoolPolicy::Reconstruct) {
item->destruct();
}
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
item->next = free;
free = item;
- lock.unlock();
}
} // namespace marl
diff --git a/include/marl/sal.h b/include/marl/sal.h
deleted file mode 100644
index de47a49..0000000
--- a/include/marl/sal.h
+++ /dev/null
@@ -1,42 +0,0 @@
-// Copyright 2019 The Marl 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
-//
-// https://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.
-
-// Stubs SAL annotation macros for platforms that do not support them.
-// See
-// https://docs.microsoft.com/en-us/visualstudio/code-quality/annotating-locking-behavior?view=vs-2019
-
-#ifndef marl_sal_h
-#define marl_sal_h
-
-#ifndef _Acquires_lock_
-#define _Acquires_lock_(...)
-#endif
-
-#ifndef _Guarded_by_
-#define _Guarded_by_(...)
-#endif
-
-#ifndef _Releases_lock_
-#define _Releases_lock_(...)
-#endif
-
-#ifndef _Requires_lock_held_
-#define _Requires_lock_held_(...)
-#endif
-
-#ifndef _When_
-#define _When_(...)
-#endif
-
-#endif // marl_sal_h
diff --git a/include/marl/scheduler.h b/include/marl/scheduler.h
index b96e147..4db4b4d 100644
--- a/include/marl/scheduler.h
+++ b/include/marl/scheduler.h
@@ -17,7 +17,7 @@
#include "debug.h"
#include "memory.h"
-#include "sal.h"
+#include "mutex.h"
#include "task.h"
#include "thread.h"
@@ -28,7 +28,6 @@
#include <deque>
#include <functional>
#include <map>
-#include <mutex>
#include <set>
#include <thread>
#include <unordered_map>
@@ -104,8 +103,6 @@
// thread that previously executed it.
class Fiber {
public:
- using Lock = std::unique_lock<std::mutex>;
-
// current() returns the currently executing fiber, or nullptr if called
// without a bound scheduler.
static Fiber* current();
@@ -122,8 +119,7 @@
// will be locked before wait() returns.
// pred will be always be called with the lock held.
// wait() must only be called on the currently executing fiber.
- _Requires_lock_held_(lock)
- void wait(Lock& lock, const Predicate& pred);
+ void wait(marl::lock& lock, const Predicate& pred);
// wait() suspends execution of this Fiber until the Fiber is woken up with
// a call to notify() and the predicate pred returns true, or sometime after
@@ -139,9 +135,8 @@
// will be locked before wait() returns.
// pred will be always be called with the lock held.
// wait() must only be called on the currently executing fiber.
- _Requires_lock_held_(lock)
template <typename Clock, typename Duration>
- inline bool wait(Lock& lock,
+ inline bool wait(marl::lock& lock,
const std::chrono::time_point<Clock, Duration>& timeout,
const Predicate& pred);
@@ -307,56 +302,51 @@
Worker(Scheduler* scheduler, Mode mode, uint32_t id);
// start() begins execution of the worker.
- void start();
+ void start() EXCLUDES(work.mutex);
// stop() ceases execution of the worker, blocking until all pending
// tasks have fully finished.
- void stop();
+ void stop() EXCLUDES(work.mutex);
// wait() suspends execution of the current task until the predicate pred
// returns true or the optional timeout is reached.
// See Fiber::wait() for more information.
- _Requires_lock_held_(lock)
- bool wait(Fiber::Lock& lock,
- const TimePoint* timeout,
- const Predicate& pred);
+ bool wait(marl::lock& lock, const TimePoint* timeout, const Predicate& pred)
+ EXCLUDES(work.mutex);
// wait() suspends execution of the current task until the fiber is
// notified, or the optional timeout is reached.
// See Fiber::wait() for more information.
- bool wait(const TimePoint* timeout);
+ bool wait(const TimePoint* timeout) EXCLUDES(work.mutex);
// suspend() suspends the currenetly executing Fiber until the fiber is
// woken with a call to enqueue(Fiber*), or automatically sometime after the
// optional timeout.
- _Requires_lock_held_(work.mutex)
- void suspend(const TimePoint* timeout);
+ void suspend(const TimePoint* timeout) REQUIRES(work.mutex);
// enqueue(Fiber*) enqueues resuming of a suspended fiber.
- void enqueue(Fiber* fiber);
+ void enqueue(Fiber* fiber) EXCLUDES(work.mutex);
// enqueue(Task&&) enqueues a new, unstarted task.
- void enqueue(Task&& task);
+ void enqueue(Task&& task) EXCLUDES(work.mutex);
// tryLock() attempts to lock the worker for task enqueing.
// If the lock was successful then true is returned, and the caller must
// call enqueueAndUnlock().
- _When_(return == true, _Acquires_lock_(work.mutex))
- bool tryLock();
+ bool tryLock() EXCLUDES(work.mutex) TRY_ACQUIRE(true, work.mutex);
// enqueueAndUnlock() enqueues the task and unlocks the worker.
// Must only be called after a call to tryLock() which returned true.
- _Requires_lock_held_(work.mutex)
- _Releases_lock_(work.mutex)
- void enqueueAndUnlock(Task&& task);
+ // _Releases_lock_(work.mutex)
+ void enqueueAndUnlock(Task&& task) REQUIRES(work.mutex) RELEASE(work.mutex);
// runUntilShutdown() processes all tasks and fibers until there are no more
// and shutdown is true, upon runUntilShutdown() returns.
- void runUntilShutdown();
+ void runUntilShutdown() REQUIRES(work.mutex);
// steal() attempts to steal a Task from the worker for another worker.
// Returns true if a task was taken and assigned to out, otherwise false.
- bool steal(Task& out);
+ bool steal(Task& out) EXCLUDES(work.mutex);
// getCurrent() returns the Worker currently bound to the current
// thread.
@@ -371,27 +361,22 @@
private:
// run() is the task processing function for the worker.
// run() processes tasks until stop() is called.
- _Requires_lock_held_(work.mutex)
- void run();
+ void run() REQUIRES(work.mutex);
// createWorkerFiber() creates a new fiber that when executed calls
// run().
- _Requires_lock_held_(work.mutex)
- Fiber* createWorkerFiber();
+ Fiber* createWorkerFiber() REQUIRES(work.mutex);
// switchToFiber() switches execution to the given fiber. The fiber
// must belong to this worker.
- _Requires_lock_held_(work.mutex)
- void switchToFiber(Fiber*);
+ void switchToFiber(Fiber*) REQUIRES(work.mutex);
// runUntilIdle() executes all pending tasks and then returns.
- _Requires_lock_held_(work.mutex)
- void runUntilIdle();
+ void runUntilIdle() REQUIRES(work.mutex);
// waitForWork() blocks until new work is available, potentially calling
// spinForWork().
- _Requires_lock_held_(work.mutex)
- void waitForWork();
+ void waitForWork() REQUIRES(work.mutex);
// spinForWork() attempts to steal work from another Worker, and keeps
// the thread awake for a short duration. This reduces overheads of
@@ -400,31 +385,28 @@
// enqueueFiberTimeouts() enqueues all the fibers that have finished
// waiting.
- _Requires_lock_held_(work.mutex)
- void enqueueFiberTimeouts();
+ void enqueueFiberTimeouts() REQUIRES(work.mutex);
- _Requires_lock_held_(work.mutex)
inline void changeFiberState(Fiber* fiber,
Fiber::State from,
- Fiber::State to) const;
+ Fiber::State to) const REQUIRES(work.mutex);
- _Requires_lock_held_(work.mutex)
- inline void setFiberState(Fiber* fiber, Fiber::State to) const;
+ inline void setFiberState(Fiber* fiber, Fiber::State to) const
+ REQUIRES(work.mutex);
// Work holds tasks and fibers that are enqueued on the Worker.
struct Work {
std::atomic<uint64_t> num = {0}; // tasks.size() + fibers.size()
- _Guarded_by_(mutex) uint64_t numBlockedFibers = 0;
- _Guarded_by_(mutex) TaskQueue tasks;
- _Guarded_by_(mutex) FiberQueue fibers;
- _Guarded_by_(mutex) WaitingFibers waiting;
- _Guarded_by_(mutex) bool notifyAdded = true;
+ GUARDED_BY(mutex) uint64_t numBlockedFibers = 0;
+ GUARDED_BY(mutex) TaskQueue tasks;
+ GUARDED_BY(mutex) FiberQueue fibers;
+ GUARDED_BY(mutex) WaitingFibers waiting;
+ GUARDED_BY(mutex) bool notifyAdded = true;
std::condition_variable added;
- std::mutex mutex;
+ marl::mutex mutex;
- _Requires_lock_held_(mutex)
template <typename F>
- inline void wait(F&&);
+ inline void wait(F&&) REQUIRES(mutex);
};
// https://en.wikipedia.org/wiki/Xorshift
@@ -472,7 +454,7 @@
Allocator* const allocator;
std::function<void()> threadInitFunc;
- std::mutex threadInitFuncMutex;
+ mutex threadInitFuncMutex;
std::array<std::atomic<int>, 8> spinningWorkers;
std::atomic<unsigned int> nextSpinningWorkerIdx = {0x8000000};
@@ -484,17 +466,18 @@
std::array<Worker*, MaxWorkerThreads> workerThreads;
struct SingleThreadedWorkers {
- std::mutex mutex;
- std::condition_variable unbind;
- std::unordered_map<std::thread::id, Allocator::unique_ptr<Worker>> byTid;
+ using WorkerByTid =
+ std::unordered_map<std::thread::id, Allocator::unique_ptr<Worker>>;
+ marl::mutex mutex;
+ GUARDED_BY(mutex) std::condition_variable unbind;
+ GUARDED_BY(mutex) WorkerByTid byTid;
};
SingleThreadedWorkers singleThreadedWorkers;
};
-_Requires_lock_held_(lock)
template <typename Clock, typename Duration>
bool Scheduler::Fiber::wait(
- Lock& lock,
+ marl::lock& lock,
const std::chrono::time_point<Clock, Duration>& timeout,
const Predicate& pred) {
using ToDuration = typename TimePoint::duration;
diff --git a/include/marl/ticket.h b/include/marl/ticket.h
index 59a38fc..6aa21ed 100644
--- a/include/marl/ticket.h
+++ b/include/marl/ticket.h
@@ -105,14 +105,14 @@
inline ~Record();
inline void done();
- inline void callAndUnlock(std::unique_lock<std::mutex>& lock);
+ inline void callAndUnlock(marl::lock& lock);
+ inline void unlink(); // guarded by shared->mutex
ConditionVariable isCalledCondVar;
std::shared_ptr<Shared> shared;
Record* next = nullptr; // guarded by shared->mutex
Record* prev = nullptr; // guarded by shared->mutex
- inline void unlink(); // guarded by shared->mutex
OnCall onCall; // guarded by shared->mutex
bool isCalled = false; // guarded by shared->mutex
std::atomic<bool> isDone = {false};
@@ -120,7 +120,7 @@
// Data shared between all tickets and the queue.
struct Shared {
- std::mutex mutex;
+ marl::mutex mutex;
Record tail;
};
@@ -136,7 +136,7 @@
Ticket::Ticket(Loan<Record>&& record) : record(std::move(record)) {}
void Ticket::wait() const {
- std::unique_lock<std::mutex> lock(record->shared->mutex);
+ marl::lock lock(record->shared->mutex);
record->isCalledCondVar.wait(lock, [this] { return record->isCalled; });
}
@@ -146,7 +146,7 @@
template <typename Function>
void Ticket::onCall(Function&& f) const {
- std::unique_lock<std::mutex> lock(record->shared->mutex);
+ marl::lock lock(record->shared->mutex);
if (record->isCalled) {
marl::schedule(std::move(f));
return;
@@ -192,7 +192,7 @@
f(std::move(Ticket(std::move(rec))));
});
last->next = &shared->tail;
- std::unique_lock<std::mutex> lock(shared->mutex);
+ marl::lock lock(shared->mutex);
first->prev = shared->tail.prev;
shared->tail.prev = last.get();
if (first->prev == nullptr) {
@@ -216,7 +216,7 @@
if (isDone.exchange(true)) {
return;
}
- std::unique_lock<std::mutex> lock(shared->mutex);
+ marl::lock lock(shared->mutex);
auto callNext = (prev == nullptr && next != nullptr) ? next : nullptr;
unlink();
if (callNext != nullptr) {
@@ -225,7 +225,7 @@
}
}
-void Ticket::Record::callAndUnlock(std::unique_lock<std::mutex>& lock) {
+void Ticket::Record::callAndUnlock(marl::lock& lock) {
if (isCalled) {
return;
}
@@ -233,7 +233,7 @@
OnCall callback;
std::swap(callback, onCall);
isCalledCondVar.notify_all();
- lock.unlock();
+ lock.unlock_no_tsa();
if (callback) {
marl::schedule(std::move(callback));
diff --git a/include/marl/tsa.h b/include/marl/tsa.h
new file mode 100644
index 0000000..84d0623
--- /dev/null
+++ b/include/marl/tsa.h
@@ -0,0 +1,80 @@
+// Copyright 2020 The Marl 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
+//
+// https://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.
+
+// Stubs Thread-Safty-Analysis annotation macros for platforms that do not
+// support them.
+// See https://clang.llvm.org/docs/ThreadSafetyAnalysis.html
+
+#ifndef marl_tsa_h
+#define marl_tsa_h
+
+// Enable thread safety attributes only with clang.
+// The attributes can be safely erased when compiling with other compilers.
+#if defined(__clang__) && (!defined(SWIG))
+#define THREAD_ANNOTATION_ATTRIBUTE__(x) __attribute__((x))
+#else
+#define THREAD_ANNOTATION_ATTRIBUTE__(x) // no-op
+#endif
+
+#define CAPABILITY(x) THREAD_ANNOTATION_ATTRIBUTE__(capability(x))
+
+#define SCOPED_CAPABILITY THREAD_ANNOTATION_ATTRIBUTE__(scoped_lockable)
+
+#define GUARDED_BY(x) THREAD_ANNOTATION_ATTRIBUTE__(guarded_by(x))
+
+#define PT_GUARDED_BY(x) THREAD_ANNOTATION_ATTRIBUTE__(pt_guarded_by(x))
+
+#define ACQUIRED_BEFORE(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(acquired_before(__VA_ARGS__))
+
+#define ACQUIRED_AFTER(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(acquired_after(__VA_ARGS__))
+
+#define REQUIRES(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(requires_capability(__VA_ARGS__))
+
+#define REQUIRES_SHARED(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(requires_shared_capability(__VA_ARGS__))
+
+#define ACQUIRE(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(acquire_capability(__VA_ARGS__))
+
+#define ACQUIRE_SHARED(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(acquire_shared_capability(__VA_ARGS__))
+
+#define RELEASE(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(release_capability(__VA_ARGS__))
+
+#define RELEASE_SHARED(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(release_shared_capability(__VA_ARGS__))
+
+#define TRY_ACQUIRE(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(try_acquire_capability(__VA_ARGS__))
+
+#define TRY_ACQUIRE_SHARED(...) \
+ THREAD_ANNOTATION_ATTRIBUTE__(try_acquire_shared_capability(__VA_ARGS__))
+
+#define EXCLUDES(...) THREAD_ANNOTATION_ATTRIBUTE__(locks_excluded(__VA_ARGS__))
+
+#define ASSERT_CAPABILITY(x) THREAD_ANNOTATION_ATTRIBUTE__(assert_capability(x))
+
+#define ASSERT_SHARED_CAPABILITY(x) \
+ THREAD_ANNOTATION_ATTRIBUTE__(assert_shared_capability(x))
+
+#define RETURN_CAPABILITY(x) THREAD_ANNOTATION_ATTRIBUTE__(lock_returned(x))
+
+#define NO_THREAD_SAFETY_ANALYSIS \
+ THREAD_ANNOTATION_ATTRIBUTE__(no_thread_safety_analysis)
+
+#endif // marl_tsa_h
diff --git a/include/marl/waitgroup.h b/include/marl/waitgroup.h
index 7adbb19..a53a446 100644
--- a/include/marl/waitgroup.h
+++ b/include/marl/waitgroup.h
@@ -70,7 +70,7 @@
std::atomic<unsigned int> count = {0};
ConditionVariable cv;
- std::mutex mutex;
+ marl::mutex mutex;
};
const std::shared_ptr<Data> data;
};
@@ -91,7 +91,7 @@
MARL_ASSERT(data->count > 0, "marl::WaitGroup::done() called too many times");
auto count = --data->count;
if (count == 0) {
- std::unique_lock<std::mutex> lock(data->mutex);
+ marl::lock lock(data->mutex);
data->cv.notify_all();
return true;
}
@@ -99,7 +99,7 @@
}
void WaitGroup::wait() const {
- std::unique_lock<std::mutex> lock(data->mutex);
+ marl::lock lock(data->mutex);
data->cv.wait(lock, [this] { return data->count == 0; });
}
diff --git a/src/conditionvariable_test.cpp b/src/conditionvariable_test.cpp
index 2930e4d..18ece7c 100644
--- a/src/conditionvariable_test.cpp
+++ b/src/conditionvariable_test.cpp
@@ -18,17 +18,16 @@
#include "marl_test.h"
#include <condition_variable>
-#include <mutex>
TEST_F(WithoutBoundScheduler, ConditionVariable) {
bool trigger[3] = {false, false, false};
bool signal[3] = {false, false, false};
- std::mutex mutex;
+ marl::mutex mutex;
marl::ConditionVariable cv;
std::thread thread([&] {
for (int i = 0; i < 3; i++) {
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
cv.wait(lock, [&] {
EXPECT_TRUE(lock.owns_lock());
return trigger[i];
@@ -45,7 +44,7 @@
for (int i = 0; i < 3; i++) {
{
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
trigger[i] = true;
cv.notify_one();
cv.wait(lock, [&] {
@@ -66,12 +65,12 @@
TEST_P(WithBoundScheduler, ConditionVariable) {
bool trigger[3] = {false, false, false};
bool signal[3] = {false, false, false};
- std::mutex mutex;
+ marl::mutex mutex;
marl::ConditionVariable cv;
std::thread thread([&] {
for (int i = 0; i < 3; i++) {
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
cv.wait(lock, [&] {
EXPECT_TRUE(lock.owns_lock());
return trigger[i];
@@ -88,7 +87,7 @@
for (int i = 0; i < 3; i++) {
{
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
trigger[i] = true;
cv.notify_one();
cv.wait(lock, [&] {
@@ -113,14 +112,14 @@
// they are early-unblocked, along with expected lock state.
TEST_P(WithBoundScheduler, ConditionVariableTimeouts) {
for (int i = 0; i < 10; i++) {
- std::mutex mutex;
+ marl::mutex mutex;
marl::ConditionVariable cv;
bool signaled = false; // guarded by mutex
auto wg = marl::WaitGroup(100);
for (int j = 0; j < 100; j++) {
marl::schedule([=, &mutex, &cv, &signaled] {
{
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
cv.wait_for(lock, std::chrono::milliseconds(j), [&] {
EXPECT_TRUE(lock.owns_lock());
return signaled;
@@ -134,7 +133,7 @@
}
std::this_thread::sleep_for(std::chrono::milliseconds(50));
{
- std::unique_lock<std::mutex> lock(mutex);
+ marl::lock lock(mutex);
signaled = true;
cv.notify_all();
}
diff --git a/src/scheduler.cpp b/src/scheduler.cpp
index c97d9ec..533892a 100644
--- a/src/scheduler.cpp
+++ b/src/scheduler.cpp
@@ -99,7 +99,7 @@
MARL_ASSERT(bound == nullptr, "Scheduler already bound");
bound = this;
{
- std::unique_lock<std::mutex> lock(singleThreadedWorkers.mutex);
+ marl::lock lock(singleThreadedWorkers.mutex);
auto worker =
allocator->make_unique<Worker>(this, Worker::Mode::SingleThreaded, -1);
worker->start();
@@ -113,7 +113,7 @@
auto worker = Scheduler::Worker::getCurrent();
worker->stop();
{
- std::unique_lock<std::mutex> lock(bound->singleThreadedWorkers.mutex);
+ marl::lock lock(bound->singleThreadedWorkers.mutex);
auto tid = std::this_thread::get_id();
auto it = bound->singleThreadedWorkers.byTid.find(tid);
MARL_ASSERT(it != bound->singleThreadedWorkers.byTid.end(),
@@ -137,9 +137,11 @@
Scheduler::~Scheduler() {
{
// Wait until all the single threaded workers have been unbound.
- std::unique_lock<std::mutex> lock(singleThreadedWorkers.mutex);
- singleThreadedWorkers.unbind.wait(
- lock, [this] { return singleThreadedWorkers.byTid.size() == 0; });
+ marl::lock lock(singleThreadedWorkers.mutex);
+ lock.wait(singleThreadedWorkers.unbind,
+ [this]() REQUIRES(singleThreadedWorkers.mutex) {
+ return singleThreadedWorkers.byTid.size() == 0;
+ });
}
// Release all worker threads.
@@ -148,12 +150,12 @@
}
void Scheduler::setThreadInitializer(const std::function<void()>& func) {
- std::unique_lock<std::mutex> lock(threadInitFuncMutex);
+ marl::lock lock(threadInitFuncMutex);
threadInitFunc = func;
}
const std::function<void()>& Scheduler::getThreadInitializer() {
- std::unique_lock<std::mutex> lock(threadInitFuncMutex);
+ marl::lock lock(threadInitFuncMutex);
return threadInitFunc;
}
@@ -251,7 +253,7 @@
worker->enqueue(this);
}
-void Scheduler::Fiber::wait(Lock& lock, const Predicate& pred) {
+void Scheduler::Fiber::wait(marl::lock& lock, const Predicate& pred) {
worker->wait(lock, nullptr, pred);
}
@@ -374,7 +376,7 @@
mainFiber = Fiber::createFromCurrentThread(scheduler->allocator, 0);
currentFiber = mainFiber.get();
{
- std::unique_lock<std::mutex> lock(work.mutex);
+ marl::lock lock(work.mutex);
run();
}
mainFiber.reset();
@@ -401,7 +403,7 @@
break;
}
case Mode::SingleThreaded: {
- std::unique_lock<std::mutex> lock(work.mutex);
+ marl::lock lock(work.mutex);
shutdown = true;
runUntilShutdown();
Worker::current = nullptr;
@@ -415,14 +417,13 @@
bool Scheduler::Worker::wait(const TimePoint* timeout) {
DBG_LOG("%d: WAIT(%d)", (int)id, (int)currentFiber->id);
{
- std::unique_lock<std::mutex> lock(work.mutex);
+ marl::lock lock(work.mutex);
suspend(timeout);
}
return timeout == nullptr || std::chrono::system_clock::now() < *timeout;
}
-_Requires_lock_held_(waitLock)
-bool Scheduler::Worker::wait(Fiber::Lock& waitLock,
+bool Scheduler::Worker::wait(lock& waitLock,
const TimePoint* timeout,
const Predicate& pred) {
DBG_LOG("%d: WAIT(%d)", (int)id, (int)currentFiber->id);
@@ -435,7 +436,7 @@
// enqueued (via Fiber::notify()) between the waitLock.unlock() and fiber
// switch, otherwise the Fiber::notify() call may be ignored and the fiber
// is never woken.
- waitLock.unlock();
+ waitLock.unlock_no_tsa();
// suspend the fiber.
suspend(timeout);
@@ -444,7 +445,7 @@
work.mutex.unlock();
// Re-lock to either return due to timeout, or call pred().
- waitLock.lock();
+ waitLock.lock_no_tsa();
// Check timeout.
if (timeout != nullptr && std::chrono::system_clock::now() >= *timeout) {
@@ -456,7 +457,6 @@
return true;
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::suspend(
const std::chrono::system_clock::time_point* timeout) {
// Current fiber is yielding as it is blocked.
@@ -496,33 +496,34 @@
setFiberState(currentFiber, Fiber::State::Running);
}
-_When_(return == true, _Acquires_lock_(work.mutex))
bool Scheduler::Worker::tryLock() {
return work.mutex.try_lock();
}
void Scheduler::Worker::enqueue(Fiber* fiber) {
- std::unique_lock<std::mutex> lock(work.mutex);
- DBG_LOG("%d: ENQUEUE(%d %s)", (int)id, (int)fiber->id,
- Fiber::toString(fiber->state));
- switch (fiber->state) {
- case Fiber::State::Running:
- case Fiber::State::Queued:
- return; // Nothing to do here - task is already queued or running.
- case Fiber::State::Waiting:
- work.waiting.erase(fiber);
- break;
- case Fiber::State::Idle:
- case Fiber::State::Yielded:
- break;
+ bool notify = false;
+ {
+ marl::lock lock(work.mutex);
+ DBG_LOG("%d: ENQUEUE(%d %s)", (int)id, (int)fiber->id,
+ Fiber::toString(fiber->state));
+ switch (fiber->state) {
+ case Fiber::State::Running:
+ case Fiber::State::Queued:
+ return; // Nothing to do here - task is already queued or running.
+ case Fiber::State::Waiting:
+ work.waiting.erase(fiber);
+ break;
+ case Fiber::State::Idle:
+ case Fiber::State::Yielded:
+ break;
+ }
+ notify = work.notifyAdded;
+ work.fibers.push_back(std::move(fiber));
+ MARL_ASSERT(!work.waiting.contains(fiber),
+ "fiber is unexpectedly in the waiting list");
+ setFiberState(fiber, Fiber::State::Queued);
+ work.num++;
}
- bool notify = work.notifyAdded;
- work.fibers.push_back(std::move(fiber));
- MARL_ASSERT(!work.waiting.contains(fiber),
- "fiber is unexpectedly in the waiting list");
- setFiberState(fiber, Fiber::State::Queued);
- work.num++;
- lock.unlock();
if (notify) {
work.added.notify_one();
@@ -534,8 +535,6 @@
enqueueAndUnlock(std::move(task));
}
-_Requires_lock_held_(work.mutex)
-_Releases_lock_(work.mutex)
void Scheduler::Worker::enqueueAndUnlock(Task&& task) {
auto notify = work.notifyAdded;
work.tasks.push_back(std::move(task));
@@ -564,21 +563,21 @@
return true;
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::run() {
if (mode == Mode::MultiThreaded) {
MARL_NAME_THREAD("Thread<%.2d> Fiber<%.2d>", int(id), Fiber::current()->id);
// This is the entry point for a multi-threaded worker.
// Start with a regular condition-variable wait for work. This avoids
// starting the thread with a spinForWork().
- work.wait([this] { return work.num > 0 || work.waiting || shutdown; });
+ work.wait([this]() REQUIRES(work.mutex) {
+ return work.num > 0 || work.waiting || shutdown;
+ });
}
ASSERT_FIBER_STATE(currentFiber, Fiber::State::Running);
runUntilShutdown();
switchToFiber(mainFiber.get());
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::runUntilShutdown() {
while (!shutdown || work.num > 0 || work.numBlockedFibers > 0U) {
waitForWork();
@@ -586,7 +585,6 @@
}
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::waitForWork() {
MARL_ASSERT(work.num == work.fibers.size() + work.tasks.size(),
"work.num out of sync");
@@ -601,7 +599,7 @@
work.mutex.lock();
}
- work.wait([this] {
+ work.wait([this]() REQUIRES(work.mutex) {
return work.num > 0 || (shutdown && work.numBlockedFibers == 0U);
});
if (work.waiting) {
@@ -609,7 +607,6 @@
}
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::enqueueFiberTimeouts() {
auto now = std::chrono::system_clock::now();
while (auto fiber = work.waiting.take(now)) {
@@ -620,7 +617,6 @@
}
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::changeFiberState(Fiber* fiber,
Fiber::State from,
Fiber::State to) const {
@@ -631,7 +627,6 @@
fiber->state = to;
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::setFiberState(Fiber* fiber, Fiber::State to) const {
DBG_LOG("%d: SET_FIBER_STATE(%d %s -> %s)", (int)id, (int)fiber->id,
Fiber::toString(fiber->state), Fiber::toString(to));
@@ -659,7 +654,7 @@
}
if (scheduler->stealWork(this, rng(), stolen)) {
- std::unique_lock<std::mutex> lock(work.mutex);
+ marl::lock lock(work.mutex);
work.tasks.emplace_back(std::move(stolen));
work.num++;
return;
@@ -669,7 +664,6 @@
}
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::runUntilIdle() {
ASSERT_FIBER_STATE(currentFiber, Fiber::State::Running);
MARL_ASSERT(work.num == work.fibers.size() + work.tasks.size(),
@@ -713,18 +707,16 @@
}
}
-_Requires_lock_held_(work.mutex)
Scheduler::Fiber* Scheduler::Worker::createWorkerFiber() {
auto fiberId = static_cast<uint32_t>(workerFibers.size() + 1);
DBG_LOG("%d: CREATE(%d)", (int)id, (int)fiberId);
auto fiber = Fiber::create(scheduler->allocator, fiberId, FiberStackSize,
- [&] { run(); });
+ [&]() REQUIRES(work.mutex) { run(); });
auto ptr = fiber.get();
workerFibers.push_back(std::move(fiber));
return ptr;
}
-_Requires_lock_held_(work.mutex)
void Scheduler::Worker::switchToFiber(Fiber* to) {
DBG_LOG("%d: SWITCH(%d -> %d)", (int)id, (int)currentFiber->id, (int)to->id);
MARL_ASSERT(to == mainFiber.get() || idleFibers.count(to) == 0,
@@ -737,18 +729,15 @@
////////////////////////////////////////////////////////////////////////////////
// Scheduler::Worker::Work
////////////////////////////////////////////////////////////////////////////////
-_Requires_lock_held_(mutex)
template <typename F>
void Scheduler::Worker::Work::wait(F&& f) {
- std::unique_lock<std::mutex> lock(mutex, std::adopt_lock);
notifyAdded = true;
if (waiting) {
- added.wait_until(lock, waiting.next(), f);
+ mutex.wait_until_locked(added, waiting.next(), f);
} else {
- added.wait(lock, f);
+ mutex.wait_locked(added, f);
}
notifyAdded = false;
- lock.release(); // Keep the lock held.
}
} // namespace marl
