From cb73246ada6fc9251ef9baa96a1308420fdf021e Mon Sep 17 00:00:00 2001 From: hai007 Date: Thu, 11 Mar 2021 14:00:45 -0800 Subject: [PATCH] Internal change PiperOrigin-RevId: 362371994 --- cpp/core/internal/base_pcp_handler.cc | 49 ++++--- cpp/core/internal/base_pcp_handler.h | 48 ++++--- cpp/core/internal/base_pcp_handler_test.cc | 6 +- cpp/core/internal/p2p_cluster_pcp_handler.cc | 130 +++++++++++-------- 4 files changed, 134 insertions(+), 99 deletions(-) diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index 0832a037..dcad3554 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -83,7 +83,7 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client, GetStringValueOfSupportedMediums(options).c_str()); ConnectionOptions advertising_options = options.CompatibleOptions(); RunOnPcpHandlerThread([this, client, &service_id, &info, &advertising_options, - &response]() { + &response]() RUN_ON_PCP_HANDLER_THREAD() { // The endpoint id inside of the advertisement is different to high // visibility and low visibility mode. In order to decide if client should // grab the high visibility or low visibility id, it needs to tell client @@ -118,7 +118,7 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client, void BasePcpHandler::StopAdvertising(ClientProxy* client) { CountDownLatch latch(1); - RunOnPcpHandlerThread([this, client, &latch]() { + RunOnPcpHandlerThread([this, client, &latch]() RUN_ON_PCP_HANDLER_THREAD() { StopAdvertisingImpl(client); client->StoppedAdvertising(); latch.CountDown(); @@ -153,7 +153,7 @@ Status BasePcpHandler::StartDiscovery(ClientProxy* client, NEARBY_LOG(INFO, "StartDiscovery with supported mediums: %s", GetStringValueOfSupportedMediums(options).c_str()); RunOnPcpHandlerThread([this, client, service_id, discovery_options, &listener, - &response]() { + &response]() RUN_ON_PCP_HANDLER_THREAD() { // Ask the implementation to attempt to start discovery. auto result = StartDiscoveryImpl(client, service_id, discovery_options); if (!result.status.Ok()) { @@ -174,7 +174,7 @@ Status BasePcpHandler::StartDiscovery(ClientProxy* client, void BasePcpHandler::StopDiscovery(ClientProxy* client) { CountDownLatch latch(1); - RunOnPcpHandlerThread([this, client, &latch]() { + RunOnPcpHandlerThread([this, client, &latch]() RUN_ON_PCP_HANDLER_THREAD() { StopDiscoveryImpl(client); client->StoppedDiscovery(); latch.CountDown(); @@ -187,10 +187,11 @@ void BasePcpHandler::InjectEndpoint( ClientProxy* client, const std::string& service_id, const OutOfBandConnectionMetadata& metadata) { CountDownLatch latch(1); - RunOnPcpHandlerThread([this, client, service_id, metadata, &latch]() { - InjectEndpointImpl(client, service_id, metadata); - latch.CountDown(); - }); + RunOnPcpHandlerThread([this, client, service_id, metadata, &latch]() + RUN_ON_PCP_HANDLER_THREAD() { + InjectEndpointImpl(client, service_id, metadata); + latch.CountDown(); + }); WaitForLatch(absl::StrCat("InjectEndpoint(", service_id, ")"), &latch); } @@ -235,17 +236,18 @@ EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() { std::unique_ptr ukey2, const std::string& auth_token, const ByteArray& raw_auth_token) { - RunOnPcpHandlerThread([this, endpoint_id, - raw_ukey2 = ukey2.release(), auth_token, - raw_auth_token]() mutable { - OnEncryptionSuccessRunnable( - endpoint_id, std::unique_ptr(raw_ukey2), - auth_token, raw_auth_token); - }); + RunOnPcpHandlerThread( + [this, endpoint_id, raw_ukey2 = ukey2.release(), auth_token, + raw_auth_token]() RUN_ON_PCP_HANDLER_THREAD() mutable { + OnEncryptionSuccessRunnable( + endpoint_id, std::unique_ptr(raw_ukey2), + auth_token, raw_auth_token); + }); }, .on_failure_cb = [this](const std::string& endpoint_id, EndpointChannel* channel) { - RunOnPcpHandlerThread([this, endpoint_id, channel]() { + RunOnPcpHandlerThread([this, endpoint_id, + channel]() RUN_ON_PCP_HANDLER_THREAD() { NEARBY_LOG(ERROR, "Encryption failed for %s on medium %d", endpoint_id.c_str(), channel->GetMedium()); OnEncryptionFailureRunnable(endpoint_id, channel); @@ -341,7 +343,8 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, const ConnectionRequestInfo& info, const ConnectionOptions& options) { auto result = std::make_shared>(); - RunOnPcpHandlerThread([this, client, &info, options, endpoint_id, result]() { + RunOnPcpHandlerThread([this, client, &info, options, endpoint_id, + result]() RUN_ON_PCP_HANDLER_THREAD() { absl::Time start_time = SystemClock::ElapsedRealtime(); // If we already have a pending connection, then we shouldn't allow any more @@ -618,7 +621,8 @@ Status BasePcpHandler::AcceptConnection( const PayloadListener& payload_listener) { Future response; RunOnPcpHandlerThread( - [this, client, endpoint_id, payload_listener, &response]() { + [this, client, endpoint_id, payload_listener, + &response]() RUN_ON_PCP_HANDLER_THREAD() { NEARBY_LOG(INFO, "AcceptConnection: id=%s", endpoint_id.c_str()); if (!pending_connections_.count(endpoint_id)) { NEARBY_LOG(INFO, "AcceptConnection: no pending connection for id=%s", @@ -671,7 +675,8 @@ Status BasePcpHandler::AcceptConnection( Status BasePcpHandler::RejectConnection(ClientProxy* client, const std::string& endpoint_id) { Future response; - RunOnPcpHandlerThread([this, client, endpoint_id, &response]() { + RunOnPcpHandlerThread([this, client, endpoint_id, + &response]() RUN_ON_PCP_HANDLER_THREAD() { NEARBY_LOG(INFO, "RejectConnection: id=%s", endpoint_id.c_str()); if (!pending_connections_.count(endpoint_id)) { NEARBY_LOG(INFO, "RejectConnection: no pending connection for id=%s", @@ -725,7 +730,8 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame, ClientProxy* client, proto::connections::Medium medium) { CountDownLatch latch(1); - RunOnPcpHandlerThread([this, client, endpoint_id, frame, &latch]() { + RunOnPcpHandlerThread([this, client, endpoint_id, frame, + &latch]() RUN_ON_PCP_HANDLER_THREAD() { NEARBY_LOG(INFO, "OnConnectionResponse: id=%s", endpoint_id.c_str()); if (client->HasRemoteEndpointResponded(endpoint_id)) { @@ -773,7 +779,8 @@ void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client, barrier.CountDown(); return; } - RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() mutable { + RunOnPcpHandlerThread([this, client, endpoint_id, + barrier]() RUN_ON_PCP_HANDLER_THREAD() mutable { auto item = pending_alarms_.find(endpoint_id); if (item != pending_alarms_.end()) { auto& alarm = item->second; diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index 3c8827d9..9ffcebb7 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -61,6 +61,11 @@ enum class WebRtcState { kUnconnectable = 2, }; +// Annotations for methods that need to run on PCP handler thread. +// Use only in BasePcpHandler and derived classes. +#define RUN_ON_PCP_HANDLER_THREAD() \ + ABSL_EXCLUSIVE_LOCKS_REQUIRED(GetPcpHandlerThread()) + // A base implementation of the PcpHandler interface that takes care of all // bookkeeping and handshake protocols that are common across all PcpHandler // implementations -- thus, every concrete PcpHandler implementation must extend @@ -226,12 +231,12 @@ class BasePcpHandler : public PcpHandler, ConnectionOptions GetConnectionOptions() const; ConnectionOptions GetDiscoveryOptions() const; - // @PcpHandlerThread void OnEndpointFound(ClientProxy* client, - std::shared_ptr endpoint); + std::shared_ptr endpoint) + RUN_ON_PCP_HANDLER_THREAD(); - // @PcpHandlerThread - void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint); + void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint) + RUN_ON_PCP_HANDLER_THREAD(); Exception OnIncomingConnection( ClientProxy* client, const ByteArray& remote_endpoint_info, @@ -244,30 +249,30 @@ class BasePcpHandler : public PcpHandler, virtual bool CanSendOutgoingConnection(ClientProxy* client) const; virtual bool CanReceiveIncomingConnection(ClientProxy* client) const; - // @PcpHandlerThread virtual StartOperationResult StartAdvertisingImpl( ClientProxy* client, const std::string& service_id, const std::string& local_endpoint_id, - const ByteArray& local_endpoint_info, - const ConnectionOptions& options) = 0; - // @PcpHandlerThread - virtual Status StopAdvertisingImpl(ClientProxy* client) = 0; + const ByteArray& local_endpoint_info, const ConnectionOptions& options) + RUN_ON_PCP_HANDLER_THREAD() = 0; + + virtual Status StopAdvertisingImpl(ClientProxy* client) + RUN_ON_PCP_HANDLER_THREAD() = 0; - // @PcpHandlerThread virtual StartOperationResult StartDiscoveryImpl( ClientProxy* client, const std::string& service_id, - const ConnectionOptions& options) = 0; - // @PcpHandlerThread - virtual Status StopDiscoveryImpl(ClientProxy* client) = 0; + const ConnectionOptions& options) RUN_ON_PCP_HANDLER_THREAD() = 0; - // @PcpHandlerThread - virtual Status InjectEndpointImpl( - ClientProxy* client, const std::string& service_id, - const OutOfBandConnectionMetadata& metadata) = 0; + virtual Status StopDiscoveryImpl(ClientProxy* client) + RUN_ON_PCP_HANDLER_THREAD() = 0; + + virtual Status InjectEndpointImpl(ClientProxy* client, + const std::string& service_id, + const OutOfBandConnectionMetadata& metadata) + RUN_ON_PCP_HANDLER_THREAD() = 0; - // @PcpHandlerThread virtual ConnectImplResult ConnectImpl(ClientProxy* client, - DiscoveredEndpoint* endpoint) = 0; + DiscoveredEndpoint* endpoint) + RUN_ON_PCP_HANDLER_THREAD() = 0; virtual std::vector GetConnectionMediumsByPriority() = 0; @@ -289,6 +294,11 @@ class BasePcpHandler : public PcpHandler, const string& endpoint_id, const ByteArray& endpoint_info); + SingleThreadExecutor* GetPcpHandlerThread() + ABSL_LOCK_RETURNED(serial_executor_) { + return &serial_executor_; + } + Mediums* mediums_; EndpointManager* endpoint_manager_; EndpointChannelManager* channel_manager_; diff --git a/cpp/core/internal/base_pcp_handler_test.cc b/cpp/core/internal/base_pcp_handler_test.cc index 1f3a9657..fd1ac030 100644 --- a/cpp/core/internal/base_pcp_handler_test.cc +++ b/cpp/core/internal/base_pcp_handler_test.cc @@ -153,10 +153,12 @@ class MockPcpHandler : public BasePcpHandler { // Mock adapters for protected non-virtual methods of a base class. void OnEndpointFound(ClientProxy* client, - std::shared_ptr endpoint) { + std::shared_ptr endpoint) + ABSL_NO_THREAD_SAFETY_ANALYSIS { BasePcpHandler::OnEndpointFound(client, std::move(endpoint)); } - void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint) { + void OnEndpointLost(ClientProxy* client, const DiscoveredEndpoint& endpoint) + ABSL_NO_THREAD_SAFETY_ANALYSIS { BasePcpHandler::OnEndpointLost(client, endpoint); } std::vector GetDiscoveredEndpoints( diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.cc b/cpp/core/internal/p2p_cluster_pcp_handler.cc index 847f617d..f8cef073 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.cc +++ b/cpp/core/internal/p2p_cluster_pcp_handler.cc @@ -206,7 +206,8 @@ bool P2pClusterPcpHandler::IsRecognizedBluetoothEndpoint( void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler( ClientProxy* client, const std::string& service_id, BluetoothDevice device) { - RunOnPcpHandlerThread([this, client, service_id, device]() { + RunOnPcpHandlerThread([this, client, service_id, + device]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -244,7 +245,8 @@ void P2pClusterPcpHandler::BluetoothDeviceDiscoveredHandler( void P2pClusterPcpHandler::BluetoothNameChangedHandler( ClientProxy* client, const std::string& service_id, BluetoothDevice device) { - RunOnPcpHandlerThread([this, client, service_id, device]() { + RunOnPcpHandlerThread([this, client, service_id, + device]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -322,7 +324,8 @@ void P2pClusterPcpHandler::BluetoothDeviceLostHandler( ClientProxy* client, const std::string& service_id, BluetoothDevice& device) { const std::string& device_name_string = device.GetName(); - RunOnPcpHandlerThread([this, client, service_id, device_name_string]() { + RunOnPcpHandlerThread([this, client, service_id, + device_name_string]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -406,7 +409,8 @@ void P2pClusterPcpHandler::BlePeripheralDiscoveredHandler( const std::string& service_id, const ByteArray& advertisement_bytes, bool fast_advertisement) { RunOnPcpHandlerThread([this, client, &peripheral, service_id, - advertisement_bytes, fast_advertisement]() { + advertisement_bytes, + fast_advertisement]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -482,7 +486,8 @@ void P2pClusterPcpHandler::BlePeripheralLostHandler( std::string peripheral_name = peripheral.GetName(); NEARBY_LOG(INFO, "Ble: [LOST, SCHED] peripheral_name=%s", peripheral_name.c_str()); - RunOnPcpHandlerThread([this, client, service_id, &peripheral]() { + RunOnPcpHandlerThread([this, client, service_id, + &peripheral]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG(INFO, @@ -552,7 +557,8 @@ bool P2pClusterPcpHandler::IsRecognizedWifiLanEndpoint( void P2pClusterPcpHandler::WifiLanServiceDiscoveredHandler( ClientProxy* client, WifiLanService& wifi_lan_service, const std::string& service_id) { - RunOnPcpHandlerThread([this, client, service_id, &wifi_lan_service]() { + RunOnPcpHandlerThread([this, client, service_id, + &wifi_lan_service]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG( @@ -597,7 +603,8 @@ void P2pClusterPcpHandler::WifiLanServiceLostHandler( NEARBY_LOG(INFO, "WifiLan: [LOST, SCHED] wifi_lan_service=%p, service_info_name=%s", &wifi_lan_service, nsd_service_info.GetServiceInfoName().c_str()); - RunOnPcpHandlerThread([this, client, service_id, nsd_service_info]() { + RunOnPcpHandlerThread([this, client, service_id, + nsd_service_info]() RUN_ON_PCP_HANDLER_THREAD() { // Make sure we are still discovering before proceeding. if (!client->IsDiscovering()) { NEARBY_LOG( @@ -822,18 +829,20 @@ proto::connections::Medium P2pClusterPcpHandler::StartBluetoothAdvertising( std::string(local_endpoint_info).c_str()); return; } - RunOnPcpHandlerThread([this, client, local_endpoint_info, - socket = std::move(socket)]() mutable { - std::string remote_device_name = - socket.GetRemoteDevice().GetName(); - auto channel = absl::make_unique( - remote_device_name, socket); - ByteArray remote_device_info{remote_device_name}; + RunOnPcpHandlerThread( + [this, client, local_endpoint_info, + socket = + std::move(socket)]() RUN_ON_PCP_HANDLER_THREAD() mutable { + std::string remote_device_name = + socket.GetRemoteDevice().GetName(); + auto channel = absl::make_unique( + remote_device_name, socket); + ByteArray remote_device_info{remote_device_name}; - OnIncomingConnection(client, remote_device_info, - std::move(channel), - proto::connections::Medium::BLUETOOTH); - }); + OnIncomingConnection(client, remote_device_info, + std::move(channel), + proto::connections::Medium::BLUETOOTH); + }); }})) { NEARBY_LOG(INFO, "BT failed to start accepting connections for service=%s", service_id.c_str()); @@ -949,21 +958,22 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising( std::string(local_endpoint_info).c_str()); return; } - RunOnPcpHandlerThread([this, client, local_endpoint_info, - service_id, - socket = std::move(socket)]() mutable { - std::string remote_peripheral_name = - socket.GetRemotePeripheral().GetName(); - auto channel = absl::make_unique( - remote_peripheral_name, socket); - ByteArray remote_peripheral_info = - socket.GetRemotePeripheral().GetAdvertisementBytes( - service_id); + RunOnPcpHandlerThread( + [this, client, local_endpoint_info, service_id, + socket = std::move(socket)]() + RUN_ON_PCP_HANDLER_THREAD() mutable { + std::string remote_peripheral_name = + socket.GetRemotePeripheral().GetName(); + auto channel = absl::make_unique( + remote_peripheral_name, socket); + ByteArray remote_peripheral_info = + socket.GetRemotePeripheral().GetAdvertisementBytes( + service_id); - OnIncomingConnection(client, remote_peripheral_info, - std::move(channel), - proto::connections::Medium::BLE); - }); + OnIncomingConnection(client, remote_peripheral_info, + std::move(channel), + proto::connections::Medium::BLE); + }); }})) { NEARBY_LOGS(INFO) << "Ble failed to start accepting connections for service_id=" @@ -988,18 +998,21 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising( std::string(local_endpoint_info).c_str()); return; } - RunOnPcpHandlerThread([this, client, local_endpoint_info, - socket = std::move(socket)]() mutable { - std::string remote_device_name = - socket.GetRemoteDevice().GetName(); - auto channel = absl::make_unique( - remote_device_name, socket); - ByteArray remote_device_info{remote_device_name}; + RunOnPcpHandlerThread( + [this, client, local_endpoint_info, + socket = std::move(socket)]() + RUN_ON_PCP_HANDLER_THREAD() mutable { + std::string remote_device_name = + socket.GetRemoteDevice().GetName(); + auto channel = + absl::make_unique( + remote_device_name, socket); + ByteArray remote_device_info{remote_device_name}; - OnIncomingConnection(client, remote_device_info, - std::move(channel), - proto::connections::Medium::BLUETOOTH); - }); + OnIncomingConnection( + client, remote_device_info, std::move(channel), + proto::connections::Medium::BLUETOOTH); + }); }})) { NEARBY_LOGS(INFO) << "BT failed to start accepting connections for service_id=" @@ -1129,20 +1142,22 @@ proto::connections::Medium P2pClusterPcpHandler::StartWifiLanAdvertising( std::string(local_endpoint_info).c_str()); return; } - RunOnPcpHandlerThread([this, client, local_endpoint_info, - socket = std::move(socket)]() mutable { - std::string remote_service_info_name = - socket.GetRemoteWifiLanService() - .GetServiceInfo() - .GetServiceInfoName(); - auto channel = absl::make_unique( - remote_service_info_name, socket); - ByteArray remote_service_info{remote_service_info_name}; + RunOnPcpHandlerThread( + [this, client, local_endpoint_info, + socket = std::move(socket)]() + RUN_ON_PCP_HANDLER_THREAD() mutable { + std::string remote_service_info_name = + socket.GetRemoteWifiLanService() + .GetServiceInfo() + .GetServiceInfoName(); + auto channel = absl::make_unique( + remote_service_info_name, socket); + ByteArray remote_service_info{remote_service_info_name}; - OnIncomingConnection(client, remote_service_info, - std::move(channel), - proto::connections::Medium::WIFI_LAN); - }); + OnIncomingConnection( + client, remote_service_info, std::move(channel), + proto::connections::Medium::WIFI_LAN); + }); }})) { NEARBY_LOG(INFO, "WifiLan failed to start accepting connections for service=%s", @@ -1257,7 +1272,8 @@ P2pClusterPcpHandler::StartListeningForWebRtcConnections( } RunOnPcpHandlerThread( - [this, client, socket = std::move(socket)]() { + [this, client, + socket = std::move(socket)]() RUN_ON_PCP_HANDLER_THREAD() { std::string remote_device_name = "WebRtcSocket"; auto channel = absl::make_unique( remote_device_name, socket);