Merge pull request #1250 from mapbox/1250-separate-thread-object

Separate objects from Thread management

Makefile

@@ -53,6 +53,7 @@ Xcode/mbgl: config/$(HOST).gypi styles/styles SMCalloutView
 Makefile/test: test/test.gyp config/$(HOST).gypi styles/styles SMCalloutView
 	deps/run_gyp test/test.gyp $(CONFIG_$(HOST)) $(LIBS_$(HOST)) --generator-output=./build/$(HOST) -f make
 
+.PHONY: test
 test: Makefile/test
 	$(MAKE) -C build/$(HOST) BUILDTYPE=$(BUILDTYPE) test
 

bin/render.cpp

@@ -7,7 +7,7 @@
 #include <mbgl/platform/default/headless_display.hpp>
 #include <mbgl/platform/log.hpp>
 #include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 
 #pragma GCC diagnostic push
 #ifndef __clang__

gyp/asset-fs.gypi

@@ -12,6 +12,7 @@
 
       'include_dirs': [
         '../include',
+        '../src',
       ],
 
       'variables': {

gyp/asset-zip.gypi

@@ -13,6 +13,7 @@
 
       'include_dirs': [
         '../include',
+        '../src',
       ],
 
       'variables': {

gyp/http-curl.gypi

@@ -12,6 +12,7 @@
 
       'include_dirs': [
         '../include',
+        '../src',
       ],
 
       'variables': {

gyp/http-nsurl.gypi

@@ -12,6 +12,7 @@
 
       'include_dirs': [
         '../include',
+        '../src',
       ],
 
       'variables': {

gyp/platform-osx.gypi

@@ -26,6 +26,7 @@
           '<@(uv_static_libs)',
         ],
         'ldflags': [
+          '-framework Foundation',
           '-framework ImageIO',
           '-framework CoreServices',
           '-framework OpenGL',

include/mbgl/android/native_map_view.hpp

@@ -4,7 +4,7 @@
 #include <mbgl/map/map.hpp>
 #include <mbgl/map/view.hpp>
 #include <mbgl/util/noncopyable.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 #include <mbgl/storage/default_file_source.hpp>
 
 #include <string>

include/mbgl/storage/default/sqlite_cache.hpp

@@ -1,54 +0,0 @@
-#ifndef MBGL_STORAGE_DEFAULT_SQLITE_CACHE
-#define MBGL_STORAGE_DEFAULT_SQLITE_CACHE
-
-#include <mbgl/storage/file_cache.hpp>
-
-#include <string>
-#include <thread>
-
-typedef struct uv_loop_s uv_loop_t;
-
-namespace mapbox { namespace util { template<typename... Types> class variant; } }
-namespace mapbox { namespace sqlite { class Database; class Statement; } }
-
-namespace mbgl {
-
-namespace util { template <typename T> class AsyncQueue; }
-
-class SQLiteCache : public FileCache {
-    struct GetAction;
-    struct PutAction;
-    struct RefreshAction;
-    struct StopAction;
-    using Action = mapbox::util::variant<GetAction, PutAction, RefreshAction, StopAction>;
-    using Queue = util::AsyncQueue<Action>;
-
-public:
-    SQLiteCache(const std::string &path = ":memory:");
-    ~SQLiteCache();
-
-    void get(const Resource &resource, std::function<void(std::unique_ptr<Response>)> callback);
-    void put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint);
-
-private:
-    struct ActionDispatcher;
-    void process(GetAction &action);
-    void process(PutAction &action);
-    void process(RefreshAction &action);
-    void process(StopAction &action);
-
-    void createDatabase();
-    void createSchema();
-
-    const std::string path;
-    uv_loop_t *loop = nullptr;
-    Queue *queue = nullptr;
-    std::thread thread;
-    std::unique_ptr<::mapbox::sqlite::Database> db;
-    std::unique_ptr<::mapbox::sqlite::Statement> getStmt, putStmt, refreshStmt;
-    bool schema = false;
-};
-
-}
-
-#endif

include/mbgl/storage/default_file_source.hpp

@@ -4,24 +4,18 @@
 #include <mbgl/storage/file_source.hpp>
 #include <mbgl/storage/file_cache.hpp>
 
-#include <set>
-#include <unordered_map>
-#include <thread>
-
-namespace mapbox { namespace util { template<typename... Types> class variant; } }
-
 namespace mbgl {
 
-namespace util { template <typename T> class AsyncQueue; }
-
-class SharedRequestBase;
+namespace util {
+template <typename T> class Thread;
+}
 
 class DefaultFileSource : public FileSource {
 public:
     DefaultFileSource(FileCache *cache, const std::string &root = "");
-    DefaultFileSource(FileCache *cache, uv_loop_t *loop, const std::string &root = "");
     ~DefaultFileSource() override;
 
+    // FileSource API
     Request *request(const Resource &resource, uv_loop_t *loop, const Environment &env,
                      Callback callback) override;
     void cancel(Request *request) override;
@@ -29,37 +23,10 @@ public:
 
     void abort(const Environment &env) override;
 
-    void notify(SharedRequestBase *sharedRequest, const std::set<Request *> &observers,
-                std::shared_ptr<const Response> response, FileCache::Hint hint);
-
 public:
-    const std::string assetRoot;
-
+    class Impl;
 private:
-    struct ActionDispatcher;
-    struct AddRequestAction;
-    struct RemoveRequestAction;
-    struct ResultAction;
-    struct StopAction;
-    struct AbortAction;
-    using Action = mapbox::util::variant<AddRequestAction, RemoveRequestAction, ResultAction,
-                                         StopAction, AbortAction>;
-    using Queue = util::AsyncQueue<Action>;
-
-    void process(AddRequestAction &action);
-    void process(RemoveRequestAction &action);
-    void process(ResultAction &action);
-    void process(StopAction &action);
-    void process(AbortAction &action);
-
-    SharedRequestBase *find(const Resource &resource);
-
-    std::unordered_map<Resource, SharedRequestBase *, Resource::Hash> pending;
-
-    uv_loop_t *loop = nullptr;
-    FileCache *cache = nullptr;
-    Queue *queue = nullptr;
-    std::thread thread;
+    const std::unique_ptr<util::Thread<Impl>> thread;
 };
 
 }

include/mbgl/storage/file_cache.hpp

@@ -16,9 +16,9 @@ public:
     virtual ~FileCache() = default;
 
     enum class Hint : uint8_t { Full, Refresh, No };
+    using Callback = std::function<void(std::unique_ptr<Response>)>;
 
-    virtual void get(const Resource &resource,
-                     std::function<void(std::unique_ptr<Response>)> callback) = 0;
+    virtual void get(const Resource &resource, Callback callback) = 0;
     virtual void put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint) = 0;
 };
 

include/mbgl/storage/sqlite_cache.hpp

@@ -0,0 +1,30 @@
+#ifndef MBGL_STORAGE_DEFAULT_SQLITE_CACHE
+#define MBGL_STORAGE_DEFAULT_SQLITE_CACHE
+
+#include <mbgl/storage/file_cache.hpp>
+
+#include <string>
+
+namespace mbgl {
+
+namespace util {
+template <typename T> class Thread;
+}
+
+class SQLiteCache : public FileCache {
+public:
+    SQLiteCache(const std::string &path = ":memory:");
+    ~SQLiteCache() override;
+
+    // FileCache API
+    void get(const Resource &resource, Callback callback) override;
+    void put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint) override;
+
+private:
+    class Impl;
+    const std::unique_ptr<util::Thread<Impl>> thread;
+};
+
+}
+
+#endif

linux/main.cpp

@@ -6,7 +6,7 @@
 #include <mbgl/platform/default/settings_json.hpp>
 #include <mbgl/platform/default/glfw_view.hpp>
 #include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 
 #include <signal.h>
 #include <getopt.h>

macosx/main.mm

@@ -4,7 +4,7 @@
 #include <mbgl/platform/darwin/Reachability.h>
 #include <mbgl/platform/default/glfw_view.hpp>
 #include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 #include <mbgl/storage/network_status.hpp>
 
 #include <mbgl/util/geo.hpp>

platform/darwin/http_request_nsurl.mm

@@ -1,5 +1,5 @@
-#include <mbgl/storage/default/http_request.hpp>
-#include <mbgl/storage/default/http_context.hpp>
+#include <mbgl/storage/http_request.hpp>
+#include <mbgl/storage/http_context.hpp>
 #include <mbgl/storage/response.hpp>
 #include <mbgl/util/uv.hpp>
 
