Asynchronous GeometryTile deletion (#2051)

Co-authored-by: Tim Sylvester <tim@wetdogweather.com>

src/mbgl/util/thread_pool.cpp

@@ -2,6 +2,7 @@
 
 #include <mbgl/platform/settings.hpp>
 #include <mbgl/platform/thread.hpp>
+#include <mbgl/util/monotonic_timer.hpp>
 #include <mbgl/util/platform.hpp>
 #include <mbgl/util/string.hpp>
 
@@ -10,11 +11,16 @@ namespace mbgl {
 ThreadedSchedulerBase::~ThreadedSchedulerBase() = default;
 
 void ThreadedSchedulerBase::terminate() {
+    // Run any leftover render jobs
+    runRenderJobs();
+
     {
         std::lock_guard<std::mutex> lock(mutex);
         terminated = true;
     }
-    cv.notify_all();
+
+    // Wake up all threads so that they shut down
+    cvAvailable.notify_all();
 }
 
 std::thread ThreadedSchedulerBase::makeSchedulerThread(size_t index) {
@@ -25,13 +31,18 @@ std::thread ThreadedSchedulerBase::makeSchedulerThread(size_t index) {
             platform::setCurrentThreadPriority(*priority);
         }
 
-        platform::setCurrentThreadName(std::string{"Worker "} + util::toString(index + 1));
+        platform::setCurrentThreadName("Worker " + util::toString(index + 1));
         platform::attachThread();
 
+        owningThreadPool.set(this);
+
         while (true) {
             std::unique_lock<std::mutex> lock(mutex);
+            if (queue.empty() && !pendingItems) {
+                cvEmpty.notify_all();
+            }
 
-            cv.wait(lock, [this] { return !queue.empty() || terminated; });
+            cvAvailable.wait(lock, [this] { return !queue.empty() || terminated; });
 
             if (terminated) {
                 platform::detachThread();
@@ -40,20 +51,83 @@ std::thread ThreadedSchedulerBase::makeSchedulerThread(size_t index) {
 
             auto function = std::move(queue.front());
             queue.pop();
+
+            if (function) {
+                pendingItems++;
+            }
+
             lock.unlock();
-            if (function) function();
+
+            if (function) {
+                const auto cleanup = [&] {
+                    // destroy the function and release its captures before unblocking `waitForEmpty`
+                    function = {};
+                    pendingItems--;
+                    if (queue.empty() && !pendingItems) {
+                        cvEmpty.notify_all();
+                    }
+                };
+                try {
+                    function();
+                    cleanup();
+                } catch (...) {
+                    lock.lock();
+                    if (handler) {
+                        handler(std::current_exception());
+                    }
+                    cleanup();
+                    if (handler) {
+                        continue;
+                    }
+                    throw;
+                }
+            }
         }
     });
 }
 
-void ThreadedSchedulerBase::schedule(std::function<void()> fn) {
+void ThreadedSchedulerBase::schedule(std::function<void()>&& fn) {
     assert(fn);
-    {
-        std::lock_guard<std::mutex> lock(mutex);
-        queue.push(std::move(fn));
+    if (fn) {
+        {
+            // We need to block if adding adding a new task from a thread not controlled by this
+            // pool.  Tasks are added by other tasks, so we must not block a thread we do control
+            // or `waitForEmpty` will deadlock.
+            std::unique_lock<std::mutex> addLock(addMutex, std::defer_lock);
+            if (!thisThreadIsOwned()) {
+                addLock.lock();
+            }
+            std::lock_guard<std::mutex> lock(mutex);
+            queue.push(std::move(fn));
+        }
+        cvAvailable.notify_one();
     }
+}
 
-    cv.notify_one();
+std::size_t ThreadedSchedulerBase::waitForEmpty(Milliseconds timeout) {
+    // Must not be called from a thread in our pool, or we would deadlock
+    assert(!thisThreadIsOwned());
+    if (!thisThreadIsOwned()) {
+        const auto startTime = util::MonotonicTimer::now();
+        const auto isDone = [&] {
+            return queue.empty() && pendingItems == 0;
+        };
+        // Block any other threads from adding new items
+        std::scoped_lock<std::mutex> addLock(addMutex);
+        std::unique_lock<std::mutex> lock(mutex);
+        while (!isDone()) {
+            if (timeout > Milliseconds::zero()) {
+                const auto elapsed = util::MonotonicTimer::now() - startTime;
+                if (timeout <= elapsed || !cvEmpty.wait_for(lock, timeout - elapsed, isDone)) {
+                    break;
+                }
+            } else {
+                cvEmpty.wait(lock, isDone);
+            }
+        }
+        return queue.size() + pendingItems;
+    }
+    return 0;
 }
 
 } // namespace mbgl