/bitcoin/src/scheduler.cpp

Source
// Copyright (c) 2015-present The Bitcoin Core developers
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.

#include <scheduler.h>

#include <sync.h>
#include <util/time.h>

#include <cassert>
#include <functional>
#include <utility>

CScheduler::CScheduler() = default;

CScheduler::~CScheduler()
{
    assert(nThreadsServicingQueue == 0);
    if (stopWhenEmpty) assert(taskQueue.empty());
}


void CScheduler::serviceQueue()
{
    WAIT_LOCK(newTaskMutex, lock);
    ++nThreadsServicingQueue;

    // newTaskMutex is locked throughout this loop EXCEPT
    // when the thread is waiting or when the user's function
    // is called.
    while (!shouldStop()) {
        try {
            while (!shouldStop() && taskQueue.empty()) {
                // Wait until there is something to do.
                newTaskScheduled.wait(lock);
            }

            // Wait until either there is a new task, or until
            // the time of the first item on the queue:

            while (!shouldStop() && !taskQueue.empty()) {
                std::chrono::steady_clock::time_point timeToWaitFor = taskQueue.begin()->first;
                if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) {
                    break; // Exit loop after timeout, it means we reached the time of the event
                }
            }

            // If there are multiple threads, the queue can empty while we're waiting (another
            // thread may service the task we were waiting on).
            if (shouldStop() || taskQueue.empty())
                continue;

            Function f = taskQueue.begin()->second;
            taskQueue.erase(taskQueue.begin());

            {
                // Unlock before calling f, so it can reschedule itself or another task
                // without deadlocking:
                REVERSE_LOCK(lock, newTaskMutex);
                f();
            }
        } catch (...) {
            --nThreadsServicingQueue;
            throw;
        }
    }
    --nThreadsServicingQueue;
    newTaskScheduled.notify_one();
}

void CScheduler::schedule(CScheduler::Function f, std::chrono::steady_clock::time_point t)
{
    {
        LOCK(newTaskMutex);
        taskQueue.insert(std::make_pair(t, f));
    }
    newTaskScheduled.notify_one();
}

void CScheduler::MockForward(std::chrono::seconds delta_seconds)
{
    assert(delta_seconds > 0s && delta_seconds <= 1h);

    {
        LOCK(newTaskMutex);

        // use temp_queue to maintain updated schedule
        std::multimap<std::chrono::steady_clock::time_point, Function> temp_queue;

        for (const auto& element : taskQueue) {
            temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
        }

        // point taskQueue to temp_queue
        taskQueue = std::move(temp_queue);
    }

    // notify that the taskQueue needs to be processed
    newTaskScheduled.notify_one();
}

static void Repeat(CScheduler& s, CScheduler::Function f, std::chrono::milliseconds delta)
{
    f();
    s.scheduleFromNow([=, &s] { Repeat(s, f, delta); }, delta);
}

void CScheduler::scheduleEvery(CScheduler::Function f, std::chrono::milliseconds delta)
{
    scheduleFromNow([this, f, delta] { Repeat(*this, f, delta); }, delta);
}

size_t CScheduler::getQueueInfo(std::chrono::steady_clock::time_point& first,
                                std::chrono::steady_clock::time_point& last) const
{
    LOCK(newTaskMutex);
    size_t result = taskQueue.size();
    if (!taskQueue.empty()) {
        first = taskQueue.begin()->first;
        last = taskQueue.rbegin()->first;
    }
    return result;
}

bool CScheduler::AreThreadsServicingQueue() const
{
    LOCK(newTaskMutex);
    return nThreadsServicingQueue;
}


void SerialTaskRunner::MaybeScheduleProcessQueue()
{
    {
        LOCK(m_callbacks_mutex);
        // Try to avoid scheduling too many copies here, but if we
        // accidentally have two ProcessQueue's scheduled at once its
        // not a big deal.
        if (m_are_callbacks_running) return;
        if (m_callbacks_pending.empty()) return;
    }
    m_scheduler.schedule([this] { this->ProcessQueue(); }, std::chrono::steady_clock::now());
}

void SerialTaskRunner::ProcessQueue()
{
    std::function<void()> callback;
    {
        LOCK(m_callbacks_mutex);
        if (m_are_callbacks_running) return;
        if (m_callbacks_pending.empty()) return;
        m_are_callbacks_running = true;

        callback = std::move(m_callbacks_pending.front());
        m_callbacks_pending.pop_front();
    }

    // RAII the setting of fCallbacksRunning and calling MaybeScheduleProcessQueue
    // to ensure both happen safely even if callback() throws.
    struct RAIICallbacksRunning {
        SerialTaskRunner* instance;
        explicit RAIICallbacksRunning(SerialTaskRunner* _instance) : instance(_instance) {}
        ~RAIICallbacksRunning()
        {
            {
                LOCK(instance->m_callbacks_mutex);
                instance->m_are_callbacks_running = false;
            }
            instance->MaybeScheduleProcessQueue();
        }
    } raiicallbacksrunning(this);

    callback();
}

