mollyim/webrtc
1
0
Fork 0
mirror of https://github.com/mollyim/webrtc.git synced 2025-05-14 06:10:40 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Improve thread annotations for TurnServer

Browse source 
Bug: webrtc:12339
Change-Id: I317485a392ad6cdf77ebf4ea8a7066f8ba0245bb
Reviewed-on: https://webrtc-review.googlesource.com/c/src/+/212502
Reviewed-by: Taylor <deadbeef@webrtc.org>
Commit-Queue: Niels Moller <nisse@webrtc.org>
Cr-Commit-Position: refs/heads/master@{#33516}
This commit is contained in:
Niels Möller 2021-03-18 15:44:24 +01:00 committed by Commit Bot
parent 2f5f5fa171
commit 76b51e21bc
2 changed files with 72 additions and 72 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

41
p2p/base/turn_server.cc
View file

@ -128,7 +128,7 @@ TurnServer::TurnServer(rtc::Thread* thread)
enable_otu_nonce_(false) {}
TurnServer::~TurnServer() {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
for (InternalSocketMap::iterator it = server_sockets_.begin();
it != server_sockets_.end(); ++it) {
rtc::AsyncPacketSocket* socket = it->first;
@ -144,7 +144,7 @@ TurnServer::~TurnServer() {
void TurnServer::AddInternalSocket(rtc::AsyncPacketSocket* socket,
ProtocolType proto) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
RTC_DCHECK(server_sockets_.end() == server_sockets_.find(socket));
server_sockets_[socket] = proto;
socket->SignalReadPacket.connect(this, &TurnServer::OnInternalPacket);
@ -152,7 +152,7 @@ void TurnServer::AddInternalSocket(rtc::AsyncPacketSocket* socket,
void TurnServer::AddInternalServerSocket(rtc::AsyncSocket* socket,
ProtocolType proto) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
RTC_DCHECK(server_listen_sockets_.end() ==
server_listen_sockets_.find(socket));
server_listen_sockets_[socket] = proto;
@ -162,20 +162,19 @@ void TurnServer::AddInternalServerSocket(rtc::AsyncSocket* socket,
void TurnServer::SetExternalSocketFactory(
rtc::PacketSocketFactory* factory,
const rtc::SocketAddress& external_addr) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
external_socket_factory_.reset(factory);
external_addr_ = external_addr;
}
void TurnServer::OnNewInternalConnection(rtc::AsyncSocket* socket) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
RTC_DCHECK(server_listen_sockets_.find(socket) !=
server_listen_sockets_.end());
AcceptConnection(socket);
}
void TurnServer::AcceptConnection(rtc::AsyncSocket* server_socket) {
RTC_DCHECK(thread_checker_.IsCurrent());
// Check if someone is trying to connect to us.
rtc::SocketAddress accept_addr;
rtc::AsyncSocket* accepted_socket = server_socket->Accept(&accept_addr);
@ -192,7 +191,7 @@ void TurnServer::AcceptConnection(rtc::AsyncSocket* server_socket) {
void TurnServer::OnInternalSocketClose(rtc::AsyncPacketSocket* socket,
int err) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
DestroyInternalSocket(socket);
}
@ -201,7 +200,7 @@ void TurnServer::OnInternalPacket(rtc::AsyncPacketSocket* socket,
size_t size,
const rtc::SocketAddress& addr,
const int64_t& /* packet_time_us */) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
// Fail if the packet is too small to even contain a channel header.
if (size < TURN_CHANNEL_HEADER_SIZE) {
return;
@ -228,7 +227,6 @@ void TurnServer::OnInternalPacket(rtc::AsyncPacketSocket* socket,
void TurnServer::HandleStunMessage(TurnServerConnection* conn,
const char* data,
size_t size) {
RTC_DCHECK(thread_checker_.IsCurrent());
TurnMessage msg;
rtc::ByteBufferReader buf(data, size);
if (!msg.Read(&buf) || (buf.Length() > 0)) {
@ -294,7 +292,6 @@ void TurnServer::HandleStunMessage(TurnServerConnection* conn,
}
bool TurnServer::GetKey(const StunMessage* msg, std::string* key) {
RTC_DCHECK(thread_checker_.IsCurrent());
const StunByteStringAttribute* username_attr =
msg->GetByteString(STUN_ATTR_USERNAME);
if (!username_attr) {
@ -310,7 +307,6 @@ bool TurnServer::CheckAuthorization(TurnServerConnection* conn,
const char* data,
size_t size,
const std::string& key) {
RTC_DCHECK(thread_checker_.IsCurrent());
// RFC 5389, 10.2.2.
RTC_DCHECK(IsStunRequestType(msg->type()));
const StunByteStringAttribute* mi_attr =
@ -369,7 +365,6 @@ bool TurnServer::CheckAuthorization(TurnServerConnection* conn,
void TurnServer::HandleBindingRequest(TurnServerConnection* conn,
const StunMessage* req) {
RTC_DCHECK(thread_checker_.IsCurrent());
StunMessage response;
InitResponse(req, &response);
@ -384,7 +379,6 @@ void TurnServer::HandleBindingRequest(TurnServerConnection* conn,
void TurnServer::HandleAllocateRequest(TurnServerConnection* conn,
const TurnMessage* msg,
const std::string& key) {
RTC_DCHECK(thread_checker_.IsCurrent());
// Check the parameters in the request.
const StunUInt32Attribute* transport_attr =
msg->GetUInt32(STUN_ATTR_REQUESTED_TRANSPORT);
@ -414,7 +408,6 @@ void TurnServer::HandleAllocateRequest(TurnServerConnection* conn,
}
std::string TurnServer::GenerateNonce(int64_t now) const {
RTC_DCHECK(thread_checker_.IsCurrent());
// Generate a nonce of the form hex(now + HMAC-MD5(nonce_key_, now))
std::string input(reinterpret_cast<const char*>(&now), sizeof(now));
std::string nonce = rtc::hex_encode(input.c_str(), input.size());
@ -425,7 +418,6 @@ std::string TurnServer::GenerateNonce(int64_t now) const {
}
bool TurnServer::ValidateNonce(const std::string& nonce) const {
RTC_DCHECK(thread_checker_.IsCurrent());
// Check the size.
if (nonce.size() != kNonceSize) {
return false;
@ -452,7 +444,6 @@ bool TurnServer::ValidateNonce(const std::string& nonce) const {
}
TurnServerAllocation* TurnServer::FindAllocation(TurnServerConnection* conn) {
RTC_DCHECK(thread_checker_.IsCurrent());
AllocationMap::const_iterator it = allocations_.find(*conn);
return (it != allocations_.end()) ? it->second.get() : nullptr;
}
@ -460,7 +451,6 @@ TurnServerAllocation* TurnServer::FindAllocation(TurnServerConnection* conn) {
TurnServerAllocation* TurnServer::CreateAllocation(TurnServerConnection* conn,
int proto,
const std::string& key) {
RTC_DCHECK(thread_checker_.IsCurrent());
rtc::AsyncPacketSocket* external_socket =
(external_socket_factory_)
? external_socket_factory_->CreateUdpSocket(external_addr_, 0, 0)
@ -481,7 +471,7 @@ void TurnServer::SendErrorResponse(TurnServerConnection* conn,
const StunMessage* req,
int code,
const std::string& reason) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
TurnMessage resp;
InitErrorResponse(req, code, reason, &resp);
RTC_LOG(LS_INFO) << "Sending error response, type=" << resp.type()
@ -493,7 +483,6 @@ void TurnServer::SendErrorResponseWithRealmAndNonce(TurnServerConnection* conn,
const StunMessage* msg,
int code,
const std::string& reason) {
RTC_DCHECK(thread_checker_.IsCurrent());
TurnMessage resp;
InitErrorResponse(msg, code, reason, &resp);
@ -513,7 +502,6 @@ void TurnServer::SendErrorResponseWithAlternateServer(
TurnServerConnection* conn,
const StunMessage* msg,
const rtc::SocketAddress& addr) {
RTC_DCHECK(thread_checker_.IsCurrent());
TurnMessage resp;
InitErrorResponse(msg, STUN_ERROR_TRY_ALTERNATE,
STUN_ERROR_REASON_TRY_ALTERNATE_SERVER, &resp);
@ -523,7 +511,7 @@ void TurnServer::SendErrorResponseWithAlternateServer(
}
void TurnServer::SendStun(TurnServerConnection* conn, StunMessage* msg) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
rtc::ByteBufferWriter buf;
// Add a SOFTWARE attribute if one is set.
if (!software_.empty()) {
@ -536,13 +524,12 @@ void TurnServer::SendStun(TurnServerConnection* conn, StunMessage* msg) {
void TurnServer::Send(TurnServerConnection* conn,
const rtc::ByteBufferWriter& buf) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
rtc::PacketOptions options;
conn->socket()->SendTo(buf.Data(), buf.Length(), conn->src(), options);
}
void TurnServer::OnAllocationDestroyed(TurnServerAllocation* allocation) {
RTC_DCHECK(thread_checker_.IsCurrent());
// Removing the internal socket if the connection is not udp.
rtc::AsyncPacketSocket* socket = allocation->conn()->socket();
InternalSocketMap::iterator iter = server_sockets_.find(socket);
@ -562,7 +549,6 @@ void TurnServer::OnAllocationDestroyed(TurnServerAllocation* allocation) {
}
void TurnServer::DestroyInternalSocket(rtc::AsyncPacketSocket* socket) {
RTC_DCHECK(thread_checker_.IsCurrent());
InternalSocketMap::iterator iter = server_sockets_.find(socket);
if (iter != server_sockets_.end()) {
rtc::AsyncPacketSocket* socket = iter->first;
@ -573,13 +559,14 @@ void TurnServer::DestroyInternalSocket(rtc::AsyncPacketSocket* socket) {
// deleting an object from within a callback from that object).
sockets_to_delete_.push_back(
std::unique_ptr<rtc::AsyncPacketSocket>(socket));
invoker_.AsyncInvoke<void>(RTC_FROM_HERE, rtc::Thread::Current(),
[this] { FreeSockets(); });
invoker_.AsyncInvoke<void>(RTC_FROM_HERE, rtc::Thread::Current(), [this] {
RTC_DCHECK_RUN_ON(thread_);
FreeSockets();
});
}
}
void TurnServer::FreeSockets() {
RTC_DCHECK(thread_checker_.IsCurrent());
sockets_to_delete_.clear();
}

103
p2p/base/turn_server.h
View file

@ -129,8 +129,8 @@ class TurnServerAllocation : public rtc::MessageHandlerAutoCleanup,
void OnChannelDestroyed(Channel* channel);
void OnMessage(rtc::Message* msg) override;
TurnServer* server_;
rtc::Thread* thread_;
TurnServer* const server_;
rtc::Thread* const thread_;
TurnServerConnection conn_;
std::unique_ptr<rtc::AsyncPacketSocket> external_socket_;
std::string key_;
@ -183,53 +183,53 @@ class TurnServer : public sigslot::has_slots<> {
// Gets/sets the realm value to use for the server.
const std::string& realm() const {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
return realm_;
}
void set_realm(const std::string& realm) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
realm_ = realm;
}
// Gets/sets the value for the SOFTWARE attribute for TURN messages.
const std::string& software() const {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
return software_;
}
void set_software(const std::string& software) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
software_ = software;
}
const AllocationMap& allocations() const {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
return allocations_;
}
// Sets the authentication callback; does not take ownership.
void set_auth_hook(TurnAuthInterface* auth_hook) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
auth_hook_ = auth_hook;
}
void set_redirect_hook(TurnRedirectInterface* redirect_hook) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
redirect_hook_ = redirect_hook;
}
void set_enable_otu_nonce(bool enable) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
enable_otu_nonce_ = enable;
}
// If set to true, reject CreatePermission requests to RFC1918 addresses.
void set_reject_private_addresses(bool filter) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
reject_private_addresses_ = filter;
}
void set_enable_permission_checks(bool enable) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
enable_permission_checks_ = enable;
}
@ -244,18 +244,22 @@ class TurnServer : public sigslot::has_slots<> {
const rtc::SocketAddress& address);
// For testing only.
std::string SetTimestampForNextNonce(int64_t timestamp) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
ts_for_next_nonce_ = timestamp;
return GenerateNonce(timestamp);
}
void SetStunMessageObserver(std::unique_ptr<StunMessageObserver> observer) {
RTC_DCHECK(thread_checker_.IsCurrent());
RTC_DCHECK_RUN_ON(thread_);
stun_message_observer_ = std::move(observer);
}
private:
std::string GenerateNonce(int64_t now) const;
// All private member functions and variables should have access restricted to
// thread_. But compile-time annotations are missing for members access from
// TurnServerAllocation (via friend declaration), and the On* methods, which
// are called via sigslot.
std::string GenerateNonce(int64_t now) const RTC_RUN_ON(thread_);
void OnInternalPacket(rtc::AsyncPacketSocket* socket,
const char* data,
size_t size,
@ -265,29 +269,32 @@ class TurnServer : public sigslot::has_slots<> {
void OnNewInternalConnection(rtc::AsyncSocket* socket);
// Accept connections on this server socket.
void AcceptConnection(rtc::AsyncSocket* server_socket);
void AcceptConnection(rtc::AsyncSocket* server_socket) RTC_RUN_ON(thread_);
void OnInternalSocketClose(rtc::AsyncPacketSocket* socket, int err);
void HandleStunMessage(TurnServerConnection* conn,
const char* data,
size_t size);
void HandleBindingRequest(TurnServerConnection* conn, const StunMessage* msg);
size_t size) RTC_RUN_ON(thread_);
void HandleBindingRequest(TurnServerConnection* conn, const StunMessage* msg)
RTC_RUN_ON(thread_);
void HandleAllocateRequest(TurnServerConnection* conn,
const TurnMessage* msg,
const std::string& key);
const std::string& key) RTC_RUN_ON(thread_);
bool GetKey(const StunMessage* msg, std::string* key);
bool GetKey(const StunMessage* msg, std::string* key) RTC_RUN_ON(thread_);
bool CheckAuthorization(TurnServerConnection* conn,
StunMessage* msg,
const char* data,
size_t size,
const std::string& key);
bool ValidateNonce(const std::string& nonce) const;
const std::string& key) RTC_RUN_ON(thread_);
bool ValidateNonce(const std::string& nonce) const RTC_RUN_ON(thread_);
TurnServerAllocation* FindAllocation(TurnServerConnection* conn);
TurnServerAllocation* FindAllocation(TurnServerConnection* conn)
RTC_RUN_ON(thread_);
TurnServerAllocation* CreateAllocation(TurnServerConnection* conn,
int proto,
const std::string& key);
const std::string& key)
RTC_RUN_ON(thread_);
void SendErrorResponse(TurnServerConnection* conn,
const StunMessage* req,
@ -297,55 +304,61 @@ class TurnServer : public sigslot::has_slots<> {
void SendErrorResponseWithRealmAndNonce(TurnServerConnection* conn,
const StunMessage* req,
int code,
const std::string& reason);
const std::string& reason)
RTC_RUN_ON(thread_);
void SendErrorResponseWithAlternateServer(TurnServerConnection* conn,
const StunMessage* req,
const rtc::SocketAddress& addr);
const rtc::SocketAddress& addr)
RTC_RUN_ON(thread_);
void SendStun(TurnServerConnection* conn, StunMessage* msg);
void Send(TurnServerConnection* conn, const rtc::ByteBufferWriter& buf);
void OnAllocationDestroyed(TurnServerAllocation* allocation);
void DestroyInternalSocket(rtc::AsyncPacketSocket* socket);
void OnAllocationDestroyed(TurnServerAllocation* allocation)
RTC_RUN_ON(thread_);
void DestroyInternalSocket(rtc::AsyncPacketSocket* socket)
RTC_RUN_ON(thread_);
// Just clears |sockets_to_delete_|; called asynchronously.
void FreeSockets();
void FreeSockets() RTC_RUN_ON(thread_);
typedef std::map<rtc::AsyncPacketSocket*, ProtocolType> InternalSocketMap;
typedef std::map<rtc::AsyncSocket*, ProtocolType> ServerSocketMap;
rtc::Thread* thread_;
webrtc::SequenceChecker thread_checker_;
std::string nonce_key_;
std::string realm_;
std::string software_;
TurnAuthInterface* auth_hook_;
TurnRedirectInterface* redirect_hook_;
rtc::Thread* const thread_;
const std::string nonce_key_;
std::string realm_ RTC_GUARDED_BY(thread_);
std::string software_ RTC_GUARDED_BY(thread_);
TurnAuthInterface* auth_hook_ RTC_GUARDED_BY(thread_);
TurnRedirectInterface* redirect_hook_ RTC_GUARDED_BY(thread_);
// otu - one-time-use. Server will respond with 438 if it's
// sees the same nonce in next transaction.
bool enable_otu_nonce_;
bool enable_otu_nonce_ RTC_GUARDED_BY(thread_);
bool reject_private_addresses_ = false;
// Check for permission when receiving an external packet.
bool enable_permission_checks_ = true;
InternalSocketMap server_sockets_;
ServerSocketMap server_listen_sockets_;
InternalSocketMap server_sockets_ RTC_GUARDED_BY(thread_);
ServerSocketMap server_listen_sockets_ RTC_GUARDED_BY(thread_);
// Used when we need to delete a socket asynchronously.
std::vector<std::unique_ptr<rtc::AsyncPacketSocket>> sockets_to_delete_;
std::unique_ptr<rtc::PacketSocketFactory> external_socket_factory_;
rtc::SocketAddress external_addr_;
std::vector<std::unique_ptr<rtc::AsyncPacketSocket>> sockets_to_delete_
RTC_GUARDED_BY(thread_);
std::unique_ptr<rtc::PacketSocketFactory> external_socket_factory_
RTC_GUARDED_BY(thread_);
rtc::SocketAddress external_addr_ RTC_GUARDED_BY(thread_);
AllocationMap allocations_;
AllocationMap allocations_ RTC_GUARDED_BY(thread_);
rtc::AsyncInvoker invoker_;
// For testing only. If this is non-zero, the next NONCE will be generated
// from this value, and it will be reset to 0 after generating the NONCE.
int64_t ts_for_next_nonce_ = 0;
int64_t ts_for_next_nonce_ RTC_GUARDED_BY(thread_) = 0;
// For testing only. Used to observe STUN messages received.
std::unique_ptr<StunMessageObserver> stun_message_observer_;
std::unique_ptr<StunMessageObserver> stun_message_observer_
RTC_GUARDED_BY(thread_);
friend class TurnServerAllocation;
};