tei/MapLibreParticleLab
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add back CURL requesting

Browse Source
This commit is contained in:
Konstantin Käfer
2014-09-24 16:09:26 +02:00
parent 0fc1f66868
commit 992d9ff051
14 changed files with 997 additions and 635 deletions
Download Patch File Download Diff File Expand all files Collapse all files

828
common/curl_request.cpp
View File

@@ -1,414 +1,414 @@
#include <mbgl/platform/platform.hpp>
#include <mbgl/platform/request.hpp>
#include <mbgl/util/uv_detail.hpp>
#include <mbgl/util/std.hpp>
#include <queue>
#include <boost/lockfree/queue.hpp>
#include <curl/curl.h>
// This file contains code from http://curl.haxx.se/libcurl/c/multi-uv.html:
/***************************************************************************
* _ _ ____ _
* Project ___| | | | _ \| |
* / __| | | | |_) | |
* | (__| |_| | _ <| |___
* \___|\___/|_| \_\_____|
*
* Copyright (C) 1998 - 2013, Daniel Stenberg, <daniel@haxx.se>, 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 mbgl {
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<std::shared_ptr<mbgl::platform::Request> *> add_queue(8);
static boost::lockfree::queue<std::shared_ptr<mbgl::platform::Request> *> 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 *> curl_handle_cache;
class CURLRequest : public mbgl::platform::Request {
public:
CURLRequest(const std::string &url,
std::function<void(mbgl::platform::Response *)> callback,
std::shared_ptr<uv::loop> loop)
: Request(url, callback, loop) {}
CURL *curl = nullptr;
};
// 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<Request> *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.
(*req)->complete();
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.
((CURLRequest *)req->get())->curl = nullptr;
// Delete the shared_ptr pointer we created earlier.
delete req;
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 = nullptr;
if (socketp) {
context = (curl_context *)socketp;
} else if (action != CURL_POLL_REMOVE) {
context = create_curl_context(s);
}
if (context) {
curl_multi_assign(curl_multi, s, (void *)context);
if (action == CURL_POLL_IN || action == CURL_POLL_INOUT) {
uv_poll_start(&context->poll_handle, UV_READABLE, curl_perform);
}
if (action == CURL_POLL_OUT || action == CURL_POLL_INOUT) {
uv_poll_start(&context->poll_handle, UV_WRITABLE, curl_perform);
}
if (action == CURL_POLL_REMOVE && socketp) {
uv_poll_stop(&context->poll_handle);
destroy_curl_context(context);
curl_multi_assign(curl_multi, s, NULL);
}
}
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;
}
// Compares the beginning of the (non-zero-terminated!) data buffer with the (zero-terminated!)
// header string. If the data buffer contains the header string at the beginning, it returns
// the length of the header string == begin of the value, otherwise it returns npos.
// The comparison of the header is ASCII-case-insensitive.
size_t header_matches(const char *header, const char *buffer, size_t length) {
const size_t header_length = strlen(header);
if (length < header_length) return std::string::npos;
size_t i = 0;
while (i < length && i < header_length && std::tolower(buffer[i]) == header[i]) {
i++;
}
return i == header_length ? i : std::string::npos;
}
size_t curl_header_cb(char *buffer, size_t size, size_t nmemb, void *userp) {
const size_t length = size * nmemb;
size_t begin = std::string::npos;
if ((begin = header_matches("last-modified: ", buffer, length)) != std::string::npos) {
const std::string value { buffer + begin, length - begin - 2 /* remove \r\n */ };
static_cast<Response *>(userp)->setLastModified(value.c_str());
} else if ((begin = header_matches("cache-control: ", buffer, length)) != std::string::npos) {
const std::string value { buffer + begin, length - begin - 2 /* remove \r\n */ };
static_cast<Response *>(userp)->setCacheControl(value.c_str());
}
return length;
}
// 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<Request> *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;
}
// 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();
}
((CURLRequest *)req->get())->curl = handle;
curl_easy_setopt(handle, CURLOPT_PRIVATE, req);
curl_easy_setopt(handle, CURLOPT_CAINFO, "ca-bundle.crt");
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_HEADERFUNCTION, curl_header_cb);
curl_easy_setopt(handle, CURLOPT_HEADERDATA, (*req)->res.get());
curl_easy_setopt(handle, CURLOPT_ACCEPT_ENCODING, "gzip, 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<Request> *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 = ((CURLRequest *)req->get())->curl;
if (handle && !(*req)->cancelled) {
remove_curl_handle(handle);
((CURLRequest *)req->get())->curl = nullptr;
}
(*req)->cancelled = true;
delete req;
req = nullptr;
}
}
void thread_init_cb() {
curl_global_init(CURL_GLOBAL_ALL);
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
std::shared_ptr<platform::Request>
platform::request_http(const std::string &url,
std::function<void(Response *)> callback,
std::shared_ptr<uv::loop> loop) {
using namespace request;
init_thread_once(thread_init_cb);
std::shared_ptr<CURLRequest> req = std::make_shared<CURLRequest>(url, callback, loop);
// 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<Request>(req));
uv_async_send(&async_add);
return req;
}
// Cancels an HTTP request.
void platform::cancel_request_http(const std::shared_ptr<Request> &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<Request>(req));
uv_async_send(&async_cancel);
}
}
} // end namespace mbgl
//
//#include <mbgl/platform/platform.hpp>
//#include <mbgl/platform/request.hpp>
//#include <mbgl/util/uv_detail.hpp>
//#include <mbgl/util/std.hpp>
//
//#include <queue>
//#include <boost/lockfree/queue.hpp>
//
//#include <curl/curl.h>
//
//// This file contains code from http://curl.haxx.se/libcurl/c/multi-uv.html:
//
///***************************************************************************
// * _ _ ____ _
// * Project ___| | | | _ \| |
// * / __| | | | |_) | |
// * | (__| |_| | _ <| |___
// * \___|\___/|_| \_\_____|
// *
// * Copyright (C) 1998 - 2013, Daniel Stenberg, <daniel@haxx.se>, 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 mbgl {
//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<std::shared_ptr<mbgl::platform::Request> *> add_queue(8);
//static boost::lockfree::queue<std::shared_ptr<mbgl::platform::Request> *> 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 *> curl_handle_cache;
//
//
//class CURLRequest : public mbgl::platform::Request {
//public:
// CURLRequest(const std::string &url,
// std::function<void(mbgl::platform::Response *)> callback,
// std::shared_ptr<uv::loop> loop)
// : Request(url, callback, loop) {}
//
// CURL *curl = nullptr;
//};
//
//
//// 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<Request> *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.
// (*req)->complete();
//
// 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.
// ((CURLRequest *)req->get())->curl = nullptr;
//
// // Delete the shared_ptr pointer we created earlier.
// delete req;
// 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 = nullptr;
//
// if (socketp) {
// context = (curl_context *)socketp;
// } else if (action != CURL_POLL_REMOVE) {
// context = create_curl_context(s);
// }
//
// if (context) {
// curl_multi_assign(curl_multi, s, (void *)context);
// if (action == CURL_POLL_IN || action == CURL_POLL_INOUT) {
// uv_poll_start(&context->poll_handle, UV_READABLE, curl_perform);
// }
// if (action == CURL_POLL_OUT || action == CURL_POLL_INOUT) {
// uv_poll_start(&context->poll_handle, UV_WRITABLE, curl_perform);
// }
// if (action == CURL_POLL_REMOVE && socketp) {
// uv_poll_stop(&context->poll_handle);
// destroy_curl_context(context);
// curl_multi_assign(curl_multi, s, NULL);
// }
// }
//
// 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;
//}
//
//// Compares the beginning of the (non-zero-terminated!) data buffer with the (zero-terminated!)
//// header string. If the data buffer contains the header string at the beginning, it returns
//// the length of the header string == begin of the value, otherwise it returns npos.
//// The comparison of the header is ASCII-case-insensitive.
//size_t header_matches(const char *header, const char *buffer, size_t length) {
// const size_t header_length = strlen(header);
// if (length < header_length) return std::string::npos;
// size_t i = 0;
// while (i < length && i < header_length && std::tolower(buffer[i]) == header[i]) {
// i++;
// }
// return i == header_length ? i : std::string::npos;
//}
//
//size_t curl_header_cb(char *buffer, size_t size, size_t nmemb, void *userp) {
// const size_t length = size * nmemb;
//
// size_t begin = std::string::npos;
// if ((begin = header_matches("last-modified: ", buffer, length)) != std::string::npos) {
// const std::string value { buffer + begin, length - begin - 2 /* remove \r\n */ };
// static_cast<Response *>(userp)->setLastModified(value.c_str());
// } else if ((begin = header_matches("cache-control: ", buffer, length)) != std::string::npos) {
// const std::string value { buffer + begin, length - begin - 2 /* remove \r\n */ };
// static_cast<Response *>(userp)->setCacheControl(value.c_str());
// }
//
// return length;
//}
//
//// 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<Request> *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;
// }
//
// // 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();
// }
//
// ((CURLRequest *)req->get())->curl = handle;
//
// curl_easy_setopt(handle, CURLOPT_PRIVATE, req);
// curl_easy_setopt(handle, CURLOPT_CAINFO, "ca-bundle.crt");
// 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_HEADERFUNCTION, curl_header_cb);
// curl_easy_setopt(handle, CURLOPT_HEADERDATA, (*req)->res.get());
// curl_easy_setopt(handle, CURLOPT_ACCEPT_ENCODING, "gzip, 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<Request> *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 = ((CURLRequest *)req->get())->curl;
// if (handle && !(*req)->cancelled) {
// remove_curl_handle(handle);
// ((CURLRequest *)req->get())->curl = nullptr;
// }
// (*req)->cancelled = true;
//
// delete req;
// req = nullptr;
// }
//}
//
//void thread_init_cb() {
// curl_global_init(CURL_GLOBAL_ALL);
//
// 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
//
//
//std::shared_ptr<platform::Request>
//platform::request_http(const std::string &url,
// std::function<void(Response *)> callback,
// std::shared_ptr<uv::loop> loop) {
// using namespace request;
// init_thread_once(thread_init_cb);
// std::shared_ptr<CURLRequest> req = std::make_shared<CURLRequest>(url, callback, loop);
//
// // 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<Request>(req));
// uv_async_send(&async_add);
//
// return req;
//}
//
//// Cancels an HTTP request.
//void platform::cancel_request_http(const std::shared_ptr<Request> &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<Request>(req));
// uv_async_send(&async_cancel);
// }
//}
//} // end namespace mbgl