void SerialTaskRunner::insert(std::function<void()> func)
{
    {
        LOCK(m_callbacks_mutex);
        m_callbacks_pending.emplace_back(std::move(func));
    }
    MaybeScheduleProcessQueue();
}

void SerialTaskRunner::flush()
{
    assert(!m_scheduler.AreThreadsServicingQueue());
    bool should_continue = true;
    while (should_continue) {
        ProcessQueue();
        LOCK(m_callbacks_mutex);
        should_continue = !m_callbacks_pending.empty();
    }
}

size_t SerialTaskRunner::size()
{
    LOCK(m_callbacks_mutex);
    return m_callbacks_pending.size();
}

Coverage Report

Created: 2026-07-14 18:13

Line	Count	Source
1		// Copyright (c) 2015-present The Bitcoin Core developers
2		// Distributed under the MIT software license, see the accompanying
3		// file COPYING or http://www.opensource.org/licenses/mit-license.php.
4
5		#include <scheduler.h>
6
7		#include <sync.h>
8		#include <util/time.h>
9
10		#include <cassert>
11		#include <functional>
12		#include <utility>
13
14	27	CScheduler::CScheduler() = default;
15
16		CScheduler::~CScheduler()
17	150k	{
18	150k	assert(nThreadsServicingQueue == 0); Branch (18:5): [True: 150k, False: 0]
19	150k	if (stopWhenEmpty) assert(taskQueue.empty()); Branch (19:9): [True: 0, False: 150k] Branch (19:24): [True: 0, False: 0]
20	150k	}
21
22
23		void CScheduler::serviceQueue()
24	27	{
25	27	WAIT_LOCK(newTaskMutex, lock);
26	27	++nThreadsServicingQueue;
27
28		// newTaskMutex is locked throughout this loop EXCEPT
29		// when the thread is waiting or when the user's function
30		// is called.
31	2.07M	while (!shouldStop()) { Branch (31:12): [True: 1.92M, False: 150k]
32	1.92M	try {
33	1.92M	while (!shouldStop() && taskQueue.empty()) { Branch (33:20): [True: 1.92M, False: 0] Branch (33:37): [True: 0, False: 1.92M]
34		// Wait until there is something to do.
35	0	newTaskScheduled.wait(lock);
36	0	}
37
38		// Wait until either there is a new task, or until
39		// the time of the first item on the queue:
40
41	2.34M	while (!shouldStop() && !taskQueue.empty()) { Branch (41:20): [True: 2.34M, False: 282] Branch (41:37): [True: 2.34M, False: 0]
42	2.34M	std::chrono::steady_clock::time_point timeToWaitFor = taskQueue.begin()->first;
43	2.34M	if (newTaskScheduled.wait_until(lock, timeToWaitFor) == std::cv_status::timeout) { Branch (43:21): [True: 1.92M, False: 420k]
44	1.92M	break; // Exit loop after timeout, it means we reached the time of the event
45	1.92M	}
46	2.34M	}
47
48		// If there are multiple threads, the queue can empty while we're waiting (another
49		// thread may service the task we were waiting on).
50	1.92M	if (shouldStop() \|\| taskQueue.empty()) Branch (50:17): [True: 372, False: 1.92M] Branch (50:33): [True: 0, False: 1.92M]
51	150k	continue;
52
53	1.77M	Function f = taskQueue.begin()->second;
54	1.77M	taskQueue.erase(taskQueue.begin());
55
56	1.77M	{
57		// Unlock before calling f, so it can reschedule itself or another task
58		// without deadlocking:
59	1.77M	REVERSE_LOCK(lock, newTaskMutex);
60	1.77M	f();
61	1.77M	}
62	1.77M	} catch (...) {
63	0	--nThreadsServicingQueue;
64	0	throw;
65	0	}
66	1.92M	}
67	150k	--nThreadsServicingQueue;
68	150k	newTaskScheduled.notify_one();
69	150k	}
70
71		void CScheduler::schedule(CScheduler::Function f, std::chrono::steady_clock::time_point t)
72	2.20M	{
73	2.20M	{
74	2.20M	LOCK(newTaskMutex);
75	2.20M	taskQueue.insert(std::make_pair(t, f));
76	2.20M	}
77	2.20M	newTaskScheduled.notify_one();
78	2.20M	}
79
80		void CScheduler::MockForward(std::chrono::seconds delta_seconds)
81	85.7k	{
82	85.7k	assert(delta_seconds > 0s && delta_seconds <= 1h); Branch (82:5): [True: 85.7k, False: 0] Branch (82:5): [True: 85.7k, False: 0] Branch (82:5): [True: 85.7k, False: 0]
83
84	85.7k	{
85	85.7k	LOCK(newTaskMutex);
86
87		// use temp_queue to maintain updated schedule
88	85.7k	std::multimap<std::chrono::steady_clock::time_point, Function> temp_queue;
89
90	835k	for (const auto& element : taskQueue) { Branch (90:34): [True: 835k, False: 85.7k]
91	835k	temp_queue.emplace_hint(temp_queue.cend(), element.first - delta_seconds, element.second);
92	835k	}
93
94		// point taskQueue to temp_queue
95	85.7k	taskQueue = std::move(temp_queue);
96	85.7k	}
97
98		// notify that the taskQueue needs to be processed
99	85.7k	newTaskScheduled.notify_one();
100	85.7k	}
101
102		static void Repeat(CScheduler& s, CScheduler::Function f, std::chrono::milliseconds delta)
103	550k	{
104	550k	f();
105	550k	s.scheduleFromNow([=, &s] { Repeat(s, f, delta); }, delta);
106	550k	}
107
108		void CScheduler::scheduleEvery(CScheduler::Function f, std::chrono::milliseconds delta)
109	108	{
110	109k	scheduleFromNow([this, f, delta] { Repeat(*this, f, delta); }, delta);
111	108	}
112
113		size_t CScheduler::getQueueInfo(std::chrono::steady_clock::time_point& first,
114		std::chrono::steady_clock::time_point& last) const
115	0	{
116	0	LOCK(newTaskMutex);
117	0	size_t result = taskQueue.size();
118	0	if (!taskQueue.empty()) { Branch (118:9): [True: 0, False: 0]
119	0	first = taskQueue.begin()->first;
120	0	last = taskQueue.rbegin()->first;
121	0	}
122	0	return result;
123	0	}
124
125		bool CScheduler::AreThreadsServicingQueue() const
126	150k	{
127	150k	LOCK(newTaskMutex);
128	150k	return nThreadsServicingQueue;
129	150k	}
130
131
132		void SerialTaskRunner::MaybeScheduleProcessQueue()
133	2.30M	{
134	2.30M	{
135	2.30M	LOCK(m_callbacks_mutex);
136		// Try to avoid scheduling too many copies here, but if we
137		// accidentally have two ProcessQueue's scheduled at once its
138		// not a big deal.
139	2.30M	if (m_are_callbacks_running) return; Branch (139:13): [True: 111k, False: 2.19M]
140	2.19M	if (m_callbacks_pending.empty()) return; Branch (140:13): [True: 602k, False: 1.59M]
141	2.19M	}
142	1.59M	m_scheduler.schedule([this] { this->ProcessQueue(); }, std::chrono::steady_clock::now());
143	1.59M	}
144
145		void SerialTaskRunner::ProcessQueue()
146	1.45M	{
147	1.45M	std::function<void()> callback;
148	1.45M	{
149	1.45M	LOCK(m_callbacks_mutex);
150	1.45M	if (m_are_callbacks_running) return; Branch (150:13): [True: 0, False: 1.45M]
151	1.45M	if (m_callbacks_pending.empty()) return; Branch (151:13): [True: 377k, False: 1.07M]
152	1.07M	m_are_callbacks_running = true;
153
154	1.07M	callback = std::move(m_callbacks_pending.front());
155	1.07M	m_callbacks_pending.pop_front();
156	1.07M	}
157
158		// RAII the setting of fCallbacksRunning and calling MaybeScheduleProcessQueue
159		// to ensure both happen safely even if callback() throws.
160	0	struct RAIICallbacksRunning {
161	1.07M	SerialTaskRunner* instance;
162	1.07M	explicit RAIICallbacksRunning(SerialTaskRunner* _instance) : instance(_instance) {}
163	1.07M	~RAIICallbacksRunning()
164	1.07M	{
165	1.07M	{
166	1.07M	LOCK(instance->m_callbacks_mutex);
167	1.07M	instance->m_are_callbacks_running = false;
168	1.07M	}
169	1.07M	instance->MaybeScheduleProcessQueue();
170	1.07M	}
171	1.07M	} raiicallbacksrunning(this);
172
173	1.07M	callback();
174	1.07M	}
175
176		void SerialTaskRunner::insert(std::function<void()> func)
177	1.22M	{
178	1.22M	{
179	1.22M	LOCK(m_callbacks_mutex);
180	1.22M	m_callbacks_pending.emplace_back(std::move(func));
181	1.22M	}
182	1.22M	MaybeScheduleProcessQueue();
183	1.22M	}
184
185		void SerialTaskRunner::flush()
186	150k	{
187	150k	assert(!m_scheduler.AreThreadsServicingQueue()); Branch (187:5): [True: 150k, False: 0]
188	150k	bool should_continue = true;
189	300k	while (should_continue) { Branch (189:12): [True: 150k, False: 150k]
190	150k	ProcessQueue();
191	150k	LOCK(m_callbacks_mutex);
192	150k	should_continue = !m_callbacks_pending.empty();
193	150k	}
194	150k	}
195
196		size_t SerialTaskRunner::size()
197	118k	{
198	118k	LOCK(m_callbacks_mutex);
199	118k	return m_callbacks_pending.size();
200	118k	}