Throttler / Timer ; use current scheduler

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

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)

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

include/mbgl/actor/actor_ref.hpp

@@ -2,6 +2,7 @@
 
 #include <mbgl/actor/mailbox.hpp>
 #include <mbgl/actor/message.hpp>
+#include <mbgl/actor/scheduler.hpp>
 
 #include <memory>
 
@@ -55,6 +56,11 @@ class ActorRef {
 
         return future;
     }
+
+    template <typename Fn, class... Args>
+    std::unique_ptr<Scheduler::Scheduled> invokeDelayed(Duration timeout, Fn fn, Args&&... args, Scheduler& scheduler = *Scheduler::GetCurrent()) {
+        return scheduler.schedule(timeout, weakMailbox, actor::makeMessage(*object, fn, std::forward<Args>(args)...));
+    }
 
 private:
     Object* object;

include/mbgl/actor/scheduler.hpp

@@ -1,10 +1,14 @@
 #pragma once
 
+#include <mbgl/util/chrono.hpp>
+#include <mbgl/util/async_request.hpp>
+
 #include <memory>
 
 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<Mailbox>) = 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<Scheduled> schedule(Duration timeout, std::weak_ptr<Mailbox>, std::unique_ptr<Message>) = 0;
+
     // Set/Get the current Scheduler for this thread
     static Scheduler* GetCurrent();
     static void SetCurrent(Scheduler*);

include/mbgl/util/run_loop.hpp

@@ -62,12 +62,14 @@ class RunLoop : public Scheduler,
         push(task);
         return std::make_unique<WorkRequest>(task);
     }
-                    
+
     void schedule(std::weak_ptr<Mailbox> mailbox) override {
         invoke([mailbox] () {
             Mailbox::maybeReceive(mailbox);
         });
     }
+
+    std::unique_ptr<Scheduled> schedule(Duration, std::weak_ptr<Mailbox>, std::unique_ptr<Message>) override;
 
     class Impl;
 

platform/android/src/run_loop.cpp

@@ -1,9 +1,9 @@
 #include "run_loop_impl.hpp"
 
 #include <mbgl/util/platform.hpp>
-#include <mbgl/util/thread_local.hpp>
 #include <mbgl/util/thread.hpp>
 #include <mbgl/util/timer.hpp>
+#include <mbgl/util/scheduled_timer.hpp>
 #include <mbgl/actor/scheduler.hpp>
 
 #include <android/looper.h>
@@ -200,7 +200,7 @@ Milliseconds RunLoop::Impl::processRunnables() {
 }
 
 RunLoop* RunLoop::Get() {
-    assert(static_cast<RunLoop*>(Scheduler::GetCurrent()));
+    assert(dynamic_cast<RunLoop*>(Scheduler::GetCurrent()));
     return static_cast<RunLoop*>(Scheduler::GetCurrent());
 }
 
@@ -258,5 +258,9 @@ void RunLoop::removeWatch(int) {
     throw std::runtime_error("Not implemented.");
 }
 
+std::unique_ptr<Scheduler::Scheduled> RunLoop::schedule(Duration timeout, std::weak_ptr<Mailbox> mailbox, std::unique_ptr<Message> message) {
+    return std::make_unique<ScheduledTimer>(*this, timeout, std::move(mailbox), std::move(message));
+}
+
 } // namespace util
 } // namespace mbgl

platform/darwin/src/run_loop.cpp

@@ -1,6 +1,11 @@
 #include <mbgl/util/run_loop.hpp>
 #include <mbgl/util/async_task.hpp>
+#include <mbgl/util/timer.hpp>
+#include <mbgl/util/platform.hpp>
+#include <mbgl/actor/mailbox.hpp>
+#include <mbgl/actor/message.hpp>
 #include <mbgl/actor/scheduler.hpp>
+#include <mbgl/util/scheduled_timer.hpp>
 
 #include <CoreFoundation/CoreFoundation.h>
 
@@ -13,7 +18,7 @@ class RunLoop::Impl {
 };
 
 RunLoop* RunLoop::Get() {
-    assert(static_cast<RunLoop*>(Scheduler::GetCurrent()));
+    assert(dynamic_cast<RunLoop*>(Scheduler::GetCurrent()));
     return static_cast<RunLoop*>(Scheduler::GetCurrent());
 }
 
@@ -45,6 +50,10 @@ void RunLoop::runOnce() {
 void RunLoop::stop() {
     invoke([&] { CFRunLoopStop(CFRunLoopGetCurrent()); });
 }
+
+std::unique_ptr<Scheduler::Scheduled> RunLoop::schedule(Duration timeout, std::weak_ptr<Mailbox> mailbox, std::unique_ptr<Message> message) {
+    return std::make_unique<ScheduledTimer>(*this, timeout, std::move(mailbox), std::move(message));
+}
 
 } // namespace util
 } // namespace mbgl

