90d6f8bf62
libglim is an Apache-licensed C++ wrapper for lmdb, and rather than rolling our own it seems prudent to use it. Note: lmdb is not included in it, and unless something happens as did with libunbound, should be installed via each OS' package manager or equivalent.
402 lines
19 KiB
C++
402 lines
19 KiB
C++
#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
|