blob: 80e5925f1a6449907d368fe5d9ae6a4386b86ddc (
plain) (
blame)
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
|
// Copyright Yahoo. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "resender.h"
#include "routingnode.h"
#include <vespa/messagebus/error.h>
#include <vespa/messagebus/errorcode.h>
#include <vespa/messagebus/tracelevel.h>
#include <vespa/vespalib/util/stringfmt.h>
using namespace std::chrono;
namespace mbus {
Resender::Resender(IRetryPolicy::SP retryPolicy)
: _queue_mutex(),
_queue(),
_retryPolicy(retryPolicy)
{ }
Resender::~Resender()
{
while (!_queue.empty()) {
_queue.top().second->discard();
_queue.pop();
}
}
void
Resender::resendScheduled()
{
using NodeList = std::vector<RoutingNode*>;
NodeList sendList;
time_point now = steady_clock::now();
{
std::lock_guard guard(_queue_mutex);
while (!_queue.empty() && _queue.top().first <= now) {
sendList.push_back(_queue.top().second);
_queue.pop();
}
}
for (RoutingNode *node : sendList) {
node->getTrace().trace(mbus::TraceLevel::COMPONENT, "Resender resending message.");
node->send();
}
}
bool
Resender::canRetry(uint32_t errorCode) const
{
return _retryPolicy->canRetry(errorCode);
}
bool
Resender::shouldRetry(const Reply &reply) const
{
uint32_t numErrors = reply.getNumErrors();
if (numErrors == 0) {
return false;
}
for (uint32_t i = 0; i < numErrors; ++i) {
if (!_retryPolicy->canRetry(reply.getError(i).getCode())) {
return false;
}
}
return true;
}
bool
Resender::scheduleRetry(RoutingNode &node)
{
Message &msg = node.getMessage();
if (!msg.getRetryEnabled()) {
return false;
}
uint32_t retry = msg.getRetry() + 1;
double delay = node.getReplyRef().getRetryDelay();
if (delay < 0) {
delay = _retryPolicy->getRetryDelay(retry);
}
milliseconds delayMS(long(delay * 1000));
if (msg.getTimeRemainingNow() <= delayMS) {
node.addError(ErrorCode::TIMEOUT, "Timeout exceeded by resender, giving up.");
return false;
}
node.prepareForRetry(); // consumes the reply
node.getTrace().trace(
TraceLevel::COMPONENT,
vespalib::make_string("Message scheduled for retry %u in %.3f seconds.", retry, delay));
msg.setRetry(retry);
std::lock_guard guard(_queue_mutex);
_queue.push(Entry(steady_clock::now() + delayMS, &node));
return true;
}
} // namespace mbus
|
