From 720a6c64c4bd9e70f52a89d909560f6f26125887 Mon Sep 17 00:00:00 2001 From: Henning Baldersheim Date: Sat, 28 Mar 2020 01:02:36 +0000 Subject: Use atomics to avoid locking instructions in hot path. --- messagebus/src/vespa/messagebus/network/rpcservice.cpp | 10 +++------- .../src/vespa/messagebus/network/rpcserviceaddress.cpp | 2 +- .../src/vespa/messagebus/network/rpcserviceaddress.h | 5 ++--- messagebus/src/vespa/messagebus/network/rpcservicepool.cpp | 6 ++---- messagebus/src/vespa/messagebus/network/rpctarget.cpp | 11 ++++++----- messagebus/src/vespa/messagebus/network/rpctarget.h | 14 +++++++------- messagebus/src/vespa/messagebus/network/rpctargetpool.cpp | 9 +++++---- 7 files changed, 26 insertions(+), 31 deletions(-) diff --git a/messagebus/src/vespa/messagebus/network/rpcservice.cpp b/messagebus/src/vespa/messagebus/network/rpcservice.cpp index 6e7c73b38ee..337cea3322b 100644 --- a/messagebus/src/vespa/messagebus/network/rpcservice.cpp +++ b/messagebus/src/vespa/messagebus/network/rpcservice.cpp @@ -14,7 +14,7 @@ RPCService::RPCService(const Mirror &mirror, _addressList() { } -RPCService::~RPCService() {} +RPCService::~RPCService() = default; RPCServiceAddress::UP RPCService::resolve() @@ -22,9 +22,7 @@ RPCService::resolve() if (_pattern.find("tcp/") == 0) { size_t pos = _pattern.find_last_of('/'); if (pos != string::npos && pos < _pattern.size() - 1) { - RPCServiceAddress::UP ret(new RPCServiceAddress( - _pattern, - _pattern.substr(0, pos))); + auto ret = std::make_unique(_pattern, _pattern.substr(0, pos)); if (!ret->isMalformed()) { return ret; } @@ -37,9 +35,7 @@ RPCService::resolve() if (!_addressList.empty()) { _addressIdx = (_addressIdx + 1) % _addressList.size(); const AddressList::value_type &entry = _addressList[_addressIdx]; - return RPCServiceAddress::UP(new RPCServiceAddress( - entry.first, - entry.second)); + return std::make_unique(entry.first, entry.second); } } return RPCServiceAddress::UP(); diff --git a/messagebus/src/vespa/messagebus/network/rpcserviceaddress.cpp b/messagebus/src/vespa/messagebus/network/rpcserviceaddress.cpp index eac33195caa..e76832f0620 100644 --- a/messagebus/src/vespa/messagebus/network/rpcserviceaddress.cpp +++ b/messagebus/src/vespa/messagebus/network/rpcserviceaddress.cpp @@ -16,7 +16,7 @@ RPCServiceAddress::RPCServiceAddress(const string &serviceName, } } -RPCServiceAddress::~RPCServiceAddress() {} +RPCServiceAddress::~RPCServiceAddress() = default; bool RPCServiceAddress::isMalformed() diff --git a/messagebus/src/vespa/messagebus/network/rpcserviceaddress.h b/messagebus/src/vespa/messagebus/network/rpcserviceaddress.h index 142e654cc06..99a9f383e75 100644 --- a/messagebus/src/vespa/messagebus/network/rpcserviceaddress.h +++ b/messagebus/src/vespa/messagebus/network/rpcserviceaddress.h @@ -32,8 +32,7 @@ public: * @param serviceName The full service name of the address. * @param connectionSpec The connection specification. */ - RPCServiceAddress(const string &serviceName, - const string &connectionSpec); + RPCServiceAddress(const string &serviceName, const string &connectionSpec); ~RPCServiceAddress(); /** @@ -69,7 +68,7 @@ public: * * @param target The target to set. */ - void setTarget(RPCTarget::SP target) { _target = target; } + void setTarget(RPCTarget::SP target) { _target = std::move(target); } /** * Returns the RPC target to be used when communicating with the remove service. Make sure that {@link diff --git a/messagebus/src/vespa/messagebus/network/rpcservicepool.cpp b/messagebus/src/vespa/messagebus/network/rpcservicepool.cpp index b306cf29cf9..cad28b9f601 100644 --- a/messagebus/src/vespa/messagebus/network/rpcservicepool.cpp +++ b/messagebus/src/vespa/messagebus/network/rpcservicepool.cpp @@ -14,9 +14,7 @@ RPCServicePool::RPCServicePool(RPCNetwork &net, uint32_t maxSize) : assert(maxSize > 0); } -RPCServicePool::~RPCServicePool() -{ -} +RPCServicePool::~RPCServicePool() = default; RPCServiceAddress::UP RPCServicePool::resolve(const string &pattern) @@ -24,7 +22,7 @@ RPCServicePool::resolve(const string &pattern) if (_lru.hasKey(pattern)) { return _lru[pattern]->resolve(); } else { - RPCService::UP service(new RPCService(_net.getMirror(), pattern)); + auto service = std::make_unique(_net.getMirror(), pattern); auto result = service->resolve(); _lru[pattern] = std::move(service); return result; diff --git a/messagebus/src/vespa/messagebus/network/rpctarget.cpp b/messagebus/src/vespa/messagebus/network/rpctarget.cpp index 63470b6b707..58f1a6f0137 100644 --- a/messagebus/src/vespa/messagebus/network/rpctarget.cpp +++ b/messagebus/src/vespa/messagebus/network/rpctarget.cpp @@ -25,11 +25,12 @@ RPCTarget::~RPCTarget() void RPCTarget::resolveVersion(duration timeout, RPCTarget::IVersionHandler &handler) { - bool hasVersion = false; bool shouldInvoke = false; - { + ResolveState state = _state.load(std::memory_order_relaxed); + bool hasVersion = (state == VERSION_RESOLVED); + if ( ! hasVersion ) { vespalib::MonitorGuard guard(_lock); - if (_state == VERSION_RESOLVED || _state == PROCESSING_HANDLERS) { + if (state == PROCESSING_HANDLERS) { while (_state == PROCESSING_HANDLERS) { guard.wait(); } @@ -54,11 +55,11 @@ RPCTarget::resolveVersion(duration timeout, RPCTarget::IVersionHandler &handler) bool RPCTarget::isValid() const { - vespalib::MonitorGuard guard(_lock); if (_target.IsValid()) { return true; } - if (_state == TARGET_INVOKED || _state == PROCESSING_HANDLERS) { + ResolveState state = _state.load(std::memory_order_relaxed); + if (state == TARGET_INVOKED || state == PROCESSING_HANDLERS) { return true; // keep alive until RequestDone() is called } return false; diff --git a/messagebus/src/vespa/messagebus/network/rpctarget.h b/messagebus/src/vespa/messagebus/network/rpctarget.h index b6488f25cb7..d927292f26d 100644 --- a/messagebus/src/vespa/messagebus/network/rpctarget.h +++ b/messagebus/src/vespa/messagebus/network/rpctarget.h @@ -50,13 +50,13 @@ private: }; typedef std::unique_ptr Version_UP; - vespalib::Monitor _lock; - FRT_Supervisor &_orb; - string _name; - FRT_Target &_target; - ResolveState _state; - Version_UP _version; - HandlerList _versionHandlers; + vespalib::Monitor _lock; + FRT_Supervisor &_orb; + string _name; + FRT_Target &_target; + std::atomic _state; + Version_UP _version; + HandlerList _versionHandlers; public: /** diff --git a/messagebus/src/vespa/messagebus/network/rpctargetpool.cpp b/messagebus/src/vespa/messagebus/network/rpctargetpool.cpp index dd977efec08..cc09e44c460 100644 --- a/messagebus/src/vespa/messagebus/network/rpctargetpool.cpp +++ b/messagebus/src/vespa/messagebus/network/rpctargetpool.cpp @@ -63,19 +63,20 @@ RPCTargetPool::size() RPCTarget::SP RPCTargetPool::getTarget(FRT_Supervisor &orb, const RPCServiceAddress &address) { + const string & spec = address.getConnectionSpec(); + uint64_t currentTime = _timer->getMilliTime(); vespalib::LockGuard guard(_lock); - string spec = address.getConnectionSpec(); - TargetMap::iterator it = _targets.find(spec); + auto it = _targets.find(spec); if (it != _targets.end()) { Entry &entry = it->second; if (entry._target->isValid()) { - entry._lastUse = _timer->getMilliTime(); + entry._lastUse = currentTime; return entry._target; } _targets.erase(it); } auto ret = std::make_shared(spec, orb); - _targets.insert(TargetMap::value_type(spec, Entry(ret, _timer->getMilliTime()))); + _targets.insert(TargetMap::value_type(spec, Entry(ret, currentTime))); return ret; } -- cgit v1.2.3 From 506d2d5491448fea5683ea8d34957b10e048847b Mon Sep 17 00:00:00 2001 From: Henning Baldersheim Date: Sun, 29 Mar 2020 21:14:26 +0000 Subject: Only start LRU mode once half full --- messagebus/src/vespa/messagebus/network/rpcservice.cpp | 3 +-- staging_vespalib/src/vespa/vespalib/stllike/lrucache_map.hpp | 2 +- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/messagebus/src/vespa/messagebus/network/rpcservice.cpp b/messagebus/src/vespa/messagebus/network/rpcservice.cpp index 337cea3322b..fd1b84f545f 100644 --- a/messagebus/src/vespa/messagebus/network/rpcservice.cpp +++ b/messagebus/src/vespa/messagebus/network/rpcservice.cpp @@ -5,8 +5,7 @@ namespace mbus { -RPCService::RPCService(const Mirror &mirror, - const string &pattern) : +RPCService::RPCService(const Mirror &mirror, const string &pattern) : _mirror(mirror), _pattern(pattern), _addressIdx(random()), diff --git a/staging_vespalib/src/vespa/vespalib/stllike/lrucache_map.hpp b/staging_vespalib/src/vespa/vespalib/stllike/lrucache_map.hpp index d8d55c9b8c4..c8a3db878b8 100644 --- a/staging_vespalib/src/vespa/vespalib/stllike/lrucache_map.hpp +++ b/staging_vespalib/src/vespa/vespalib/stllike/lrucache_map.hpp @@ -267,7 +267,7 @@ typename lrucache_map

::internal_iterator lrucache_map

::findAndRef(const K & key) { internal_iterator found = HashTable::find(key); - if (found != HashTable::end()) { + if (found != HashTable::end() && (size()*2 > capacity())) { ref(found); } return found; -- cgit v1.2.3 From 42f80e15c96ef559799a17c68964f2a86e5bc934 Mon Sep 17 00:00:00 2001 From: Henning Baldersheim Date: Mon, 30 Mar 2020 10:25:32 +0000 Subject: Use original logic to avoid race condition. --- messagebus/src/vespa/messagebus/network/rpctarget.cpp | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/messagebus/src/vespa/messagebus/network/rpctarget.cpp b/messagebus/src/vespa/messagebus/network/rpctarget.cpp index 58f1a6f0137..ea21010e21c 100644 --- a/messagebus/src/vespa/messagebus/network/rpctarget.cpp +++ b/messagebus/src/vespa/messagebus/network/rpctarget.cpp @@ -26,12 +26,13 @@ void RPCTarget::resolveVersion(duration timeout, RPCTarget::IVersionHandler &handler) { bool shouldInvoke = false; - ResolveState state = _state.load(std::memory_order_relaxed); + ResolveState state = _state.load(std::memory_order_acquire); bool hasVersion = (state == VERSION_RESOLVED); if ( ! hasVersion ) { vespalib::MonitorGuard guard(_lock); - if (state == PROCESSING_HANDLERS) { - while (_state == PROCESSING_HANDLERS) { + state = _state.load(std::memory_order_relaxed); + if (state == VERSION_RESOLVED || state == PROCESSING_HANDLERS) { + while (_state.load(std::memory_order::memory_order_relaxed) == PROCESSING_HANDLERS) { guard.wait(); } hasVersion = true; -- cgit v1.2.3