81
common/http_request_baton_cocoa.mm
View File

@@ -15,9 +15,11 @@ namespace mbgl {
dispatch_once_t request_initialize = 0;
NSURLSession *session = nullptr;
void HTTPRequestBaton::start() {
void HTTPRequestBaton::start(const util::ptr<HTTPRequestBaton> &ptr) {
assert(uv_thread_self() == ptr->thread_id);
// Starts the request.
assert(!ptr);
util::ptr<HTTPRequestBaton> baton = ptr;
// Create a C locale
static locale_t locale = newlocale(LC_ALL_MASK, nullptr, nullptr);
@@ -32,11 +34,11 @@ void HTTPRequestBaton::start() {
// TODO: add a delegate to the session that prohibits caching, since we handle this ourselves.
});
NSMutableURLRequest *request = [[NSMutableURLRequest alloc] initWithURL:[NSURL URLWithString:@(path.c_str())]];
if (response && response->modified) {
NSMutableURLRequest *request = [[NSMutableURLRequest alloc] initWithURL:[NSURL URLWithString:@(baton->path.c_str())]];
if (baton->response && baton->response->modified) {
struct tm *timeinfo;
char buffer[32];
const time_t modified = response->modified;
const time_t modified = baton->response->modified;
timeinfo = std::gmtime(&modified);
strftime_l(buffer, 32, "%a, %d %b %Y %H:%M:%S GMT", timeinfo, locale);
[request addValue:@(buffer) forHTTPHeaderField:@"If-Modified-Since"];
@@ -48,23 +50,23 @@ void HTTPRequestBaton::start() {
if ([error code] == NSURLErrorCancelled) {
// The response code remains at 0 to indicate cancelation.
// In addition, we don't need any response object.
response.reset();
type = HTTPResponseType::Canceled;
baton->response.reset();
baton->type = HTTPResponseType::Canceled;
} else {
// TODO: Use different codes for host not found, timeout, invalid URL etc.
// These can be categorized in temporary and permanent errors.
response = std::make_unique<Response>();
response->code = [(NSHTTPURLResponse *)res statusCode];
response->message = [[error localizedDescription] UTF8String];
baton->response = std::make_unique<Response>();
baton->response->code = [(NSHTTPURLResponse *)res statusCode];
baton->response->message = [[error localizedDescription] UTF8String];
switch ([error code]) {
case NSURLErrorBadServerResponse: // 5xx errors
type = HTTPResponseType::TemporaryError;
baton->type = HTTPResponseType::TemporaryError;
break;
case NSURLErrorTimedOut:
case NSURLErrorUserCancelledAuthentication:
type = HTTPResponseType::SingularError; // retry immediately
baton->type = HTTPResponseType::SingularError; // retry immediately
break;
case NSURLErrorNetworkConnectionLost:
@@ -75,28 +77,28 @@ void HTTPRequestBaton::start() {
case NSURLErrorInternationalRoamingOff:
case NSURLErrorCallIsActive:
case NSURLErrorDataNotAllowed:
type = HTTPResponseType::ConnectionError;
baton->type = HTTPResponseType::ConnectionError;
break;
default:
type = HTTPResponseType::PermanentError;
baton->type = HTTPResponseType::PermanentError;
}
}
} else if ([res isKindOfClass:[NSHTTPURLResponse class]]) {
const long code = [(NSHTTPURLResponse *)res statusCode];
if (code == 304) {
// Assume a Response object already exists.
assert(response);
assert(baton->response);
} else {
response = std::make_unique<Response>();
response->code = code;
response->data = {(const char *)[data bytes], [data length]};
baton->response = std::make_unique<Response>();
baton->response->code = code;
baton->response->data = {(const char *)[data bytes], [data length]};
}
if (code == 304) {
type = HTTPResponseType::NotModified;
baton->type = HTTPResponseType::NotModified;
} else if (code == 200) {
type = HTTPResponseType::Successful;
baton->type = HTTPResponseType::Successful;
} else {
assert(!"code must be either 200 or 304");
}
@@ -104,47 +106,36 @@ void HTTPRequestBaton::start() {
NSDictionary *headers = [(NSHTTPURLResponse *)res allHeaderFields];
NSString *cache_control = [headers objectForKey:@"Cache-Control"];
if (cache_control) {
response->expires = Response::parseCacheControl([cache_control UTF8String]);
baton->response->expires = Response::parseCacheControl([cache_control UTF8String]);
}
NSString *last_modified = [headers objectForKey:@"Last-Modified"];
if (last_modified) {
response->modified = parse_date([last_modified UTF8String]);
baton->response->modified = parse_date([last_modified UTF8String]);
}
} else {
// This should never happen.
type = HTTPResponseType::PermanentError;
response = std::make_unique<Response>();
response->code = -1;
response->message = "response class is not NSHTTPURLResponse";
baton->type = HTTPResponseType::PermanentError;
baton->response = std::make_unique<Response>();
baton->response->code = -1;
baton->response->message = "response class is not NSHTTPURLResponse";
}
uv_async_send(async);
uv_async_send(baton->async);
}];
[task resume];
ptr = const_cast<void *>(CFBridgingRetain(task));
baton->ptr = const_cast<void *>(CFBridgingRetain(task));
}
void HTTPRequestBaton::cleanup() {
if (ptr) {
CFBridgingRelease(ptr);
ptr = nullptr;
}
}
void HTTPRequestBaton::stop(const util::ptr<HTTPRequestBaton> &ptr) {
assert(uv_thread_self() == ptr->thread_id);
assert(ptr->ptr);
void HTTPRequestBaton::cancel() {
// After this function returns, the HTTPRequestBaton object may cease to exist at any time.
// try to stop the request
if (ptr) {
NSURLSessionDataTask *task = CFBridgingRelease(ptr);
ptr = nullptr;
[task cancel];
} else {
// Currently, there is no request in progress. We can delete the async right away.
uv_async_send(async);
}
NSURLSessionDataTask *task = CFBridgingRelease(ptr->ptr);
ptr->ptr = nullptr;
[task cancel];
}
}

421
common/http_request_baton_curl.cpp Normal file
View File

@@ -0,0 +1,421 @@
#include <mbgl/storage/http_request_baton.hpp>
#include <mbgl/util/uv-messenger.h>
#include <mbgl/util/std.hpp>
#include <mbgl/util/ptr.hpp>
#include <uv.h>
#include <curl/curl.h>
#include <queue>
#include <cassert>
// This file contains code from http://curl.haxx.se/libcurl/c/multi-uv.html:
/***************************************************************************
* _ _ ____ _
* Project ___| | | | _ \| |
* / __| | | | |_) | |
* | (__| |_| | _ <| |___
* \___|\___/|_| \_\_____|
*
* Copyright (C) 1998 - 2013, Daniel Stenberg, <daniel@haxx.se>, 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.
*/
// Handles the request thread + messaging to the thread.
static uv_once_t once;
static uv_loop_t loop;
static uv_messenger_t start_messenger;
static uv_messenger_t stop_messenger;
static uv_thread_t thread;
static unsigned long thread_id;
// 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 *multi = nullptr;
// CURL share handles are used for sharing session state (e.g.)
static uv_mutex_t share_mutex;
static CURLSH *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 *> handles;
namespace mbgl {
struct curl_context {
uv_poll_t poll_handle;
curl_socket_t sockfd;
};
// Locks the CURL share handle
void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) {
uv_mutex_lock(&share_mutex);
}
// Unlocks the CURL share handle
void curl_share_unlock(CURL *, curl_lock_data, void *) {
uv_mutex_unlock(&share_mutex);
}
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 destroy_curl_context(curl_context *context) {
uv_close((uv_handle_t *)&context->poll_handle, [](uv_handle_t *handle) {
delete (curl_context *)handle->data;
});
}
// This function must run in the CURL thread.
// It is either called when the request is completed, or when we try to cancel the request.
void finish_request(const util::ptr<HTTPRequestBaton> &baton) {
assert(uv_thread_self() == thread_id);
if (baton->ptr) {
CURL *handle = (CURL *)baton->ptr;
CURLMcode error = curl_multi_remove_handle(multi, handle);
if (error != CURLM_OK) {
baton->response = std::make_unique<Response>();
baton->response->code = -1;
baton->response->message = curl_multi_strerror(error);
}
// Destroy the shared pointer. We still have one pointing to it
util::ptr<HTTPRequestBaton> *baton_ptr = nullptr;
curl_easy_getinfo(handle, CURLINFO_PRIVATE, (char *)&baton_ptr);
curl_easy_setopt(handle, CURLOPT_PRIVATE, nullptr);
delete baton_ptr;
curl_easy_reset(handle);
handles.push(handle);
baton->ptr = nullptr;
}
}
void curl_perform(uv_poll_t *req, int status, int events) {
int running_handles;
int flags = 0;
curl_context *context = (curl_context *)req;
CURLMsg *message;
int pending;
uv_timer_stop(&timeout);
if (events & UV_READABLE) {
flags |= CURL_CSELECT_IN;
}
if (events & UV_WRITABLE) {
flags |= CURL_CSELECT_OUT;
}
curl_multi_socket_action(multi, context->sockfd, flags, &running_handles);
while ((message = curl_multi_info_read(multi, &pending))) {
switch (message->msg) {
case CURLMSG_DONE: {
util::ptr<HTTPRequestBaton> *baton_ptr = nullptr;
curl_easy_getinfo(message->easy_handle, CURLINFO_PRIVATE, (char *)&baton_ptr);
util::ptr<HTTPRequestBaton> baton = *baton_ptr;
// Add human-readable error code
if (message->data.result != CURLE_OK) {
baton->response->message = curl_easy_strerror(message->data.result);
baton->response->code = -1;
switch (message->data.result) {
case CURLE_COULDNT_RESOLVE_PROXY:
case CURLE_COULDNT_RESOLVE_HOST:
case CURLE_COULDNT_CONNECT:
baton->type = HTTPResponseType::ConnectionError;
break;
case CURLE_OPERATION_TIMEDOUT:
baton->type = HTTPResponseType::TemporaryError;
break;
default:
baton->type = HTTPResponseType::PermanentError;
}
} else {
long code = 0;
curl_easy_getinfo(message->easy_handle, CURLINFO_RESPONSE_CODE, &code);
if (code != 304) {
baton->response->code = code;
}
if (code == 304) {
baton->type = HTTPResponseType::NotModified;
} else if (code == 200) {
baton->type = HTTPResponseType::Successful;
} else if (code >= 500 && code < 600) {
baton->type = HTTPResponseType::TemporaryError;
} else if (code >= 400 && code < 500) {
baton->type = HTTPResponseType::PermanentError;
} else {
assert(!"code must be either 200 or 304");
}
}
// We're currently in the CURL request thread.
finish_request(baton);
if (baton->async) {
uv_async_send(baton->async);
baton->async = 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 = nullptr;
if (socketp) {
context = (curl_context *)socketp;
} else if (action != CURL_POLL_REMOVE) {
context = create_curl_context(s);
}
if (context) {
curl_multi_assign(multi, s, (void *)context);
if (action == CURL_POLL_IN || action == CURL_POLL_INOUT) {
uv_poll_start(&context->poll_handle, UV_READABLE, curl_perform);
}
if (action == CURL_POLL_OUT || action == CURL_POLL_INOUT) {
uv_poll_start(&context->poll_handle, UV_WRITABLE, curl_perform);
}
if (action == CURL_POLL_REMOVE && socketp) {
uv_poll_stop(&context->poll_handle);
destroy_curl_context(context);
curl_multi_assign(multi, s, NULL);
}
}
return 0;
}
void on_timeout(uv_timer_t *req) {
int running_handles;
CURLMcode error = curl_multi_socket_action(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);
}
}
void thread_init(void *) {
#ifdef __APPLE__
pthread_setname_np("CURL");
#endif
thread_id = uv_thread_self();
uv_timer_init(&loop, &timeout);
CURLSHcode share_error;
share = curl_share_init();
share_error = curl_share_setopt(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(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;
multi = curl_multi_init();
multi_error = curl_multi_setopt(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(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(multi);
multi = nullptr;
curl_share_cleanup(share);
share = nullptr;
thread_id = -1;
}
// 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 *const contents, const size_t size, const size_t nmemb, void *const userp) {
((std::string *)userp)->append((char *)contents, size * nmemb);
return size * nmemb;
}
// Compares the beginning of the (non-zero-terminated!) data buffer with the (zero-terminated!)
// header string. If the data buffer contains the header string at the beginning, it returns
// the length of the header string == begin of the value, otherwise it returns npos.
// The comparison of the header is ASCII-case-insensitive.
size_t header_matches(const char *const header, const char *const buffer, const size_t length) {
const size_t header_length = strlen(header);
if (length < header_length) return std::string::npos;
size_t i = 0;
while (i < length && i < header_length && std::tolower(buffer[i]) == header[i]) {
i++;
}
return i == header_length ? i : std::string::npos;
}
size_t curl_header_cb(char * const buffer, const size_t size, const size_t nmemb, void *const userp) {
const size_t length = size * nmemb;
size_t begin = std::string::npos;
if ((begin = header_matches("last-modified: ", buffer, length)) != std::string::npos) {
const std::string value { buffer + begin, length - begin - 2 }; // remove \r\n
static_cast<Response *>(userp)->modified = curl_getdate(value.c_str(), nullptr);
} else if ((begin = header_matches("cache-control: ", buffer, length)) != std::string::npos) {
const std::string value { buffer + begin, length - begin - 2 }; // remove \r\n
static_cast<Response *>(userp)->expires = Response::parseCacheControl(value.c_str());
}
return length;
}
// This function must run in the CURL thread.
void start_request(void *const ptr) {
assert(uv_thread_self() == thread_id);
util::ptr<HTTPRequestBaton> &baton = *(util::ptr<HTTPRequestBaton> *)ptr;
assert(baton);
CURL *handle = nullptr;
if (!handles.empty()) {
handle = handles.front();
handles.pop();
} else {
handle = curl_easy_init();
}
baton->ptr = handle;
if (!baton->response) {
baton->response = std::make_unique<Response>();
}
// Carry on the shared pointer in the private information of the CURL handle.
curl_easy_setopt(handle, CURLOPT_PRIVATE, ptr);
curl_easy_setopt(handle, CURLOPT_CAINFO, "ca-bundle.crt");
curl_easy_setopt(handle, CURLOPT_URL, baton->path.c_str());
curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, curl_write_cb);
curl_easy_setopt(handle, CURLOPT_WRITEDATA, &baton->response->data);
curl_easy_setopt(handle, CURLOPT_HEADERFUNCTION, curl_header_cb);
curl_easy_setopt(handle, CURLOPT_HEADERDATA, baton->response.get());
curl_easy_setopt(handle, CURLOPT_ACCEPT_ENCODING, "gzip, deflate");
curl_easy_setopt(handle, CURLOPT_SHARE, share);
// Start requesting the information.
curl_multi_add_handle(multi, handle);
}
// This function must run in the CURL thread.
void stop_request(void *const ptr) {
assert(uv_thread_self() == thread_id);
util::ptr<HTTPRequestBaton> &baton = *(util::ptr<HTTPRequestBaton> *)ptr;
assert(baton);
if (baton->async) {
baton->type = HTTPResponseType::Canceled;
// We can still stop the request because it is still in progress.
finish_request(baton);
uv_async_send(baton->async);
baton->async = nullptr;
} else {
// If the async handle is gone, it means that the actual request has been completed before
// we got a chance to cancel it. In this case, this is a no-op. It is likely that
// the pointer below is the last lifeline of the HTTPRequestBaton. This means we're going
// to delete the HTTPRequestBaton in the current (CURL) thread.
}
delete (util::ptr<HTTPRequestBaton> *)ptr;
}
void create_thread() {
uv_mutex_init(&share_mutex);
uv_loop_init(&loop);
uv_messenger_init(&loop, &start_messenger, start_request);
uv_messenger_init(&loop, &stop_messenger, stop_request);
uv_thread_create(&thread, thread_init, nullptr);
}
// This function must be run from the main thread (== where the HTTPRequestBaton was created)
void HTTPRequestBaton::start(const util::ptr<HTTPRequestBaton> &ptr) {
assert(uv_thread_self() == ptr->thread_id);
uv_once(&once, create_thread);
uv_messenger_send(&start_messenger, new util::ptr<HTTPRequestBaton>(ptr));
}
// This function must be run from the main thread (== where the HTTPRequestBaton was created)
void HTTPRequestBaton::stop(const util::ptr<HTTPRequestBaton> &ptr) {
assert(uv_thread_self() == ptr->thread_id);
uv_once(&once, create_thread);
uv_messenger_send(&stop_messenger, new util::ptr<HTTPRequestBaton>(ptr));
}
}

14
include/mbgl/storage/http_request.hpp
View File

@@ -3,12 +3,14 @@
#include <mbgl/storage/resource_type.hpp>
#include <mbgl/storage/base_request.hpp>
#include <mbgl/storage/http_request_baton.hpp>
#include <string>
#include <memory>
#include <cassert>
typedef struct uv_loop_s uv_loop_t;
typedef struct uv_timer_s uv_timer_t;
namespace mbgl {
@@ -25,14 +27,22 @@ public:
void cancel();
private:
void loadedCacheEntry(std::unique_ptr<Response> &&response);
void handleCacheResponse(std::unique_ptr<Response> &&response, uv_loop_t *loop);
void handleHTTPResponse(HTTPResponseType responseType, std::unique_ptr<Response> &&response, uv_loop_t *loop);
void startRequest(uv_loop_t *loop);
void removeCacheBaton();
void removeHTTPBaton();
private:
const unsigned long thread_id;
CacheRequestBaton *cache_baton = nullptr;
HTTPRequestBaton *http_baton = nullptr;
util::ptr<HTTPRequestBaton> http_baton;
uv_timer_t *backoff_timer = nullptr;
util::ptr<SQLiteStore> store;
const ResourceType type;
uint8_t attempts = 0;
friend struct HTTPRequestBaton;
};

40
include/mbgl/storage/http_request_baton.hpp
View File

@@ -2,11 +2,11 @@
#define MBGL_STORAGE_HTTP_REQUEST_BATON
#include <mbgl/storage/response.hpp>
#include <mbgl/util/ptr.hpp>
#include <string>
typedef struct uv_async_s uv_async_t;
typedef struct uv_timer_s uv_timer_t;
namespace mbgl {
@@ -44,44 +44,28 @@ enum class HTTPResponseType : int8_t {
};
struct HTTPRequestBaton {
HTTPRequestBaton();
HTTPRequestBaton(const std::string &path);
~HTTPRequestBaton();
const unsigned long thread_id;
const std::string path;
HTTPRequest *request = nullptr;
std::string path;
uv_async_t *async = nullptr;
uv_timer_t *timer = nullptr;
std::unique_ptr<Response> response;
void *ptr = nullptr;
HTTPResponseType type = HTTPResponseType::Unknown;
uint8_t attempts = 0;
std::unique_ptr<Response> response;
// Implementation specific use.
void *ptr = nullptr;
// IMPLEMENT THESE 3 PLATFORM SPECIFIC FUNCTIONS:
// Begin the HTTP request. Platform-specific implementation.
void start();
static void start(const util::ptr<HTTPRequestBaton> &ptr);
// This will be called so that the baton can release resources. It is expected that the ptr is
// null after this call. Platform-specific implementation.
void cleanup();
// This will be called to cancel the HTTP request (if possible). Platform-specific implementation.
void cancel();
// Called when the request should be canceled. This will call cancel() in turn.
void stop();
// After calling reset(), it is safe to call start() again on this baton object.
void reset();
// Will call start() after the specified timeout.
void retry(uint64_t delay);
// Will be
static void notify(uv_async_t *async);
// This will be called to stop/cancel the HTTP request (if possible). Platform-specific implementation.
static void stop(const util::ptr<HTTPRequestBaton> &ptr);
};
}

2
ios/mapbox-gl-cocoa

Submodule ios/mapbox-gl-cocoa updated: 461024ae40...f3829151ab

2
linux/mapboxgl-app.gyp
View File

@@ -15,7 +15,7 @@
'../common/platform_default.cpp',
'../common/glfw_view.hpp',
'../common/glfw_view.cpp',
'../common/curl_request.cpp',
'../common/http_request_baton_curl.cpp',
'../common/stderr_log.hpp',
'../common/stderr_log.cpp',
],

3
mapboxgl.gyp
View File

@@ -241,6 +241,9 @@
'<@(curl_libraries)',
'<@(sqlite3_libraries)',
],
'OTHER_CPLUSPLUSFLAGS': [
'<@(uv_cflags)',
],
'OTHER_CFLAGS': [
'<@(uv_cflags)',
],

1
src/storage/file_request.cpp
View File

@@ -178,6 +178,7 @@ void FileRequest::cancel() {
ptr = nullptr;
}
notify();
}
FileRequest::~FileRequest() {

110
src/storage/http_request.cpp
View File

@@ -33,7 +33,7 @@ HTTPRequest::HTTPRequest(ResourceType type_, const std::string &path, uv_loop_t
std::unique_ptr<CacheRequestBaton> baton((CacheRequestBaton *)ptr);
if (baton->request) {
assert(uv_thread_self() == baton->request->thread_id);
baton->request->loadedCacheEntry(std::move(response));
baton->request->handleCacheResponse(std::move(response), baton->loop);
if (baton->request) {
// If we called notify(), the request object may already have ceased to exist.
baton->request->cache_baton = nullptr;
@@ -42,7 +42,7 @@ HTTPRequest::HTTPRequest(ResourceType type_, const std::string &path, uv_loop_t
}, cache_baton);
}
void HTTPRequest::loadedCacheEntry(std::unique_ptr<Response> &&response) {
void HTTPRequest::handleCacheResponse(std::unique_ptr<Response> &&response, uv_loop_t *loop) {
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>(
@@ -60,25 +60,92 @@ void HTTPRequest::loadedCacheEntry(std::unique_ptr<Response> &&response) {
}
}
// Either, the cache entry is expired, or there is no cached version. In both cases,
// we need to do a new request. The requesting code will issue a conditional request in case
// we already have a modified date on the response object.
http_baton = new HTTPRequestBaton;
http_baton->request = this;
http_baton->path = std::move(cache_baton->path);
http_baton->async = new uv_async_t;
http_baton->response = std::move(response);
http_baton->async->data = http_baton;
uv_async_init(cache_baton->loop, http_baton->async, HTTPRequestBaton::notify);
http_baton->reset();
http_baton->start();
startRequest(loop);
}
void HTTPRequest::cancel() {
void HTTPRequest::startRequest(uv_loop_t *loop) {
assert(uv_thread_self() == thread_id);
assert(!http_baton);
http_baton = std::make_shared<HTTPRequestBaton>(path);
http_baton->request = this;
http_baton->async = new uv_async_t;
http_baton->response = std::move(response);
http_baton->async->data = new util::ptr<HTTPRequestBaton>(http_baton);
uv_async_init(loop, http_baton->async, [](uv_async_t *async) {
util::ptr<HTTPRequestBaton> &http_baton = *(util::ptr<HTTPRequestBaton> *)async->data;
if (http_baton->request) {
HTTPRequest *request = http_baton->request;
request->http_baton.reset();
http_baton->request = nullptr;
request->handleHTTPResponse(http_baton->type, std::move(http_baton->response), async->loop);
}
delete (util::ptr<HTTPRequestBaton> *)async->data;
uv_close((uv_handle_t *)async, [](uv_handle_t *handle) {
uv_async_t *async = (uv_async_t *)handle;
delete async;
});
});
HTTPRequestBaton::start(http_baton);
}
void HTTPRequest::handleHTTPResponse(HTTPResponseType responseType, std::unique_ptr<Response> &&res, uv_loop_t *loop) {
assert(uv_thread_self() == thread_id);
assert(!http_baton);
assert(!response);
switch (responseType) {
case HTTPResponseType::TemporaryError:
case HTTPResponseType::SingularError:
case HTTPResponseType::ConnectionError:
fprintf(stderr, "error: %s\n", res->message.c_str());
// TODO: retry
break;
case HTTPResponseType::Canceled:
response.reset();
notify();
break;
case HTTPResponseType::PermanentError:
response = std::move(res);
notify();
break;
case HTTPResponseType::Successful:
store->put(path, type, *res);
response = std::move(res);
notify();
break;
case HTTPResponseType::NotModified:
store->updateExpiration(path, res->expires);
response = std::move(res);
notify();
break;
default:
assert(!"Response wasn't set");
break;
}
}
void HTTPRequest::removeHTTPBaton() {
assert(uv_thread_self() == thread_id);
if (http_baton) {
http_baton->request = nullptr;
HTTPRequestBaton::stop(http_baton);
http_baton.reset();
}
}
void HTTPRequest::removeCacheBaton() {
assert(uv_thread_self() == thread_id);
if (cache_baton) {
// Make sure that this object doesn't accidentally get accessed when it is destructed before
// Make sre that this object doesn't accidentally get accessed when it is destructed before
// the callback returned. They are being run in the same thread, so just setting it to
// null is sufficient.
// Note: We don't manually delete the CacheRequestBaton since it'll be deleted by the
@@ -86,13 +153,12 @@ void HTTPRequest::cancel() {
cache_baton->request = nullptr;
cache_baton = nullptr;
}
}
if (http_baton) {
http_baton->request = nullptr;
http_baton->stop();
http_baton = nullptr;
}
void HTTPRequest::cancel() {
assert(uv_thread_self() == thread_id);
removeCacheBaton();
removeHTTPBaton();
notify();
}

116
src/storage/http_request_baton.cpp
View File

@@ -1,126 +1,12 @@
#include <mbgl/storage/http_request_baton.hpp>
#include <mbgl/storage/http_request.hpp>
#include <mbgl/storage/sqlite_store.hpp>
#include <mbgl/util/uv.hpp>
#include <uv.h>
#include <cassert>
namespace mbgl {
HTTPRequestBaton::HTTPRequestBaton() : thread_id(uv_thread_self()) {
HTTPRequestBaton::HTTPRequestBaton(const std::string &path_) : thread_id(uv_thread_self()), path(path_) {
}
HTTPRequestBaton::~HTTPRequestBaton() {
assert(uv_thread_self() == thread_id);
if (timer) {
timer->data = nullptr;
}
}
void HTTPRequestBaton::reset() {
attempts++;
response.reset();
type = HTTPResponseType::Unknown;
}
void HTTPRequestBaton::retry(uint64_t delay) {
assert(uv_thread_self() == thread_id);
assert(!timer);
timer = new uv_timer_t;
timer->data = this;
uv_timer_init(async->loop, timer);
uv_timer_start(timer, [](uv_timer_t *timer) {
std::unique_ptr<uv_timer_t, uv::deleter> timer_guard(timer);
uv_timer_stop(timer);
if (timer->data) {
HTTPRequestBaton *baton = reinterpret_cast<HTTPRequestBaton *>(timer->data);
assert(uv_thread_self() == baton->thread_id);
baton->timer = nullptr;
baton->start();
}
}, delay, 0);
}
void HTTPRequestBaton::stop() {
if (timer) {
uv_timer_stop(timer);
uv::deleter()(timer);
timer = nullptr;
}
cancel();
}
void HTTPRequestBaton::notify(uv_async_t *async_unguarded) {
// Ensure deletion of pointers.
std::unique_ptr<uv_async_t, uv::deleter> async(async_unguarded);
std::unique_ptr<HTTPRequestBaton> baton((HTTPRequestBaton *)async->data);
baton->cleanup();
if (baton->request) {
assert(uv_thread_self() == baton->request->thread_id);
switch (baton->type) {
case HTTPResponseType::TemporaryError:
case HTTPResponseType::ConnectionError:
// Exponential back-off:
if (baton->attempts >= 4) {
// Fail immediately.
baton->request->response = std::move(baton->response);
baton->request->notify();
} else {
baton->reset();
baton->retry(1000 * (1 << baton->attempts));
// Make sure we're not deleting the objects, since we're now reusing them.
baton.release();
async.release();
}
break;
case HTTPResponseType::SingularError:
// Retry immediately.
if (baton->attempts >= 4) {
// Fail immediately.
baton->request->response = std::move(baton->response);
baton->request->notify();
} else {
baton->reset();
baton->start();
// Make sure we're not deleting the objects, since we're now reusing them.
baton.release();
async.release();
}
break;
case HTTPResponseType::Canceled:
baton->request->response.reset();
baton->request->notify();
break;
case HTTPResponseType::PermanentError:
case HTTPResponseType::Successful:
baton->request->store->put(baton->path, baton->request->type, *baton->response);
baton->request->response = std::move(baton->response);
baton->request->notify();
break;
case HTTPResponseType::NotModified:
baton->request->store->updateExpiration(baton->path, baton->response->expires);
baton->request->response = std::move(baton->response);
baton->request->notify();
break;
default:
// This should never happen.
assert(false);
break;
}
// Note: after calling notify(), the baton object may cease to exist.
} else {
// The HTTPRequest object has already been destructed. There is nobody we need to notify
// anymore. All we need to do is to cleanup.
}
}
}

2
src/util/uv-channel.c
View File

@@ -21,7 +21,7 @@ int uv_chan_init(uv_chan_t *chan) {
}
void uv_chan_send(uv_chan_t *chan, void *data) {
uv__chan_item_t *item = malloc(sizeof(uv__chan_item_t));
uv__chan_item_t *item = (uv__chan_item_t *)malloc(sizeof(uv__chan_item_t));
item->data = data;
uv_mutex_lock(&chan->mutex);

2
src/util/uv-messenger.c
View File

@@ -47,7 +47,7 @@ int uv_messenger_init(uv_loop_t *loop, uv_messenger_t *msgr, uv_messenger_cb cal
}
void uv_messenger_send(uv_messenger_t *msgr, void *data) {
uv__messenger_item_t *item = malloc(sizeof(uv__messenger_item_t));
uv__messenger_item_t *item = (uv__messenger_item_t *)malloc(sizeof(uv__messenger_item_t));
item->data = data;
uv_mutex_lock(&msgr->mutex);

10
src/util/uv-worker.c
View File

@@ -69,7 +69,7 @@ void uv__worker_thread_loop(void *ptr) {
#endif
uv__worker_item_t *item = NULL;
while ((item = uv_chan_receive(&worker->chan)) != NULL) {
while ((item = (uv__worker_item_t *)uv_chan_receive(&worker->chan)) != NULL) {
assert(item->work_cb);
item->work_cb(item->data);
@@ -86,7 +86,7 @@ void uv__worker_thread_loop(void *ptr) {
uv_chan_send(&worker->chan, NULL);
// Create a new worker item that acts as a terminate flag for this thread.
item = malloc(sizeof(uv__worker_item_t));
item = (uv__worker_item_t *)malloc(sizeof(uv__worker_item_t));
item->data = worker_thread;
item->work_cb = NULL;
item->after_work_cb = NULL;
@@ -100,7 +100,7 @@ int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char *
worker->name = name;
worker->count = 0;
worker->close_cb = NULL;
worker->msgr = malloc(sizeof(uv_messenger_t));
worker->msgr = (uv_messenger_t *)malloc(sizeof(uv_messenger_t));
int ret = uv_messenger_init(loop, worker->msgr, uv__worker_after);
if (ret < 0) {
free(worker->msgr);
@@ -111,7 +111,7 @@ int uv_worker_init(uv_worker_t *worker, uv_loop_t *loop, int count, const char *
// Initialize all worker threads.
for (int i = 0; i < count; i++) {
uv__worker_thread_t *worker_thread = malloc(sizeof(uv__worker_thread_t));
uv__worker_thread_t *worker_thread = (uv__worker_thread_t *)malloc(sizeof(uv__worker_thread_t));
worker_thread->worker = worker;
ret = uv_thread_create(&worker_thread->thread, uv__worker_thread_loop, worker_thread);
if (ret < 0) return ret;
@@ -132,7 +132,7 @@ void uv_worker_send(uv_worker_t *worker, void *data, uv_worker_cb work_cb,
// worker item will instead be freed in the worker thread.
assert(work_cb);
uv__worker_item_t *item = malloc(sizeof(uv__worker_item_t));
uv__worker_item_t *item = (uv__worker_item_t *)malloc(sizeof(uv__worker_item_t));
item->work_cb = work_cb;
item->after_work_cb = after_work_cb;
item->data = data;