diff options
Diffstat (limited to 'external/glim/runner.hpp')
| -rw-r--r-- | external/glim/runner.hpp | 402 |
1 files changed, 0 insertions, 402 deletions
diff --git a/external/glim/runner.hpp b/external/glim/runner.hpp deleted file mode 100644 index 57c7e5248..000000000 --- a/external/glim/runner.hpp +++ /dev/null @@ -1,402 +0,0 @@ -#ifndef _GLIM_RUNNER_INCLUDED -#define _GLIM_RUNNER_INCLUDED - -#include <algorithm> // min -#include <atomic> -#include <condition_variable> -#include <chrono> -#include <functional> -#include <mutex> -#include <memory> -#include <stdexcept> -#include <thread> -#include <unordered_map> - -#include <curl/curl.h> -#include <event2/event.h> // cf. hiperfifo.cpp at http://article.gmane.org/gmane.comp.web.curl.library/37752 - -#include <boost/intrusive_ptr.hpp> -#include <boost/lockfree/queue.hpp> // http://www.boost.org/doc/libs/1_53_0/doc/html/boost/lockfree/queue.html -#include <boost/log/trivial.hpp> - -#include <time.h> -#include <stdlib.h> // rand -#include <sys/eventfd.h> - -#include "gstring.hpp" -#include "exception.hpp" - -namespace glim { - -/// Listens to messages returned by `curl_multi_info_read`. -/// NB: When CURL is queued with `addToCURLM` the CURL's `CURLOPT_PRIVATE` must point to the instance of `CurlmInformationListener`. -struct CurlmInformationListener { - enum FreeOptions {REMOVE_CURL_FROM_CURLM = 1, CURL_CLEANUP = 2, DELETE_LISTENER = 4, REMOVE_CLEAN_DELETE = 1|2|4}; - virtual FreeOptions information (CURLMsg*, CURLM*) = 0; - virtual ~CurlmInformationListener() {} -}; - -/// Listener deferring to a lambda. -struct FunCurlmLisneter: public glim::CurlmInformationListener { - std::function <void(CURLMsg*, CURLM*)> _fun; - FreeOptions _freeOptions; - FunCurlmLisneter (std::function <void(CURLMsg*, CURLM*)>&& fun, FreeOptions freeOptions): _fun (std::move (fun)), _freeOptions (freeOptions) {} - virtual FreeOptions information (CURLMsg* msg, CURLM* curlm) override { - if (__builtin_expect ((bool) _fun, 1)) - try {_fun (msg, curlm);} catch (const std::exception& ex) {BOOST_LOG_TRIVIAL (error) << "FunCurlmLisneter] " << ex.what();} - return _freeOptions; - } -}; - -/// Running cURL jobs in a single thread. -/// NB: The RunnerV2 *must* be allocated with `boost::intrusive_ptr` (typically you'd use `RunnerV2::instance()`). -class RunnerV2 { - std::atomic_int_fast32_t _references {0}; // For intrusive_ptr. - CURLM* _multi = nullptr; ///< Initialized in `run`. Should not be used outside of it. - int _eventFd = 0; ///< Used to give the `curl_multi_wait` some work when there's no cURL descriptors and to wake it from `withCURLM`. - boost::lockfree::queue<CURL*, boost::lockfree::capacity<64>> _queue; ///< `CURL` handles waiting to be added to `CURL_MULTI`. - std::thread _thread; - std::atomic_bool _running {false}; /// True if the `_thread` is running. - - using FreeOptions = CurlmInformationListener::FreeOptions; - - friend inline void intrusive_ptr_add_ref (RunnerV2*); - friend inline void intrusive_ptr_release (RunnerV2*); - - void run() noexcept { - try { - if (__builtin_expect (_references <= 0, 0)) GTHROW ("RunnerV2] Must be allocated with boost::intrusive_ptr!"); - _running = true; // NB: _running only becomes true if we're in the intrusive_ptr. ^^ - pthread_setname_np (pthread_self(), "Runner"); - _multi = curl_multi_init(); if (__builtin_expect (_multi == nullptr, 0)) GTHROW ("!curl_multi_init"); - _eventFd = eventfd (0, EFD_CLOEXEC | EFD_NONBLOCK); // Used to pause `curl_multi_wait` when there's no other jobs. - if (__builtin_expect (_eventFd == -1, 0)) GTHROW (std::string ("eventfd: ") + ::strerror (errno)); - while (__builtin_expect (_references > 0, 0)) { - // Reset the CURL_EVENT_FD value to 0, so that the `curl_multi_wait` can sleep. - if (__builtin_expect (_eventFd > 0, 1)) {eventfd_t count = 0; eventfd_read (_eventFd, &count);} - - // Add the queued CURL handles to our CURLM. - CURL* easy = nullptr; while (_queue.pop (easy)) curl_multi_add_handle (_multi, easy); - - // Run the cURL. - int runningHandles = 0; - CURLMcode rc = curl_multi_perform (_multi, &runningHandles); // http://curl.haxx.se/libcurl/c/curl_multi_perform.html - if (__builtin_expect (rc != CURLM_OK, 0)) BOOST_LOG_TRIVIAL (error) << "Runner] curl_multi_perform: " << curl_multi_strerror (rc); - - // Process the finished handles. - for (;;) { - int messagesLeft = 0; CURLMsg* msg = curl_multi_info_read (_multi, &messagesLeft); if (msg) try { - CURL* curl = msg->easy_handle; CurlmInformationListener* listener = 0; - if (__builtin_expect (curl_easy_getinfo (curl, CURLINFO_PRIVATE, &listener) == CURLE_OK, 1)) { - using FOP = CurlmInformationListener::FreeOptions; - FOP fop = listener->information (msg, _multi); - if (fop & FOP::REMOVE_CURL_FROM_CURLM) curl_multi_remove_handle (_multi, curl); - if (fop & FOP::CURL_CLEANUP) curl_easy_cleanup (curl); - if (fop & FOP::DELETE_LISTENER) delete listener; - } else { - curl_multi_remove_handle (_multi, curl); - curl_easy_cleanup (curl); - } - } catch (const std::exception& ex) {BOOST_LOG_TRIVIAL (error) << "Runner] " << ex.what();} - if (messagesLeft == 0) break; - } - - // Wait on the cURL file descriptors. - int descriptors = 0; - curl_waitfd waitfd = {_eventFd, CURL_WAIT_POLLIN, 0}; - eventfd_t eValue = 0; eventfd_read (_eventFd, &eValue); // Reset the curlEventFd value to zero. - rc = curl_multi_wait (_multi, &waitfd, 1, 100, &descriptors); // http://curl.haxx.se/libcurl/c/curl_multi_wait.html - if (__builtin_expect (rc != CURLM_OK, 0)) BOOST_LOG_TRIVIAL (error) << "Runner] curl_multi_wait: " << curl_multi_strerror (rc); - } - } catch (const std::exception& ex) {BOOST_LOG_TRIVIAL (error) << "Runner] " << ex.what();} - // Delayed destruction: when we're in intrusive_ptr (_running == true) but no longer referenced. - if (_running && _references == 0) delete this; // http://www.parashift.com/c++-faq-lite/delete-this.html - else _running = false; - } -public: - RunnerV2() { - // Start a thread using CURLM in a thread-safe way (that is, from this single thread only). - // NB: Handles *can* be passed between threads: http://article.gmane.org/gmane.comp.web.curl.library/33188 - _thread = std::thread (&RunnerV2::run, this); - } - ~RunnerV2() { - _thread.detach(); - } - - /// A singletone instance of the Runner used in order for different programes to reuse the same cURL thread. - static boost::intrusive_ptr<RunnerV2>& instance() { - static boost::intrusive_ptr<RunnerV2> INSTANCE (new RunnerV2()); - return INSTANCE; - } - - /// Schedule a CURL handler to be executed in the cURL thread. - /// NB: If the handle have a `CURLOPT_PRIVATE` option then it MUST point to an instance of `CurlmInformationListener`. - void addToCURLM (CURL* easyHandle) { - if (__builtin_expect (!_queue.push (easyHandle), 0)) GTHROW ("Can't push CURL* into the queue."); - if (__builtin_expect (_eventFd > 0, 1)) eventfd_write (_eventFd, 1); // Will wake the `curl_multi_wait` up, in order to run the `curl_multi_add_handle`. - } - - /// Schedule a CURL handler to be executed in the cURL thread. - /// NB: `CURLOPT_PRIVATE` is overwritten with a pointer to `FunCurlmLisneter`. - void addToCURLM (CURL* easyHandle, std::function <void(CURLMsg*, CURLM*)>&& listener, - FreeOptions freeOptions = static_cast<FreeOptions> (FreeOptions::REMOVE_CURL_FROM_CURLM | FreeOptions::DELETE_LISTENER)) { - FunCurlmLisneter* funListener = new FunCurlmLisneter (std::move (listener), freeOptions); // Will be deleted by the Runner. - curl_easy_setopt (easyHandle, CURLOPT_PRIVATE, funListener); // Tells `addToCURLM` to call this listener later. - addToCURLM (easyHandle); - } -}; - -inline void intrusive_ptr_add_ref (RunnerV2* runner) {++ runner->_references;} -inline void intrusive_ptr_release (RunnerV2* runner) {if (-- runner->_references == 0 && !runner->_running) delete runner;} - -/// Run CURLM requests and completion handlers, as well as other periodic jobs. -class Runner { - G_DEFINE_EXCEPTION (RunnerEx); - /// Free CURL during stack unwinding. - struct FreeCurl { - Runner* runner; CURL* curl; - FreeCurl (Runner* runner, CURL* curl): runner (runner), curl (curl) {} - ~FreeCurl() { - runner->_handlers.erase (curl); - curl_multi_remove_handle (runner->_curlm, curl); - curl_easy_cleanup (curl); - } - }; - public: - struct JobInfo; - /// The job must return `true` if Runner is to continue invoking it. - typedef std::function<bool(JobInfo& jobInfo)> job_t; - struct JobInfo { - job_t job; - float pauseSec = 1.0f; - struct timespec ran = {0, 0}; - }; -protected: - typedef std::function<void(CURLMsg*)> handler_t; - typedef std::function<void(const char* error)> errlog_t; - std::shared_ptr<struct event_base> _evbase; - errlog_t _errlog; - std::recursive_mutex _mutex; - typedef std::unique_ptr<struct event, void(*)(struct event*)> event_t; - std::unordered_map<CURL*, std::pair<handler_t, event_t>> _handlers; - /// Functions to run periodically. - typedef std::unordered_map<gstring, JobInfo> jobs_map_t; - jobs_map_t _jobs; - CURLM* _curlm = nullptr; - struct event* _timer = nullptr; - - /// Schedule a function to be run on the event loop. Useful to run all cURL methods on the single event loop thread. - template<typename F> - void doInEv (F fun, struct timeval after = {0, 0}) { - struct Dugout {F fun; struct event* timer; Dugout (F&& fun): fun (std::move (fun)), timer (nullptr) {}} *dugout = new Dugout (std::move (fun)); - event_callback_fn cb = [](evutil_socket_t, short, void* dugout_)->void { - Dugout* dugout = static_cast<Dugout*> (dugout_); - event_free (dugout->timer); dugout->timer = nullptr; - F fun = std::move (dugout->fun); delete dugout; - fun(); - }; - dugout->timer = evtimer_new (_evbase.get(), cb, dugout); - evtimer_add (dugout->timer, &after); - } - - bool shouldRun (jobs_map_t::value_type& entry, const struct timespec& ct) { - JobInfo& jobInfo = entry.second; - if (jobInfo.pauseSec <= 0.f) return true; // Run always. - if (jobInfo.ran.tv_sec == 0) {jobInfo.ran = ct; return true;} - float delta = (float)(ct.tv_sec - jobInfo.ran.tv_sec); - delta += (float)(ct.tv_nsec - jobInfo.ran.tv_nsec) / 1000000000.0f; - if (delta >= jobInfo.pauseSec) {jobInfo.ran = ct; return true;} - return false; - } - - /// Used for debugging. - static uint64_t ms() { - return std::chrono::duration_cast<std::chrono::milliseconds> (std::chrono::system_clock::now().time_since_epoch()) .count(); - } - /// Tells CURL to check its sockets. - void callCurlWithTimeout() { - //std::cout << __LINE__ << ',' << ms() << ": callCurlWithTimeout" << std::endl; - int running_handles = 0; - CURLMcode rc = curl_multi_socket_action (_curlm, CURL_SOCKET_TIMEOUT, 0, &running_handles); - if (rc != CURLM_OK) {GSTRING_ON_STACK (err, 256) << "glim::Runner: curl_multi_socket_action: " << curl_multi_strerror (rc); _errlog (err.c_str());} - } - - /// Should only be run when the _mutex is locked. - void checkForFinishedCurlJobs() { - //std::cout << __LINE__ << ',' << ms() << ": checkForFinishedCurlJobs" << std::endl; - nextMessage: - int msgs_in_queue = 0; - CURLMsg* msg = curl_multi_info_read (_curlm, &msgs_in_queue); - if (msg) try { - auto curl = msg->easy_handle; - FreeCurl freeCurl (this, curl); - auto it = _handlers.find (curl); - if (it != _handlers.end()) it->second.first (msg); - if (msgs_in_queue > 0) goto nextMessage; - } catch (const std::exception& ex) { - char eBuf[512]; gstring err (sizeof(eBuf), eBuf, false, 0); - err << "glim::Runner: handler: " << ex.what(); - _errlog (err.c_str()); - } - } - /// Will reset the timer unless there is a shorter timer already set. - void restartTimer (uint32_t nextInMicro = 100000) { // 100ms = 100000µs - struct timeval tv; - if (event_pending (_timer, EV_TIMEOUT, &tv) && !tv.tv_sec && tv.tv_usec < nextInMicro) return; // Already have a shorter timeout. - tv = {0, nextInMicro}; - evtimer_add (_timer, &tv); - } - static void evTimerCB (evutil_socket_t, short, void* runner_) { - //std::cout << __LINE__ << ',' << ms() << ": evTimerCB" << std::endl; - Runner* runner = (Runner*) runner_; - runner->callCurlWithTimeout(); - runner->run(); - } - /// event_callback_fn: There is an activity on a socket we are monitoring for CURL. - static void evSocketCB (evutil_socket_t sock, short events, void* runner_) { - //std::cout << __LINE__ << ',' << ms() << ": evSocketCB; sock: " << sock << "; events: " << events << std::endl; - Runner* runner = (Runner*) runner_; - int ev_bitmask = (events & EV_READ ? CURL_CSELECT_IN : 0) | (events & EV_WRITE ? CURL_CSELECT_OUT : 0); - int running_handles = 0; - CURLMcode rc = curl_multi_socket_action (runner->_curlm, sock, ev_bitmask, &running_handles); - if (rc != CURLM_OK) {GSTRING_ON_STACK (err, 256) << "glim::Runner: curl_multi_socket_action: " << curl_multi_strerror (rc); runner->_errlog (err.c_str());} - } - static void deleteEvent (struct event* ev) { - //std::cout << __LINE__ << ',' << ms() << ": deleteEvent: " << ev << std::endl; - event_del (ev); event_free (ev); - }; - /// curl_socket_callback: CURL asks us to monitor the socket. - static int curlSocketCB (CURL* easy, curl_socket_t sock, int what, void* runner_, void* socketp) { - //std::cout << __LINE__ << ',' << ms() << ": curlSocketCB; sock: " << sock << "; what: " << what; - //std::cout << " (" << (what == 0 ? "none" : what == 1 ? "in" : what == 2 ? "out" : what == 3 ? "inout" : what == 4 ? "remove" : "?") << ")" << std::endl; - Runner* runner = (Runner*) runner_; - std::lock_guard<std::recursive_mutex> lock (runner->_mutex); - if (what & CURL_POLL_REMOVE) { - auto it = runner->_handlers.find (easy); if (it != runner->_handlers.end()) it->second.second.reset(); - // We can't run `checkForFinishedCurlJobs` from there or bad things would happen - // (`curl_multi_remove_handle` will be called while we are still in the `curl_multi_socket_action`), - // but we can schedule the check via the libevent timer. - runner->restartTimer (0); - } else { - auto it = runner->_handlers.find (easy); if (it != runner->_handlers.end() && !it->second.second) { - event_callback_fn cb = evSocketCB; - struct event* ev = event_new (runner->_evbase.get(), sock, EV_READ | EV_WRITE | EV_ET | EV_PERSIST, cb, runner); - event_add (ev, nullptr); - //std::cout << __LINE__ << ',' << ms() << ": new event: " << ev << std::endl; - it->second.second = event_t (ev, deleteEvent); - } - } - return 0; - } - /// curl_multi_timer_callback: Schedule a CURL timer event or if `timeout_ms` is 0 then run immediately. - static int curlTimerCB (CURLM* multi, long timeout_ms, void* runner_) { - //std::cout << __LINE__ << ',' << ms() << ": curlTimerCB; timeout_ms: " << timeout_ms << std::endl; - if (timeout_ms == -1) return 0; // CURL tells us it doesn't need no timer. - Runner* runner = (Runner*) runner_; - if (timeout_ms == 0) { // CURL tells us it wants to run NOW. - runner->callCurlWithTimeout(); - return 0; - } - // CURL asks us to run it `timeout_ms` from now. - runner->restartTimer (std::min ((uint32_t) timeout_ms, (uint32_t) 100) * 1000); // We wait no more than 100ms. - return 0; - } -public: - Runner (std::shared_ptr<struct event_base> evbase, errlog_t errlog): _evbase (evbase), _errlog (errlog) { - doInEv ([this]() { - std::lock_guard<std::recursive_mutex> lock (_mutex); - _curlm = curl_multi_init(); if (!_curlm) GNTHROW (RunnerEx, "!curl_multi_init"); - auto check = [this](CURLMcode rc) {if (rc != CURLM_OK) {curl_multi_cleanup (_curlm); GNTHROW (RunnerEx, "curl_multi_setopt: " + std::to_string (rc));}}; - check (curl_multi_setopt (_curlm, CURLMOPT_SOCKETDATA, this)); - curl_socket_callback socketCB = curlSocketCB; check (curl_multi_setopt (_curlm, CURLMOPT_SOCKETFUNCTION, socketCB)); - check (curl_multi_setopt (_curlm, CURLMOPT_TIMERDATA, this)); - curl_multi_timer_callback curlTimerCB_ = curlTimerCB; check (curl_multi_setopt (_curlm, CURLMOPT_TIMERFUNCTION, curlTimerCB_)); - event_callback_fn evTimerCB_ = evTimerCB; _timer = evtimer_new (_evbase.get(), evTimerCB_, this); - restartTimer(); - }); - } - ~Runner() { - //std::cout << __LINE__ << ',' << ms() << ": ~Runner" << std::endl; - std::lock_guard<std::recursive_mutex> lock (_mutex); - if (_timer) {evtimer_del (_timer); event_free (_timer); _timer = nullptr;} - doInEv ([curlm = _curlm, handlers = std::move (_handlers)]() { - for (auto it = handlers.begin(), end = handlers.end(); it != end; ++it) { - curl_multi_remove_handle (curlm, it->first); - curl_easy_cleanup (it->first); - } - if (curlm) {curl_multi_cleanup (curlm);} - }); - _curlm = nullptr; - } - - /** Turns HTTP Pipelining on (or off). - * See http://curl.haxx.se/libcurl/c/curl_multi_setopt.html#CURLMOPTPIPELINING */ - Runner& pipeline (long enabled = 1) { - CURLMcode rc = curl_multi_setopt (_curlm, CURLMOPT_PIPELINING, enabled); - if (rc != CURLM_OK) GNTHROW (RunnerEx, "curl_multi_setopt: " + std::to_string (rc)); - return *this; - } - - /// Wait for the operation to complete, then call the `handler`, then free the `curl`. - void multi (CURL* curl, handler_t handler) { - { std::lock_guard<std::recursive_mutex> lock (_mutex); - _handlers.insert (std::make_pair (curl, std::make_pair (std::move (handler), event_t (nullptr, nullptr)))); } - doInEv ([this,curl]() { - curl_multi_add_handle (_curlm, curl); - }); - } - /// Register a new job to be run on the thread loop. - JobInfo& job (const gstring& name) { - std::lock_guard<std::recursive_mutex> lock (_mutex); - return _jobs[name]; - } - /// Register a new job to be run on the thread loop. - void schedule (const gstring& name, float pauseSec, job_t job) { - struct timespec ct; if (pauseSec > 0.f) clock_gettime (CLOCK_MONOTONIC, &ct); - std::lock_guard<std::recursive_mutex> lock (_mutex); - JobInfo& jobInfo = _jobs[name]; - jobInfo.job = job; - jobInfo.pauseSec = pauseSec; - if (pauseSec > 0.f) jobInfo.ran = ct; // If we need a pause then we also need to know when the job was scheduled. - } - /// Register a new job to be run on the thread loop. - void schedule (float pauseSec, job_t job) { - // Find a unique job name. - anotherName: - GSTRING_ON_STACK (name, 64) << "job" << rand(); - if (_jobs.find (name) != _jobs.end()) goto anotherName; - schedule (name, pauseSec, std::move (job)); - } - void removeJob (const gstring& name) { - std::lock_guard<std::recursive_mutex> lock (_mutex); - _jobs.erase (name); - } - /// Invoked automatically from a libevent timer; can also be invoked manually. - void run() { - _mutex.lock(); - checkForFinishedCurlJobs(); - // Run non-CURL jobs. Copy jobs into a local array in order not to run them with the `_mutex` locked. - struct timespec ct; clock_gettime (CLOCK_MONOTONIC, &ct); - JobInfo jobs[_jobs.size()]; gstring jobNames[_jobs.size()]; int jn = -1; { - for (auto it = _jobs.begin(), end = _jobs.end(); it != end; ++it) if (shouldRun (*it, ct)) { - ++jn; jobNames[jn] = it->first; jobs[jn] = it->second; - } } - _mutex.unlock(); - - for (; jn >= 0; --jn) try { - if (!jobs[jn].job (jobs[jn])) removeJob (jobNames[jn]); - } catch (const std::exception& ex) { - char eBuf[512]; gstring err (sizeof(eBuf), eBuf, false, 0); - err << "glim::Runner: error in job " << jobNames[jn] << ": " << ex.what(); - _errlog (err.c_str()); - } - restartTimer(); - } - - /// Expose CURLM. Useful for curl_multi_setopt (http://curl.haxx.se/libcurl/c/curl_multi_setopt.html). - CURLM* curlm() const {return _curlm;} -}; - -} // namespace glim - -#endif // _GLIM_RUNNER_INCLUDED |