mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 12:28:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

misc: remove lock mutex template argument

Browse Source
This commit is contained in:
Adrien Béraud committed 2026-03-15 00:00:33 -04:00
1 parent 22d219c695
commit 99da0be2b7
20 files changed
+227 -227
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

include/opendht/dht_proxy_server.h
+2 -2
View File
@@ -425,7 +425,7 @@ private:
p.pack(clientId);
}
if (sessionCtx) {
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
p.pack("sid");
p.pack(sessionCtx->sessionId);
}
@@ -478,7 +478,7 @@ private:
p.pack("exp");
p.pack(to_time_t(expiration));
if (sessionCtx) {
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
p.pack("sid");
p.pack(sessionCtx->sessionId);
}
include/opendht/dhtrunner.h
+1 -1
View File
@@ -499,7 +499,7 @@ public:
*/
time_point loop()
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return loop_();
}
include/opendht/logger.h
+2 -2
View File
@@ -131,7 +131,7 @@ struct OPENDHT_PUBLIC Logger
std::shared_ptr<Logger> createChild(std::string tag)
{
std::lock_guard<std::mutex> lock {children_mutex_};
std::lock_guard lock {children_mutex_};
auto child = std::make_shared<Logger>(*this, std::move(tag));
children_.push_back(child);
return child;
@@ -139,7 +139,7 @@ struct OPENDHT_PUBLIC Logger
void setFilter(std::string_view filter)
{
std::lock_guard<std::mutex> lock {children_mutex_};
std::lock_guard lock {children_mutex_};
enable_ = filter.empty() or tag_.find(filter) != std::string_view::npos;
for (auto it = children_.begin(); it != children_.end();) {
if (auto c = it->lock()) {
include/opendht/network_utils.h
+6 -6
View File
@@ -58,7 +58,7 @@ public:
inline void setOnReceive(OnReceive&& cb)
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
rx_callback = std::move(cb);
}
@@ -67,12 +67,12 @@ public:
SockAddr getBound(sa_family_t family = AF_UNSPEC) const
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
return getBoundRef(family);
}
in_port_t getPort(sa_family_t family = AF_UNSPEC) const
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
return getBoundRef(family).getPort();
}
@@ -102,7 +102,7 @@ protected:
inline void onReceived(PacketList&& packets)
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
if (rx_callback) {
auto r = rx_callback(std::move(packets));
if (not r.empty() and toRecycle_.size() < RX_QUEUE_MAX_SIZE)
@@ -134,12 +134,12 @@ public:
bool hasIPv4() const override
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
return s4 != -1;
}
bool hasIPv6() const override
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
return s6 != -1;
}
include/opendht/thread_pool.h
+4 -4
View File
@@ -104,7 +104,7 @@ public:
void run(std::function<void()>&& task)
{
std::lock_guard<std::mutex> lock(state_->mtx);
std::lock_guard lock(state_->mtx);
if (state_->shutdown_)
return;
state_->pendingTasks++;
@@ -125,7 +125,7 @@ private:
void destroy(bool wait = true)
{
std::unique_lock<std::mutex> lock(mtx);
std::unique_lock lock(mtx);
if (destroyed)
return;
if (wait) {
@@ -141,7 +141,7 @@ private:
void run(const std::function<void()>& task)
{
{
std::lock_guard<std::mutex> lock(mtx);
std::lock_guard lock(mtx);
pendingTasks--;
ongoingTasks++;
}
@@ -149,7 +149,7 @@ private:
return;
task();
{
std::lock_guard<std::mutex> lock(mtx);
std::lock_guard lock(mtx);
ongoingTasks--;
cv.notify_all();
}
src/crypto.cpp
+2 -2
View File
@@ -446,7 +446,7 @@ PrivateKey::getPublicKey() const
const std::shared_ptr<PublicKey>&
PrivateKey::getSharedPublicKey() const
{
std::lock_guard<std::mutex> lock(publicKeyMutex_);
std::lock_guard lock(publicKeyMutex_);
if (not publicKey_) {
auto pk = std::make_shared<PublicKey>();
if (auto err = gnutls_pubkey_import_privkey(pk->pk, key, GNUTLS_KEY_KEY_CERT_SIGN | GNUTLS_KEY_CRL_SIGN, 0))
@@ -959,7 +959,7 @@ Certificate::getPublicKey() const
const std::shared_ptr<PublicKey>&
Certificate::getSharedPublicKey() const
{
std::lock_guard<std::mutex> lock(publicKeyMutex_);
std::lock_guard lock(publicKeyMutex_);
if (not publicKey_) {
auto pk = std::make_shared<PublicKey>();
if (auto err = gnutls_pubkey_import_x509(pk->pk, cert, 0))
src/dht_proxy_client.cpp
+45 -45
View File
@@ -195,7 +195,7 @@ DhtProxyClient::stop()
if (not isDestroying_.exchange(true)) {
std::shared_ptr<http::Resolver> resolver;
{
std::lock_guard<std::mutex> l(resolverLock_);
std::lock_guard l(resolverLock_);
resolver = std::move(resolver_);
}
if (resolver)
@@ -204,7 +204,7 @@ DhtProxyClient::stop()
if (infoState_)
infoState_->cancel = true;
{
std::lock_guard<std::mutex> lock(requestLock_);
std::lock_guard lock(requestLock_);
for (auto& request : requests_)
request.second->cancel();
}
@@ -219,7 +219,7 @@ DhtProxyClient::stop()
std::vector<Sp<Value>>
DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const
{
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto s = searches_.find(k);
if (s == searches_.end())
return {};
@@ -229,7 +229,7 @@ DhtProxyClient::getLocal(const InfoHash& k, const Value::Filter& filter) const
Sp<Value>
DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const
{
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto s = searches_.find(k);
if (s == searches_.end())
return {};
@@ -239,7 +239,7 @@ DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const
void
DhtProxyClient::cancelAllListeners()
{
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
if (logger_)
logger_->debug("[proxy:client] [listeners] [{} searches] cancel all", searches_.size());
for (auto& s : searches_) {
@@ -266,7 +266,7 @@ DhtProxyClient::shutdown(ShutdownCallback cb, bool)
NodeStatus
DhtProxyClient::getStatus(sa_family_t af) const
{
std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
std::lock_guard l(lockCurrentProxyInfos_);
switch (af) {
case AF_INET:
return statusIpv4_;
@@ -280,7 +280,7 @@ DhtProxyClient::getStatus(sa_family_t af) const
bool
DhtProxyClient::isRunning(sa_family_t af) const
{
std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
std::lock_guard l(lockCurrentProxyInfos_);
switch (af) {
case AF_INET:
return statusIpv4_ != NodeStatus::Disconnected;
@@ -297,7 +297,7 @@ DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr, const time_point& /*n
// Exec all currently stored callbacks
decltype(callbacks_) callbacks;
{
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks = std::move(callbacks_);
}
for (auto& callback : callbacks)
@@ -357,7 +357,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
}
if (not values.empty() and cb) {
{
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([opstate, cb, values = std::move(values)]() {
if (not opstate->stop.load() and not cb(values)) {
opstate->stop.store(true);
@@ -382,7 +382,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
}
if (donecb) {
{
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([donecb, opstate]() {
donecb(opstate->ok, {});
opstate->stop.store(true);
@@ -391,12 +391,12 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
loopSignal_();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
@@ -419,7 +419,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
std::shared_ptr<std::atomic_bool> ok;
if (permanent) {
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
ok = std::make_shared<std::atomic_bool>(true);
auto& search = searches_[key];
if (val->id) {
@@ -443,7 +443,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
if (ok)
*ok = result;
if (cb) {
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([cb, result]() { cb(result, {}); });
}
loopSignal_();
@@ -464,7 +464,7 @@ DhtProxyClient::handleRefreshPut(const asio::error_code& ec, InfoHash key, Value
}
if (logger_)
logger_->debug("[proxy:client] [put] [refresh {}]", key.to_view());
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto search = searches_.find(key);
if (search != searches_.end()) {
auto p = search->second.puts.find(id);
@@ -480,7 +480,7 @@ DhtProxyClient::handleRefreshPut(const asio::error_code& ec, InfoHash key, Value
std::shared_ptr<http::Request>
DhtProxyClient::buildRequest(const std::string& target)
{
std::unique_lock<std::mutex> l(resolverLock_);
std::unique_lock l(resolverLock_);
auto resolver = resolver_;
l.unlock();
if (not resolver)
@@ -535,7 +535,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
auto id = dht::Value(parsedValue).id;
val->id = id;
if (permanent) {
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto& search = searches_[key];
auto it = search.pendingPuts.find(val);
if (it != search.pendingPuts.end()) {
@@ -565,12 +565,12 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallbackSimple cb,
if (cb)
cb(ok);
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
@@ -640,7 +640,7 @@ DhtProxyClient::getProxyInfos()
logger_->debug("[proxy:client] [info] requesting proxy server node information");
auto infoState = std::make_shared<InfoState>();
{
std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
std::lock_guard l(lockCurrentProxyInfos_);
if (infoState_)
infoState_->cancel = true;
infoState_ = infoState;
@@ -655,7 +655,7 @@ DhtProxyClient::getProxyInfos()
auto resolver = std::make_shared<http::Resolver>(httpContext_, proxyUrl_, logger_);
queryProxyInfo(infoState, resolver, AF_INET);
queryProxyInfo(infoState, resolver, AF_INET6);
std::lock_guard<std::mutex> l(resolverLock_);
std::lock_guard l(resolverLock_);
resolver_ = resolver;
}
@@ -701,7 +701,7 @@ DhtProxyClient::queryProxyInfo(const Sp<InfoState>& infoState, const Sp<http::Re
}
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
@@ -709,7 +709,7 @@ DhtProxyClient::queryProxyInfo(const Sp<InfoState>& infoState, const Sp<http::Re
if (infoState->cancel.load())
return;
{
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
@@ -724,7 +724,7 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
{
if (isDestroying_)
return;
std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
std::unique_lock l(lockCurrentProxyInfos_);
auto oldStatus = std::max(statusIpv4_, statusIpv6_);
auto& status = family == AF_INET ? statusIpv4_ : statusIpv6_;
auto& pubAddress = family == AF_INET ? publicAddressV4_ : publicAddressV6_;
@@ -761,7 +761,7 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
addresses.emplace_back(publicAddressV4_);
if (publicAddressV6_)
addresses.emplace_back(publicAddressV6_);
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back(
[cb = publicAddressChangedCb_, addresses = std::move(addresses)]() { cb(std::move(addresses)); });
}
@@ -785,7 +785,7 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
if (not onConnectCallbacks_.empty()) {
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([cbs = std::move(onConnectCallbacks_)]() mutable {
while (not cbs.empty()) {
cbs.front()();
@@ -820,7 +820,7 @@ DhtProxyClient::parsePublicAddress(const Json::Value& val)
std::vector<SockAddr>
DhtProxyClient::getPublicAddress(sa_family_t family)
{
std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
std::lock_guard l(lockCurrentProxyInfos_);
std::vector<SockAddr> result;
if (publicAddressV6_ && family != AF_INET)
result.emplace_back(publicAddressV6_);
@@ -837,7 +837,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
if (isDestroying_)
return 0;
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto& search = searches_[key];
auto query = std::make_shared<Query>(Select {}, std::move(where));
return search.ops
@@ -874,7 +874,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
system_clock::time_point t) {
if (opstate->stop)
return false;
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto s = searches_.find(key);
if (s != searches_.end()) {
auto l = s->second.listeners.find(token);
@@ -929,7 +929,7 @@ DhtProxyClient::handleResubscribe(const asio::error_code& ec,
}
if (opstate->stop)
return;
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto s = searches_.find(key);
if (s != searches_.end()) {
auto l = s->second.listeners.find(token);
@@ -948,7 +948,7 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken)
if (logger_)
logger_->debug("[proxy:client] [search {}] cancel listen {}", key.to_view(), gtoken);
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
// find the listener in cache
auto it = searches_.find(key);
if (it == searches_.end())
@@ -979,7 +979,7 @@ DhtProxyClient::handleExpireListener(const asio::error_code& ec, const InfoHash&
if (logger_)
logger_->debug("[proxy:client] [listen {}] expire listener", key.to_view());
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto search = searches_.find(key);
if (search == searches_.end())
return;
@@ -1015,12 +1015,12 @@ DhtProxyClient::handleExpireListener(const asio::error_code& ec, const InfoHash&
opFailed();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
@@ -1097,7 +1097,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
if (cb) {
auto expired = json.get("expired", Json::Value(false)).asBool();
{
std::lock_guard<std::mutex> lock(lockCallbacks_);
std::lock_guard lock(lockCallbacks_);
callbacks_.emplace_back([cb, value, opstate, expired]() {
if (not opstate->stop.load()
and not cb({value}, expired, system_clock::time_point::min()))
@@ -1124,12 +1124,12 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t& header,
opFailed();
}
if (not isDestroying_) {
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->add_on_status_callback([request, seconds = this->listenKeepIdle()](unsigned status_code) {
@@ -1153,7 +1153,7 @@ DhtProxyClient::opFailed()
if (logger_)
logger_->error("[proxy:client] proxy request failed");
{
std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
std::lock_guard l(lockCurrentProxyInfos_);
statusIpv4_ = NodeStatus::Disconnected;
statusIpv6_ = NodeStatus::Disconnected;
}
@@ -1186,7 +1186,7 @@ DhtProxyClient::restartListeners(const asio::error_code& ec)
if (logger_)
logger_->d("[proxy:client] [listeners] refresh permanent puts");
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
for (auto& search : searches_) {
const auto& key = search.first;
for (auto& put : search.second.puts) {
@@ -1249,7 +1249,7 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
#ifdef OPENDHT_PUSH_NOTIFICATIONS
{
// If a push notification is received, the proxy is up and running
std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
std::lock_guard l(lockCurrentProxyInfos_);
auto oldStatus = std::max(statusIpv4_, statusIpv6_);
if (oldStatus != NodeStatus::Connected)
launchConnectedCbs_ = true;
@@ -1264,7 +1264,7 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
logger_->debug("[proxy:client] [push] ignoring push for other session");
return PushNotificationResult::IgnoredWrongSession;
}
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto timeout = notification.find("timeout");
if (timeout != notification.cend()) {
InfoHash key(timeout->second);
@@ -1326,11 +1326,11 @@ DhtProxyClient::pushNotificationReceived([[maybe_unused]] const std::map<std::st
ids.emplace_back(std::stoull(substr));
}
{
std::lock_guard<std::mutex> lockCb(lockCallbacks_);
std::lock_guard lockCb(lockCallbacks_);
callbacks_.emplace_back([this, key, token, opstate, ids, sendTime]() {
if (opstate->stop)
return;
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
auto s = searches_.find(key);
if (s == searches_.end())
return;
@@ -1423,7 +1423,7 @@ DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& tok
{
#ifdef OPENDHT_PUSH_NOTIFICATIONS
{
std::unique_lock<std::mutex> l(lockCurrentProxyInfos_);
std::unique_lock l(lockCurrentProxyInfos_);
if (deviceKey_ == token)
return;
deviceKey_ = token;
@@ -1432,7 +1432,7 @@ DhtProxyClient::setPushNotificationToken([[maybe_unused]] const std::string& tok
}
if (logger_)
logger_->debug("[proxy:client] [push] token changed, resubscribing");
std::lock_guard<std::mutex> lock(searchLock_);
std::lock_guard lock(searchLock_);
for (auto& search : searches_) {
for (auto& listener : search.second.listeners) {
resubscribe(search.first, listener.first, listener.second);
src/dht_proxy_server.cpp
+22 -22
View File
@@ -176,7 +176,7 @@ DhtProxyServer::ConnectionListener::state_changed(const restinio::connection_sta
void
DhtProxyServer::onConnectionClosed(restinio::connection_id_t id)
{
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
auto it = listeners_.find(id);
if (it != listeners_.end()) {
dht_->cancelListen(it->second.hash, std::move(it->second.token));
@@ -381,13 +381,13 @@ DhtProxyServer::saveState(Os& stream)
msgpack::packer<Os> pk(&stream);
pk.pack_map(2);
{
std::lock_guard<std::mutex> lock(lockSearchPuts_);
std::lock_guard lock(lockSearchPuts_);
pk.pack("puts");
pk.pack(puts_);
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
{
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
pk.pack("pushListeners");
pk.pack(pushListeners_);
}
@@ -408,7 +408,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
if (oh.get().type != msgpack::type::MAP)
continue;
if (auto puts = findMapValue(oh.get(), "puts"sv)) {
std::lock_guard<std::mutex> lock(lockSearchPuts_);
std::lock_guard lock(lockSearchPuts_);
puts_ = puts->as<decltype(puts_)>();
if (logger_)
logger_->debug("[proxy:server] Loading {} persistent puts", puts_.size());
@@ -432,7 +432,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
json["to"] = clientId;
json["vid"] = std::to_string(vid);
if (sessionCtx) {
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
}
return json;
@@ -459,7 +459,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
}
#ifdef OPENDHT_PUSH_NOTIFICATIONS
if (auto pushListeners = findMapValue(oh.get(), "pushListeners"sv)) {
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
pushListeners_ = pushListeners->as<decltype(pushListeners_)>();
if (logger_)
logger_->debug("[proxy:server] Loading {} push listeners", pushListeners_.size());
@@ -496,7 +496,7 @@ DhtProxyServer::loadState(Is& is, size_t size)
Json::Value json;
json["timeout"] = infoHash;
json["to"] = clientId;
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
return json;
};
@@ -546,7 +546,7 @@ DhtProxyServer::~DhtProxyServer()
saveState(stateFile);
}
if (dht_) {
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
for (auto& l : listeners_) {
dht_->cancelListen(l.second.hash, std::move(l.second.token));
if (l.second.response)
@@ -817,7 +817,7 @@ DhtProxyServer::listen(restinio::request_handle_t request, restinio::router::rou
auto response = std::make_shared<ResponseByPartsBuilder>(
initHttpResponse(request->create_response<ResponseByParts>()));
response->flush();
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
// save the listener to handle a disconnect
auto& session = listeners_[request->connection_id()];
session.hash = infoHash;
@@ -931,7 +931,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
logger_->debug("[proxy:server] [subscribe {}] [client {}] [session {}]", infoHash, clientId, sessionId);
// Insert new or return existing push listeners of a token
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
auto& pushListener = pushListeners_[pushToken];
auto& pushListeners = pushListener.listeners[infoHash];
@@ -945,7 +945,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
listIt->clientId = clientId;
listIt->sessionCtx = std::make_shared<PushSessionContext>(sessionId);
} else {
std::lock_guard<std::mutex> l(listIt->sessionCtx->lock);
std::lock_guard l(listIt->sessionCtx->lock);
listIt->sessionCtx->sessionId = sessionId;
}
auto& listener = *listIt;
@@ -963,7 +963,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::
Json::Value json;
json["timeout"] = h;
json["to"] = clientId;
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
return json;
};
@@ -1110,7 +1110,7 @@ DhtProxyServer::handleCancelPushListen(const asio::error_code& ec,
}
if (logger_)
logger_->debug("[proxy:server] [listen:push {}] cancelled for {}", key, clientId);
std::lock_guard<std::mutex> lock(lockListener_);
std::lock_guard lock(lockListener_);
auto pushListener = pushListeners_.find(pushToken);
if (pushListener == pushListeners_.end())
@@ -1151,7 +1151,7 @@ DhtProxyServer::handlePushListen(const InfoHash& infoHash,
json["t"] = Json::Value::Int64(duration_cast<milliseconds>(system_clock::now().time_since_epoch()).count());
{
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
}
@@ -1302,12 +1302,12 @@ DhtProxyServer::sendPushNotification(
if (state == http::Request::State::DONE) {
if (logger_ and response.status_code != 200)
logger_->error("[proxy:server] [notification] push failed: {}", response.status_code);
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
});
{
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_[reqid] = request;
}
request->send();
@@ -1330,7 +1330,7 @@ DhtProxyServer::sendPushNotification(
if (logger_)
logger_->error("[proxy:server] [notification] error send push: {}", e.what());
if (reqid) {
std::lock_guard<std::mutex> l(requestLock_);
std::lock_guard l(requestLock_);
requests_.erase(reqid);
}
}
@@ -1349,7 +1349,7 @@ DhtProxyServer::handleCancelPermamentPut(const asio::error_code& ec, const InfoH
}
if (logger_)
logger_->debug("[proxy:server] [put {}] cancel permament put {}", key, vid);
std::lock_guard<std::mutex> lock(lockSearchPuts_);
std::lock_guard lock(lockSearchPuts_);
auto sPuts = puts_.find(key);
if (sPuts == puts_.end())
return;
@@ -1406,7 +1406,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
sessionId = pVal["session_id"].asString();
topic = pVal["topic"].asString();
}
std::lock_guard<std::mutex> lock(lockSearchPuts_);
std::lock_guard lock(lockSearchPuts_);
auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT;
auto& sPuts = puts_[infoHash];
if (value->id == Value::INVALID_ID) {
@@ -1423,7 +1423,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
if (not pp.second.sessionCtx)
pp.second.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
else {
std::lock_guard<std::mutex> l(pp.second.sessionCtx->lock);
std::lock_guard l(pp.second.sessionCtx->lock);
pp.second.sessionCtx->sessionId = sessionId;
}
}
@@ -1458,7 +1458,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
if (not pput.sessionCtx)
pput.sessionCtx = std::make_shared<PushSessionContext>(sessionId);
else {
std::lock_guard<std::mutex> l(pput.sessionCtx->lock);
std::lock_guard l(pput.sessionCtx->lock);
pput.sessionCtx->sessionId = sessionId;
}
}
@@ -1475,7 +1475,7 @@ DhtProxyServer::put(restinio::request_handle_t request, restinio::router::route_
json["timeout"] = infoHash.toString();
json["to"] = clientId;
json["vid"] = std::to_string(vid);
std::lock_guard<std::mutex> l(sessionCtx->lock);
std::lock_guard l(sessionCtx->lock);
json["s"] = sessionCtx->sessionId;
return json;
};
src/dhtrunner.cpp
+73 -73
View File
@@ -76,7 +76,7 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, Config& co
void
DhtRunner::run(const Config& config, Context&& context)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
auto expected = State::Idle;
if (not running.compare_exchange_strong(expected, State::Running)) {
if (context.logger)
@@ -141,7 +141,7 @@ DhtRunner::run(const Config& config, Context&& context)
context.sock->setOnReceive([&](net::PacketList&& pkts) {
net::PacketList ret;
{
std::lock_guard<std::mutex> lck(sock_mtx);
std::lock_guard lck(sock_mtx);
rcv.splice(rcv.end(), std::move(pkts));
size_t dropped = 0;
while (rcv.size() > net::RX_QUEUE_MAX_SIZE) {
@@ -197,19 +197,19 @@ DhtRunner::run(const Config& config, Context&& context)
return;
dht_thread = std::thread([this]() {
while (running != State::Idle) {
std::unique_lock<std::mutex> lk(dht_mtx);
std::unique_lock lk(dht_mtx);
time_point wakeup = loop_();
auto hasJobToDo = [this]() {
if (running == State::Idle)
return true;
{
std::lock_guard<std::mutex> lck(sock_mtx);
std::lock_guard lck(sock_mtx);
if (not rcv.empty())
return true;
}
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (not pending_ops_prio.empty())
return true;
auto s = getStatus();
@@ -284,7 +284,7 @@ DhtRunner::run(const Config& config, Context&& context)
void
DhtRunner::shutdown(ShutdownCallback cb, bool stop)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
auto expected = State::Running;
if (not running.compare_exchange_strong(expected, State::Stopping)) {
if (expected == State::Stopping and ongoing_ops) {
@@ -344,7 +344,7 @@ DhtRunner::checkShutdown()
{
decltype(shutdownCallbacks_) cbs;
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (running != State::Stopping or ongoing_ops)
return false;
cbs = std::move(shutdownCallbacks_);
@@ -359,7 +359,7 @@ void
DhtRunner::join()
{
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (running.exchange(State::Idle) == State::Idle)
return;
cv.notify_all();
@@ -378,7 +378,7 @@ DhtRunner::join()
dht_thread.join();
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (ongoing_ops and logger_) {
logger_->warn("[runner {:p}] stopping with {:d} remaining ops", fmt::ptr(this), ongoing_ops.load());
}
@@ -388,7 +388,7 @@ DhtRunner::join()
shutdownCallbacks_.clear();
}
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
resetDht();
status4 = NodeStatus::Disconnected;
status6 = NodeStatus::Disconnected;
@@ -398,7 +398,7 @@ DhtRunner::join()
SockAddr
DhtRunner::getBound(sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
if (auto sock = dht_->getSocket())
return sock->getBound(af);
@@ -408,7 +408,7 @@ DhtRunner::getBound(sa_family_t af) const
in_port_t
DhtRunner::getBoundPort(sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
if (auto sock = dht_->getSocket())
return sock->getPort(af);
@@ -418,7 +418,7 @@ DhtRunner::getBoundPort(sa_family_t af) const
void
DhtRunner::dumpTables() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
dht_->dumpTables();
}
@@ -443,7 +443,7 @@ DhtRunner::getNodeId() const
std::pair<size_t, size_t>
DhtRunner::getStoreSize() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
return {};
return dht_->getStoreSize();
@@ -452,7 +452,7 @@ DhtRunner::getStoreSize() const
void
DhtRunner::setStorageLimit(size_t limit)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
throw std::runtime_error("dht is not running");
return dht_->setStorageLimit(limit);
@@ -461,7 +461,7 @@ DhtRunner::setStorageLimit(size_t limit)
size_t
DhtRunner::getStorageLimit() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
throw std::runtime_error("dht is not running");
return dht_->getStorageLimit();
@@ -470,7 +470,7 @@ DhtRunner::getStorageLimit() const
void
DhtRunner::setLocalStorageLimit(size_t limit)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
throw std::runtime_error("dht is not running");
return dht_->setLocalStorageLimit(limit);
@@ -479,7 +479,7 @@ DhtRunner::setLocalStorageLimit(size_t limit)
size_t
DhtRunner::getLocalStorageLimit() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
throw std::runtime_error("dht is not running");
return dht_->getLocalStorageLimit();
@@ -488,7 +488,7 @@ DhtRunner::getLocalStorageLimit() const
std::vector<NodeExport>
DhtRunner::exportNodes() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
return {};
return dht_->exportNodes();
@@ -497,7 +497,7 @@ DhtRunner::exportNodes() const
std::vector<ValuesExport>
DhtRunner::exportValues() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (!dht_)
return {};
return dht_->exportValues();
@@ -506,7 +506,7 @@ DhtRunner::exportValues() const
void
DhtRunner::setLogger(const Sp<Logger>& logger)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
logger_ = logger;
if (dht_)
dht_->setLogger(logger);
@@ -515,7 +515,7 @@ DhtRunner::setLogger(const Sp<Logger>& logger)
void
DhtRunner::setLogFilter(const InfoHash& f)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
dht_->setLogFilter(f);
}
@@ -523,21 +523,21 @@ DhtRunner::setLogFilter(const InfoHash& f)
void
DhtRunner::registerType(const ValueType& type)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
dht_->registerType(type);
}
void
DhtRunner::registerInsecureType(const ValueType& type)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
dht_->registerInsecureType(type);
}
void
DhtRunner::importValues(const std::vector<ValuesExport>& values)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
dht_->importValues(values);
}
@@ -548,7 +548,7 @@ DhtRunner::getNodesStats(sa_family_t af,
unsigned* cached_return,
unsigned* incoming_return) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
const auto stats = dht_->getNodesStats(af);
if (good_return)
*good_return = stats.good_nodes;
@@ -564,7 +564,7 @@ DhtRunner::getNodesStats(sa_family_t af,
NodeStats
DhtRunner::getNodesStats(sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getNodesStats(af);
}
@@ -572,7 +572,7 @@ NodeInfo
DhtRunner::getNodeInfo() const
{
NodeInfo info {};
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
info = dht_->getNodeInfo();
info.ongoing_ops = ongoing_ops;
@@ -582,7 +582,7 @@ DhtRunner::getNodeInfo() const
void
DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
ongoing_ops++;
pending_ops_prio.emplace([cb = std::move(cb), this](SecureDht& dht) {
auto sinfo = std::make_shared<NodeInfo>();
@@ -597,44 +597,44 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
std::vector<unsigned>
DhtRunner::getNodeMessageStats(bool in) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getNodeMessageStats(in);
}
std::string
DhtRunner::getStorageLog() const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getStorageLog();
}
std::string
DhtRunner::getStorageLog(const InfoHash& f) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getStorageLog(f);
}
std::string
DhtRunner::getRoutingTablesLog(sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getRoutingTablesLog(af);
}
std::string
DhtRunner::getSearchesLog(sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getSearchesLog(af);
}
std::string
DhtRunner::getSearchLog(const InfoHash& f, sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
return dht_->getSearchLog(f, af);
}
std::vector<SockAddr>
DhtRunner::getPublicAddress(sa_family_t af) const
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
return dht_->getPublicAddress(af);
return {};
@@ -651,7 +651,7 @@ DhtRunner::getPublicAddressStr(sa_family_t af) const
void
DhtRunner::getPublicAddress(std::function<void(std::vector<SockAddr>&&)> cb, sa_family_t af)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
ongoing_ops++;
pending_ops_prio.emplace([cb = std::move(cb), this, af](SecureDht& dht) {
cb(dht.getPublicAddress(af));
@@ -663,14 +663,14 @@ DhtRunner::getPublicAddress(std::function<void(std::vector<SockAddr>&&)> cb, sa_
void
DhtRunner::registerCertificate(const std::shared_ptr<crypto::Certificate>& cert)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
dht_->registerCertificate(cert);
}
void
DhtRunner::setLocalCertificateStore(CertificateStoreQueryLegacy&& query_method)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
dht_->setLocalCertificateStore(std::forward<CertificateStoreQueryLegacy>(query_method));
}
@@ -678,7 +678,7 @@ DhtRunner::setLocalCertificateStore(CertificateStoreQueryLegacy&& query_method)
void
DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
}
@@ -691,7 +691,7 @@ DhtRunner::loop_()
decltype(pending_ops) ops {};
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
auto s = getStatus();
ops = (pending_ops_prio.empty()
&& (s == NodeStatus::Connected or s == NodeStatus::Disconnected or running == State::Stopping))
@@ -707,7 +707,7 @@ DhtRunner::loop_()
decltype(rcv) received {};
decltype(rcv) received_treated {};
{
std::lock_guard<std::mutex> lck(sock_mtx);
std::lock_guard lck(sock_mtx);
// move to stack
received = std::move(rcv);
}
@@ -742,7 +742,7 @@ DhtRunner::loop_()
}
if (not received_treated.empty()) {
std::lock_guard<std::mutex> lck(sock_mtx);
std::lock_guard lck(sock_mtx);
if (rcv_free.size() < net::RX_QUEUE_MAX_SIZE)
rcv_free.splice(rcv_free.end(), std::move(received_treated));
}
@@ -766,7 +766,7 @@ DhtRunner::loop_()
void
DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f, Where w)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (dcb)
@@ -788,7 +788,7 @@ DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb,
void
DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (done_cb)
@@ -806,7 +806,7 @@ std::future<size_t>
DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
{
auto ret_token = std::make_shared<std::promise<size_t>>();
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
ret_token->set_value(0);
@@ -828,7 +828,7 @@ DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f, Wher
void
DhtRunner::cancelListen(InfoHash h, size_t token)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (running != State::Running)
return;
ongoing_ops++;
@@ -842,7 +842,7 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
void
DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (running != State::Running)
return;
ongoing_ops++;
@@ -856,7 +856,7 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
void
DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -873,7 +873,7 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created
void
DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -896,7 +896,7 @@ DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, tim
void
DhtRunner::cancelPut(const InfoHash& h, Value::Id id)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops.emplace([=](SecureDht& dht) { dht.cancelPut(h, id); });
cv.notify_all();
}
@@ -904,7 +904,7 @@ DhtRunner::cancelPut(const InfoHash& h, Value::Id id)
void
DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops.emplace([=](SecureDht& dht) { dht.cancelPut(h, value->id); });
cv.notify_all();
}
@@ -912,7 +912,7 @@ DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value)
void
DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -941,7 +941,7 @@ DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple c
void
DhtRunner::putEncrypted(InfoHash hash, const PkId& to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -958,7 +958,7 @@ DhtRunner::putEncrypted(InfoHash hash, const PkId& to, std::shared_ptr<Value> va
void
DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -991,7 +991,7 @@ DhtRunner::putEncrypted(InfoHash hash,
DoneCallback cb,
bool permanent)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -1015,7 +1015,7 @@ DhtRunner::putEncrypted(
void
DhtRunner::bootstrap(const std::string& host, const std::string& service)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops_prio.emplace([host, service](SecureDht& dht) mutable { dht.addBootstrap(host, service); });
cv.notify_all();
}
@@ -1023,7 +1023,7 @@ DhtRunner::bootstrap(const std::string& host, const std::string& service)
void
DhtRunner::bootstrap(const std::string& hostService)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
auto [h, s] = splitPort(hostService);
pending_ops_prio.emplace(
[host = std::string(h), service = std::string(s)](SecureDht& dht) { dht.addBootstrap(host, service); });
@@ -1033,7 +1033,7 @@ DhtRunner::bootstrap(const std::string& hostService)
void
DhtRunner::clearBootstrap()
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops_prio.emplace([](SecureDht& dht) mutable { dht.clearBootstrap(); });
cv.notify_all();
}
@@ -1045,7 +1045,7 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple cb)
cb(false);
return;
}
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
ongoing_ops++;
pending_ops_prio.emplace([cb = bindOpDoneCallback(std::move(cb)), nodes = std::move(nodes)](SecureDht& dht) mutable {
auto rem = cb ? std::make_shared<std::pair<size_t, bool>>(nodes.size(), false) : nullptr;
@@ -1068,7 +1068,7 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple cb)
void
DhtRunner::bootstrap(SockAddr addr, DoneCallbackSimple cb)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
if (cb)
@@ -1085,7 +1085,7 @@ DhtRunner::bootstrap(SockAddr addr, DoneCallbackSimple cb)
void
DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (running != State::Running)
return;
pending_ops_prio.emplace([id, address](SecureDht& dht) mutable { dht.insertNode(id, address); });
@@ -1095,7 +1095,7 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
void
DhtRunner::bootstrap(std::vector<NodeExport> nodes)
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
if (running != State::Running)
return;
pending_ops_prio.emplace([nodes = std::move(nodes)](SecureDht& dht) {
@@ -1108,7 +1108,7 @@ DhtRunner::bootstrap(std::vector<NodeExport> nodes)
void
DhtRunner::connectivityChanged()
{
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops_prio.emplace([=](SecureDht& dht) {
dht.connectivityChanged();
#ifdef OPENDHT_PEER_DISCOVERY
@@ -1122,7 +1122,7 @@ DhtRunner::connectivityChanged()
void
DhtRunner::findCertificate(InfoHash hash, std::function<void(const Sp<crypto::Certificate>&)> cb)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
cb({});
@@ -1141,7 +1141,7 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const Sp<crypto::Ce
void
DhtRunner::findCertificate(PkId hash, std::function<void(const std::shared_ptr<crypto::Certificate>&)> cb)
{
std::unique_lock<std::mutex> lck(storage_mtx);
std::unique_lock lck(storage_mtx);
if (running != State::Running) {
lck.unlock();
return;
@@ -1167,7 +1167,7 @@ void
DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeId)
{
#ifdef OPENDHT_PROXY_CLIENT
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (config_.proxy_server == proxy and config_.push_node_id == pushNodeId)
return;
config_.proxy_server = proxy;
@@ -1194,7 +1194,7 @@ DhtRunner::enableProxy(bool proxify)
config_.client_identity,
[this] {
if (config_.threaded) {
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops_prio.emplace([=](SecureDht&) mutable {});
cv.notify_all();
}
@@ -1221,7 +1221,7 @@ DhtRunner::enableProxy(bool proxify)
void
DhtRunner::forwardAllMessages(bool forward)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
if (dht_)
dht_->forwardAllMessages(forward);
}
@@ -1232,7 +1232,7 @@ DhtRunner::forwardAllMessages(bool forward)
void
DhtRunner::setPushNotificationToken(const std::string& token)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
config_.push_token = token;
if (dht_)
@@ -1245,7 +1245,7 @@ DhtRunner::setPushNotificationToken(const std::string& token)
void
DhtRunner::setPushNotificationTopic(const std::string& topic)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
config_.push_topic = topic;
if (dht_)
@@ -1258,7 +1258,7 @@ DhtRunner::setPushNotificationTopic(const std::string& topic)
void
DhtRunner::setPushNotificationPlatform(const std::string& platform)
{
std::lock_guard<std::mutex> lck(dht_mtx);
std::lock_guard lck(dht_mtx);
#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
config_.push_platform = platform;
if (dht_)
@@ -1274,7 +1274,7 @@ DhtRunner::pushNotificationReceived([[maybe_unused]] const std::map<std::string,
#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
auto ret_token = std::make_shared<std::promise<PushNotificationResult>>();
auto future = ret_token->get_future();
std::lock_guard<std::mutex> lck(storage_mtx);
std::lock_guard lck(storage_mtx);
pending_ops_prio.emplace([ret_token, this, data](SecureDht&) {
if (dht_)
ret_token->set_value(dht_->pushNotificationReceived(data));
src/http.cpp
+20 -20
View File
@@ -239,7 +239,7 @@ Connection::~Connection()
void
Connection::close()
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
asio::error_code ec;
if (ssl_socket_) {
if (ssl_socket_->is_open())
@@ -490,7 +490,7 @@ ocspValidateResponse(const OscpRequestInfo& info,
void
Connection::set_ssl_verification(const std::string& hostname, const asio::ssl::verify_mode verify_mode)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (ssl_socket_) {
// Set SNI Hostname (many hosts need this to handshake successfully)
SSL_set_tlsext_host_name(ssl_socket_->asio_ssl_stream().native_handle(), hostname.c_str());
@@ -576,7 +576,7 @@ Connection::input()
std::string
Connection::read_bytes(size_t bytes)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (bytes == 0)
bytes = read_buf_.in_avail();
std::string content;
@@ -597,7 +597,7 @@ Connection::read_until(const char delim)
void
Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!ssl_socket_ && !socket_) {
cb(asio::error::operation_aborted, {});
return;
@@ -629,7 +629,7 @@ Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, Conn
void
Connection::async_handshake(HandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (ssl_socket_) {
std::weak_ptr<Connection> wthis = shared_from_this();
ssl_socket_
@@ -663,7 +663,7 @@ Connection::async_handshake(HandlerCb cb)
void
Connection::async_write(BytesHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!is_open()) {
if (cb)
asio::post(ctx_, [cb]() { cb(asio::error::broken_pipe, 0); });
@@ -680,7 +680,7 @@ Connection::async_write(BytesHandlerCb cb)
void
Connection::async_read_until(const char* delim, BytesHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!is_open()) {
if (cb)
asio::post(ctx_, [cb]() { cb(asio::error::broken_pipe, 0); });
@@ -697,7 +697,7 @@ Connection::async_read_until(const char* delim, BytesHandlerCb cb)
void
Connection::async_read_until(char delim, BytesHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!is_open()) {
if (cb)
asio::post(ctx_, [cb]() { cb(asio::error::broken_pipe, 0); });
@@ -714,7 +714,7 @@ Connection::async_read_until(char delim, BytesHandlerCb cb)
void
Connection::async_read(size_t bytes, BytesHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!is_open()) {
if (cb)
asio::post(ctx_, [cb]() { cb(asio::error::broken_pipe, 0); });
@@ -731,7 +731,7 @@ Connection::async_read(size_t bytes, BytesHandlerCb cb)
void
Connection::async_read_some(size_t bytes, BytesHandlerCb cb)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!is_open()) {
if (cb)
asio::post(ctx_, [cb]() { cb(asio::error::broken_pipe, 0); });
@@ -751,7 +751,7 @@ Connection::async_read_some(size_t bytes, BytesHandlerCb cb)
void
Connection::set_keepalive(uint32_t seconds)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!ssl_socket_ && !socket_)
return;
@@ -871,7 +871,7 @@ Resolver::~Resolver()
{
decltype(cbs_) cbs;
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
cbs = std::move(cbs_);
}
while (not cbs.empty()) {
@@ -907,7 +907,7 @@ filter(const std::vector<asio::ip::tcp::endpoint>& epts, sa_family_t family)
void
Resolver::add_callback(ResolverCb cb, sa_family_t family)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (!completed_)
cbs_.emplace(family == AF_UNSPEC ? std::move(cb)
: [cb, family](const asio::error_code& ec,
@@ -946,7 +946,7 @@ Resolver::resolve(std::string_view host, std::string_view service)
}
decltype(cbs_) cbs;
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
ec_ = ec;
endpoints_ = std::vector<asio::ip::tcp::endpoint> {endpoints.begin(),
endpoints.end()};
@@ -1390,7 +1390,7 @@ Request::connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, HandlerCb cb)
if (not sthis)
return;
auto& this_ = *sthis;
std::lock_guard<std::mutex> lock(this_.mutex_);
std::lock_guard lock(this_.mutex_);
if (ec == asio::error::operation_aborted) {
this_.terminate(ec);
return;
@@ -1443,7 +1443,7 @@ Request::send()
[wthis](const asio::error_code& ec, std::vector<asio::ip::tcp::endpoint> endpoints) {
if (auto sthis = wthis.lock()) {
auto& this_ = *sthis;
std::lock_guard<std::mutex> lock(this_.mutex_);
std::lock_guard lock(this_.mutex_);
if (ec) {
if (this_.logger_)
this_.logger_->error("[http:request:{:d}] resolve error: {:s}", this_.id_, ec.message());
@@ -1518,7 +1518,7 @@ Request::terminate(const asio::error_code& ec)
void
Request::handle_request(const asio::error_code& ec)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (ec and ec != asio::error::eof) {
terminate(ec);
return;
@@ -1542,7 +1542,7 @@ Request::handle_request(const asio::error_code& ec)
void
Request::handle_response(const asio::error_code& ec, size_t /* n_bytes */)
{
std::lock_guard<std::mutex> lock(mutex_);
std::lock_guard lock(mutex_);
if (ec && ec != asio::error::eof) {
terminate(ec);
return;
@@ -1660,11 +1660,11 @@ const Response&
Request::await()
{
std::mutex mtx;
std::unique_lock<std::mutex> lock(mtx);
std::unique_lock lock(mtx);
std::condition_variable cv;
bool ok {false};
add_on_done_callback([&](const Response&) {
std::lock_guard<std::mutex> lk(mtx);
std::lock_guard lk(mtx);
ok = true;
cv.notify_all();
});
src/network_utils.cpp
+5 -5
View File
@@ -107,14 +107,14 @@ UdpSocket::UdpSocket(in_port_t port, const std::shared_ptr<Logger>& l)
SockAddr bind6;
bind6.setFamily(AF_INET6);
bind6.setPort(port);
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
openSockets(bind4, bind6);
}
UdpSocket::UdpSocket(const SockAddr& bind4, const SockAddr& bind6, const std::shared_ptr<Logger>& l)
: logger(l)
{
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
openSockets(bind4, bind6);
}
@@ -161,7 +161,7 @@ UdpSocket::sendTo(const SockAddr& dest, const uint8_t* data, size_t size, [[mayb
if (logger)
logger->d("Can't send message to %s: %s", dest.toString().c_str(), strerror(err));
if (err == EPIPE || err == ENOTCONN || err == ECONNRESET) {
std::lock_guard<std::mutex> lk(lock);
std::lock_guard lk(lock);
auto bind4 = std::move(bound4), bind6 = std::move(bound6);
openSockets(bind4, bind6);
return sendTo(dest, data, size, false);
@@ -292,7 +292,7 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
if (err == EPIPE || err == ENOTCONN || err == ECONNRESET) {
if (not running)
break;
std::unique_lock<std::mutex> lk(lock, std::try_to_lock);
std::unique_lock lk(lock, std::try_to_lock);
if (lk.owns_lock()) {
if (not running)
break;
@@ -337,7 +337,7 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
close(stop_readfd);
if (stopfd != -1)
close(stopfd);
std::unique_lock<std::mutex> lk(lock, std::try_to_lock);
std::unique_lock lk(lock, std::try_to_lock);
if (lk.owns_lock()) {
s4 = -1;
s6 = -1;
src/peer_discovery.cpp
+10 -10
View File
@@ -123,7 +123,7 @@ PeerDiscovery::DomainPeerDiscovery::~DomainPeerDiscovery()
void
PeerDiscovery::DomainPeerDiscovery::startDiscovery(std::string_view type, ServiceDiscoveredCallback callback)
{
std::lock_guard<std::mutex> lck(dmtx_);
std::lock_guard lck(dmtx_);
callbackmap_[std::string(type)] = callback;
if (not drunning_) {
drunning_ = true;
@@ -137,7 +137,7 @@ PeerDiscovery::DomainPeerDiscovery::startDiscovery(std::string_view type, Servic
void
PeerDiscovery::DomainPeerDiscovery::loopListener()
{
std::lock_guard<std::mutex> lck(dmtx_);
std::lock_guard lck(dmtx_);
if (not drunning_)
return;
sockFd_.async_receive_from(asio::buffer(receiveBuf_),
@@ -163,7 +163,7 @@ PeerDiscovery::DomainPeerDiscovery::loopListener()
continue;
ServiceDiscoveredCallback cb;
{
std::lock_guard<std::mutex> lck(dmtx_);
std::lock_guard lck(dmtx_);
if (drunning_) {
auto callback = callbackmap_.find(o.key.as<std::string_view>());
if (callback != callbackmap_.end())
@@ -189,7 +189,7 @@ PeerDiscovery::DomainPeerDiscovery::loopListener()
void
PeerDiscovery::DomainPeerDiscovery::query(const asio::ip::udp::endpoint& peer)
{
std::lock_guard<std::mutex> lck(dmtx_);
std::lock_guard lck(dmtx_);
if (not drunning_)
return;
@@ -209,7 +209,7 @@ PeerDiscovery::DomainPeerDiscovery::query(const asio::ip::udp::endpoint& peer)
void
PeerDiscovery::DomainPeerDiscovery::publish(const asio::ip::udp::endpoint& peer)
{
std::lock_guard<std::mutex> lck(mtx_);
std::lock_guard lck(mtx_);
if (not lrunning_)
return;
@@ -229,7 +229,7 @@ PeerDiscovery::DomainPeerDiscovery::startPublish(std::string_view type, const ms
msgpack::sbuffer pack_buf_c(pack_buf.size());
pack_buf_c.write(pack_buf.data(), pack_buf.size());
std::lock_guard<std::mutex> lck(mtx_);
std::lock_guard lck(mtx_);
messages_[std::string(type)] = std::move(pack_buf_c);
reloadMessages();
lrunning_ = true;
@@ -239,7 +239,7 @@ PeerDiscovery::DomainPeerDiscovery::startPublish(std::string_view type, const ms
bool
PeerDiscovery::DomainPeerDiscovery::stopDiscovery(std::string_view type)
{
std::lock_guard<std::mutex> lck(dmtx_);
std::lock_guard lck(dmtx_);
auto it = callbackmap_.find(type);
if (it != callbackmap_.end()) {
callbackmap_.erase(it);
@@ -253,7 +253,7 @@ PeerDiscovery::DomainPeerDiscovery::stopDiscovery(std::string_view type)
bool
PeerDiscovery::DomainPeerDiscovery::stopPublish(std::string_view type)
{
std::lock_guard<std::mutex> lck(mtx_);
std::lock_guard lck(mtx_);
auto it = messages_.find(type);
if (it != messages_.end()) {
messages_.erase(it);
@@ -282,11 +282,11 @@ void
PeerDiscovery::DomainPeerDiscovery::stop()
{
{
std::lock_guard<std::mutex> lck(dmtx_);
std::lock_guard lck(dmtx_);
stopDiscovery();
}
{
std::lock_guard<std::mutex> lck(mtx_);
std::lock_guard lck(mtx_);
stopPublish();
}
}
tests/test_dhtproxy.cpp
+15 -15
View File
@@ -57,11 +57,11 @@ DhtProxyTester::tearDown()
std::condition_variable cv;
std::mutex cv_m;
nodeProxy->shutdown([&] {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
});
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&] { return done; }));
serverProxy.reset();
nodeProxy.reset();
@@ -81,11 +81,11 @@ DhtProxyTester::testGetPut()
auto val_data = val.data;
{
nodePeer.put(key, std::move(val), [&](bool) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
});
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; }));
}
@@ -101,7 +101,7 @@ DhtProxyTester::testListen()
std::condition_variable cv;
std::mutex cv_m;
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
auto key = dht::InfoHash::get("GLaDOs");
bool done = false;
@@ -111,7 +111,7 @@ DhtProxyTester::testListen()
auto firstVal_data = firstVal.data;
nodePeer.put(key, std::move(firstVal), [&](bool ok) {
CPPUNIT_ASSERT(ok);
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
});
@@ -121,7 +121,7 @@ DhtProxyTester::testListen()
std::vector<dht::Blob> values;
auto token = nodeClient.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
if (not expired) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
for (const auto& value : v)
values.emplace_back(value->data);
done = true;
@@ -156,7 +156,7 @@ DhtProxyTester::testResubscribeGetValues()
bool done = false;
std::condition_variable cv;
std::mutex cv_m;
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
auto key = dht::InfoHash::get("GLaDOs");
// If a peer sent a value, the listen operation from the client
@@ -164,7 +164,7 @@ DhtProxyTester::testResubscribeGetValues()
dht::Value firstVal {"Hey! It's been a long time. How have you been?"};
auto firstVal_data = firstVal.data;
nodePeer.put(key, std::move(firstVal), [&](bool ok) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
CPPUNIT_ASSERT(ok);
done = true;
cv.notify_all();
@@ -186,7 +186,7 @@ DhtProxyTester::testResubscribeGetValues()
std::vector<std::shared_ptr<dht::Value>> values;
auto ftoken = nodeClient.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
if (not expired) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
values.insert(values.end(), v.begin(), v.end());
done = true;
cv.notify_all();
@@ -218,14 +218,14 @@ DhtProxyTester::testPutGet40KChars()
mtu.emplace_back((i % 2) ? 'T' : 'M');
std::condition_variable cv;
std::mutex cv_m;
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
bool done_put = false;
bool done_get = false;
// Act
dht::Value val {mtu};
nodePeer.put(key, std::move(val), [&](bool ok) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done_put = ok;
cv.notify_all();
});
@@ -238,7 +238,7 @@ DhtProxyTester::testPutGet40KChars()
return true;
},
[&](bool ok) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done_get = ok;
cv.notify_all();
});
@@ -316,12 +316,12 @@ DhtProxyTester::testShutdownStop()
std::mutex cv_m;
nodeTest->shutdown(
[&] {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
},
true);
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&] { return done; }));
}
CPPUNIT_ASSERT_EQUAL(2 * C, callback_count.load());
tests/test_dhtproxy_stress.cpp
+7 -7
View File
@@ -51,11 +51,11 @@ DhtProxyStressTester::tearDown()
std::condition_variable cv;
std::mutex cv_m;
nodeProxy->shutdown([&] {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
});
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&] { return done; }));
serverProxy.reset();
nodeProxy.reset();
@@ -70,7 +70,7 @@ DhtProxyStressTester::testRepeatValues()
std::condition_variable cv;
std::mutex cv_m;
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
auto key = dht::InfoHash::get("GLaDOs");
bool done = false;
@@ -80,7 +80,7 @@ DhtProxyStressTester::testRepeatValues()
auto firstVal_data = firstVal.data;
nodePeer.put(key, std::move(firstVal), [&](bool ok) {
CPPUNIT_ASSERT(ok);
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
});
@@ -90,7 +90,7 @@ DhtProxyStressTester::testRepeatValues()
std::vector<dht::Blob> values;
auto token = nodeClient.listen(key, [&](const std::vector<std::shared_ptr<dht::Value>>& v, bool expired) {
if (not expired) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
for (const auto& value : v)
values.emplace_back(value->data);
done = true;
@@ -132,7 +132,7 @@ DhtProxyStressTester::testRepeatValues()
val->id = id++;
nodePeer.put(key, val, [&](bool ok) {
CPPUNIT_ASSERT(ok);
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
});
@@ -145,7 +145,7 @@ DhtProxyStressTester::testRepeatValues()
if (not expired) {
for (const auto& value : v) {
if (value->id == val->id) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
done = true;
cv.notify_all();
break;
tests/test_http.cpp
+5 -5
View File
@@ -338,7 +338,7 @@ HttpTester::test_send_json()
// Arrange
std::condition_variable cv;
std::mutex cv_m;
std::unique_lock<std::mutex> lk(cv_m);
std::unique_lock lk(cv_m);
bool done = false;
unsigned int status = 0;
@@ -350,7 +350,7 @@ HttpTester::test_send_json()
"http://127.0.0.1:8080/key/key",
json,
[&](Json::Value value, const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
resp_val = std::move(value);
status = response.status_code;
done = true;
@@ -368,7 +368,7 @@ HttpTester::test_send_json()
request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"http://google.ca",
[&](const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
status = response.status_code;
done = true;
cv.notify_all();
@@ -382,7 +382,7 @@ HttpTester::test_send_json()
request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"https://google.ca",
[&](const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
status = response.status_code;
done = true;
cv.notify_all();
@@ -396,7 +396,7 @@ HttpTester::test_send_json()
request = std::make_shared<dht::http::Request>(serverProxy->io_context(),
"https://google.ca/sdbjklwGBIP",
[&](const dht::http::Response& response) {
std::lock_guard<std::mutex> lk(cv_m);
std::lock_guard lk(cv_m);
status = response.status_code;
done = true;
cv.notify_all();
tests/test_peerdiscovery.cpp
+3 -3
View File
@@ -54,7 +54,7 @@ PeerDiscoveryTester::testMulticastToTwoNodes()
unsigned countDht {0};
unsigned countJami {0};
{
std::unique_lock<std::mutex> l(lock);
std::unique_lock l(lock);
dht::PeerDiscovery testDht(MULTICAST_PORT);
dht::PeerDiscovery testJami(MULTICAST_PORT);
@@ -63,7 +63,7 @@ PeerDiscoveryTester::testMulticastToTwoNodes()
CPPUNIT_ASSERT_EQUAL(dhtNode.nodeid, v.nodeid);
CPPUNIT_ASSERT_EQUAL(dhtNode.nid, v.nid);
{
std::lock_guard<std::mutex> l(lock);
std::lock_guard l(lock);
countDht++;
}
cv.notify_all();
@@ -75,7 +75,7 @@ PeerDiscoveryTester::testMulticastToTwoNodes()
CPPUNIT_ASSERT_EQUAL(jamiNode.cha, v.cha);
CPPUNIT_ASSERT_EQUAL(jamiNode.str, v.str);
{
std::lock_guard<std::mutex> l(lock);
std::lock_guard l(lock);
countJami++;
}
cv.notify_all();
tools/dhtnode.cpp
+1 -1
View File
@@ -649,7 +649,7 @@ main(int argc, char** argv)
});
// wait for shutdown
std::unique_lock<std::mutex> lk(m);
std::unique_lock lk(m);
cv.wait(lk, [&]() { return done; });
node->join();
tools/dhtscanner.cpp
+1 -1
View File
@@ -126,7 +126,7 @@ main(int argc, char** argv)
{
std::mutex m;
std::unique_lock<std::mutex> lk(m);
std::unique_lock lk(m);
cv.wait(lk, [&]() { return done.load() == 0; });
}
tools/perftest.cpp
+2 -2
View File
@@ -68,7 +68,7 @@ benchPingPong(unsigned netSize, unsigned n_parallel)
unsigned max = PINGPONG_MAX * n_parallel;
auto ping = [&](DhtRunner& node, InfoHash h) {
std::lock_guard<std::mutex> lk(m);
std::lock_guard lk(m);
if (i < max) {
i++;
node.put(h, Value("hey"));
@@ -93,7 +93,7 @@ benchPingPong(unsigned netSize, unsigned n_parallel)
ping(pong_node, locs[i].first);
{
std::unique_lock<std::mutex> lk(m);
std::unique_lock lk(m);
if (not cv.wait_for(lk, std::chrono::minutes(1), [&]() { return i == max; })) {
throw std::runtime_error(std::string("Timeout: ") + std::to_string(i));
}
tools/tools_common.h
+1 -1
View File
@@ -421,7 +421,7 @@ struct ServiceRunner
{
bool wait()
{
std::unique_lock<std::mutex> lock(m);
std::unique_lock lock(m);
cv.wait(lock, [&] { return terminate.load(); });
return !terminate;
}