]>
Commit | Line | Data |
---|---|---|
928b950e GA |
1 | // Copyright (c) 2015 The Bitcoin Core developers |
2 | // Distributed under the MIT software license, see the accompanying | |
bc909a7a | 3 | // file COPYING or https://www.opensource.org/licenses/mit-license.php . |
928b950e GA |
4 | |
5 | #include "scheduler.h" | |
6 | ||
6b51b9b1 CR |
7 | #include "reverselock.h" |
8 | ||
928b950e GA |
9 | #include <assert.h> |
10 | #include <boost/bind.hpp> | |
11 | #include <utility> | |
12 | ||
f5010548 | 13 | CScheduler::CScheduler() : nThreadsServicingQueue(0), stopRequested(false), stopWhenEmpty(false) |
928b950e GA |
14 | { |
15 | } | |
16 | ||
17 | CScheduler::~CScheduler() | |
18 | { | |
19 | assert(nThreadsServicingQueue == 0); | |
20 | } | |
21 | ||
928b950e GA |
22 | void CScheduler::serviceQueue() |
23 | { | |
24 | boost::unique_lock<boost::mutex> lock(newTaskMutex); | |
25 | ++nThreadsServicingQueue; | |
26 | ||
27 | // newTaskMutex is locked throughout this loop EXCEPT | |
28 | // when the thread is waiting or when the user's function | |
29 | // is called. | |
f5010548 | 30 | while (!shouldStop()) { |
928b950e | 31 | try { |
f5010548 | 32 | while (!shouldStop() && taskQueue.empty()) { |
928b950e GA |
33 | // Wait until there is something to do. |
34 | newTaskScheduled.wait(lock); | |
35 | } | |
f5010548 GA |
36 | |
37 | // Wait until either there is a new task, or until | |
38 | // the time of the first item on the queue: | |
928b950e | 39 | |
ef1d5060 CF |
40 | // Some boost versions have a conflicting overload of wait_until that returns void. |
41 | // Explicitly use a template here to avoid hitting that overload. | |
f5010548 | 42 | while (!shouldStop() && !taskQueue.empty() && |
ef1d5060 | 43 | newTaskScheduled.wait_until<>(lock, taskQueue.begin()->first) != boost::cv_status::timeout) { |
928b950e GA |
44 | // Keep waiting until timeout |
45 | } | |
c72a4732 | 46 | |
928b950e GA |
47 | // If there are multiple threads, the queue can empty while we're waiting (another |
48 | // thread may service the task we were waiting on). | |
f5010548 | 49 | if (shouldStop() || taskQueue.empty()) |
928b950e GA |
50 | continue; |
51 | ||
52 | Function f = taskQueue.begin()->second; | |
53 | taskQueue.erase(taskQueue.begin()); | |
54 | ||
626c5e69 CR |
55 | { |
56 | // Unlock before calling f, so it can reschedule itself or another task | |
57 | // without deadlocking: | |
6b51b9b1 | 58 | reverse_lock<boost::unique_lock<boost::mutex> > rlock(lock); |
626c5e69 CR |
59 | f(); |
60 | } | |
928b950e GA |
61 | } catch (...) { |
62 | --nThreadsServicingQueue; | |
63 | throw; | |
64 | } | |
65 | } | |
f5010548 GA |
66 | --nThreadsServicingQueue; |
67 | } | |
68 | ||
69 | void CScheduler::stop(bool drain) | |
70 | { | |
71 | { | |
72 | boost::unique_lock<boost::mutex> lock(newTaskMutex); | |
73 | if (drain) | |
74 | stopWhenEmpty = true; | |
75 | else | |
76 | stopRequested = true; | |
77 | } | |
78 | newTaskScheduled.notify_all(); | |
928b950e GA |
79 | } |
80 | ||
81 | void CScheduler::schedule(CScheduler::Function f, boost::chrono::system_clock::time_point t) | |
82 | { | |
83 | { | |
84 | boost::unique_lock<boost::mutex> lock(newTaskMutex); | |
85 | taskQueue.insert(std::make_pair(t, f)); | |
86 | } | |
87 | newTaskScheduled.notify_one(); | |
88 | } | |
89 | ||
90 | void CScheduler::scheduleFromNow(CScheduler::Function f, int64_t deltaSeconds) | |
91 | { | |
92 | schedule(f, boost::chrono::system_clock::now() + boost::chrono::seconds(deltaSeconds)); | |
93 | } | |
94 | ||
95 | static void Repeat(CScheduler* s, CScheduler::Function f, int64_t deltaSeconds) | |
96 | { | |
97 | f(); | |
98 | s->scheduleFromNow(boost::bind(&Repeat, s, f, deltaSeconds), deltaSeconds); | |
99 | } | |
100 | ||
101 | void CScheduler::scheduleEvery(CScheduler::Function f, int64_t deltaSeconds) | |
102 | { | |
103 | scheduleFromNow(boost::bind(&Repeat, this, f, deltaSeconds), deltaSeconds); | |
104 | } | |
f5010548 GA |
105 | |
106 | size_t CScheduler::getQueueInfo(boost::chrono::system_clock::time_point &first, | |
107 | boost::chrono::system_clock::time_point &last) const | |
108 | { | |
109 | boost::unique_lock<boost::mutex> lock(newTaskMutex); | |
110 | size_t result = taskQueue.size(); | |
111 | if (!taskQueue.empty()) { | |
112 | first = taskQueue.begin()->first; | |
113 | last = taskQueue.rbegin()->first; | |
114 | } | |
115 | return result; | |
116 | } |