Fixed crash during devices scanning

PiperOrigin-RevId: 518127975

internal/platform/implementation/windows/ble_medium.cc

@@ -136,6 +136,8 @@ BleMedium::BleMedium(api::BluetoothAdapter& adapter)
 bool BleMedium::StartAdvertising(
     const std::string& service_id, const ByteArray& advertisement_bytes,
     const std::string& fast_advertisement_service_uuid) {
+  absl::MutexLock lock(&mutex_);
+
   try {
     if (!adapter_->IsEnabled()) {
       NEARBY_LOGS(WARNING) << "BLE cannot start advertising because the "
@@ -226,6 +228,8 @@ bool BleMedium::StartAdvertising(
 }
 
 bool BleMedium::StopAdvertising(const std::string& service_id) {
+  absl::MutexLock lock(&mutex_);
+
   try {
     if (!adapter_->IsEnabled()) {
       NEARBY_LOGS(WARNING) << "BLE cannot stop advertising because the "
@@ -267,6 +271,8 @@ bool BleMedium::StartScanning(
     const std::string& service_id,
     const std::string& fast_advertisement_service_uuid,
     DiscoveredPeripheralCallback callback) {
+  absl::MutexLock lock(&mutex_);
+
   try {
     if (!adapter_->IsEnabled()) {
       NEARBY_LOGS(WARNING) << "BLE cannot start scanning because the "
@@ -284,10 +290,7 @@ bool BleMedium::StartScanning(
 
     service_id_ = service_id;
     advertisement_received_callback_ = std::move(callback);
-    {
+    peripheral_map_.clear();
-      absl::MutexLock lock(&peripheral_map_mutex_);
-      peripheral_map_.clear();
-    }
 
     watcher_ = BluetoothLEAdvertisementWatcher();
     watcher_token_ = watcher_.Stopped({this, &BleMedium::WatcherHandler});
@@ -324,6 +327,8 @@ bool BleMedium::StartScanning(
 }
 
 bool BleMedium::StopScanning(const std::string& service_id) {
+  absl::MutexLock lock(&mutex_);
+
   try {
     if (!adapter_->IsEnabled()) {
       NEARBY_LOGS(WARNING) << "BLE cannot stop scanning because the "
@@ -365,12 +370,16 @@ bool BleMedium::StopScanning(const std::string& service_id) {
 
 bool BleMedium::StartAcceptingConnections(const std::string& service_id,
                                           AcceptedConnectionCallback callback) {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << "Windows Ble StartAcceptingConnections: service_id="
                     << service_id;
   return true;
 }
 
 bool BleMedium::StopAcceptingConnections(const std::string& service_id) {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << "Windows Ble StopAcceptingConnections: service_id="
                     << service_id;
   return true;
@@ -379,6 +388,8 @@ bool BleMedium::StopAcceptingConnections(const std::string& service_id) {
 std::unique_ptr<api::BleSocket> BleMedium::Connect(
     api::BlePeripheral& remote_peripheral, const std::string& service_id,
     CancellationFlag* cancellation_flag) {
+  absl::MutexLock lock(&mutex_);
+
   if (cancellation_flag->Cancelled()) {
     NEARBY_LOGS(ERROR) << "Windows BLE Connect: Has been cancelled: "
                           "service_id="
@@ -395,6 +406,8 @@ std::unique_ptr<api::BleSocket> BleMedium::Connect(
 void BleMedium::PublisherHandler(
     BluetoothLEAdvertisementPublisher publisher,
     BluetoothLEAdvertisementPublisherStatusChangedEventArgs args) {
+  absl::MutexLock lock(&mutex_);
+
   // This method is called when publisher's status is changed.
   switch (args.Status()) {
     case BluetoothLEAdvertisementPublisherStatus::Created:
@@ -486,6 +499,8 @@ void BleMedium::PublisherHandler(
 void BleMedium::WatcherHandler(
     BluetoothLEAdvertisementWatcher watcher,
     BluetoothLEAdvertisementWatcherStoppedEventArgs args) {
+  absl::MutexLock lock(&mutex_);
+
   // This method is called when watcher stopped. Args give more detailed
   // information on the reason.
   switch (args.Error()) {
@@ -548,6 +563,8 @@ void BleMedium::WatcherHandler(
 void BleMedium::AdvertisementReceivedHandler(
     BluetoothLEAdvertisementWatcher watcher,
     BluetoothLEAdvertisementReceivedEventArgs args) {
+  absl::MutexLock lock(&mutex_);
+
   // Handle all BLE advertisements and determine whether the BLE Medium
   // Advertisement Scan Response packet (containing Copresence UUID 0xFEF3 in
   // 0x16 Service Data) has been received in the handler
@@ -583,33 +600,30 @@ void BleMedium::AdvertisementReceivedHandler(
 
       BlePeripheral* peripheral_ptr = nullptr;
 
-      {
+      if (peripheral_map_.contains(peripheral_name)) {
-        absl::MutexLock lock(&peripheral_map_mutex_);
+        if (peripheral_map_[peripheral_name]->GetAdvertisementBytes(
-        if (peripheral_map_.contains(peripheral_name)) {
+                service_id_) != advertisement_data) {
-          if (peripheral_map_[peripheral_name]->GetAdvertisementBytes(
+          NEARBY_LOGS(INFO) << "BLE reports lost device: " << peripheral_name;
-                  service_id_) != advertisement_data) {
-            NEARBY_LOGS(INFO) << "BLE reports lost device: " << peripheral_name;
 
-            // Lost the device first and then report discovered the device.
+          // Lost the device first and then report discovered the device.
-            advertisement_received_callback_.peripheral_lost_cb(
+          advertisement_received_callback_.peripheral_lost_cb(
-                /*ble_peripheral*/ *peripheral_map_[peripheral_name],
+              /*ble_peripheral*/ *peripheral_map_[peripheral_name],
-                /*service_id*/ service_id_);
+              /*service_id*/ service_id_);
 
-          } else {
+        } else {
-            // The device already reported to discovery, don't need to call it
+          // The device already reported to discovery, don't need to call it
-            // again.
+          // again.
-            return;
+          return;
-          }
         }
-
-        auto peripheral = std::make_unique<BlePeripheral>();
-        peripheral->SetName(peripheral_name);
-        peripheral->SetAdvertisementBytes(advertisement_data);
-
-        peripheral_map_[peripheral_name] = std::move(peripheral);
-        peripheral_ptr = peripheral_map_[peripheral_name].get();
       }
 
+      auto peripheral = std::make_unique<BlePeripheral>();
+      peripheral->SetName(peripheral_name);
+      peripheral->SetAdvertisementBytes(advertisement_data);
+
+      peripheral_map_[peripheral_name] = std::move(peripheral);
+      peripheral_ptr = peripheral_map_[peripheral_name].get();
+
       // Received Fast Advertisement packet
       if (unconsumed_buffer_length <= 27) {
         NEARBY_LOGS(INFO)

internal/platform/implementation/windows/ble_medium.h

@@ -22,6 +22,7 @@
 #include <string>
 #include <utility>
 
+#include "absl/base/thread_annotations.h"
 #include "absl/container/flat_hash_map.h"
 #include "absl/synchronization/mutex.h"
 #include "internal/platform/implementation/ble.h"
@@ -47,33 +48,40 @@ class BleMedium : public api::BleMedium {
 
   bool StartAdvertising(
       const std::string& service_id, const ByteArray& advertisement_bytes,
-      const std::string& fast_advertisement_service_uuid) override;
+      const std::string& fast_advertisement_service_uuid) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
-  bool StopAdvertising(const std::string& service_id) override;
+  bool StopAdvertising(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Returns true once the BLE scan has been initiated.
   bool StartScanning(const std::string& service_id,
                      const std::string& fast_advertisement_service_uuid,
-                     DiscoveredPeripheralCallback callback) override;
+                     DiscoveredPeripheralCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Returns true once BLE scanning for service_id is well and truly stopped;
   // after this returns, there must be no more invocations of the
   // DiscoveredPeripheralCallback passed in to StartScanning() for service_id.
-  bool StopScanning(const std::string& service_id) override;
+  bool StopScanning(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Returns true once BLE socket connection requests to service_id can be
   // accepted.
   bool StartAcceptingConnections(const std::string& service_id,
-                                 AcceptedConnectionCallback callback) override;
+                                 AcceptedConnectionCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
-  bool StopAcceptingConnections(const std::string& service_id) override;
+  bool StopAcceptingConnections(const std::string& service_id) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Connects to a BLE peripheral.
   // On success, returns a new BleSocket.
   // On error, returns nullptr.
   std::unique_ptr<api::BleSocket> Connect(api::BlePeripheral& peripheral,
                                           const std::string& service_id,
-                                          CancellationFlag* cancellation_flag);
+                                          CancellationFlag* cancellation_flag)
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
   void PublisherHandler(
@@ -91,6 +99,7 @@ class BleMedium : public api::BleMedium {
                       ::winrt::Windows::Devices::Bluetooth::Advertisement::
                           BluetoothLEAdvertisementWatcherStoppedEventArgs args);
 
+  mutable absl::Mutex mutex_;
   BluetoothAdapter* adapter_;
   std::string service_id_;
 
@@ -98,9 +107,8 @@ class BleMedium : public api::BleMedium {
 
   // Map to protect the pointer for BlePeripheral because
   // DiscoveredPeripheralCallback only keeps the pointer to the object
-  absl::Mutex peripheral_map_mutex_;
   absl::flat_hash_map<std::string, std::unique_ptr<BlePeripheral>>
-      peripheral_map_ ABSL_GUARDED_BY(peripheral_map_mutex_);
+      peripheral_map_ ABSL_GUARDED_BY(mutex_);
 
   // WinRT objects
   ::winrt::Windows::Devices::Bluetooth::Advertisement::

internal/platform/implementation/windows/bluetooth_classic_medium.cc

@@ -26,6 +26,7 @@
 #include <string>
 #include <utility>
 
+#include "absl/synchronization/mutex.h"
 #include "internal/platform/cancellation_flag.h"
 #include "internal/platform/cancellation_flag_listener.h"
 #include "internal/platform/exception.h"
@@ -101,6 +102,8 @@ BluetoothClassicMedium::~BluetoothClassicMedium() {}
 
 void BluetoothClassicMedium::OnScanModeChanged(
     BluetoothAdapter::ScanMode scanMode) {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << __func__
                     << ": OnScanModeChanged is called with scanMode: "
                     << static_cast<int>(scanMode);
@@ -151,6 +154,8 @@ void BluetoothClassicMedium::OnScanModeChanged(
 
 bool BluetoothClassicMedium::StartDiscovery(
     BluetoothClassicMedium::DiscoveryCallback discovery_callback) {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << "StartDiscovery is called.";
 
   bool result = false;
@@ -164,6 +169,8 @@ bool BluetoothClassicMedium::StartDiscovery(
 }
 
 bool BluetoothClassicMedium::StopDiscovery() {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << "StopDiscovery is called.";
 
   bool result = false;
@@ -214,8 +221,11 @@ void BluetoothClassicMedium::InitializeDeviceWatcher() {
 std::unique_ptr<api::BluetoothSocket> BluetoothClassicMedium::ConnectToService(
     api::BluetoothDevice& remote_device, const std::string& service_uuid,
     CancellationFlag* cancellation_flag) {
+  absl::MutexLock lock(&mutex_);
+
   try {
-    NEARBY_LOGS(INFO) << "ConnectToService is called.";
+    NEARBY_LOGS(INFO) << "ConnectToService is called. device:"
+                      << remote_device.GetName();
     if (service_uuid.empty()) {
       NEARBY_LOGS(ERROR) << __func__ << ": service_uuid not specified.";
       return nullptr;
@@ -318,9 +328,13 @@ std::unique_ptr<api::BluetoothSocket> BluetoothClassicMedium::ConnectToService(
         rfcomm_socket->Connect(requested_service.ConnectionHostName(),
                                requested_service.ConnectionServiceName());
     if (!success) {
+      NEARBY_LOGS(INFO) << "Failed to connect Bluetooth device:"
+                        << remote_device.GetName();
       return nullptr;
     }
 
+    NEARBY_LOGS(INFO) << "Connected to Bluetooth device:"
+                      << remote_device.GetName();
     return std::move(rfcomm_socket);
   } catch (std::exception exception) {
     // We will log and eat the exception since the caller
@@ -416,6 +430,8 @@ bool BluetoothClassicMedium::CheckSdp(RfcommDeviceService requestedService) {
 std::unique_ptr<api::BluetoothServerSocket>
 BluetoothClassicMedium::ListenForService(const std::string& service_name,
                                          const std::string& service_uuid) {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << "ListenForService is called with service name: "
                     << service_name << ".";
   if (service_uuid.empty()) {
@@ -450,6 +466,8 @@ BluetoothClassicMedium::ListenForService(const std::string& service_name,
 
 api::BluetoothDevice* BluetoothClassicMedium::GetRemoteDevice(
     const std::string& mac_address) {
+  absl::MutexLock lock(&mutex_);
+
   return new BluetoothDevice(mac_address);
 }
 
@@ -489,6 +507,8 @@ bool BluetoothClassicMedium::StopScanning() {
 
 winrt::fire_and_forget BluetoothClassicMedium::DeviceWatcher_Added(
     DeviceWatcher sender, DeviceInformation deviceInfo) {
+  absl::MutexLock lock(&mutex_);
+
   NEARBY_LOGS(INFO) << "Device added " << winrt::to_string(deviceInfo.Id());
   IMapView<winrt::hstring, IInspectable> properties = deviceInfo.Properties();
   DumpDeviceInformation(properties);
@@ -565,6 +585,8 @@ winrt::fire_and_forget BluetoothClassicMedium::DeviceWatcher_Added(
 
 winrt::fire_and_forget BluetoothClassicMedium::DeviceWatcher_Updated(
     DeviceWatcher sender, DeviceInformationUpdate deviceInfoUpdate) {
+  absl::MutexLock lock(&mutex_);
+
   auto it = discovered_devices_by_id_.find(deviceInfoUpdate.Id());
 
   if (it == discovered_devices_by_id_.end()) {
@@ -613,6 +635,8 @@ winrt::fire_and_forget BluetoothClassicMedium::DeviceWatcher_Updated(
 
 winrt::fire_and_forget BluetoothClassicMedium::DeviceWatcher_Removed(
     DeviceWatcher sender, DeviceInformationUpdate deviceInfo) {
+  absl::MutexLock lock(&mutex_);
+
   auto it = discovered_devices_by_id_.find(deviceInfo.Id());
 
   if (it == discovered_devices_by_id_.end()) {
@@ -702,7 +726,11 @@ bool BluetoothClassicMedium::StartAdvertising(bool radio_discoverable) {
       return false;
     }
 
-    server_socket_->SetCloseNotifier([&]() { StopAdvertising(); });
+    server_socket_->SetCloseNotifier([&]() {
+      absl::MutexLock lock(&mutex_);
+      NEARBY_LOGS(INFO) << __func__ << ": Server socket is closed.";
+      StopAdvertising();
+    });
 
     // Set the SDP attributes and start Bluetooth advertising
     InitializeServiceSdpAttributes(rfcomm_provider_, service_name_);

internal/platform/implementation/windows/bluetooth_classic_medium.h

@@ -19,6 +19,7 @@
 #include <memory>
 #include <string>
 
+#include "absl/base/thread_annotations.h"
 #include "internal/platform/implementation/bluetooth_classic.h"
 #include "internal/platform/implementation/windows/bluetooth_adapter.h"
 #include "internal/platform/implementation/windows/bluetooth_classic_device.h"
@@ -104,14 +105,15 @@ class BluetoothClassicMedium : public api::BluetoothClassicMedium {
   ~BluetoothClassicMedium() override;
 
   // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#startDiscovery()
-  bool StartDiscovery(DiscoveryCallback discovery_callback) override;
+  bool StartDiscovery(DiscoveryCallback discovery_callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#cancelDiscovery()
   //
   // Returns true once discovery is well and truly stopped; after this returns,
   // there must be no more invocations of the DiscoveryCallback passed in to
   // StartDiscovery().
-  bool StopDiscovery() override;
+  bool StopDiscovery() override ABSL_LOCKS_EXCLUDED(mutex_);
 
   // A combination of
   // https://developer.android.com/reference/android/bluetooth/BluetoothDevice.html#createInsecureRfcommSocketToServiceRecord
@@ -128,7 +130,7 @@ class BluetoothClassicMedium : public api::BluetoothClassicMedium {
   // On error, throw's an exception
   std::unique_ptr<api::BluetoothSocket> ConnectToService(
       api::BluetoothDevice& remote_device, const std::string& service_uuid,
-      CancellationFlag* cancellation_flag) override;
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
 
   // https://developer.android.com/reference/android/bluetooth/BluetoothAdapter.html#listenUsingInsecureRfcommWithServiceRecord
   //
@@ -140,17 +142,18 @@ class BluetoothClassicMedium : public api::BluetoothClassicMedium {
   //
   //  Returns nullptr error.
   std::unique_ptr<api::BluetoothServerSocket> ListenForService(
-      const std::string& service_name,
+      const std::string& service_name, const std::string& service_uuid) override
-      const std::string& service_uuid) override;
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
-  api::BluetoothDevice* GetRemoteDevice(
+  api::BluetoothDevice* GetRemoteDevice(const std::string& mac_address) override
-      const std::string& mac_address) override;
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
  private:
-  bool StartScanning();
+  bool StartScanning() ABSL_SHARED_LOCKS_REQUIRED(mutex_);
-  bool StopScanning();
+  bool StopScanning() ABSL_SHARED_LOCKS_REQUIRED(mutex_);
-  bool StartAdvertising(bool radio_discoverable);
+  bool StartAdvertising(bool radio_discoverable)
-  bool StopAdvertising();
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
+  bool StopAdvertising() ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   bool InitializeServiceSdpAttributes(RfcommServiceProvider rfcomm_provider,
                                       std::string service_name);
   bool IsWatcherStarted();
@@ -192,7 +195,7 @@ class BluetoothClassicMedium : public api::BluetoothClassicMedium {
   // hstring is the only type of string winrt understands.
   // https://docs.microsoft.com/en-us/uwp/cpp-ref-for-winrt/hstring
   std::map<winrt::hstring, std::unique_ptr<BluetoothDevice>>
-      discovered_devices_by_id_;
+      discovered_devices_by_id_ ABSL_GUARDED_BY(mutex_);
 
   BluetoothAdapter& bluetooth_adapter_;
 
@@ -204,6 +207,9 @@ class BluetoothClassicMedium : public api::BluetoothClassicMedium {
   std::unique_ptr<BluetoothServerSocket> server_socket_ = nullptr;
   BluetoothServerSocket* raw_server_socket_ = nullptr;
   bool is_radio_discoverable_ = false;
+
+  // Used to enable thread safe for APIs.
+  mutable absl::Mutex mutex_;
 };
 
 }  // namespace windows

internal/platform/implementation/windows/wifi_lan.h

@@ -223,33 +223,37 @@ class WifiLanMedium : public api::WifiLanMedium {
   ~WifiLanMedium() override = default;
 
   // Check if a network connection to a primary router exist.
-  bool IsNetworkConnected() const override;
+  bool IsNetworkConnected() const override ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Starts to advertising
-  bool StartAdvertising(const NsdServiceInfo& nsd_service_info) override;
+  bool StartAdvertising(const NsdServiceInfo& nsd_service_info) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Stops to advertising
-  bool StopAdvertising(const NsdServiceInfo& nsd_service_info) override;
+  bool StopAdvertising(const NsdServiceInfo& nsd_service_info) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Starts to discovery
   bool StartDiscovery(const std::string& service_type,
-                      DiscoveredServiceCallback callback) override;
+                      DiscoveredServiceCallback callback) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   // Returns true once WifiLan discovery for service_type is well and truly
   // stopped; after this returns, there must be no more invocations of the
   // DiscoveredServiceCallback passed in to StartDiscovery() for service_type.
-  bool StopDiscovery(const std::string& service_type) override;
+  bool StopDiscovery(const std::string& service_type) override
+      ABSL_LOCKS_EXCLUDED(mutex_);
 
   std::unique_ptr<api::WifiLanSocket> ConnectToService(
       const NsdServiceInfo& remote_service_info,
-      CancellationFlag* cancellation_flag) override;
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
 
   std::unique_ptr<api::WifiLanSocket> ConnectToService(
       const std::string& ip_address, int port,
-      CancellationFlag* cancellation_flag) override;
+      CancellationFlag* cancellation_flag) override ABSL_LOCKS_EXCLUDED(mutex_);
 
   std::unique_ptr<api::WifiLanServerSocket> ListenForService(
-      int port = 0) override;
+      int port = 0) override ABSL_LOCKS_EXCLUDED(mutex_);
 
   // DnsServiceDeRegister is a async process, after operation finish, callback
   // will call this method to notify the waiting method StopAdvertising to
@@ -290,14 +294,14 @@ class WifiLanMedium : public api::WifiLanMedium {
                               absl::Duration timeout = absl::Seconds(1));
 
   // Methods to manage discovred services.
-  void ClearDiscoveredServices() ABSL_LOCKS_EXCLUDED(mutex_);
+  void ClearDiscoveredServices() ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   std::optional<NsdServiceInfo> GetDiscoveredService(absl::string_view id)
-      ABSL_LOCKS_EXCLUDED(mutex_);
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   void UpdateDiscoveredService(absl::string_view id,
                                const NsdServiceInfo& nsd_service_info)
-      ABSL_LOCKS_EXCLUDED(mutex_);
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
   void RemoveDiscoveredService(absl::string_view id)
-      ABSL_LOCKS_EXCLUDED(mutex_);
+      ABSL_SHARED_LOCKS_REQUIRED(mutex_);
 
   // From mDNS device information, to build NsdServiceInfo.
   // the properties are from DeviceInformation and DeviceInformationUpdate.
@@ -321,6 +325,12 @@ class WifiLanMedium : public api::WifiLanMedium {
 
   void RestartScanning();
 
+  // Internal methods to avoid deadlock.
+  bool InternalStopAdvertising(const NsdServiceInfo& nsd_service_info);
+  std::unique_ptr<api::WifiLanSocket> InternalConnectToService(
+      const std::string& ip_address, int port,
+      CancellationFlag* cancellation_flag);
+
   //
   // Dns-sd related properties
   //
@@ -357,7 +367,7 @@ class WifiLanMedium : public api::WifiLanMedium {
       port_to_server_socket_map_;
 
   // Used to protect the access to mDNS instances and scanning related data.
-  absl::Mutex mutex_;
+  mutable absl::Mutex mutex_;
 
   // Keeps the map from device id to service during scanning.
   absl::flat_hash_map<std::string, NsdServiceInfo> discovered_services_map_

internal/platform/implementation/windows/wifi_lan_medium.cc

@@ -65,6 +65,8 @@ constexpr absl::Duration kConnectServiceTimeout = absl::Seconds(3);
 }  // namespace
 
 bool WifiLanMedium::IsNetworkConnected() const {
+  absl::MutexLock lock(&mutex_);
+
   // connection_profile will be null when there's no network adapter or
   // connection to a network. For example, WiFi isn't connected to an AP/hotspot
   // and ethernet isn't connected to a router/hub/switch.
@@ -73,6 +75,8 @@ bool WifiLanMedium::IsNetworkConnected() const {
 }
 
 bool WifiLanMedium::StartAdvertising(const NsdServiceInfo& nsd_service_info) {
+  absl::MutexLock lock(&mutex_);
+
   bool socket_found = false;
   WifiLanServerSocket* server_socket_ptr = nullptr;
   for (const auto& server_socket : port_to_server_socket_map_) {
@@ -155,7 +159,7 @@ bool WifiLanMedium::StartAdvertising(const NsdServiceInfo& nsd_service_info) {
     NEARBY_LOGS(WARNING) << "advertising instance name was changed due to have "
                             "same name instance was running.";
     // stop the service and return false
-    StopAdvertising(nsd_service_info);
+    InternalStopAdvertising(nsd_service_info);
     return false;
   }
 
@@ -194,6 +198,12 @@ void WifiLanMedium::NotifyDnsServiceUnregistered(DWORD status) {
 }
 
 bool WifiLanMedium::StopAdvertising(const NsdServiceInfo& nsd_service_info) {
+  absl::MutexLock lock(&mutex_);
+  return InternalStopAdvertising(nsd_service_info);
+}
+
+bool WifiLanMedium::InternalStopAdvertising(
+    const NsdServiceInfo& nsd_service_info) {
   // Need to use Win32 API to deregister the Dnssd instance
   if (!IsAdvertising()) {
     NEARBY_LOGS(WARNING)
@@ -252,6 +262,7 @@ bool WifiLanMedium::StopAdvertising(const NsdServiceInfo& nsd_service_info) {
 // Returns true once the WifiLan discovery has been initiated.
 bool WifiLanMedium::StartDiscovery(const std::string& service_type,
                                    DiscoveredServiceCallback callback) {
+  absl::MutexLock lock(&mutex_);
   if (IsDiscovering()) {
     NEARBY_LOGS(WARNING) << "discovery already running for service type ="
                          << service_type;
@@ -305,6 +316,8 @@ bool WifiLanMedium::StartDiscovery(const std::string& service_type,
 // stopped; after this returns, there must be no more invocations of the
 // DiscoveredServiceCallback passed in to StartDiscovery() for service_id.
 bool WifiLanMedium::StopDiscovery(const std::string& service_type) {
+  absl::MutexLock lock(&mutex_);
+
   if (!IsDiscovering()) {
     NEARBY_LOGS(WARNING) << "no discovering service to stop.";
     return false;
@@ -321,17 +334,30 @@ bool WifiLanMedium::StopDiscovery(const std::string& service_type) {
 std::unique_ptr<api::WifiLanSocket> WifiLanMedium::ConnectToService(
     const NsdServiceInfo& remote_service_info,
     CancellationFlag* cancellation_flag) {
-  NEARBY_LOGS(ERROR)
+  absl::MutexLock lock(&mutex_);
+  NEARBY_LOGS(INFO)
       << "connect to service by NSD service info. service type is "
       << remote_service_info.GetServiceType();
 
-  return ConnectToService(remote_service_info.GetIPAddress(),
+  return InternalConnectToService(remote_service_info.GetIPAddress(),
-                          remote_service_info.GetPort(), cancellation_flag);
+                                  remote_service_info.GetPort(),
+                                  cancellation_flag);
 }
 
 std::unique_ptr<api::WifiLanSocket> WifiLanMedium::ConnectToService(
     const std::string& ip_address, int port,
     CancellationFlag* cancellation_flag) {
+  absl::MutexLock lock(&mutex_);
+  NEARBY_LOGS(INFO) << "connect to service by NSD service info. ip_address:"
+                    << ipaddr_4bytes_to_dotdecimal_string(ip_address) << ":"
+                    << port;
+
+  return InternalConnectToService(ip_address, port, cancellation_flag);
+}
+
+std::unique_ptr<api::WifiLanSocket> WifiLanMedium::InternalConnectToService(
+    const std::string& ip_address, int port,
+    CancellationFlag* cancellation_flag) {
   NEARBY_LOGS(INFO) << "ConnectToService is called.";
   if (ip_address.empty() || ip_address.length() != 4 || port == 0) {
     NEARBY_LOGS(ERROR) << "no valid service address and port to connect.";
@@ -423,6 +449,7 @@ std::unique_ptr<api::WifiLanSocket> WifiLanMedium::ConnectToService(
 
 std::unique_ptr<api::WifiLanServerSocket> WifiLanMedium::ListenForService(
     int port) {
+  absl::MutexLock lock(&mutex_);
   // check current status
   const auto& it = port_to_server_socket_map_.find(port);
   if (it != port_to_server_socket_map_.end()) {
@@ -576,6 +603,8 @@ ExceptionOr<NsdServiceInfo> WifiLanMedium::GetNsdServiceInformation(
 
 fire_and_forget WifiLanMedium::Watcher_DeviceAdded(
     DeviceWatcher sender, DeviceInformation deviceInfo) {
+  absl::MutexLock lock(&mutex_);
+
   // need to read IP address and port information from deviceInfo
   ExceptionOr<NsdServiceInfo> nsd_service_info_except =
       GetNsdServiceInformation(deviceInfo.Properties(),
@@ -624,6 +653,8 @@ fire_and_forget WifiLanMedium::Watcher_DeviceAdded(
 
 fire_and_forget WifiLanMedium::Watcher_DeviceUpdated(
     DeviceWatcher sender, DeviceInformationUpdate deviceInfoUpdate) {
+  absl::MutexLock lock(&mutex_);
+
   ExceptionOr<NsdServiceInfo> nsd_service_info_except =
       GetNsdServiceInformation(deviceInfoUpdate.Properties(),
                                /*is_device_found*/ true);
@@ -687,6 +718,7 @@ fire_and_forget WifiLanMedium::Watcher_DeviceUpdated(
 
 fire_and_forget WifiLanMedium::Watcher_DeviceRemoved(
     DeviceWatcher sender, DeviceInformationUpdate deviceInfoUpdate) {
+  absl::MutexLock lock(&mutex_);
   // need to read IP address and port information from deviceInfo
   ExceptionOr<NsdServiceInfo> nsd_service_info_except =
       GetNsdServiceInformation(deviceInfoUpdate.Properties(),
@@ -715,13 +747,11 @@ fire_and_forget WifiLanMedium::Watcher_DeviceRemoved(
 }
 
 void WifiLanMedium::ClearDiscoveredServices() {
-  absl::MutexLock lock(&mutex_);
   discovered_services_map_.clear();
 }
 
 std::optional<NsdServiceInfo> WifiLanMedium::GetDiscoveredService(
     absl::string_view id) {
-  absl::MutexLock lock(&mutex_);
   auto it = discovered_services_map_.find(id);
   if (it == discovered_services_map_.end()) {
     return std::nullopt;
@@ -732,12 +762,10 @@ std::optional<NsdServiceInfo> WifiLanMedium::GetDiscoveredService(
 
 void WifiLanMedium::UpdateDiscoveredService(
     absl::string_view id, const NsdServiceInfo& nsd_service_info) {
-  absl::MutexLock lock(&mutex_);
   discovered_services_map_[id] = nsd_service_info;
 }
 
 void WifiLanMedium::RemoveDiscoveredService(absl::string_view id) {
-  absl::MutexLock lock(&mutex_);
   auto it = discovered_services_map_.find(id);
   if (it != discovered_services_map_.end()) {
     discovered_services_map_.erase(it);