#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
