From 25deb0962f286e853c049aa73fe2da976552d565 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 12 Jul 2024 16:51:00 -0700 Subject: [PATCH] Implement Bluetooth Multiplex. PiperOrigin-RevId: 651921709 --- .../implementation/base_endpoint_channel.h | 1 + .../implementation/base_pcp_handler.cc | 26 +- .../bluetooth_endpoint_channel.cc | 7 + .../bluetooth_endpoint_channel.h | 1 + connections/implementation/client_proxy.cc | 27 +- connections/implementation/client_proxy.h | 27 +- .../implementation/client_proxy_test.cc | 42 ++++ connections/implementation/endpoint_channel.h | 5 +- .../mediums/bluetooth_classic.cc | 136 ++++++++++- .../mediums/bluetooth_classic.h | 13 + .../implementation/mediums/multiplex/BUILD | 2 +- .../multiplex/multiplex_output_stream.cc | 109 ++++----- .../multiplex/multiplex_output_stream.h | 40 +-- .../mediums/multiplex/multiplex_socket.cc | 230 ++++++++++-------- .../mediums/multiplex/multiplex_socket.h | 38 +-- .../multiplex/multiplex_socket_test.cc | 55 +++-- .../mediums/webrtc/webrtc_socket_impl.cc | 5 +- .../mediums/webrtc/webrtc_socket_impl.h | 13 +- .../implementation/mediums/webrtc_socket.h | 3 +- .../implementation/p2p_cluster_pcp_handler.cc | 4 +- internal/platform/BUILD | 3 + internal/platform/array_blocking_queue.h | 104 ++++++++ internal/platform/blocking_queue_stream.cc | 72 ++++++ internal/platform/blocking_queue_stream.h | 52 ++++ internal/platform/bluetooth_classic.cc | 49 +++- internal/platform/bluetooth_classic.h | 77 +++++- internal/platform/feature_flags.h | 1 + internal/platform/socket.h | 17 +- 28 files changed, 853 insertions(+), 306 deletions(-) create mode 100644 internal/platform/array_blocking_queue.h create mode 100644 internal/platform/blocking_queue_stream.cc create mode 100644 internal/platform/blocking_queue_stream.h diff --git a/connections/implementation/base_endpoint_channel.h b/connections/implementation/base_endpoint_channel.h index 889cf4fa..3908de52 100644 --- a/connections/implementation/base_endpoint_channel.h +++ b/connections/implementation/base_endpoint_channel.h @@ -28,6 +28,7 @@ #include "internal/platform/input_stream.h" #include "internal/platform/mutex.h" #include "internal/platform/output_stream.h" +#include "internal/platform/socket.h" namespace nearby { namespace connections { diff --git a/connections/implementation/base_pcp_handler.cc b/connections/implementation/base_pcp_handler.cc index 14bdbe4d..779eaae2 100644 --- a/connections/implementation/base_pcp_handler.cc +++ b/connections/implementation/base_pcp_handler.cc @@ -71,6 +71,7 @@ #include "internal/platform/exception.h" #include "internal/platform/feature_flags.h" #include "internal/platform/future.h" +#include "internal/platform/implementation/system_clock.h" #include "internal/platform/logging.h" #include "internal/platform/mutex_lock.h" #include "internal/platform/prng.h" @@ -1513,6 +1514,11 @@ void BasePcpHandler::OnIncomingFrame( client->SetRemoteOsInfo(endpoint_id, connection_response.os_info()); } + if (connection_response.has_multiplex_socket_bitmask()) { + client->SetRemoteMultiplexSocketBitmask( + endpoint_id, connection_response.multiplex_socket_bitmask()); + } + if (connection_response.has_safe_to_disconnect_version()) { NEARBY_LOGS(INFO) << "[safe-to-disconnect]: endpoint_id=" << endpoint_id @@ -2125,8 +2131,8 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client, bool can_close_immediately) { // Short-circuit immediately if we're not in an actionable state yet. We will // be called again once the other side has made their decision. - if (!client->IsConnectionAccepted(endpoint_id) && - !client->IsConnectionRejected(endpoint_id)) { + bool is_connection_accepted = client->IsConnectionAccepted(endpoint_id); + if (!is_connection_accepted && !client->IsConnectionRejected(endpoint_id)) { if (!client->HasLocalEndpointResponded(endpoint_id)) { NEARBY_LOGS(INFO) << "ConnectionResult: local client did not respond; endpoint_id=" @@ -2150,7 +2156,8 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client, auto pair = pending_connections_.extract(it); BasePcpHandler::PendingConnectionInfo& connection_info = pair.mapped(); - bool is_connection_accepted = client->IsConnectionAccepted(endpoint_id); + Medium medium = + channel_manager_->GetChannelForEndpoint(endpoint_id)->GetMedium(); Status response_code; if (is_connection_accepted) { @@ -2173,6 +2180,17 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client, std::move(context))) { response_code = {Status::kEndpointUnknown}; } + + std::shared_ptr channel = + channel_manager_->GetChannelForEndpoint(endpoint_id); + if (channel != nullptr) { + if (!channel->EnableMultiplexSocket()) { + NEARBY_LOGS(INFO) + << "MultiplexSocket is not implemented for this channel."; + } + } else { + NEARBY_LOGS(INFO) << "channel is null"; + } } else { NEARBY_LOGS(INFO) << "Pending connection rejected; endpoint_id=" << endpoint_id; @@ -2202,8 +2220,6 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client, return; } - Medium medium = - channel_manager_->GetChannelForEndpoint(endpoint_id)->GetMedium(); client->GetAnalyticsRecorder().OnConnectionEstablished( endpoint_id, medium, connection_info.connection_token); diff --git a/connections/implementation/bluetooth_endpoint_channel.cc b/connections/implementation/bluetooth_endpoint_channel.cc index 04278cf9..2b1aa65e 100644 --- a/connections/implementation/bluetooth_endpoint_channel.cc +++ b/connections/implementation/bluetooth_endpoint_channel.cc @@ -62,5 +62,12 @@ void BluetoothEndpointChannel::CloseImpl() { } } +bool BluetoothEndpointChannel::EnableMultiplexSocket() { + NEARBY_LOGS(INFO) << "BluetoothEndpointChannel MultiplexSocket will be " + "enabled if the Bluetooth MultiplexSocket is valid"; + bluetooth_socket_.EnableMultiplexSocket(); + return true; +} + } // namespace connections } // namespace nearby diff --git a/connections/implementation/bluetooth_endpoint_channel.h b/connections/implementation/bluetooth_endpoint_channel.h index 3c44d71a..b2176388 100644 --- a/connections/implementation/bluetooth_endpoint_channel.h +++ b/connections/implementation/bluetooth_endpoint_channel.h @@ -33,6 +33,7 @@ class BluetoothEndpointChannel final : public BaseEndpointChannel { location::nearby::proto::connections::Medium GetMedium() const override; int GetMaxTransmitPacketSize() const override; + bool EnableMultiplexSocket() override; private: static constexpr int kDefaultBTMaxTransmitPacketSize = 1980; // 990 * 2 Bytes diff --git a/connections/implementation/client_proxy.cc b/connections/implementation/client_proxy.cc index e60810db..ec788d4a 100644 --- a/connections/implementation/client_proxy.cc +++ b/connections/implementation/client_proxy.cc @@ -757,15 +757,14 @@ bool ClientProxy::HasRemoteEndpointResponded( void ClientProxy::LocalEndpointAcceptedConnection( const std::string& endpoint_id, PayloadListener listener) { MutexLock lock(&mutex_); - if (HasLocalEndpointResponded(endpoint_id)) { NEARBY_LOGS(INFO) << "ClientProxy [Local Accepted]: local endpoint has responded; id=" << endpoint_id; return; } - AppendConnectionStatus(endpoint_id, Connection::kLocalEndpointAccepted); + NEARBY_LOGS(INFO) << "ClientProxy [Local Accepted]: id=" << endpoint_id; ConnectionPair* item = LookupConnection(endpoint_id); if (item != nullptr) { item->second = std::move(listener); @@ -1196,6 +1195,8 @@ std::int32_t ClientProxy::GetLocalMultiplexSocketBitmask() const { if (NearbyFlags::GetInstance().GetBoolFlag( config_package_nearby::nearby_connections_feature:: kEnableMultiplex)) { + NEARBY_LOGS(INFO) << "ClientProxy [GetLocalMultiplexSocketBitmask]: " + << kBtMultiplexEnabled; return kBtMultiplexEnabled; } return 0; @@ -1203,33 +1204,47 @@ std::int32_t ClientProxy::GetLocalMultiplexSocketBitmask() const { void ClientProxy::SetRemoteMultiplexSocketBitmask( absl::string_view endpoint_id, int remote_multiplex_socket_bitmask) { - MutexLock lock(&mutex_); ConnectionPair* item = LookupConnection(endpoint_id); if (item != nullptr) { item->first.remote_multiplex_socket_bitmask = remote_multiplex_socket_bitmask; + NEARBY_LOGS(INFO) << "ClientProxy [SetRemoteMultiplexSocketBitmask]: " + << remote_multiplex_socket_bitmask; + } +} + +bool ClientProxy::IsLocalMultiplexSocketSupported(Medium medium) { + int bitmask = GetLocalMultiplexSocketBitmask(); + switch (medium) { + case Medium::BLUETOOTH: + NEARBY_LOGS(INFO) << "ClientProxy [IsLocalMultiplexSocketSupported]: " + << (bitmask & kBtMultiplexEnabled); + return (bitmask & kBtMultiplexEnabled) != 0; + case Medium::WIFI_LAN: + return (bitmask & kWifiLanMultiplexEnabled) != 0; + default: + return false; } } std::optional ClientProxy::GetRemoteMultiplexSocketBitmask( absl::string_view endpoint_id) const { - MutexLock lock(&mutex_); const ConnectionPair* item = LookupConnection(endpoint_id); if (item != nullptr) { return item->first.remote_multiplex_socket_bitmask; } return std::nullopt; } + bool ClientProxy::IsMultiplexSocketSupported(absl::string_view endpoint_id, Medium medium) { - MutexLock lock(&mutex_); ConnectionPair* item = LookupConnection(endpoint_id); if (item == nullptr) { return false; } - int combined_result = GetLocalMultiplexSocketBitmask() & item->first.remote_multiplex_socket_bitmask; + switch (medium) { case Medium::BLUETOOTH: return (combined_result & kBtMultiplexEnabled) != 0; diff --git a/connections/implementation/client_proxy.h b/connections/implementation/client_proxy.h index 1d34c91c..a38fab6a 100644 --- a/connections/implementation/client_proxy.h +++ b/connections/implementation/client_proxy.h @@ -314,12 +314,27 @@ class ClientProxy final { // Sets the multiplex socket supports status for remote device. void SetRemoteMultiplexSocketBitmask(absl::string_view endpoint_id, int remote_multiplex_socket_bitmask); + // Returns true if the multiplex socket is supported for the given medium. + bool IsLocalMultiplexSocketSupported(Medium medium); + // Gets the multiplex socket supports status for remote device. std::optional GetRemoteMultiplexSocketBitmask( absl::string_view endpoint_id) const; // Returns true if the multiplex socket is supported for the given medium. bool IsMultiplexSocketSupported(absl::string_view endpoint_id, Medium medium); + /** Bitmask for bt multiplex connection support. */ + // Note. Deprecates the first and second bit of BT_MULTIPLEX_ENABLED and + // WIFI_LAN_MULTIPLEX_ENABLED and shift them to the third and the forth bit. + // The reason is we need to escape the (0, 1) bit which has been set in some + // devices without salt enabled. If accompany with the devices with salted + // enabled, the frames passed cannot be decrypted and the connection shall be + // failed. Please refer to b/295925531#comment#14 for the details. + enum MultiplexSocketBitmask : uint32_t { + kBtMultiplexEnabled = 1 << 2, + kWifiLanMultiplexEnabled = 1 << 3, + }; + private: struct Connection { // Status: may be either: @@ -487,18 +502,6 @@ class ClientProxy final { bool supports_safe_to_disconnect_; bool support_auto_reconnect_; std::int32_t local_safe_to_disconnect_version_; - - /** Bitmask for bt multiplex connection support. */ - // Note. Deprecates the first and second bit of BT_MULTIPLEX_ENABLED and - // WIFI_LAN_MULTIPLEX_ENABLED and shift them to the third and the forth bit. - // The reason is we need to escape the (0, 1) bit which has been set in some - // devices without salt enabled. If accompany with the devices with salted - // enabled, the frames passed cannot be decrypted and the connection shall be - // failed. Please refer to b/295925531#comment#14 for the details. - enum MultiplexSocketBitmask : uint32_t { - kBtMultiplexEnabled = 1 << 2, - kWifiLanMultiplexEnabled = 1 << 3, - }; }; } // namespace connections diff --git a/connections/implementation/client_proxy_test.cc b/connections/implementation/client_proxy_test.cc index 3db21bac..242af16a 100644 --- a/connections/implementation/client_proxy_test.cc +++ b/connections/implementation/client_proxy_test.cc @@ -1552,6 +1552,48 @@ TEST_F(ClientProxyTest, TestAutoBwuWhenListeningWithAutoBwu) { EXPECT_TRUE(client1()->AutoUpgradeBandwidth()); } +TEST_F(ClientProxyTest, TestMultiplexSocketBitmask) { + EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(), 0); + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kEnableMultiplex, + true); + EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(), + ClientProxy::kBtMultiplexEnabled); + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kEnableMultiplex, + false); +} + +TEST_F(ClientProxyTest, TestRemoteMultiplexSocketBitmask) { + EXPECT_EQ(client1()->GetLocalMultiplexSocketBitmask(), 0); + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kEnableMultiplex, + true); + Endpoint advertising_endpoint = + StartAdvertising(client1(), advertising_connection_listener_); + OnAdvertisingConnectionInitiated(client1(), advertising_endpoint); + client1()->SetRemoteMultiplexSocketBitmask( + advertising_endpoint.id, + ClientProxy::kBtMultiplexEnabled | ClientProxy::kWifiLanMultiplexEnabled); + ASSERT_TRUE(client1() + ->GetRemoteMultiplexSocketBitmask(advertising_endpoint.id) + .has_value()); + EXPECT_EQ( + client1() + ->GetRemoteMultiplexSocketBitmask(advertising_endpoint.id) + .value(), + ClientProxy::kBtMultiplexEnabled | ClientProxy::kWifiLanMultiplexEnabled); + EXPECT_TRUE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id, + Medium::BLUETOOTH)); + EXPECT_FALSE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id, + Medium::WIFI_LAN)); + EXPECT_FALSE(client1()->IsMultiplexSocketSupported(advertising_endpoint.id, + Medium::WIFI_AWARE)); + NearbyFlags::GetInstance().OverrideBoolFlagValue( + config_package_nearby::nearby_connections_feature::kEnableMultiplex, + false); +} + } // namespace } // namespace connections } // namespace nearby diff --git a/connections/implementation/endpoint_channel.h b/connections/implementation/endpoint_channel.h index f0fc3e67..2626866e 100644 --- a/connections/implementation/endpoint_channel.h +++ b/connections/implementation/endpoint_channel.h @@ -15,7 +15,6 @@ #ifndef CORE_INTERNAL_ENDPOINT_CHANNEL_H_ #define CORE_INTERNAL_ENDPOINT_CHANNEL_H_ -#include #include #include "securegcm/d2d_connection_context_v1.h" @@ -23,7 +22,6 @@ #include "connections/implementation/analytics/packet_meta_data.h" #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" -#include "internal/platform/mutex.h" namespace nearby { namespace connections { @@ -121,6 +119,9 @@ class EndpointChannel { virtual void SetAnalyticsRecorder( analytics::AnalyticsRecorder* analytics_recorder, const std::string& endpoint_id) = 0; + + // Enables the multiplex socket on the EndpointChannel. + virtual bool EnableMultiplexSocket() {return false;} }; inline bool operator==(const EndpointChannel& lhs, const EndpointChannel& rhs) { diff --git a/connections/implementation/mediums/bluetooth_classic.cc b/connections/implementation/mediums/bluetooth_classic.cc index 5a40633d..d020256e 100644 --- a/connections/implementation/mediums/bluetooth_classic.cc +++ b/connections/implementation/mediums/bluetooth_classic.cc @@ -18,12 +18,17 @@ #include #include +#include "connections/implementation/flags/nearby_connections_feature_flags.h" #include "connections/implementation/mediums/bluetooth_radio.h" +#include "connections/implementation/mediums/multiplex/multiplex_socket.h" +#include "connections/medium_selector.h" +#include "internal/flags/nearby_flags.h" #include "internal/platform/bluetooth_adapter.h" #include "internal/platform/bluetooth_classic.h" #include "internal/platform/cancellation_flag.h" #include "internal/platform/logging.h" #include "internal/platform/mutex_lock.h" +#include "internal/platform/socket.h" #include "internal/platform/uuid.h" namespace nearby { @@ -44,6 +49,8 @@ std::string ScanModeToString(BluetoothAdapter::ScanMode mode) { } } // namespace +using MultiplexSocket = mediums::multiplex::MultiplexSocket; + BluetoothClassic::BluetoothClassic(BluetoothRadio& radio) : BluetoothClassic(radio, std::make_unique( radio.GetBluetoothAdapter())) {} @@ -52,7 +59,10 @@ BluetoothClassic::BluetoothClassic( BluetoothRadio& radio, std::unique_ptr medium) : radio_(radio), adapter_(radio_.GetBluetoothAdapter()), - medium_(std::move(medium)) {} + medium_(std::move(medium)) { + is_multiplex_enabled_ = NearbyFlags::GetInstance().GetBoolFlag( + config_package_nearby::nearby_connections_feature::kEnableMultiplex); +} BluetoothClassic::~BluetoothClassic() { // Destructor is not taking locks, but methods it is calling are. @@ -62,6 +72,19 @@ BluetoothClassic::~BluetoothClassic() { } TurnOffDiscoverability(); + { + MutexLock lock(&mutex_); + NEARBY_LOGS(INFO) << "Closing multiplex sockets for " + << multiplex_sockets_.size() << " devices"; + if (is_multiplex_enabled_) { + for (auto& [bt_mac, multiplex_socket] : multiplex_sockets_) { + NEARBY_LOGS(INFO) << "Closing multiplex sockets for " + << GetRemoteDevice(bt_mac).GetName(); + multiplex_socket->Shutdown(); + } + } + } + // All the AcceptLoopRunnable objects in here should already have gotten an // opportunity to shut themselves down cleanly in the calls to // StopAcceptingConnections() above. @@ -348,20 +371,61 @@ bool BluetoothClassic::StartAcceptingConnections( auto owned_socket = server_sockets_.emplace(service_id, std::move(socket)).first->second; + if (is_multiplex_enabled_) { + MultiplexSocket::ListenForIncomingConnection( + service_id, Medium::BLUETOOTH, + [&callback](const std::string& listening_service_id, + MediumSocket* virtual_socket) mutable { + if (callback) { + callback(listening_service_id, + *(dynamic_cast(virtual_socket))); + } + }); + } // Start the accept loop on a dedicated thread - this stays alive and // listening for new incoming connections until StopAcceptingConnections() // is invoked. accept_loops_runner_.Execute( "bt-accept", [callback = std::move(callback), server_socket = std::move(owned_socket), - service_id]() mutable { + service_id, this]() mutable { while (true) { BluetoothSocket client_socket = server_socket.Accept(); if (!client_socket.IsValid()) { + NEARBY_LOGS(INFO) << "Failed to accept connection for " + << service_id; server_socket.Close(); break; } - if (callback) { + NEARBY_LOGS(INFO) << "Accepted connection for " << service_id; + bool callback_called = false; + { + MutexLock lock(&mutex_); + if (is_multiplex_enabled_) { + MultiplexSocket* multiplex_socket = + MultiplexSocket::CreateIncomingSocket(&client_socket, + service_id); + if (multiplex_socket != nullptr && + multiplex_socket->GetVirtualSocket(service_id)) { + multiplex_sockets_.emplace( + client_socket.GetRemoteDevice().GetMacAddress(), + multiplex_socket); + MultiplexSocket::StopListeningForIncomingConnection( + service_id, Medium::BLUETOOTH); + NEARBY_LOGS(INFO) << "Multiplex virtaul socket created for " + << client_socket.GetRemoteDevice().GetName(); + if (callback) { + callback( + service_id, + *(dynamic_cast( + multiplex_socket->GetVirtualSocket(service_id)))); + callback_called = true; + } + } + } + } + if (callback && !callback_called) { + NEARBY_LOGS(INFO) << "Call back triggered for physical socket."; callback(service_id, std::move(client_socket)); } } @@ -396,6 +460,10 @@ bool BluetoothClassic::StopAcceptingConnections(const std::string& service_id) { << service_id << " because it was never started."; return false; } + if (is_multiplex_enabled_) { + MultiplexSocket::StopListeningForIncomingConnection(service_id, + Medium::BLUETOOTH); + } // Closing the BluetoothServerSocket will kick off the suicide of the thread // in accept_loops_thread_pool_ that blocks on @@ -424,6 +492,31 @@ bool BluetoothClassic::StopAcceptingConnections(const std::string& service_id) { BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device, const std::string& service_id, CancellationFlag* cancellation_flag) { + { + MutexLock lock(&mutex_); + if (is_multiplex_enabled_) { + NEARBY_LOGS(INFO) << "multiplex_sockets_ size:" + << multiplex_sockets_.size(); + auto it = multiplex_sockets_.find(bluetooth_device.GetMacAddress()); + if (it != multiplex_sockets_.end()) { + MultiplexSocket* multiplex_socket = it->second; + if (multiplex_socket->IsEnabled()) { + auto* virtual_socket = + multiplex_socket->EstablishVirtualSocket(service_id); + // Should not happen. + auto* bluetooth_socket = + dynamic_cast(virtual_socket); + if (bluetooth_socket == nullptr) { + NEARBY_LOGS(INFO) + << "Failed to cast to BluetoothSocket for " << service_id + << " with " << bluetooth_device.GetName(); + return BluetoothSocket{}; + } + return *bluetooth_socket; + } + } + } + } service_id_to_connect_attempts_count_map_[service_id] = 1; while (service_id_to_connect_attempts_count_map_[service_id] <= kConnectAttemptsLimit) { @@ -432,14 +525,14 @@ BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device, << "Attempt #" << service_id_to_connect_attempts_count_map_[service_id] << ": Cannot start creating client BT socket due to cancel."; - return BluetoothSocket(); + return BluetoothSocket{}; } - NEARBY_LOGS(INFO) << "Attempt #" - << service_id_to_connect_attempts_count_map_[service_id] - << " to connect."; auto wrapper_result = AttemptToConnect(bluetooth_device, service_id, cancellation_flag); + NEARBY_LOGS(INFO) << "Attempt #" + << service_id_to_connect_attempts_count_map_[service_id] + << " to connect: " << wrapper_result.IsValid(); if (wrapper_result.IsValid()) { return wrapper_result; } @@ -449,7 +542,7 @@ BluetoothSocket BluetoothClassic::Connect(BluetoothDevice& bluetooth_device, NEARBY_LOGS(WARNING) << "Giving up after " << kConnectAttemptsLimit << " attempts"; - return BluetoothSocket(); + return BluetoothSocket{}; } BluetoothSocket BluetoothClassic::AttemptToConnect( @@ -458,9 +551,8 @@ BluetoothSocket BluetoothClassic::AttemptToConnect( MutexLock lock(&mutex_); NEARBY_LOGS(INFO) << "BluetoothClassic::Connect: service_id=" << service_id << ", device=" << &bluetooth_device; - // Socket to return. To allow for NRVO to work, it has to be a single - // object. - BluetoothSocket socket; + // Socket to return. To allow for NRVO to work, it has to be a single object. + BluetoothSocket socket{}; if (service_id.empty()) { NEARBY_LOGS(INFO) @@ -489,7 +581,27 @@ BluetoothSocket BluetoothClassic::AttemptToConnect( if (!socket.IsValid() || cancellation_flag->Cancelled()) { NEARBY_LOGS(INFO) << "Failed to Connect via BT [service=" << service_id << "]"; - return BluetoothSocket(); + return BluetoothSocket{}; + } + + if (is_multiplex_enabled_) { + // New MultiplexSocket but default disabled, should be enabled after + // negotiated + MultiplexSocket* multiplex_socket = + MultiplexSocket::CreateOutgoingSocket(&socket, service_id); + auto* virtual_socket = multiplex_socket->GetVirtualSocket(service_id); + // Should not happen. + auto* bluetooth_socket = dynamic_cast(virtual_socket); + if (bluetooth_socket == nullptr) { + NEARBY_LOGS(INFO) << "Failed to cast to BluetoothSocket for " + << service_id << " with " << bluetooth_device.GetName(); + return BluetoothSocket{}; + } + NEARBY_LOGS(INFO) << "Multiplex socket created for " + << bluetooth_device.GetName(); + multiplex_sockets_.emplace(bluetooth_device.GetMacAddress(), + multiplex_socket); + return *bluetooth_socket; } return socket; diff --git a/connections/implementation/mediums/bluetooth_classic.h b/connections/implementation/mediums/bluetooth_classic.h index 8c3de4e6..dd60c035 100644 --- a/connections/implementation/mediums/bluetooth_classic.h +++ b/connections/implementation/mediums/bluetooth_classic.h @@ -22,7 +22,10 @@ #include "absl/base/thread_annotations.h" #include "absl/container/flat_hash_map.h" #include "absl/functional/any_invocable.h" +#include "connections/implementation/flags/nearby_connections_feature_flags.h" #include "connections/implementation/mediums/bluetooth_radio.h" +#include "connections/implementation/mediums/multiplex/multiplex_socket.h" +#include "internal/flags/nearby_flags.h" #include "internal/platform/bluetooth_adapter.h" #include "internal/platform/bluetooth_classic.h" #include "internal/platform/cancellation_flag.h" @@ -228,6 +231,16 @@ class BluetoothClassic { mutable Mutex discovery_callbacks_mutex_; absl::flat_hash_map discovery_callbacks_ ABSL_GUARDED_BY(discovery_callbacks_mutex_); + + // Whether the multiplex feature is enabled. + bool is_multiplex_enabled_ = NearbyFlags::GetInstance().GetBoolFlag( + config_package_nearby::nearby_connections_feature:: + kEnableMultiplex); + + // A map of Bluetooth MacAddress -> MultiplexSocket. + absl::flat_hash_map + multiplex_sockets_ ABSL_GUARDED_BY(mutex_); }; } // namespace connections diff --git a/connections/implementation/mediums/multiplex/BUILD b/connections/implementation/mediums/multiplex/BUILD index e7209e7e..65e57ced 100644 --- a/connections/implementation/mediums/multiplex/BUILD +++ b/connections/implementation/mediums/multiplex/BUILD @@ -30,6 +30,7 @@ cc_library( "//connections/implementation:__subpackages__", ], deps = [ + "//connections:core_types", "//connections/implementation/flags:connections_flags", "//connections/implementation/mediums:utils", "//internal/flags:nearby_flags", @@ -42,7 +43,6 @@ cc_library( "//internal/platform/implementation:types", "//proto:connections_enums_cc_proto", "//proto/mediums:multiplex_frames_cc_proto", - "@aappleby_smhasher//:libmurmur3", "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/container:flat_hash_map", "@com_google_absl//absl/container:flat_hash_set", diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream.cc b/connections/implementation/mediums/multiplex/multiplex_output_stream.cc index 80c51732..90cb6c67 100644 --- a/connections/implementation/mediums/multiplex/multiplex_output_stream.cc +++ b/connections/implementation/mediums/multiplex/multiplex_output_stream.cc @@ -20,7 +20,9 @@ #include #include "absl/strings/string_view.h" +#include "absl/time/time.h" #include "connections/implementation/mediums/multiplex/multiplex_frames.h" +#include "internal/platform/array_blocking_queue.h" #include "internal/platform/atomic_boolean.h" #include "internal/platform/base64_utils.h" #include "internal/platform/byte_array.h" @@ -38,56 +40,9 @@ namespace multiplex { namespace { using ::location::nearby::mediums::ConnectionResponseFrame; -constexpr absl::string_view TAG = "MultiplexOutputStream:"; constexpr absl::string_view kFakeSalt = "RECEIVER_CONDIMENT"; } // namespace -// Implementation for class ArrayBlockingQueue -template -void ArrayBlockingQueue::Put(const T& value) { - MutexLock lock(&queue_mutex_); - if (queue_.size() >= capacity_) { - has_space_.Wait(); - } - queue_.push(value); - has_data_.Notify(); -} - -template -T ArrayBlockingQueue::Take() { - MutexLock lock(&queue_mutex_); - if (queue_.empty()) { - has_data_.Wait(); - } - T front = queue_.front(); - queue_.pop(); - has_space_.Notify(); - return front; -} - -template -bool ArrayBlockingQueue::TryPut(const T& value) { - MutexLock lock(&queue_mutex_); - if (queue_.size() < capacity_) { - queue_.push(value); - has_data_.Notify(); - return true; - } - return false; -} - -template -std::optional ArrayBlockingQueue::TryTake() { - MutexLock lock(&queue_mutex_); - if (!queue_.empty()) { - T front = queue_.front(); - queue_.pop(); - has_space_.Notify(); - return front; - } - return std::nullopt; -} - // Implementation for class MultiplexOutputStream MultiplexOutputStream::MultiplexOutputStream(OutputStream* physical_writer, AtomicBoolean& is_enabled) @@ -98,25 +53,25 @@ MultiplexOutputStream::MultiplexOutputStream(OutputStream* physical_writer, Exception MultiplexOutputStream::WaitForResult(const std::string& method_name, Future* future) { if (!future) { - NEARBY_LOGS(INFO) << TAG << "No future to wait for; return with error."; + NEARBY_LOGS(INFO) << "No future to wait for; return with error."; return {Exception::kFailed}; } - NEARBY_LOGS(INFO) << TAG << "Waiting for future to complete: " << method_name; + NEARBY_LOGS(INFO) << "Waiting for future to complete: " << method_name; ExceptionOr result = future->Get(FeatureFlags::GetInstance() .GetFlags() .mediums_frame_write_timeout_millis); if (!result.ok()) { - NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name + NEARBY_LOGS(INFO) << "Future:[" << method_name << "] completed with exception:" << result.exception(); return {Exception::kFailed}; } if (result.result()) { - NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name + NEARBY_LOGS(INFO) << "Future:[" << method_name << "] completed with success."; return {Exception::kSuccess}; } - NEARBY_LOGS(INFO) << TAG << "Future:[" << method_name + NEARBY_LOGS(INFO) << "Future:[" << method_name << "] completed with failure."; return {Exception::kFailed}; } @@ -156,7 +111,7 @@ bool MultiplexOutputStream::WriteConnectionResponseFrame( bool MultiplexOutputStream::Close(const std::string& service_id) { auto item = virtual_output_streams_.find(service_id); if (item == virtual_output_streams_.end()) { - NEARBY_LOGS(WARNING) << TAG << "Failed to close VirtualOutputStream(" + NEARBY_LOGS(INFO) << "Don't need to close VirtualOutputStream(" << service_id << ") because it's already gone."; return false; } @@ -223,7 +178,7 @@ MultiplexOutputStream::MultiplexWriter::MultiplexWriter( MultiplexOutputStream::MultiplexWriter::~MultiplexWriter() { Close(); - writer_thread_.Shutdown(); + // writer_thread_.Shutdown(); physical_writer_ = nullptr; } @@ -245,7 +200,7 @@ void MultiplexOutputStream::MultiplexWriter::EnqueueToSend( } void MultiplexOutputStream::MultiplexWriter::StartWriting() { - NEARBY_LOGS(INFO) << TAG << "Writing loop started."; + NEARBY_LOGS(INFO) << "Writing loop started."; while (true) { auto enqueued_frame = data_queue_.TryTake(); if (enqueued_frame != std::nullopt) { @@ -256,19 +211,23 @@ void MultiplexOutputStream::MultiplexWriter::StartWriting() { MutexLock lock(&writing_mutex_); if (data_queue_.Empty() && is_writing_ && !is_closed_) { is_writing_ = false; - NEARBY_LOGS(INFO) << TAG << "Waiting for data_queue_ has data."; + NEARBY_LOGS(INFO) << "Waiting for data_queue_ has data."; Exception wait_succeeded = is_writing_cond_.Wait(); if (!wait_succeeded.Ok()) { NEARBY_LOGS(WARNING) - << TAG << __func__ - << ": Failure waiting to wait: " << wait_succeeded.value; + << "Failure waiting to wait: " << wait_succeeded.value; return; } } - if (is_closed_) break; + if (is_closed_) { + NEARBY_LOGS(INFO) << "Notify to close_writing_thread"; + MutexLock lock(&close_writing_thread_mutex_); + close_writing_thread_cond_.Notify(); + break; + } } } - NEARBY_LOGS(INFO) << TAG << "Writing loop stopped."; + NEARBY_LOGS(INFO) << "Writing loop stopped."; } void MultiplexOutputStream::MultiplexWriter::Write( @@ -292,13 +251,28 @@ void MultiplexOutputStream::MultiplexWriter::Write( } void MultiplexOutputStream::MultiplexWriter::Close() { - MutexLock lock(&writing_mutex_); - is_closed_ = true; - if (is_write_loop_running_) { - NEARBY_LOGS(INFO) << TAG << "Stop writing loop and Shutdown writer thread."; + if (is_closed_) { + NEARBY_LOGS(INFO) << "MultiplexWriter is already closed."; + return; + } + NEARBY_LOGS(INFO) << "Stop writing loop and Shutdown writer thread."; + { + MutexLock lock(&writing_mutex_); + is_closed_ = true; + if (!is_write_loop_running_) { + writer_thread_.Shutdown(); + return; + } is_write_loop_running_ = false; is_writing_cond_.Notify(); } + NEARBY_LOGS(INFO) << "Wait to close_writing_thread"; + { + MutexLock lock(&close_writing_thread_mutex_); + close_writing_thread_cond_.Wait(absl::Milliseconds(20)); + NEARBY_LOGS(INFO) << "Shutdown writer thread."; + writer_thread_.Shutdown(); + } } MultiplexOutputStream::VirtualOutputStream::VirtualOutputStream( @@ -315,9 +289,9 @@ MultiplexOutputStream::VirtualOutputStream::VirtualOutputStream( Exception MultiplexOutputStream::VirtualOutputStream::Write( const ByteArray& data) { - if (is_closed_) { + if (is_closed_.Get()) { NEARBY_LOGS(WARNING) - << TAG << "Failed to write data because the VirtualOutputStream for " + << "Failed to write data because the VirtualOutputStream for " << service_id_ << " closed"; return {Exception::kIo}; } @@ -336,8 +310,7 @@ Exception MultiplexOutputStream::VirtualOutputStream::Write( // true to let the remote handle correctly. if ((service_id_hash_salt_ == kFakeSalt) && !should_pass_salt) { should_pass_salt = true; - NEARBY_LOGS(INFO) << TAG - << "service_idHashSalt is still a fake one and " + NEARBY_LOGS(INFO) << "service_idHashSalt is still a fake one and " "not changed yet; continue to pass salt."; } } diff --git a/connections/implementation/mediums/multiplex/multiplex_output_stream.h b/connections/implementation/mediums/multiplex/multiplex_output_stream.h index ea16394f..cc41073f 100644 --- a/connections/implementation/mediums/multiplex/multiplex_output_stream.h +++ b/connections/implementation/mediums/multiplex/multiplex_output_stream.h @@ -15,14 +15,12 @@ #ifndef CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_ #define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_OUTPUT_STREAM_H_ -#include #include -#include -#include #include #include "absl/base/thread_annotations.h" #include "absl/container/flat_hash_map.h" +#include "internal/platform/array_blocking_queue.h" #include "internal/platform/atomic_boolean.h" #include "internal/platform/byte_array.h" #include "internal/platform/exception.h" @@ -37,40 +35,6 @@ namespace nearby { namespace connections { namespace mediums { namespace multiplex { - -/** - * Payload from different services/clients will be put into an - * ArrayBlockingQueue before sending to ensure each client has equal chance to - * send its data. Since C++ doesn't provide ArrayBlockingQueue as Java, we - * implement one here. - */ -template -class ArrayBlockingQueue { - public: - explicit ArrayBlockingQueue(size_t capacity) : capacity_(capacity) {} - void Put(const T& value); - T Take(); - bool TryPut(const T& value); - // Returns std::nullopt if the queue is empty. - std::optional TryTake(); - - size_t Size() const { - MutexLock lock(&queue_mutex_); - return queue_.size(); - } - bool Empty() const { - MutexLock lock(&queue_mutex_); - return queue_.empty(); - } - - private: - std::queue queue_; - mutable Mutex queue_mutex_; - ConditionVariable has_data_{&queue_mutex_}; - ConditionVariable has_space_{&queue_mutex_}; - const size_t capacity_; -}; - /** * A helper class to send out the {@code MultiplexControlFrame} and the outgoing * data from clients. It schedules control and data frames with priority below @@ -173,6 +137,8 @@ class MultiplexOutputStream { ConditionVariable is_writing_cond_{&writing_mutex_}; bool is_writing_ ABSL_GUARDED_BY(writing_mutex_) = false; bool is_closed_ = false; + mutable Mutex close_writing_thread_mutex_; + ConditionVariable close_writing_thread_cond_{&close_writing_thread_mutex_}; // The single thread to write all enqueued frames. SingleThreadExecutor writer_thread_; diff --git a/connections/implementation/mediums/multiplex/multiplex_socket.cc b/connections/implementation/mediums/multiplex/multiplex_socket.cc index 6d3a7a2e..50ad72a1 100644 --- a/connections/implementation/mediums/multiplex/multiplex_socket.cc +++ b/connections/implementation/mediums/multiplex/multiplex_socket.cc @@ -23,11 +23,14 @@ #include "absl/container/flat_hash_map.h" #include "absl/functional/any_invocable.h" #include "absl/strings/string_view.h" +#include "absl/time/time.h" #include "connections/implementation/mediums/multiplex/multiplex_frames.h" #include "connections/implementation/mediums/multiplex/multiplex_output_stream.h" #include "connections/implementation/mediums/utils.h" #include "internal/platform/base64_utils.h" +#include "internal/platform/bluetooth_classic.h" #include "internal/platform/byte_array.h" +#include "internal/platform/count_down_latch.h" #include "internal/platform/exception.h" #include "internal/platform/feature_flags.h" #include "internal/platform/future.h" @@ -74,7 +77,20 @@ void MultiplexSocket::StopListeningForIncomingConnection( MultiplexSocket::MultiplexSocket(MediumSocket* physical_socket) : physical_socket_(physical_socket), multiplex_output_stream_{&physical_socket->GetOutputStream(), enabled_}, - physical_reader_(&physical_socket->GetInputStream()) {} + physical_reader_(&physical_socket->GetInputStream()) { + NEARBY_LOGS(INFO) << "physical_socket_: " << physical_socket_; + switch (physical_socket_->GetMedium()) { + case Medium::BLUETOOTH: + medium_ = Medium::BLUETOOTH; + bluetooth_socket_ = + std::move(*static_cast(physical_socket)); + break; + default: + medium_ = Medium::UNKNOWN_MEDIUM; + NEARBY_LOGS(ERROR) << __func__ << "Unsupported medium: " + << physical_socket_->GetMedium(); + } +} absl::flat_hash_map, MultiplexIncomingConnectionCb>& @@ -104,6 +120,7 @@ MultiplexSocket* MultiplexSocket::CreateIncomingSocket( storage_bt; multiplex_incoming_socket = new (&storage_bt) MultiplexSocket(physical_socket); + break; case Medium::BLE: static std::aligned_storage_t>( - []() { multiplex_incoming_socket->OnPhysicalSocketClosed(); }); - physical_socket->AddOnSocketClosedListener( - std::move(on_physical_socket_closed_listener)); - NEARBY_LOGS(INFO) << __func__ - << "CreateIncomingSocket with serviceId=" << service_id + NEARBY_LOGS(INFO) << "CreateIncomingSocket with serviceId=" << service_id << ", serviceIdHashSalt=" << kFakeSalt; + multiplex_incoming_socket->CreateFirstVirtualSocket(service_id, (std::string)kFakeSalt); multiplex_incoming_socket->StartReaderThread(); @@ -172,17 +183,9 @@ MultiplexSocket* MultiplexSocket::CreateOutgoingSocket( << physical_socket->GetMedium(); return multiplex_outgoing_socket; } - - auto on_physical_socket_closed_listener = - std::make_unique>( - []() { multiplex_outgoing_socket->OnPhysicalSocketClosed(); }); - physical_socket->AddOnSocketClosedListener( - std::move(on_physical_socket_closed_listener)); - NEARBY_LOGS(INFO) << __func__ - << "CreateOutgoingSocket with serviceId=" << service_id + NEARBY_LOGS(INFO) << "CreateOutgoingSocket with serviceId=" << service_id << ", serviceIdHashSalt=" << service_id_hash_salt; - NEARBY_LOGS(INFO) << __func__ << "multiplex_outgoing_socket:" - << multiplex_outgoing_socket; + multiplex_outgoing_socket->CreateFirstVirtualSocket(service_id, service_id_hash_salt); multiplex_outgoing_socket->StartReaderThread(); @@ -204,16 +207,20 @@ MediumSocket* MultiplexSocket::CreateFirstVirtualSocket( MutexLock lock(&virtual_socket_mutex_); std::string salted_service_id_hash_key = GenerateServiceIdHashKeyWithSalt(service_id, service_id_hash_salt); - + NEARBY_LOGS(INFO) << __func__ << " for service_id=" << service_id + << ", salt=" << service_id_hash_salt + << ", salted_service_id_hash_key=" + << salted_service_id_hash_key; MediumSocket* virtual_socket = physical_socket_->CreateVirtualSocket( salted_service_id_hash_key, output_stream, physical_socket_->GetMedium(), &virtual_sockets_); virtual_socket->AddOnSocketClosedListener( std::make_unique>( - [this, &service_id]() { OnVirtualSocketClosed(service_id); })); + [this, service_id]() { OnVirtualSocketClosed(service_id); })); if (!IsEnabled()) { + NEARBY_LOGS(INFO) << __func__ << ": Register multiplex enabled callback"; virtual_socket->RegisterMultiplexEnabledCallback(enable_cb_); } @@ -228,22 +235,31 @@ MediumSocket* MultiplexSocket::CreateVirtualSocket( std::string salted_service_id_hash_key = GenerateServiceIdHashKeyWithSalt(service_id, service_id_hash_salt); + NEARBY_LOGS(INFO) << __func__ << "service_id=" << service_id + << ", salt=" << service_id_hash_salt + << ", salted_service_id_hash_key=" + << salted_service_id_hash_key; + MediumSocket* virtual_socket = physical_socket_->CreateVirtualSocket( salted_service_id_hash_key, output_stream, physical_socket_->GetMedium(), &virtual_sockets_); virtual_socket->AddOnSocketClosedListener( std::make_unique>( - [this, &service_id]() { OnVirtualSocketClosed(service_id); })); + [this, service_id]() { OnVirtualSocketClosed(service_id); })); return virtual_socket; } MediumSocket* MultiplexSocket::GetVirtualSocket(const std::string& service_id) { MutexLock lock(&virtual_socket_mutex_); + NEARBY_LOGS(INFO) << __func__ << " service_id=" << service_id << ", Salt=" + << multiplex_output_stream_.GetServiceIdHashSalt(service_id) + << ", virtual_sockets_.size()=" << virtual_sockets_.size(); auto item = virtual_sockets_.find(GenerateServiceIdHashKeyWithSalt( service_id, multiplex_output_stream_.GetServiceIdHashSalt(service_id))); if (item == virtual_sockets_.end()) { + NEARBY_LOGS(INFO) << "Not found!"; return nullptr; } return item->second.get(); @@ -254,6 +270,16 @@ int MultiplexSocket::GetVirtualSocketCount() { return virtual_sockets_.size(); } +void MultiplexSocket::ListVirtualSocket() { + NEARBY_LOGS(INFO) << __func__ << " virtual_sockets_.size()=" + << virtual_sockets_.size(); + for (auto& [service_id_hash_key, virtual_socket] : virtual_sockets_) { + NEARBY_LOGS(INFO) << __func__ + << " service_id_hash_key=" << service_id_hash_key + << ", virtual_socket=" << virtual_socket; + } +} + std::shared_ptr> MultiplexSocket::RegisterConnectionResponse(const std::string& service_id) { auto future = std::make_shared>(); @@ -270,7 +296,8 @@ void MultiplexSocket::UnRegisterConnectionResponse( MediumSocket* MultiplexSocket::EstablishVirtualSocket( const std::string& service_id) { if (!IsEnabled()) { - NEARBY_LOGS(ERROR) << __func__ << "EstablishVirtualSocket disabled"; + NEARBY_LOGS(ERROR) + << "MultiplexSocket is disabled, cannot establish virtual socket."; return nullptr; } @@ -293,22 +320,19 @@ MediumSocket* MultiplexSocket::EstablishVirtualSocket( ConnectionResponseCode response_code = result.GetResult(); switch (response_code) { case ConnectionResponseFrame::CONNECTION_ACCEPTED: - NEARBY_LOGS(INFO) << __func__ - << "EstablishVirtualSocket after remote response to" + NEARBY_LOGS(INFO) << "EstablishVirtualSocket after remote response to" " accept the connection with service_id=" << service_id << ", service_id_hash_salt=" << service_id_hash_salt; return CreateVirtualSocket(service_id, service_id_hash_salt); case ConnectionResponseFrame::NOT_LISTENING: - NEARBY_LOGS(ERROR) << __func__ - << "EstablishVirtualSocket failed for service_id=" + NEARBY_LOGS(ERROR) << "EstablishVirtualSocket failed for service_id=" << service_id << ", service_id_hash_salt=" << service_id_hash_salt << " with response code=NOT_LISTENING"; break; default: - NEARBY_LOGS(ERROR) << __func__ - << "EstablishVirtualSocket failed for service_id=" + NEARBY_LOGS(ERROR) << "EstablishVirtualSocket failed for service_id=" << service_id << ", service_id_hash_salt=" << service_id_hash_salt << " with response code=UNKNOWN_RESPONSE_CODE"; @@ -319,8 +343,7 @@ MediumSocket* MultiplexSocket::EstablishVirtualSocket( void MultiplexSocket::StartReaderThread() { if (is_shutdown_) { - NEARBY_LOGS(WARNING) << __func__ - << "Stop to start reader thread since socket is " + NEARBY_LOGS(WARNING) << "Stop to start reader thread since socket is " "shutdown."; return; } @@ -358,21 +381,9 @@ void MultiplexSocket::StartReaderThread() { } } if (fail) { - { - MutexLock lock(&virtual_socket_mutex_); - if (virtual_sockets_.empty()) { - NEARBY_LOGS(INFO) - << __func__ - << "The reader thread stopped because all virtual socket " - "closed."; - } else { - NEARBY_LOGS(ERROR) << __func__ - << "The reader thread stopped because " - "unexpected IOException"; - } - } return; } + ExceptionOr frame_exc = multiplex::FromBytes(bytes.result()); if (!frame_exc.ok()) { @@ -388,11 +399,11 @@ void MultiplexSocket::StartReaderThread() { // the feature at this point. NEARBY_LOGS(INFO) << __func__ - << "Received a multiplex frame while not enabled, enable " + << " Received a multiplex frame while not enabled, enable " "multiplex."; Enable(); } - auto frame = frame_exc.result(); + const auto& frame = frame_exc.result(); auto salted_service_id_hash = ByteArray{std::move(frame.header().salted_service_id_hash())}; auto service_id_hash_salt = frame.header().has_service_id_hash_salt() @@ -404,12 +415,14 @@ void MultiplexSocket::StartReaderThread() { frame.control_frame()); break; case MultiplexFrame::DATA_FRAME: + NEARBY_LOGS(VERBOSE) + << "service_id_hash_salt: " << service_id_hash_salt; HandleDataFrame(salted_service_id_hash, service_id_hash_salt, frame.data_frame()); break; default: NEARBY_LOGS(WARNING) - << __func__ << "Received MultiplexFrame with unknown frame type " + << __func__ << " Received MultiplexFrame with unknown frame type " << frame.frame_type(); } } @@ -417,7 +430,6 @@ void MultiplexSocket::StartReaderThread() { } void MultiplexSocket::HandleOfflineFrame(const ByteArray& bytes) { - // Only pass the data when there's only 1 VirtualSocket. MutexLock lock(&virtual_socket_mutex_); NEARBY_LOGS(INFO) << __func__ << " Virtual_socket num:" << virtual_sockets_.size(); @@ -427,6 +439,7 @@ void MultiplexSocket::HandleOfflineFrame(const ByteArray& bytes) { NEARBY_LOGS(WARNING) << "Expected one live socket, but found null."; return; } + NEARBY_LOGS(INFO) << __func__ << "FeedIncomingData:" << std::string(bytes); item->second->FeedIncomingData(Base64Utils::IntToBytes(bytes.size())); item->second->FeedIncomingData(bytes); } @@ -474,8 +487,7 @@ void MultiplexSocket::HandleConnectionRequest( const ByteArray& salted_service_id_hash, const std::string& service_id_hash_salt) { if (!IsEnabled()) { - NEARBY_LOGS(WARNING) << __func__ - << "Received a CONNECTION_REQUEST frame on medium " + NEARBY_LOGS(WARNING) << "Received a CONNECTION_REQUEST frame on medium " << Medium_Name(physical_socket_->GetMedium()) << " but status is disabled, ignore it."; return; @@ -496,14 +508,13 @@ void MultiplexSocket::HandleConnectionRequest( } if (incoming_connection_callback == nullptr || listening_service_id.empty()) { - NEARBY_LOGS(INFO) << __func__ - << "There's no client listening for hash salt : " + NEARBY_LOGS(INFO) << "There's no client listening for hash salt : " << service_id_hash_salt << ", hash key : " << salted_service_id_hash_key << " on medium " << Medium_Name(physical_socket_->GetMedium()); - NEARBY_LOGS(INFO) << __func__ << "Dump incomingConnectionCallbacks : " + NEARBY_LOGS(INFO) << "The size of incomingConnectionCallbacks : " << GetIncomingConnectionCallbacks().size(); if (!multiplex_output_stream_.WriteConnectionResponseFrame( salted_service_id_hash, service_id_hash_salt, @@ -512,8 +523,7 @@ void MultiplexSocket::HandleConnectionRequest( } return; } - NEARBY_LOGS(INFO) << __func__ - << "Accept new virtual socket request service ID : " + NEARBY_LOGS(INFO) << "Accept new virtual socket request service ID : " << listening_service_id << ", hash salt : " << service_id_hash_salt << ", hash key : " << salted_service_id_hash_key @@ -523,14 +533,12 @@ void MultiplexSocket::HandleConnectionRequest( if (!multiplex_output_stream_.WriteConnectionResponseFrame( salted_service_id_hash, service_id_hash_salt, ConnectionResponseFrame::CONNECTION_ACCEPTED)) { - NEARBY_LOGS(INFO) << __func__ - << "Failed to write CONNECTION_ACCEPTED frame."; + NEARBY_LOGS(INFO) << "Failed to write CONNECTION_ACCEPTED frame."; return; } NEARBY_LOGS(VERBOSE) - << __func__ - << "establishVirtualSocket after local device accept the connection " + << "EstablishVirtualSocket after local device accept the connection " "with serviceId=" << listening_service_id << ", serviceIdHashSalt=" << service_id_hash_salt; MediumSocket* virtual_socket = @@ -576,25 +584,11 @@ void MultiplexSocket::HandleDisconnection( auto item = virtual_sockets_.find(salted_service_id_hash_key); if (item != virtual_sockets_.end()) { NEARBY_LOGS(INFO) - << __func__ << "Received a DISCONNECTION frame to disconnect virtual socket for " "salted service ID Hash Key " << salted_service_id_hash_key; - if (item->second != nullptr) { - item->second->Close(); - } - virtual_sockets_.erase(item); - // physical_socket_->RemoveVirtualSocket(salted_service_id_hash_key); - - if (virtual_sockets_.empty()) { - NEARBY_LOGS(INFO) << __func__ - << "Close the physical socket because all services " - "disconnected."; - physical_socket_->Close(); - } } else { NEARBY_LOGS(WARNING) - << __func__ << "Received a DISCONNECTION frame but there's no alive socket to " "disconnect for service ID Hash Key " << salted_service_id_hash_key; @@ -622,15 +616,13 @@ void MultiplexSocket::HandleDataFrame(const ByteArray& salted_service_id_hash, } if (virtual_socket != nullptr) { - NEARBY_LOGS(INFO) - << __func__ + NEARBY_LOGS(VERBOSE) << "Received a DATA frame to feed virtual socket for salted service ID " "Hash Key " << salted_service_id_hash_key; virtual_socket->FeedIncomingData(ByteArray(frame.data())); } else { NEARBY_LOGS(WARNING) - << __func__ << "Received a DATA frame but there's no alive socket to feed for " "salted service ID Hash Key " << salted_service_id_hash_key; @@ -642,32 +634,50 @@ void MultiplexSocket::OnPhysicalSocketClosed() { } void MultiplexSocket::OnVirtualSocketClosed(const std::string& service_id) { - RunOffloadThread("VirtualSocketClosed", [this, service_id]() { + NEARBY_LOGS(INFO) << __func__ << " for service_id:" << service_id; + CountDownLatch latch(1); + bool shutdown = false; + RunOffloadThread("VirtualSocketClosed", [this, service_id, &latch, + &shutdown]() { + NEARBY_LOGS(INFO) << "Try to close Virtual socket: " << service_id; + MediumSocket* virtual_socket = GetVirtualSocket(service_id); { MutexLock lock(&virtual_socket_mutex_); - MediumSocket* virtual_socket = GetVirtualSocket(service_id); + NEARBY_LOGS(INFO) << "virtual_socket:" << virtual_socket; if (virtual_socket != nullptr) { - virtual_sockets_.erase(GenerateServiceIdHashKeyWithSalt( + auto salted_service_id_hash_key = GenerateServiceIdHashKeyWithSalt( service_id, - multiplex_output_stream_.GetServiceIdHashSalt(service_id))); - NEARBY_LOGS(INFO) << __func__ << "Virtual socket(" << service_id - << ") disconnected"; + multiplex_output_stream_.GetServiceIdHashSalt(service_id)); multiplex_output_stream_.Close(service_id); - virtual_socket->Close(); + virtual_sockets_.erase(salted_service_id_hash_key); + NEARBY_LOGS(INFO) << "Erase Virtual socket with service_id: " + << service_id + << ", hash_key: " << salted_service_id_hash_key; + ListVirtualSocket(); if (virtual_sockets_.empty()) { - NEARBY_LOGS(INFO) << __func__ - << "Close the physical socket because all virtual " + NEARBY_LOGS(INFO) << "Close the physical socket because all virtual " "sockets disconnected."; - physical_socket_->Close(); + Shutdown(); + shutdown = true; } - return; + } else { + NEARBY_LOGS(INFO) << "Virtual socket(" << service_id + << ") not found"; } - - NEARBY_LOGS(INFO) << __func__ << "Virtual socket(" << service_id - << ") not found"; } + latch.CountDown(); }); + + if (!latch.Await(absl::Milliseconds(1000)).result()) { + NEARBY_LOGS(ERROR) << "Timeout to close virtual socket"; + } + if (shutdown) { + NEARBY_LOGS(INFO) + << "Shutdown single_thread_offloader_ and physical_reader_thread_"; + single_thread_offloader_.Shutdown(); + physical_reader_thread_.Shutdown(); + } } MediumSocket* MultiplexSocket::ReMapAndGetVirtualSocket( @@ -675,8 +685,7 @@ MediumSocket* MultiplexSocket::ReMapAndGetVirtualSocket( const std::string& service_id_hash_salt) { std::string salted_service_id_hash_key = GenerateServiceIdHashKey(salted_service_id_hash); - NEARBY_LOGS(VERBOSE) << __func__ - << "reMapAndGetVirtualSocket with serviceIdHashSalt=" + NEARBY_LOGS(VERBOSE) << "ReMapAndGetVirtualSocket with serviceIdHashSalt=" << service_id_hash_salt << ", saltedServiceIdHashKey=" << salted_service_id_hash_key; { @@ -695,17 +704,19 @@ MediumSocket* MultiplexSocket::ReMapAndGetVirtualSocket( (hash_key == salted_service_id_hash_key)) { return virtual_socket.get(); } else { - NEARBY_LOGS(INFO) << __func__ << "Remap the virtualSockets."; - virtual_sockets_.erase(hash_key); + NEARBY_LOGS(INFO) << "Remap the virtualSockets."; output_stream->SetserviceIdHashSalt(service_id_hash_salt); - virtual_sockets_.emplace(salted_service_id_hash_key, virtual_socket); - - return virtual_socket.get(); + auto virtual_socket_tmp = virtual_socket; + NEARBY_LOGS(INFO) << "virtual_socket before:" << virtual_socket; + virtual_sockets_.erase(hash_key); + virtual_sockets_[salted_service_id_hash_key] = virtual_socket_tmp; + ListVirtualSocket(); + return virtual_socket_tmp.get(); } } } - NEARBY_LOGS(INFO) << __func__ << "Failed to remap the virtualSockets."; + NEARBY_LOGS(INFO) << "Failed to remap the virtualSockets."; return nullptr; } @@ -715,9 +726,13 @@ void MultiplexSocket::RunOffloadThread(const std::string& name, } void MultiplexSocket::Shutdown() { - NEARBY_LOGS(INFO) << __func__ << " shutdown"; + NEARBY_LOGS(INFO) << __func__ << " start"; + if (is_shutdown_) { + NEARBY_LOGS(INFO) << __func__ << " Already shutdown"; + return; + } { - MutexLock lock(&virtual_socket_mutex_); + // MutexLock lock(&virtual_socket_mutex_); for (auto& [hash_key, virtual_socket] : virtual_sockets_) { if (virtual_socket != nullptr) { virtual_socket->Close(); @@ -727,14 +742,23 @@ void MultiplexSocket::Shutdown() { } multiplex_output_stream_.Shutdown(); - physical_socket_->Close(); + switch (medium_) { + case Medium::BLUETOOTH: + bluetooth_socket_.Close(); + break; + case Medium::UNKNOWN_MEDIUM: + NEARBY_LOGS(INFO) << __func__ << " Unknown medium"; + break; + default: + break; + } GetIncomingConnectionCallbacks().clear(); connection_response_futures_.clear(); - physical_reader_thread_.Shutdown(); - single_thread_offloader_.Shutdown(); is_shutdown_ = true; + enabled_.Set(false); + NEARBY_LOGS(INFO) << __func__ << " end"; } } // namespace multiplex diff --git a/connections/implementation/mediums/multiplex/multiplex_socket.h b/connections/implementation/mediums/multiplex/multiplex_socket.h index 2740ce50..2dcbdb19 100644 --- a/connections/implementation/mediums/multiplex/multiplex_socket.h +++ b/connections/implementation/mediums/multiplex/multiplex_socket.h @@ -15,31 +15,25 @@ #ifndef CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_ #define CORE_INTERNAL_MEDIUMS_MULTIPLEX_MULTIPLEX_SOCKET_H_ -#include -#include #include -#include -#include #include #include -#include "absl/base/thread_annotations.h" #include "absl/container/flat_hash_map.h" #include "absl/functional/any_invocable.h" #include "connections/implementation/mediums/multiplex/multiplex_output_stream.h" +#include "connections/medium_selector.h" #include "internal/platform/atomic_boolean.h" +#include "internal/platform/ble.h" +#include "internal/platform/bluetooth_classic.h" #include "internal/platform/byte_array.h" -#include "internal/platform/count_down_latch.h" -#include "internal/platform/exception.h" -#include "internal/platform/feature_flags.h" #include "internal/platform/future.h" #include "internal/platform/input_stream.h" #include "internal/platform/logging.h" #include "internal/platform/mutex.h" -#include "internal/platform/output_stream.h" -#include "internal/platform/settable_future.h" #include "internal/platform/single_thread_executor.h" #include "internal/platform/socket.h" +#include "internal/platform/wifi_lan.h" #include "proto/connections_enums.pb.h" #include "proto/mediums/multiplex_frames.pb.h" @@ -69,9 +63,10 @@ class MultiplexSocket { static MultiplexSocket* CreateOutgoingSocket(MediumSocket* physical_socket, const std::string& service_id); - // A Table of service Id as row key, medium type as column key, and {@link - // IncomingConnectionCallback} as value. Non-empty while the client starts - // listening for incoming virtual socket. + // A Table of service Id as row key, medium type as column key, and + // MultiplexIncomingConnectionCb as value. Non-empty while the client starts + // listening for incoming virtual socket. The MultiplexIncomingConnectionCb + // will be called when the incoming virtual socket is established. static absl::flat_hash_map< std::pair, MultiplexIncomingConnectionCb>& @@ -106,10 +101,14 @@ class MultiplexSocket { // Gets the virtual socket count. int GetVirtualSocketCount(); + void ListVirtualSocket(); + // Establishes the virtual socket by service id. MediumSocket* EstablishVirtualSocket(const std::string& service_id); // Shuts down the multiplex socket. void Shutdown(); + bool IsShutdown() { return is_shutdown_; } + void SetShutdown(bool is_shutdown) { is_shutdown_ = is_shutdown; } private: explicit MultiplexSocket(MediumSocket* physical_socket); @@ -167,10 +166,18 @@ class MultiplexSocket { // The physical socket connect to the remote device. MediumSocket* physical_socket_; + // The output stream to manage all outgoing frames from all clients. MultiplexOutputStream multiplex_output_stream_; - // The {@link InputStream} of the physical socket. + // The {@link InputStream} of the physical socket. It is used to read the + // incoming MultiplexFrame from the physical socket. InputStream* physical_reader_; + // The medium type of the physical socket. + Medium medium_; + // Save the phyical socket here, so it can be closed when all the virtual + // socket is gone. + BluetoothSocket bluetooth_socket_; + WifiLanSocket wifi_lan_socket_; // The callback to enable the MultiplexSocket. std::shared_ptr> enable_cb_ = @@ -191,7 +198,8 @@ class MultiplexSocket { // MultiplexSocket object mutable Mutex virtual_socket_mutex_; absl::flat_hash_map> - virtual_sockets_ ABSL_GUARDED_BY(virtual_socket_mutex_); + // virtual_sockets_ ABSL_GUARDED_BY(virtual_socket_mutex_); + virtual_sockets_; // The thread to receive incoming MultiplexFrame from the physical socket. SingleThreadExecutor physical_reader_thread_; diff --git a/connections/implementation/mediums/multiplex/multiplex_socket_test.cc b/connections/implementation/mediums/multiplex/multiplex_socket_test.cc index e114e9a6..36b812b4 100644 --- a/connections/implementation/mediums/multiplex/multiplex_socket_test.cc +++ b/connections/implementation/mediums/multiplex/multiplex_socket_test.cc @@ -90,11 +90,18 @@ class FakeSocket : public MediumSocket { InputStream& GetInputStream() override { return *reader_1_; } OutputStream& GetOutputStream() override { return *writer_2_; } - void Close() override { + Exception Close() override { + if (IsVirtualSocket()) { + NEARBY_LOGS(INFO) << "Multiplex: Closing virtual socket: " << this; + CloseLocal(); + return {Exception::kSuccess}; + } + NEARBY_LOGS(INFO) << "Multiplex: Closing physical socket: " << this; reader_1_->Close(); reader_2_->Close(); writer_1_->Close(); writer_2_->Close(); + return {Exception::kSuccess}; } MediumSocket* CreateVirtualSocket( @@ -148,13 +155,12 @@ class FakeSocket : public MediumSocket { }; TEST(MultiplexSocketTest, CreateSuccessAndReaderThreadStarted) { - testing::NiceMock fake_socket{Medium::WIFI_LAN}; + testing::NiceMock fake_socket{Medium::BLUETOOTH}; MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1), - Medium::WIFI_LAN); + Medium::BLUETOOTH); MultiplexSocket* multiplex_socket_incoming = MultiplexSocket::CreateIncomingSocket(&fake_socket, std::string(SERVICE_ID_1)); - ASSERT_NE(multiplex_socket_incoming, nullptr); FakeSocket* virtual_socket = (FakeSocket*)multiplex_socket_incoming->GetVirtualSocket( @@ -189,10 +195,9 @@ TEST(MultiplexSocketTest, CreateSuccessAndReaderThreadStarted) { ByteArray data = result.result(); NEARBY_LOGS(INFO) << "Received " << data.size() << " bytes of data."; EXPECT_NE(data.size(), 0); + absl::SleepFor(absl::Milliseconds(100)); fake_socket.reader_1_->Close(); EXPECT_EQ(multiplex_socket_incoming->GetVirtualSocketCount(), 1); - multiplex_socket_incoming->Shutdown(); - EXPECT_EQ(multiplex_socket_incoming->GetVirtualSocketCount(), 0); } TEST(MultiplexSocketTest, CreateFail_MediumNotSupport) { @@ -208,11 +213,11 @@ TEST(MultiplexSocketTest, CreateFail_MediumNotSupport) { TEST(MultiplexSocketTest, EstablishVirtualSocket_ReturnNullWhenMultiplexSocketDisabled) { - testing::NiceMock fake_socket{Medium::WIFI_LAN}; + testing::NiceMock fake_socket{Medium::BLE}; MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1), - Medium::WIFI_LAN); + Medium::BLE); MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2), - Medium::WIFI_LAN); + Medium::BLE); MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket( &fake_socket, std::string(SERVICE_ID_1)); ASSERT_NE(multiplex_socket, nullptr); @@ -221,24 +226,36 @@ TEST(MultiplexSocketTest, multiplex_socket->EstablishVirtualSocket(std::string(SERVICE_ID_2)); EXPECT_EQ(socket, nullptr); absl::SleepFor(absl::Milliseconds(100)); - + FakeSocket* virtual_socket = + (FakeSocket*)multiplex_socket->GetVirtualSocket( + std::string(SERVICE_ID_1)); + if (virtual_socket == nullptr) { + NEARBY_LOGS(INFO) << "Virtual socket not found for " << SERVICE_ID_1; + return; + } fake_socket.reader_1_->Close(); EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 1); - multiplex_socket->Shutdown(); + virtual_socket->Close(); EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 0); } TEST(MultiplexSocketTest, EstablishVirtualSocket_TimeoutBecauseNoConnectionResponse) { - testing::NiceMock fake_socket{Medium::BLUETOOTH}; + testing::NiceMock fake_socket{Medium::WIFI_LAN}; MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1), - Medium::BLUETOOTH); + Medium::WIFI_LAN); MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2), - Medium::BLUETOOTH); + Medium::WIFI_LAN); MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket( &fake_socket, std::string(SERVICE_ID_1)); ASSERT_NE(multiplex_socket, nullptr); multiplex_socket->Enable(); + FakeSocket* virtual_socket = (FakeSocket*)multiplex_socket->GetVirtualSocket( + std::string(SERVICE_ID_1)); + if (virtual_socket == nullptr) { + NEARBY_LOGS(INFO) << "Virtual socket not found for " << SERVICE_ID_1; + return; + } SingleThreadExecutor executor; CountDownLatch latch(1); @@ -268,17 +285,17 @@ TEST(MultiplexSocketTest, absl::SleepFor(absl::Milliseconds(100)); fake_socket.reader_1_->Close(); EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 1); - multiplex_socket->Shutdown(); + virtual_socket->Close(); EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 0); } TEST(MultiplexSocketTest, EstablishVirtualSocket_RemoteAccepted) { - testing::NiceMock fake_socket{Medium::BLE}; + testing::NiceMock fake_socket{Medium::BLUETOOTH}; MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_1), - Medium::BLE); + Medium::BLUETOOTH); MultiplexSocket::StopListeningForIncomingConnection(std::string(SERVICE_ID_2), - Medium::BLE); + Medium::BLUETOOTH); MultiplexSocket* multiplex_socket = MultiplexSocket::CreateOutgoingSocket( &fake_socket, std::string(SERVICE_ID_1)); @@ -351,8 +368,6 @@ TEST(MultiplexSocketTest, fake_socket.reader_1_->Close(); EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 2); - multiplex_socket->Shutdown(); - EXPECT_EQ(multiplex_socket->GetVirtualSocketCount(), 0); } } // namespace multiplex diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc b/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc index 96d1b9bb..3df46999 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl.cc @@ -90,9 +90,9 @@ InputStream& WebRtcSocket::GetInputStream() { return *pipe_input_; } OutputStream& WebRtcSocket::GetOutputStream() { return output_stream_; } -void WebRtcSocket::Close() { +Exception WebRtcSocket::Close() { NEARBY_LOGS(INFO) << "WebRtcSocket::Close(" << name_ << ") this: " << this; - if (closed_.Set(true)) return; + if (closed_.Set(true)) return {Exception::kSuccess}; ClosePipe(); // NOTE: This call blocks and triggers a state change on the siginaling thread @@ -101,6 +101,7 @@ void WebRtcSocket::Close() { data_channel_->Close(); NEARBY_LOGS(INFO) << "WebRtcSocket::Close(" << name_ << ") this: " << this << " done"; + return {Exception::kSuccess}; } void WebRtcSocket::OnStateChange() { diff --git a/connections/implementation/mediums/webrtc/webrtc_socket_impl.h b/connections/implementation/mediums/webrtc/webrtc_socket_impl.h index 723280b8..2c11f57f 100644 --- a/connections/implementation/mediums/webrtc/webrtc_socket_impl.h +++ b/connections/implementation/mediums/webrtc/webrtc_socket_impl.h @@ -15,17 +15,20 @@ #ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_ #define CORE_INTERNAL_MEDIUMS_WEBRTC_WEBRTC_SOCKET_IMPL_H_ -#ifndef NO_WEBRTC - +#include +#include #include -#include "connections/listeners.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/listeners.h" +#include "internal/platform/runnable.h" +#ifndef NO_WEBRTC #include "internal/platform/atomic_boolean.h" #include "internal/platform/condition_variable.h" #include "internal/platform/input_stream.h" #include "internal/platform/mutex.h" #include "internal/platform/output_stream.h" -#include "internal/platform/pipe.h" #include "internal/platform/single_thread_executor.h" #include "internal/platform/socket.h" #include "webrtc/api/data_channel_interface.h" @@ -54,7 +57,7 @@ class WebRtcSocket : public Socket, public webrtc::DataChannelObserver { // Overrides for nearby::Socket: InputStream& GetInputStream() override; OutputStream& GetOutputStream() override; - void Close() override; + Exception Close() override; // webrtc::DataChannelObserver: void OnStateChange() override; diff --git a/connections/implementation/mediums/webrtc_socket.h b/connections/implementation/mediums/webrtc_socket.h index 3c1370e6..724e247a 100644 --- a/connections/implementation/mediums/webrtc_socket.h +++ b/connections/implementation/mediums/webrtc_socket.h @@ -15,6 +15,7 @@ #ifndef CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_ #define CORE_INTERNAL_MEDIUMS_WEBRTC_SOCKET_H_ +#include "internal/platform/exception.h" #ifndef NO_WEBRTC #include @@ -38,7 +39,7 @@ class WebRtcSocketWrapper final { OutputStream& GetOutputStream() { return impl_->GetOutputStream(); } - void Close() { return impl_->Close(); } + Exception Close() { return impl_->Close(); } bool IsValid() const { return impl_ != nullptr; } diff --git a/connections/implementation/p2p_cluster_pcp_handler.cc b/connections/implementation/p2p_cluster_pcp_handler.cc index 306429ce..3bac53c6 100644 --- a/connections/implementation/p2p_cluster_pcp_handler.cc +++ b/connections/implementation/p2p_cluster_pcp_handler.cc @@ -549,7 +549,7 @@ void P2pClusterPcpHandler::BlePeripheralDiscoveredHandler( NEARBY_LOGS(ERROR) << ble_status_or.status().ToString(); return; } - auto advertisement = ble_status_or.value(); + const auto& advertisement = ble_status_or.value(); // Make sure the BLE advertisement points to a valid // endpoint we're discovering. @@ -726,7 +726,7 @@ void P2pClusterPcpHandler::BleV2PeripheralDiscoveredHandler( NEARBY_LOGS(ERROR) << ble_status_or.status(); return; } - auto advertisement = ble_status_or.value(); + const auto& advertisement = ble_status_or.value(); // Make sure the BLE advertisement points to a valid // endpoint we're discovering. diff --git a/internal/platform/BUILD b/internal/platform/BUILD index ce968d3d..3a37e6e9 100644 --- a/internal/platform/BUILD +++ b/internal/platform/BUILD @@ -280,6 +280,7 @@ cc_test( cc_library( name = "types", srcs = [ + "blocking_queue_stream.cc", "clock_impl.cc", "device_info_impl.cc", "monitored_runnable.cc", @@ -289,8 +290,10 @@ cc_library( "timer_impl.cc", ], hdrs = [ + "array_blocking_queue.h", "atomic_boolean.h", "atomic_reference.h", + "blocking_queue_stream.h", "borrowable.h", "cancelable.h", "cancelable_alarm.h", diff --git a/internal/platform/array_blocking_queue.h b/internal/platform/array_blocking_queue.h new file mode 100644 index 00000000..1752d8cc --- /dev/null +++ b/internal/platform/array_blocking_queue.h @@ -0,0 +1,104 @@ +// Copyright 2024 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef PLATFORM_PUBLIC_ARRAY_BLOCKING_QUEUE_H_ +#define PLATFORM_PUBLIC_ARRAY_BLOCKING_QUEUE_H_ + +#include +#include +#include + +#include "internal/platform/condition_variable.h" +#include "internal/platform/logging.h" +#include "internal/platform/mutex.h" +#include "internal/platform/mutex_lock.h" + +namespace nearby { + +/** + * Payload from different services/clients will be put into an + * ArrayBlockingQueue before sending to ensure each client has equal chance to + * send its data. Since C++ doesn't provide ArrayBlockingQueue as Java, we + * implement one here. + */ +template +class ArrayBlockingQueue { + public: + explicit ArrayBlockingQueue(size_t capacity) : capacity_(capacity) {} + + void Put(const T& value) { + MutexLock lock(&queue_mutex_); + if (queue_.size() >= capacity_) { + has_space_.Wait(); + } + queue_.push(value); + NEARBY_LOGS(INFO) << "ArrayBlockingQueue::Put()"; + has_data_.Notify(); + } + + T Take() { + MutexLock lock(&queue_mutex_); + if (queue_.empty()) { + has_data_.Wait(); + } + T front = queue_.front(); + queue_.pop(); + NEARBY_LOGS(INFO) << "ArrayBlockingQueue::Take()"; + has_space_.Notify(); + return front; + } + + bool TryPut(const T& value) { + MutexLock lock(&queue_mutex_); + if (queue_.size() < capacity_) { + queue_.push(value); + has_data_.Notify(); + return true; + } + return false; + } + + // Returns std::nullopt if the queue is empty. + std::optional TryTake() { + MutexLock lock(&queue_mutex_); + if (!queue_.empty()) { + T front = queue_.front(); + queue_.pop(); + has_space_.Notify(); + return front; + } + return std::nullopt; + } + + size_t Size() const { + MutexLock lock(&queue_mutex_); + return queue_.size(); + } + + bool Empty() const { + MutexLock lock(&queue_mutex_); + return queue_.empty(); + } + + private: + std::queue queue_; + mutable Mutex queue_mutex_; + ConditionVariable has_data_{&queue_mutex_}; + ConditionVariable has_space_{&queue_mutex_}; + const size_t capacity_; +}; + +} // namespace nearby + +#endif // PLATFORM_PUBLIC_ARRAY_BLOCKING_QUEUE_H_ diff --git a/internal/platform/blocking_queue_stream.cc b/internal/platform/blocking_queue_stream.cc new file mode 100644 index 00000000..171b3395 --- /dev/null +++ b/internal/platform/blocking_queue_stream.cc @@ -0,0 +1,72 @@ +// Copyright 2024 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "internal/platform/blocking_queue_stream.h" + +#include + +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/logging.h" + +namespace nearby { + +BlockingQueueStream::BlockingQueueStream() { + NEARBY_LOGS(INFO) << "Create a BlockingQueueStream with size " + << FeatureFlags::GetInstance() + .GetFlags() + .blocking_queue_stream_queue_capacity; +} + +ExceptionOr BlockingQueueStream::Read(std::int64_t size) { + if (is_closed_) { + NEARBY_LOGS(INFO) + << "Failed to read BlockingQueueStream because it was closed."; + return ExceptionOr(Exception::kInterrupted); + } + NEARBY_LOGS(INFO) << "BlockingQueueStream read " << size << " bytes"; + return ExceptionOr(blocking_queue_.Take()); +} + +void BlockingQueueStream::Write(const ByteArray& bytes) { + if (is_closed_) { + NEARBY_LOGS(INFO) + << "Failed to write BlockingQueueStream because it was closed."; + return; + } + is_writing_ = true; + blocking_queue_.Put(bytes); + is_writing_ = false; + NEARBY_LOGS(VERBOSE) << "BlockingQueueStream wrote " << bytes.size() + << " bytes"; +} + +Exception BlockingQueueStream::Close() { + if (is_closed_) { + NEARBY_LOGS(INFO) << "InputBlockingQueueStream has already been closed."; + return {Exception::kSuccess}; + } + if (is_writing_) { + NEARBY_LOGS(INFO) + << "BlockingQueueStream is waiting for writing, read first to unblock"; + blocking_queue_.TryTake(); + } + blocking_queue_.TryPut(queue_end_); + is_closed_ = true; + NEARBY_LOGS(INFO) << "InputBlockingQueueStream is closed."; + return {Exception::kSuccess}; +} + +} // namespace nearby diff --git a/internal/platform/blocking_queue_stream.h b/internal/platform/blocking_queue_stream.h new file mode 100644 index 00000000..72d76c14 --- /dev/null +++ b/internal/platform/blocking_queue_stream.h @@ -0,0 +1,52 @@ +// Copyright 2024 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// https://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef PLATFORM_PUBLIC_BLOCKING_QUEUE_STREAM_H_ +#define PLATFORM_PUBLIC_BLOCKING_QUEUE_STREAM_H_ + +#include + +#include "internal/platform/array_blocking_queue.h" +#include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" +#include "internal/platform/feature_flags.h" +#include "internal/platform/input_stream.h" +#include "internal/platform/mutex.h" + +namespace nearby { +class BlockingQueueStream : public InputStream { + public: + BlockingQueueStream(); + ~BlockingQueueStream() override = default; + + ExceptionOr Read(std::int64_t size) override; + void Write(const ByteArray& bytes); + Exception Close() override; + bool IsWriting() const { + return is_writing_; + } + + private: + mutable Mutex mutex_; + ArrayBlockingQueue blocking_queue_{FeatureFlags::GetInstance() + .GetFlags() + .blocking_queue_stream_queue_capacity}; + ByteArray queue_end_{0}; + bool is_writing_ = false; + bool is_closed_ = false; +}; +} // namespace nearby + +#endif // #ifndef PLATFORM_PUBLIC_BLOCKING_QUEUE_STREAM_H_ + diff --git a/internal/platform/bluetooth_classic.cc b/internal/platform/bluetooth_classic.cc index 3f313f4b..6969833f 100644 --- a/internal/platform/bluetooth_classic.cc +++ b/internal/platform/bluetooth_classic.cc @@ -14,10 +14,57 @@ #include "internal/platform/bluetooth_classic.h" +#include +#include +#include + +#include "absl/container/flat_hash_map.h" +#include "internal/platform/bluetooth_adapter.h" +#include "internal/platform/cancellation_flag.h" +#include "internal/platform/implementation/bluetooth_classic.h" #include "internal/platform/logging.h" #include "internal/platform/mutex_lock.h" +#include "internal/platform/output_stream.h" +#include "internal/platform/socket.h" namespace nearby { +using location::nearby::proto::connections::Medium; + +MediumSocket* BluetoothSocket::CreateVirtualSocket(OutputStream* outputstream) { + if (IsVirtualSocket()) { + NEARBY_LOGS(WARNING) + << "Creating the virtual socket on a virtual socket is not allowed."; + return nullptr; + } + auto virtual_socket = std::make_shared(outputstream); + return virtual_socket.get(); +} + +MediumSocket* BluetoothSocket::CreateVirtualSocket( + const std::string& salted_service_id_hash_key, OutputStream* outputstream, + Medium medium, + absl::flat_hash_map>* + virtual_sockets_ptr) { + if (IsVirtualSocket()) { + NEARBY_LOGS(WARNING) + << "Creating the virtual socket on a virtual socket is not allowed."; + return nullptr; + } + + auto virtual_socket = std::make_shared(outputstream); + virtual_socket->impl_ = this->impl_; + NEARBY_LOGS(WARNING) << "Created the virtual socket for Medium: " + << Medium_Name(virtual_socket->GetMedium()); + + if (virtual_sockets_ptr_ == nullptr) { + virtual_sockets_ptr_ = virtual_sockets_ptr; + } + + (*virtual_sockets_ptr_)[salted_service_id_hash_key] = virtual_socket; + NEARBY_LOGS(INFO) << "virtual_sockets_ size: " + << virtual_sockets_ptr_->size(); + return virtual_socket.get(); +} BluetoothClassicMedium::~BluetoothClassicMedium() { NEARBY_LOG(INFO, "~BluetoothClassicMedium: observer_list_ size: %d", @@ -54,7 +101,7 @@ bool BluetoothClassicMedium::StartDiscovery(DiscoveryCallback callback) { device.GetName().c_str()); MutexLock lock(&mutex_); auto pair = devices_.emplace( - &device, absl::make_unique()); + &device, std::make_unique()); auto& context = *pair.first->second; if (!pair.second) { NEARBY_LOG(INFO, "Adding (again) device=%p, impl=%p", diff --git a/internal/platform/bluetooth_classic.h b/internal/platform/bluetooth_classic.h index 1a6b81df..fe883211 100644 --- a/internal/platform/bluetooth_classic.h +++ b/internal/platform/bluetooth_classic.h @@ -15,14 +15,15 @@ #ifndef PLATFORM_PUBLIC_BLUETOOTH_CLASSIC_H_ #define PLATFORM_PUBLIC_BLUETOOTH_CLASSIC_H_ +#include #include #include #include #include #include "absl/container/flat_hash_map.h" -#include "absl/container/flat_hash_set.h" #include "internal/base/observer_list.h" +#include "internal/platform/blocking_queue_stream.h" #include "internal/platform/bluetooth_adapter.h" #include "internal/platform/byte_array.h" #include "internal/platform/cancellation_flag.h" @@ -34,29 +35,79 @@ #include "internal/platform/logging.h" #include "internal/platform/mutex.h" #include "internal/platform/output_stream.h" +#include "internal/platform/socket.h" namespace nearby { // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html. -class BluetoothSocket final { +class BluetoothSocket : public MediumSocket { public: - BluetoothSocket() = default; + BluetoothSocket() + : MediumSocket(location::nearby::proto::connections::Medium::BLUETOOTH) { + }; BluetoothSocket(const BluetoothSocket&) = default; BluetoothSocket& operator=(const BluetoothSocket&) = default; + +// Creates a physical BluetoothSocket from a platform implementation. explicit BluetoothSocket(std::unique_ptr socket) - : impl_(socket.release()) {} - ~BluetoothSocket() = default; + : MediumSocket(location::nearby::proto::connections::Medium::BLUETOOTH), + impl_(socket.release()) {} + +// Creates a virtual BluetoothSocket from a virtual output stream. + explicit BluetoothSocket(OutputStream* virtual_output_stream) + : MediumSocket(location::nearby::proto::connections::Medium::BLUETOOTH), + blocking_queue_input_stream_(std::make_shared()), + virtual_output_stream_(virtual_output_stream), + is_virtual_socket_(true) {} + + ~BluetoothSocket() override = default; // Returns the InputStream of this connected BluetoothSocket. - InputStream& GetInputStream() { return impl_->GetInputStream(); } + InputStream& GetInputStream() override { + return IsVirtualSocket() ? *blocking_queue_input_stream_ + : impl_->GetInputStream(); + } // Returns the OutputStream of this connected BluetoothSocket. - OutputStream& GetOutputStream() { return impl_->GetOutputStream(); } + OutputStream& GetOutputStream() override { + return IsVirtualSocket() ? *virtual_output_stream_ + : impl_->GetOutputStream(); + } // Closes both input and output streams, marks Socket as closed. // After this call object should be treated as not connected. // Returns Exception::kIo on error, Exception::kSuccess otherwise. - Exception Close() { return impl_->Close(); } + Exception Close() override { + if (IsVirtualSocket()) { + NEARBY_LOGS(INFO) << "Multiplex: Closing virtual socket: " << this; + blocking_queue_input_stream_->Close(); + virtual_output_stream_->Close(); + CloseLocal(); + return {Exception::kSuccess}; + } + NEARBY_LOGS(INFO) << "Multiplex: Closing physical socket: " << this; + return impl_->Close(); + } + + // Returns true if this is a virtual socket. + bool IsVirtualSocket() override { return is_virtual_socket_; } + + // Creates a virtual socket only with outputstream. + MediumSocket* CreateVirtualSocket(OutputStream* outputstream) override; + MediumSocket* CreateVirtualSocket( + const std::string& salted_service_id_hash_key, OutputStream* outputstream, + location::nearby::proto::connections::Medium medium, + absl::flat_hash_map>* + virtual_sockets_ptr) override; + + /** Feeds the received incoming data to the client. */ + void FeedIncomingData(ByteArray data) override { + if (!IsVirtualSocket()) { + NEARBY_LOGS(INFO) << "Feeding data on a physical socket is not allowed."; + return; + } + blocking_queue_input_stream_->Write(data); + } // https://developer.android.com/reference/android/bluetooth/BluetoothSocket.html#getRemoteDevice() BluetoothDevice GetRemoteDevice() { @@ -73,7 +124,10 @@ class BluetoothSocket final { // BluetoothServerSocket::Accept(). // These methods may also return an invalid socket if connection failed for // any reason. - bool IsValid() const { return impl_ != nullptr; } + bool IsValid() const { + if (is_virtual_socket_) return true; + return impl_ != nullptr; + } // Returns reference to platform implementation. // This is used to communicate with platform code, and for debugging purposes. @@ -84,6 +138,11 @@ class BluetoothSocket final { private: std::shared_ptr impl_; + absl::flat_hash_map>* + virtual_sockets_ptr_ = nullptr; + std::shared_ptr blocking_queue_input_stream_ = nullptr; + OutputStream* virtual_output_stream_ = nullptr; + bool is_virtual_socket_ = false; }; // https://developer.android.com/reference/android/bluetooth/BluetoothServerSocket.html. diff --git a/internal/platform/feature_flags.h b/internal/platform/feature_flags.h index c9e9c9b5..9df7fcbe 100644 --- a/internal/platform/feature_flags.h +++ b/internal/platform/feature_flags.h @@ -110,6 +110,7 @@ class FeatureFlags { // The maximum size of frame we'll attempt to read, to avoid a remote device // from triggering an OutOfMemory error. std::uint32_t connection_max_frame_length = 1048576; + std::uint32_t blocking_queue_stream_queue_capacity = 10; }; static const FeatureFlags& GetInstance() { diff --git a/internal/platform/socket.h b/internal/platform/socket.h index 16fc983e..c1a34d64 100644 --- a/internal/platform/socket.h +++ b/internal/platform/socket.h @@ -23,6 +23,7 @@ #include "absl/container/flat_hash_set.h" #include "absl/functional/any_invocable.h" #include "internal/platform/byte_array.h" +#include "internal/platform/exception.h" #include "internal/platform/input_stream.h" #include "internal/platform/output_stream.h" #include "proto/connections_enums.pb.h" @@ -38,7 +39,7 @@ class Socket { virtual InputStream& GetInputStream() = 0; virtual OutputStream& GetOutputStream() = 0; - virtual void Close() = 0; + virtual Exception Close() = 0; }; class MediumSocket : public Socket { @@ -52,6 +53,11 @@ class MediumSocket : public Socket { return medium_; } + /** Creates a virtual socket only with outputstream. */ + virtual MediumSocket* CreateVirtualSocket(OutputStream* outputstream) { + return this; + } + /** Creates a virtual socket. */ virtual MediumSocket* CreateVirtualSocket( const std::string& salted_service_id_hash_key, OutputStream* outputstream, @@ -62,7 +68,9 @@ class MediumSocket : public Socket { } /** Feeds the received incoming data to the client. */ - virtual void FeedIncomingData(ByteArray data) {} + virtual void FeedIncomingData(ByteArray data) { +// NEARBY_LOGS(INFO) << "FeedIncomingData: do nothing"; + } /** Returns true if the socket is a virtual socket. */ virtual bool IsVirtualSocket() { @@ -86,16 +94,15 @@ class MediumSocket : public Socket { if (!IsVirtualSocket()) { return; } - for (auto& callback : multiplex_socket_enabled_cbs_) { - callback.get(); + (*callback)(); } } /** Closes the local socket. */ void CloseLocal() { for (auto& listener : socket_closed_listeners_) { - listener.get(); + (*listener)(); } }