From 0de17ea5f47e9e6c4b67170d4d04b0425e67ce5c Mon Sep 17 00:00:00 2001 From: "Thiago Marcos P. Santos" Date: Tue, 13 Jun 2017 18:44:13 +0300 Subject: [core] Added the ThreadedObject Actor model version of util::Thread. --- src/mbgl/storage/file_source_request.cpp | 37 +++++++ src/mbgl/storage/file_source_request.hpp | 31 ++++++ src/mbgl/util/threaded_object.hpp | 163 +++++++++++++++++++++++++++++++ 3 files changed, 231 insertions(+) create mode 100644 src/mbgl/storage/file_source_request.cpp create mode 100644 src/mbgl/storage/file_source_request.hpp create mode 100644 src/mbgl/util/threaded_object.hpp (limited to 'src') diff --git a/src/mbgl/storage/file_source_request.cpp b/src/mbgl/storage/file_source_request.cpp new file mode 100644 index 0000000000..8a6fb21181 --- /dev/null +++ b/src/mbgl/storage/file_source_request.cpp @@ -0,0 +1,37 @@ +#include + +#include +#include + +namespace mbgl { + +FileSourceRequest::FileSourceRequest(FileSource::Callback&& callback) + : responseCallback(callback) + , mailbox(std::make_shared(*util::RunLoop::Get())) { +} + +FileSourceRequest::~FileSourceRequest() { + if (cancelCallback) { + cancelCallback(); + } + + mailbox->close(); +} + +void FileSourceRequest::onCancel(std::function&& callback) { + cancelCallback = std::move(callback); +} + +void FileSourceRequest::setResponse(const Response& response) { + // Copy, because calling the callback will sometimes self + // destroy this object. We cannot move because this method + // can be called more than one. + auto callback = responseCallback; + callback(response); +} + +ActorRef FileSourceRequest::actor() { + return ActorRef(*this, mailbox); +} + +} // namespace mbgl diff --git a/src/mbgl/storage/file_source_request.hpp b/src/mbgl/storage/file_source_request.hpp new file mode 100644 index 0000000000..6bd0d44df6 --- /dev/null +++ b/src/mbgl/storage/file_source_request.hpp @@ -0,0 +1,31 @@ +#pragma once + +#include +#include +#include + +#include +#include + +namespace mbgl { + +class Mailbox; + +class FileSourceRequest : public AsyncRequest { +public: + FileSourceRequest(FileSource::Callback&& callback); + ~FileSourceRequest() final; + + void onCancel(std::function&& callback); + void setResponse(const Response& res); + + ActorRef actor(); + +private: + FileSource::Callback responseCallback = nullptr; + std::function cancelCallback = nullptr; + + std::shared_ptr mailbox; +}; + +} // namespace mbgl diff --git a/src/mbgl/util/threaded_object.hpp b/src/mbgl/util/threaded_object.hpp new file mode 100644 index 0000000000..7310eed2b1 --- /dev/null +++ b/src/mbgl/util/threaded_object.hpp @@ -0,0 +1,163 @@ +#pragma once + +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace mbgl { +namespace util { + +// 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 `ThreadedObject<>` object is destructed, the destructor waits +// for thread termination. The `ThreadedObject<>` 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 `ThreadedObject<>` +// - A `RunLoop` is created for the `Object` thread. +// - `Object` can use `Timer` and do asynchronous I/O, like wait for sockets events. +// +template +class ThreadedObject : public Scheduler { +public: + template + ThreadedObject(const std::string& name, Args&&... args) { + std::promise running; + + thread = std::thread([&] { + platform::setCurrentThreadName(name); + platform::makeThreadLowPriority(); + + util::RunLoop loop_(util::RunLoop::Type::New); + loop = &loop_; + + object = std::make_unique>(*this, std::forward(args)...); + running.set_value(); + + loop->run(); + loop = nullptr; + }); + + running.get_future().get(); + } + + ~ThreadedObject() override { + MBGL_VERIFY_THREAD(tid); + + if (paused) { + resume(); + } + + std::promise joinable; + + // Kill the actor, so we don't get more + // messages posted on this scheduler after + // we delete the RunLoop. + loop->invoke([&] { + object.reset(); + joinable.set_value(); + }); + + joinable.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 `ThreadedObject<>` gets destroyed. + ActorRef> actor() const { + 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>(); + resumed = std::make_unique>(); + + auto pausing = paused->get_future(); + + loop->invoke([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); + + void schedule(std::weak_ptr mailbox) override { + { + std::lock_guard lock(mutex); + queue.push(mailbox); + } + + loop->invoke([this] { receive(); }); + } + + void receive() { + std::unique_lock lock(mutex); + + auto mailbox = queue.front(); + queue.pop(); + lock.unlock(); + + Mailbox::maybeReceive(mailbox); + } + + std::mutex mutex; + std::queue> queue; + std::thread thread; + std::unique_ptr> object; + + std::unique_ptr> paused; + std::unique_ptr> resumed; + + util::RunLoop* loop = nullptr; +}; + +} // namespace util +} // namespace mbgl -- cgit v1.2.1