diff --git a/common/curl_request.cpp b/common/curl_request.cpp new file mode 100644 index 000000000..dc9db0269 --- /dev/null +++ b/common/curl_request.cpp @@ -0,0 +1,428 @@ +#include "curl_request.hpp" + +#include +#include +#include + +#include +#include + +#include + +// This file contains code from http://curl.haxx.se/libcurl/c/multi-uv.html: + +/*************************************************************************** + * _ _ ____ _ + * Project ___| | | | _ \| | + * / __| | | | |_) | | + * | (__| |_| | _ <| |___ + * \___|\___/|_| \_\_____| + * + * Copyright (C) 1998 - 2013, Daniel Stenberg, , et al. + * + * This software is licensed as described in the file COPYING, which + * you should have received as part of this distribution. The terms + * are also available at http://curl.haxx.se/docs/copyright.html. + * + * You may opt to use, copy, modify, merge, publish, distribute and/or sell + * copies of the Software, and permit persons to whom the Software is + * furnished to do so, under the terms of the COPYING file. + * + * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY + * KIND, either express or implied. + * + ***************************************************************************/ + +/* Example application code using the multi socket interface to download + multiple files at once, but instead of using curl_multi_perform and + curl_multi_wait, which uses select(), we use libuv. + It supports epoll, kqueue, etc. on unixes and fast IO completion ports on + Windows, which means, it should be very fast on all platforms.. + + Written by Clemens Gruber, based on an outdated example from uvbook and + some tests from libuv. + + Requires libuv and (of course) libcurl. + + See http://nikhilm.github.com/uvbook/ for more information on libuv. +*/ + +namespace llmr { +namespace platform { +namespace request { + +struct curl_context { + uv_poll_t poll_handle; + curl_socket_t sockfd; +}; + +// Handles the request thread + messaging to the thread. +static uv_thread_t thread; +static uv::once init_thread_once; +static uv_loop_t *loop = nullptr; +static uv_async_t async_add; +static uv_async_t async_cancel; + +// Stores pointers (!) to shared_ptrs. We use shared_ptrs so that request objects don't get +// auto-destructed while they're in progress. The TileData object retains a weak_ptr to this +// request, so we have to use a shared_ptr here to ensure that this object stays alive. +static boost::lockfree::queue *> add_queue(8); +static boost::lockfree::queue *> cancel_queue(8); + +// Used as the CURL timer function to periodically check for socket updates. +static uv_timer_t timeout; + +// CURL multi handle that we use to request multiple URLs at the same time, without having to block +// and spawn threads. +static CURLM *curl_multi = nullptr; + +// CURL share handles are used for sharing session state (e.g.) +static uv::mutex curl_share_mutex; +static CURLSH *curl_share = nullptr; + +// A queue that we use for storing resuable CURL easy handles to avoid creating and destroying them +// all the time. +static std::queue curl_handle_cache; + +// Implementation starts here. + +// Locks the CURL share handle +void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) { curl_share_mutex.lock(); } + +// Unlocks the CURL share handle +void curl_share_unlock(CURL *, curl_lock_data, void *) { curl_share_mutex.unlock(); } + +curl_context *create_curl_context(curl_socket_t sockfd) { + curl_context *context = new curl_context; + context->sockfd = sockfd; + + uv_poll_init_socket(loop, &context->poll_handle, sockfd); + context->poll_handle.data = context; + + return context; +} + +void curl_close_cb(uv_handle_t *handle) { + curl_context *context = (curl_context *)handle->data; + free(context); +} + +void destroy_curl_context(curl_context *context) { + uv_close((uv_handle_t *)&context->poll_handle, curl_close_cb); +} + +void remove_curl_handle(CURL *handle) { + CURLMcode error = curl_multi_remove_handle(curl_multi, handle); + if (error != CURLM_OK) { + throw std::runtime_error(std::string("CURL multi error: ") + curl_multi_strerror(error)); + } + + curl_easy_reset(handle); + curl_handle_cache.push(handle); +} + +void curl_perform(uv_poll_t *req, int /*status*/, int events) { + int running_handles; + int flags = 0; + curl_context *context; + CURLMsg *message; + int pending; + + uv_timer_stop(&timeout); + + if (events & UV_READABLE) + flags |= CURL_CSELECT_IN; + if (events & UV_WRITABLE) + flags |= CURL_CSELECT_OUT; + + context = (curl_context *)req; + + curl_multi_socket_action(curl_multi, context->sockfd, flags, &running_handles); + + while ((message = curl_multi_info_read(curl_multi, &pending))) { + switch (message->msg) { + case CURLMSG_DONE: { + std::shared_ptr *req = nullptr; + curl_easy_getinfo(message->easy_handle, CURLINFO_PRIVATE, (char *)&req); + + // Add human-readable error code + if (message->data.result != CURLE_OK) { + (*req)->res->error_message = curl_easy_strerror(message->data.result); + (*req)->res->code = -1; + } else { + curl_easy_getinfo(message->easy_handle, CURLINFO_RESPONSE_CODE, &(*req)->res->code); + } + + // We're currently in the CURL request thread. We're going to schedule a uv_work request + // that executes the background function in a threadpool, and tell it to call the + // after callback back in the main uv loop. + + uv_work_t *work = new uv_work_t(); + + // We're passing on the pointer we created to the work structure. + // It is going to be deleted in the after_work_cb; + work->data = req; + + // Executes the background_function in a libuv thread pool, and the after_work_cb back + // in the *main* event loop. + uv_queue_work(uv_default_loop(), work, Request::work_cb, Request::after_work_cb); + + CURL *handle = message->easy_handle; + remove_curl_handle(handle); + + // We're setting this to NULL because there might still be shared_ptrs around that could + // be cancelled. + (*req)->curl = nullptr; + + break; + } + + default: + // This should never happen, because there are no other message types. + throw std::runtime_error("CURLMSG returned unknown message type"); + } + } +} + +int handle_socket(CURL * /*easy*/, curl_socket_t s, int action, void * /*userp*/, void *socketp) { + curl_context *context; + if (action == CURL_POLL_IN || action == CURL_POLL_OUT) { + if (socketp) { + context = (curl_context *)socketp; + } else { + context = create_curl_context(s); + } + curl_multi_assign(curl_multi, s, (void *)context); + } + + switch (action) { + case CURL_POLL_IN: + uv_poll_start(&context->poll_handle, UV_READABLE, curl_perform); + break; + case CURL_POLL_OUT: + uv_poll_start(&context->poll_handle, UV_WRITABLE, curl_perform); + break; + case CURL_POLL_REMOVE: + if (socketp) { + uv_poll_stop(&((curl_context *)socketp)->poll_handle); + destroy_curl_context((curl_context *)socketp); + curl_multi_assign(curl_multi, s, NULL); + } + break; + default: + abort(); + } + + return 0; +} + +void on_timeout(uv_timer_t * /*req*/) { + int running_handles; + CURLMcode error = + curl_multi_socket_action(curl_multi, CURL_SOCKET_TIMEOUT, 0, &running_handles); + if (error != CURLM_OK) { + throw std::runtime_error(std::string("CURL multi error: ") + curl_multi_strerror(error)); + } +} + +void start_timeout(CURLM * /*multi*/, long timeout_ms, void * /*userp*/) { + if (timeout_ms <= 0) { + on_timeout(&timeout); + } else { + uv_timer_start(&timeout, on_timeout, timeout_ms, 0); + } +} + +// This function is the first function called in the request thread. It sets up the CURL share/multi +// handles and runs the thread loop. +void thread_init(void * /*ptr*/) { + uv_timer_init(loop, &timeout); + + CURLSHcode share_error; + curl_share = curl_share_init(); + + share_error = curl_share_setopt(curl_share, CURLSHOPT_LOCKFUNC, curl_share_lock); + if (share_error != CURLSHE_OK) { + throw std::runtime_error(std::string("CURL share error: ") + curl_share_strerror(share_error)); + } + + share_error = curl_share_setopt(curl_share, CURLSHOPT_UNLOCKFUNC, curl_share_unlock); + if (share_error != CURLSHE_OK) { + throw std::runtime_error(std::string("CURL share error: ") + curl_share_strerror(share_error)); + } + + CURLMcode multi_error; + curl_multi = curl_multi_init(); + + multi_error = curl_multi_setopt(curl_multi, CURLMOPT_SOCKETFUNCTION, handle_socket); + if (multi_error != CURLM_OK) { + throw std::runtime_error(std::string("CURL multi error: ") + curl_multi_strerror(multi_error)); + } + multi_error = curl_multi_setopt(curl_multi, CURLMOPT_TIMERFUNCTION, start_timeout); + if (multi_error != CURLM_OK) { + throw std::runtime_error(std::string("CURL multi error: ") + curl_multi_strerror(multi_error)); + + } + + // Main event loop. This will not return until the request loop is terminated. + uv_run(loop, UV_RUN_DEFAULT); + + curl_multi_cleanup(curl_multi); + curl_multi = nullptr; + curl_share_cleanup(curl_share); + curl_share = nullptr; + + // Clean up all the CURL easy handles that we kept around for potential future reuse. + while (!curl_handle_cache.empty()) { + curl_easy_cleanup(curl_handle_cache.front()); + curl_handle_cache.pop(); + } +} + +// This function is called when we have new data for a request. We just append it to the string +// containing the previous data. +size_t curl_write_cb(void *contents, size_t size, size_t nmemb, void *userp) { + ((std::string *)userp)->append((char *)contents, size * nmemb); + return size * nmemb; +} + +// This callback is called in the request event loop (on the request thread). +// It initializes newly queued up download requests and adds them to the CURL +// multi handle. +void async_add_cb(uv_async_t * /*async*/) { + std::shared_ptr *req = nullptr; + while (add_queue.pop(req)) { + // Make sure that we're not starting requests that have been cancelled + // already by async_cancel_cb. + if ((*req)->cancelled) { + delete req; + continue; + } + + (*req)->res = std::make_unique(); + + // Obtain a curl handle (and try to reuse existing handles before creating new ones). + CURL *handle = nullptr; + if (!curl_handle_cache.empty()) { + handle = curl_handle_cache.front(); + curl_handle_cache.pop(); + } else { + handle = curl_easy_init(); + } + + (*req)->curl = handle; + + curl_easy_setopt(handle, CURLOPT_PRIVATE, req); + curl_easy_setopt(handle, CURLOPT_URL, (*req)->url.c_str()); + curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curl_write_cb); + curl_easy_setopt(handle, CURLOPT_WRITEDATA, &(*req)->res->body); + curl_easy_setopt(handle, CURLOPT_ACCEPT_ENCODING, "deflate"); + curl_easy_setopt(handle, CURLOPT_SHARE, curl_share); + curl_multi_add_handle(curl_multi, handle); + } +} + +void async_cancel_cb(uv_async_t * /*async*/) { + std::shared_ptr *req = nullptr; + while (cancel_queue.pop(req)) { + // It is possible that the request has not yet been started, but that it already has been + // added to the queue for scheduling new requests. In this case, the CURL handle is invalid + // and we manually mark the Request as cancelled. + CURL *handle = (*req)->curl; + if (handle && !(*req)->cancelled) { + remove_curl_handle(handle); + (*req)->curl = nullptr; + } + (*req)->cancelled = true; + + delete req; + req = nullptr; + } +} + +void thread_init_cb() { + loop = uv_loop_new(); + uv_async_init(loop, &async_add, async_add_cb); + uv_async_init(loop, &async_cancel, async_cancel_cb); + uv_thread_create(&thread, thread_init, nullptr); +} +} // end namespace request +} // end namespace platform +} // end namespace llmr + +using namespace llmr::platform; + +Request::Request(const std::string &url, + std::function background_function, + std::function foreground_callback) + : url(url), + background_function(background_function), + foreground_callback(foreground_callback) { + + // Add a check handle without a callback to keep the default loop running. + // We don't have a real handler attached to the default loop right from the + // beginning, because we're using asynchronous messaging to perform the actual + // CURL request in the request thread. Only after the request is complete, we + // create an actual work request that is attached to the default loop. + check = new uv_check_t(); + uv_check_init(uv_default_loop(), check); + uv_check_start(check, [](uv_check_t *) {}); +} + +Request::~Request() { + // We need to remove our no-op handle again to allow the main event loop to exit. + uv_check_stop(check); +} + +void Request::work_cb(uv_work_t *work) { + std::shared_ptr &req = *static_cast *>(work->data); + req->background_function(req->res.get()); +} + +// This callback is executed in the main loop. +void Request::after_work_cb(uv_work_t *work, int /*status*/) { + std::shared_ptr &req = *static_cast *>(work->data); + + req->foreground_callback(); + + // This finally deletes the *pointer* to the shared pointer we've been holding on since we + // pushed it on the add_queue on request creation. + delete static_cast *>(work->data); +} + + + + + +namespace llmr { + +std::shared_ptr +platform::request_http(const std::string &url, + std::function background_function, + std::function foreground_callback) { + using namespace request; + init_thread_once(thread_init_cb); + std::shared_ptr req = std::make_shared(url, background_function, foreground_callback); + + // Note that we are creating a new shared_ptr pointer(!) because the lockless queue can't store + // objects with nontrivial destructors. We have to make absolutely sure that we manually delete + // the shared_ptr when we pop it from the queue. + add_queue.push(new std::shared_ptr(req)); + uv_async_send(&async_add); + + return req; +} + +// Cancels an HTTP request. +void platform::cancel_request_http(const std::shared_ptr &req) { + if (req) { + using namespace request; + + // Note that we are creating a new shared_ptr pointer(!) because the lockless queue can't + // store objects with nontrivial destructors. We have to make absolutely shure that we + // manually delete the shared_ptr when we pop it from the queue. + cancel_queue.push(new std::shared_ptr(req)); + uv_async_send(&async_cancel); + } +} +} \ No newline at end of file diff --git a/common/curl_request.hpp b/common/curl_request.hpp new file mode 100644 index 000000000..3a512836e --- /dev/null +++ b/common/curl_request.hpp @@ -0,0 +1,44 @@ +#ifndef LLMR_COMMON_CURL_REQUEST +#define LLMR_COMMON_CURL_REQUEST + +#include +#include + +#include + +// Forward definition. +typedef void CURL; +typedef struct uv_work_s uv_work_t; +typedef struct uv_check_s uv_check_t; + +namespace llmr { +namespace platform { + +struct Response; + +class Request : public std::enable_shared_from_this, private util::noncopyable { +public: + Request(const std::string &url, + std::function background_function, + std::function foreground_callback); + ~Request(); + +public: + static void work_cb(uv_work_t *work); + static void after_work_cb(uv_work_t *work, int status); + + +public: + const std::string url; + const std::function background_function; + const std::function foreground_callback; + std::unique_ptr res; + CURL *curl = nullptr; + bool cancelled = false; + + uv_check_t *check = nullptr; +}; +} +} + +#endif diff --git a/include/llmr/util/uv.hpp b/include/llmr/util/uv.hpp new file mode 100644 index 000000000..3a1f48980 --- /dev/null +++ b/include/llmr/util/uv.hpp @@ -0,0 +1,46 @@ +#ifndef LLMR_UTIL_UV +#define LLMR_UTIL_UV + +#include +#include +#include +#include + +namespace uv { + +typedef void (*callback)(); + +class mutex { +public: + inline mutex() { assert(uv_mutex_init(&mtx) == 0); } + inline ~mutex() { uv_mutex_destroy(&mtx); } + inline void lock() { uv_mutex_lock(&mtx); } + inline void unlock() { uv_mutex_unlock(&mtx); } + inline bool trylock() { return uv_mutex_trylock(&mtx); } + +private: + uv_mutex_t mtx; +}; + +class lock { +public: + lock(mutex &mtx) : mtx(mtx) { mtx.lock(); } + ~lock() { mtx.unlock(); } + +private: + mutex &mtx; +}; + +class once { +public: + void operator()(void (*callback)(void)) { + uv_once(&o, callback); + } + +private: + uv_once_t o = UV_ONCE_INIT; +}; + +} + +#endif diff --git a/test/headless.cpp b/test/headless.cpp index d1059f750..0dbf79d40 100644 --- a/test/headless.cpp +++ b/test/headless.cpp @@ -8,29 +8,16 @@ #include #include +#include + double llmr::platform::elapsed() { return 0; } void llmr::platform::restart() { - // TODO + // noop } -std::shared_ptr -llmr::platform::request_http(const std::string &url, std::function background_function, - std::function foreground_callback) { - // TODO - return nullptr; -} - -void llmr::platform::cancel_request_http(const std::shared_ptr &req) { - // TODO -} - - - - - TEST(Headless, initialize) { llmr::Settings settings; @@ -110,11 +97,16 @@ TEST(Headless, initialize) { return; } + map.setup(); map.resize(width, height); map.loadSettings(); map.update(); + + // Run the loop. It will terminate when we don't have any further listeners. + uv_run(uv_default_loop(), UV_RUN_DEFAULT); + map.render(); diff --git a/test/test.gyp b/test/test.gyp index a04b4ba78..98e9b125b 100644 --- a/test/test.gyp +++ b/test/test.gyp @@ -21,6 +21,33 @@ ], }, }, + { + 'target_name': 'link_curl', + 'type': 'none', + 'direct_dependent_settings': { + 'conditions': [ + ['OS == "mac"', { + 'xcode_settings': { + 'OTHER_CPLUSPLUSFLAGS':[ + '<@(curl_cflags)', + ], + 'OTHER_LDFLAGS': [ + '<@(curl_libraries)', + ], + }, + }, { + 'cflags': [ + '<@(curl_cflags)', + ], + 'link_settings': { + 'libraries': [ + '<@(curl_libraries)', + ], + }, + }], + ], + }, + }, { "target_name": "rotation_range", "product_name": "test_rotation_range", @@ -47,11 +74,14 @@ "sources": [ "./main.cpp", "./headless.cpp", + "../common/curl_request.hpp", + "../common/curl_request.cpp", ], "dependencies": [ "../deps/gtest/gtest.gyp:gtest", "../llmr.gyp:llmr-x86", "link_gl", + "link_curl", ] }, {