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
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#pragma once
#include "storagelinkqueued.h"
#include <vespa/storageframework/generic/thread/thread.h>
#include <vespa/storageframework/generic/component/component.h>
#include <vespa/vespalib/util/stringfmt.h>
#include <sstream>
#include <chrono>
#include <cassert>
namespace storage {
template<typename Message>
void
StorageLinkQueued::Dispatcher<Message>::terminate() {
if (_thread) {
_thread->interrupt();
{
std::lock_guard<std::mutex> guard(_sync);
_syncCond.notify_one();
}
_thread->join();
_thread.reset();
}
}
template<typename Message>
StorageLinkQueued::Dispatcher<Message>::Dispatcher(StorageLinkQueued& parent, unsigned int maxQueueSize, bool replyDispatcher)
: _parent(parent),
_maxQueueSize(maxQueueSize),
_sync(),
_syncCond(),
_messages(),
_replyDispatcher(replyDispatcher)
{
std::ostringstream name;
name << "Queued storage " << (_replyDispatcher ? "up" : "down")
<< "link - " << _parent.getName();
_component = std::make_unique<framework::Component>(parent.getComponentRegister(), name.str());
}
template<typename Message>
StorageLinkQueued::Dispatcher<Message>::~Dispatcher() {
terminate();
}
template<typename Message>
void StorageLinkQueued::Dispatcher<Message>::start()
{
assert( ! _thread);
_thread = _component->startThread(*this, 5s, 100ms);
}
template<typename Message>
void StorageLinkQueued::Dispatcher<Message>::add(const std::shared_ptr<Message>& m)
{
std::unique_lock<std::mutex> guard(_sync);
if ( ! _thread) start();
while ((_messages.size() > _maxQueueSize) && !_thread->interrupted()) {
_syncCond.wait_for(guard, 100ms);
}
_messages.push_back(m);
_syncCond.notify_one();
}
template<typename Message>
void StorageLinkQueued::Dispatcher<Message>::run(framework::ThreadHandle& h)
{
while (!h.interrupted()) {
h.registerTick(framework::PROCESS_CYCLE);
std::shared_ptr<Message> message;
{
std::unique_lock<std::mutex> guard(_sync);
while (!h.interrupted() && _messages.empty()) {
_syncCond.wait_for(guard, 100ms);
h.registerTick(framework::WAIT_CYCLE);
}
if (h.interrupted()) break;
message.swap(_messages.front());
}
try {
send(message);
} catch (std::exception& e) {
_parent.logError(vespalib::make_string(
"When running command %s, caught exception %s. "
"Discarding message",
message->toString().c_str(),
e.what()).c_str());
}
{
// Since flush() only waits for stack to be empty, we must
// pop stack AFTER send have been called.
std::lock_guard<std::mutex> guard(_sync);
_messages.pop_front();
_syncCond.notify_one();
}
}
_parent.logDebug("Finished storage link queued thread");
}
template<typename Message>
void StorageLinkQueued::Dispatcher<Message>::flush()
{
using namespace std::chrono_literals;
std::unique_lock<std::mutex> guard(_sync);
while (!_messages.empty()) {
_syncCond.wait_for(guard, 100ms);
}
}
}
|
