From 2d4edf5b18e7029be355017f436c96a9f66fd9a0 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 12 Feb 2021 12:18:55 -0800 Subject: [PATCH 1/5] cl/356658736 Add Stop() to ServiceController API. --- cpp/core/internal/BUILD | 1 - cpp/core/internal/mock_service_controller.h | 1 + .../internal/offline_service_controller.cc | 12 ++ .../internal/offline_service_controller.h | 2 +- cpp/core/internal/service_controller.h | 7 + .../internal/service_controller_router.cc | 7 +- cpp/core/internal/service_controller_router.h | 3 +- .../internal/stoppable_service_controller.h | 131 ------------------ 8 files changed, 25 insertions(+), 139 deletions(-) delete mode 100644 cpp/core/internal/stoppable_service_controller.h diff --git a/cpp/core/internal/BUILD b/cpp/core/internal/BUILD index 803df6ad..f6829072 100644 --- a/cpp/core/internal/BUILD +++ b/cpp/core/internal/BUILD @@ -62,7 +62,6 @@ cc_library( "pcp_manager.h", "service_controller.h", "service_controller_router.h", - "stoppable_service_controller.h", "webrtc_bwu_handler.h", "webrtc_endpoint_channel.h", "wifi_lan_bwu_handler.h", diff --git a/cpp/core/internal/mock_service_controller.h b/cpp/core/internal/mock_service_controller.h index d89e47e1..a6745d77 100644 --- a/cpp/core/internal/mock_service_controller.h +++ b/cpp/core/internal/mock_service_controller.h @@ -17,6 +17,7 @@ namespace connections { */ class MockServiceController : public ServiceController { public: + MOCK_METHOD(void, Stop, (), (override)); MOCK_METHOD(Status, StartAdvertising, (ClientProxy * client, const std::string& service_id, const ConnectionOptions& options, diff --git a/cpp/core/internal/offline_service_controller.cc b/cpp/core/internal/offline_service_controller.cc index c1008a4f..d4c0a97a 100644 --- a/cpp/core/internal/offline_service_controller.cc +++ b/cpp/core/internal/offline_service_controller.cc @@ -17,48 +17,57 @@ void OfflineServiceController::Stop() { Status OfflineServiceController::StartAdvertising( ClientProxy* client, const std::string& service_id, const ConnectionOptions& options, const ConnectionRequestInfo& info) { + if (stop_) return {Status::kOutOfOrderApiCall}; return pcp_manager_.StartAdvertising(client, service_id, options, info); } void OfflineServiceController::StopAdvertising(ClientProxy* client) { + if (stop_) return; pcp_manager_.StopAdvertising(client); } Status OfflineServiceController::StartDiscovery( ClientProxy* client, const std::string& service_id, const ConnectionOptions& options, const DiscoveryListener& listener) { + if (stop_) return {Status::kOutOfOrderApiCall}; return pcp_manager_.StartDiscovery(client, service_id, options, listener); } void OfflineServiceController::StopDiscovery(ClientProxy* client) { + if (stop_) return; pcp_manager_.StopDiscovery(client); } void OfflineServiceController::InjectEndpoint( ClientProxy* client, const std::string& service_id, const OutOfBandConnectionMetadata& metadata) { + if (stop_) return; pcp_manager_.InjectEndpoint(client, service_id, metadata); } Status OfflineServiceController::RequestConnection( ClientProxy* client, const std::string& endpoint_id, const ConnectionRequestInfo& info, const ConnectionOptions& options) { + if (stop_) return {Status::kOutOfOrderApiCall}; return pcp_manager_.RequestConnection(client, endpoint_id, info, options); } Status OfflineServiceController::AcceptConnection( ClientProxy* client, const std::string& endpoint_id, const PayloadListener& listener) { + if (stop_) return {Status::kOutOfOrderApiCall}; return pcp_manager_.AcceptConnection(client, endpoint_id, listener); } Status OfflineServiceController::RejectConnection( ClientProxy* client, const std::string& endpoint_id) { + if (stop_) return {Status::kOutOfOrderApiCall}; return pcp_manager_.RejectConnection(client, endpoint_id); } void OfflineServiceController::InitiateBandwidthUpgrade( ClientProxy* client, const std::string& endpoint_id) { + if (stop_) return; NEARBY_LOGS(INFO) << "Client " << client->GetClientId() << " initiated a manual bandwidth upgrade with endpoint id=" << endpoint_id; @@ -68,16 +77,19 @@ void OfflineServiceController::InitiateBandwidthUpgrade( void OfflineServiceController::SendPayload( ClientProxy* client, const std::vector& endpoint_ids, Payload payload) { + if (stop_) return; payload_manager_.SendPayload(client, endpoint_ids, std::move(payload)); } Status OfflineServiceController::CancelPayload(ClientProxy* client, std::int64_t payload_id) { + if (stop_) return {Status::kOutOfOrderApiCall}; return payload_manager_.CancelPayload(client, payload_id); } void OfflineServiceController::DisconnectFromEndpoint( ClientProxy* client, const std::string& endpoint_id) { + if (stop_) return; endpoint_manager_.UnregisterEndpoint(client, endpoint_id); } diff --git a/cpp/core/internal/offline_service_controller.h b/cpp/core/internal/offline_service_controller.h index bc77e201..e74c7a02 100644 --- a/cpp/core/internal/offline_service_controller.h +++ b/cpp/core/internal/offline_service_controller.h @@ -61,7 +61,7 @@ class OfflineServiceController : public ServiceController { void DisconnectFromEndpoint(ClientProxy* client, const std::string& endpoint_id) override; - void Stop(); + void Stop() override; private: // Note that the order of declaration of these is crucial, because we depend diff --git a/cpp/core/internal/service_controller.h b/cpp/core/internal/service_controller.h index ecd80e8a..aa11b040 100644 --- a/cpp/core/internal/service_controller.h +++ b/cpp/core/internal/service_controller.h @@ -34,6 +34,13 @@ class ServiceController { ServiceController(const ServiceController&) = delete; ServiceController& operator=(const ServiceController&) = delete; + // Stops and disables service controller. + // + // When service controller is stopped all API call fail early. + // Note that all Core, ClientProxy objects referencing this service + // controller are affected. + virtual void Stop() = 0; + // Starts advertising an endpoint for a local app. virtual Status StartAdvertising(ClientProxy* client, const std::string& service_id, diff --git a/cpp/core/internal/service_controller_router.cc b/cpp/core/internal/service_controller_router.cc index e28d842d..6a44ad4a 100644 --- a/cpp/core/internal/service_controller_router.cc +++ b/cpp/core/internal/service_controller_router.cc @@ -40,7 +40,7 @@ ServiceControllerRouter::~ServiceControllerRouter() { .GetFlags() .disable_released_service_controller) { if (service_controller_) { - service_controller_->Shutdown(); + service_controller_->Stop(); } } else { service_controller_.reset(); @@ -427,7 +427,7 @@ void ServiceControllerRouter::ReleaseServiceControllerForClient( if (FeatureFlags::GetInstance() .GetFlags() .disable_released_service_controller) { - service_controller_->Shutdown(); + service_controller_->Stop(); } if (clients_.empty()) { @@ -477,8 +477,7 @@ Status ServiceControllerRouter::UpdateCurrentServiceControllerAndStrategy( return {Status::kError}; } - service_controller_ = absl::make_unique( - service_controller_factory_()); + service_controller_.reset(service_controller_factory_()); current_strategy_ = strategy; return {Status::kSuccess}; diff --git a/cpp/core/internal/service_controller_router.h b/cpp/core/internal/service_controller_router.h index 9a2817e2..26eb7bf2 100644 --- a/cpp/core/internal/service_controller_router.h +++ b/cpp/core/internal/service_controller_router.h @@ -7,7 +7,6 @@ #include "core/internal/client_proxy.h" #include "core/internal/service_controller.h" -#include "core/internal/stoppable_service_controller.h" #include "core/options.h" #include "core/params.h" #include "platform/base/runnable.h" @@ -106,7 +105,7 @@ class ServiceControllerRouter { absl::flat_hash_set clients_; std::function service_controller_factory_; - std::unique_ptr service_controller_; + std::unique_ptr service_controller_; Strategy current_strategy_; SingleThreadExecutor serializer_; }; diff --git a/cpp/core/internal/stoppable_service_controller.h b/cpp/core/internal/stoppable_service_controller.h deleted file mode 100644 index 4e1f1be6..00000000 --- a/cpp/core/internal/stoppable_service_controller.h +++ /dev/null @@ -1,131 +0,0 @@ -#ifndef CORE_INTERNAL_STOPPABLE_SERVICE_CONTROLLER_H_ -#define CORE_INTERNAL_STOPPABLE_SERVICE_CONTROLLER_H_ - -#include - -#include "core/internal/service_controller.h" -#include "core/status.h" -#include "platform/public/atomic_boolean.h" - -namespace location { -namespace nearby { -namespace connections { - -// A ServiceController proxy that can be shut down. -// When shut down, the API calls are not forwarded to the real controller. -// StoppableServiceController takes over ownership of ServiceController. -class StoppableServiceController : public ServiceController { - public: - explicit StoppableServiceController(ServiceController* controller) - : service_controller_{controller} {} - ~StoppableServiceController() override = default; - - void Shutdown() { stopped_.Set(true); } - - Status StartAdvertising(ClientProxy* client, const std::string& service_id, - const ConnectionOptions& options, - const ConnectionRequestInfo& info) override { - if (stopped_) { - return {Status::kError}; - } - return service_controller_->StartAdvertising(client, service_id, options, - info); - } - - void StopAdvertising(ClientProxy* client) override { - if (stopped_) { - return; - } - service_controller_->StopAdvertising(client); - } - - Status StartDiscovery(ClientProxy* client, const std::string& service_id, - const ConnectionOptions& options, - const DiscoveryListener& listener) override { - if (stopped_) { - return {Status::kError}; - } - return service_controller_->StartDiscovery(client, service_id, options, - listener); - } - void StopDiscovery(ClientProxy* client) override { - if (stopped_) { - return; - } - service_controller_->StopDiscovery(client); - } - - void InjectEndpoint(ClientProxy* client, const std::string& service_id, - const OutOfBandConnectionMetadata& metadata) override { - if (stopped_) { - return; - } - service_controller_->InjectEndpoint(client, service_id, metadata); - } - - Status RequestConnection(ClientProxy* client, const std::string& endpoint_id, - const ConnectionRequestInfo& info, - const ConnectionOptions& options) override { - if (stopped_) { - return {Status::kError}; - } - return service_controller_->RequestConnection(client, endpoint_id, info, - options); - } - Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id, - const PayloadListener& listener) override { - if (stopped_) { - return {Status::kError}; - } - return service_controller_->AcceptConnection(client, endpoint_id, listener); - } - Status RejectConnection(ClientProxy* client, - const std::string& endpoint_id) override { - if (stopped_) { - return {Status::kError}; - } - return service_controller_->RejectConnection(client, endpoint_id); - } - - void InitiateBandwidthUpgrade(ClientProxy* client, - const std::string& endpoint_id) override { - if (stopped_) { - return; - } - service_controller_->InitiateBandwidthUpgrade(client, endpoint_id); - } - - void SendPayload(ClientProxy* client, - const std::vector& endpoint_ids, - Payload payload) override { - if (stopped_) { - return; - } - service_controller_->SendPayload(client, endpoint_ids, std::move(payload)); - } - - Status CancelPayload(ClientProxy* client, Payload::Id payload_id) override { - if (stopped_) { - return {Status::kError}; - } - return service_controller_->CancelPayload(client, payload_id); - } - - void DisconnectFromEndpoint(ClientProxy* client, - const std::string& endpoint_id) override { - if (stopped_) { - return; - } - service_controller_->DisconnectFromEndpoint(client, endpoint_id); - } - - private: - std::unique_ptr service_controller_; - AtomicBoolean stopped_{false}; -}; - -} // namespace connections -} // namespace nearby -} // namespace location - -#endif // CORE_INTERNAL_STOPPABLE_SERVICE_CONTROLLER_H_ From f7a2df1e87bf5e4353e270f4ed6b44d3994d926f Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 12 Feb 2021 12:22:48 -0800 Subject: [PATCH 2/5] cl/356667540 Add logs for bwu manager. --- cpp/core/internal/bwu_manager.cc | 51 ++++++++++++++++++++++++++++++-- 1 file changed, 48 insertions(+), 3 deletions(-) diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 47742df4..9c2f33fc 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -110,6 +110,8 @@ void BwuManager::Shutdown() { void BwuManager::InitiateBwuForEndpoint(ClientProxy* client, const std::string& endpoint_id, Medium new_medium) { + NEARBY_LOG(INFO, "InitiateBwuForEndpoint for endpoint %s with medium %d", + endpoint_id.c_str(), new_medium); RunOnBwuManagerThread([this, client, endpoint_id, new_medium]() { Medium proposed_medium = ChooseBestUpgradeMedium( client->GetUpgradeMediums(endpoint_id).GetMediums(true)); @@ -184,6 +186,8 @@ void BwuManager::InitiateBwuForEndpoint(ClientProxy* client, void BwuManager::OnIncomingFrame(OfflineFrame& frame, const std::string& endpoint_id, ClientProxy* client, Medium medium) { + NEARBY_LOG(INFO, "OnIncomingFrame for endpoint %s with medium: %d", + endpoint_id.c_str(), medium); if (parser::GetFrameType(frame) != V1Frame::BANDWIDTH_UPGRADE_NEGOTIATION) return; auto bwu_frame = frame.v1().bandwidth_upgrade_negotiation(); @@ -198,6 +202,7 @@ void BwuManager::OnIncomingFrame(OfflineFrame& frame, void BwuManager::OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, CountDownLatch barrier) { + NEARBY_LOG(INFO, "OnEndpointDisconnect for endpoint %s", endpoint_id.c_str()); RunOnBwuManagerThread([this, client, endpoint_id, barrier]() mutable { if (medium_ == Medium::UNKNOWN_MEDIUM) { barrier.CountDown(); @@ -233,6 +238,7 @@ void BwuManager::OnEndpointDisconnect(ClientProxy* client, } BwuHandler* BwuManager::SetCurrentBwuHandler(Medium medium) { + NEARBY_LOG(INFO, "SetCurrentBwuHandler to %d", medium); handler_ = nullptr; medium_ = medium; if (medium != Medium::UNKNOWN_MEDIUM) { @@ -245,6 +251,7 @@ BwuHandler* BwuManager::SetCurrentBwuHandler(Medium medium) { } void BwuManager::Revert() { + NEARBY_LOG(INFO, "Revert reseting medium %d", medium_); if (handler_) { handler_->Revert(); medium_ = Medium::UNKNOWN_MEDIUM; @@ -255,6 +262,8 @@ void BwuManager::Revert() { void BwuManager::OnBwuNegotiationFrame(ClientProxy* client, const BwuNegotiationFrame& frame, const string& endpoint_id) { + NEARBY_LOG(INFO, "OnBwuNegotiationFrame for endpoint %s", + endpoint_id.c_str()); switch (frame.event_type()) { case BwuNegotiationFrame::UPGRADE_PATH_AVAILABLE: ProcessBwuPathAvailableEvent(client, endpoint_id, @@ -278,6 +287,8 @@ void BwuManager::OnBwuNegotiationFrame(ClientProxy* client, void BwuManager::OnIncomingConnection( ClientProxy* client, std::unique_ptr mutable_connection) { + NEARBY_LOG(INFO, "OnIncomingConnection service id: %s", + client->GetServiceId().c_str()); std::shared_ptr connection( mutable_connection.release()); RunOnBwuManagerThread([this, client, connection]() { @@ -315,7 +326,6 @@ void BwuManager::OnIncomingConnection( if (item.empty()) return; mapped_client = item.mapped(); } - CancelRetryUpgradeAlarm(endpoint_id); if (mapped_client == nullptr) { // This was never a fully EstablishedConnection, no need to provide a @@ -340,6 +350,9 @@ void BwuManager::RunOnBwuManagerThread(Runnable runnable) { void BwuManager::RunUpgradeProtocol( ClientProxy* client, const std::string& endpoint_id, std::unique_ptr new_channel) { + NEARBY_LOG(INFO, "RunUpgradeProtocol new channel @%d name: %s, medium: %d", + new_channel.get(), new_channel->GetName().c_str(), + new_channel->GetMedium()); // First, register this new EndpointChannel as *the* EndpointChannel to use // for this endpoint here onwards. NOTE: We pause this new EndpointChannel // until we've completely drained the old EndpointChannel to avoid out of @@ -379,6 +392,9 @@ void BwuManager::RunUpgradeProtocol( void BwuManager::ProcessBwuPathAvailableEvent( ClientProxy* client, const string& endpoint_id, const UpgradePathInfo& upgrade_path_info) { + NEARBY_LOG(INFO, "ProcessBwuPathAvailableEvent for endpoint %s medium %d.", + endpoint_id.c_str(), + parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium())); if (in_progress_upgrades_.contains(endpoint_id)) { NEARBY_LOG(INFO, "Invoking duplicate ProcessBwuPathAvailableEvent for %s", endpoint_id.c_str()); @@ -394,15 +410,14 @@ void BwuManager::ProcessBwuPathAvailableEvent( return; } } - Medium medium = parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium()); if (medium_ == Medium::UNKNOWN_MEDIUM) { SetCurrentBwuHandler(medium); } - // Check for the correct medium so we don't process an incorrect OfflineFrame. if (medium != medium_) { + NEARBY_LOG(INFO, "Medium not matching"); RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info); return; } @@ -410,6 +425,7 @@ void BwuManager::ProcessBwuPathAvailableEvent( auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id, upgrade_path_info); if (channel == nullptr) { + NEARBY_LOG(INFO, "Failed to get new channel."); RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info); return; } @@ -426,6 +442,10 @@ std::unique_ptr BwuManager::ProcessBwuPathAvailableEventInternal( ClientProxy* client, const string& endpoint_id, const UpgradePathInfo& upgrade_path_info) { + NEARBY_LOG(INFO, + "ProcessBwuPathAvailableEventInternal for endpoint %s medium %d", + endpoint_id.c_str(), + parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium())); std::unique_ptr channel = handler_->CreateUpgradedEndpointChannel(client, client->GetServiceId(), endpoint_id, upgrade_path_info); @@ -480,6 +500,9 @@ BwuManager::ProcessBwuPathAvailableEventInternal( void BwuManager::RunUpgradeFailedProtocol( ClientProxy* client, const std::string& endpoint_id, const UpgradePathInfo& upgrade_path_info) { + NEARBY_LOG(INFO, "RunUpgradeFailedProtocol for endpoint %s medium %d", + endpoint_id.c_str(), + parser::UpgradePathInfoMediumToMedium(upgrade_path_info.medium())); // We attempted to connect to the new medium that the remote device has set up // for us but we failed. We need to let the remote device know so that they // can pick another medium for us to try. @@ -515,6 +538,9 @@ void BwuManager::RunUpgradeFailedProtocol( bool BwuManager::ReadClientIntroductionFrame(EndpointChannel* channel, ClientIntroduction& introduction) { + NEARBY_LOG(INFO, + "ReadClientIntroductionFrame with channel name: %s, medium: %d", + channel->GetName().c_str(), channel->GetMedium()); CancelableAlarm timeout_alarm( "BwuManager::ReadClientIntroductionFrame", [channel]() { @@ -545,6 +571,9 @@ bool BwuManager::ReadClientIntroductionFrame(EndpointChannel* channel, } bool BwuManager::ReadClientIntroductionAckFrame(EndpointChannel* channel) { + NEARBY_LOG(INFO, + "ReadClientIntroductionFrame with channel name: %s, medium: %d", + channel->GetName().c_str(), channel->GetMedium()); CancelableAlarm timeout_alarm( "BwuManager::ReadClientIntroductionAckFrame", [channel]() { @@ -572,11 +601,16 @@ bool BwuManager::ReadClientIntroductionAckFrame(EndpointChannel* channel) { } bool BwuManager::WriteClientIntroductionAckFrame(EndpointChannel* channel) { + NEARBY_LOG(INFO, + "WriteClientIntroductionAckFrame channel name: %s, medium: %d", + channel->GetName().c_str(), channel->GetMedium()); return channel->Write(parser::ForBwuIntroductionAck()).Ok(); } void BwuManager::ProcessLastWriteToPriorChannelEvent( ClientProxy* client, const std::string& endpoint_id) { + NEARBY_LOG(INFO, "ProcessLastWriteToPriorChannelEvent for endpoint %s", + endpoint_id.c_str()); // By this point in the upgrade protocol, there is the guarantee that both // involved endpoints have registered a new EndpointChannel with the // EndpointChannelManager as the official channel for communication; given @@ -621,6 +655,8 @@ void BwuManager::ProcessLastWriteToPriorChannelEvent( void BwuManager::ProcessSafeToClosePriorChannelEvent( ClientProxy* client, const std::string& endpoint_id) { + NEARBY_LOG(INFO, "ProcessSafeToClosePriorChannelEvent for endpoint %s", + endpoint_id.c_str()); // By this point in the upgrade protocol, there's no more writes happening // over the prior EndpointChannel, and the remote device has given us the // go-ahead to close this EndpointChannel [1], so we can safely close it @@ -690,6 +726,9 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent( void BwuManager::ProcessUpgradeFailureEvent( ClientProxy* client, const std::string& endpoint_id, const UpgradePathInfo& upgrade_info) { + NEARBY_LOG(INFO, "ProcessUpgradeFailureEvent for endpoint %s from medium: %d", + endpoint_id.c_str(), + parser::UpgradePathInfoMediumToMedium(upgrade_info.medium())); // The remote device failed to upgrade to the new medium we set up for them. // That's alright! We'll just try the next available medium (if there is // one). @@ -738,6 +777,10 @@ void BwuManager::RetryUpgradeMediums(ClientProxy* client, const std::string& endpoint_id, std::vector upgrade_mediums) { Medium next_medium = ChooseBestUpgradeMedium(upgrade_mediums); + NEARBY_LOG( + INFO, + "RetryUpgradeMediums for endpoint %s after ChooseBestUpgradeMedium: %d", + endpoint_id.c_str(), next_medium); // If current medium is not WiFi and we have not succeeded with upgrading // yet, retry upgrade. @@ -875,6 +918,7 @@ absl::Duration BwuManager::CalculateNextRetryDelay( } void BwuManager::CancelRetryUpgradeAlarm(const std::string& endpoint_id) { + NEARBY_LOG(INFO, "CancelRetryUpgradeAlarm for %s", endpoint_id.c_str()); auto item = retry_upgrade_alarms_.extract(endpoint_id); if (item.empty()) return; auto& pair = item.mapped(); @@ -882,6 +926,7 @@ void BwuManager::CancelRetryUpgradeAlarm(const std::string& endpoint_id) { } void BwuManager::CancelAllRetryUpgradeAlarms() { + NEARBY_LOG(INFO, "CancelAllRetryUpgradeAlarms invoked"); for (const auto& item : retry_upgrade_alarms_) { const std::string& endpoint_id = item.first; CancelRetryUpgradeAlarm(endpoint_id); From 7cc0cd614d16a141fda5456ac03a76abcc3d525d Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 12 Feb 2021 12:23:56 -0800 Subject: [PATCH 3/5] cl/356907843 Prevent closing socket twice. --- cpp/core/internal/mediums/webrtc/webrtc_socket.cc | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cpp/core/internal/mediums/webrtc/webrtc_socket.cc b/cpp/core/internal/mediums/webrtc/webrtc_socket.cc index 159cfaed..ff7f1f18 100644 --- a/cpp/core/internal/mediums/webrtc/webrtc_socket.cc +++ b/cpp/core/internal/mediums/webrtc/webrtc_socket.cc @@ -49,9 +49,8 @@ InputStream& WebRtcSocket::GetInputStream() { return pipe_.GetInputStream(); } OutputStream& WebRtcSocket::GetOutputStream() { return output_stream_; } void WebRtcSocket::Close() { - if (IsClosed()) return; + if (closed_.Set(true)) return; - closed_.Set(true); pipe_.GetInputStream().Close(); pipe_.GetOutputStream().Close(); data_channel_->Close(); From 420ee4927bdd74ed1bbf385041a4a1be8cb3906e Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 12 Feb 2021 12:25:59 -0800 Subject: [PATCH 4/5] cl/357006793 Close all channels on duplicate BwuPathAvailable frame. --- cpp/core/internal/bwu_manager.cc | 27 ++++++++++++++++++++++----- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 9c2f33fc..6cfeab48 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -401,12 +401,29 @@ void BwuManager::ProcessBwuPathAvailableEvent( if (FeatureFlags::GetInstance() .GetFlags() .disallow_out_of_order_bwu_avail_event) { - NEARBY_LOG(WARNING, - "BandwidthUpgradeManager is ignoring bandwidth upgrade for " - "endpoint %s because we're already upgrading bandwidth for " - "that endpoint. Something may have gone wrong, as it seems " - "we're out of sync with the remote device.", + NEARBY_LOG(ERROR, + "BandwidthUpgradeManager received a duplicate bandwidth " + "upgrade for endpoint %s. We're out of sync with the remote " + "device and cannot recover; closing all channels.", endpoint_id.c_str()); + + auto item = previous_endpoint_channels_.extract(endpoint_id); + if (!item.empty()) { + std::shared_ptr previous_endpoint_channel = + item.mapped(); + if (previous_endpoint_channel) { + previous_endpoint_channel->Close(DisconnectionReason::IO_ERROR); + } + } + std::shared_ptr new_channel = + channel_manager_->GetChannelForEndpoint(endpoint_id); + if (new_channel) { + // The upgraded channel never finished upgrading, and therefore is still + // paused. + new_channel->Resume(); + new_channel->Close(DisconnectionReason::IO_ERROR); + } + return; } } From 468be3186c6db8c9d143d295d1c391075fa5d339 Mon Sep 17 00:00:00 2001 From: hai007 Date: Fri, 12 Feb 2021 12:27:38 -0800 Subject: [PATCH 5/5] cl/357028940 Do not log raw binary data. --- cpp/core/internal/base_pcp_handler.cc | 2 +- cpp/core/internal/mediums/ble.cc | 3 +-- cpp/core/internal/p2p_cluster_pcp_handler.cc | 10 ++-------- 3 files changed, 4 insertions(+), 11 deletions(-) diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index 3bebcb01..63702034 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -90,7 +90,7 @@ Status BasePcpHandler::StartAdvertising(ClientProxy* client, response.Set({Status::kSuccess}); }); return WaitForResult( - absl::StrCat("StartAdvertising(", std::string(info.endpoint_info), ")"), + absl::StrCat("StartAdvertising(", service_id, ")"), client->GetClientId(), &response); } diff --git a/cpp/core/internal/mediums/ble.cc b/cpp/core/internal/mediums/ble.cc index d4cd2543..8b5bb0d5 100644 --- a/cpp/core/internal/mediums/ble.cc +++ b/cpp/core/internal/mediums/ble.cc @@ -69,8 +69,7 @@ bool Ble::StartAdvertising(const std::string& service_id, return false; } - NEARBY_LOGS(INFO) << "Turning on BLE advertising with advertisement bytes=" - << advertisement_bytes.data() << "(" + NEARBY_LOGS(INFO) << "Turning on BLE advertising (advertisement size=" << advertisement_bytes.size() << ")" << ", service id=" << service_id << ", fast advertisement service uuid=" diff --git a/cpp/core/internal/p2p_cluster_pcp_handler.cc b/cpp/core/internal/p2p_cluster_pcp_handler.cc index 2f7e32bd..03f06d0e 100644 --- a/cpp/core/internal/p2p_cluster_pcp_handler.cc +++ b/cpp/core/internal/p2p_cluster_pcp_handler.cc @@ -1000,10 +1000,8 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising( } NEARBY_LOG(INFO, - "P2pClusterPcpHandler::StartBleAdvertising: service=%s: " - "make advertisement; id=%s, name=%s", - service_id.c_str(), local_endpoint_id.c_str(), - std::string(local_endpoint_info).c_str()); + "P2pClusterPcpHandler::StartBleAdvertising: service=%s, id=%s", + service_id.c_str(), local_endpoint_id.c_str()); // Generate a BleAdvertisement. If a fast advertisement service UUID was // provided, create a fast BleAdvertisement. ByteArray advertisement_bytes; @@ -1031,10 +1029,6 @@ proto::connections::Medium P2pClusterPcpHandler::StartBleAdvertising( "BleAdvertisement failed"); ble_medium_.StopAcceptingConnections(service_id); return proto::connections::UNKNOWN_MEDIUM; - } else { - NEARBY_LOGS(INFO) << "P2pClusterPcpHandler::StartBleAdvertising: generate " - "BleAdvertisement succeeded; advertisement_bytes=" - << advertisement_bytes.data(); } NEARBY_LOG(