Skip to content
Snippets Groups Projects
Commit cdd1f24e authored by Sébastien Blin's avatar Sébastien Blin Committed by Adrien Béraud
Browse files

connectionManager: simplify states and mutex usage 

All structures are now stored in a map<<InfoHash, Vid>, ConnectionInfo> instead
of 3 maps to manipulate states. This avoids a lot of complexity and potential
crashes

Change-Id: I003c4bd9a94b6c160911e2dcc8c4a15835a749c9
parent f175ed5f
Branches
Tags
No related merge requests found
Show whitespace changes
Inline Side-by-side
src/jamidht/connectionmanager.cpp
+ 204
− 249
View file @ cdd1f24e
......@@ -28,6 +28,7 @@
#include <opendht/thread_pool.h>
#include <opendht/value.h>
#include <algorithm>
#include <mutex>
#include <map>
#include <condition_variable>
......@@ -48,11 +49,16 @@ struct ConnectionInfo
PeerConnectionRequest response_ {};
std::mutex mutex_ {};
std::unique_ptr<IceTransport> ice_ {nullptr};
// Used to store currently non ready TLS Socket
std::unique_ptr<TlsSocketEndpoint> tls_ {nullptr};
std::shared_ptr<MultiplexedSocket> socket_ {};
};
class ConnectionManager::Impl : public std::enable_shared_from_this<ConnectionManager::Impl>
{
public:
using ConnectionKey = std::pair<dht::InfoHash /* device id */, dht::Value::Id /* uid */>;
explicit Impl(JamiAccount& account)
: account {account}
{}
......@@ -60,39 +66,26 @@ public:
void removeUnusedConnections(const std::string& deviceId = "")
{
{
std::lock_guard<std::mutex> lk(nonReadySocketsMutex_);
for (auto& listSocks : nonReadySockets_) {
if (!deviceId.empty() && listSocks.first != deviceId)
continue;
for (auto& tlsSock : listSocks.second) {
if (tlsSock.second)
tlsSock.second->shutdown();
}
}
if (deviceId.empty()) {
dht::ThreadPool::io().run([nrs = std::make_shared<decltype(nonReadySockets_)>(
std::move(nonReadySockets_))] { nrs->clear(); });
} else {
nonReadySockets_.erase(deviceId);
}
}
{
std::lock_guard<std::mutex> lk(msocketsMutex_);
for (auto& listSocks : multiplexedSockets_) {
if (!deviceId.empty() && listSocks.first != deviceId)
continue;
for (auto& mxSock : listSocks.second) {
if (mxSock.second)
mxSock.second->shutdown();
}
std::lock_guard<std::mutex> lk(infosMtx_);
for (auto it = infos_.begin(); it != infos_.end();) {
auto& [key, info] = *it;
bool erased = false;
if (info) {
if (info->tls_)
info->tls_->shutdown();
if (info->socket_)
info->socket_->shutdown();
if (!deviceId.empty() && key.first == dht::InfoHash(deviceId)) {
erased = true;
it = infos_.erase(it);
}
}
if (!erased)
++it;
}
if (deviceId.empty()) {
dht::ThreadPool::io().run([ms = std::make_shared<decltype(multiplexedSockets_)>(
std::move(multiplexedSockets_))] { ms->clear(); });
} else {
multiplexedSockets_.erase(deviceId);
}
dht::ThreadPool::io().run([infos = std::make_shared<decltype(infos_)>(
std::move(infos_))] { infos->clear(); });
}
}
void shutdown()
......@@ -104,21 +97,24 @@ public:
std::lock_guard<std::mutex> lk(connectCbsMtx_);
pendingCbs_.clear();
}
for (auto& connection : connectionsInfos_) {
for (auto& info : connection.second) {
if (info.second.ice_) {
info.second.ice_->cancelOperations();
info.second.ice_->stop();
}
info.second.responseCv_.notify_all();
{
std::lock_guard<std::mutex> lk(infosMtx_);
for (auto& [key, connInfo] : infos_) {
if (!connInfo)
continue;
if (connInfo->ice_) {
connInfo->ice_->cancelOperations();
connInfo->ice_->stop();
}
connInfo->responseCv_.notify_all();
}
// This is called when the account is only disabled.
// Move this on the thread pool because each
// IceTransport takes 500ms to delete, and it's sequential
// So, it can increase quickly the time to unregister an account
dht::ThreadPool::io().run([co = std::make_shared<decltype(connectionsInfos_)>(
std::move(connectionsInfos_))] { co->clear(); });
dht::ThreadPool::io().run(
[co = std::make_shared<decltype(infos_)>(std::move(infos_))] { co->clear(); });
}
removeUnusedConnections();
}
......@@ -148,9 +144,7 @@ public:
void onDhtPeerRequest(const PeerConnectionRequest& req,
const std::shared_ptr<dht::crypto::Certificate>& cert);
void addNewMultiplexedSocket(const std::string& deviceId,
const dht::Value::Id& vid,
std::unique_ptr<TlsSocketEndpoint>&& tlsSocket);
void addNewMultiplexedSocket(const std::string& deviceId, const dht::Value::Id& vid);
void onPeerResponse(const PeerConnectionRequest& req);
void onDhtConnected(const std::string& deviceId);
......@@ -164,20 +158,19 @@ public:
JamiAccount& account;
std::mutex infosMtx_ {};
// Note: Someone can ask multiple sockets, so to avoid any race condition,
// each device can have multiple multiplexed sockets.
std::map<std::string /* device id */, std::map<dht::Value::Id /* uid */, ConnectionInfo>>
connectionsInfos_ {};
// Used to store currently non ready TLS Socket
std::mutex nonReadySocketsMutex_ {};
std::map<std::string /* device id */,
std::map<dht::Value::Id /* uid */, std::unique_ptr<TlsSocketEndpoint>>>
nonReadySockets_ {};
std::mutex msocketsMutex_ {};
// Note: Multiplexed sockets is also stored in ChannelSockets, so has to be shared_ptr
std::map<std::string /* device id */,
std::map<dht::Value::Id /* uid */, std::shared_ptr<MultiplexedSocket>>>
multiplexedSockets_ {};
std::map<ConnectionKey, std::shared_ptr<ConnectionInfo>> infos_ {};
std::shared_ptr<ConnectionInfo> getInfo(const dht::InfoHash& deviceId, const dht::Value::Id& id)
{
std::lock_guard<std::mutex> lk(infosMtx_);
auto it = infos_.find({deviceId, id});
if (it == infos_.end())
return {};
return it->second;
}
// key: Stored certificate PublicKey id (normaly it's the DeviceId)
// value: pair of shared_ptr<Certificate> and associated RingId
......@@ -216,16 +209,14 @@ void
ConnectionManager::Impl::connectDeviceStartIce(const std::string& deviceId,
const dht::Value::Id& vid)
{
auto tit = connectionsInfos_.find(deviceId);
if (tit == connectionsInfos_.end())
return;
auto it = tit->second.find(vid);
if (it == tit->second.end())
auto info = getInfo(dht::InfoHash(deviceId), vid);
if (!info) {
return;
}
std::pair<std::string, dht::Value::Id> cbId(deviceId, vid);
std::unique_lock<std::mutex> lk {it->second.mutex_};
auto& ice = it->second.ice_;
std::unique_lock<std::mutex> lk(info->mutex_);
auto& ice = info->ice_;
auto onError = [&]() {
ice.reset();
......@@ -271,16 +262,16 @@ ConnectionManager::Impl::connectDeviceStartIce(const std::string& deviceId,
// Wait for call to onResponse() operated by DHT
if (isDestroying_)
return; // This avoid to wait new negotiation when destroying
it->second.responseCv_.wait_for(lk, DHT_MSG_TIMEOUT);
info->responseCv_.wait_for(lk, DHT_MSG_TIMEOUT);
if (isDestroying_)
return; // The destructor can wake a pending wait here.
if (!it->second.responseReceived_) {
if (!info->responseReceived_) {
JAMI_ERR("no response from DHT to E2E request.");
onError();
return;
}
auto& response = it->second.response_;
auto& response = info->response_;
if (!ice)
return;
auto sdp = IceTransport::parse_SDP(response.ice_msg, *ice);
......@@ -300,16 +291,13 @@ ConnectionManager::Impl::connectDeviceOnNegoDone(
const dht::Value::Id& vid,
const std::shared_ptr<dht::crypto::Certificate>& cert)
{
auto tit = connectionsInfos_.find(deviceId);
if (tit == connectionsInfos_.end())
return;
auto it = tit->second.find(vid);
if (it == tit->second.end())
auto info = getInfo(dht::InfoHash(deviceId), vid);
if (!info)
return;
std::pair<std::string, dht::Value::Id> cbId(deviceId, vid);
std::unique_lock<std::mutex> lk {it->second.mutex_};
auto& ice = it->second.ice_;
std::unique_lock<std::mutex> lk {info->mutex_};
auto& ice = info->ice_;
auto onError = [&]() {
std::lock_guard<std::mutex> lk(connectCbsMtx_);
......@@ -328,29 +316,25 @@ ConnectionManager::Impl::connectDeviceOnNegoDone(
}
// Build socket
std::lock_guard<std::mutex> lknrs(nonReadySocketsMutex_);
auto endpoint = std::make_unique<IceSocketEndpoint>(std::shared_ptr<IceTransport>(
std::move(ice)),
true);
// Negotiate a TLS session
JAMI_DBG() << account << "Start TLS session";
auto tlsSocket = std::make_unique<TlsSocketEndpoint>(std::move(endpoint),
info->tls_ = std::make_unique<TlsSocketEndpoint>(std::move(endpoint),
account.identity(),
account.dhParams(),
*cert);
auto& nonReadyIt = nonReadySockets_[deviceId][vid];
nonReadyIt = std::move(tlsSocket);
nonReadyIt->setOnReady([w = weak(),
deviceId = std::move(deviceId),
vid = std::move(vid),
name = std::move(name)](bool ok) {
info->tls_->setOnReady(
[w = weak(), deviceId = std::move(deviceId), vid = std::move(vid), name = std::move(name)](
bool ok) {
auto sthis = w.lock();
if (!sthis)
return;
auto mSockIt = sthis->multiplexedSockets_[deviceId];
if (mSockIt.find(vid) != mSockIt.end())
auto info = sthis->getInfo(dht::InfoHash(deviceId), vid);
if (!info)
return;
if (!ok) {
JAMI_ERR() << "TLS connection failure for peer " << deviceId;
......@@ -363,21 +347,11 @@ ConnectionManager::Impl::connectDeviceOnNegoDone(
sthis->pendingCbs_.erase(cbIt);
}
} else {
// The socket is ready, store it in multiplexedSockets_
std::lock_guard<std::mutex> lkmSockets(sthis->msocketsMutex_);
std::lock_guard<std::mutex> lknrs(sthis->nonReadySocketsMutex_);
auto nonReadyIt = sthis->nonReadySockets_.find(deviceId);
if (nonReadyIt != sthis->nonReadySockets_.end()) {
sthis->addNewMultiplexedSocket(deviceId, vid, std::move(nonReadyIt->second[vid]));
nonReadyIt->second.erase(vid);
if (nonReadyIt->second.empty()) {
sthis->nonReadySockets_.erase(nonReadyIt);
}
}
// The socket is ready, store it
sthis->addNewMultiplexedSocket(deviceId, vid);
// Finally, open the channel
auto mxSockIt = sthis->multiplexedSockets_.at(deviceId);
if (!mxSockIt.empty())
sthis->sendChannelRequest(mxSockIt.rbegin()->second, name, deviceId, vid);
if (info->socket_)
sthis->sendChannelRequest(info->socket_, name, deviceId, vid);
}
});
}
......@@ -425,16 +399,25 @@ ConnectionManager::Impl::connectDevice(const std::string& deviceId,
}
}
std::shared_ptr<MultiplexedSocket> sock;
{
// Test if a socket already exists for this device
std::lock_guard<std::mutex> lk(sthis->msocketsMutex_);
auto it = sthis->multiplexedSockets_.find(deviceId);
if (it != sthis->multiplexedSockets_.end() && !it->second.empty()) {
std::lock_guard<std::mutex> lk(sthis->infosMtx_);
auto it = std::find_if(sthis->infos_.begin(),
sthis->infos_.end(),
[deviceId](const auto& item) {
auto& [key, value] = item;
return key.first == dht::InfoHash(deviceId);
});
if (it != sthis->infos_.end() && it->second) {
sock = it->second->socket_;
}
}
if (sock) {
JAMI_DBG("Peer already connected. Add a new channel");
sthis->sendChannelRequest(it->second.rbegin()->second, name, deviceId, vid);
sthis->sendChannelRequest(sock, name, deviceId, vid);
return;
}
}
// If no socket exists, we need to initiate an ICE connection.
auto& iceTransportFactory = Manager::instance().getIceTransportFactory();
auto ice_config = sthis->account.getIceOptions();
......@@ -501,15 +484,19 @@ ConnectionManager::Impl::connectDevice(const std::string& deviceId,
});
};
auto& connectionInfo = sthis->connectionsInfos_[deviceId][vid];
std::unique_lock<std::mutex> lk {connectionInfo.mutex_};
connectionInfo.ice_ = iceTransportFactory
auto info = std::make_shared<ConnectionInfo>();
{
std::lock_guard<std::mutex> lk(sthis->infosMtx_);
sthis->infos_[{dht::InfoHash(deviceId), vid}] = info;
}
std::unique_lock<std::mutex> lk {info->mutex_};
info->ice_ = iceTransportFactory
.createUTransport(sthis->account.getAccountID().c_str(),
1,
false,
ice_config);
if (!connectionInfo.ice_) {
if (!info->ice_) {
JAMI_ERR("Cannot initialize ICE session.");
std::lock_guard<std::mutex> lk(sthis->connectCbsMtx_);
auto cbIt = sthis->pendingCbs_.find(cbId);
......@@ -567,16 +554,14 @@ ConnectionManager::Impl::onPeerResponse(const PeerConnectionRequest& req)
{
auto device = req.from;
JAMI_INFO() << account << " New response received from " << device.toString().c_str();
auto& infos = connectionsInfos_[device.toString().c_str()];
auto it = infos.find(req.id);
if (it == infos.end()) {
auto info = getInfo(device, req.id);
if (!info) {
JAMI_WARN() << account << " respond received, but cannot find request";
return;
}
auto& connectionInfo = it->second;
connectionInfo.responseReceived_ = true;
connectionInfo.response_ = std::move(req);
connectionInfo.responseCv_.notify_one();
info->responseReceived_ = true;
info->response_ = std::move(req);
info->responseCv_.notify_one();
}
void
......@@ -652,15 +637,12 @@ ConnectionManager::Impl::answerTo(IceTransport& ice,
void
ConnectionManager::Impl::onRequestStartIce(const PeerConnectionRequest& req)
{
auto tit = connectionsInfos_.find(req.from.toString());
if (tit == connectionsInfos_.end())
return;
auto it = tit->second.find(req.id);
if (it == tit->second.end())
auto info = getInfo(req.from, req.id);
if (!info)
return;
std::unique_lock<std::mutex> lk {it->second.mutex_};
auto& ice = it->second.ice_;
std::unique_lock<std::mutex> lk {info->mutex_};
auto& ice = info->ice_;
if (!ice) {
JAMI_ERR("No ICE detected");
if (connReadyCb_)
......@@ -687,20 +669,16 @@ ConnectionManager::Impl::onRequestStartIce(const PeerConnectionRequest& req)
void
ConnectionManager::Impl::onRequestOnNegoDone(const PeerConnectionRequest& req)
{
auto deviceId = req.from.toString();
auto tit = connectionsInfos_.find(deviceId);
if (tit == connectionsInfos_.end())
return;
auto it = tit->second.find(req.id);
if (it == tit->second.end())
auto info = getInfo(req.from, req.id);
if (!info)
return;
std::unique_lock<std::mutex> lk {it->second.mutex_};
auto& ice = it->second.ice_;
std::unique_lock<std::mutex> lk {info->mutex_};
auto& ice = info->ice_;
if (!ice) {
JAMI_ERR("No ICE detected");
if (connReadyCb_)
connReadyCb_(deviceId, "", nullptr);
connReadyCb_(req.from.toString(), "", nullptr);
return;
}
......@@ -710,14 +688,13 @@ ConnectionManager::Impl::onRequestOnNegoDone(const PeerConnectionRequest& req)
answerTo(*ice, req.id, req.from);
// Build socket
std::lock_guard<std::mutex> lknrs(nonReadySocketsMutex_);
auto endpoint = std::make_unique<IceSocketEndpoint>(std::shared_ptr<IceTransport>(
std::move(ice)),
false);
// init TLS session
auto ph = req.from;
auto tlsSocket = std::make_unique<TlsSocketEndpoint>(
info->tls_ = std::make_unique<TlsSocketEndpoint>(
std::move(endpoint),
account.identity(),
account.dhParams(),
......@@ -729,32 +706,22 @@ ConnectionManager::Impl::onRequestOnNegoDone(const PeerConnectionRequest& req)
return shared->validatePeerCertificate(cert, peer_h) && peer_h == ph;
});
auto& nonReadyIt = nonReadySockets_[deviceId][req.id];
nonReadyIt = std::move(tlsSocket);
nonReadyIt->setOnReady([w = weak(), deviceId, vid = std::move(req.id)](bool ok) {
info->tls_->setOnReady(
[w = weak(), deviceId = std::move(req.from), vid = std::move(req.id)](bool ok) {
auto shared = w.lock();
if (!shared)
return;
if (shared->multiplexedSockets_[deviceId].find(vid)
!= shared->multiplexedSockets_[deviceId].end())
auto info = shared->getInfo(deviceId, vid);
if (!info)
return;
if (!ok) {
JAMI_ERR() << "TLS connection failure for peer " << deviceId;
if (shared->connReadyCb_)
shared->connReadyCb_(deviceId, "", nullptr);
shared->connReadyCb_(deviceId.toString(), "", nullptr);
} else {
// The socket is ready, store it in multiplexedSockets_
std::lock_guard<std::mutex> lk(shared->msocketsMutex_);
std::lock_guard<std::mutex> lknrs(shared->nonReadySocketsMutex_);
auto nonReadyIt = shared->nonReadySockets_.find(deviceId);
if (nonReadyIt != shared->nonReadySockets_.end()) {
JAMI_DBG("Connection to %s is ready", deviceId.c_str());
shared->addNewMultiplexedSocket(deviceId, vid, std::move(nonReadyIt->second[vid]));
nonReadyIt->second.erase(vid);
if (nonReadyIt->second.empty()) {
shared->nonReadySockets_.erase(nonReadyIt);
}
}
// The socket is ready, store it
JAMI_DBG("Connection to %s is ready", deviceId.to_c_str());
shared->addNewMultiplexedSocket(deviceId.toString(), vid);
}
});
}
......@@ -803,7 +770,7 @@ ConnectionManager::Impl::onDhtPeerRequest(const PeerConnectionRequest& req,
return;
}
dht::ThreadPool::io().run([w = std::move(w), deviceId, req = std::move(req)] {
dht::ThreadPool::io().run([w = std::move(w), req = std::move(req)] {
auto shared = w.lock();
if (!shared)
return;
......@@ -822,7 +789,7 @@ ConnectionManager::Impl::onDhtPeerRequest(const PeerConnectionRequest& req,
return;
}
dht::ThreadPool::io().run([w = std::move(w), deviceId, req = std::move(req)] {
dht::ThreadPool::io().run([w = std::move(w), req = std::move(req)] {
auto shared = w.lock();
if (!shared)
return;
......@@ -830,16 +797,18 @@ ConnectionManager::Impl::onDhtPeerRequest(const PeerConnectionRequest& req,
});
};
// 1. Create a new Multiplexed Socket
// Negotiate a new ICE socket
auto& connectionInfo = connectionsInfos_[deviceId][req.id];
std::unique_lock<std::mutex> lk {connectionInfo.mutex_};
connectionInfo.ice_ = iceTransportFactory.createUTransport(account.getAccountID().c_str(),
auto info = std::make_shared<ConnectionInfo>();
{
std::lock_guard<std::mutex> lk(infosMtx_);
infos_[{req.from, req.id}] = info;
}
std::unique_lock<std::mutex> lk {info->mutex_};
info->ice_ = iceTransportFactory.createUTransport(account.getAccountID().c_str(),
1,
true,
ice_config);
if (not connectionInfo.ice_) {
if (not info->ice_) {
JAMI_ERR("Cannot initialize ICE session.");
if (connReadyCb_)
connReadyCb_(deviceId, "", nullptr);
......@@ -849,25 +818,26 @@ ConnectionManager::Impl::onDhtPeerRequest(const PeerConnectionRequest& req,
void
ConnectionManager::Impl::addNewMultiplexedSocket(const std::string& deviceId,
const dht::Value::Id& vid,
std::unique_ptr<TlsSocketEndpoint>&& tlsSocket)
const dht::Value::Id& vid)
{
// mSocketsMutex_ MUST be locked
auto mSock = std::make_shared<MultiplexedSocket>(deviceId, std::move(tlsSocket));
mSock->setOnReady(
auto info = getInfo(dht::InfoHash(deviceId), vid);
if (!info)
return;
info->socket_ = std::make_shared<MultiplexedSocket>(deviceId, std::move(info->tls_));
info->socket_->setOnReady(
[w = weak()](const std::string& deviceId, const std::shared_ptr<ChannelSocket>& socket) {
if (auto sthis = w.lock())
if (sthis->connReadyCb_)
sthis->connReadyCb_(deviceId, socket->name(), socket);
});
mSock->setOnRequest(
info->socket_->setOnRequest(
[w = weak()](const std::string& deviceId, const uint16_t&, const std::string& name) {
if (auto sthis = w.lock())
if (sthis->channelReqCb_)
return sthis->channelReqCb_(deviceId, name);
return false;
});
mSock->onShutdown([w = weak(), deviceId, vid]() {
info->socket_->onShutdown([w = weak(), deviceId, vid]() {
auto sthis = w.lock();
if (!sthis)
return;
......@@ -890,45 +860,20 @@ ConnectionManager::Impl::addNewMultiplexedSocket(const std::string& deviceId,
auto sthis = w.lock();
if (!sthis)
return;
// Delete the socket
std::lock_guard<std::mutex> lk(sthis->msocketsMutex_);
auto mxSockIt = sthis->multiplexedSockets_.find(deviceId);
if (mxSockIt != sthis->multiplexedSockets_.end()) {
auto vidIt = mxSockIt->second.find(vid);
if (vidIt != mxSockIt->second.end() && vidIt->second) {
vidIt->second->shutdown();
}
}
auto connIt = sthis->connectionsInfos_.find(deviceId);
if (connIt != sthis->connectionsInfos_.end()) {
auto it = connIt->second.find(vid);
if (it != connIt->second.end()) {
if (it->second.ice_)
it->second.ice_->cancelOperations();
connIt->second.erase(vid);
if (connIt->second.empty())
sthis->connectionsInfos_.erase(deviceId);
}
// This will close the TLS Session
std::lock_guard<std::mutex> lk(sthis->nonReadySocketsMutex_);
auto nonReadyIt = sthis->nonReadySockets_.find(deviceId);
if (nonReadyIt != sthis->nonReadySockets_.end()) {
nonReadyIt->second.erase(vid);
if (nonReadyIt->second.empty()) {
sthis->nonReadySockets_.erase(nonReadyIt);
}
}
}
auto info = sthis->getInfo(dht::InfoHash(deviceId), vid);
if (!info)
return;
if (mxSockIt != sthis->multiplexedSockets_.end()) {
mxSockIt->second.erase(vid);
if (mxSockIt->second.empty())
sthis->multiplexedSockets_.erase(mxSockIt);
}
if (info->socket_)
info->socket_->shutdown();
if (info && info->ice_)
info->ice_->cancelOperations();
std::lock_guard<std::mutex> lk(sthis->infosMtx_);
sthis->infos_.erase({dht::InfoHash(deviceId), vid});
});
});
multiplexedSockets_[deviceId][vid] = std::move(mSock);
}
bool
......@@ -979,18 +924,28 @@ ConnectionManager::closeConnectionsWith(const std::string& deviceId)
}
}
}
auto it = pimpl_->connectionsInfos_.find(deviceId);
if (it != pimpl_->connectionsInfos_.end()) {
for (auto& info : it->second) {
if (info.second.ice_) {
info.second.ice_->cancelOperations();
info.second.ice_->stop();
std::vector<std::shared_ptr<ConnectionInfo>> connInfos;
{
std::lock_guard<std::mutex> lk(pimpl_->infosMtx_);
for (auto iter = pimpl_->infos_.begin(); iter != pimpl_->infos_.end();) {
auto const& [key, value] = *iter;
if (key.first == dht::InfoHash(deviceId)) {
connInfos.emplace_back(value);
iter = pimpl_->infos_.erase(iter);
} else {
iter++;
}
}
}
info.second.responseCv_.notify_all();
if (info.second.ice_) {
std::unique_lock<std::mutex> lk {info.second.mutex_};
info.second.ice_.reset();
for (auto& info : connInfos) {
if (info->ice_) {
info->ice_->cancelOperations();
info->ice_->stop();
}
info->responseCv_.notify_all();
if (info->ice_) {
std::unique_lock<std::mutex> lk {info->mutex_};
info->ice_.reset();
}
}
// This will close the TLS Session
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment