add Thread<T> and RunLoop

Thread<T> is a generic thread management object that starts a thread, creates an object T in that thread and calls its .start() function. When the Thread<T> object is destructed (which must happen in the thread that created it), it'll call the object's .stop() function, and wait for thread termination. The .stop() function must somehow get the thread to terminate. Note that object T's constructor/destructor/start()/stop() must be protected/private, and Thread<T> must be declared as a friend class. All public functions may be called from any thread and are exposed through operator->().

RunLoop encapsulates a libuv runloop and has a facility of dispatching function objects to the thread. It can be used either as a worker thread class by itself, or it can be derived from. This commit converts SQLiteCache to derive from RunLoop and removes the custom code used previously for managing the cache thread.

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();
 }