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 Vespa.ai. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include "protocolrepository.h"
#include <cassert>
#include <vespa/log/log.h>
LOG_SETUP(".protocolrepository");
namespace mbus {
ProtocolRepository::ProtocolRepository() : _numProtocols(0) {}
ProtocolRepository::~ProtocolRepository() = default;
void
ProtocolRepository::clearPolicyCache()
{
std::lock_guard guard(_lock);
_routingPolicyCache.clear();
}
IProtocol::SP
ProtocolRepository::putProtocol(const IProtocol::SP & protocol)
{
const string &name = protocol->getName();
const auto numProtocols = _numProtocols.load();
size_t protocolIndex = numProtocols;
for (size_t i(0); i < numProtocols; i++) {
if (_protocols[i].first == name) {
protocolIndex = i;
break;
}
}
if (protocolIndex == numProtocols) {
assert(numProtocols < MAX_PROTOCOLS);
_protocols[protocolIndex].first = name;
_protocols[protocolIndex].second = nullptr;
// nullptr may be observed after increment but before protocol pointer
// update; this is fine as it has the same behavior as if the protocol
// has not yet been added.
const auto beforeAdd = _numProtocols.fetch_add(1, std::memory_order_release);
assert(beforeAdd == numProtocols); // Sanity check for racing inserters
} else {
clearPolicyCache();
}
_protocols[protocolIndex].second.store(protocol.get(), std::memory_order_release);
IProtocol::SP prev = _activeProtocols[name];
_activeProtocols[name] = protocol;
return prev;
}
IProtocol *
ProtocolRepository::getProtocol(const string &name)
{
const auto numProtocols = _numProtocols.load(std::memory_order_acquire);
for (size_t i(0); i < numProtocols; i++) {
if (_protocols[i].first == name) {
return _protocols[i].second.load(std::memory_order_acquire);
}
}
return nullptr;
}
IRoutingPolicy::SP
ProtocolRepository::getRoutingPolicy(const string &protocolName,
const string &policyName,
const string &policyParam)
{
string cacheKey = protocolName;
cacheKey.append('.').append(policyName).append(".").append(policyParam);
std::lock_guard guard(_lock);
RoutingPolicyCache::iterator cit = _routingPolicyCache.find(cacheKey);
if (cit != _routingPolicyCache.end()) {
return cit->second;
}
ProtocolMap::iterator pit = _activeProtocols.find(protocolName);
if (pit == _activeProtocols.end()) {
LOG(error, "Protocol '%s' not supported.", protocolName.c_str());
return IRoutingPolicy::SP();
}
IRoutingPolicy::UP policy;
try {
policy = pit->second->createPolicy(policyName, policyParam);
} catch (const std::exception &e) {
LOG(error, "Protocol '%s' threw an exception; %s", protocolName.c_str(), e.what());
}
if (policy.get() == nullptr) {
LOG(error, "Protocol '%s' failed to create routing policy '%s' with parameter '%s'.",
protocolName.c_str(), policyName.c_str(), policyParam.c_str());
return IRoutingPolicy::SP();
}
IRoutingPolicy::SP ret(policy.release());
_routingPolicyCache[cacheKey] = ret;
return ret;
}
}
|
