[core] Decouple Scheduler interface from actor model

So that it is possible to schedule normal `std::function` and use `mapbox::base::WeakPtr`.
This commit is contained in:
Mikhail Pozdnyakov
2019-10-04 12:30:41 +03:00
parent 198e345339
commit bef635765a
13 changed files with 38 additions and 39 deletions

View File

@@ -2,6 +2,7 @@
#include <mbgl/util/optional.hpp>
#include <functional>
#include <memory>
#include <mutex>
#include <queue>
@@ -34,6 +35,7 @@ public:
void receive();
static void maybeReceive(std::weak_ptr<Mailbox>);
static std::function<void()> makeClosure(std::weak_ptr<Mailbox>);
private:
optional<Scheduler*> scheduler;

View File

@@ -1,5 +1,6 @@
#pragma once
#include <functional>
#include <memory>
namespace mbgl {
@@ -31,12 +32,9 @@ class Mailbox;
class Scheduler {
public:
virtual ~Scheduler() = default;
// Used by a Mailbox when it has a message in its queue to request that it
// be scheduled. Specifically, the scheduler is expected to asynchronously
// call `receive() on the given mailbox, provided it still exists at that
// time.
virtual void schedule(std::weak_ptr<Mailbox>) = 0;
// Enqueues a function for execution.
virtual void schedule(std::function<void()>) = 0;
// Set/Get the current Scheduler for this thread
static Scheduler* GetCurrent();

View File

@@ -72,12 +72,8 @@ public:
push(Priority::Default, task);
return std::make_unique<WorkRequest>(task);
}
void schedule(std::weak_ptr<Mailbox> mailbox) override {
invoke([mailbox] () {
Mailbox::maybeReceive(mailbox);
});
}
void schedule(std::function<void()> fn) override { invoke(std::move(fn)); }
class Impl;

View File

@@ -43,7 +43,7 @@ ActorRef<Renderer> MapRenderer::actor() const {
return *rendererRef;
}
void MapRenderer::schedule(std::weak_ptr<Mailbox> scheduled) {
void MapRenderer::schedule(std::function<void()> scheduled) {
// Create a runnable
android::UniqueEnv _env = android::AttachEnv();
auto runnable = std::make_unique<MapRendererRunnable>(*_env, std::move(scheduled));

View File

@@ -67,7 +67,7 @@ public:
// From Scheduler. Schedules by using callbacks to the
// JVM to process the mailbox on the right thread.
void schedule(std::weak_ptr<Mailbox> scheduled) override;
void schedule(std::function<void()> scheduled) override;
void requestRender();

View File

@@ -5,9 +5,8 @@
namespace mbgl {
namespace android {
MapRendererRunnable::MapRendererRunnable(jni::JNIEnv& env, std::weak_ptr<Mailbox> mailbox_)
: mailbox(std::move(mailbox_)) {
MapRendererRunnable::MapRendererRunnable(jni::JNIEnv& env, std::function<void()> function_)
: function(std::move(function_)) {
// Create the Java peer and hold on to a global reference
// Not using a weak reference here as this might oerflow
// the weak reference table on some devices
@@ -21,7 +20,7 @@ MapRendererRunnable::MapRendererRunnable(jni::JNIEnv& env, std::weak_ptr<Mailbox
MapRendererRunnable::~MapRendererRunnable() = default;
void MapRendererRunnable::run(jni::JNIEnv&) {
Mailbox::maybeReceive(mailbox);
if (function) function();
}
jni::Global<jni::Object<MapRendererRunnable>> MapRendererRunnable::peer() {

View File

@@ -24,7 +24,7 @@ public:
static void registerNative(jni::JNIEnv&);
MapRendererRunnable(jni::JNIEnv&, std::weak_ptr<Mailbox>);
MapRendererRunnable(jni::JNIEnv&, std::function<void()>);
// Only for jni registration, unused
MapRendererRunnable(jni::JNIEnv&) {
@@ -40,7 +40,7 @@ public:
private:
jni::Global<jni::Object<MapRendererRunnable>> javaPeer;
std::weak_ptr<Mailbox> mailbox;
std::function<void()> function;
};
} // namespace android

View File

@@ -13,10 +13,9 @@ QMapboxGLScheduler::~QMapboxGLScheduler()
MBGL_VERIFY_THREAD(tid);
}
void QMapboxGLScheduler::schedule(std::weak_ptr<mbgl::Mailbox> mailbox)
{
void QMapboxGLScheduler::schedule(std::function<void()> function) {
std::lock_guard<std::mutex> lock(m_taskQueueMutex);
m_taskQueue.push(mailbox);
m_taskQueue.push(std::move(function));
// Need to force the main thread to wake
// up this thread and process the events.
@@ -25,14 +24,15 @@ void QMapboxGLScheduler::schedule(std::weak_ptr<mbgl::Mailbox> mailbox)
void QMapboxGLScheduler::processEvents()
{
std::queue<std::weak_ptr<mbgl::Mailbox>> taskQueue;
std::queue<std::function<void()>> taskQueue;
{
std::unique_lock<std::mutex> lock(m_taskQueueMutex);
std::swap(taskQueue, m_taskQueue);
}
while (!taskQueue.empty()) {
mbgl::Mailbox::maybeReceive(taskQueue.front());
auto& function = taskQueue.front();
if (function) function();
taskQueue.pop();
}
}

View File

@@ -1,6 +1,5 @@
#pragma once
#include <mbgl/actor/mailbox.hpp>
#include <mbgl/actor/scheduler.hpp>
#include <mbgl/util/util.hpp>
@@ -19,7 +18,7 @@ public:
virtual ~QMapboxGLScheduler();
// mbgl::Scheduler implementation.
void schedule(std::weak_ptr<mbgl::Mailbox> scheduled) final;
void schedule(std::function<void()> scheduled) final;
void processEvents();
@@ -30,5 +29,5 @@ private:
MBGL_STORE_THREAD(tid);
std::mutex m_taskQueueMutex;
std::queue<std::weak_ptr<mbgl::Mailbox>> m_taskQueue;
std::queue<std::function<void()>> m_taskQueue;
};

View File

@@ -27,7 +27,7 @@ void Mailbox::open(Scheduler& scheduler_) {
}
if (!queue.empty()) {
(*scheduler)->schedule(shared_from_this());
(*scheduler)->schedule(makeClosure(shared_from_this()));
}
}
@@ -57,7 +57,7 @@ void Mailbox::push(std::unique_ptr<Message> message) {
bool wasEmpty = queue.empty();
queue.push(std::move(message));
if (wasEmpty && scheduler) {
(*scheduler)->schedule(shared_from_this());
(*scheduler)->schedule(makeClosure(shared_from_this()));
}
}
@@ -84,14 +84,20 @@ void Mailbox::receive() {
(*message)();
if (!wasEmpty) {
(*scheduler)->schedule(shared_from_this());
(*scheduler)->schedule(makeClosure(shared_from_this()));
}
}
// static
void Mailbox::maybeReceive(std::weak_ptr<Mailbox> mailbox) {
if (auto locked = mailbox.lock()) {
locked->receive();
}
}
// static
std::function<void()> Mailbox::makeClosure(std::weak_ptr<Mailbox> mailbox) {
return [mailbox]() { maybeReceive(mailbox); };
}
} // namespace mbgl

View File

@@ -26,11 +26,10 @@ ThreadPool::ThreadPool(std::size_t count) {
return;
}
auto mailbox = queue.front();
auto function = std::move(queue.front());
queue.pop();
lock.unlock();
Mailbox::maybeReceive(mailbox);
if (function) function();
}
});
}
@@ -49,10 +48,10 @@ ThreadPool::~ThreadPool() {
}
}
void ThreadPool::schedule(std::weak_ptr<Mailbox> mailbox) {
void ThreadPool::schedule(std::function<void()> fn) {
{
std::lock_guard<std::mutex> lock(mutex);
queue.push(mailbox);
queue.push(std::move(fn));
}
cv.notify_one();

View File

@@ -15,11 +15,11 @@ public:
explicit ThreadPool(std::size_t count);
~ThreadPool() override;
void schedule(std::weak_ptr<Mailbox>) override;
void schedule(std::function<void()>) override;
private:
std::vector<std::thread> threads;
std::queue<std::weak_ptr<Mailbox>> queue;
std::queue<std::function<void()>> queue;
std::mutex mutex;
std::condition_variable cv;
bool terminate{ false };

View File

@@ -101,7 +101,7 @@ TEST(Actor, DestructionBlocksOnSend) {
EXPECT_TRUE(waited.load());
}
void schedule(std::weak_ptr<Mailbox>) final {
void schedule(std::function<void()>) final {
promise.set_value();
future.wait();
std::this_thread::sleep_for(1ms);