From 6c58101b12ee8e599aff3541f10c1fbfe5a027fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Sat, 1 Mar 2014 13:05:13 +0100 Subject: [PATCH 01/11] multithreaded curl for linux --- Makefile | 6 ++ include/llmr/map/tile.hpp | 2 +- include/llmr/platform/platform.hpp | 9 +-- linux/llmr-app.gyp | 56 ++++++++++++++--- linux/main.cpp | 98 ++++++++++++++++++++---------- macosx/main.mm | 17 ++++-- src/map/tile.cpp | 2 +- 7 files changed, 135 insertions(+), 55 deletions(-) diff --git a/Makefile b/Makefile index 854b5fb81..c6f4eb0b3 100644 --- a/Makefile +++ b/Makefile @@ -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 diff --git a/include/llmr/map/tile.hpp b/include/llmr/map/tile.hpp index d738c99bb..aba9e8a28 100644 --- a/include/llmr/map/tile.hpp +++ b/include/llmr/map/tile.hpp @@ -95,7 +95,7 @@ private: // Source data std::string data; const Style& style; - platform::Request req; + platform::Request *req = nullptr; }; } diff --git a/include/llmr/platform/platform.hpp b/include/llmr/platform/platform.hpp index ef137c206..fba922d1e 100644 --- a/include/llmr/platform/platform.hpp +++ b/include/llmr/platform/platform.hpp @@ -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; -}; +struct 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 background_function, std::function foreground_callback); +Request *request_http(std::string url, std::function background_function, std::function 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(); diff --git a/linux/llmr-app.gyp b/linux/llmr-app.gyp index 06d7ed822..79c1d4560 100644 --- a/linux/llmr-app.gyp +++ b/linux/llmr-app.gyp @@ -13,16 +13,52 @@ "./settings.cpp", "./settings.hpp" ], - 'product_extension': 'app', - 'link_settings': { - 'libraries': [ - '<@(glfw3_libraries)', - '-lcurl' - ], - }, - 'cflags': [ - '<@(png_cflags)', - '<@(glfw3_cflags)' + 'conditions': [ + ['OS == "mac"', { + 'product_extension': 'app', + 'mac_bundle_resources': [ + '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': '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" diff --git a/linux/main.cpp b/linux/main.cpp index 1e6c409a0..328027414 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -5,15 +5,28 @@ #include "settings.hpp" #include #include +#include #include -#include +#include + +namespace llmr { +namespace platform { + +struct Request { + pthread_t thread; + pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; + std::string url; + bool done = false; + std::function background_function; + std::function foreground_callback; +}; + +} +} + +std::forward_list requests; -std::list> futures; -std::list>::iterator f_it; -std::list> callbacks; -std::list>::iterator c_it; -std::chrono::milliseconds zero (0); class MapView { public: @@ -159,21 +172,22 @@ 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 cb = *c_it; - futures.erase(f_it++); - callbacks.erase(c_it++); - cb(); + bool& dirty = this->dirty; + requests.remove_if([&dirty](llmr::platform::Request *req) { + pthread_mutex_lock(&req->mutex); + if (req->done) { + pthread_mutex_unlock(&req->mutex); + req->foreground_callback(); + pthread_join(req->thread, nullptr); + pthread_mutex_destroy(&req->mutex); + delete req; + dirty = true; + return true; } else { - f_it++; - c_it++; + pthread_mutex_unlock(&req->mutex); + return false; } - } + }); if (dirty) { try { @@ -185,7 +199,7 @@ public: fps(); } - if (dirty || futures.size()) { + if (!requests.empty() || dirty) { glfwPollEvents(); } else { glfwWaitEvents(); @@ -237,29 +251,29 @@ 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); + ((std::string *)userp)->append((char *)contents, size * nmemb); return size * nmemb; } -void request_http_async(std::string url, std::function func) { - +void *request_http(void *ptr) { + Request *req = static_cast(ptr); Response res; + // TODO: use curl multi to be able to cancel + CURL *curl; CURLcode code; curl = curl_easy_init(); if (curl) { - curl_easy_setopt(curl, CURLOPT_URL, url.c_str()); + curl_easy_setopt(curl, CURLOPT_URL, req->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"); @@ -268,14 +282,33 @@ void request_http_async(std::string url, std::function func) { curl_easy_cleanup(curl); } - func(res); + req->background_function(res); + + pthread_mutex_lock(&req->mutex); + req->done = true; + pthread_mutex_unlock(&req->mutex); + + pthread_exit(nullptr); } -void request_http(std::string url, std::function func, std::function cb) { - futures.emplace_back(std::async(std::launch::async, request_http_async, url, func)); - callbacks.push_back(cb); +Request *request_http(std::string url, std::function background_function, std::function foreground_callback) { + Request *req = new Request(); + req->url = url; + req->background_function = background_function; + req->foreground_callback = foreground_callback; + requests.push_front(req); + int rc = pthread_create(&req->thread, nullptr, request_http, (void *)req); + if (rc) { + fprintf(stderr, "http request failed\n"); + return nullptr; + } + + return req; } +void cancel_request_http(Request *request) { + fprintf(stderr, "cancel %p\n", request); +} double time() { return glfwGetTime(); @@ -285,10 +318,13 @@ double time() { } int main() { + curl_global_init(CURL_GLOBAL_ALL); + mapView = new MapView(); mapView->init(); int ret = mapView->run(); mapView->settings.sync(); delete mapView; + pthread_exit(NULL); return ret; } diff --git a/macosx/main.mm b/macosx/main.mm index 973db9a86..bed18013c 100644 --- a/macosx/main.mm +++ b/macosx/main.mm @@ -240,7 +240,12 @@ void restart() { [[NSApplication sharedApplication] postEvent: [NSEvent eventWithCGEvent:event] atStart:NO]; } -Request request_http(std::string url, std::function background_function, std::function foreground_callback) +struct Request { + int16_t identifier; + std::string original_url; +}; + +Request *request_http(std::string url, std::function background_function, std::function 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 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]; }]; } diff --git a/src/map/tile.cpp b/src/map/tile.cpp index 6d88050a7..ff0824621 100644 --- a/src/map/tile.cpp +++ b/src/map/tile.cpp @@ -74,7 +74,7 @@ void Tile::request() { // Note: Somehow this feels slower than the change to request_http() std::shared_ptr 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); From bc1d6774c53a7a04ce8e3d203f233591ac7d1e45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Sat, 1 Mar 2014 17:52:59 +0100 Subject: [PATCH 02/11] make linux downloads cancellable --- linux/main.cpp | 52 ++++++++++++++++++++++++++++++++++++++++++++------ 1 file changed, 46 insertions(+), 6 deletions(-) diff --git a/linux/main.cpp b/linux/main.cpp index 328027414..325b34ad3 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -18,6 +18,7 @@ struct Request { pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; std::string url; bool done = false; + bool cancel = false; std::function background_function; std::function foreground_callback; }; @@ -25,9 +26,19 @@ struct Request { } } +pthread_mutex_t curl_share_mutex = PTHREAD_MUTEX_INITIALIZER; +CURLSH *curl_share; std::forward_list requests; +void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) { + pthread_mutex_lock(&curl_share_mutex); +} + +void curl_share_unlock(CURL *, curl_lock_data, void *) { + pthread_mutex_unlock(&curl_share_mutex); +} + class MapView { public: MapView() : @@ -173,7 +184,7 @@ public: int run() { while (!glfwWindowShouldClose(window)) { bool& dirty = this->dirty; - requests.remove_if([&dirty](llmr::platform::Request *req) { + requests.remove_if([&dirty](llmr::platform::Request * req) { pthread_mutex_lock(&req->mutex); if (req->done) { pthread_mutex_unlock(&req->mutex); @@ -257,11 +268,22 @@ void restart() { } } -static size_t WriteCallback(void *contents, size_t size, size_t nmemb, void *userp) { +static size_t curl_write_callback(void *contents, size_t size, size_t nmemb, void *userp) { ((std::string *)userp)->append((char *)contents, size * nmemb); return size * nmemb; } + +static int curl_progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { + Request *req = static_cast(ptr); + + pthread_mutex_lock(&req->mutex); + bool cancel = req->cancel; + pthread_mutex_unlock(&req->mutex); + + return cancel; +} + void *request_http(void *ptr) { Request *req = static_cast(ptr); Response res; @@ -274,15 +296,21 @@ void *request_http(void *ptr) { if (curl) { curl_easy_setopt(curl, CURLOPT_URL, req->url.c_str()); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, WriteCallback); + 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); curl_easy_cleanup(curl); - } - req->background_function(res); + if (code != CURLE_ABORTED_BY_CALLBACK) { + req->background_function(res); + } + } pthread_mutex_lock(&req->mutex); req->done = true; @@ -307,7 +335,13 @@ Request *request_http(std::string url, std::function background } void cancel_request_http(Request *request) { - fprintf(stderr, "cancel %p\n", request); + auto it = std::find(requests.begin(), requests.end(), request); + if (it != requests.end()) { + Request *req = *it; + pthread_mutex_lock(&req->mutex); + req->cancel = true; + pthread_mutex_unlock(&req->mutex); + } } double time() { @@ -317,14 +351,20 @@ double time() { } } + int main() { 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); mapView = new MapView(); mapView->init(); int ret = mapView->run(); mapView->settings.sync(); delete mapView; + + curl_share_cleanup(curl_share); pthread_exit(NULL); return ret; } From 7fdff4176b842004872db6d94cde11562abff2e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Sat, 1 Mar 2014 18:01:59 +0100 Subject: [PATCH 03/11] use atomics instead of mutexes --- linux/main.cpp | 21 ++++----------------- 1 file changed, 4 insertions(+), 17 deletions(-) diff --git a/linux/main.cpp b/linux/main.cpp index 325b34ad3..a7830c6f2 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -7,6 +7,7 @@ #include #include #include +#include #include @@ -15,10 +16,9 @@ namespace platform { struct Request { pthread_t thread; - pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; std::string url; - bool done = false; - bool cancel = false; + std::atomic done; + std::atomic cancel; std::function background_function; std::function foreground_callback; }; @@ -185,17 +185,13 @@ public: while (!glfwWindowShouldClose(window)) { bool& dirty = this->dirty; requests.remove_if([&dirty](llmr::platform::Request * req) { - pthread_mutex_lock(&req->mutex); if (req->done) { - pthread_mutex_unlock(&req->mutex); req->foreground_callback(); pthread_join(req->thread, nullptr); - pthread_mutex_destroy(&req->mutex); delete req; dirty = true; return true; } else { - pthread_mutex_unlock(&req->mutex); return false; } }); @@ -276,12 +272,7 @@ static size_t curl_write_callback(void *contents, size_t size, size_t nmemb, voi static int curl_progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { Request *req = static_cast(ptr); - - pthread_mutex_lock(&req->mutex); - bool cancel = req->cancel; - pthread_mutex_unlock(&req->mutex); - - return cancel; + return req->cancel; } void *request_http(void *ptr) { @@ -312,9 +303,7 @@ void *request_http(void *ptr) { } } - pthread_mutex_lock(&req->mutex); req->done = true; - pthread_mutex_unlock(&req->mutex); pthread_exit(nullptr); } @@ -338,9 +327,7 @@ void cancel_request_http(Request *request) { auto it = std::find(requests.begin(), requests.end(), request); if (it != requests.end()) { Request *req = *it; - pthread_mutex_lock(&req->mutex); req->cancel = true; - pthread_mutex_unlock(&req->mutex); } } From e134e6a8f3c4fdfef81d1a4f4e74b04853ffa4db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Sun, 2 Mar 2014 11:26:54 +0100 Subject: [PATCH 04/11] add threadpooled requests --- include/llmr/platform/platform.hpp | 2 +- linux/llmr-app.gyp | 4 +- linux/main.cpp | 102 ++---------------- linux/request.cpp | 166 +++++++++++++++++++++++++++++ linux/request.hpp | 33 ++++++ 5 files changed, 210 insertions(+), 97 deletions(-) create mode 100644 linux/request.cpp create mode 100644 linux/request.hpp diff --git a/include/llmr/platform/platform.hpp b/include/llmr/platform/platform.hpp index fba922d1e..d280ebb3a 100644 --- a/include/llmr/platform/platform.hpp +++ b/include/llmr/platform/platform.hpp @@ -15,7 +15,7 @@ namespace platform { // Restarts painting. This could for example trigger the event loop of the controlling application. void restart(); -struct Request; +class Request; struct Response { int16_t code = -1; diff --git a/linux/llmr-app.gyp b/linux/llmr-app.gyp index 79c1d4560..6c5002d0e 100644 --- a/linux/llmr-app.gyp +++ b/linux/llmr-app.gyp @@ -11,7 +11,9 @@ "sources": [ "./main.cpp", "./settings.cpp", - "./settings.hpp" + "./settings.hpp", + "./request.cpp", + "./request.hpp" ], 'conditions': [ ['OS == "mac"', { diff --git a/linux/main.cpp b/linux/main.cpp index a7830c6f2..f8da4fb95 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -1,44 +1,14 @@ #include #include -#include #include -#include "settings.hpp" #include #include -#include -#include -#include -#include +#include "settings.hpp" +#include "request.hpp" -namespace llmr { -namespace platform { - -struct Request { - pthread_t thread; - std::string url; - std::atomic done; - std::atomic cancel; - std::function background_function; - std::function foreground_callback; -}; - -} -} - -pthread_mutex_t curl_share_mutex = PTHREAD_MUTEX_INITIALIZER; -CURLSH *curl_share; std::forward_list requests; - -void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) { - pthread_mutex_lock(&curl_share_mutex); -} - -void curl_share_unlock(CURL *, curl_lock_data, void *) { - pthread_mutex_unlock(&curl_share_mutex); -} - class MapView { public: MapView() : @@ -187,7 +157,6 @@ public: requests.remove_if([&dirty](llmr::platform::Request * req) { if (req->done) { req->foreground_callback(); - pthread_join(req->thread, nullptr); delete req; dirty = true; return true; @@ -264,70 +233,17 @@ void restart() { } } -static size_t curl_write_callback(void *contents, size_t size, size_t nmemb, void *userp) { - ((std::string *)userp)->append((char *)contents, size * nmemb); - return size * nmemb; -} - - -static int curl_progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { - Request *req = static_cast(ptr); - return req->cancel; -} - -void *request_http(void *ptr) { - Request *req = static_cast(ptr); - Response res; - - // TODO: use curl multi to be able to cancel - - CURL *curl; - CURLcode code; - curl = curl_easy_init(); - - if (curl) { - 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); - curl_easy_cleanup(curl); - - if (code != CURLE_ABORTED_BY_CALLBACK) { - req->background_function(res); - } - } - - req->done = true; - - pthread_exit(nullptr); -} - Request *request_http(std::string url, std::function background_function, std::function foreground_callback) { - Request *req = new Request(); - req->url = url; - req->background_function = background_function; - req->foreground_callback = foreground_callback; + Request *req = new Request(url, background_function, foreground_callback); requests.push_front(req); - int rc = pthread_create(&req->thread, nullptr, request_http, (void *)req); - if (rc) { - fprintf(stderr, "http request failed\n"); - return nullptr; - } - return req; } void cancel_request_http(Request *request) { - auto it = std::find(requests.begin(), requests.end(), request); - if (it != requests.end()) { - Request *req = *it; - req->cancel = true; + for (Request *req : requests) { + if (req == request) { + req->cancel(); + } } } @@ -341,9 +257,6 @@ double time() { int main() { 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); mapView = new MapView(); mapView->init(); @@ -351,7 +264,6 @@ int main() { mapView->settings.sync(); delete mapView; - curl_share_cleanup(curl_share); pthread_exit(NULL); return ret; } diff --git a/linux/request.cpp b/linux/request.cpp new file mode 100644 index 000000000..5985e22b6 --- /dev/null +++ b/linux/request.cpp @@ -0,0 +1,166 @@ +#include "request.hpp" + +#include +#include +#include + +#include + +using namespace llmr::platform; + +class Threadpool; + +class Worker { +public: + Worker(Threadpool& pool); + + static void *loop(void *ptr) { + pthread_setname_np("worker"); + static_cast(ptr)->loop(); + pthread_exit(nullptr); + } + + void loop(); + +private: + Threadpool& pool; + pthread_t thread; +}; + + +class Threadpool { + friend class Worker; + +public: + typedef void (*Callback)(void *); + Threadpool(Callback callback); + + void add(void *data); + +private: + pthread_mutex_t mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER; + pthread_cond_t condition = PTHREAD_COND_INITIALIZER; + std::forward_list workers; + Callback callback; + std::queue tasks; +}; + + +Threadpool::Threadpool(Callback callback) + : callback(callback) { + const int worker_count = 4; + fprintf(stderr, "workers: %d\n", worker_count); + for (int i = 0; i < worker_count; i++) { + workers.emplace_front(*this); + } +} + +void Threadpool::add(void *data) { + pthread_mutex_lock(&mutex); + tasks.push(data); + pthread_mutex_unlock(&mutex); + pthread_cond_signal(&condition); +} + +Worker::Worker(Threadpool& pool) + : pool(pool) { + pthread_create(&thread, nullptr, loop, (void *)this); +} + +void Worker::loop() { + pthread_mutex_lock(&pool.mutex); + while (true) { + if (pool.tasks.size()) { + void *task = pool.tasks.front(); + pool.tasks.pop(); + pthread_mutex_unlock(&pool.mutex); + pool.callback(task); + pthread_mutex_lock(&pool.mutex); + } else { + pthread_cond_wait(&pool.condition, &pool.mutex); + } + } +} + +Threadpool *request_pool = nullptr; +CURLSH *curl_share = nullptr; +pthread_mutex_t curl_share_mutex = PTHREAD_MUTEX_INITIALIZER; + + +void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) { + pthread_mutex_lock(&curl_share_mutex); +} + +void curl_share_unlock(CURL *, curl_lock_data, void *) { + pthread_mutex_unlock(&curl_share_mutex); +} + + +Request::Request(std::string url, std::function bg, std::function fg) + : done(false), + cancelled(false), + url(url), + background_function(bg), + foreground_callback(fg) { + if (!request_pool) { + request_pool = new Threadpool(request); + } + if (!curl_share) { + // 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); + } + request_pool->add(this); +} + + +size_t Request::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::progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { + Request *req = static_cast(ptr); + bool cancel = req->cancelled; + if (cancel) { + fprintf(stderr, "cancel download %s\n", req->url.c_str()); + } + return cancel; +} + +void Request::request(void *ptr) { + Request *req = static_cast(ptr); + Response res; + + fprintf(stderr, "[%p] request %s\n", pthread_self(), req->url.c_str()); + + // TODO: use curl multi to be able to cancel, or to + + CURL *curl = curl_easy_init(); + CURLcode code; + + curl_easy_setopt(curl, CURLOPT_URL, req->url.c_str()); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_callback); + curl_easy_setopt(curl, CURLOPT_WRITEDATA, &res.body); + curl_easy_setopt(curl, CURLOPT_PROGRESSFUNCTION, 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); + curl_easy_cleanup(curl); + + if (code != CURLE_ABORTED_BY_CALLBACK) { + req->background_function(res); + } + + req->done = true; +} + +void Request::cancel() { + cancelled = true; +} diff --git a/linux/request.hpp b/linux/request.hpp new file mode 100644 index 000000000..337ab7522 --- /dev/null +++ b/linux/request.hpp @@ -0,0 +1,33 @@ +#ifndef LLMR_LINUX_REQUEST +#define LLMR_LINUX_REQUEST + +#include +#include +#include + +namespace llmr { +namespace platform { + +struct Response; + +class Request { +public: + Request(std::string url, std::function bg, std::function fg); + + static void request(void *); + static size_t write_callback(void *, size_t, size_t, void *); + static int progress_callback(void *, double, double, double, double); + void cancel(); + +public: + std::atomic done; + std::atomic cancelled; + const std::string url; + const std::function background_function; + const std::function foreground_callback; +}; + +} +} + +#endif From b26f3d319bbb8400b5608ced69534260134915ea Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 12:17:32 +0100 Subject: [PATCH 05/11] add threadpool implementation --- include/llmr/util/threadpool.hpp | 43 +++++++++ linux/llmr-app.gyp | 5 +- linux/main.cpp | 7 +- linux/request.cpp | 151 ++++++++----------------------- linux/request.hpp | 24 ++++- src/util/threadpool.cpp | 51 +++++++++++ 6 files changed, 164 insertions(+), 117 deletions(-) create mode 100644 include/llmr/util/threadpool.hpp create mode 100644 src/util/threadpool.cpp diff --git a/include/llmr/util/threadpool.hpp b/include/llmr/util/threadpool.hpp new file mode 100644 index 000000000..736e65c52 --- /dev/null +++ b/include/llmr/util/threadpool.hpp @@ -0,0 +1,43 @@ +#ifndef LLMR_UTIL_THREADPOOL +#define LLMR_UTIL_THREADPOOL + +#include +#include +#include + +namespace llmr { +namespace util { + +class Threadpool { + class Worker { + public: + Worker(Threadpool& pool); + ~Worker(); + static void *loop(void *ptr); + + private: + Threadpool& pool; + pthread_t thread; + }; + +public: + typedef void (*Callback)(void *); + Threadpool(Callback callback, int max_workers = 4); + + void add(void *data); + +private: + const int max_workers; + pthread_mutex_t mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER; + pthread_cond_t condition = PTHREAD_COND_INITIALIZER; + std::forward_list workers; + int worker_count = 0; + Callback callback; + std::queue tasks; +}; + +} +} + +#endif + diff --git a/linux/llmr-app.gyp b/linux/llmr-app.gyp index 6c5002d0e..0fa77085e 100644 --- a/linux/llmr-app.gyp +++ b/linux/llmr-app.gyp @@ -18,8 +18,9 @@ 'conditions': [ ['OS == "mac"', { 'product_extension': 'app', + 'mac_bundle': 1, 'mac_bundle_resources': [ - 'Icon.icns', + '../macosx/Icon.icns', ], 'link_settings': { 'libraries': [ @@ -42,7 +43,7 @@ '-lcurl', ], 'SDKROOT': 'macosx', - 'INFOPLIST_FILE': 'Info.plist', + 'INFOPLIST_FILE': '../macosx/Info.plist', 'CLANG_CXX_LIBRARY': 'libc++', 'CLANG_CXX_LANGUAGE_STANDARD':'c++11', 'MACOSX_DEPLOYMENT_TARGET':'10.9', diff --git a/linux/main.cpp b/linux/main.cpp index f8da4fb95..e9b6e4c36 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -258,12 +258,17 @@ 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; - pthread_exit(NULL); + + llmr::platform::Request::finish(); + + curl_global_cleanup(); return ret; } diff --git a/linux/request.cpp b/linux/request.cpp index 5985e22b6..b2881f789 100644 --- a/linux/request.cpp +++ b/linux/request.cpp @@ -1,128 +1,32 @@ #include "request.hpp" -#include -#include -#include +#include +#include +#include #include using namespace llmr::platform; -class Threadpool; -class Worker { -public: - Worker(Threadpool& pool); +llmr::util::Threadpool *Request::pool = nullptr; +CURLSH *Request::curl_share = nullptr; +pthread_mutex_t Request::curl_share_mutex = PTHREAD_MUTEX_INITIALIZER; - static void *loop(void *ptr) { - pthread_setname_np("worker"); - static_cast(ptr)->loop(); - pthread_exit(nullptr); - } - - void loop(); - -private: - Threadpool& pool; - pthread_t thread; -}; - - -class Threadpool { - friend class Worker; - -public: - typedef void (*Callback)(void *); - Threadpool(Callback callback); - - void add(void *data); - -private: - pthread_mutex_t mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER; - pthread_cond_t condition = PTHREAD_COND_INITIALIZER; - std::forward_list workers; - Callback callback; - std::queue tasks; -}; - - -Threadpool::Threadpool(Callback callback) - : callback(callback) { - const int worker_count = 4; - fprintf(stderr, "workers: %d\n", worker_count); - for (int i = 0; i < worker_count; i++) { - workers.emplace_front(*this); - } -} - -void Threadpool::add(void *data) { - pthread_mutex_lock(&mutex); - tasks.push(data); - pthread_mutex_unlock(&mutex); - pthread_cond_signal(&condition); -} - -Worker::Worker(Threadpool& pool) - : pool(pool) { - pthread_create(&thread, nullptr, loop, (void *)this); -} - -void Worker::loop() { - pthread_mutex_lock(&pool.mutex); - while (true) { - if (pool.tasks.size()) { - void *task = pool.tasks.front(); - pool.tasks.pop(); - pthread_mutex_unlock(&pool.mutex); - pool.callback(task); - pthread_mutex_lock(&pool.mutex); - } else { - pthread_cond_wait(&pool.condition, &pool.mutex); - } - } -} - -Threadpool *request_pool = nullptr; -CURLSH *curl_share = nullptr; -pthread_mutex_t curl_share_mutex = PTHREAD_MUTEX_INITIALIZER; - - -void curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) { +void Request::curl_share_lock(CURL *, curl_lock_data, curl_lock_access, void *) { pthread_mutex_lock(&curl_share_mutex); } -void curl_share_unlock(CURL *, curl_lock_data, void *) { +void Request::curl_share_unlock(CURL *, curl_lock_data, void *) { pthread_mutex_unlock(&curl_share_mutex); } - -Request::Request(std::string url, std::function bg, std::function fg) - : done(false), - cancelled(false), - url(url), - background_function(bg), - foreground_callback(fg) { - if (!request_pool) { - request_pool = new Threadpool(request); - } - if (!curl_share) { - // 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); - } - request_pool->add(this); -} - - -size_t Request::write_callback(void *contents, size_t size, size_t nmemb, void *userp) { +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::progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { +int Request::curl_progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { Request *req = static_cast(ptr); bool cancel = req->cancelled; if (cancel) { @@ -131,21 +35,46 @@ int Request::progress_callback(void *ptr, double dltotal, double dlnow, double u return cancel; } +void Request::initialize() { + pool = new llmr::util::Threadpool(Request::request); + + // 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() { + delete pool; + curl_share_cleanup(curl_share); +} + +Request::Request(std::string url, std::function bg, std::function fg) + : done(false), + cancelled(false), + url(url), + background_function(bg), + foreground_callback(fg) { + assert(pool); + pool->add(this); +} + void Request::request(void *ptr) { + assert(curl_share); + Request *req = static_cast(ptr); Response res; - fprintf(stderr, "[%p] request %s\n", pthread_self(), req->url.c_str()); - - // TODO: use curl multi to be able to cancel, or to + // TODO: use curl multi to be able to cancel, or to CURL *curl = curl_easy_init(); CURLcode code; curl_easy_setopt(curl, CURLOPT_URL, req->url.c_str()); - curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, write_callback); + curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, curl_write_callback); curl_easy_setopt(curl, CURLOPT_WRITEDATA, &res.body); - curl_easy_setopt(curl, CURLOPT_PROGRESSFUNCTION, progress_callback); + 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"); diff --git a/linux/request.hpp b/linux/request.hpp index 337ab7522..ff72afc96 100644 --- a/linux/request.hpp +++ b/linux/request.hpp @@ -5,7 +5,13 @@ #include #include + namespace llmr { + +namespace util { +class Threadpool; +} + namespace platform { struct Response; @@ -14,17 +20,29 @@ class Request { public: Request(std::string url, std::function bg, std::function fg); - static void request(void *); - static size_t write_callback(void *, size_t, size_t, void *); - static int progress_callback(void *, double, double, double, double); + static void initialize(); + static void finish(); + void cancel(); +private: + 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: std::atomic done; std::atomic cancelled; const std::string url; const std::function background_function; const std::function foreground_callback; + +private: + static util::Threadpool* pool; + static CURLSH *curl_share; + static pthread_mutex_t curl_share_mutex; }; } diff --git a/src/util/threadpool.cpp b/src/util/threadpool.cpp new file mode 100644 index 000000000..4f7a84cf8 --- /dev/null +++ b/src/util/threadpool.cpp @@ -0,0 +1,51 @@ +#include + +using namespace llmr::util; + +Threadpool::Threadpool(Callback callback, int max_workers) + : max_workers(max_workers), + callback(callback) { +} + +void Threadpool::add(void *data) { + if (worker_count < max_workers) { + workers.emplace_front(*this); + worker_count++; + } + + pthread_mutex_lock(&mutex); + tasks.push(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(ptr); + Threadpool& pool = worker->pool; + + pthread_setname_np("worker"); + pthread_mutex_lock(&pool.mutex); + while (true) { + if (pool.tasks.size()) { + void *task = pool.tasks.front(); + pool.tasks.pop(); + pthread_mutex_unlock(&pool.mutex); + pool.callback(task); + pthread_mutex_lock(&pool.mutex); + } else { + pthread_cond_wait(&pool.condition, &pool.mutex); + } + } + + return nullptr; +} From c076da0ae2f2e8c8221b727d9197ae5257c703f9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 12:26:04 +0100 Subject: [PATCH 06/11] make threadpool implementation callback agnostic --- include/llmr/util/threadpool.hpp | 8 ++++---- linux/request.cpp | 4 ++-- src/util/threadpool.cpp | 13 ++++++------- 3 files changed, 12 insertions(+), 13 deletions(-) diff --git a/include/llmr/util/threadpool.hpp b/include/llmr/util/threadpool.hpp index 736e65c52..f47bd5bf8 100644 --- a/include/llmr/util/threadpool.hpp +++ b/include/llmr/util/threadpool.hpp @@ -22,9 +22,10 @@ class Threadpool { public: typedef void (*Callback)(void *); - Threadpool(Callback callback, int max_workers = 4); + typedef std::pair Task; + Threadpool(int max_workers = 4); - void add(void *data); + void add(Callback callback, void *data); private: const int max_workers; @@ -32,8 +33,7 @@ private: pthread_cond_t condition = PTHREAD_COND_INITIALIZER; std::forward_list workers; int worker_count = 0; - Callback callback; - std::queue tasks; + std::queue tasks; }; } diff --git a/linux/request.cpp b/linux/request.cpp index b2881f789..2a5d5d9be 100644 --- a/linux/request.cpp +++ b/linux/request.cpp @@ -36,7 +36,7 @@ int Request::curl_progress_callback(void *ptr, double dltotal, double dlnow, dou } void Request::initialize() { - pool = new llmr::util::Threadpool(Request::request); + pool = new llmr::util::Threadpool(); // curl init curl_global_init(CURL_GLOBAL_ALL); @@ -57,7 +57,7 @@ Request::Request(std::string url, std::function bg, s background_function(bg), foreground_callback(fg) { assert(pool); - pool->add(this); + pool->add(request, this); } void Request::request(void *ptr) { diff --git a/src/util/threadpool.cpp b/src/util/threadpool.cpp index 4f7a84cf8..cf533c22b 100644 --- a/src/util/threadpool.cpp +++ b/src/util/threadpool.cpp @@ -2,19 +2,18 @@ using namespace llmr::util; -Threadpool::Threadpool(Callback callback, int max_workers) - : max_workers(max_workers), - callback(callback) { +Threadpool::Threadpool(int max_workers) + : max_workers(max_workers) { } -void Threadpool::add(void *data) { +void Threadpool::add(Callback callback, void *data) { if (worker_count < max_workers) { workers.emplace_front(*this); worker_count++; } pthread_mutex_lock(&mutex); - tasks.push(data); + tasks.push(std::make_pair(callback, data)); pthread_mutex_unlock(&mutex); pthread_cond_signal(&condition); } @@ -37,10 +36,10 @@ void *Threadpool::Worker::loop(void *ptr) { pthread_mutex_lock(&pool.mutex); while (true) { if (pool.tasks.size()) { - void *task = pool.tasks.front(); + Threadpool::Task task = pool.tasks.front(); pool.tasks.pop(); pthread_mutex_unlock(&pool.mutex); - pool.callback(task); + task.first(task.second); pthread_mutex_lock(&pool.mutex); } else { pthread_cond_wait(&pool.condition, &pool.mutex); From f850856cecf3358f5cf63a8cfae00a1bd65a6d8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 12:55:32 +0100 Subject: [PATCH 07/11] move requests to global threadpool --- include/llmr/util/threadpool.hpp | 13 +++++++++---- linux/main.cpp | 7 ++++++- linux/request.cpp | 13 ++----------- linux/request.hpp | 6 ------ src/util/threadpool.cpp | 12 ++++++++---- 5 files changed, 25 insertions(+), 26 deletions(-) diff --git a/include/llmr/util/threadpool.hpp b/include/llmr/util/threadpool.hpp index f47bd5bf8..817cdc651 100644 --- a/include/llmr/util/threadpool.hpp +++ b/include/llmr/util/threadpool.hpp @@ -4,30 +4,33 @@ #include #include #include +#include +#include "std.hpp" namespace llmr { namespace util { class Threadpool { +private: class Worker { public: - Worker(Threadpool& pool); + Worker(Threadpool& pool, const std::string name); ~Worker(); static void *loop(void *ptr); private: Threadpool& pool; + const std::string name; pthread_t thread; }; public: - typedef void (*Callback)(void *); - typedef std::pair Task; Threadpool(int max_workers = 4); - + typedef void (*Callback)(void *); void add(Callback callback, void *data); private: + typedef std::pair Task; const int max_workers; pthread_mutex_t mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER; pthread_cond_t condition = PTHREAD_COND_INITIALIZER; @@ -36,6 +39,8 @@ private: std::queue tasks; }; +extern std::unique_ptr threadpool; + } } diff --git a/linux/main.cpp b/linux/main.cpp index e9b6e4c36..a6814ab92 100644 --- a/linux/main.cpp +++ b/linux/main.cpp @@ -39,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); diff --git a/linux/request.cpp b/linux/request.cpp index 2a5d5d9be..b9066bf24 100644 --- a/linux/request.cpp +++ b/linux/request.cpp @@ -9,7 +9,6 @@ using namespace llmr::platform; -llmr::util::Threadpool *Request::pool = nullptr; CURLSH *Request::curl_share = nullptr; pthread_mutex_t Request::curl_share_mutex = PTHREAD_MUTEX_INITIALIZER; @@ -28,16 +27,10 @@ size_t Request::curl_write_callback(void *contents, size_t size, size_t nmemb, v int Request::curl_progress_callback(void *ptr, double dltotal, double dlnow, double ultotal, double ulnow) { Request *req = static_cast(ptr); - bool cancel = req->cancelled; - if (cancel) { - fprintf(stderr, "cancel download %s\n", req->url.c_str()); - } - return cancel; + return req->cancelled; } void Request::initialize() { - pool = new llmr::util::Threadpool(); - // curl init curl_global_init(CURL_GLOBAL_ALL); curl_share = curl_share_init(); @@ -46,7 +39,6 @@ void Request::initialize() { } void Request::finish() { - delete pool; curl_share_cleanup(curl_share); } @@ -56,8 +48,7 @@ Request::Request(std::string url, std::function bg, s url(url), background_function(bg), foreground_callback(fg) { - assert(pool); - pool->add(request, this); + llmr::util::threadpool->add(request, this); } void Request::request(void *ptr) { diff --git a/linux/request.hpp b/linux/request.hpp index ff72afc96..ae673e0da 100644 --- a/linux/request.hpp +++ b/linux/request.hpp @@ -7,11 +7,6 @@ namespace llmr { - -namespace util { -class Threadpool; -} - namespace platform { struct Response; @@ -40,7 +35,6 @@ public: const std::function foreground_callback; private: - static util::Threadpool* pool; static CURLSH *curl_share; static pthread_mutex_t curl_share_mutex; }; diff --git a/src/util/threadpool.cpp b/src/util/threadpool.cpp index cf533c22b..31d4fdc39 100644 --- a/src/util/threadpool.cpp +++ b/src/util/threadpool.cpp @@ -1,15 +1,18 @@ #include +#include using namespace llmr::util; +std::unique_ptr llmr::util::threadpool = std::make_unique(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) { - workers.emplace_front(*this); worker_count++; + workers.emplace_front(*this, std::string("worker ") + std::to_string(worker_count)); } pthread_mutex_lock(&mutex); @@ -18,8 +21,9 @@ void Threadpool::add(Callback callback, void *data) { pthread_cond_signal(&condition); } -Threadpool::Worker::Worker(Threadpool& pool) - : pool(pool) { +Threadpool::Worker::Worker(Threadpool& pool, const std::string name) + : pool(pool), + name(name) { pthread_create(&thread, nullptr, loop, (void *)this); } @@ -32,7 +36,7 @@ void *Threadpool::Worker::loop(void *ptr) { Worker *worker = static_cast(ptr); Threadpool& pool = worker->pool; - pthread_setname_np("worker"); + pthread_setname_np(worker->name.c_str()); pthread_mutex_lock(&pool.mutex); while (true) { if (pool.tasks.size()) { From f1f146b01e190570e4308a8781ae18f6931c7948 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 13:01:09 +0100 Subject: [PATCH 08/11] remove thread naming --- include/llmr/util/threadpool.hpp | 5 +---- src/util/threadpool.cpp | 9 ++++----- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/include/llmr/util/threadpool.hpp b/include/llmr/util/threadpool.hpp index 817cdc651..88b9df1a3 100644 --- a/include/llmr/util/threadpool.hpp +++ b/include/llmr/util/threadpool.hpp @@ -4,8 +4,6 @@ #include #include #include -#include -#include "std.hpp" namespace llmr { namespace util { @@ -14,13 +12,12 @@ class Threadpool { private: class Worker { public: - Worker(Threadpool& pool, const std::string name); + Worker(Threadpool& pool); ~Worker(); static void *loop(void *ptr); private: Threadpool& pool; - const std::string name; pthread_t thread; }; diff --git a/src/util/threadpool.cpp b/src/util/threadpool.cpp index 31d4fdc39..3b69302b1 100644 --- a/src/util/threadpool.cpp +++ b/src/util/threadpool.cpp @@ -1,4 +1,5 @@ #include +#include #include using namespace llmr::util; @@ -12,7 +13,7 @@ Threadpool::Threadpool(int max_workers) void Threadpool::add(Callback callback, void *data) { if (worker_count < max_workers) { worker_count++; - workers.emplace_front(*this, std::string("worker ") + std::to_string(worker_count)); + workers.emplace_front(*this); } pthread_mutex_lock(&mutex); @@ -21,9 +22,8 @@ void Threadpool::add(Callback callback, void *data) { pthread_cond_signal(&condition); } -Threadpool::Worker::Worker(Threadpool& pool, const std::string name) - : pool(pool), - name(name) { +Threadpool::Worker::Worker(Threadpool& pool) + : pool(pool) { pthread_create(&thread, nullptr, loop, (void *)this); } @@ -36,7 +36,6 @@ void *Threadpool::Worker::loop(void *ptr) { Worker *worker = static_cast(ptr); Threadpool& pool = worker->pool; - pthread_setname_np(worker->name.c_str()); pthread_mutex_lock(&pool.mutex); while (true) { if (pool.tasks.size()) { From e13b519745a5f1f3cfad66b9ef28c9292d550531 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 13:05:01 +0100 Subject: [PATCH 09/11] we're not doing recursive locking --- include/llmr/util/threadpool.hpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/include/llmr/util/threadpool.hpp b/include/llmr/util/threadpool.hpp index 88b9df1a3..c3649c60e 100644 --- a/include/llmr/util/threadpool.hpp +++ b/include/llmr/util/threadpool.hpp @@ -29,7 +29,7 @@ public: private: typedef std::pair Task; const int max_workers; - pthread_mutex_t mutex = PTHREAD_RECURSIVE_MUTEX_INITIALIZER; + pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; pthread_cond_t condition = PTHREAD_COND_INITIALIZER; std::forward_list workers; int worker_count = 0; From 880dc2d841af85e6651444f64ee09dcc72e85825 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 14:57:12 +0100 Subject: [PATCH 10/11] reuse curl handles --- linux/request.cpp | 25 +++++++++++++++++++++++-- linux/request.hpp | 4 ++++ 2 files changed, 27 insertions(+), 2 deletions(-) diff --git a/linux/request.cpp b/linux/request.cpp index b9066bf24..2c4ebd1bb 100644 --- a/linux/request.cpp +++ b/linux/request.cpp @@ -51,15 +51,37 @@ Request::Request(std::string url, std::function bg, s 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(ptr); Response res; + pthread_once(&key_once, create_key); // TODO: use curl multi to be able to cancel, or to - CURL *curl = curl_easy_init(); + 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()); @@ -72,7 +94,6 @@ void Request::request(void *ptr) { curl_easy_setopt(curl, CURLOPT_SHARE, curl_share); code = curl_easy_perform(curl); curl_easy_getinfo(curl, CURLINFO_RESPONSE_CODE, &res.code); - curl_easy_cleanup(curl); if (code != CURLE_ABORTED_BY_CALLBACK) { req->background_function(res); diff --git a/linux/request.hpp b/linux/request.hpp index ae673e0da..a61437c94 100644 --- a/linux/request.hpp +++ b/linux/request.hpp @@ -21,6 +21,8 @@ public: 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); @@ -28,6 +30,8 @@ private: static void curl_share_unlock(CURL *, curl_lock_data, void *); public: + static pthread_key_t key; + static pthread_once_t key_once; std::atomic done; std::atomic cancelled; const std::string url; From 45385d112b940c2826fa83fdce9bee5eb4f4f6a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Konstantin=20Ka=CC=88fer?= Date: Mon, 3 Mar 2014 14:58:15 +0100 Subject: [PATCH 11/11] delete key --- linux/request.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/linux/request.cpp b/linux/request.cpp index 2c4ebd1bb..4d760656b 100644 --- a/linux/request.cpp +++ b/linux/request.cpp @@ -39,6 +39,7 @@ void Request::initialize() { } void Request::finish() { + pthread_key_delete(key); curl_share_cleanup(curl_share); } @@ -51,7 +52,6 @@ Request::Request(std::string url, std::function bg, s llmr::util::threadpool->add(request, this); } - pthread_key_t Request::key; pthread_once_t Request::key_once = PTHREAD_ONCE_INIT;