core: rename mbgl directories to mln (#4511)
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
This commit is contained in:
168
include/mln/util/thread.hpp
Normal file
168
include/mln/util/thread.hpp
Normal file
@@ -0,0 +1,168 @@
|
||||
#pragma once
|
||||
|
||||
#include <mln/actor/actor.hpp>
|
||||
#include <mln/actor/mailbox.hpp>
|
||||
#include <mln/actor/scheduler.hpp>
|
||||
#include <mln/platform/thread.hpp>
|
||||
#include <mln/util/platform.hpp>
|
||||
#include <mln/util/run_loop.hpp>
|
||||
#include <mln/util/util.hpp>
|
||||
#include <mln/util/instrumentation.hpp>
|
||||
|
||||
#include <cassert>
|
||||
#include <future>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <utility>
|
||||
|
||||
namespace mln {
|
||||
namespace util {
|
||||
|
||||
/// @brief Manages a thread with `Object`.
|
||||
///
|
||||
/// Upon creation of this object, it launches a thread and creates an object of
|
||||
/// type `Object` in that thread. When the `Thread<>` object is destructed, the
|
||||
/// destructor waits for thread termination. The `Thread<>` constructor blocks
|
||||
/// until the thread and the `Object` are fully created, so after the object
|
||||
/// creation, it's safe to obtain the `Object` stored in this thread. The thread
|
||||
/// created will always have low priority on the platforms that support setting
|
||||
/// thread priority.
|
||||
///
|
||||
/// The following properties make this class different from `ThreadPool`:
|
||||
///
|
||||
/// - Only one thread is created.
|
||||
/// - `Object` will live in a single thread, providing thread affinity.
|
||||
/// - It is safe to use `ThreadLocal` in an `Object` managed by `Thread<>`
|
||||
/// - A `RunLoop` is created for the `Object` thread.
|
||||
/// - `Object` can use `Timer` and do asynchronous I/O, like wait for sockets
|
||||
/// events.
|
||||
template <typename Object>
|
||||
class Thread {
|
||||
public:
|
||||
template <typename TupleArgs>
|
||||
Thread(std::function<void()> prioritySetter_, const std::string& name, TupleArgs&& args) {
|
||||
std::promise<void> running_;
|
||||
running = running_.get_future();
|
||||
thread = std::thread([this,
|
||||
name,
|
||||
capturedArgs = std::forward<TupleArgs>(args),
|
||||
runningPromise = std::move(running_),
|
||||
prioritySetter = std::move(prioritySetter_)]() mutable {
|
||||
platform::setCurrentThreadName(name);
|
||||
instrumentation::setThreadName(name);
|
||||
|
||||
if (prioritySetter) prioritySetter();
|
||||
platform::attachThread();
|
||||
|
||||
// narrowing the scope to release the Object before we detach the thread
|
||||
{
|
||||
util::RunLoop loop_(util::RunLoop::Type::New);
|
||||
loop = &loop_;
|
||||
EstablishedActor<Object> establishedActor(loop_, object, std::move(capturedArgs));
|
||||
|
||||
runningPromise.set_value();
|
||||
|
||||
loop->run();
|
||||
|
||||
(void)establishedActor;
|
||||
|
||||
loop = nullptr;
|
||||
}
|
||||
|
||||
platform::detachThread();
|
||||
});
|
||||
}
|
||||
|
||||
template <typename... Args>
|
||||
Thread(const std::string& name, Args&&... args)
|
||||
: Thread([] { platform::makeThreadLowPriority(); }, name, std::make_tuple(std::forward<Args>(args)...)) {}
|
||||
|
||||
template <typename... Args>
|
||||
Thread(const std::function<void()>& prioritySetter, const std::string& name, Args&&... args)
|
||||
: Thread(prioritySetter, name, std::make_tuple(std::forward<Args>(args)...)) {}
|
||||
|
||||
~Thread() {
|
||||
if (paused) {
|
||||
resume();
|
||||
}
|
||||
|
||||
std::promise<void> stoppable;
|
||||
|
||||
running.wait();
|
||||
|
||||
// Invoke a noop task on the run loop to ensure that we're executing
|
||||
// run() before we call stop()
|
||||
loop->invoke([&] { stoppable.set_value(); });
|
||||
|
||||
stoppable.get_future().get();
|
||||
|
||||
loop->stop();
|
||||
thread.join();
|
||||
}
|
||||
|
||||
/// Returns a non-owning reference to `Object` that
|
||||
/// can be used to send messages to `Object`. It is safe
|
||||
/// to the non-owning reference to outlive this object
|
||||
/// and be used after the `Thread<>` gets destroyed.
|
||||
ActorRef<std::decay_t<Object>> actor() { return object.self(); }
|
||||
|
||||
/// Pauses the `Object` thread. It will prevent the object to wake
|
||||
/// up from events such as timers and file descriptor I/O. Messages
|
||||
/// sent to a paused `Object` will be queued and only processed after
|
||||
/// `resume()` is called.
|
||||
void pause() {
|
||||
MBGL_VERIFY_THREAD(tid);
|
||||
|
||||
assert(!paused);
|
||||
|
||||
paused = std::make_unique<std::promise<void>>();
|
||||
resumed = std::make_unique<std::promise<void>>();
|
||||
|
||||
auto pausing = paused->get_future();
|
||||
|
||||
running.wait();
|
||||
|
||||
loop->invoke(RunLoop::Priority::High, [this] {
|
||||
auto resuming = resumed->get_future();
|
||||
paused->set_value();
|
||||
resuming.get();
|
||||
});
|
||||
|
||||
pausing.get();
|
||||
}
|
||||
|
||||
// Resumes the `Object` thread previously paused by `pause()`.
|
||||
void resume() {
|
||||
MBGL_VERIFY_THREAD(tid);
|
||||
|
||||
assert(paused);
|
||||
|
||||
resumed->set_value();
|
||||
|
||||
resumed.reset();
|
||||
paused.reset();
|
||||
}
|
||||
|
||||
private:
|
||||
MBGL_STORE_THREAD(tid);
|
||||
|
||||
AspiringActor<Object> object;
|
||||
|
||||
std::thread thread;
|
||||
|
||||
std::future<void> running;
|
||||
|
||||
std::unique_ptr<std::promise<void>> paused;
|
||||
std::unique_ptr<std::promise<void>> resumed;
|
||||
|
||||
util::RunLoop* loop = nullptr;
|
||||
};
|
||||
|
||||
/// Returns function, that once invoked, will set a thread priority for
|
||||
/// a thread `threadType` based on a setting provided by corresponding
|
||||
/// Settings' platform::EXPERIMENTAL_THREAD_PRIORITY_* value.
|
||||
std::function<void()> makeThreadPrioritySetter(std::string threadType);
|
||||
|
||||
} // namespace util
|
||||
} // namespace mln
|
||||
Reference in New Issue
Block a user