diff --git a/cmake/core-files.cmake b/cmake/core-files.cmake index 1a86da6f247..f1a0be23bbc 100644 --- a/cmake/core-files.cmake +++ b/cmake/core-files.cmake @@ -606,13 +606,14 @@ set(MBGL_CORE_FILES src/mbgl/util/premultiply.cpp src/mbgl/util/rapidjson.hpp src/mbgl/util/rect.hpp + src/mbgl/util/scheduled_timer.cpp + src/mbgl/util/scheduled_timer.hpp src/mbgl/util/std.hpp src/mbgl/util/stopwatch.cpp src/mbgl/util/stopwatch.hpp src/mbgl/util/string.cpp src/mbgl/util/thread.hpp src/mbgl/util/thread_local.hpp - src/mbgl/util/throttler.cpp src/mbgl/util/throttler.hpp src/mbgl/util/tile_coordinate.hpp src/mbgl/util/tile_cover.cpp diff --git a/cmake/node.cmake b/cmake/node.cmake index 502edd82933..a5c450eae7d 100644 --- a/cmake/node.cmake +++ b/cmake/node.cmake @@ -11,6 +11,9 @@ add_nodejs_module(mbgl-node set_target_properties("mbgl-node" PROPERTIES CXX_STANDARD 14) target_sources(mbgl-node + PRIVATE platform/default/async_task.cpp + PRIVATE platform/default/run_loop.cpp + PRIVATE platform/default/timer.cpp PRIVATE platform/node/src/node_logging.hpp PRIVATE platform/node/src/node_logging.cpp PRIVATE platform/node/src/node_map.hpp @@ -30,16 +33,16 @@ target_compile_options(mbgl-node ) target_include_directories(mbgl-node + PRIVATE include + PRIVATE src PRIVATE platform/default ) -# Use node-provided uv.h. This is not part of loop-uv.cmake because loop-uv.cmake is also -# used by linux/config.cmake, where we need to use headers provided by mason's libuv. +# Use node-provided uv.h target_include_directories(mbgl-loop-uv PUBLIC ${NODEJS_INCLUDE_DIRS}) target_link_libraries(mbgl-node PRIVATE mbgl-core - PRIVATE mbgl-loop-uv ) target_add_mason_package(mbgl-node PRIVATE geojson) diff --git a/cmake/test-files.cmake b/cmake/test-files.cmake index 19412e61335..7d68e31db61 100644 --- a/cmake/test-files.cmake +++ b/cmake/test-files.cmake @@ -135,6 +135,7 @@ set(MBGL_TEST_FILES test/util/text_conversions.test.cpp test/util/thread.test.cpp test/util/thread_local.test.cpp + test/util/throttler.test.cpp test/util/tile_cover.test.cpp test/util/timer.test.cpp test/util/token.test.cpp diff --git a/include/mbgl/actor/actor_ref.hpp b/include/mbgl/actor/actor_ref.hpp index 958ee3777c3..7b6903ce17e 100644 --- a/include/mbgl/actor/actor_ref.hpp +++ b/include/mbgl/actor/actor_ref.hpp @@ -2,6 +2,7 @@ #include #include +#include #include @@ -55,6 +56,11 @@ class ActorRef { return future; } + + template + std::unique_ptr invokeDelayed(Duration timeout, Fn fn, Args&&... args, Scheduler& scheduler = *Scheduler::GetCurrent()) { + return scheduler.schedule(timeout, weakMailbox, actor::makeMessage(*object, fn, std::forward(args)...)); + } private: Object* object; diff --git a/include/mbgl/actor/scheduler.hpp b/include/mbgl/actor/scheduler.hpp index d8a26ebeab1..75d191c5114 100644 --- a/include/mbgl/actor/scheduler.hpp +++ b/include/mbgl/actor/scheduler.hpp @@ -1,10 +1,14 @@ #pragma once +#include +#include + #include namespace mbgl { class Mailbox; +class Message; /* A `Scheduler` is responsible for coordinating the processing of messages by @@ -33,6 +37,21 @@ class Scheduler { virtual ~Scheduler() = default; virtual void schedule(std::weak_ptr) = 0; + /* + A token to the scheduled action, at the end of + the token's life-time, the scheduled action is + canceled. + */ + class Scheduled : public AsyncRequest { + public: + ~Scheduled() override = default; + + virtual bool isFinished() = 0; + }; + + // Schedule message delivery at a later time + virtual std::unique_ptr schedule(Duration timeout, std::weak_ptr, std::unique_ptr) = 0; + // Set/Get the current Scheduler for this thread static Scheduler* GetCurrent(); static void SetCurrent(Scheduler*); diff --git a/include/mbgl/util/run_loop.hpp b/include/mbgl/util/run_loop.hpp index acbea802739..cef03824f1c 100644 --- a/include/mbgl/util/run_loop.hpp +++ b/include/mbgl/util/run_loop.hpp @@ -62,12 +62,14 @@ class RunLoop : public Scheduler, push(task); return std::make_unique(task); } - + void schedule(std::weak_ptr mailbox) override { invoke([mailbox] () { Mailbox::maybeReceive(mailbox); }); } + + std::unique_ptr schedule(Duration, std::weak_ptr, std::unique_ptr) override; class Impl; diff --git a/platform/android/src/run_loop.cpp b/platform/android/src/run_loop.cpp index dff7d1d9845..9ce9278b8c2 100644 --- a/platform/android/src/run_loop.cpp +++ b/platform/android/src/run_loop.cpp @@ -1,9 +1,9 @@ #include "run_loop_impl.hpp" #include -#include #include #include +#include #include #include @@ -200,7 +200,7 @@ Milliseconds RunLoop::Impl::processRunnables() { } RunLoop* RunLoop::Get() { - assert(static_cast(Scheduler::GetCurrent())); + assert(dynamic_cast(Scheduler::GetCurrent())); return static_cast(Scheduler::GetCurrent()); } @@ -258,5 +258,9 @@ void RunLoop::removeWatch(int) { throw std::runtime_error("Not implemented."); } +std::unique_ptr RunLoop::schedule(Duration timeout, std::weak_ptr mailbox, std::unique_ptr message) { + return std::make_unique(*this, timeout, std::move(mailbox), std::move(message)); +} + } // namespace util } // namespace mbgl diff --git a/platform/darwin/src/run_loop.cpp b/platform/darwin/src/run_loop.cpp index 2ba8f8415bb..206fee05318 100644 --- a/platform/darwin/src/run_loop.cpp +++ b/platform/darwin/src/run_loop.cpp @@ -1,6 +1,11 @@ #include #include +#include +#include +#include +#include #include +#include #include @@ -13,7 +18,7 @@ class RunLoop::Impl { }; RunLoop* RunLoop::Get() { - assert(static_cast(Scheduler::GetCurrent())); + assert(dynamic_cast(Scheduler::GetCurrent())); return static_cast(Scheduler::GetCurrent()); } @@ -45,6 +50,10 @@ void RunLoop::runOnce() { void RunLoop::stop() { invoke([&] { CFRunLoopStop(CFRunLoopGetCurrent()); }); } + +std::unique_ptr RunLoop::schedule(Duration timeout, std::weak_ptr mailbox, std::unique_ptr message) { + return std::make_unique(*this, timeout, std::move(mailbox), std::move(message)); +} } // namespace util } // namespace mbgl diff --git a/platform/default/mbgl/util/default_thread_pool.cpp b/platform/default/mbgl/util/default_thread_pool.cpp index d3950bb8aa1..0ea891d30dd 100644 --- a/platform/default/mbgl/util/default_thread_pool.cpp +++ b/platform/default/mbgl/util/default_thread_pool.cpp @@ -1,14 +1,32 @@ #include -#include #include #include +#include namespace mbgl { + ThreadPool::ThreadPool(std::size_t count) { + std::promise timerThreadRunning; + + timerThread = std::thread([&]() { + platform::setCurrentThreadName("Timer thread"); + platform::makeThreadLowPriority(); + + util::RunLoop loop(util::RunLoop::Type::New); + timerLoop = &loop; + timerThreadRunning.set_value(); + + loop.run(); + }); + + timerThreadRunning.get_future().get(); + threads.reserve(count); for (std::size_t i = 0; i < count; ++i) { - threads.emplace_back([this, i]() { + auto scheduler = this; + threads.emplace_back([this, scheduler, i]() { + Scheduler::SetCurrent(scheduler); platform::setCurrentThreadName(std::string{ "Worker " } + util::toString(i + 1)); while (true) { @@ -38,11 +56,16 @@ ThreadPool::~ThreadPool() { terminate = true; } + // Stop the threadpool threads cv.notify_all(); for (auto& thread : threads) { thread.join(); } + + // Stop the timer thread + timerLoop->stop(); + timerThread.join(); } void ThreadPool::schedule(std::weak_ptr mailbox) { @@ -53,5 +76,10 @@ void ThreadPool::schedule(std::weak_ptr mailbox) { cv.notify_one(); } + + +std::unique_ptr ThreadPool::schedule(Duration timeout, std::weak_ptr mailbox, std::unique_ptr message) { + return std::make_unique(*timerLoop, timeout, std::move(mailbox), std::move(message)); +} } // namespace mbgl diff --git a/platform/default/mbgl/util/default_thread_pool.hpp b/platform/default/mbgl/util/default_thread_pool.hpp index a14d16d7713..9a07b179787 100644 --- a/platform/default/mbgl/util/default_thread_pool.hpp +++ b/platform/default/mbgl/util/default_thread_pool.hpp @@ -1,6 +1,7 @@ #pragma once #include +#include #include #include @@ -15,6 +16,8 @@ class ThreadPool : public Scheduler { ~ThreadPool() override; void schedule(std::weak_ptr) override; + + std::unique_ptr schedule(Duration, std::weak_ptr, std::unique_ptr) override; private: std::vector threads; @@ -22,6 +25,10 @@ class ThreadPool : public Scheduler { std::mutex mutex; std::condition_variable cv; bool terminate { false }; + + // Dedicated thread for timers + std::thread timerThread; + util::RunLoop* timerLoop; }; } // namespace mbgl diff --git a/platform/default/run_loop.cpp b/platform/default/run_loop.cpp index 6375dba78ec..a3c659ecf41 100644 --- a/platform/default/run_loop.cpp +++ b/platform/default/run_loop.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include #include @@ -215,6 +216,10 @@ void RunLoop::removeWatch(int fd) { uv_close(reinterpret_cast(&watch->poll), &Watch::onClose); } + +std::unique_ptr RunLoop::schedule(Duration timeout, std::weak_ptr mailbox, std::unique_ptr message) { + return std::make_unique(*this, timeout, std::move(mailbox), std::move(message)); +} } // namespace util } // namespace mbgl diff --git a/platform/node/src/node_thread_pool.hpp b/platform/node/src/node_thread_pool.hpp index d412e53d3da..91e4ed48a8a 100644 --- a/platform/node/src/node_thread_pool.hpp +++ b/platform/node/src/node_thread_pool.hpp @@ -18,6 +18,12 @@ class NodeThreadPool : public mbgl::Scheduler { ~NodeThreadPool(); void schedule(std::weak_ptr) override; + + std::unique_ptr schedule(mbgl::Duration, std::weak_ptr, std::unique_ptr) override { + // Not implemented + assert(false); + return {}; + }; private: util::AsyncQueue>* queue; diff --git a/platform/qt/src/run_loop.cpp b/platform/qt/src/run_loop.cpp index 71ea19032a3..c4af173e7fc 100644 --- a/platform/qt/src/run_loop.cpp +++ b/platform/qt/src/run_loop.cpp @@ -1,6 +1,7 @@ #include "run_loop_impl.hpp" #include +#include #include @@ -119,6 +120,10 @@ void RunLoop::removeWatch(int fd) { impl->readPoll.erase(readPollIter); } } + +std::unique_ptr RunLoop::schedule(Duration timeout, std::weak_ptr mailbox, std::unique_ptr message) { + return std::make_unique(*this, timeout, std::move(mailbox), std::move(message)); +} } } diff --git a/src/mbgl/tile/geometry_tile.cpp b/src/mbgl/tile/geometry_tile.cpp index 20056e355d1..d1064654bf8 100644 --- a/src/mbgl/tile/geometry_tile.cpp +++ b/src/mbgl/tile/geometry_tile.cpp @@ -42,7 +42,7 @@ GeometryTile::GeometryTile(const OverscaledTileID& id_, parameters.pixelRatio), glyphManager(parameters.glyphManager), imageManager(parameters.imageManager), - placementThrottler(Milliseconds(300), [this] { invokePlacement(); }), + placementThrottler(ActorRef(*this, mailbox), Milliseconds(300)), lastYStretch(1.0f) { } @@ -86,7 +86,7 @@ void GeometryTile::setPlacementConfig(const PlacementConfig& desiredConfig) { ++correlationID; requestedConfig = desiredConfig; - placementThrottler.invoke(); + placementThrottler.invoke(&GeometryTile::invokePlacement); } void GeometryTile::invokePlacement() { diff --git a/src/mbgl/tile/geometry_tile.hpp b/src/mbgl/tile/geometry_tile.hpp index c45762742bd..8814e9facfb 100644 --- a/src/mbgl/tile/geometry_tile.hpp +++ b/src/mbgl/tile/geometry_tile.hpp @@ -141,7 +141,7 @@ class GeometryTile : public Tile, public GlyphRequestor, ImageRequestor { std::unordered_map> symbolBuckets; std::unique_ptr collisionTile; - util::Throttler placementThrottler; + util::Throttler placementThrottler; float lastYStretch; public: diff --git a/src/mbgl/util/scheduled_timer.cpp b/src/mbgl/util/scheduled_timer.cpp new file mode 100644 index 00000000000..517924da48f --- /dev/null +++ b/src/mbgl/util/scheduled_timer.cpp @@ -0,0 +1,61 @@ +#include + +#include +#include + +namespace mbgl { + + +class ScheduledTimer::Impl { +public: + // Will be called on the timer thread so it has access to the timer loop + void start(mbgl::Duration timeout, ActorRef task) { + // Create the timer on this thread and start it + timer = std::make_unique(); + timer->start(timeout, Duration::zero(), [task=std::move(task)]() mutable { + task.invoke(&ScheduledTimer::onTimer); + }); + } + + void cancel() { + if (timer) { + timer->stop(); + } + } + +private: + std::unique_ptr timer; +}; + +ScheduledTimer::ScheduledTimer(Scheduler& timerScheduler, + mbgl::Duration timeout, + std::weak_ptr destMailbox_, + std::unique_ptr message_) + : mailbox(std::make_shared(*Scheduler::GetCurrent())) + , impl(std::make_unique>(timerScheduler)) + , destMailbox(std::move(destMailbox_)) + , message(std::move(message_)) { + + // Let the timer start on the timer scheduler + impl->invoke(&Impl::start, timeout, ActorRef(*this, mailbox)); +} + +ScheduledTimer::~ScheduledTimer() { + impl->invoke(&ScheduledTimer::Impl::cancel); + mailbox->close(); +}; + +bool ScheduledTimer::isFinished() { + return finished; +} + +void ScheduledTimer::onTimer() { + if (!message) return; + + finished = true; + if (auto mb = destMailbox.lock()) { + mb->push(std::move(message)); + } +} + +} // namespace mbgl diff --git a/src/mbgl/util/scheduled_timer.hpp b/src/mbgl/util/scheduled_timer.hpp new file mode 100644 index 00000000000..cfc304eeb70 --- /dev/null +++ b/src/mbgl/util/scheduled_timer.hpp @@ -0,0 +1,48 @@ +#pragma once + +#include + +#include + +namespace mbgl { + +class Mailbox; +class Message; +class Scheduler; + +namespace util { +class RunLoop; +} // namespace util + +// Bridges to the dedicated timer thread +class ScheduledTimer : public Scheduler::Scheduled { +public: + + ScheduledTimer(Scheduler& timerScheduler, + mbgl::Duration timeout, + std::weak_ptr destMailbox, + std::unique_ptr message); + + ~ScheduledTimer() override; + + bool isFinished() override; + +private: + void onTimer(); + + // The Timer's own mailbox + std::shared_ptr mailbox; + + class Impl; + friend class Impl; + std::unique_ptr> impl; + + // The destination Mailbox and message + // to deliver + std::weak_ptr destMailbox; + std::unique_ptr message; + + std::atomic finished { false }; +}; + +} // namespace mbgl diff --git a/src/mbgl/util/thread.hpp b/src/mbgl/util/thread.hpp index 572f46080e1..4186a10f5a4 100644 --- a/src/mbgl/util/thread.hpp +++ b/src/mbgl/util/thread.hpp @@ -82,6 +82,10 @@ class Thread : public Scheduler { loop->stop(); thread.join(); } + + std::unique_ptr schedule(Duration timeout, std::weak_ptr mailbox, std::unique_ptr message) override { + return loop->schedule(timeout, std::move(mailbox), std::move(message)); + } // Returns a non-owning reference to `Object` that // can be used to send messages to `Object`. It is safe diff --git a/src/mbgl/util/throttler.cpp b/src/mbgl/util/throttler.cpp deleted file mode 100644 index 910810ce2fd..00000000000 --- a/src/mbgl/util/throttler.cpp +++ /dev/null @@ -1,36 +0,0 @@ -#include - -namespace mbgl { -namespace util { - -Throttler::Throttler(Duration frequency_, std::function&& function_) - : frequency(frequency_) - , function(std::move(function_)) - , pendingInvocation(false) - , lastInvocation(TimePoint::min()) -{} - -void Throttler::invoke() { - if (pendingInvocation) { - return; - } - - Duration timeToNextInvocation = lastInvocation == TimePoint::min() - ? Duration::zero() - : (lastInvocation + frequency) - Clock::now(); - - if (timeToNextInvocation <= Duration::zero()) { - lastInvocation = Clock::now(); - function(); - } else { - pendingInvocation = true; - timer.start(timeToNextInvocation, Duration::zero(), [this]{ - pendingInvocation = false; - lastInvocation = Clock::now(); - function(); - }); - } -} - -} // namespace util -} // namespace mbgl diff --git a/src/mbgl/util/throttler.hpp b/src/mbgl/util/throttler.hpp index 175de7ccaf7..0424c1c2a55 100644 --- a/src/mbgl/util/throttler.hpp +++ b/src/mbgl/util/throttler.hpp @@ -1,21 +1,49 @@ +#include +#include +#include #include -#include namespace mbgl { namespace util { +/** + * Throttles invocations to the Delegate Actor, throttling is done + * over all methods, not on a per method base. + */ +template class Throttler { + public: - Throttler(Duration frequency, std::function&& function); + Throttler(ActorRef delegate_, Duration frequency_) + : delegate(std::move(delegate_)) + , frequency(frequency_) { + } - void invoke(); + template + void invoke(Fn fn, Args&&... args) { + if (pendingInvocation && !pendingInvocation->isFinished()) { + return; + } + + Duration timeToNextInvocation = lastInvocation == TimePoint::min() + ? Duration::zero() + : (lastInvocation + frequency) - Clock::now(); + + if (timeToNextInvocation <= Duration::zero()) { + lastInvocation = Clock::now(); + delegate.invoke(fn, std::forward(args)...); + } else { + pendingInvocation = delegate.invokeDelayed(timeToNextInvocation, fn, std::forward(args)...); + lastInvocation = Clock::now() + timeToNextInvocation; + } + } + private: + ActorRef delegate; Duration frequency; - std::function function; - Timer timer; - bool pendingInvocation; - TimePoint lastInvocation; + std::unique_ptr pendingInvocation; + TimePoint lastInvocation = TimePoint::min(); }; } // namespace util diff --git a/test/actor/actor.test.cpp b/test/actor/actor.test.cpp index 4c7fc3666d9..5f85982b5ee 100644 --- a/test/actor/actor.test.cpp +++ b/test/actor/actor.test.cpp @@ -92,6 +92,11 @@ TEST(Actor, DestructionBlocksOnSend) { std::this_thread::sleep_for(1ms); waited = true; } + + std::unique_ptr schedule(Duration, std::weak_ptr, std::unique_ptr) final { + assert(false); + return {}; + } }; struct Test { diff --git a/test/util/throttler.test.cpp b/test/util/throttler.test.cpp new file mode 100644 index 00000000000..52163c6ef8e --- /dev/null +++ b/test/util/throttler.test.cpp @@ -0,0 +1,105 @@ +#include +#include +#include +#include + +#include + +#include +#include +#include +#include + +using namespace mbgl; +using namespace mbgl::util; +using namespace std::chrono_literals; + + +TEST(Throttler, Basic) { + // Ensure the delegate is called + + // Throttler needs a RunLoop + RunLoop loop; + + struct Test { + std::promise promise; + + Test(ActorRef, std::promise promise_) : promise(std::move(promise_)) {} + + void doIt() { + promise.set_value(); + } + }; + + std::promise promise; + std::future future = promise.get_future(); + + ThreadPool pool { 2 }; + Actor test(pool, std::move(promise)); + + Throttler throttler { test.self(), Milliseconds { 300 } }; + + throttler.invoke(&Test::doIt); + + future.wait(); + +} + +TEST(Throttler, Throttling) { + // Ensure invocations are throttled + + RunLoop loop; + + static Milliseconds frequency { 200 }; + + struct Test { + TimePoint lastInvocation = TimePoint::min(); + int& invocations; + + Test(ActorRef, int& invocations_) : invocations(invocations_) { } + + void doIt() { + invocations++; + ASSERT_GE(std::chrono::time_point_cast(Clock::now() - (frequency * .9)).time_since_epoch().count(), + std::chrono::time_point_cast(lastInvocation).time_since_epoch().count()); + lastInvocation = Clock::now(); + } + }; + + std::promise promise; + std::future future = promise.get_future(); + + int invocations = 0; + Actor test(loop, invocations); + + Throttler throttler { test.self(), frequency }; + + // Invoke three times in a row + throttler.invoke(&Test::doIt); // Will be executed immediately + throttler.invoke(&Test::doIt); // Will be executed delayed + throttler.invoke(&Test::doIt); // Will not be executed as there is a pending execution + + // Let the actor process it's mailbox do it's thing + auto start = Clock::now(); + while (Clock::now() < start + std::chrono::milliseconds(100)) { + loop.runOnce(); + std::this_thread::sleep_for (std::chrono::milliseconds(5)); + } + + // Should only see the initial invocation + ASSERT_EQ(invocations, 1); + + // Let the runloop process the pending execution + while (invocations < 2) { + std::this_thread::sleep_for (std::chrono::milliseconds(5)); + loop.runOnce(); + } + + // Another invocation should be executed immediately + throttler.invoke(&Test::doIt); + while (invocations < 3) { + std::this_thread::sleep_for (std::chrono::milliseconds(5)); + loop.runOnce(); + } + +}