18 assert(nThreadsServicingQueue == 0);
19 if (stopWhenEmpty)
assert(taskQueue.empty());
26 ++nThreadsServicingQueue;
42 std::chrono::system_clock::time_point timeToWaitFor = taskQueue.begin()->first;
43 if (
newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) {
53 Function f = taskQueue.begin()->second;
54 taskQueue.erase(taskQueue.begin());
63 --nThreadsServicingQueue;
67 --nThreadsServicingQueue;
75 taskQueue.insert(std::make_pair(t, f));
82 assert(delta_seconds.count() > 0 && delta_seconds < std::chrono::hours{1});
88 std::multimap<std::chrono::system_clock::time_point, Function> temp_queue;
90 for (
const auto& element : taskQueue) {
91 temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
95 taskQueue = std::move(temp_queue);
114 std::chrono::system_clock::time_point& last)
const 117 size_t result = taskQueue.size();
118 if (!taskQueue.empty()) {
119 first = taskQueue.begin()->first;
120 last = taskQueue.rbegin()->first;
128 return nThreadsServicingQueue;
135 LOCK(m_cs_callbacks_pending);
139 if (m_are_callbacks_running)
return;
140 if (m_callbacks_pending.empty())
return;
147 std::function<void()> callback;
149 LOCK(m_cs_callbacks_pending);
150 if (m_are_callbacks_running)
return;
151 if (m_callbacks_pending.empty())
return;
152 m_are_callbacks_running =
true;
154 callback = std::move(m_callbacks_pending.front());
155 m_callbacks_pending.pop_front();
160 struct RAIICallbacksRunning {
163 ~RAIICallbacksRunning()
167 instance->m_are_callbacks_running =
false;
171 } raiicallbacksrunning(
this);
181 LOCK(m_cs_callbacks_pending);
182 m_callbacks_pending.emplace_back(std::move(func));
184 MaybeScheduleProcessQueue();
189 assert(!m_pscheduler->AreThreadsServicingQueue());
190 bool should_continue =
true;
191 while (should_continue) {
193 LOCK(m_cs_callbacks_pending);
194 should_continue = !m_callbacks_pending.empty();
200 LOCK(m_cs_callbacks_pending);
201 return m_callbacks_pending.size();
Class used by CScheduler clients which may schedule multiple jobs which are required to be run serial...
const std::chrono::seconds now
void scheduleEvery(Function f, std::chrono::milliseconds delta)
Repeat f until the scheduler is stopped.
static void Repeat(CScheduler &s, CScheduler::Function f, std::chrono::milliseconds delta)
void MaybeScheduleProcessQueue()
void MockForward(std::chrono::seconds delta_seconds)
Mock the scheduler to fast forward in time.
std::function< void()> Function
void AddToProcessQueue(std::function< void()> func)
Add a callback to be executed.
std::condition_variable newTaskScheduled
void schedule(Function f, std::chrono::system_clock::time_point t)
Call func at/after time t.
RecursiveMutex m_cs_callbacks_pending
void serviceQueue()
Services the queue 'forever'.
#define WAIT_LOCK(cs, name)
size_t CallbacksPending()
bool AreThreadsServicingQueue() const
Returns true if there are threads actively running in serviceQueue()
Simple class for background tasks that should be run periodically or once "after a while"...
size_t getQueueInfo(std::chrono::system_clock::time_point &first, std::chrono::system_clock::time_point &last) const
Returns number of tasks waiting to be serviced, and first and last task times.
void EmptyQueue()
Processes all remaining queue members on the calling thread, blocking until queue is empty Must be ca...
void scheduleFromNow(Function f, std::chrono::milliseconds delta)
Call f once after the delta has passed.
bool shouldStop() const EXCLUSIVE_LOCKS_REQUIRED(newTaskMutex)