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

Merge pull request #85 from mapbox/threaded-curl

Browse Source 
add threading for libcurl
This commit is contained in:
Konstantin Käfer
2014-03-03 15:07:33 +01:00
parent a31fab1e22 45385d112b
commit dadc8f151c
11 changed files with 365 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
Makefile
View File

@@ -20,6 +20,12 @@ linux: config.gypi src linux/llmr-app.gyp
make -C build/linux-make V=$(V)
./build/linux-make/out/Release/llmr.app
# build OS X app with Xcode
lproj: config.gypi src linux/llmr-app.gyp
deps/run_gyp linux/llmr-app.gyp -Goutput_dir=./out/ --depth=. --generator-output=./ -f xcode
open ./linux/llmr-app.xcodeproj
# build just xcode project for libllmr
xcode: config.gypi llmr.gyp
deps/run_gyp llmr.gyp -Goutput_dir=./out/ --depth=. --generator-output=./ -f xcode

2
include/llmr/map/tile.hpp
View File

@@ -95,7 +95,7 @@ private:
// Source data
std::string data;
const Style& style;
platform::Request req;
platform::Request *req = nullptr;
};
}

9
include/llmr/platform/platform.hpp
View File

@@ -15,10 +15,7 @@ namespace platform {
// Restarts painting. This could for example trigger the event loop of the controlling application.
void restart();
struct Request {
int16_t identifier = -1;
std::string original_url;
};
class Request;
struct Response {
int16_t code = -1;
@@ -26,10 +23,10 @@ struct Response {
};
// Makes an HTTP request of a URL on a background thread, calls a function with the results on the same thread, and finally calls a callback function on the main thread. Returns a cancellable request.
Request request_http(std::string url, std::function<void(Response&)> background_function, std::function<void()> foreground_callback);
Request *request_http(std::string url, std::function<void(Response&)> background_function, std::function<void()> foreground_callback);
// Cancels an HTTP request.
void cancel_request_http(Request request);
void cancel_request_http(Request *request);
// Returns a relative timestamp in seconds. This value must be monotonic.
double time();

45
include/llmr/util/threadpool.hpp Normal file
View File

@@ -0,0 +1,45 @@
#ifndef LLMR_UTIL_THREADPOOL
#define LLMR_UTIL_THREADPOOL
#include <pthread.h>
#include <forward_list>
#include <queue>
namespace llmr {
namespace util {
class Threadpool {
private:
class Worker {
public:
Worker(Threadpool& pool);
~Worker();
static void *loop(void *ptr);
private:
Threadpool& pool;
pthread_t thread;
};
public:
Threadpool(int max_workers = 4);
typedef void (*Callback)(void *);
void add(Callback callback, void *data);
private:
typedef std::pair<Callback, void *> Task;
const int max_workers;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t condition = PTHREAD_COND_INITIALIZER;
std::forward_list<Worker> workers;
int worker_count = 0;
std::queue<Task> tasks;
};
extern std::unique_ptr<Threadpool> threadpool;
}
}
#endif

61
linux/llmr-app.gyp
View File

@@ -11,18 +11,57 @@
"sources": [
"./main.cpp",
"./settings.cpp",
"./settings.hpp"
"./settings.hpp",
"./request.cpp",
"./request.hpp"
],
'product_extension': 'app',
'link_settings': {
'libraries': [
'<@(glfw3_libraries)',
'-lcurl'
],
},
'cflags': [
'<@(png_cflags)',
'<@(glfw3_cflags)'
'conditions': [
['OS == "mac"', {
'product_extension': 'app',
'mac_bundle': 1,
'mac_bundle_resources': [
'../macosx/Icon.icns',
],
'link_settings': {
'libraries': [
'$(SDKROOT)/System/Library/Frameworks/Cocoa.framework',
'$(SDKROOT)/System/Library/Frameworks/IOKit.framework',
'$(SDKROOT)/System/Library/Frameworks/OpenGL.framework',
'$(SDKROOT)/System/Library/Frameworks/CoreVideo.framework',
],
},
'xcode_settings': {
'ARCHS': [ "x86_64" ],
'SDKROOT': 'macosx',
'SUPPORTED_PLATFORMS':'macosx',
'OTHER_CPLUSPLUSFLAGS':[
'<@(glfw3_cflags)'
],
'OTHER_LDFLAGS': [
'-stdlib=libc++',
'<@(glfw3_libraries)',
'-lcurl',
],
'SDKROOT': 'macosx',
'INFOPLIST_FILE': '../macosx/Info.plist',
'CLANG_CXX_LIBRARY': 'libc++',
'CLANG_CXX_LANGUAGE_STANDARD':'c++11',
'MACOSX_DEPLOYMENT_TARGET':'10.9',
'GCC_VERSION': 'com.apple.compilers.llvm.clang.1_0',
'CLANG_ENABLE_OBJC_ARC': 'YES'
},
}, {
'link_settings': {
'libraries': [
'<@(glfw3_libraries)',
'-lcurl'
],
},
'cflags': [
'<@(png_cflags)',
'<@(glfw3_cflags)'
],
}]
],
"dependencies": [
"../llmr.gyp:llmr-x86"

93
linux/main.cpp
View File

@@ -1,19 +1,13 @@
#include <llmr/llmr.hpp>
#include <GLFW/glfw3.h>
#include <curl/curl.h>
#include <llmr/platform/platform.hpp>
#include "settings.hpp"
#include <future>
#include <list>
#include <chrono>
#include <thread>
#include "settings.hpp"
#include "request.hpp"
std::list<std::future<void>> futures;
std::list<std::future<void>>::iterator f_it;
std::list<std::function<void()>> callbacks;
std::list<std::function<void()>>::iterator c_it;
std::chrono::milliseconds zero (0);
std::forward_list<llmr::platform::Request *> requests;
class MapView {
public:
@@ -45,8 +39,13 @@ public:
glfwSetWindowUserPointer(window, this);
glfwMakeContextCurrent(window);
int width, height;
glfwGetWindowSize(window, &width, &height);
int fb_width, fb_height;
glfwGetFramebufferSize(window, &fb_width, &fb_height);
settings.load();
map.setup();
map.setup((double)fb_width / width);
resize(window, 0, 0);
@@ -159,21 +158,17 @@ public:
int run() {
while (!glfwWindowShouldClose(window)) {
f_it = futures.begin();
c_it = callbacks.begin();
while (f_it != futures.end()) {
if (f_it->wait_for(zero) == std::future_status::ready) {
std::function<void()> cb = *c_it;
futures.erase(f_it++);
callbacks.erase(c_it++);
cb();
bool& dirty = this->dirty;
requests.remove_if([&dirty](llmr::platform::Request * req) {
if (req->done) {
req->foreground_callback();
delete req;
dirty = true;
return true;
} else {
f_it++;
c_it++;
return false;
}
}
});
if (dirty) {
try {
@@ -185,7 +180,7 @@ public:
fps();
}
if (dirty || futures.size()) {
if (!requests.empty() || dirty) {
glfwPollEvents();
} else {
glfwWaitEvents();
@@ -237,46 +232,26 @@ MapView *mapView = nullptr;
namespace llmr {
namespace platform {
void restart(void *) {
void restart() {
if (mapView) {
mapView->dirty = true;
}
}
static size_t WriteCallback(void *contents, size_t size, size_t nmemb, void *userp) {
((std::string*)userp)->append((char*)contents, size *nmemb);
return size * nmemb;
Request *request_http(std::string url, std::function<void(Response&)> background_function, std::function<void()> foreground_callback) {
Request *req = new Request(url, background_function, foreground_callback);
requests.push_front(req);
return req;
}
void request_http_async(std::string url, std::function<void(Response&)> func) {
Response res;
CURL *curl;
CURLcode code;
curl = curl_easy_init();
if (curl) {
curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &res.body);
curl_easy_setopt(curl, CURLOPT_ACCEPT_ENCODING, "deflate");
code = curl_easy_perform(curl);
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &res.code);
curl_easy_cleanup(curl);
void cancel_request_http(Request *request) {
for (Request *req : requests) {
if (req == request) {
req->cancel();
}
}
func(res);
}
void request_http(std::string url, std::function<void(Response&)> func, std::function<void()> cb) {
futures.emplace_back(std::async(std::launch::async, request_http_async, url, func));
callbacks.push_back(cb);
}
double time() {
return glfwGetTime();
}
@@ -284,11 +259,21 @@ double time() {
}
}
int main() {
curl_global_init(CURL_GLOBAL_ALL);
llmr::platform::Request::initialize();
mapView = new MapView();
mapView->init();
int ret = mapView->run();
mapView->settings.sync();
delete mapView;
llmr::platform::Request::finish();
curl_global_cleanup();
return ret;
}

107
linux/request.cpp Normal file
View File

@@ -0,0 +1,107 @@
#include "request.hpp"
#include <atomic>
#include <cassert>
#include <llmr/util/threadpool.hpp>
#include <llmr/platform/platform.hpp>
using namespace llmr::platform;
CURLSH *Request::curl_share = nullptr;
pthread_mutex_t Request::curl_share_mutex = PTHREAD_MUTEX_INITIALIZER;
void Request::curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) {
pthread_mutex_lock(&curl_share_mutex);
}
void Request::curl_share_unlock(CURL *, curl_lock_data, void *) {
pthread_mutex_unlock(&curl_share_mutex);
}
size_t Request::curl_write_callback(void *contents, size_t size, size_t nmemb, void *userp) {
((std::string *)userp)->append((char *)contents, size * nmemb);
return size * nmemb;
}
int Request::curl_progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) {
Request *req = static_cast<Request *>(ptr);
return req->cancelled;
}
void Request::initialize() {
// curl init
curl_global_init(CURL_GLOBAL_ALL);
curl_share = curl_share_init();
curl_share_setopt(curl_share, CURLSHOPT_LOCKFUNC, curl_share_lock);
curl_share_setopt(curl_share, CURLSHOPT_UNLOCKFUNC, curl_share_unlock);
}
void Request::finish() {
pthread_key_delete(key);
curl_share_cleanup(curl_share);
}
Request::Request(std::string url, std::function<void(platform::Response&)> bg, std::function<void()> fg)
: done(false),
cancelled(false),
url(url),
background_function(bg),
foreground_callback(fg) {
llmr::util::threadpool->add(request, this);
}
pthread_key_t Request::key;
pthread_once_t Request::key_once = PTHREAD_ONCE_INIT;
void Request::create_key() {
pthread_key_create(&key, delete_key);
}
void Request::delete_key(void *ptr) {
if (ptr != nullptr) {
curl_easy_cleanup(ptr);
}
}
void Request::request(void *ptr) {
assert(curl_share);
Request *req = static_cast<Request *>(ptr);
Response res;
pthread_once(&key_once, create_key);
// TODO: use curl multi to be able to cancel, or to
CURL *curl = nullptr;
if ((curl = pthread_getspecific(key)) == nullptr) {
curl = curl_easy_init();
pthread_setspecific(key, curl);
}
curl_easy_reset(curl);
CURLcode code;
curl_easy_setopt(curl, CURLOPT_URL, req->url.c_str());
curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_write_callback);
curl_easy_setopt(curl, CURLOPT_WRITEDATA, &res.body);
curl_easy_setopt(curl, CURLOPT_PROGRESSFUNCTION, curl_progress_callback);
curl_easy_setopt(curl, CURLOPT_PROGRESSDATA, req);
curl_easy_setopt(curl, CURLOPT_NOPROGRESS, 0);
curl_easy_setopt(curl, CURLOPT_ACCEPT_ENCODING, "deflate");
curl_easy_setopt(curl, CURLOPT_SHARE, curl_share);
code = curl_easy_perform(curl);
curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &res.code);
if (code != CURLE_ABORTED_BY_CALLBACK) {
req->background_function(res);
}
req->done = true;
}
void Request::cancel() {
cancelled = true;
}

49
linux/request.hpp Normal file
View File

@@ -0,0 +1,49 @@
#ifndef LLMR_LINUX_REQUEST
#define LLMR_LINUX_REQUEST
#include <string>
#include <functional>
#include <curl/curl.h>
namespace llmr {
namespace platform {
struct Response;
class Request {
public:
Request(std::string url, std::function<void(platform::Response&)> bg, std::function<void()> fg);
static void initialize();
static void finish();
void cancel();
private:
static void create_key();
static void delete_key(void *ptr);
static void request(void *);
static size_t curl_write_callback(void *, size_t, size_t, void *);
static int curl_progress_callback(void *, double, double, double, double);
static void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *);
static void curl_share_unlock(CURL *, curl_lock_data, void *);
public:
static pthread_key_t key;
static pthread_once_t key_once;
std::atomic<bool> done;
std::atomic<bool> cancelled;
const std::string url;
const std::function<void(platform::Response&)> background_function;
const std::function<void()> foreground_callback;
private:
static CURLSH *curl_share;
static pthread_mutex_t curl_share_mutex;
};
}
}
#endif

17
macosx/main.mm
View File

@@ -240,7 +240,12 @@ void restart() {
[[NSApplication sharedApplication] postEvent: [NSEvent eventWithCGEvent:event] atStart:NO];
}
Request request_http(std::string url, std::function<void(Response&)> background_function, std::function<void()> foreground_callback)
struct Request {
int16_t identifier;
std::string original_url;
};
Request *request_http(std::string url, std::function<void(Response&)> background_function, std::function<void()> foreground_callback)
{
NSURLSessionDataTask *task = [[NSURLSession sharedSession] dataTaskWithURL:[NSURL URLWithString:@(url.c_str())] completionHandler:^(NSData *data, NSURLResponse *response, NSError *error)
{
@@ -262,20 +267,20 @@ Request request_http(std::string url, std::function<void(Response&)> background_
[task resume];
Request req;
Request *req = new Request();
req.identifier = task.taskIdentifier;
req.original_url = url;
req->identifier = task.taskIdentifier;
req->original_url = url;
return req;
}
void cancel_request_http(Request request)
void cancel_request_http(Request *request)
{
[[NSURLSession sharedSession] getTasksWithCompletionHandler:^(NSArray *dataTasks, NSArray *uploadTasks, NSArray *downloadTasks)
{
for (NSURLSessionDownloadTask *task in downloadTasks)
if (task.taskIdentifier == request.identifier)
if (task.taskIdentifier == request->identifier)
return [task cancel];
}];
}

2
src/map/tile.cpp
View File

@@ -74,7 +74,7 @@ void Tile::request() {
// Note: Somehow this feels slower than the change to request_http()
std::shared_ptr<Tile> tile = shared_from_this();
platform::Request request = platform::request_http(url, [=](platform::Response& res) {
platform::Request *request = platform::request_http(url, [=](platform::Response& res) {
if (res.code == 200 && tile->state != obsolete) {
tile->state = Tile::loaded;
tile->data.swap(res.body);

53
src/util/threadpool.cpp Normal file
View File

@@ -0,0 +1,53 @@
#include <llmr/util/threadpool.hpp>
#include <llmr/util/std.hpp>
#include <thread>
using namespace llmr::util;
std::unique_ptr<Threadpool> llmr::util::threadpool = std::make_unique<Threadpool>(std::thread::hardware_concurrency());
Threadpool::Threadpool(int max_workers)
: max_workers(max_workers) {
}
void Threadpool::add(Callback callback, void *data) {
if (worker_count < max_workers) {
worker_count++;
workers.emplace_front(*this);
}
pthread_mutex_lock(&mutex);
tasks.push(std::make_pair(callback, data));
pthread_mutex_unlock(&mutex);
pthread_cond_signal(&condition);
}
Threadpool::Worker::Worker(Threadpool& pool)
: pool(pool) {
pthread_create(&thread, nullptr, loop, (void *)this);
}
Threadpool::Worker::~Worker() {
pthread_cancel(thread);
}
void *Threadpool::Worker::loop(void *ptr) {
Worker *worker = static_cast<Worker *>(ptr);
Threadpool& pool = worker->pool;
pthread_mutex_lock(&pool.mutex);
while (true) {
if (pool.tasks.size()) {
Threadpool::Task task = pool.tasks.front();
pool.tasks.pop();
pthread_mutex_unlock(&pool.mutex);
task.first(task.second);
pthread_mutex_lock(&pool.mutex);
} else {
pthread_cond_wait(&pool.condition, &pool.mutex);
}
}
return nullptr;
}