[core] Introduce Scheduler::GetSequenced() API

The newly introduced `Scheduler::GetSequenced()` returns
sequenced schedulers from the cache limited to 10 instances,
preventing from spawning too many threads.

include/mbgl/actor/scheduler.hpp

@@ -69,8 +69,21 @@ public:
     // Get the scheduler for asynchronous tasks. This method
     // will lazily initialize a shared worker pool when ran
     // from the first time.
+    // The scheduled tasks might run in parallel on different
+    // threads.
+    // TODO : Rename to GetPool()
     static std::shared_ptr<Scheduler> GetBackground();
 
+    // Get the *sequenced* scheduler for asynchronous tasks.
+    // Unlike the method above, the returned scheduler
+    // (once stored) represents a single thread, thus each
+    // newly scheduled task is guarantied to run after the
+    // previously scheduled one.
+    //
+    // Sequenced scheduler can be used for running tasks
+    // on the same thread-unsafe object.
+    static std::shared_ptr<Scheduler> GetSequenced();
+
 protected:
     template <typename TaskFn, typename ReplyFn>
     void scheduleAndReplyValue(const TaskFn& task,

src/mbgl/actor/scheduler.cpp

@@ -41,4 +41,31 @@ std::shared_ptr<Scheduler> Scheduler::GetBackground() {
     return scheduler;
 }
 
+// static
+std::shared_ptr<Scheduler> Scheduler::GetSequenced() {
+    const std::size_t kSchedulersCount = 10;
+    static std::vector<std::weak_ptr<Scheduler>> weaks(kSchedulersCount);
+    static std::mutex mtx;
+    static std::size_t lastUsedIndex = 0u;
+
+    std::lock_guard<std::mutex> lock(mtx);
+
+    if (++lastUsedIndex == kSchedulersCount) lastUsedIndex = 0u;
+
+    std::shared_ptr<Scheduler> result;
+    for (std::size_t i = 0; i < kSchedulersCount; ++i) {
+        auto& weak = weaks[i];
+        if (auto scheduler = weak.lock()) {
+            if (lastUsedIndex == i) result = scheduler;
+            continue;
+        }
+        result = std::make_shared<SequencedScheduler>();
+        weak = result;
+        lastUsedIndex = i;
+        break;
+    }
+
+    return result;
+}
+
 } //namespace mbgl

test/util/async_task.test.cpp

@@ -169,4 +169,47 @@ TEST(AsyncTask, scheduleAndReplyValue) {
     auto sheduler = Scheduler::GetBackground();
     sheduler->scheduleAndReplyValue(runInBackground, onResult);
     loop.run();
+}
+
+TEST(AsyncTask, SequencedScheduler) {
+    RunLoop loop;
+    std::thread::id caller_id = std::this_thread::get_id();
+    std::thread::id bg_id;
+    int count = 0;
+
+    auto first = [caller_id, &bg_id, &count]() {
+        EXPECT_EQ(0, count);
+        bg_id = std::this_thread::get_id();
+        EXPECT_NE(caller_id, bg_id);
+        count++;
+    };
+    auto second = [&bg_id, &count]() {
+        EXPECT_EQ(1, count);
+        EXPECT_EQ(bg_id, std::this_thread::get_id());
+        count++;
+    };
+    auto third = [&bg_id, &count, &loop]() {
+        EXPECT_EQ(2, count);
+        EXPECT_EQ(bg_id, std::this_thread::get_id());
+        loop.stop();
+    };
+
+    auto sheduler = Scheduler::GetSequenced();
+
+    sheduler->schedule(first);
+    sheduler->schedule(second);
+    sheduler->schedule(third);
+    loop.run();
+}
+
+TEST(AsyncTask, MultipleSequencedSchedulers) {
+    std::vector<std::shared_ptr<Scheduler>> shedulers;
+
+    for (int i = 0; i < 10; ++i) {
+        auto scheduler = Scheduler::GetSequenced();
+        EXPECT_TRUE(std::none_of(
+            shedulers.begin(), shedulers.end(), [&scheduler](const auto &item) { return item == scheduler; }));
+        shedulers.emplace_back(std::move(scheduler));
+    }
+    EXPECT_EQ(shedulers.front(), Scheduler::GetSequenced());
 }