platform/default/mbgl/util/default_thread_pool.cpp

@@ -1,14 +1,32 @@
 #include <mbgl/util/default_thread_pool.hpp>
-#include <mbgl/actor/mailbox.hpp>
 #include <mbgl/util/platform.hpp>
 #include <mbgl/util/string.hpp>
+#include <mbgl/util/scheduled_timer.hpp>
 
 namespace mbgl {
+
 
 ThreadPool::ThreadPool(std::size_t count) {
+    std::promise<void> 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> mailbox) {
@@ -53,5 +76,10 @@ void ThreadPool::schedule(std::weak_ptr<Mailbox> mailbox) {
 
     cv.notify_one();
 }
+
+
+std::unique_ptr<Scheduler::Scheduled> ThreadPool::schedule(Duration timeout, std::weak_ptr<Mailbox> mailbox, std::unique_ptr<Message> message) {
+    return std::make_unique<ScheduledTimer>(*timerLoop, timeout, std::move(mailbox), std::move(message));
+}
 
 } // namespace mbgl

platform/default/mbgl/util/default_thread_pool.hpp

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <mbgl/actor/scheduler.hpp>
+#include <mbgl/util/run_loop.hpp>
 
 #include <condition_variable>
 #include <mutex>
@@ -15,13 +16,19 @@ class ThreadPool : public Scheduler {
     ~ThreadPool() override;
 
     void schedule(std::weak_ptr<Mailbox>) override;
+
+    std::unique_ptr<Scheduled> schedule(Duration, std::weak_ptr<Mailbox>, std::unique_ptr<Message>) override;
 
 private:
     std::vector<std::thread> threads;
     std::queue<std::weak_ptr<Mailbox>> queue;
     std::mutex mutex;
     std::condition_variable cv;
     bool terminate { false };
+
+    // Dedicated thread for timers
+    std::thread timerThread;
+    util::RunLoop* timerLoop;
 };
 
 } // namespace mbgl

platform/default/run_loop.cpp

@@ -1,6 +1,7 @@
 #include <mbgl/util/run_loop.hpp>
 #include <mbgl/util/async_task.hpp>
 #include <mbgl/util/thread_local.hpp>
+#include <mbgl/util/scheduled_timer.hpp>
 #include <mbgl/actor/scheduler.hpp>
 
 #include <uv.h>
@@ -215,6 +216,10 @@ void RunLoop::removeWatch(int fd) {
 
     uv_close(reinterpret_cast<uv_handle_t*>(&watch->poll), &Watch::onClose);
 }
+
+std::unique_ptr<Scheduler::Scheduled> RunLoop::schedule(Duration timeout, std::weak_ptr<Mailbox> mailbox, std::unique_ptr<Message> message) {
+    return std::make_unique<ScheduledTimer>(*this, timeout, std::move(mailbox), std::move(message));
+}
 
 } // namespace util
 } // namespace mbgl

platform/node/src/node_thread_pool.hpp

@@ -18,6 +18,12 @@ class NodeThreadPool : public mbgl::Scheduler {
     ~NodeThreadPool();
 
     void schedule(std::weak_ptr<mbgl::Mailbox>) override;
+
+    std::unique_ptr<Scheduled> schedule(mbgl::Duration, std::weak_ptr<mbgl::Mailbox>, std::unique_ptr<mbgl::Message>) override {
+        // Not implemented
+        assert(false);
+        return {};
+    };
 
 private:
     util::AsyncQueue<std::weak_ptr<mbgl::Mailbox>>* queue;

platform/qt/src/run_loop.cpp

@@ -1,6 +1,7 @@
 #include "run_loop_impl.hpp"
 
 #include <mbgl/actor/scheduler.hpp>
+#include <mbgl/util/scheduled_timer.hpp>
 
 #include <QCoreApplication>
 
@@ -119,6 +120,10 @@ void RunLoop::removeWatch(int fd) {
         impl->readPoll.erase(readPollIter);
     }
 }
+
+std::unique_ptr<Scheduler::Scheduled> RunLoop::schedule(Duration timeout, std::weak_ptr<Mailbox> mailbox, std::unique_ptr<Message> message) {
+    return std::make_unique<ScheduledTimer>(*this, timeout, std::move(mailbox), std::move(message));
+}
 
 }
 }

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<GeometryTile>(*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() {

src/mbgl/tile/geometry_tile.hpp

@@ -141,7 +141,7 @@ class GeometryTile : public Tile, public GlyphRequestor, ImageRequestor {
     std::unordered_map<std::string, std::shared_ptr<Bucket>> symbolBuckets;
     std::unique_ptr<CollisionTile> collisionTile;
 
-    util::Throttler placementThrottler;
+    util::Throttler<GeometryTile> placementThrottler;
     float lastYStretch;
 
 public: