Internal change

PiperOrigin-RevId: 362371994
2026-09-15 15:16:12 -04:00 · 2021-03-11 14:01:08 -08:00
parent b3d4ac6ffc
commit cb73246ada
4 changed files with 134 additions and 99 deletions
@@ -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<UKey2Handshake> 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<UKey2Handshake>(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<UKey2Handshake>(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<Future<Status>>();
-  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<Status> 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<Status> 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;
@@ -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<DiscoveredEndpoint> endpoint);
+                       std::shared_ptr<DiscoveredEndpoint> 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<proto::connections::Medium>
  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_;
@@ -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<DiscoveredEndpoint> endpoint) {
+                       std::shared_ptr<DiscoveredEndpoint> 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<BasePcpHandler::DiscoveredEndpoint*> GetDiscoveredEndpoints(
@@ -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<BluetoothEndpointChannel>(
-                  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<BluetoothEndpointChannel>(
+                      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<BleEndpointChannel>(
-                    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<BleEndpointChannel>(
+                            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<BluetoothEndpointChannel>(
-                      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<BluetoothEndpointChannel>(
+                                  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<WifiLanEndpointChannel>(
-                  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<WifiLanEndpointChannel>(
+                          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<WebRtcEndpointChannel>(
                        remote_device_name, socket);