blob: 342494c5a7a3ac6294c153e276ecd768845c77be (
plain)
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
|
// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab
/*
* Ceph - scalable distributed file system
*
* Copyright (C) 2014 CohortFS, LLC
*
* This is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License version 2.1, as published by the Free Software
* Foundation. See file COPYING.
*
*/
#include <string>
#include "QueueStrategy.h"
#define dout_subsys ceph_subsys_ms
#include "common/debug.h"
QueueStrategy::QueueStrategy(int _n_threads)
: n_threads(_n_threads),
stop(false),
mqueue(),
disp_threads()
{
}
void QueueStrategy::ds_dispatch(Message *m) {
msgr->ms_fast_preprocess(m);
if (msgr->ms_can_fast_dispatch(m)) {
msgr->ms_fast_dispatch(m);
return;
}
std::lock_guard l{lock};
mqueue.push_back(*m);
if (disp_threads.size()) {
if (! disp_threads.empty()) {
QSThread *thrd = &disp_threads.front();
disp_threads.pop_front();
thrd->cond.notify_all();
}
}
}
void QueueStrategy::entry(QSThread *thrd)
{
for (;;) {
ceph::ref_t<Message> m;
std::unique_lock l{lock};
for (;;) {
if (! mqueue.empty()) {
m = ceph::ref_t<Message>(&mqueue.front(), false);
mqueue.pop_front();
break;
}
if (stop)
break;
disp_threads.push_front(*thrd);
thrd->cond.wait(l);
}
l.unlock();
if (stop) {
if (!m) break;
continue;
}
get_messenger()->ms_deliver_dispatch(m);
}
}
void QueueStrategy::shutdown()
{
QSThread *thrd;
std::lock_guard l{lock};
stop = true;
while (disp_threads.size()) {
thrd = &(disp_threads.front());
disp_threads.pop_front();
thrd->cond.notify_all();
}
}
void QueueStrategy::wait()
{
std::unique_lock l{lock};
ceph_assert(stop);
for (auto& thread : threads) {
l.unlock();
// join outside of lock
thread->join();
l.lock();
}
}
void QueueStrategy::start()
{
ceph_assert(!stop);
std::lock_guard l{lock};
threads.reserve(n_threads);
for (int ix = 0; ix < n_threads; ++ix) {
std::string thread_name = "ms_qs_";
thread_name.append(std::to_string(ix));
auto thrd = std::make_unique<QSThread>(this);
thrd->create(thread_name.c_str());
threads.emplace_back(std::move(thrd));
}
}
|
