Squashed 'third_party/marl/' changes from 539094011..748d3c161 748d3c161 Add usage recommendations to README.md 2939fcfed Replace marl::parallelize() 4b4600208 Fix stupid typo in front page code samples. 742bba9c8 Fix clang-tidy and go-vet warnings. git-subtree-dir: third_party/marl git-subtree-split: 748d3c16165c7dcff018273cfaf9d8f26a33333a
diff --git a/CMakeLists.txt b/CMakeLists.txt index 005b855..4707a86 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt
@@ -242,6 +242,7 @@ ${MARL_SRC_DIR}/marl_test.h ${MARL_SRC_DIR}/memory_test.cpp ${MARL_SRC_DIR}/osfiber_test.cpp + ${MARL_SRC_DIR}/parallelize_test.cpp ${MARL_SRC_DIR}/pool_test.cpp ${MARL_SRC_DIR}/scheduler_test.cpp ${MARL_SRC_DIR}/ticket_test.cpp
diff --git a/README.md b/README.md index cea4522..06408dc 100644 --- a/README.md +++ b/README.md
@@ -33,31 +33,31 @@ constexpr int numTasks = 10; // Create an event that is manually reset. - marl::Event sayHellow(marl::Event::Mode::Manual); + marl::Event sayHello(marl::Event::Mode::Manual); // Create a WaitGroup with an initial count of numTasks. - marl::WaitGroup saidHellow(numTasks); + marl::WaitGroup saidHello(numTasks); // Schedule some tasks to run asynchronously. for (int i = 0; i < numTasks; i++) { // Each task will run on one of the 4 worker threads. marl::schedule([=] { // All marl primitives are capture-by-value. // Decrement the WaitGroup counter when the task has finished. - defer(saidHellow.done()); + defer(saidHello.done()); printf("Task %d waiting to say hello...\n", i); // Blocking in a task? // The scheduler will find something else for this thread to do. - sayHellow.wait(); + sayHello.wait(); printf("Hello from task %d!\n", i); }); } - sayHellow.signal(); // Unblock all the tasks. + sayHello.signal(); // Unblock all the tasks. - saidHellow.wait(); // Wait for all tasks to complete. + saidHello.wait(); // Wait for all tasks to complete. printf("All tasks said hello.\n"); @@ -123,6 +123,81 @@ add_subdirectory(${MARL_DIR}) ``` +### Usage Recommendations + +#### Capture marl synchronization primitves by value + +All marl synchronization primitves aside from `marl::ConditionVariable` should be lambda-captured by **value**: + +```c++ +marl::Event event; +marl::schedule([=]{ // [=] Good, [&] Bad. + event.signal(); +}) +``` + +Internally, these primitives hold a shared pointer to the primitive state. By capturing by value we avoid common issues where the primitive may be destructed before the last reference is used. + +#### Create one instance of `marl::Scheduler`, use it for the lifetime of the process. + +`marl::Scheduler::setWorkerThreadCount()` is an expensive operation as it spawn a number of hardware threads. \ +Destructing the `marl::Scheduler` requires waiting on all tasks to complete. + +Multiple `marl::Scheduler`s may fight each other for hardware thread utilization. + +For these reasons, it is recommended to create a single `marl::Scheduler` for the lifetime of your process. + +For example: + +```c++ +int main() { + marl::Scheduler scheduler; + scheduler.bind(); + scheduler.setWorkerThreadCount(marl::Thread::numLogicalCPUs()); + defer(scheduler.unbind()); + + return do_program_stuff(); +} +``` + +#### Bind the scheduler to externally created threads + +In order to call `marl::schedule()` the scheduler must be bound to the calling thread. Failure to bind the scheduler to the thread before calling `marl::schedule()` will result in undefined behavior. + +`marl::Scheduler` may be simultaneously bound to any number of threads, and the scheduler can be retrieved from a bound thread with `marl::Scheduler::get()`. + +A typical way to pass the scheduler from one thread to another would be: + +```c++ +std::thread spawn_new_thread() { + // Grab the scheduler from the currently running thread. + marl::Scheduler* scheduler = marl::Scheduler::get(); + + // Spawn the new thread. + return std::thread([=] { + // Bind the scheduler to the new thread. + scheduler->bind(); + defer(scheduler->unbind()); + + // You can now safely call `marl::schedule()` + run_thread_logic(); + }); +} + +``` + +Always remember to unbind the scheduler before terminating the thread. Forgetting to unbind will result in the `marl::Scheduler` destructor blocking indefinitely. + +#### Don't use externally blocking calls in marl tasks + +The `marl::Scheduler` internally holds a number of worker threads which will execute the scheduled tasks. If a marl task becomes blocked on a marl synchronization primitive, marl can yield from the blocked task and continue execution of other scheduled tasks. + +Calling a non-marl blocking function on a marl worker thread will prevent that worker thread from being able to switch to execute other tasks until the blocking function has returned. Examples of these non-marl blocking functions include: [`std::mutex::lock()`](https://en.cppreference.com/w/cpp/thread/mutex/lock), [`std::condition_variable::wait()`](https://en.cppreference.com/w/cpp/thread/condition_variable/wait), [`accept()`](http://man7.org/linux/man-pages/man2/accept.2.html). + +Short blocking calls are acceptable, such as a mutex lock to access a data structure. However be careful that you do not use a marl blocking call with a `std::mutex` lock held - the marl task may yield with the lock held, and block other tasks from re-locking the mutex. This sort of situation may end up with a deadlock. + +If you need to make a blocking call from a marl worker thread, you may wish to use [`marl::blocking_call()`](https://github.com/google/marl/blob/master/include/marl/blockingcall.h), which will spawn a new thread for performing the call, allowing the marl worker to continue processing other scheduled tasks. + --- Note: This is not an officially supported Google product
diff --git a/examples/hello_task.cpp b/examples/hello_task.cpp index 6dfff3c..3206c3c 100644 --- a/examples/hello_task.cpp +++ b/examples/hello_task.cpp
@@ -32,31 +32,31 @@ constexpr int numTasks = 10; // Create an event that is manually reset. - marl::Event sayHellow(marl::Event::Mode::Manual); + marl::Event sayHello(marl::Event::Mode::Manual); // Create a WaitGroup with an initial count of numTasks. - marl::WaitGroup saidHellow(numTasks); + marl::WaitGroup saidHello(numTasks); // Schedule some tasks to run asynchronously. for (int i = 0; i < numTasks; i++) { // Each task will run on one of the 4 worker threads. marl::schedule([=] { // All marl primitives are capture-by-value. // Decrement the WaitGroup counter when the task has finished. - defer(saidHellow.done()); + defer(saidHello.done()); printf("Task %d waiting to say hello...\n", i); // Blocking in a task? // The scheduler will find something else for this thread to do. - sayHellow.wait(); + sayHello.wait(); printf("Hello from task %d!\n", i); }); } - sayHellow.signal(); // Unblock all the tasks. + sayHello.signal(); // Unblock all the tasks. - saidHellow.wait(); // Wait for all tasks to complete. + saidHello.wait(); // Wait for all tasks to complete. printf("All tasks said hello.\n");
diff --git a/include/marl/parallelize.h b/include/marl/parallelize.h new file mode 100644 index 0000000..d7ceadc --- /dev/null +++ b/include/marl/parallelize.h
@@ -0,0 +1,50 @@ +// 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. + +#ifndef marl_parallelize_h +#define marl_parallelize_h + +#include "scheduler.h" +#include "waitgroup.h" + +namespace marl { + +namespace detail { + +void parallelizeChain(WaitGroup*) {} + +template <typename F, typename... L> +void parallelizeChain(WaitGroup* wg, F&& f, L&&... l) { + schedule([=] { + f(); + wg->done(); + }); + parallelizeChain(wg, std::forward<L>(l)...); +} + +} // namespace detail + +// parallelize() schedules all the function parameters and waits for them to +// complete. These functions may execute concurrently. +// Each function must take no parameters. +template <typename... FUNCTIONS> +inline void parallelize(FUNCTIONS&&... functions) { + WaitGroup wg(sizeof...(FUNCTIONS)); + detail::parallelizeChain(&wg, functions...); + wg.wait(); +} + +} // namespace marl + +#endif // marl_parallelize_h
diff --git a/include/marl/scheduler.h b/include/marl/scheduler.h index 4db4b4d..0acfdb9 100644 --- a/include/marl/scheduler.h +++ b/include/marl/scheduler.h
@@ -255,7 +255,7 @@ // take() returns the next fiber that has exceeded its timeout, or nullptr // if there are no fibers that have yet exceeded their timeouts. - inline Fiber* take(const TimePoint& timepoint); + inline Fiber* take(const TimePoint& timeout); // next() returns the timepoint of the next fiber to timeout. // next() can only be called if operator bool() returns true.
diff --git a/include/marl/utils.h b/include/marl/utils.h deleted file mode 100644 index 5ecae80..0000000 --- a/include/marl/utils.h +++ /dev/null
@@ -1,52 +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. - -#ifndef marl_util_h -#define marl_util_h - -#include "scheduler.h" -#include "waitgroup.h" - -namespace marl { - -// parallelize() is used to split a number of work items into N smaller batches -// which can be processed in parallel with the function f(). -// numTotal is the total number of work items to process. -// numPerTask is the maximum number of work items to process per call to f(). -// There will always be at least one call to f(). -// F must be a function with the signature: -// void(COUNTER taskIndex, COUNTER first, COUNTER count) -// COUNTER is any integer type. -template <typename F, typename COUNTER> -inline void parallelize(COUNTER numTotal, COUNTER numPerTask, const F& f) { - auto numTasks = (numTotal + numPerTask - 1) / numPerTask; - WaitGroup wg(numTasks - 1); - for (unsigned int task = 1; task < numTasks; task++) { - schedule([=] { - auto first = task * numPerTask; - auto count = std::min(first + numPerTask, numTotal) - first; - f(task, first, count); - wg.done(); - }); - } - - // Run the first chunk on this fiber to reduce the amount of time spent - // waiting. - f(0, 0, std::min(numPerTask, numTotal)); - wg.wait(); -} - -} // namespace marl - -#endif // marl_util_h
diff --git a/src/parallelize_test.cpp b/src/parallelize_test.cpp new file mode 100644 index 0000000..c111d40 --- /dev/null +++ b/src/parallelize_test.cpp
@@ -0,0 +1,27 @@ +// 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. + +#include "marl_test.h" + +#include "marl/parallelize.h" + +TEST_P(WithBoundScheduler, Parallelize) { + bool a = false; + bool b = false; + bool c = false; + marl::parallelize([&] { a = true; }, [&] { b = true; }, [&] { c = true; }); + ASSERT_TRUE(a); + ASSERT_TRUE(b); + ASSERT_TRUE(c); +}
diff --git a/src/scheduler.cpp b/src/scheduler.cpp index 533892a..b6443d1 100644 --- a/src/scheduler.cpp +++ b/src/scheduler.cpp
@@ -120,7 +120,7 @@ "singleThreadedWorker not found"); MARL_ASSERT(it->second.get() == worker, "worker is not bound?"); bound->singleThreadedWorkers.byTid.erase(it); - if (bound->singleThreadedWorkers.byTid.size() == 0) { + if (bound->singleThreadedWorkers.byTid.empty()) { bound->singleThreadedWorkers.unbind.notify_one(); } } @@ -140,7 +140,7 @@ marl::lock lock(singleThreadedWorkers.mutex); lock.wait(singleThreadedWorkers.unbind, [this]() REQUIRES(singleThreadedWorkers.mutex) { - return singleThreadedWorkers.byTid.size() == 0; + return singleThreadedWorkers.byTid.empty(); }); } @@ -149,9 +149,9 @@ setWorkerThreadCount(0); } -void Scheduler::setThreadInitializer(const std::function<void()>& func) { +void Scheduler::setThreadInitializer(const std::function<void()>& init) { marl::lock lock(threadInitFuncMutex); - threadInitFunc = func; + threadInitFunc = init; } const std::function<void()>& Scheduler::getThreadInitializer() { @@ -299,15 +299,15 @@ // Scheduler::WaitingFibers //////////////////////////////////////////////////////////////////////////////// Scheduler::WaitingFibers::operator bool() const { - return fibers.size() > 0; + return !fibers.empty(); } -Scheduler::Fiber* Scheduler::WaitingFibers::take(const TimePoint& timepoint) { +Scheduler::Fiber* Scheduler::WaitingFibers::take(const TimePoint& timeout) { if (!*this) { return nullptr; } auto it = timeouts.begin(); - if (timepoint < it->timepoint) { + if (timeout < it->timepoint) { return nullptr; } auto fiber = it->fiber; @@ -324,9 +324,9 @@ return timeouts.begin()->timepoint; } -void Scheduler::WaitingFibers::add(const TimePoint& timepoint, Fiber* fiber) { - timeouts.emplace(Timeout{timepoint, fiber}); - bool added = fibers.emplace(fiber, timepoint).second; +void Scheduler::WaitingFibers::add(const TimePoint& timeout, Fiber* fiber) { + timeouts.emplace(Timeout{timeout, fiber}); + bool added = fibers.emplace(fiber, timeout).second; (void)added; MARL_ASSERT(added, "WaitingFibers::add() fiber already waiting"); } @@ -474,13 +474,13 @@ work.numBlockedFibers++; - if (work.fibers.size() > 0) { + if (!work.fibers.empty()) { // There's another fiber that has become unblocked, resume that. work.num--; auto to = take(work.fibers); ASSERT_FIBER_STATE(to, Fiber::State::Queued); switchToFiber(to); - } else if (idleFibers.size() > 0) { + } else if (!idleFibers.empty()) { // There's an old fiber we can reuse, resume that. auto to = take(idleFibers); ASSERT_FIBER_STATE(to, Fiber::State::Idle); @@ -518,7 +518,7 @@ break; } notify = work.notifyAdded; - work.fibers.push_back(std::move(fiber)); + work.fibers.push_back(fiber); MARL_ASSERT(!work.waiting.contains(fiber), "fiber is unexpectedly in the waiting list"); setFiberState(fiber, Fiber::State::Queued); @@ -552,8 +552,7 @@ if (!work.mutex.try_lock()) { return false; } - if (work.tasks.size() == 0 || - work.tasks.front().is(Task::Flags::SameThread)) { + if (work.tasks.empty() || work.tasks.front().is(Task::Flags::SameThread)) { work.mutex.unlock(); return false; } @@ -668,12 +667,12 @@ ASSERT_FIBER_STATE(currentFiber, Fiber::State::Running); MARL_ASSERT(work.num == work.fibers.size() + work.tasks.size(), "work.num out of sync"); - while (work.fibers.size() > 0 || work.tasks.size() > 0) { + while (!work.fibers.empty() || !work.tasks.empty()) { // Note: we cannot take and store on the stack more than a single fiber // or task at a time, as the Fiber may yield and these items may get // held on suspended fiber stack. - while (work.fibers.size() > 0) { + while (!work.fibers.empty()) { work.num--; auto fiber = take(work.fibers); // Sanity checks, @@ -690,7 +689,7 @@ changeFiberState(currentFiber, Fiber::State::Idle, Fiber::State::Running); } - if (work.tasks.size() > 0) { + if (!work.tasks.empty()) { work.num--; auto task = take(work.tasks); work.mutex.unlock();
diff --git a/tools/bench/bench.go b/tools/bench/bench.go index 81f2d72..ad81181 100644 --- a/tools/bench/bench.go +++ b/tools/bench/bench.go
@@ -69,7 +69,6 @@ case nil: return b, nil case errWrongFormat: - break default: return Benchmark{}, err }