17 files changed, 279 insertions, 180 deletions

diff --git a/android/cpp/native_map_view.cpp b/android/cpp/native_map_view.cpp
index 39a777bff2..340bbae10b 100644
--- a/android/cpp/native_map_view.cpp
+++ b/android/cpp/native_map_view.cpp
@@ -58,7 +58,7 @@ void log_gl_string(GLenum name, const char *label) {
 NativeMapView::NativeMapView(JNIEnv *env, jobject obj_)
     : mbgl::View(*this),
       fileCache(mbgl::android::cachePath + "/mbgl-cache.db"),
-      fileSource(&fileCache),
+      fileSource(fileCache),
       map(*this, fileSource) {
     mbgl::Log::Debug(mbgl::Event::Android, "NativeMapView::NativeMapView");
 
diff --git a/bin/render.cpp b/bin/render.cpp
index 01f6929092..879bf840a2 100644
--- a/bin/render.cpp
+++ b/bin/render.cpp
@@ -2,6 +2,7 @@
 #include <mbgl/util/image.hpp>
 #include <mbgl/util/std.hpp>
 #include <mbgl/util/io.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <mbgl/platform/default/headless_view.hpp>
 #include <mbgl/platform/default/headless_display.hpp>
@@ -66,8 +67,8 @@ int main(int argc, char *argv[]) {
 
     using namespace mbgl;
 
-    mbgl::SQLiteCache cache(cache_file);
-    mbgl::DefaultFileSource fileSource(&cache);
+    mbgl::util::Thread<mbgl::SQLiteCache> cache(cache_file);
+    mbgl::DefaultFileSource fileSource(cache);
 
     // Try to load the token from the environment.
     if (!token.size()) {
diff --git a/include/mbgl/android/native_map_view.hpp b/include/mbgl/android/native_map_view.hpp
index 21784f5315..4a8d9365bf 100644
--- a/include/mbgl/android/native_map_view.hpp
+++ b/include/mbgl/android/native_map_view.hpp
@@ -4,6 +4,7 @@
 #include <mbgl/map/map.hpp>
 #include <mbgl/map/view.hpp>
 #include <mbgl/util/noncopyable.hpp>
+#include <mbgl/util/thread.hpp>
 #include <mbgl/storage/default/sqlite_cache.hpp>
 #include <mbgl/storage/default_file_source.hpp>
 
@@ -61,7 +62,7 @@ private:
 
     ANativeWindow *window = nullptr;
 
-    mbgl::SQLiteCache fileCache;
+    mbgl::util::Thread<mbgl::SQLiteCache> fileCache;
     mbgl::DefaultFileSource fileSource;
     mbgl::Map map;
 
diff --git a/include/mbgl/storage/default/sqlite_cache.hpp b/include/mbgl/storage/default/sqlite_cache.hpp
index fe80a41b52..899cc8a892 100644
--- a/include/mbgl/storage/default/sqlite_cache.hpp
+++ b/include/mbgl/storage/default/sqlite_cache.hpp
@@ -2,50 +2,40 @@
 #define MBGL_STORAGE_DEFAULT_SQLITE_CACHE
 
 #include <mbgl/storage/file_cache.hpp>
+#include <mbgl/util/run_loop.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>;
+class SQLiteCache : public FileCache, protected util::RunLoop {
+    friend class util::Thread<SQLiteCache>;
 
-public:
+private:
     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);
+public:
+    // 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:
-    struct ActionDispatcher;
-    void process(GetAction &action);
-    void process(PutAction &action);
-    void process(RefreshAction &action);
-    void process(StopAction &action);
-
     void createDatabase();
     void createSchema();
 
+    void processGet(const Resource& resource, Callback callback);
+    void processPut(const Resource& resource, std::shared_ptr<const Response> response);
+    void processRefresh(const Resource& resource, int64_t expires);
+
+private:
     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;
+    std::unique_ptr<::mapbox::sqlite::Statement> getStmt;
+    std::unique_ptr<::mapbox::sqlite::Statement> putStmt;
+    std::unique_ptr<::mapbox::sqlite::Statement> refreshStmt;
     bool schema = false;
 };
 
diff --git a/include/mbgl/storage/file_cache.hpp b/include/mbgl/storage/file_cache.hpp
index 97e75a5d85..f815d5b8c2 100644
--- a/include/mbgl/storage/file_cache.hpp
+++ b/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;
 };
 
diff --git a/include/mbgl/util/run_loop.hpp b/include/mbgl/util/run_loop.hpp
new file mode 100644
index 0000000000..f0318d2026
--- /dev/null
+++ b/include/mbgl/util/run_loop.hpp
@@ -0,0 +1,52 @@
+#ifndef MBGL_UTIL_RUN_LOOP
+#define MBGL_UTIL_RUN_LOOP
+
+#include <mutex>
+#include <functional>
+#include <queue>
+
+namespace uv {
+class async;
+class loop;
+}
+
+namespace mbgl {
+namespace util {
+
+template <typename T> class Thread;
+
+class RunLoop {
+    friend Thread<RunLoop>;
+
+protected:
+    // These are called by the Thread<> wrapper.
+    RunLoop();
+    ~RunLoop();
+
+    // Called by the Thread<> wrapper to start the loop. When you implement this
+    // method in a child class, you *must* call this function as the last action.
+    void start();
+
+protected:
+    // Called by the Thread<> wrapper to terminate this loop.
+    void stop();
+
+private:
+    // Invokes function in the run loop.
+    void process();
+
+public:
+    // Schedules a function to be executed as part of this run loop.
+    void invoke(std::function<void()> fn);
+
+private:
+    const std::unique_ptr<uv::loop> runloop;
+    const std::unique_ptr<uv::async> runloopAsync;
+    std::mutex runloopMutex;
+    std::queue<std::function<void()>> runloopQueue;
+};
+
+}
+}
+
+#endif
\ No newline at end of file
diff --git a/include/mbgl/util/thread.hpp b/include/mbgl/util/thread.hpp
new file mode 100644
index 0000000000..943039c5d6
--- /dev/null
+++ b/include/mbgl/util/thread.hpp
@@ -0,0 +1,52 @@
+#ifndef MBGL_UTIL_THREAD
+#define MBGL_UTIL_THREAD
+
+#include <future>
+#include <thread>
+
+namespace mbgl {
+namespace util {
+
+template <class Object>
+class Thread {
+public:
+    template <class... Args>
+    Thread(Args&&... args);
+    Thread(const Thread&) = delete;
+    Thread(Thread&&) = delete;
+    Thread& operator=(const Thread&) = delete;
+    Thread& operator=(Thread&&) = delete;
+    ~Thread();
+
+    inline Object* operator->() const { return &object; }
+    inline operator Object*() const { return &object; }
+
+private:
+    std::thread thread;
+    Object& object;
+};
+
+template <class Object>
+template <class... Args>
+Thread<Object>::Thread(Args&&... args)
+    : object([&]() -> Object& {
+          std::promise<Object&> promise;
+          thread = std::thread([&] {
+              Object context(::std::forward<Args>(args)...);
+              promise.set_value(context);
+              context.start();
+          });
+          return promise.get_future().get();
+      }()) {
+}
+
+template <class Object>
+Thread<Object>::~Thread() {
+    object.stop();
+    thread.join();
+}
+
+}
+}
+
+#endif
diff --git a/linux/main.cpp b/linux/main.cpp
index 4d5474c02f..71c2424af5 100644
--- a/linux/main.cpp
+++ b/linux/main.cpp
@@ -7,6 +7,7 @@
 #include <mbgl/platform/default/glfw_view.hpp>
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <signal.h>
 #include <getopt.h>
@@ -65,8 +66,8 @@ int main(int argc, char *argv[]) {
 
     view = mbgl::util::make_unique<GLFWView>();
 
-    mbgl::SQLiteCache cache("/tmp/mbgl-cache.db");
-    mbgl::DefaultFileSource fileSource(&cache);
+    mbgl::util::Thread<mbgl::SQLiteCache> cache("/tmp/mbgl-cache.db");
+    mbgl::DefaultFileSource fileSource(cache);
     mbgl::Map map(*view, fileSource);
 
     // Load settings
diff --git a/macosx/main.mm b/macosx/main.mm
index 087544f7c4..fb12282e28 100644
--- a/macosx/main.mm
+++ b/macosx/main.mm
@@ -6,6 +6,7 @@
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/default/sqlite_cache.hpp>
 #include <mbgl/storage/network_status.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <mbgl/util/geo.hpp>
 
@@ -103,8 +104,8 @@ const std::string &defaultCacheDatabase() {
 int main() {
     GLFWView view;
 
-    mbgl::SQLiteCache cache(defaultCacheDatabase());
-    mbgl::DefaultFileSource fileSource(&cache);
+    mbgl::util::Thread<mbgl::SQLiteCache> cache(defaultCacheDatabase());
+    mbgl::DefaultFileSource fileSource(cache);
     mbgl::Map map(view, fileSource);
 
     URLHandler *handler = [[URLHandler alloc] init];
diff --git a/platform/default/sqlite_cache.cpp b/platform/default/sqlite_cache.cpp
index a3114098c8..b084a7f2a7 100644
--- a/platform/default/sqlite_cache.cpp
+++ b/platform/default/sqlite_cache.cpp
@@ -2,9 +2,6 @@
 #include <mbgl/storage/default/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/platform/log.hpp>
@@ -12,10 +9,6 @@
 #include "sqlite3.hpp"
 #include <sqlite3.h>
 
-#include <uv.h>
-
-#include <cassert>
-
 namespace mbgl {
 
 std::string removeAccessTokenFromURL(const std::string &url) {
@@ -67,90 +60,28 @@ 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]() {
+    : path(path_) {
 #ifdef __APPLE__
-          pthread_setname_np("SQLite Cache");
+    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());
-          }
-      }) {
 }
 
 SQLiteCache::~SQLiteCache() {
-    if (thread.joinable()) {
-        if (queue) {
-            queue->send(StopAction{ });
-        }
-        thread.join();
-        uv_loop_delete(loop);
-    }
-}
-
-
-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 });
+    // 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::createDatabase() {
     db = util::make_unique<Database>(path.c_str(), ReadWrite | Create);
-
-    createSchema();
 }
 
 void SQLiteCache::createSchema() {
@@ -171,7 +102,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 +122,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.
+    invoke(std::bind(&SQLiteCache::processGet, this, resource, callback));
+}
+
+void SQLiteCache::processGet(const Resource &resource, Callback callback) {
     try {
         // This is called in the SQLite event loop.
         if (!db) {
@@ -212,7 +150,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 +163,29 @@ void SQLiteCache::process(GetAction &action) {
             if (getStmt->get<int>(5)) { // == compressed
                 response->data = util::decompress(response->data);
             }
-            action.callback(std::move(response));
+            callback(std::move(response));
         } else {
             // There is no data.
-            action.callback(nullptr);
+            callback(nullptr);
         }
     } catch (mapbox::sqlite::Exception& ex) {
         Log::Error(Event::Database, ex.code, ex.what());
-        action.callback(nullptr);
+        callback(nullptr);
+    }
+}
+
+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) {
+        invoke(std::bind(&SQLiteCache::processPut, this, resource, response));
+    } else if (hint == Hint::Refresh) {
+        invoke(std::bind(&SQLiteCache::processRefresh, this, resource, response->expires));
     }
 }
 