@@ -47,7 +47,7 @@ class HTTPNSURLContext;
 
 class HTTPRequestImpl {
 public:
-    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::unique_ptr<Response> response);
+    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::shared_ptr<const Response> response);
     ~HTTPRequestImpl();
 
     void cancel();
@@ -66,7 +66,7 @@ private:
     HTTPRequest *request = nullptr;
     NSURLSessionDataTask *task = nullptr;
     std::unique_ptr<Response> response;
-    std::unique_ptr<Response> existingResponse;
+    const std::shared_ptr<const Response> existingResponse;
     ResponseStatus status = ResponseStatus::PermanentError;
     uv_async_t *async = nullptr;
     int attempts = 0;
@@ -118,10 +118,10 @@ HTTPNSURLContext::~HTTPNSURLContext() {
 // -------------------------------------------------------------------------------------------------
 
 HTTPRequestImpl::HTTPRequestImpl(HTTPRequest *request_, uv_loop_t *loop,
-                                 std::unique_ptr<Response> existingResponse_)
+                                 std::shared_ptr<const Response> existingResponse_)
     : context(HTTPNSURLContext::Get(loop)),
       request(request_),
-      existingResponse(std::move(existingResponse_)),
+      existingResponse(existingResponse_),
       async(new uv_async_t) {
     assert(request);
     context->addRequest(request);
@@ -326,12 +326,12 @@ void HTTPRequestImpl::handleResult(NSData *data, NSURLResponse *res, NSError *er
 
         if (responseCode == 304) {
             if (existingResponse) {
-                // We're going to reuse the old response object, but need to copy over the new
-                // expires value (if possible).
-                std::swap(response, existingResponse);
-                if (existingResponse->expires) {
-                    response->expires = existingResponse->expires;
-                }
+                // We're going to copy over the existing response's data.
+                response->status = existingResponse->status;
+                response->message = existingResponse->message;
+                response->modified = existingResponse->modified;
+                response->etag = existingResponse->etag;
+                response->data = existingResponse->data;
                 status = ResponseStatus::NotModified;
             } else {
                 // This is an unsolicited 304 response and should only happen on malfunctioning
@@ -397,7 +397,7 @@ void HTTPRequestImpl::restart(uv_timer_t *timer, int) {
 
 // -------------------------------------------------------------------------------------------------
 
-HTTPRequest::HTTPRequest(DefaultFileSource *source, const Resource &resource)
+HTTPRequest::HTTPRequest(DefaultFileSource::Impl *source, const Resource &resource)
     : SharedRequestBase(source, resource) {
 }
 
@@ -409,11 +409,11 @@ HTTPRequest::~HTTPRequest() {
     }
 }
 
-void HTTPRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void HTTPRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     assert(!ptr);
-    ptr = new HTTPRequestImpl(this, loop, std::move(response));
+    ptr = new HTTPRequestImpl(this, loop, response);
 }
 
 void HTTPRequest::retryImmediately() {

platform/default/asset_request_fs.cpp

@@ -1,4 +1,4 @@
-#include <mbgl/storage/default/asset_request.hpp>
+#include <mbgl/storage/asset_request.hpp>
 #include <mbgl/storage/response.hpp>
 #include <mbgl/util/std.hpp>
 #include <mbgl/util/util.hpp>
@@ -207,7 +207,7 @@ void AssetRequestImpl::cleanup(uv_fs_t *req) {
 
 // -------------------------------------------------------------------------------------------------
 
-AssetRequest::AssetRequest(DefaultFileSource *source_, const Resource &resource_)
+AssetRequest::AssetRequest(DefaultFileSource::Impl *source_, const Resource &resource_)
     : SharedRequestBase(source_, resource_) {
     assert(algo::starts_with(resource.url, "asset://"));
 }
@@ -220,7 +220,7 @@ AssetRequest::~AssetRequest() {
     }
 }
 
-void AssetRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void AssetRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     // We're ignoring the existing response if any.

platform/default/asset_request_zip.cpp

@@ -1,5 +1,5 @@
-#include <mbgl/storage/default/asset_request.hpp>
-#include <mbgl/storage/default/thread_context.hpp>
+#include <mbgl/storage/asset_request.hpp>
+#include <mbgl/storage/thread_context.hpp>
 #include <mbgl/android/jni.hpp>
 #include <mbgl/storage/response.hpp>
 #include <mbgl/platform/log.hpp>
@@ -275,7 +275,7 @@ void AssetRequestImpl::cancel() {
 
 // -------------------------------------------------------------------------------------------------
 
-AssetRequest::AssetRequest(DefaultFileSource *source_, const Resource &resource_)
+AssetRequest::AssetRequest(DefaultFileSource::Impl *source_, const Resource &resource_)
     : SharedRequestBase(source_, resource_) {
     assert(algo::starts_with(resource.url, "asset://"));
 }
@@ -288,7 +288,7 @@ AssetRequest::~AssetRequest() {
     }
 }
 
-void AssetRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void AssetRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     // We're ignoring the existing response if any.

platform/default/http_request_curl.cpp

@@ -1,5 +1,5 @@
-#include <mbgl/storage/default/http_request.hpp>
-#include <mbgl/storage/default/http_context.hpp>
+#include <mbgl/storage/http_request.hpp>
+#include <mbgl/storage/http_context.hpp>
 #include <mbgl/storage/response.hpp>
 #include <mbgl/util/chrono.hpp>
 #include <mbgl/platform/log.hpp>
@@ -95,7 +95,7 @@ class HTTPRequestImpl {
     MBGL_STORE_THREAD(tid)
 
 public:
-    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::unique_ptr<Response> response);
+    HTTPRequestImpl(HTTPRequest *request, uv_loop_t *loop, std::shared_ptr<const Response> response);
     ~HTTPRequestImpl();
 
     void handleResult(CURLcode code);
@@ -123,7 +123,7 @@ private:
     std::unique_ptr<Response> response;
 
     // In case of revalidation requests, this will store the old response.
-    std::unique_ptr<Response> existingResponse;
+    const std::shared_ptr<const Response> existingResponse;
 
     CURL *handle = nullptr;
     curl_slist *headers = nullptr;
@@ -425,10 +425,10 @@ static CURLcode sslctx_function(CURL * /* curl */, void *sslctx, void * /* parm
 }
 #endif
 
-HTTPRequestImpl::HTTPRequestImpl(HTTPRequest *request_, uv_loop_t *loop, std::unique_ptr<Response> response_)
+HTTPRequestImpl::HTTPRequestImpl(HTTPRequest *request_, uv_loop_t *loop, std::shared_ptr<const Response> response_)
     : context(HTTPCURLContext::Get(loop)),
       request(request_),
-      existingResponse(std::move(response_)),
+      existingResponse(response_),
       handle(context->getHandle()) {
     assert(request);
     context->addRequest(request);
@@ -688,12 +688,12 @@ void HTTPRequestImpl::handleResult(CURLcode code) {
 
         if (responseCode == 304) {
             if (existingResponse) {
-                // We're going to reuse the old response object, but need to copy over the new
-                // expires value (if possible).
-                std::swap(response, existingResponse);
-                if (existingResponse->expires) {
-                    response->expires = existingResponse->expires;
-                }
+                // We're going to copy over the existing response's data.
+                response->status = existingResponse->status;
+                response->message = existingResponse->message;
+                response->modified = existingResponse->modified;
+                response->etag = existingResponse->etag;
+                response->data = existingResponse->data;
                 return finish(ResponseStatus::NotModified);
             } else {
                 // This is an unsolicited 304 response and should only happen on malfunctioning
@@ -722,7 +722,7 @@ void HTTPRequestImpl::handleResult(CURLcode code) {
 
 // -------------------------------------------------------------------------------------------------
 
-HTTPRequest::HTTPRequest(DefaultFileSource *source_, const Resource &resource_)
+HTTPRequest::HTTPRequest(DefaultFileSource::Impl *source_, const Resource &resource_)
     : SharedRequestBase(source_, resource_) {
 }
 
@@ -734,11 +734,11 @@ HTTPRequest::~HTTPRequest() {
     }
 }
 
-void HTTPRequest::start(uv_loop_t *loop, std::unique_ptr<Response> response) {
+void HTTPRequest::start(uv_loop_t *loop, std::shared_ptr<const Response> response) {
     MBGL_VERIFY_THREAD(tid);
 
     assert(!ptr);
-    ptr = new HTTPRequestImpl(this, loop, std::move(response));
+    ptr = new HTTPRequestImpl(this, loop, response);
 }
 
 void HTTPRequest::retryImmediately() {

platform/default/sqlite_cache.cpp

@@ -1,21 +1,15 @@
-#include <mbgl/storage/default/sqlite_cache.hpp>
-#include <mbgl/storage/default/request.hpp>
+#include "sqlite_cache_impl.hpp"
+#include <mbgl/storage/request.hpp>
 #include <mbgl/storage/response.hpp>
 
-#include <mbgl/util/util.hpp>
-#include <mbgl/util/async_queue.hpp>
-#include <mbgl/util/variant.hpp>
 #include <mbgl/util/compression.hpp>
 #include <mbgl/util/io.hpp>
+#include <mbgl/util/thread.hpp>
 #include <mbgl/platform/log.hpp>
 
 #include "sqlite3.hpp"
 #include <sqlite3.h>
 
-#include <uv.h>
-
-#include <cassert>
-
 namespace mbgl {
 
 std::string removeAccessTokenFromURL(const std::string &url) {
@@ -67,93 +61,34 @@ std::string unifyMapboxURLs(const std::string &url) {
 
 using namespace mapbox::sqlite;
 
-struct SQLiteCache::GetAction {
-    const Resource resource;
-    const std::function<void(std::unique_ptr<Response>)> callback;
-};
-
-struct SQLiteCache::PutAction {
-    const Resource resource;
-    const std::shared_ptr<const Response> response;
-};
-
-struct SQLiteCache::RefreshAction {
-    const Resource resource;
-    const int64_t expires;
-};
-
-struct SQLiteCache::StopAction {
-};
-
-struct SQLiteCache::ActionDispatcher {
-    SQLiteCache &cache;
-    template <typename T> void operator()(T &t) { cache.process(t); }
-};
-
 SQLiteCache::SQLiteCache(const std::string& path_)
-    : path(path_),
-      loop(uv_loop_new()),
-      queue(new Queue(loop, [this](Action& action) {
-          mapbox::util::apply_visitor(ActionDispatcher{ *this }, action);
-      })),
-      thread([this]() {
-#ifdef __APPLE__
-          pthread_setname_np("SQLite Cache");
-#endif
-          uv_run(loop, UV_RUN_DEFAULT);
-
-          try {
-              getStmt.reset();
-              putStmt.reset();
-              refreshStmt.reset();
-              db.reset();
-          } catch (mapbox::sqlite::Exception& ex) {
-              Log::Error(Event::Database, ex.code, ex.what());
-          }
-      }) {
+    : thread(util::make_unique<util::Thread<Impl>>("SQLite Cache", path_)) {
 }
 
-SQLiteCache::~SQLiteCache() {
-    if (thread.joinable()) {
-        if (queue) {
-            queue->send(StopAction{ });
-        }
-        thread.join();
-        uv_loop_delete(loop);
+SQLiteCache::~SQLiteCache() = default;
+
+SQLiteCache::Impl::Impl(const std::string& path_)
+    : path(path_) {
+}
+
+SQLiteCache::Impl::~Impl() {
+    // Deleting these SQLite objects may result in exceptions, but we're in a destructor, so we
+    // can't throw anything.
+    try {
+        getStmt.reset();
+        putStmt.reset();
+        refreshStmt.reset();
+        db.reset();
+    } catch (mapbox::sqlite::Exception& ex) {
+        Log::Error(Event::Database, ex.code, ex.what());
     }
 }
 
-
-void SQLiteCache::get(const Resource &resource, std::function<void(std::unique_ptr<Response>)> callback) {
-    // Can be called from any thread, but most likely from the file source thread.
-    // Will try to load the URL from the SQLite database and call the callback when done.
-    // Note that the callback is probably going to invoked from another thread, so the caller
-    // must make sure that it can run in that thread.
-    assert(queue);
-    queue->send(GetAction{ resource, callback });
-}
-
-void SQLiteCache::put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint) {
-    // Can be called from any thread, but most likely from the file source thread. We are either
-    // storing a new response or updating the currently stored response, potentially setting a new
-    // expiry date.
-    assert(queue);
-    assert(response);
-
-    if (hint == Hint::Full) {
-        queue->send(PutAction{ resource, response });
-    } else if (hint == Hint::Refresh) {
-        queue->send(RefreshAction{ resource, response->expires });
-    }
-}
-
-void SQLiteCache::createDatabase() {
+void SQLiteCache::Impl::createDatabase() {
     db = util::make_unique<Database>(path.c_str(), ReadWrite | Create);
-
-    createSchema();
 }
 
-void SQLiteCache::createSchema() {
+void SQLiteCache::Impl::createSchema() {
     constexpr const char *const sql = ""
         "CREATE TABLE IF NOT EXISTS `http_cache` ("
         "    `url` TEXT PRIMARY KEY NOT NULL,"
@@ -171,7 +106,6 @@ void SQLiteCache::createSchema() {
         db->exec(sql);
         schema = true;
     } catch (mapbox::sqlite::Exception &ex) {
-
         if (ex.code == SQLITE_NOTADB) {
             Log::Warning(Event::Database, "Trashing invalid database");
             db.reset();
@@ -192,7 +126,15 @@ void SQLiteCache::createSchema() {
     }
 }
 
-void SQLiteCache::process(GetAction &action) {
+void SQLiteCache::get(const Resource &resource, Callback callback) {
+    // Can be called from any thread, but most likely from the file source thread.
+    // Will try to load the URL from the SQLite database and call the callback when done.
+    // Note that the callback is probably going to invoked from another thread, so the caller
+    // must make sure that it can run in that thread.
+    thread->invokeWithResult(&Impl::get, callback, resource);
+}
+
+std::unique_ptr<Response> SQLiteCache::Impl::get(const Resource &resource) {
     try {
         // This is called in the SQLite event loop.
         if (!db) {
@@ -212,7 +154,7 @@ void SQLiteCache::process(GetAction &action) {
             getStmt->reset();
         }
 
-        const std::string unifiedURL = unifyMapboxURLs(action.resource.url);
+        const std::string unifiedURL = unifyMapboxURLs(resource.url);
         getStmt->bind(1, unifiedURL.c_str());
         if (getStmt->run()) {
             // There is data.
@@ -225,18 +167,29 @@ void SQLiteCache::process(GetAction &action) {
             if (getStmt->get<int>(5)) { // == compressed
                 response->data = util::decompress(response->data);
             }
-            action.callback(std::move(response));
+            return std::move(response);
         } else {
             // There is no data.
-            action.callback(nullptr);
+            return nullptr;
         }
     } catch (mapbox::sqlite::Exception& ex) {
         Log::Error(Event::Database, ex.code, ex.what());
-        action.callback(nullptr);
+        return nullptr;
     }
 }
 
-void SQLiteCache::process(PutAction &action) {
+void SQLiteCache::put(const Resource &resource, std::shared_ptr<const Response> response, Hint hint) {
+    // Can be called from any thread, but most likely from the file source thread. We are either
+    // storing a new response or updating the currently stored response, potentially setting a new
+    // expiry date.
+    if (hint == Hint::Full) {
+        thread->invoke(&Impl::put, resource, std::move(response));
+    } else if (hint == Hint::Refresh) {
+        thread->invoke(&Impl::refresh, resource, int64_t(response->expires));
+    }
+}
+
+void SQLiteCache::Impl::put(const Resource& resource, std::shared_ptr<const Response> response) {
     try {
         if (!db) {
             createDatabase();
@@ -255,37 +208,37 @@ void SQLiteCache::process(PutAction &action) {
             putStmt->reset();
         }
 
-        const std::string unifiedURL = unifyMapboxURLs(action.resource.url);
+        const std::string unifiedURL = unifyMapboxURLs(resource.url);
         putStmt->bind(1 /* url */, unifiedURL.c_str());
-        putStmt->bind(2 /* status */, int(action.response->status));
-        putStmt->bind(3 /* kind */, int(action.resource.kind));
-        putStmt->bind(4 /* modified */, action.response->modified);
-        putStmt->bind(5 /* etag */, action.response->etag.c_str());
-        putStmt->bind(6 /* expires */, action.response->expires);
+        putStmt->bind(2 /* status */, int(response->status));
+        putStmt->bind(3 /* kind */, int(resource.kind));
+        putStmt->bind(4 /* modified */, response->modified);
+        putStmt->bind(5 /* etag */, response->etag.c_str());
+        putStmt->bind(6 /* expires */, response->expires);
 
         std::string data;
-        if (action.resource.kind != Resource::Image) {
+        if (resource.kind != Resource::Image) {
             // Do not compress images, since they are typically compressed already.
-            data = util::compress(action.response->data);
+            data = util::compress(response->data);
         }
 
-        if (!data.empty() && data.size() < action.response->data.size()) {
+        if (!data.empty() && data.size() < response->data.size()) {
             // Store the compressed data when it is smaller than the original
             // uncompressed data.
             putStmt->bind(7 /* data */, data, false); // do not retain the string internally.
             putStmt->bind(8 /* compressed */, true);
         } else {
-            putStmt->bind(7 /* data */, action.response->data, false); // do not retain the string internally.
+            putStmt->bind(7 /* data */, response->data, false); // do not retain the string internally.
             putStmt->bind(8 /* compressed */, false);
         }
 
         putStmt->run();
     } catch (mapbox::sqlite::Exception& ex) {
         Log::Error(Event::Database, ex.code, ex.what());
-        }
+    }
 }
 
-void SQLiteCache::process(RefreshAction &action) {
+void SQLiteCache::Impl::refresh(const Resource& resource, int64_t expires) {
     try {
         if (!db) {
             createDatabase();
@@ -302,8 +255,8 @@ void SQLiteCache::process(RefreshAction &action) {
             refreshStmt->reset();
         }
 
-        const std::string unifiedURL = unifyMapboxURLs(action.resource.url);
-        refreshStmt->bind(1, int64_t(action.expires));
+        const std::string unifiedURL = unifyMapboxURLs(resource.url);
+        refreshStmt->bind(1, int64_t(expires));
         refreshStmt->bind(2, unifiedURL.c_str());
         refreshStmt->run();
     } catch (mapbox::sqlite::Exception& ex) {
@@ -311,10 +264,4 @@ void SQLiteCache::process(RefreshAction &action) {
     }
 }
 
-void SQLiteCache::process(StopAction &) {
-    assert(queue);
-    queue->stop();
-    queue = nullptr;
-}
-
 }

platform/default/sqlite_cache_impl.hpp

@@ -0,0 +1,39 @@
+#ifndef MBGL_STORAGE_DEFAULT_SQLITE_CACHE_IMPL
+#define MBGL_STORAGE_DEFAULT_SQLITE_CACHE_IMPL
+
+#include <mbgl/storage/sqlite_cache.hpp>
+
+namespace mapbox {
+namespace sqlite {
+class Database;
+class Statement;
+}
+}
+
+namespace mbgl {
+
+class SQLiteCache::Impl {
+public:
+    Impl(const std::string &path = ":memory:");
+    ~Impl();
+
+    std::unique_ptr<Response> get(const Resource&);
+    void put(const Resource& resource, std::shared_ptr<const Response> response);
+    void refresh(const Resource& resource, int64_t expires);
+
+private:
+    void createDatabase();
+    void createSchema();
+
+    const std::string path;
+    std::unique_ptr<::mapbox::sqlite::Database> db;
+    std::unique_ptr<::mapbox::sqlite::Statement> getStmt;
+    std::unique_ptr<::mapbox::sqlite::Statement> putStmt;
+    std::unique_ptr<::mapbox::sqlite::Statement> refreshStmt;
+    bool schema = false;
+};
+
+
+}
+
+#endif

platform/ios/MGLMapView.mm

@@ -11,7 +11,7 @@
 #include <mbgl/platform/platform.hpp>
 #include <mbgl/platform/darwin/reachability.h>
 #include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 #include <mbgl/storage/network_status.hpp>
 #include <mbgl/util/geo.hpp>
 

src/mbgl/storage/asset_request.hpp

@@ -7,9 +7,9 @@ namespace mbgl {
 
 class AssetRequest : public SharedRequestBase {
 public:
-    AssetRequest(DefaultFileSource *source, const Resource &resource);
+    AssetRequest(DefaultFileSource::Impl *source, const Resource &resource);
 
-    void start(uv_loop_t *loop, std::unique_ptr<Response> response = nullptr);
+    void start(uv_loop_t *loop, std::shared_ptr<const Response> response = nullptr);
     void cancel();
 
 private:

src/mbgl/storage/default_file_source.cpp

@@ -1,17 +1,16 @@
-#include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/request.hpp>
-#include <mbgl/storage/default/asset_request.hpp>
-#include <mbgl/storage/default/http_request.hpp>
+#include <mbgl/storage/default_file_source_impl.hpp>
+#include <mbgl/storage/request.hpp>
+#include <mbgl/storage/asset_request.hpp>
+#include <mbgl/storage/http_request.hpp>
 
 #include <mbgl/storage/response.hpp>
 #include <mbgl/platform/platform.hpp>
 
-#include <mbgl/util/async_queue.hpp>
-#include <mbgl/util/util.hpp>
-
-#include <mbgl/util/variant.hpp>
+#include <mbgl/util/uv_detail.hpp>
 #include <mbgl/util/chrono.hpp>
+#include <mbgl/util/thread.hpp>
 #include <mbgl/platform/log.hpp>
+#include <mbgl/map/environment.hpp>
 
 #pragma GCC diagnostic push
 #ifndef __clang__
@@ -21,7 +20,6 @@
 #include <boost/algorithm/string.hpp>
 #pragma GCC diagnostic pop
 
-#include <thread>
 #include <algorithm>
 #include <cassert>
 
@@ -30,75 +28,19 @@ namespace algo = boost::algorithm;
 
 namespace mbgl {
 
-struct DefaultFileSource::ActionDispatcher {
-    DefaultFileSource &fileSource;
-    template <typename T> void operator()(T &t) { fileSource.process(t); }
-};
-
-struct DefaultFileSource::AddRequestAction {
-    Request *const request;
-};
-
-struct DefaultFileSource::RemoveRequestAction {
-    Request *const request;
-};
-
-struct DefaultFileSource::ResultAction {
-    const Resource resource;
-    std::unique_ptr<Response> response;
-};
-
-struct DefaultFileSource::StopAction {
-};
-
-struct DefaultFileSource::AbortAction {
-    const Environment &env;
-};
-
-
-DefaultFileSource::DefaultFileSource(FileCache *cache_, const std::string &root)
-    : assetRoot(root.empty() ? platform::assetRoot() : root),
-      loop(uv_loop_new()),
-      cache(cache_),
-      queue(new Queue(loop, [this](Action &action) {
-          mapbox::util::apply_visitor(ActionDispatcher{*this}, action);
-      })),
-      thread([this]() {
-#ifdef __APPLE__
-          pthread_setname_np("FileSource");
-#endif
-          uv_run(loop, UV_RUN_DEFAULT);
-      }) {
+DefaultFileSource::Impl::Impl(FileCache* cache_, const std::string& root)
+    : assetRoot(root.empty() ? platform::assetRoot() : root), cache(cache_) {
 }
 
-DefaultFileSource::DefaultFileSource(FileCache *cache_, uv_loop_t *loop_, const std::string &root)
-    : assetRoot(root.empty() ? platform::assetRoot() : root),
-      loop(loop_),
-      cache(cache_),
-      queue(new Queue(loop, [this](Action &action) {
-          mapbox::util::apply_visitor(ActionDispatcher{*this}, action);
-      })) {
-    // Make sure that the queue doesn't block the loop from exiting.
-    queue->unref();
+DefaultFileSource::DefaultFileSource(FileCache* cache, const std::string& root)
+    : thread(util::make_unique<util::Thread<Impl>>("FileSource", cache, root)) {
 }
 
 DefaultFileSource::~DefaultFileSource() {
     MBGL_VERIFY_THREAD(tid);
-
-    if (thread.joinable()) {
-        if (queue) {
-            queue->send(StopAction{ });
-        }
-        thread.join();
-        uv_loop_delete(loop);
-    } else {
-        // Assume that the loop we received is running in the current thread.
-        StopAction action {};
-        process(action);
-    }
 }
 
-SharedRequestBase *DefaultFileSource::find(const Resource &resource) {
+SharedRequestBase *DefaultFileSource::Impl::find(const Resource &resource) {
     // We're using a set of pointers here instead of a map between url and SharedRequestBase because
     // we need to find the requests both by pointer and by URL. Given that the number of requests
     // is generally very small (typically < 10 at a time), hashing by URL incurs too much overhead
@@ -110,40 +52,37 @@ SharedRequestBase *DefaultFileSource::find(const Resource &resource) {
     return nullptr;
 }
 
-Request *DefaultFileSource::request(const Resource &resource, uv_loop_t *l, const Environment &env,
+Request* DefaultFileSource::request(const Resource& resource,
+                                    uv_loop_t* l,
+                                    const Environment& env,
                                     Callback callback) {
     auto req = new Request(resource, l, env, std::move(callback));
 
     // This function can be called from any thread. Make sure we're executing the actual call in the
-    // file source loop by sending it over the queue. It will be processed in processAction().
-    queue->send(AddRequestAction{ req });
+    // file source loop by sending it over the queue.
+    thread->invoke(&Impl::add, std::move(req), thread->get());
+
     return req;
 }
 
-void DefaultFileSource::request(const Resource &resource, const Environment &env,
-                                Callback callback) {
-    auto req = new Request(resource, nullptr, env, std::move(callback));
-
-    // This function can be called from any thread. Make sure we're executing the actual call in the
-    // file source loop by sending it over the queue. It will be processed in processAction().
-    queue->send(AddRequestAction{ req });
+void DefaultFileSource::request(const Resource& resource, const Environment& env, Callback callback) {
+    request(resource, nullptr, env, std::move(callback));
 }
 
 void DefaultFileSource::cancel(Request *req) {
     req->cancel();
 
     // This function can be called from any thread. Make sure we're executing the actual call in the
-    // file source loop by sending it over the queue. It will be processed in processAction().
-    queue->send(RemoveRequestAction{ req });
+    // file source loop by sending it over the queue.
+    thread->invoke(&Impl::cancel, std::move(req));
 }
 
 void DefaultFileSource::abort(const Environment &env) {
-    queue->send(AbortAction{ env });
+    thread->invoke(&Impl::abort, std::ref(env));
 }
 
-
-void DefaultFileSource::process(AddRequestAction &action) {
-    const Resource &resource = action.request->resource;
+void DefaultFileSource::Impl::add(Request* req, uv_loop_t* loop) {
+    const Resource &resource = req->resource;
 
     // We're adding a new Request.
     SharedRequestBase *sharedRequest = find(resource);
@@ -155,11 +94,6 @@ void DefaultFileSource::process(AddRequestAction &action) {
             sharedRequest = new HTTPRequest(this, resource);
         }
 
-        // Make sure the loop stays alive when we're not running the file source in it's own thread.
-        if (!thread.joinable() && pending.empty()) {
-            queue->ref();
-        }
-
         const bool inserted = pending.emplace(resource, sharedRequest).second;
         assert(inserted);
         (void (inserted)); // silence unused variable warning on Release builds.
@@ -170,21 +104,21 @@ void DefaultFileSource::process(AddRequestAction &action) {
         } else {
             // Otherwise, first check the cache for existing data so that we can potentially
             // revalidate the information without having to redownload everything.
-            cache->get(resource, [this, resource](std::unique_ptr<Response> response) {
-                queue->send(ResultAction { resource, std::move(response) });
+            cache->get(resource, [this, resource, loop](std::unique_ptr<Response> response) {
+                processResult(resource, std::move(response), loop);
             });
         }
     }
-    sharedRequest->subscribe(action.request);
+    sharedRequest->subscribe(req);
 }
 
-void DefaultFileSource::process(RemoveRequestAction &action) {
-    SharedRequestBase *sharedRequest = find(action.request->resource);
+void DefaultFileSource::Impl::cancel(Request* req) {
+    SharedRequestBase *sharedRequest = find(req->resource);
     if (sharedRequest) {
         // If the number of dependent requests of the SharedRequestBase drops to zero, the
         // unsubscribe callback triggers the removal of the SharedRequestBase pointer from the list
         // of pending requests and initiates cancelation.
-        sharedRequest->unsubscribe(action.request);
+        sharedRequest->unsubscribe(req);
     } else {
         // There is no request for this URL anymore. Likely, the request already completed
         // before we got around to process the cancelation request.
@@ -192,24 +126,24 @@ void DefaultFileSource::process(RemoveRequestAction &action) {
 
     // Send a message back to the requesting thread and notify it that this request has been
     // canceled and is now safe to be deleted.
-    action.request->destruct();
+    req->destruct();
 }
 
-void DefaultFileSource::process(ResultAction &action) {
-    SharedRequestBase *sharedRequest = find(action.resource);
+void DefaultFileSource::Impl::processResult(const Resource& resource, std::shared_ptr<const Response> response, uv_loop_t* loop) {
+    SharedRequestBase *sharedRequest = find(resource);
     if (sharedRequest) {
-        if (action.response) {
+        if (response) {
             // This entry was stored in the cache. Now determine if we need to revalidate.
             const int64_t now = std::chrono::duration_cast<std::chrono::seconds>(
                                     SystemClock::now().time_since_epoch()).count();
-            if (action.response->expires > now) {
+            if (response->expires > now) {
                 // The response is fresh. We're good to notify the caller.
-                sharedRequest->notify(std::move(action.response), FileCache::Hint::No);
+                sharedRequest->notify(response, FileCache::Hint::No);
                 sharedRequest->cancel();
                 return;
             } else {
                 // The cached response is stale. Now run the real request.
-                sharedRequest->start(loop, std::move(action.response));
+                sharedRequest->start(loop, response);
             }
         } else {
             // There is no response. Now run the real request.
@@ -221,19 +155,8 @@ void DefaultFileSource::process(ResultAction &action) {
     }
 }
 
-// A stop action means the file source is about to be destructed. We need to cancel all requests
-// for all environments.
-void DefaultFileSource::process(StopAction &) {
-    // There may not be any pending requests in this file source anymore. You must terminate all
-    // Map objects before deleting the FileSource.
-    assert(pending.empty());
-    assert(queue);
-    queue->stop();
-    queue = nullptr;
-}
-
 // Aborts all requests that are part of the current environment.
-void DefaultFileSource::process(AbortAction &action) {
+void DefaultFileSource::Impl::abort(const Environment& env) {
     // Construct a cancellation response.
     auto res = util::make_unique<Response>();
     res->status = Response::Error;
@@ -243,7 +166,7 @@ void DefaultFileSource::process(AbortAction &action) {
     // Iterate through all pending requests and remove them in case they're abandoned.
     util::erase_if(pending, [&](const std::pair<Resource, SharedRequestBase *> &it) -> bool {
         // Obtain all pending requests that are in the current environment.
-        const auto aborted = it.second->removeAllInEnvironment(action.env);
+        const auto aborted = it.second->removeAllInEnvironment(env);
 
         // Notify all observers.
         for (auto req : aborted) {
@@ -260,7 +183,7 @@ void DefaultFileSource::process(AbortAction &action) {
     });
 }
 
-void DefaultFileSource::notify(SharedRequestBase *sharedRequest,
+void DefaultFileSource::Impl::notify(SharedRequestBase *sharedRequest,
                                const std::set<Request *> &observers,
                                std::shared_ptr<const Response> response, FileCache::Hint hint) {
     // First, remove the request, since it might be destructed at any point now.
@@ -278,11 +201,6 @@ void DefaultFileSource::notify(SharedRequestBase *sharedRequest,
             req->notify(response);
         }
     }
-
-    if (!thread.joinable() && pending.empty()) {
-        // When there are no pending requests, we're going to allow the queue to stop.
-        queue->unref();
-    }
 }
 
 }

src/mbgl/storage/default_file_source_impl.hpp

@@ -0,0 +1,36 @@
+#ifndef MBGL_STORAGE_DEFAULT_DEFAULT_FILE_SOURCE_IMPL
+#define MBGL_STORAGE_DEFAULT_DEFAULT_FILE_SOURCE_IMPL
+
+#include <mbgl/storage/default_file_source.hpp>
+
+#include <set>
+#include <unordered_map>
+
+namespace mbgl {
+
+class SharedRequestBase;
+
+class DefaultFileSource::Impl {
+public:
+    Impl(FileCache *cache, const std::string &root = "");
+
+    void notify(SharedRequestBase *sharedRequest, const std::set<Request *> &observers,
+                std::shared_ptr<const Response> response, FileCache::Hint hint);
+    SharedRequestBase *find(const Resource &resource);
+
+    void add(Request* request, uv_loop_t* loop);
+    void cancel(Request* request);
+    void abort(const Environment& env);
+
+    const std::string assetRoot;
+
+private:
+    void processResult(const Resource& resource, std::shared_ptr<const Response> response, uv_loop_t* loop);
+
+    std::unordered_map<Resource, SharedRequestBase *, Resource::Hash> pending;
+    FileCache *cache = nullptr;
+};
+
+}
+
+#endif

src/mbgl/storage/http_request.hpp

@@ -7,9 +7,9 @@ namespace mbgl {
 
 class HTTPRequest : public SharedRequestBase {
 public:
-    HTTPRequest(DefaultFileSource *source, const Resource &resource);
+    HTTPRequest(DefaultFileSource::Impl *source, const Resource &resource);
 
-    void start(uv_loop_t *loop, std::unique_ptr<Response> response = nullptr);
+    void start(uv_loop_t *loop, std::shared_ptr<const Response> response = nullptr);
     void cancel();
 
     void retryImmediately();

src/mbgl/storage/request.cpp

@@ -1,4 +1,4 @@
-#include <mbgl/storage/default/request.hpp>
+#include <mbgl/storage/request.hpp>
 
 #include <mbgl/storage/response.hpp>
 

src/mbgl/storage/shared_request_base.hpp

@@ -3,8 +3,8 @@
 
 #include <mbgl/storage/resource.hpp>
 #include <mbgl/storage/file_cache.hpp>
-#include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/request.hpp>
+#include <mbgl/storage/default_file_source_impl.hpp>
+#include <mbgl/storage/request.hpp>
 #include <mbgl/util/util.hpp>
 #include <mbgl/util/noncopyable.hpp>
 
@@ -26,18 +26,17 @@ protected:
     MBGL_STORE_THREAD(tid)
 
 public:
-    SharedRequestBase(DefaultFileSource *source_, const Resource &resource_)
+    SharedRequestBase(DefaultFileSource::Impl *source_, const Resource &resource_)
         : resource(resource_), source(source_) {}
 
-    virtual void start(uv_loop_t *loop, std::unique_ptr<Response> response = nullptr) = 0;
+    virtual void start(uv_loop_t *loop, std::shared_ptr<const Response> response = nullptr) = 0;
     virtual void cancel() = 0;
 
-    void notify(std::unique_ptr<Response> response, FileCache::Hint hint) {
+    void notify(std::shared_ptr<const Response> response, FileCache::Hint hint) {
         MBGL_VERIFY_THREAD(tid);
 
         if (source) {
-            source->notify(this, observers, std::shared_ptr<const Response>(std::move(response)),
-                           hint);
+            source->notify(this, observers, response, hint);
         }
     }
 
@@ -96,7 +95,7 @@ public:
     const Resource resource;
 
 protected:
-    DefaultFileSource *source = nullptr;
+    DefaultFileSource::Impl *source = nullptr;
 
 private:
     std::set<Request *> observers;

src/mbgl/util/run_loop.cpp

@@ -0,0 +1,39 @@
+#include <mbgl/util/run_loop.hpp>
+
+namespace mbgl {
+namespace util {
+
+uv::tls<RunLoop> RunLoop::current;
+
+RunLoop::RunLoop()
+    : async(*loop, std::bind(&RunLoop::process, this)) {
+}
+
+void RunLoop::withMutex(std::function<void()>&& fn) {
+    std::lock_guard<std::mutex> lock(mutex);
+    fn();
+}
+
+void RunLoop::process() {
+    Queue queue_;
+    withMutex([&] { queue_.swap(queue); });
+
+    while (!queue_.empty()) {
+        (queue_.front())();
+        queue_.pop();
+    }
+}
+
+void RunLoop::run() {
+    assert(!current.get());
+    current.set(this);
+    loop.run();
+    current.set(nullptr);
+}
+
+void RunLoop::stop() {
+    invoke([&] { async.unref(); });
+}
+
+}
+}

src/mbgl/util/run_loop.hpp

@@ -0,0 +1,100 @@
+#ifndef MBGL_UTIL_RUN_LOOP
+#define MBGL_UTIL_RUN_LOOP
+
+#include <mbgl/util/uv_detail.hpp>
+
+#include <functional>
+#include <queue>
+#include <mutex>
+
+namespace mbgl {
+namespace util {
+
+class RunLoop {
+public:
+    RunLoop();
+
+    void run();
+    void stop();
+
+    // Invoke fn() in the runloop thread.
+    template <class Fn>
+    void invoke(Fn&& fn) {
+        withMutex([&] { queue.push(Message(std::move(fn))); });
+        async.send();
+    }
+
+    // Invoke fn() in the runloop thread, then invoke callback(result) in the current thread.
+    template <class Fn, class R>
+    void invokeWithResult(Fn&& fn, std::function<void (R)> callback) {
+        RunLoop* outer = current.get();
+        assert(outer);
+
+        invoke([fn, callback, outer] {
+            /*
+                With C++14, we could write:
+
+                outer->invoke([callback, result = std::move(fn())] () mutable {
+                    callback(std::move(result));
+                });
+
+                Instead we're using a workaround with std::bind
+                to obtain move-capturing semantics with C++11:
+                  http://stackoverflow.com/a/12744730/52207
+            */
+            outer->invoke(std::bind([callback] (R& result) {
+                callback(std::move(result));
+            }, std::move(fn())));
+        });
+    }
+
+    uv_loop_t* get() { return *loop; }
+
+private:
+    struct Message {
+        struct Base {
+            virtual void operator()() = 0;
+            virtual ~Base() = default;
+        };
+
+        template <class F>
+        struct Invoker : Base {
+            Invoker(F&& f) : func(std::move(f)) {}
+            void operator()() override { func(); }
+            F func;
+        }; 
+
+        Message() = default;
+        Message(Message&&) = default;
+        ~Message() = default;
+        Message& operator=(Message&&) = default;
+
+        // copy members implicitly deleted
+
+        template <class Fn>
+        Message(Fn fn)
+            : p_fn(new Invoker<Fn>(std::move(fn))) {
+        }
+
+        void operator()() const { (*p_fn)(); }
+        std::unique_ptr<Base> p_fn;
+    };
+
+    using Queue = std::queue<Message>;
+
+    static uv::tls<RunLoop> current;
+
+    void withMutex(std::function<void()>&&);
+    void process();
+
+    Queue queue;
+    std::mutex mutex;
+
+    uv::loop loop;
+    uv::async async;
+};
+
+}
+}
+
+#endif

src/mbgl/util/thread.hpp

@@ -0,0 +1,121 @@
+#ifndef MBGL_UTIL_THREAD
+#define MBGL_UTIL_THREAD
+
+#include <future>
+#include <thread>
+#include <functional>
+
+#include <mbgl/util/run_loop.hpp>
+
+namespace {
+
+template <::std::size_t...>
+struct index_sequence {};
+
+template <::std::size_t N, ::std::size_t... I>
+struct integer_sequence : integer_sequence<N - 1, N - 1, I...> {};
+
+template <::std::size_t... I>
+struct integer_sequence<0, I...> {
+    using type = index_sequence<I...>;
+};
+
+}
+
+namespace mbgl {
+namespace util {
+
+// Manages a thread with Object.
+
+// Upon creation of this object, it launches a thread, creates an object of type Object in that
+// thread, and then calls .start(); on that object. When the Thread<> object is destructed, the
+// Object's .stop() function is called, and the destructor waits for thread termination. The
+// Thread<> constructor blocks until the thread and the Object are fully created, so after the
+// object creation, it's safe to obtain the Object stored in this thread.
+
+template <class Object>
+class Thread {
+public:
+    template <class... Args>
+    Thread(const std::string& name, Args&&... args);
+    ~Thread();
+
+    // Invoke object->fn(args...) in the runloop thread.
+    template <typename Fn, class... Args>
+    void invoke(Fn fn, Args&&... args) {
+        loop->invoke(std::bind(fn, object, args...));
+    }
+
+    // Invoke object->fn(args...) in the runloop thread, then invoke callback(result) in the current thread.
+    template <typename Fn, class R, class... Args>
+    void invokeWithResult(Fn fn, std::function<void (R)> callback, Args&&... args) {
+        loop->invokeWithResult(std::bind(fn, object, args...), callback);
+    }
+
+    uv_loop_t* get() { return loop->get(); }
+
+private:
+    Thread(const Thread&) = delete;
+    Thread(Thread&&) = delete;
+    Thread& operator=(const Thread&) = delete;
+    Thread& operator=(Thread&&) = delete;
+
+    template <typename P, std::size_t... I>
+    void run(P&& params, index_sequence<I...>);
+
+    std::promise<void> running;
+    std::promise<void> joinable;
+
+    std::thread thread;
+
+    Object* object;
+    RunLoop* loop;
+};
+
+template <class Object>
+template <class... Args>
+Thread<Object>::Thread(const std::string& name, Args&&... args) {
+    // Note: We're using std::tuple<> to store the arguments because GCC 4.9 has a bug
+    // when expanding parameters packs captured in lambdas.
+    std::tuple<Args...> params = std::forward_as_tuple(::std::forward<Args>(args)...);
+
+    thread = std::thread([&] {
+        #ifdef __APPLE__
+        pthread_setname_np(name.c_str());
+        #else
+        (void(name));
+        #endif
+
+        constexpr auto seq = typename integer_sequence<sizeof...(Args)>::type();
+        run(std::move(params), seq);
+    });
+
+    running.get_future().get();
+}
+
+template <class Object>
+template <typename P, std::size_t... I>
+void Thread<Object>::run(P&& params, index_sequence<I...>) {
+    Object object_(std::get<I>(std::forward<P>(params))...);
+    object = &object_;
+
+    RunLoop loop_;
+    loop = &loop_;
+
+    running.set_value();
+    loop_.run();
+
+    joinable.get_future().get();
+}
+
+template <class Object>
+Thread<Object>::~Thread() {
+    loop->stop();
+    joinable.set_value();
+    thread.join();
+}
+
+}
+}
+
+#endif

src/mbgl/util/uv.cpp

@@ -3,6 +3,28 @@
 
 #include <uv.h>
 
+#if UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR <= 10
+
+int uv_key_create(uv_key_t* key) {
+  return -pthread_key_create(key, NULL);
+}
+
+void uv_key_delete(uv_key_t* key) {
+  if (pthread_key_delete(*key))
+    abort();
+}
+
+void* uv_key_get(uv_key_t* key) {
+  return pthread_getspecific(*key);
+}
+
+void uv_key_set(uv_key_t* key, void* value) {
+  if (pthread_setspecific(*key, value))
+    abort();
+}
+
+#endif
+
 namespace uv {
 
 std::string cwd() {

src/mbgl/util/uv_detail.hpp

@@ -11,6 +11,21 @@
 #include <memory>
 #include <string>
 
+#if UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR <= 10
+
+// Add thread local storage to libuv API:
+// https://github.com/joyent/libuv/commit/5d2434bf71e47802841bad218d521fa254d1ca2d
+
+typedef pthread_key_t uv_key_t;
+
+UV_EXTERN int uv_key_create(uv_key_t* key);
+UV_EXTERN void uv_key_delete(uv_key_t* key);
+UV_EXTERN void* uv_key_get(uv_key_t* key);
+UV_EXTERN void uv_key_set(uv_key_t* key, void* value);
+
+#endif
+
+
 namespace uv {
 
 template <class T>
@@ -41,10 +56,19 @@ public:
         uv_loop_close(l);
         delete l;
 #endif
-
     }
 
-    inline uv_loop_t *operator*() { return l; }
+    inline void run() {
+        uv_run(l, UV_RUN_DEFAULT);
+    }
+
+    inline uv_loop_t* operator*() {
+        return l;
+    }
+
+    inline uv_loop_t* get() {
+        return l;
+    }
 
 private:
     uv_loop_t *l = nullptr;
@@ -72,6 +96,14 @@ public:
         }
     }
 
+    inline void ref() {
+        uv_ref(reinterpret_cast<uv_handle_t*>(a.get()));
+    }
+
+    inline void unref() {
+        uv_unref(reinterpret_cast<uv_handle_t*>(a.get()));
+    }
+
 private:
 #if UV_VERSION_MAJOR == 0 && UV_VERSION_MINOR <= 10
     static void async_cb(uv_async_t* a, int) {
@@ -116,6 +148,22 @@ private:
     uv_rwlock_t mtx;
 };
 
+template <class T>
+class tls : public mbgl::util::noncopyable {
+public:
+    inline tls() {
+        if (uv_key_create(&key) != 0) {
+            throw std::runtime_error("failed to initialize thread local storage key");
+        }
+    }
+    inline ~tls() { uv_key_delete(&key); }
+    inline T* get() { return reinterpret_cast<T*>(uv_key_get(&key)); }
+    inline void set(T* val) { uv_key_set(&key, val); }
+
+private:
+    uv_key_t key;
+};
+
 }
 
 #endif

test/headless/headless.cpp

@@ -139,7 +139,7 @@ TEST_P(HeadlessTest, render) {
         }
 
         HeadlessView view(display);
-        mbgl::DefaultFileSource fileSource(nullptr);
+        DefaultFileSource fileSource(nullptr);
         Map map(view, fileSource);
 
         map.setClasses(classes);

test/storage/cache_response.cpp

@@ -3,7 +3,7 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 
 TEST_F(Storage, CacheResponse) {
     SCOPED_TEST(CacheResponse);
@@ -11,7 +11,7 @@ TEST_F(Storage, CacheResponse) {
     using namespace mbgl;
 
     SQLiteCache cache(":memory:");
-    DefaultFileSource fs(&cache, uv_default_loop());
+    DefaultFileSource fs(&cache);
 
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/cache" };
     auto &env = *static_cast<const Environment *>(nullptr);

test/storage/cache_revalidate.cpp

@@ -3,7 +3,7 @@
 #include <uv.h>
 
 #include <mbgl/storage/default_file_source.hpp>
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 
 TEST_F(Storage, CacheRevalidate) {
     SCOPED_TEST(CacheRevalidateSame)

test/storage/database.cpp

@@ -1,52 +1,34 @@
 #include "storage.hpp"
 #include "../fixtures/fixture_log_observer.hpp"
 
-#include <future>
-
-#include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/storage/sqlite_cache.hpp>
 #include <mbgl/storage/resource.hpp>
 #include <mbgl/storage/response.hpp>
 #include <mbgl/util/io.hpp>
+#include <mbgl/util/run_loop.hpp>
 
 #include <sqlite3.h>
 
-class ScopedTest {
-public:
-    ScopedTest(std::function<void()> destructor_) : destructor(destructor_) {}
-    ScopedTest() {}
-
-    void finish() {
-        promise.set_value();
-    }
-
-    ~ScopedTest() {
-        promise.get_future().get();
-        if (destructor) {
-            destructor();
-        }
-    }
-
-private:
-    const std::function<void()> destructor;
-    std::promise<void> promise;
-};
-
 TEST_F(Storage, DatabaseDoesNotExist) {
     using namespace mbgl;
 
     Log::setObserver(util::make_unique<FixtureLogObserver>());
 
-    ScopedTest test([&] {
-        auto observer = Log::removeObserver();
-        EXPECT_EQ(1ul, dynamic_cast<FixtureLogObserver*>(observer.get())->count({ EventSeverity::Error, Event::Database, 14, "unable to open database file" }));
-    });
-
     SQLiteCache cache("test/fixtures/404/cache.db");
 
-    cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-        EXPECT_EQ(nullptr, res.get());
-        test.finish();
+    util::RunLoop loop;
+
+    loop.invoke([&] {
+        cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+            EXPECT_EQ(nullptr, res.get());
+            loop.stop();
+        });
     });
+
+    loop.run();
+
+    auto observer = Log::removeObserver();
+    EXPECT_EQ(1ul, dynamic_cast<FixtureLogObserver*>(observer.get())->count({ EventSeverity::Error, Event::Database, 14, "unable to open database file" }));
 }
 
 void createDir(const char* name) {
@@ -81,16 +63,20 @@ TEST_F(Storage, DatabaseCreate) {
 
     Log::setObserver(util::make_unique<FixtureLogObserver>());
 
-    ScopedTest test([&] {
-        Log::removeObserver();
-    });
-
     SQLiteCache cache("test/fixtures/database/cache.db");
 
-    cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-        EXPECT_EQ(nullptr, res.get());
-        test.finish();
+    util::RunLoop loop;
+
+    loop.invoke([&] {
+        cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+            EXPECT_EQ(nullptr, res.get());
+            loop.stop();
+        });
     });
+
+    loop.run();
+
+    Log::removeObserver();
 }
 
 class FileLock {
@@ -140,19 +126,23 @@ TEST_F(Storage, DatabaseLockedRead) {
     deleteFile("test/fixtures/database/locked.db");
     FileLock guard("test/fixtures/database/locked.db");
 
-    auto cache = util::make_unique<SQLiteCache>("test/fixtures/database/locked.db");
 
-    std::promise<void> promise;
+    SQLiteCache cache("test/fixtures/database/locked.db");
 
     {
         // First request should fail.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_EQ(nullptr, res.get());
-            promise.set_value();
+
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_EQ(nullptr, res.get());
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         // Make sure that we got a few "database locked" errors
         auto observer = Log::removeObserver();
@@ -166,20 +156,21 @@ TEST_F(Storage, DatabaseLockedRead) {
     {
         // First, try getting a file (the cache value should not exist).
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
 
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_EQ(nullptr, res.get());
-            promise.set_value();
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_EQ(nullptr, res.get());
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         // Make sure that we got a no errors
         Log::removeObserver();
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }
 
 
@@ -192,21 +183,24 @@ TEST_F(Storage, DatabaseLockedWrite) {
     deleteFile("test/fixtures/database/locked.db");
     FileLock guard("test/fixtures/database/locked.db");
 
-    auto cache = util::make_unique<SQLiteCache>("test/fixtures/database/locked.db");
-
-    std::promise<void> promise;
+    SQLiteCache cache("test/fixtures/database/locked.db");
 
     {
         // Adds a file (which should fail).
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
-        auto response = std::make_shared<Response>();
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_EQ(nullptr, res.get());
-            promise.set_value();
+
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_EQ(nullptr, res.get());
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         auto observer = Log::removeObserver();
         auto flo = dynamic_cast<FixtureLogObserver*>(observer.get());
@@ -219,24 +213,25 @@ TEST_F(Storage, DatabaseLockedWrite) {
     {
         // Then, set a file and obtain it again.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
 
-        auto response = std::make_shared<Response>();
-        response->data = "Demo";
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_NE(nullptr, res.get());
-            EXPECT_EQ("Demo", res->data);
-            promise.set_value();
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            response->data = "Demo";
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_NE(nullptr, res.get());
+                EXPECT_EQ("Demo", res->data);
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         // Make sure that we got a no errors
         Log::removeObserver();
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }
 
 
@@ -249,25 +244,28 @@ TEST_F(Storage, DatabaseLockedRefresh) {
     createDir("test/fixtures/database");
     deleteFile("test/fixtures/database/locked.db");
 
-    auto cache = util::make_unique<SQLiteCache>("test/fixtures/database/locked.db");
+    SQLiteCache cache("test/fixtures/database/locked.db");
 
     // Then, lock the file and try again.
     FileLock guard("test/fixtures/database/locked.db");
 
-    std::promise<void> promise;
-
     {
         // Adds a file.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
-        auto response = std::make_shared<Response>();
-        response->data = "Demo";
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_EQ(nullptr, res.get());
-            promise.set_value();
+
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            response->data = "Demo";
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_EQ(nullptr, res.get());
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         auto observer = Log::removeObserver();
         auto flo = dynamic_cast<FixtureLogObserver*>(observer.get());
@@ -277,25 +275,26 @@ TEST_F(Storage, DatabaseLockedRefresh) {
     {
         // Then, try to refresh it.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
 
-        auto response = std::make_shared<Response>();
-        response->data = "Demo";
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Refresh);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_EQ(nullptr, res.get());
-            promise.set_value();
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            response->data = "Demo";
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Refresh);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_EQ(nullptr, res.get());
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         // Make sure that we got the right errors.
         auto observer = Log::removeObserver();
         auto flo = dynamic_cast<FixtureLogObserver*>(observer.get());
         EXPECT_EQ(4ul, flo->count({ EventSeverity::Error, Event::Database, 5, "database is locked" }));
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }
 
 
@@ -307,23 +306,26 @@ TEST_F(Storage, DatabaseDeleted) {
     createDir("test/fixtures/database");
     deleteFile("test/fixtures/database/locked.db");
 
-    auto cache = util::make_unique<SQLiteCache>("test/fixtures/database/locked.db");
-
-    std::promise<void> promise;
+    SQLiteCache cache("test/fixtures/database/locked.db");
 
     {
         // Adds a file.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
-        auto response = std::make_shared<Response>();
-        response->data = "Demo";
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_NE(nullptr, res.get());
-            EXPECT_EQ("Demo", res->data);
-            promise.set_value();
+
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            response->data = "Demo";
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_NE(nullptr, res.get());
+                EXPECT_EQ("Demo", res->data);
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         Log::removeObserver();
     }
@@ -333,24 +335,26 @@ TEST_F(Storage, DatabaseDeleted) {
     {
         // Adds a file.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
-        auto response = std::make_shared<Response>();
-        response->data = "Demo";
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_NE(nullptr, res.get());
-            EXPECT_EQ("Demo", res->data);
-            promise.set_value();
+
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            response->data = "Demo";
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_NE(nullptr, res.get());
+                EXPECT_EQ("Demo", res->data);
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         auto observer = Log::removeObserver();
         auto flo = dynamic_cast<FixtureLogObserver*>(observer.get());
         EXPECT_EQ(1ul, flo->count({ EventSeverity::Error, Event::Database, 8, "attempt to write a readonly database" }));
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }
 
 
@@ -363,29 +367,29 @@ TEST_F(Storage, DatabaseInvalid) {
     deleteFile("test/fixtures/database/invalid.db");
     writeFile("test/fixtures/database/invalid.db", "this is an invalid file");
 
-    auto cache = util::make_unique<SQLiteCache>("test/fixtures/database/invalid.db");
-
-    std::promise<void> promise;
+    SQLiteCache cache("test/fixtures/database/invalid.db");
 
     {
         // Adds a file.
         Log::setObserver(util::make_unique<FixtureLogObserver>());
-        promise = {};
-        auto response = std::make_shared<Response>();
-        response->data = "Demo";
-        cache->put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
-        cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
-            EXPECT_NE(nullptr, res.get());
-            EXPECT_EQ("Demo", res->data);
-            promise.set_value();
+
+        util::RunLoop loop;
+
+        loop.invoke([&] {
+            auto response = std::make_shared<Response>();
+            response->data = "Demo";
+            cache.put({ Resource::Unknown, "mapbox://test" }, response, FileCache::Hint::Full);
+            cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+                EXPECT_NE(nullptr, res.get());
+                EXPECT_EQ("Demo", res->data);
+                loop.stop();
+            });
         });
-        promise.get_future().get();
+
+        loop.run();
 
         auto observer = Log::removeObserver();
         auto flo = dynamic_cast<FixtureLogObserver*>(observer.get());
         EXPECT_EQ(1ul, flo->count({ EventSeverity::Warning, Event::Database, -1, "Trashing invalid database" }));
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }

test/storage/directory_reading.cpp

@@ -10,9 +10,9 @@ TEST_F(Storage, AssetReadDirectory) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    DefaultFileSource fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);

test/storage/file_reading.cpp

@@ -11,9 +11,9 @@ TEST_F(Storage, AssetEmptyFile) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    DefaultFileSource fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);
@@ -38,9 +38,9 @@ TEST_F(Storage, AssetNonEmptyFile) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    DefaultFileSource fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);
@@ -66,9 +66,9 @@ TEST_F(Storage, AssetNonExistentFile) {
     using namespace mbgl;
 
 #ifdef MBGL_ASSET_ZIP
-    DefaultFileSource fs(nullptr, uv_default_loop(), "test/fixtures/storage/assets.zip");
+    DefaultFileSource fs(nullptr, "test/fixtures/storage/assets.zip");
 #else
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 #endif
 
     auto &env = *static_cast<const Environment *>(nullptr);

test/storage/http_cancel.cpp

@@ -12,7 +12,7 @@ TEST_F(Storage, HTTPCancel) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
@@ -31,7 +31,7 @@ TEST_F(Storage, HTTPCancelMultiple) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/test" };

test/storage/http_coalescing.cpp

@@ -12,7 +12,7 @@ TEST_F(Storage, HTTPCoalescing) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 

test/storage/http_environment.cpp

@@ -13,7 +13,7 @@ TEST_F(Storage, HTTPCancelEnvironment) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     // Create two fake environment pointers. The FileSource implementation treats these as opaque
     // pointers and doesn't reach into them.

test/storage/http_error.cpp

@@ -20,7 +20,7 @@ TEST_F(Storage, HTTPError) {
     }, 500, 500);
     uv_unref((uv_handle_t *)&statusChange);
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 

test/storage/http_header_parsing.cpp

@@ -13,7 +13,7 @@ TEST_F(Storage, HTTPHeaderParsing) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 

test/storage/http_load.cpp

@@ -9,7 +9,7 @@ TEST_F(Storage, HTTPLoad) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 

test/storage/http_reading.cpp

@@ -4,13 +4,15 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 
+#include <future>
+
 TEST_F(Storage, HTTPReading) {
     SCOPED_TEST(HTTPTest)
     SCOPED_TEST(HTTP404)
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
@@ -47,7 +49,7 @@ TEST_F(Storage, HTTPNoCallback) {
 
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
@@ -59,18 +61,21 @@ TEST_F(Storage, HTTPNoCallback) {
     HTTPTest.finish();
 }
 
-TEST_F(Storage, HTTPNoCallbackNoLoop) {
-    SCOPED_TEST(HTTPTest)
-
+TEST_F(Storage, HTTPCallbackNotOnLoop) {
     using namespace mbgl;
 
-    DefaultFileSource fs(nullptr, uv_default_loop());
+    DefaultFileSource fs(nullptr);
+
+    SCOPED_TEST(HTTPTest)
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
-    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, env, nullptr);
+    std::promise<void> promise;
+    fs.request({ Resource::Unknown, "http://127.0.0.1:3000/test" }, env, [&] (const Response &) {
+        promise.set_value();
+    });
 
-    uv_run(uv_default_loop(), UV_RUN_DEFAULT);
+    promise.get_future().get();
 
     HTTPTest.finish();
 }