Convert DefaultFileSource and dependents to actor model

benchmark/api/query.benchmark.cpp

@@ -35,8 +35,8 @@ class QueryBenchmark {
     util::RunLoop loop;
     HeadlessBackend backend;
     OffscreenView view{ backend.getContext(), { 1000, 1000 } };
-    DefaultFileSource fileSource{ "benchmark/fixtures/api/cache.db", "." };
-    ThreadPool threadPool{ 4 };
+    ThreadPool threadPool{ 4, { "Worker" } };
+    DefaultFileSource fileSource{ threadPool, "benchmark/fixtures/api/cache.db", "." };
     Map map{ backend, view.size, 1, fileSource, threadPool, MapMode::Still };
     ScreenBox box{{ 0, 0 }, { 1000, 1000 }};
 };

bin/offline.cpp

@@ -1,5 +1,6 @@
 #include <mbgl/util/default_styles.hpp>
 #include <mbgl/util/run_loop.hpp>
+#include <mbgl/util/default_thread_pool.hpp>
 #include <mbgl/util/string.hpp>
 #include <mbgl/util/io.hpp>
 
@@ -55,7 +56,8 @@ int main(int argc, char *argv[]) {
     using namespace mbgl;
 
     util::RunLoop loop;
-    DefaultFileSource fileSource(output, ".");
+    ThreadPool threadPool{ 2, { "Worker" } };
+    DefaultFileSource fileSource(threadPool, output, ".");
     std::unique_ptr<OfflineRegion> region;
 
     fileSource.setAccessToken(token);

bin/render.cpp

@@ -69,7 +69,8 @@ int main(int argc, char *argv[]) {
     using namespace mbgl;
 
     util::RunLoop loop;
-    DefaultFileSource fileSource(cache_file, asset_root);
+    ThreadPool threadPool(4, { "Worker" });
+    DefaultFileSource fileSource(threadPool, cache_file, asset_root);
 
     // Try to load the token from the environment.
     if (!token.size()) {
@@ -86,7 +87,6 @@ int main(int argc, char *argv[]) {
 
     HeadlessBackend backend;
     OffscreenView view(backend.getContext(), { width * pixelRatio, height * pixelRatio });
-    ThreadPool threadPool(4);
     Map map(backend, mbgl::Size { width, height }, pixelRatio, fileSource, threadPool, MapMode::Still);
 
     if (util::isURL(style_path)) {

include/mbgl/storage/default_file_source.hpp

@@ -8,6 +8,8 @@
 
 namespace mbgl {
 
+class Scheduler;
+
 namespace util {
 template <typename T> class Thread;
 } // namespace util
@@ -21,7 +23,8 @@ class DefaultFileSource : public FileSource {
      * regions, we want the database to remain fairly small (order tens or low hundreds
      * of megabytes).
      */
-    DefaultFileSource(const std::string& cachePath,
+    DefaultFileSource(Scheduler&,
+                      const std::string& cachePath,
                       const std::string& assetRoot,
                       uint64_t maximumCacheSize = util::DEFAULT_MAX_CACHE_SIZE);
     ~DefaultFileSource() override;

platform/android/src/asset_file_source.cpp

@@ -1,9 +1,13 @@
 #include <mbgl/storage/asset_file_source.hpp>
 #include <mbgl/storage/response.hpp>
+#include <mbgl/util/run_loop.hpp>
 #include <mbgl/util/util.hpp>
 #include <mbgl/util/thread.hpp>
 #include <mbgl/util/url.hpp>
 
+#include <mbgl/actor/actor.hpp>
+#include <mbgl/actor/actor_ref.hpp>
+
 #pragma GCC diagnostic push
 #pragma GCC diagnostic ignored "-Wshadow"
 #include <zip.h>
@@ -37,20 +41,21 @@ namespace mbgl {
 
 class AssetFileRequest : public AsyncRequest {
 public:
-    AssetFileRequest(std::unique_ptr<WorkRequest> workRequest_)
-        : workRequest(std::move(workRequest_)) {
+    AssetFileRequest(std::shared_ptr<FileSource::Callback> callback_)
+        : callback(std::move(callback_)) {
     }
 
-    std::unique_ptr<WorkRequest> workRequest;
+private:
+    std::shared_ptr<FileSource::Callback> callback;
 };
 
-class AssetFileSource::Impl {
+class AssetFileSource::Worker {
 public:
-    Impl(const std::string& root_)
-        : root(root_) {
+    Worker(ActorRef<Worker>, ActorRef<AssetFileSource> parent_, std::string root_)
+        : parent(std::move(parent_)), root(std::move(root_)) {
     }
 
-    void request(const std::string& url, FileSource::Callback callback) {
+    void readFile(const std::string& url, std::weak_ptr<Callback> callback) {
         ZipHolder archive = ::zip_open(root.c_str(), 0, nullptr);
         if (!archive.archive) {
             reportError(Response::Error::Reason::Other, "Could not open zip archive", callback);
@@ -85,29 +90,55 @@ class AssetFileSource::Impl {
 
         Response response;
         response.data = buf;
-        callback(response);
+
+        parent.invoke(&AssetFileSource::respond, callback, response);
     }
 
-    void reportError(Response::Error::Reason reason, const char * message, FileSource::Callback callback) {
+    void reportError(Response::Error::Reason reason,
+                     const char* message,
+                     std::weak_ptr<Callback> callback) {
         Response response;
         response.error = std::make_unique<Response::Error>(reason, message);
-        callback(response);
+        parent.invoke(&AssetFileSource::respond, callback, response);
     }
 
 private:
+    ActorRef<AssetFileSource> parent;
     std::string root;
 };
 
-AssetFileSource::AssetFileSource(const std::string& root)
-    : thread(std::make_unique<util::Thread<Impl>>(
-        util::ThreadContext{"AssetFileSource", util::ThreadPriority::Low},
-        root)) {
+// AssetFileSource::AssetFileSource(const std::string& root)
+//     : thread(std::make_unique<util::Thread<Impl>>(
+//         util::ThreadContext{"AssetFileSource", util::ThreadPriority::Low},
+//         root)) {
+// }
+
+// AssetFileSource::~AssetFileSource() = default;
+
+// std::unique_ptr<AsyncRequest> AssetFileSource::request(const Resource& resource, Callback callback) {
+//     return thread->invokeWithCallback(&Impl::request, resource.url, callback);
+// }
+
+
+AssetFileSource::AssetFileSource(Scheduler& scheduler, const std::string& root)
+    : mailbox(std::make_shared<Mailbox>(*util::RunLoop::Get())),
+      worker(std::make_unique<Actor<Worker>>(
+          scheduler, ActorRef<AssetFileSource>(*this, mailbox), root)) {
 }
 
 AssetFileSource::~AssetFileSource() = default;
 
-std::unique_ptr<AsyncRequest> AssetFileSource::request(const Resource& resource, Callback callback) {
-    return thread->invokeWithCallback(&Impl::request, resource.url, callback);
+std::unique_ptr<AsyncRequest> AssetFileSource::request(const Resource& resource,
+                                                       Callback callback) {
+    auto cb = std::make_shared<Callback>(std::move(callback));
+    worker->invoke(&Worker::readFile, resource.url, cb);
+    return std::make_unique<AssetFileRequest>(std::move(cb));
+}
+
+void AssetFileSource::respond(std::weak_ptr<Callback> callback, Response response) {
+    if (auto locked = callback.lock()) {
+        (*locked)(response);
+    }
 }
 
 }

platform/android/src/jni.cpp

@@ -11,6 +11,7 @@
 #include "jni.hpp"
 #include "java_types.hpp"
 #include "native_map_view.hpp"
+#include "shared_thread_pool.hpp"
 #include "connectivity_listener.hpp"
 #include "style/layers/layers.hpp"
 #include "style/sources/sources.hpp"
@@ -1236,7 +1237,8 @@ void nativeScheduleTakeSnapshot(JNIEnv *env, jni::jobject* obj, jlong nativeMapV
 jlong createDefaultFileSource(JNIEnv *env, jni::jobject* obj, jni::jstring* cachePath_, jni::jstring* assetRoot_, jlong maximumCacheSize) {
     std::string cachePath = std_string_from_jstring(env, cachePath_);
     std::string assetRoot = std_string_from_jstring(env, assetRoot_);
-    mbgl::DefaultFileSource *defaultFileSource = new mbgl::DefaultFileSource(cachePath, assetRoot, maximumCacheSize);
+    mbgl::DefaultFileSource* defaultFileSource = new mbgl::DefaultFileSource(
+        mbgl::android::sharedThreadPool(), cachePath, assetRoot, maximumCacheSize);
     jlong defaultFileSourcePtr = reinterpret_cast<jlong>(defaultFileSource);
     return defaultFileSourcePtr;
 }

platform/android/src/native_map_view.cpp

@@ -1,5 +1,6 @@
 #include "native_map_view.hpp"
 #include "jni.hpp"
+#include "shared_thread_pool.hpp"
 
 #include <cstdlib>
 #include <ctime>
@@ -40,8 +41,7 @@ void log_egl_string(EGLDisplay display, EGLint name, const char *label) {
 NativeMapView::NativeMapView(JNIEnv *env_, jobject obj_, float pixelRatio, int availableProcessors_, size_t totalMemory_)
     : env(env_),
       availableProcessors(availableProcessors_),
-      totalMemory(totalMemory_),
-      threadPool(4) {
+      totalMemory(totalMemory_) {
     mbgl::Log::Debug(mbgl::Event::Android, "NativeMapView::NativeMapView");
 
     assert(env_ != nullptr);
@@ -59,12 +59,13 @@ NativeMapView::NativeMapView(JNIEnv *env_, jobject obj_, float pixelRatio, int a
     }
 
     fileSource = std::make_unique<mbgl::DefaultFileSource>(
+        sharedThreadPool(),
         mbgl::android::cachePath + "/mbgl-offline.db",
         mbgl::android::apkPath);
 
     map = std::make_unique<mbgl::Map>(
         *this, mbgl::Size{ static_cast<uint32_t>(width), static_cast<uint32_t>(height) },
-        pixelRatio, *fileSource, threadPool, MapMode::Continuous);
+        pixelRatio, *fileSource, sharedThreadPool(), MapMode::Continuous);
 
     float zoomFactor   = map->getMaxZoom() - map->getMinZoom() + 1;
     float cpuFactor    = availableProcessors;

platform/android/src/native_map_view.hpp

@@ -4,7 +4,6 @@
 #include <mbgl/map/view.hpp>
 #include <mbgl/map/backend.hpp>
 #include <mbgl/util/noncopyable.hpp>
-#include <mbgl/util/default_thread_pool.hpp>
 #include <mbgl/storage/default_file_source.hpp>
 
 #include <string>
@@ -97,7 +96,6 @@ class NativeMapView : public mbgl::View, public mbgl::Backend {
 
     // Ensure these are initialised last
     std::unique_ptr<mbgl::DefaultFileSource> fileSource;
-    mbgl::ThreadPool threadPool;
     std::unique_ptr<mbgl::Map> map;
     mbgl::EdgeInsets insets;
 

platform/android/src/shared_thread_pool.hpp

@@ -0,0 +1,15 @@
+#pragma once
+
+#include <mbgl/util/default_thread_pool.hpp>
+
+namespace mbgl {
+namespace android {
+
+// May only be called from the main/UI thread.
+inline ThreadPool& sharedThreadPool() {
+    static ThreadPool threadPool{ 4, { "Worker" } };
+    return threadPool;
+}
+
+}
+}

platform/darwin/src/MGLOfflineStorage.mm

@@ -5,6 +5,7 @@
 #import "MGLNetworkConfiguration.h"
 #import "MGLOfflinePack_Private.h"
 #import "MGLOfflineRegion_Private.h"
+#import "MGLThreadPool_Private.h"
 #import "MGLTilePyramidOfflineRegion.h"
 #import "NSValue+MGLAdditions.h"
 
@@ -130,7 +131,9 @@ - (instancetype)init {
             [[NSFileManager defaultManager] moveItemAtPath:subdirectorylessCacheURL.path toPath:cachePath error:NULL];
         }
 
-        _mbglFileSource = new mbgl::DefaultFileSource(cachePath.UTF8String, [NSBundle mainBundle].resourceURL.path.UTF8String);
+        _mbglFileSource = new mbgl::DefaultFileSource(
+            *[MGLThreadPool sharedPool].mbglThreadPool, cachePath.UTF8String,
+            [NSBundle mainBundle].resourceURL.path.UTF8String);
 
         // Observe for changes to the API base URL (and find out the current one).
         [[MGLNetworkConfiguration sharedManager] addObserver:self

platform/darwin/src/MGLThreadPool.mm

@@ -0,0 +1,39 @@
+#import "MGLThreadPool_Private.h"
+
+#import "NSProcessInfo+MGLAdditions.h"
+
+@interface MGLThreadPool ()
+
+@property (nonatomic) mbgl::ThreadPool *mbglThreadPool;
+
+@end
+
+@implementation MGLThreadPool
+
+#pragma mark - Internal
+
++ (instancetype)sharedPool {
+    if (NSProcessInfo.processInfo.mgl_isInterfaceBuilderDesignablesAgent) {
+        return nil;
+    }
+    static dispatch_once_t onceToken;
+    static MGLThreadPool *sharedThreadPool;
+    dispatch_once(&onceToken, ^{
+        sharedThreadPool = [[self alloc] init];
+    });
+    return sharedThreadPool;
+}
+
+- (instancetype)init {
+    if (self = [super init]) {
+        _mbglThreadPool = new mbgl::ThreadPool(4, { "Worker" });
+    }
+    return self;
+}
+
+- (void)dealloc {
+    delete _mbglThreadPool;
+    _mbglThreadPool = nullptr;
+}
+
+@end

platform/darwin/src/MGLThreadPool_Private.h

@@ -0,0 +1,18 @@
+#import <Foundation/Foundation.h>
+
+#include <mbgl/util/default_thread_pool.hpp>
+
+@interface MGLThreadPool : NSObject
+@end
+
+@interface MGLThreadPool (Private)
+
+/// Returns the shared instance of the `MGLThreadPool` class.
++ (instancetype)sharedPool;
+
+/**
+ The shared thread pool object owned by the shared thread pool object.
+ */
+@property (nonatomic) mbgl::ThreadPool *mbglThreadPool;
+
+@end