-void SQLiteCache::process(PutAction &action) {
+void SQLiteCache::processPut(const Resource& resource, std::shared_ptr<const Response> response) {
     try {
         if (!db) {
             createDatabase();
@@ -255,37 +204,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::processRefresh(const Resource& resource, int64_t expires) {
     try {
         if (!db) {
             createDatabase();
@@ -302,8 +251,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 +260,4 @@ void SQLiteCache::process(RefreshAction &action) {
     }
 }
 
-void SQLiteCache::process(StopAction &) {
-    assert(queue);
-    queue->stop();
-    queue = nullptr;
-}
-
 }
diff --git a/platform/ios/MGLMapView.mm b/platform/ios/MGLMapView.mm
index f70f9f348f..662b371b8c 100644
--- a/platform/ios/MGLMapView.mm
+++ b/platform/ios/MGLMapView.mm
@@ -14,6 +14,7 @@
 #include <mbgl/storage/default/sqlite_cache.hpp>
 #include <mbgl/storage/network_status.hpp>
 #include <mbgl/util/geo.hpp>
+#include <mbgl/util/thread.hpp>
 
 #import "MGLTypes.h"
 #import "NSString+MGLAdditions.h"
@@ -111,7 +112,7 @@ std::chrono::steady_clock::duration secondsAsDuration(float duration)
 
 mbgl::Map *mbglMap = nullptr;
 MBGLView *mbglView = nullptr;
-mbgl::SQLiteCache *mbglFileCache = nullptr;
+mbgl::util::Thread<mbgl::SQLiteCache> *mbglFileCache = nullptr;
 mbgl::DefaultFileSource *mbglFileSource = nullptr;
 
 - (instancetype)initWithFrame:(CGRect)frame
@@ -269,8 +270,8 @@ mbgl::DefaultFileSource *mbglFileSource = nullptr;
     // setup mbgl map
     //
     mbglView = new MBGLView(self);
-    mbglFileCache  = new mbgl::SQLiteCache(defaultCacheDatabase());
-    mbglFileSource = new mbgl::DefaultFileSource(mbglFileCache);
+    mbglFileCache  = new mbgl::util::Thread<mbgl::SQLiteCache>(defaultCacheDatabase());
+    mbglFileSource = new mbgl::DefaultFileSource(*mbglFileCache);
     mbglMap = new mbgl::Map(*mbglView, *mbglFileSource);
     mbglView->resize(self.bounds.size.width, self.bounds.size.height, _glView.contentScaleFactor, _glView.drawableWidth, _glView.drawableHeight);
 
diff --git a/src/mbgl/storage/default_file_source.cpp b/src/mbgl/storage/default_file_source.cpp
index ca8d423b1b..841a56bcef 100644
--- a/src/mbgl/storage/default_file_source.cpp
+++ b/src/mbgl/storage/default_file_source.cpp
@@ -122,11 +122,7 @@ Request *DefaultFileSource::request(const Resource &resource, uv_loop_t *l, cons
 
 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 });
+    request(resource, nullptr, env, std::move(callback));
 }
 
 void DefaultFileSource::cancel(Request *req) {
diff --git a/src/mbgl/util/run_loop.cpp b/src/mbgl/util/run_loop.cpp
new file mode 100644
index 0000000000..3a7cf597ff
--- /dev/null
+++ b/src/mbgl/util/run_loop.cpp
@@ -0,0 +1,55 @@
+#include <mbgl/util/run_loop.hpp>
+#include <mbgl/util/uv_detail.hpp>
+#include <mbgl/util/std.hpp>
+
+#include <uv.h>
+
+namespace /* anonymous */ {
+inline void critical_section(std::mutex& mutex, std::function<void()> fn) {
+    std::lock_guard<std::mutex> lock(mutex);
+    fn();
+}
+}
+
+namespace mbgl {
+namespace util {
+
+RunLoop::RunLoop()
+    : runloop(util::make_unique<uv::loop>()),
+      runloopAsync(util::make_unique<uv::async>(runloop->get(), std::bind(&RunLoop::process, this))) {
+}
+
+// Define here since we can't destroy the uv::* objects from just the header file.
+RunLoop::~RunLoop() = default;
+
+void RunLoop::start() {
+    runloop->run();
+}
+
+void RunLoop::stop() {
+    critical_section(runloopMutex, [this] { runloopQueue.push(nullptr); });
+    runloopAsync->send();
+}
+
+void RunLoop::process() {
+    std::queue<std::function<void()>> queue;
+    critical_section(runloopMutex, [this, &queue] { queue.swap(runloopQueue); });
+    while (!queue.empty()) {
+        if (queue.front()) {
+            queue.front()();
+        } else {
+            runloopAsync->unref();
+        }
+        queue.pop();
+    }
+}
+
+void RunLoop::invoke(std::function<void()> fn) {
+    if (fn) {
+        critical_section(runloopMutex, [this, &fn] { runloopQueue.push(fn); });
+        runloopAsync->send();
+    }
+}
+
+}
+}
diff --git a/src/mbgl/util/uv_detail.hpp b/src/mbgl/util/uv_detail.hpp
index 9d479da425..6acef1b386 100644
--- a/src/mbgl/util/uv_detail.hpp
+++ b/src/mbgl/util/uv_detail.hpp
@@ -41,10 +41,19 @@ public:
         uv_loop_close(l);
         delete l;
 #endif
+    }
+
+    inline void run() {
+        uv_run(l, UV_RUN_DEFAULT);
+    }
 
+    inline uv_loop_t* operator*() {
+        return l;
     }
 
-    inline uv_loop_t *operator*() { return l; }
+    inline uv_loop_t* get() {
+        return l;
+    }
 
 private:
     uv_loop_t *l = nullptr;
@@ -72,6 +81,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) {
diff --git a/test/storage/cache_response.cpp b/test/storage/cache_response.cpp
index ac0dc4c565..f5d7c4bf3c 100644
--- a/test/storage/cache_response.cpp
+++ b/test/storage/cache_response.cpp
@@ -4,14 +4,15 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, CacheResponse) {
     SCOPED_TEST(CacheResponse);
 
     using namespace mbgl;
 
-    SQLiteCache cache(":memory:");
-    DefaultFileSource fs(&cache, uv_default_loop());
+    util::Thread<SQLiteCache> cache(":memory:");
+    DefaultFileSource fs(cache, uv_default_loop());
 
     const Resource resource { Resource::Unknown, "http://127.0.0.1:3000/cache" };
     auto &env = *static_cast<const Environment *>(nullptr);
diff --git a/test/storage/cache_revalidate.cpp b/test/storage/cache_revalidate.cpp
index bd32042b94..29f25eb775 100644
--- a/test/storage/cache_revalidate.cpp
+++ b/test/storage/cache_revalidate.cpp
@@ -4,6 +4,7 @@
 
 #include <mbgl/storage/default_file_source.hpp>
 #include <mbgl/storage/default/sqlite_cache.hpp>
+#include <mbgl/util/thread.hpp>
 
 TEST_F(Storage, CacheRevalidate) {
     SCOPED_TEST(CacheRevalidateSame)
@@ -12,8 +13,8 @@ TEST_F(Storage, CacheRevalidate) {
 
     using namespace mbgl;
 
-    SQLiteCache cache(":memory:");
-    DefaultFileSource fs(&cache);
+    util::Thread<SQLiteCache> cache(":memory:");
+    DefaultFileSource fs(cache);
 
     auto &env = *static_cast<const Environment *>(nullptr);
 
diff --git a/test/storage/database.cpp b/test/storage/database.cpp
index 1a2b618a57..ffa82ecb98 100644
--- a/test/storage/database.cpp
+++ b/test/storage/database.cpp
@@ -7,6 +7,7 @@
 #include <mbgl/storage/resource.hpp>
 #include <mbgl/storage/response.hpp>
 #include <mbgl/util/io.hpp>
+#include <mbgl/util/thread.hpp>
 
 #include <sqlite3.h>
 
@@ -41,9 +42,9 @@ TEST_F(Storage, DatabaseDoesNotExist) {
         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");
+    util::Thread<SQLiteCache> cache("test/fixtures/404/cache.db");
 
-    cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+    cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
         EXPECT_EQ(nullptr, res.get());
         test.finish();
     });
@@ -85,9 +86,9 @@ TEST_F(Storage, DatabaseCreate) {
         Log::removeObserver();
     });
 
-    SQLiteCache cache("test/fixtures/database/cache.db");
+    util::Thread<SQLiteCache> cache("test/fixtures/database/cache.db");
 
-    cache.get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
+    cache->get({ Resource::Unknown, "mapbox://test" }, [&] (std::unique_ptr<Response> res) {
         EXPECT_EQ(nullptr, res.get());
         test.finish();
     });
@@ -140,7 +141,8 @@ 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");
+
+    util::Thread<SQLiteCache> cache("test/fixtures/database/locked.db");
 
     std::promise<void> promise;
 
@@ -177,9 +179,6 @@ TEST_F(Storage, DatabaseLockedRead) {
         // Make sure that we got a no errors
         Log::removeObserver();
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }
 
 
@@ -192,7 +191,7 @@ 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");
+    util::Thread<SQLiteCache> cache("test/fixtures/database/locked.db");
 
     std::promise<void> promise;
 
@@ -234,9 +233,6 @@ TEST_F(Storage, DatabaseLockedWrite) {
         // Make sure that we got a no errors
         Log::removeObserver();
     }
-
-    // Explicitly delete the Cache now.
-    cache.reset();
 }
 
 
@@ -249,7 +245,7 @@ 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");
+    util::Thread<SQLiteCache> cache("test/fixtures/database/locked.db");
 
     // Then, lock the file and try again.
     FileLock guard("test/fixtures/database/locked.db");
@@ -293,9 +289,6 @@ TEST_F(Storage, DatabaseLockedRefresh) {
         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,7 +300,7 @@ 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");
+    util::Thread<SQLiteCache> cache("test/fixtures/database/locked.db");
 
     std::promise<void> promise;
 
@@ -348,9 +341,6 @@ TEST_F(Storage, DatabaseDeleted) {
         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,7 +353,7 @@ 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");
+    util::Thread<SQLiteCache> cache("test/fixtures/database/invalid.db");
 
     std::promise<void> promise;
 
@@ -385,7 +375,4 @@ TEST_F(Storage, DatabaseInvalid) {
         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();
 }