Repository navigation
Expand file tree
/
Copy pathdispatch_queue.cpp
More file actions
118 lines (96 loc) · 3.82 KB
/
Copy pathdispatch_queue.cpp
File metadata and controls
118 lines (96 loc) · 3.82 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
#include <utility>
//!
//! Created by nick on 11/22/18.
//! Copyright(c) 2018 Iris. All rights reserved.
//! Use and copying of this software and preparation of derivative
//! works based upon this software are not permitted. Any distribution
//! of this software or derivative works must comply with all applicable
//! Canadian export control laws.
//! THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
//! WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
//! OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
//! DISCLAIMED. IN NO EVENT SHALL IRIS OR ITS CONTRIBUTORS
//! BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY,
//! OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
//! PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
//! OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON
//! ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
//! OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
//! OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
//! SUCH DAMAGE.
//!
#include <cpp_redis/misc/dispatch_queue.hpp>
namespace cpp_redis {
dispatch_queue::dispatch_queue(string_t name,
const notify_callback_t ¬ify_callback,
size_t thread_cnt)
: m_name(name), m_threads(thread_cnt), m_mq(),
notify_handler(notify_callback) {
printf("Creating dispatch queue: %s\n", name.c_str());
printf("Dispatch threads: %zu\n", thread_cnt);
for (auto &i : m_threads) {
i = std::thread(&dispatch_queue::dispatch_thread_handler, this);
}
}
dispatch_queue::~dispatch_queue() {
printf("Destructor: Destroying dispatch threads...\n");
// Signal to dispatch threads that it's time to wrap up
std::unique_lock<std::mutex> lock(m_threads_lock);
m_quit = true;
lock.unlock();
m_cv.notify_all();
// Wait for threads to finish before we exit
for (size_t i = 0; i < m_threads.size(); i++) {
if (m_threads[i].joinable()) {
printf("Destructor: Joining thread %zu until completion\n", i);
m_threads[i].join();
}
}
}
void dispatch_queue::dispatch(const cpp_redis::message_type &message,
const dispatch_callback_t &op) {
std::unique_lock<std::mutex> lock(m_threads_lock);
m_mq.push({op, message});
// Manual unlocking is done before notifying, to avoid waking up
// the waiting thread only to block again (see notify_one for details)
lock.unlock();
m_cv.notify_all();
}
void dispatch_queue::dispatch(const cpp_redis::message_type &message,
dispatch_callback_t &&op) {
std::unique_lock<std::mutex> lock(m_threads_lock);
m_mq.push({std::move(op), message});
// Manual unlocking is done before notifying, to avoid waking up
// the waiting thread only to block again (see notify_one for details)
lock.unlock();
m_cv.notify_all();
}
void dispatch_queue::dispatch_thread_handler() {
std::unique_lock<std::mutex> lock(m_threads_lock);
do {
// Wait until we have data or a quit signal
m_cv.wait(lock, [this] { return (!m_mq.empty() || m_quit); });
notify_handler(m_mq.size());
// after wait, we own the lock
if (!m_quit && !m_mq.empty()) {
auto op = std::move(m_mq.front());
m_mq.pop();
// unlock now that we're done messing with the queue
lock.unlock();
auto vals = op.message.get_values();
for (auto v : vals) {
std::cout << v.second << std::endl;
}
auto res = op.callback(op.message);
lock.lock();
}
} while (!m_quit);
}
size_t dispatch_queue::size() {
std::lock_guard<std::mutex> mq_lock(m_mq_mutex);
long res = m_mq.size();
// unlock now that we're done messing with the queue
// mq_lock.unlock();
return static_cast<size_t>(res);
}
} // namespace cpp_redis