diff --git a/cpp/core_v2/core.h b/cpp/core_v2/core.h index 6f01a231..7c6f1b5f 100644 --- a/cpp/core_v2/core.h +++ b/cpp/core_v2/core.h @@ -18,6 +18,7 @@ #include #include "core_v2/internal/client_proxy.h" +#include "core_v2/internal/offline_service_controller.h" #include "core_v2/internal/service_controller.h" #include "core_v2/internal/service_controller_router.h" #include "core_v2/listeners.h" @@ -33,7 +34,8 @@ namespace connections { // This class defines the API of the Nearby Connections Core library. class Core { public: - explicit Core(std::function factory) + explicit Core(std::function factory = + []() { return new OfflineServiceController; }) : router_(factory) {} ~Core(); Core(Core&&) = default; diff --git a/cpp/core_v2/internal/BUILD b/cpp/core_v2/internal/BUILD index 5b7b31f1..e2e6c2db 100644 --- a/cpp/core_v2/internal/BUILD +++ b/cpp/core_v2/internal/BUILD @@ -27,6 +27,7 @@ cc_library( "internal_payload.cc", "internal_payload_factory.cc", "offline_frames.cc", + "offline_service_controller.cc", "p2p_cluster_pcp_handler.cc", "p2p_point_to_point_pcp_handler.cc", "p2p_star_pcp_handler.cc", @@ -51,6 +52,7 @@ cc_library( "internal_payload.h", "internal_payload_factory.h", "offline_frames.h", + "offline_service_controller.h", "p2p_cluster_pcp_handler.h", "p2p_point_to_point_pcp_handler.h", "p2p_star_pcp_handler.h", @@ -94,10 +96,12 @@ cc_library( name = "internal_test", testonly = True, srcs = [ + "offline_simulation_user.cc", "simulation_user.cc", ], hdrs = [ "mock_service_controller.h", + "offline_simulation_user.h", "simulation_user.h", ], visibility = [ @@ -110,6 +114,7 @@ cc_library( "//platform_v2/public:types", "//testing/base/public:gunit", "//absl/functional:bind_front", + "//absl/strings", ], ) @@ -127,6 +132,7 @@ cc_test( "endpoint_manager_test.cc", "internal_payload_factory_test.cc", "offline_frames_test.cc", + "offline_service_controller_test.cc", "p2p_cluster_pcp_handler_test.cc", "payload_manager_test.cc", "pcp_manager_test.cc", diff --git a/cpp/core_v2/internal/base_pcp_handler.cc b/cpp/core_v2/internal/base_pcp_handler.cc index 29886aad..30759bec 100644 --- a/cpp/core_v2/internal/base_pcp_handler.cc +++ b/cpp/core_v2/internal/base_pcp_handler.cc @@ -46,12 +46,9 @@ BasePcpHandler::BasePcpHandler(EndpointManager* endpoint_manager, pcp_(pcp) {} BasePcpHandler::~BasePcpHandler() { - // Unregister ourselves from the FrameProcessors. NEARBY_LOGS(INFO) << "BasePcpHandler: going down; strategy=" - << strategy_.GetName(); - endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE, - handle_); - + << strategy_.GetName() << "; handle=" << handle_; + DisconnectFromEndpointManager(); // Stop all the ongoing Runnables (as gracefully as possible). NEARBY_LOGS(INFO) << "BasePcpHandler: bringing down executors; strategy=" << strategy_.GetName(); @@ -61,8 +58,17 @@ BasePcpHandler::~BasePcpHandler() { << strategy_.GetName(); } +void BasePcpHandler::DisconnectFromEndpointManager() { + if (stop_.Set(true)) return; + NEARBY_LOGS(INFO) << "BasePcpHandler: Unregister from EPM; strategy=" + << strategy_.GetName() << "; handle=" << handle_; + // Unregister ourselves from EPM message dispatcher. + endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE, + handle_, true); +} + Status BasePcpHandler::StartAdvertising(ClientProxy* client, - const string& service_id, + const std::string& service_id, const ConnectionOptions& options, const ConnectionRequestInfo& info) { Future response; @@ -99,7 +105,7 @@ void BasePcpHandler::StopAdvertising(ClientProxy* client) { } Status BasePcpHandler::StartDiscovery(ClientProxy* client, - const string& service_id, + const std::string& service_id, const ConnectionOptions& options, const DiscoveryListener& listener) { Future response; @@ -136,7 +142,7 @@ void BasePcpHandler::StopDiscovery(ClientProxy* client) { WaitForLatch("stopDiscovery", &latch); } -void BasePcpHandler::WaitForLatch(const string& method_name, +void BasePcpHandler::WaitForLatch(const std::string& method_name, CountDownLatch* latch) { Exception await_exception = latch->Await(); if (!await_exception.Ok()) { @@ -146,7 +152,7 @@ void BasePcpHandler::WaitForLatch(const string& method_name, } } -Status BasePcpHandler::WaitForResult(const string& method_name, +Status BasePcpHandler::WaitForResult(const std::string& method_name, std::int64_t client_id, Future* future) { if (!future) { @@ -170,9 +176,10 @@ void BasePcpHandler::RunOnPcpHandlerThread(Runnable runnable) { EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() { return { .on_success_cb = - [this](const string& endpoint_id, + [this](const std::string& endpoint_id, std::unique_ptr ukey2, - const string& auth_token, const ByteArray& raw_auth_token) { + const std::string& auth_token, + const ByteArray& raw_auth_token) { RunOnPcpHandlerThread([this, endpoint_id, raw_ukey2 = ukey2.release(), auth_token, raw_auth_token]() mutable { @@ -182,7 +189,7 @@ EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() { }); }, .on_failure_cb = - [this](const string& endpoint_id, EndpointChannel* channel) { + [this](const std::string& endpoint_id, EndpointChannel* channel) { RunOnPcpHandlerThread([this, endpoint_id, channel]() { OnEncryptionFailureRunnable(endpoint_id, channel); }); @@ -191,8 +198,8 @@ EncryptionRunner::ResultListener BasePcpHandler::GetResultListener() { } void BasePcpHandler::OnEncryptionSuccessRunnable( - const string& endpoint_id, std::unique_ptr ukey2, - const string& auth_token, const ByteArray& raw_auth_token) { + const std::string& endpoint_id, std::unique_ptr ukey2, + const std::string& auth_token, const ByteArray& raw_auth_token) { // Quick fail if we've been removed from pending connections while we were // busy running UKEY2. auto it = pending_connections_.find(endpoint_id); @@ -217,7 +224,8 @@ void BasePcpHandler::OnEncryptionSuccessRunnable( // Set ourselves up so that we receive all acceptance/rejection messages handle_ = endpoint_manager_->RegisterFrameProcessor( - V1Frame::CONNECTION_RESPONSE, this); + V1Frame::CONNECTION_RESPONSE, + static_cast(this)); // Now we register our endpoint so that we can listen for both sides to // accept. @@ -239,7 +247,7 @@ void BasePcpHandler::OnEncryptionSuccessRunnable( } void BasePcpHandler::OnEncryptionFailureRunnable( - const string& endpoint_id, EndpointChannel* endpoint_channel) { + const std::string& endpoint_id, EndpointChannel* endpoint_channel) { auto it = pending_connections_.find(endpoint_id); if (it == pending_connections_.end()) { NEARBY_LOG(INFO, @@ -270,7 +278,7 @@ void BasePcpHandler::OnEncryptionFailureRunnable( } Status BasePcpHandler::RequestConnection(ClientProxy* client, - const string& endpoint_id, + const std::string& endpoint_id, const ConnectionRequestInfo& info) { Future result; RunOnPcpHandlerThread([this, client, &info, endpoint_id, &result]() { @@ -372,7 +380,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, } BasePcpHandler::DiscoveredEndpoint* BasePcpHandler::GetDiscoveredEndpoint( - const string& endpoint_id) { + const std::string& endpoint_id) { auto it = discovered_endpoints_.find(endpoint_id); if (it == discovered_endpoints_.end()) { return nullptr; @@ -414,15 +422,15 @@ bool BasePcpHandler::CanReceiveIncomingConnection(ClientProxy* client) const { } Exception BasePcpHandler::WriteConnectionRequestFrame( - EndpointChannel* endpoint_channel, const string& local_endpoint_id, - const string& local_endpoint_name, std::int32_t nonce, + EndpointChannel* endpoint_channel, const std::string& local_endpoint_id, + const std::string& local_endpoint_name, std::int32_t nonce, const std::vector& supported_mediums) { return endpoint_channel->Write(parser::ForConnectionRequest( local_endpoint_id, local_endpoint_name, nonce, supported_mediums)); } void BasePcpHandler::ProcessPreConnectionInitiationFailure( - const string& endpoint_id, EndpointChannel* channel, Status status, + const std::string& endpoint_id, EndpointChannel* channel, Status status, Future* result) { if (channel != nullptr) { channel->Close(); @@ -437,7 +445,7 @@ void BasePcpHandler::ProcessPreConnectionInitiationFailure( } void BasePcpHandler::ProcessPreConnectionResultFailure( - ClientProxy* client, const string& endpoint_id) { + ClientProxy* client, const std::string& endpoint_id) { auto item = pending_connections_.extract(endpoint_id); endpoint_manager_->DiscardEndpoint(client, endpoint_id); client->OnConnectionRejected(endpoint_id, {Status::kError}); @@ -462,7 +470,7 @@ bool BasePcpHandler::AutoUpgradeBandwidth() const { } Status BasePcpHandler::AcceptConnection( - ClientProxy* client, const string& endpoint_id, + ClientProxy* client, const std::string& endpoint_id, const PayloadListener& payload_listener) { Future response; RunOnPcpHandlerThread( @@ -517,7 +525,7 @@ Status BasePcpHandler::AcceptConnection( } Status BasePcpHandler::RejectConnection(ClientProxy* client, - const string& endpoint_id) { + const std::string& endpoint_id) { Future response; RunOnPcpHandlerThread([this, client, endpoint_id, &response]() { NEARBY_LOG(INFO, "RejectConnection: id=%s", endpoint_id.c_str()); @@ -573,7 +581,7 @@ Status BasePcpHandler::RejectConnection(ClientProxy* client, //} void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame, - const string& endpoint_id, + const std::string& endpoint_id, ClientProxy* client, proto::connections::Medium medium) { CountDownLatch latch(1); @@ -609,8 +617,12 @@ void BasePcpHandler::OnIncomingFrame(OfflineFrame& frame, } void BasePcpHandler::OnEndpointDisconnect(ClientProxy* client, - const string& endpoint_id, + const std::string& endpoint_id, CountDownLatch* barrier) { + if (stop_.Get()) { + if (barrier) barrier->CountDown(); + return; + } RunOnPcpHandlerThread([this, client, endpoint_id, barrier]() { auto item = pending_alarms_.find(endpoint_id); if (item != pending_alarms_.end()) { @@ -716,7 +728,7 @@ bool BasePcpHandler::IsPreferred( } Exception BasePcpHandler::OnIncomingConnection( - ClientProxy* client, const string& remote_device_name, + ClientProxy* client, const std::string& remote_device_name, std::unique_ptr channel, proto::connections::Medium medium) { absl::Time start_time = SystemClock::ElapsedRealtime(); @@ -811,7 +823,8 @@ Exception BasePcpHandler::OnIncomingConnection( return {Exception::kSuccess}; } -bool BasePcpHandler::BreakTie(ClientProxy* client, const string& endpoint_id, +bool BasePcpHandler::BreakTie(ClientProxy* client, + const std::string& endpoint_id, std::int32_t incoming_nonce, EndpointChannel* endpoint_channel) { auto it = pending_connections_.find(endpoint_id); @@ -849,7 +862,7 @@ bool BasePcpHandler::BreakTie(ClientProxy* client, const string& endpoint_id, } void BasePcpHandler::ProcessTieBreakLoss( - ClientProxy* client, const string& endpoint_id, + ClientProxy* client, const std::string& endpoint_id, BasePcpHandler::PendingConnectionInfo* info) { ProcessPreConnectionInitiationFailure(endpoint_id, info->channel.get(), {Status::kEndpointIoError}, @@ -859,7 +872,7 @@ void BasePcpHandler::ProcessTieBreakLoss( } void BasePcpHandler::InitiateBandwidthUpgrade( - ClientProxy* client, const string& endpoint_id, + ClientProxy* client, const std::string& endpoint_id, const std::vector& supported_mediums) { // When we successfully connect to a remote endpoint and a bandwidth upgrade // medium has not yet been decided, we'll pick the highest bandwidth medium @@ -908,7 +921,7 @@ proto::connections::Medium BasePcpHandler::ChooseBestUpgradeMedium( } void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client, - const string& endpoint_id, + const std::string& endpoint_id, 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. @@ -1046,12 +1059,12 @@ BasePcpHandler::PendingConnectionInfo::~PendingConnectionInfo() { } void BasePcpHandler::PendingConnectionInfo::LocalEndpointAcceptedConnection( - const string& endpoint_id, const PayloadListener& payload_listener) { + const std::string& endpoint_id, const PayloadListener& payload_listener) { client->LocalEndpointAcceptedConnection(endpoint_id, payload_listener); } void BasePcpHandler::PendingConnectionInfo::LocalEndpointRejectedConnection( - const string& endpoint_id) { + const std::string& endpoint_id) { client->LocalEndpointRejectedConnection(endpoint_id); } diff --git a/cpp/core_v2/internal/base_pcp_handler.h b/cpp/core_v2/internal/base_pcp_handler.h index 804ba8a9..8f3ec486 100644 --- a/cpp/core_v2/internal/base_pcp_handler.h +++ b/cpp/core_v2/internal/base_pcp_handler.h @@ -31,6 +31,7 @@ #include "core_v2/status.h" #include "proto/connections/offline_wire_formats.pb.h" #include "platform_v2/base/prng.h" +#include "platform_v2/public/atomic_boolean.h" #include "platform_v2/public/atomic_reference.h" #include "platform_v2/public/cancelable_alarm.h" #include "platform_v2/public/count_down_latch.h" @@ -153,6 +154,7 @@ class BasePcpHandler : public PcpHandler, Pcp GetPcp() const override { return pcp_; } Strategy GetStrategy() const override { return strategy_; } + void DisconnectFromEndpointManager(); protected: // The result of a call to startAdvertisingImpl() or startDiscoveryImpl(). @@ -437,6 +439,8 @@ class BasePcpHandler : public PcpHandler, // stops discovering because it might still be useful downstream of // discovery (eg: connection speed, etc.) ConnectionOptions discovery_options_; + + AtomicBoolean stop_{false}; Pcp pcp_; Strategy strategy_{PcpToStrategy(pcp_)}; Prng prng_; diff --git a/cpp/core_v2/internal/base_pcp_handler_test.cc b/cpp/core_v2/internal/base_pcp_handler_test.cc index ac85e21a..90cd2085 100644 --- a/cpp/core_v2/internal/base_pcp_handler_test.cc +++ b/cpp/core_v2/internal/base_pcp_handler_test.cc @@ -38,6 +38,7 @@ namespace { using ::location::nearby::proto::connections::Medium; using ::testing::_; +using ::testing::AtLeast; using ::testing::Invoke; using ::testing::MockFunction; using ::testing::Return; @@ -139,8 +140,7 @@ class MockContext { struct MockDiscoveredEndpoint : public MockPcpHandler::DiscoveredEndpoint { MockDiscoveredEndpoint(DiscoveredEndpoint endpoint, MockContext context) - : DiscoveredEndpoint(std::move(endpoint)), - context(std::move(context)) {} + : DiscoveredEndpoint(std::move(endpoint)), context(std::move(context)) {} MockContext context; }; @@ -324,148 +324,144 @@ class BasePcpHandlerTest : public ::testing::Test { }; TEST_F(BasePcpHandlerTest, ConstructorDestructorWorks) { - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); SUCCEED(); } TEST_F(BasePcpHandlerTest, StartAdvertisingChangesState) { - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartAdvertising(client.get(), pcp_handler.get()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartAdvertising(&client, &pcp_handler); } TEST_F(BasePcpHandlerTest, StopAdvertisingChangesState) { - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartAdvertising(client.get(), pcp_handler.get()); - EXPECT_CALL(*pcp_handler, StopAdvertisingImpl(client.get())).Times(1); - EXPECT_TRUE(client->IsAdvertising()); - pcp_handler->StopAdvertising(client.get()); - EXPECT_FALSE(client->IsAdvertising()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartAdvertising(&client, &pcp_handler); + EXPECT_CALL(pcp_handler, StopAdvertisingImpl(&client)).Times(1); + EXPECT_TRUE(client.IsAdvertising()); + pcp_handler.StopAdvertising(&client); + EXPECT_FALSE(client.IsAdvertising()); } TEST_F(BasePcpHandlerTest, StartDiscoveryChangesState) { - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartDiscovery(&client, &pcp_handler); } TEST_F(BasePcpHandlerTest, StopDiscoveryChangesState) { - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); - EXPECT_CALL(*pcp_handler, StopDiscoveryImpl(client.get())).Times(1); - EXPECT_TRUE(client->IsDiscovering()); - pcp_handler->StopDiscovery(client.get()); - EXPECT_FALSE(client->IsDiscovering()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartDiscovery(&client, &pcp_handler); + EXPECT_CALL(pcp_handler, StopDiscoveryImpl(&client)).Times(1); + EXPECT_TRUE(client.IsDiscovering()); + pcp_handler.StopDiscovery(&client); + EXPECT_FALSE(client.IsDiscovering()); } TEST_F(BasePcpHandlerTest, RequestConnectionChangesState) { std::string endpoint_id{"1234"}; - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartDiscovery(&client, &pcp_handler); auto channel_pair = SetupConnection(pipe_a_, pipe_b_); + auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; - RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(), - client.get(), pcp_handler.get()); + EXPECT_CALL(*channel_a, CloseImpl).Times(1); + EXPECT_CALL(*channel_b, CloseImpl).Times(1); + EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0)); + RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client, + &pcp_handler); NEARBY_LOG(INFO, "RequestConnection complete"); channel_b->Close(); + pcp_handler.DisconnectFromEndpointManager(); } TEST_F(BasePcpHandlerTest, AcceptConnectionChangesState) { std::string endpoint_id{"1234"}; - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartDiscovery(&client, &pcp_handler); auto channel_pair = SetupConnection(pipe_a_, pipe_b_); + auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; - RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(), - client.get(), pcp_handler.get()); + EXPECT_CALL(*channel_a, CloseImpl).Times(1); + EXPECT_CALL(*channel_b, CloseImpl).Times(1); + RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client, + &pcp_handler); NEARBY_LOG(INFO, "Attempting to accept connection: id=%s", endpoint_id.c_str()); - EXPECT_EQ(pcp_handler->AcceptConnection(client.get(), endpoint_id, {}), + EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}), Status{Status::kSuccess}); - NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str()); + EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0)); + NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); + pcp_handler.DisconnectFromEndpointManager(); } TEST_F(BasePcpHandlerTest, RejectConnectionChangesState) { std::string endpoint_id{"1234"}; - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartDiscovery(&client, &pcp_handler); auto channel_pair = SetupConnection(pipe_a_, pipe_b_); auto& channel_b = channel_pair.second; EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(1); RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(), - client.get(), pcp_handler.get()); - NEARBY_LOG(INFO, "Attempting to reject connection: id=%s", - endpoint_id.c_str()); - EXPECT_EQ(pcp_handler->RejectConnection(client.get(), endpoint_id), + &client, &pcp_handler); + NEARBY_LOGS(INFO) << "Attempting to reject connection: id=" << endpoint_id; + EXPECT_EQ(pcp_handler.RejectConnection(&client, endpoint_id), Status{Status::kSuccess}); - NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str()); + NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); + pcp_handler.DisconnectFromEndpointManager(); } TEST_F(BasePcpHandlerTest, OnIncomingFrameChangesState) { std::string endpoint_id{"1234"}; - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); + ClientProxy client; + EndpointChannelManager ecm; + EndpointManager em(&ecm); + MockPcpHandler pcp_handler(&em, &ecm); + StartDiscovery(&client, &pcp_handler); auto channel_pair = SetupConnection(pipe_a_, pipe_b_); + auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; - RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(), - client.get(), pcp_handler.get()); - NEARBY_LOG(INFO, "Attempting to accept connection: id=%s", - endpoint_id.c_str()); + EXPECT_CALL(*channel_a, CloseImpl).Times(1); + EXPECT_CALL(*channel_b, CloseImpl).Times(1); + RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), &client, + &pcp_handler); + NEARBY_LOGS(INFO) << "Attempting to accept connection: id=" << endpoint_id; EXPECT_CALL(mock_connection_listener_.accepted_cb, Call).Times(1); - EXPECT_EQ(pcp_handler->AcceptConnection(client.get(), endpoint_id, {}), + EXPECT_CALL(mock_connection_listener_.disconnected_cb, Call) + .Times(AtLeast(0)); + EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}), Status{Status::kSuccess}); NEARBY_LOG(INFO, "Simulating remote accept: id=%s", endpoint_id.c_str()); auto frame = parser::FromBytes(parser::ForConnectionResponse(Status::kSuccess)); - pcp_handler->OnIncomingFrame(frame.result(), endpoint_id, client.get(), - Medium::BLE); - NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str()); + pcp_handler.OnIncomingFrame(frame.result(), endpoint_id, &client, + Medium::BLE); + NEARBY_LOGS(INFO) << "Closing connection: id=" << endpoint_id; channel_b->Close(); -} - -TEST_F(BasePcpHandlerTest, OnEndpointDisconnectChangesState) { - std::string endpoint_id{"1234"}; - auto client = std::make_unique(); - auto ecm = std::make_unique(); - auto em = std::make_unique(ecm.get()); - auto pcp_handler = std::make_unique(em.get(), ecm.get()); - StartDiscovery(client.get(), pcp_handler.get()); - auto channel_pair = SetupConnection(pipe_a_, pipe_b_); - auto& channel_b = channel_pair.second; - EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(1); - RequestConnection(endpoint_id, std::move(channel_pair.first), channel_b.get(), - client.get(), pcp_handler.get()); - NEARBY_LOG(INFO, "Simulating disconnect event: id=%s", endpoint_id.c_str()); - CountDownLatch latch(1); - pcp_handler->OnEndpointDisconnect(client.get(), endpoint_id, &latch); - channel_b->Close(); - EXPECT_TRUE(latch.Await(absl::Milliseconds(5000)).result()); + pcp_handler.DisconnectFromEndpointManager(); } TEST_F(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { @@ -478,15 +474,20 @@ TEST_F(BasePcpHandlerTest, DestructorIsCalledOnProtocolEndpoint) { MockPcpHandler pcp_handler(&em, &ecm); StartDiscovery(&client, &pcp_handler); auto channel_pair = SetupConnection(pipe_a_, pipe_b_); + auto& channel_a = channel_pair.first; auto& channel_b = channel_pair.second; - RequestConnection(endpoint_id, std::move(channel_pair.first), - channel_b.get(), &client, &pcp_handler, &destroyed_flag); + EXPECT_CALL(*channel_a, CloseImpl).Times(1); + EXPECT_CALL(*channel_b, CloseImpl).Times(1); + RequestConnection(endpoint_id, std::move(channel_a), channel_b.get(), + &client, &pcp_handler, &destroyed_flag); NEARBY_LOG(INFO, "Attempting to accept connection: id=%s", endpoint_id.c_str()); EXPECT_EQ(pcp_handler.AcceptConnection(&client, endpoint_id, {}), Status{Status::kSuccess}); + EXPECT_CALL(mock_connection_listener_.rejected_cb, Call).Times(AtLeast(0)); NEARBY_LOG(INFO, "Closing connection: id=%s", endpoint_id.c_str()); channel_b->Close(); + pcp_handler.DisconnectFromEndpointManager(); } EXPECT_TRUE(destroyed_flag.load()); } diff --git a/cpp/core_v2/internal/endpoint_manager.cc b/cpp/core_v2/internal/endpoint_manager.cc index e610450c..9ae32550 100644 --- a/cpp/core_v2/internal/endpoint_manager.cc +++ b/cpp/core_v2/internal/endpoint_manager.cc @@ -214,6 +214,7 @@ EndpointManager::EndpointManager(EndpointChannelManager* manager) : channel_manager_(manager) {} EndpointManager::~EndpointManager() { + NEARBY_LOG(INFO, "EndpointManager going down"); CountDownLatch latch(1); RunOnEndpointManagerThread([this, &latch]() { NEARBY_LOG(INFO, "Bringing down endpoints"); @@ -222,10 +223,8 @@ EndpointManager::~EndpointManager() { EndpointState& state = item.second; // This will close the channel; all workers will sense that and // terminate. - NEARBY_LOG(INFO, "Bringing down endpoint channels: id=%s", - endpoint_id.c_str()); - WaitForEndpointDisconnectionProcessing(state.client, endpoint_id); channel_manager_->UnregisterChannelForEndpoint(endpoint_id); + state.barrier.Await(); } latch.CountDown(); }); @@ -250,9 +249,12 @@ EndpointManager::RegisterFrameProcessor( RunOnEndpointManagerThread([this, frame_type, &latch, processor]() { auto it = frame_processors_.find(frame_type); if (it != frame_processors_.end()) { - NEARBY_LOG(INFO, "Frame processor found, updated; type=%d", frame_type); + NEARBY_LOGS(INFO) << "Frame processor found: updated; type=" << frame_type + << "; processor=" << processor << "; self=" << this; it->second = processor; } else { + NEARBY_LOGS(INFO) << "Frame processor added; type=" << frame_type + << "; processor=" << processor << "; self=" << this; frame_processors_.emplace(frame_type, processor); } latch.CountDown(); @@ -263,14 +265,22 @@ EndpointManager::RegisterFrameProcessor( void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type, const void* handle, bool sync) { + NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: handle=" << handle; if (handle == nullptr) return; CountDownLatch latch(1); RunOnEndpointManagerThread([this, frame_type, handle, &latch, sync]() { auto it = frame_processors_.find(frame_type); - if (it == frame_processors_.end()) return; + if (it == frame_processors_.end()) { + NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: handle=" + << handle; + if (sync) latch.CountDown(); + return; + } + NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: handle=" << handle; if (it->second == handle) { frame_processors_.erase(it); - NEARBY_LOG(INFO, "Unregistered: type=%d", frame_type); + NEARBY_LOGS(INFO) << "Unregistered: type=" << frame_type + << "; processor=" << handle << "; self=" << this; } else { NEARBY_LOG(INFO, "Failed to unregister: type=%d; handle mismatch: passed=%p, " @@ -281,7 +291,8 @@ void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type, }); if (sync) { latch.Await(); - NEARBY_LOG(INFO, "Unregistered: [sync done] type=%d", frame_type); + NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type + << "; processor=" << handle << "; self=" << this; } } @@ -308,10 +319,13 @@ void EndpointManager::EnsureWorkersTerminated(const std::string& endpoint_id) { // If another instance of data and keep-alive handlers is running, it will // terminate soon; we should block until it happens. EndpointState& endpoint_state = item->second; - NEARBY_LOG(INFO, "Waiting for workers to terminate for endpoint_id='%s'", - endpoint_id.c_str()); + NEARBY_LOGS(INFO) << "Waiting for workers to terminate for id: " + << endpoint_id; endpoint_state.barrier.Await(); endpoints_.erase(item); + NEARBY_LOGS(INFO) << "Workers terminated for id: " << endpoint_id; + } else { + NEARBY_LOGS(INFO) << "EndpointState not found for id: " << endpoint_id; } } @@ -392,8 +406,8 @@ void EndpointManager::UnregisterEndpoint(ClientProxy* client, const std::string& endpoint_id) { CountDownLatch latch(1); RunOnEndpointManagerThread([this, client, endpoint_id, &latch]() { - channel_manager_->UnregisterChannelForEndpoint(endpoint_id); - RemoveEndpoint(client, endpoint_id, /*notify=*/false); + RemoveEndpoint(client, endpoint_id, + client->IsConnectedToEndpoint(endpoint_id)); latch.CountDown(); }); latch.Await(); @@ -405,7 +419,6 @@ void EndpointManager::UnregisterEndpoint(ClientProxy* client, void EndpointManager::DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id) { RunOnEndpointManagerThread([this, client, endpoint_id]() { - channel_manager_->UnregisterChannelForEndpoint(endpoint_id); RemoveEndpoint(client, endpoint_id, /*notify=*/ client->IsConnectedToEndpoint(endpoint_id)); @@ -449,25 +462,50 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client, // should be no further interactions with the endpoint. // (See b/37352254 for history) WaitForEndpointDisconnectionProcessing(client, endpoint_id); - EnsureWorkersTerminated(endpoint_id); client->OnDisconnected(endpoint_id, notify); - NEARBY_LOG(INFO, "Removed endpoint; id=%s", - endpoint_id.c_str()); + NEARBY_LOG(INFO, "Removed endpoint; id=%s", endpoint_id.c_str()); } } // @EndpointManagerThread void EndpointManager::WaitForEndpointDisconnectionProcessing( ClientProxy* client, const std::string& endpoint_id) { - CountDownLatch barrier(frame_processors_.size()); + NEARBY_LOGS(INFO) << "Wait: client=" << client << "; id=" << endpoint_id; + auto total_size = frame_processors_.size(); + NEARBY_LOGS(INFO) << "Total frame processors: " << total_size; + if (!total_size) return; + CountDownLatch barrier(total_size); + int valid = 0; for (auto& item : frame_processors_) { - auto& processor = item.second; - processor->OnEndpointDisconnect(client, endpoint_id, &barrier); + auto* processor = item.second; + NEARBY_LOGS(INFO) << "processor=" << processor << "; type=" << item.first; + if (processor) { + valid++; + processor->OnEndpointDisconnect(client, endpoint_id, &barrier); + } else { + barrier.CountDown(); + } } - barrier.Await(kProcessEndpointDisconnectionTimeout); + if (!valid) { + NEARBY_LOGS(INFO) << "No valid frame processors."; + return; + } else { + NEARBY_LOGS(INFO) << "Valid frame processors: " << valid; + } + + NEARBY_LOGS(INFO) << "Waiting for " << valid + << " frame processors to disconnect from: " << endpoint_id; + if (!barrier.Await(kProcessEndpointDisconnectionTimeout).result()) { + NEARBY_LOGS(INFO) << "Failed to disconnect frame processors from: " + << endpoint_id; + } else { + NEARBY_LOGS(INFO) + << "Finished waiting for frame processors to disconnect from: " + << endpoint_id; + } } std::vector EndpointManager::SendTransferFrameBytes( diff --git a/cpp/core_v2/internal/mediums/webrtc.h b/cpp/core_v2/internal/mediums/webrtc.h index e94bcbf4..a32bb55e 100644 --- a/cpp/core_v2/internal/mediums/webrtc.h +++ b/cpp/core_v2/internal/mediums/webrtc.h @@ -152,11 +152,11 @@ class WebRtc { std::vector<::location::nearby::mediums::IceCandidate> pending_local_ice_candidates_ ABSL_GUARDED_BY(mutex_); + WebRtcMedium medium_; std::unique_ptr connection_flow_; std::unique_ptr signaling_messenger_ ABSL_GUARDED_BY(mutex_); WebRtcSocketWrapper socket_ ABSL_GUARDED_BY(mutex_); - WebRtcMedium medium_; SingleThreadExecutor single_thread_executor_; }; diff --git a/cpp/core_v2/internal/offline_service_controller.cc b/cpp/core_v2/internal/offline_service_controller.cc new file mode 100644 index 00000000..7465fc96 --- /dev/null +++ b/cpp/core_v2/internal/offline_service_controller.cc @@ -0,0 +1,93 @@ +// Copyright 2020 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 "core_v2/internal/offline_service_controller.h" + +#include + +namespace location { +namespace nearby { +namespace connections { + +OfflineServiceController::~OfflineServiceController() { + Stop(); +} + +void OfflineServiceController::Stop() { + if (stop_.Set(true)) return; + payload_manager_.DisconnectFromEndpointManager(); + pcp_manager_.DisconnectFromEndpointManager(); +} + +Status OfflineServiceController::StartAdvertising( + ClientProxy* client, const std::string& service_id, + const ConnectionOptions& options, const ConnectionRequestInfo& info) { + return pcp_manager_.StartAdvertising(client, service_id, options, info); +} + +void OfflineServiceController::StopAdvertising(ClientProxy* client) { + pcp_manager_.StopAdvertising(client); +} + +Status OfflineServiceController::StartDiscovery( + ClientProxy* client, const std::string& service_id, + const ConnectionOptions& options, const DiscoveryListener& listener) { + return pcp_manager_.StartDiscovery(client, service_id, options, listener); +} + +void OfflineServiceController::StopDiscovery(ClientProxy* client) { + pcp_manager_.StopDiscovery(client); +} + +Status OfflineServiceController::RequestConnection( + ClientProxy* client, const std::string& endpoint_id, + const ConnectionRequestInfo& info) { + return pcp_manager_.RequestConnection(client, endpoint_id, info); +} + +Status OfflineServiceController::AcceptConnection( + ClientProxy* client, const std::string& endpoint_id, + const PayloadListener& listener) { + return pcp_manager_.AcceptConnection(client, endpoint_id, listener); +} + +Status OfflineServiceController::RejectConnection( + ClientProxy* client, const std::string& endpoint_id) { + return pcp_manager_.RejectConnection(client, endpoint_id); +} + +void OfflineServiceController::InitiateBandwidthUpgrade( + ClientProxy* client, const std::string& endpoint_id) { + // TODO(apolyudov): implement. +} + +void OfflineServiceController::SendPayload( + ClientProxy* client, const std::vector& endpoint_ids, + Payload payload) { + payload_manager_.SendPayload(client, endpoint_ids, std::move(payload)); +} + +Status OfflineServiceController::CancelPayload(ClientProxy* client, + std::int64_t payload_id) { + return payload_manager_.CancelPayload(client, payload_id); +} + +void OfflineServiceController::DisconnectFromEndpoint( + ClientProxy* client, const std::string& endpoint_id) { + endpoint_manager_.UnregisterEndpoint(client, endpoint_id); +} + +} // namespace connections +} // namespace nearby +} // namespace location diff --git a/cpp/core_v2/internal/offline_service_controller.h b/cpp/core_v2/internal/offline_service_controller.h new file mode 100644 index 00000000..a4855db2 --- /dev/null +++ b/cpp/core_v2/internal/offline_service_controller.h @@ -0,0 +1,94 @@ +// Copyright 2020 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 CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_ +#define CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_ + +#include +#include +#include + +#include "core_v2/internal/client_proxy.h" +#include "core_v2/internal/endpoint_channel_manager.h" +#include "core_v2/internal/endpoint_manager.h" +#include "core_v2/internal/mediums/mediums.h" +#include "core_v2/internal/payload_manager.h" +#include "core_v2/internal/pcp_manager.h" +#include "core_v2/internal/service_controller.h" +#include "core_v2/listeners.h" +#include "core_v2/options.h" +#include "core_v2/payload.h" +#include "core_v2/status.h" + +namespace location { +namespace nearby { +namespace connections { + +class OfflineServiceController : public ServiceController { + public: + OfflineServiceController() = default; + ~OfflineServiceController() override; + + Status StartAdvertising(ClientProxy* client, + const std::string& service_id, + const ConnectionOptions& options, + const ConnectionRequestInfo& info) override; + void StopAdvertising(ClientProxy* client) override; + + Status StartDiscovery(ClientProxy* client, + const std::string& service_id, + const ConnectionOptions& options, + const DiscoveryListener& listener) override; + void StopDiscovery(ClientProxy* client) override; + + Status RequestConnection(ClientProxy* client, + const std::string& endpoint_id, + const ConnectionRequestInfo& info) override; + Status AcceptConnection(ClientProxy* client, + const std::string& endpoint_id, + const PayloadListener& listener) override; + Status RejectConnection(ClientProxy* client, + const std::string& endpoint_id) override; + + void InitiateBandwidthUpgrade(ClientProxy* client, + const std::string& endpoint_id) override; + + void SendPayload(ClientProxy* client, + const std::vector& endpoint_ids, + Payload payload) override; + Status CancelPayload(ClientProxy* client, + Payload::Id payload_id) override; + + void DisconnectFromEndpoint(ClientProxy* client, + const std::string& endpoint_id) override; + + void Stop(); + + private: + // Note that the order of declaration of these is crucial, because we depend + // on the destructors running (strictly) in the reverse order; a deviation + // from that will lead to crashes at runtime. + AtomicBoolean stop_{false}; + Mediums mediums_; + EndpointChannelManager channel_manager_; + EndpointManager endpoint_manager_{&channel_manager_}; + PayloadManager payload_manager_{endpoint_manager_}; + PcpManager pcp_manager_{mediums_, channel_manager_, endpoint_manager_}; +}; + +} // namespace connections +} // namespace nearby +} // namespace location + +#endif // CORE_V2_INTERNAL_OFFLINE_SERVICE_CONTROLLER_H_ diff --git a/cpp/core_v2/internal/offline_service_controller_test.cc b/cpp/core_v2/internal/offline_service_controller_test.cc new file mode 100644 index 00000000..2d4487ea --- /dev/null +++ b/cpp/core_v2/internal/offline_service_controller_test.cc @@ -0,0 +1,335 @@ +// Copyright 2020 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 "core_v2/internal/offline_service_controller.h" + +#include "core_v2/internal/offline_simulation_user.h" +#include "platform_v2/base/medium_environment.h" +#include "platform_v2/base/output_stream.h" +#include "platform_v2/public/count_down_latch.h" +#include "platform_v2/public/logging.h" +#include "platform_v2/public/pipe.h" +#include "platform_v2/public/system_clock.h" +#include "gmock/gmock.h" +#include "gtest/gtest.h" + +namespace location { +namespace nearby { +namespace connections { +namespace { + +using ::testing::Eq; + +constexpr absl::string_view kServiceId = "service-id"; +constexpr absl::string_view kDeviceA = "device-a"; +constexpr absl::string_view kDeviceB = "device-b"; +constexpr absl::string_view kMessage = "message"; +constexpr absl::Duration kProgressTimeout = absl::Milliseconds(1000); +constexpr absl::Duration kDefaultTimeout = absl::Milliseconds(1000); +constexpr absl::Duration kDisconnectTimeout = absl::Milliseconds(15000); + +class OfflineServiceControllerTest : public ::testing::Test { + protected: + OfflineServiceControllerTest() { env_.Stop(); } + + bool SetupConnection(OfflineSimulationUser& user_a, + OfflineSimulationUser& user_b) { + user_a.StartAdvertising(std::string(kServiceId), &connect_latch_); + user_b.StartDiscovery(std::string(kServiceId), &discover_latch_); + EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); + EXPECT_EQ(user_b.GetDiscovered().service_id, kServiceId); + EXPECT_EQ(user_b.GetDiscovered().endpoint_name, user_a.GetName()); + EXPECT_FALSE(user_b.GetDiscovered().endpoint_id.empty()); + NEARBY_LOG(INFO, "EP-B: [discovered] %s", + user_b.GetDiscovered().endpoint_id.c_str()); + user_b.RequestConnection(&connect_latch_); + EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); + EXPECT_FALSE(user_a.GetDiscovered().endpoint_id.empty()); + NEARBY_LOG(INFO, "EP-A: [discovered] %s", + user_a.GetDiscovered().endpoint_id.c_str()); + NEARBY_LOG(INFO, "Both users discovered their peers."); + user_a.AcceptConnection(&accept_latch_); + user_b.AcceptConnection(&accept_latch_); + EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result()); + NEARBY_LOG(INFO, "Both users reached connected state."); + return user_a.IsConnected() && user_b.IsConnected(); + } + + CountDownLatch discover_latch_{1}; + CountDownLatch connect_latch_{2}; + CountDownLatch accept_latch_{2}; + CountDownLatch payload_latch_{1}; + MediumEnvironment& env_ = MediumEnvironment::Instance(); +}; + +TEST_F(OfflineServiceControllerTest, CanCreateOne) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanCreateMany) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanStartAdvertising) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_FALSE(user_a.IsAdvertising()); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(user_a.IsAdvertising()); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanStartDiscoveryBeforeAdvertising) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_FALSE(user_b.IsDiscovering()); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(user_b.IsDiscovering()); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanStartDiscoveryAfterAdvertising) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_FALSE(user_b.IsDiscovering()); + EXPECT_FALSE(user_b.IsAdvertising()); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), + Eq(Status{Status::kSuccess})); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(user_a.IsAdvertising()); + EXPECT_TRUE(user_b.IsDiscovering()); + EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanStopAdvertising) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_FALSE(user_a.IsAdvertising()); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(user_a.IsAdvertising()); + user_a.StopAdvertising(); + EXPECT_FALSE(user_a.IsAdvertising()); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(user_b.IsDiscovering()); + EXPECT_FALSE(discover_latch_.Await(kDefaultTimeout).result()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanStopDiscovery) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_FALSE(user_b.IsDiscovering()); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(user_b.IsDiscovering()); + user_b.StopDiscovery(); + EXPECT_FALSE(user_b.IsDiscovering()); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), nullptr), + Eq(Status{Status::kSuccess})); + EXPECT_FALSE(discover_latch_.Await(kDefaultTimeout).result()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanConnect) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); + EXPECT_THAT(user_b.RequestConnection(&connect_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanAcceptConnection) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); + EXPECT_THAT(user_b.RequestConnection(&connect_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); + EXPECT_THAT(user_a.AcceptConnection(&accept_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_THAT(user_b.AcceptConnection(&accept_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(accept_latch_.Await(kDefaultTimeout).result()); + EXPECT_TRUE(user_a.IsConnected()); + EXPECT_TRUE(user_b.IsConnected()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanRejectConnection) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + CountDownLatch reject_latch(1); + EXPECT_THAT(user_a.StartAdvertising(std::string(kServiceId), &connect_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_THAT(user_b.StartDiscovery(std::string(kServiceId), &discover_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(discover_latch_.Await(kDefaultTimeout).result()); + EXPECT_THAT(user_b.RequestConnection(&connect_latch_), + Eq(Status{Status::kSuccess})); + EXPECT_TRUE(connect_latch_.Await(kDefaultTimeout).result()); + user_a.ExpectRejectedConnection(reject_latch); + EXPECT_THAT(user_b.RejectConnection(nullptr), Eq(Status{Status::kSuccess})); + EXPECT_TRUE(reject_latch.Await(kDefaultTimeout).result()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanSendBytePayload) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + ASSERT_TRUE(SetupConnection(user_a, user_b)); + ByteArray message(std::string{kMessage}); + user_a.SendPayload(Payload(message)); + user_b.ExpectPayload(payload_latch_); + EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); + EXPECT_EQ(user_b.GetPayload().AsBytes(), message); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanSendStreamPayload) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + ASSERT_TRUE(SetupConnection(user_a, user_b)); + ByteArray message(std::string{kMessage}); + auto pipe = std::make_shared(); + OutputStream& tx = pipe->GetOutputStream(); + user_a.SendPayload(Payload([pipe]() -> InputStream& { + return pipe->GetInputStream(); // NOLINT + })); + user_b.ExpectPayload(payload_latch_); + tx.Write(message); + EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); + EXPECT_NE(user_b.GetPayload().AsStream(), nullptr); + InputStream& rx = *user_b.GetPayload().AsStream(); + ASSERT_TRUE(user_b.WaitForProgress( + [size = message.size()](const PayloadProgressInfo& info) -> bool { + return info.bytes_transferred >= size; + }, + kProgressTimeout)); + EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanCancelStreamPayload) { + env_.Start(); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + ASSERT_TRUE(SetupConnection(user_a, user_b)); + ByteArray message(std::string{kMessage}); + auto pipe = std::make_shared(); + OutputStream& tx = pipe->GetOutputStream(); + user_a.SendPayload(Payload([pipe]() -> InputStream& { + return pipe->GetInputStream(); // NOLINT + })); + user_b.ExpectPayload(payload_latch_); + tx.Write(message); + EXPECT_TRUE(payload_latch_.Await(kDefaultTimeout).result()); + EXPECT_NE(user_b.GetPayload().AsStream(), nullptr); + InputStream& rx = *user_b.GetPayload().AsStream(); + ASSERT_TRUE(user_b.WaitForProgress( + [size = message.size()](const PayloadProgressInfo& info) -> bool { + return info.bytes_transferred >= size; + }, + kProgressTimeout)); + EXPECT_EQ(rx.Read(Pipe::kChunkSize).result(), message); + user_b.CancelPayload(); + int count = 0; + while (true) { + count++; + if (!tx.Write(message).Ok()) break; + SystemClock::Sleep(kDefaultTimeout); + } + EXPECT_TRUE(user_a.WaitForProgress( + [](const PayloadProgressInfo& info) -> bool { + return info.status == PayloadProgressInfo::Status::kCanceled; + }, + kProgressTimeout)); + EXPECT_LT(count, 10); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +TEST_F(OfflineServiceControllerTest, CanDisconnect) { + env_.Start(); + CountDownLatch disconnect_latch(1); + OfflineSimulationUser user_a(kDeviceA); + OfflineSimulationUser user_b(kDeviceB); + ASSERT_TRUE(SetupConnection(user_a, user_b)); + NEARBY_LOGS(INFO) << "Disconnecting"; + user_b.ExpectDisconnect(disconnect_latch); + user_b.Disconnect(); + EXPECT_TRUE(disconnect_latch.Await(kDisconnectTimeout).result()); + NEARBY_LOGS(INFO) << "Disconnected"; + EXPECT_FALSE(user_b.IsConnected()); + user_a.Stop(); + user_b.Stop(); + env_.Stop(); +} + +} // namespace +} // namespace connections +} // namespace nearby +} // namespace location diff --git a/cpp/core_v2/internal/offline_simulation_user.cc b/cpp/core_v2/internal/offline_simulation_user.cc new file mode 100644 index 00000000..1a58f117 --- /dev/null +++ b/cpp/core_v2/internal/offline_simulation_user.cc @@ -0,0 +1,191 @@ +// Copyright 2020 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 "core_v2/internal/offline_simulation_user.h" + +#include "core_v2/listeners.h" +#include "platform_v2/public/count_down_latch.h" +#include "platform_v2/public/system_clock.h" +#include "absl/functional/bind_front.h" + +namespace location { +namespace nearby { +namespace connections { + +void OfflineSimulationUser::OnConnectionInitiated( + const std::string& endpoint_id, const ConnectionResponseInfo& info, + bool is_outgoing) { + if (is_outgoing) { + NEARBY_LOG(INFO, "RequestConnection: initiated_cb called"); + } else { + NEARBY_LOG(INFO, "StartAdvertising: initiated_cb called"); + discovered_ = DiscoveredInfo{ + .endpoint_id = endpoint_id, + .endpoint_name = name_, + .service_id = service_id_, + }; + } + if (initiated_latch_) initiated_latch_->CountDown(); +} + +void OfflineSimulationUser::OnConnectionAccepted( + const std::string& endpoint_id) { + if (accept_latch_) accept_latch_->CountDown(); +} + +void OfflineSimulationUser::OnConnectionRejected(const std::string& endpoint_id, + Status status) { + if (reject_latch_) reject_latch_->CountDown(); +} + +void OfflineSimulationUser::OnEndpointDisconnect( + const std::string& endpoint_id) { + NEARBY_LOGS(INFO) << "OnEndpointDisconnect: self=" << this + << "; id=" << endpoint_id; + if (disconnect_latch_) disconnect_latch_->CountDown(); +} + +void OfflineSimulationUser::OnEndpointFound(const std::string& endpoint_id, + const std::string& endpoint_name, + const std::string& service_id) { + NEARBY_LOG(INFO, "Device discovered: id=%s", endpoint_id.c_str()); + discovered_ = DiscoveredInfo{ + .endpoint_id = endpoint_id, + .endpoint_name = endpoint_name, + .service_id = service_id, + }; + if (found_latch_) found_latch_->CountDown(); +} + +void OfflineSimulationUser::OnEndpointLost(const std::string& endpoint_id) { + if (lost_latch_) lost_latch_->CountDown(); +} + +void OfflineSimulationUser::OnPayload(const std::string& endpoint_id, + Payload payload) { + payload_ = std::move(payload); + if (payload_latch_) payload_latch_->CountDown(); +} + +void OfflineSimulationUser::OnPayloadProgress(const std::string& endpoint_id, + const PayloadProgressInfo& info) { + MutexLock lock(&progress_mutex_); + progress_info_ = info; + if (future_ && predicate_ && predicate_(info)) future_->Set(true); +} + +bool OfflineSimulationUser::WaitForProgress( + std::function predicate, + absl::Duration timeout) { + Future future; + { + MutexLock lock(&progress_mutex_); + if (predicate(progress_info_)) return true; + future_ = &future; + predicate_ = std::move(predicate); + } + auto response = future.Get(timeout); + { + MutexLock lock(&progress_mutex_); + future_ = nullptr; + predicate_ = nullptr; + } + return response.ok() && response.result(); +} + +Status OfflineSimulationUser::StartAdvertising(const std::string& service_id, + CountDownLatch* latch) { + initiated_latch_ = latch; + service_id_ = service_id; + ConnectionListener listener = { + .initiated_cb = + std::bind(&OfflineSimulationUser::OnConnectionInitiated, this, + std::placeholders::_1, std::placeholders::_2, false), + .accepted_cb = + absl::bind_front(&OfflineSimulationUser::OnConnectionAccepted, this), + .rejected_cb = + absl::bind_front(&OfflineSimulationUser::OnConnectionRejected, this), + .disconnected_cb = + absl::bind_front(&OfflineSimulationUser::OnEndpointDisconnect, this), + }; + return ctrl_.StartAdvertising(&client_, service_id_, options_, + { + .name = name_, + .listener = std::move(listener), + }); +} + +void OfflineSimulationUser::StopAdvertising() { + ctrl_.StopAdvertising(&client_); +} + +Status OfflineSimulationUser::StartDiscovery(const std::string& service_id, + CountDownLatch* latch) { + found_latch_ = latch; + DiscoveryListener listener = { + .endpoint_found_cb = + absl::bind_front(&OfflineSimulationUser::OnEndpointFound, this), + .endpoint_lost_cb = + absl::bind_front(&OfflineSimulationUser::OnEndpointLost, this), + }; + return ctrl_.StartDiscovery(&client_, service_id, options_, + std::move(listener)); +} + +void OfflineSimulationUser::StopDiscovery() { ctrl_.StopDiscovery(&client_); } + +Status OfflineSimulationUser::RequestConnection(CountDownLatch* latch) { + initiated_latch_ = latch; + ConnectionListener listener = { + .initiated_cb = + std::bind(&OfflineSimulationUser::OnConnectionInitiated, this, + std::placeholders::_1, std::placeholders::_2, true), + .accepted_cb = + absl::bind_front(&OfflineSimulationUser::OnConnectionAccepted, this), + .rejected_cb = + absl::bind_front(&OfflineSimulationUser::OnConnectionRejected, this), + .disconnected_cb = + absl::bind_front(&OfflineSimulationUser::OnEndpointDisconnect, this), + }; + return ctrl_.RequestConnection(&client_, discovered_.endpoint_id, + { + .name = discovered_.endpoint_name, + .listener = std::move(listener), + }); +} + +Status OfflineSimulationUser::AcceptConnection(CountDownLatch* latch) { + accept_latch_ = latch; + PayloadListener listener = { + .payload_cb = absl::bind_front(&OfflineSimulationUser::OnPayload, this), + .payload_progress_cb = + absl::bind_front(&OfflineSimulationUser::OnPayloadProgress, this), + }; + return ctrl_.AcceptConnection(&client_, discovered_.endpoint_id, + std::move(listener)); +} + +Status OfflineSimulationUser::RejectConnection(CountDownLatch* latch) { + reject_latch_ = latch; + return ctrl_.RejectConnection(&client_, discovered_.endpoint_id); +} + +void OfflineSimulationUser::Disconnect() { + NEARBY_LOGS(INFO) << "Disconnecting from id=" << discovered_.endpoint_id; + ctrl_.DisconnectFromEndpoint(&client_, discovered_.endpoint_id); +} + +} // namespace connections +} // namespace nearby +} // namespace location diff --git a/cpp/core_v2/internal/offline_simulation_user.h b/cpp/core_v2/internal/offline_simulation_user.h new file mode 100644 index 00000000..27a41d56 --- /dev/null +++ b/cpp/core_v2/internal/offline_simulation_user.h @@ -0,0 +1,175 @@ +// Copyright 2020 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 CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_ +#define CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_ + +#include + +#include "core_v2/internal/client_proxy.h" +#include "core_v2/internal/offline_service_controller.h" +#include "platform_v2/public/atomic_boolean.h" +#include "platform_v2/public/condition_variable.h" +#include "platform_v2/public/count_down_latch.h" +#include "platform_v2/public/future.h" +#include "gtest/gtest.h" +#include "absl/strings/string_view.h" + +// Test-only class to help run end-to-end simulations for nearby connections +// protocol. +// +// This is a "standalone" version of PcpManager. It can run independently, +// provided MediumEnvironment has adequate support for all medium types in use. +namespace location { +namespace nearby { +namespace connections { + +class OfflineSimulationUser { + public: + struct DiscoveredInfo { + std::string endpoint_id; + std::string endpoint_name; + std::string service_id; + + bool Empty() const { return endpoint_id.empty(); } + void Clear() { endpoint_id.clear(); } + }; + + explicit OfflineSimulationUser(absl::string_view device_name) + : name_(device_name) {} + virtual ~OfflineSimulationUser() = default; + + // Calls PcpManager::StartAdvertising(). + // If latch is provided, will call latch->CountDown() in the initiated_cb + // callback. + Status StartAdvertising(const std::string& service_id, CountDownLatch* latch); + + // Calls PcpManager::StopAdvertising(). + void StopAdvertising(); + + // Calls PcpManager::StartDiscovery(). + // If latch is provided, will call latch->CountDown() in the endpoint_found_cb + // callback. + Status StartDiscovery(const std::string& service_id, CountDownLatch* latch); + + // Calls PcpManager::StopDiscovery(). + void StopDiscovery(); + + // Calls PcpManager::RequestConnection(). + // If latch is provided, latch->CountDown() will be called in the initiated_cb + // callback. + Status RequestConnection(CountDownLatch* latch); + + // Calls PcpManager::AcceptConnection. + // If latch is provided, latch->CountDown() will be called in the accepted_cb + // callback. + Status AcceptConnection(CountDownLatch* latch); + + // Calls PcpManager::RejectConnection. + // If latch is provided, latch->CountDown() will be called in the rejected_cb + // callback. + Status RejectConnection(CountDownLatch* latch); + + // Unlike acceptance, rejection does not have to be mutual, in order to work. + // This method will allow to synchronize on the remote rejection, without + // performing a local rejection. + // latch.CountDown() will be called in the rejected_cb callback. + void ExpectRejectedConnection(CountDownLatch& latch) { + reject_latch_ = &latch; + } + + void ExpectPayload(CountDownLatch& latch) { payload_latch_ = &latch; } + void ExpectDisconnect(CountDownLatch& latch) { disconnect_latch_ = &latch; } + + const DiscoveredInfo& GetDiscovered() const { return discovered_; } + std::string GetName() const { return name_; } + + bool WaitForProgress(std::function pred, + absl::Duration timeout); + + Payload& GetPayload() { return payload_; } + void SendPayload(Payload payload) { + sender_payload_id_ = payload.GetId(); + ctrl_.SendPayload(&client_, {discovered_.endpoint_id}, std::move(payload)); + } + + Status CancelPayload() { + if (sender_payload_id_) { + return ctrl_.CancelPayload(&client_, sender_payload_id_); + } else { + return ctrl_.CancelPayload(&client_, payload_.GetId()); + } + } + + void Disconnect(); + + bool IsAdvertising() const { return client_.IsAdvertising(); } + + bool IsDiscovering() const { return client_.IsDiscovering(); } + + bool IsConnected() const { + return client_.IsConnectedToEndpoint(discovered_.endpoint_id); + } + + void Stop() { + ctrl_.Stop(); + } + + protected: + // ConnectionListener callbacks + void OnConnectionInitiated(const std::string& endpoint_id, + const ConnectionResponseInfo& info, + bool is_outgoing); + void OnConnectionAccepted(const std::string& endpoint_id); + void OnConnectionRejected(const std::string& endpoint_id, Status status); + void OnEndpointDisconnect(const std::string& endpoint_id); + + // DiscoveryListener callbacks + void OnEndpointFound(const std::string& endpoint_id, + const std::string& endpoint_name, + const std::string& service_id); + void OnEndpointLost(const std::string& endpoint_id); + + // PayloadListener callbacks + void OnPayload(const std::string& endpoint_id, Payload payload); + void OnPayloadProgress(const std::string& endpoint_id, + const PayloadProgressInfo& info); + + std::string service_id_; + DiscoveredInfo discovered_; + Mutex progress_mutex_; + ConditionVariable progress_sync_{&progress_mutex_}; + PayloadProgressInfo progress_info_; + Payload payload_; + Payload::Id sender_payload_id_ = 0; + CountDownLatch* initiated_latch_ = nullptr; + CountDownLatch* accept_latch_ = nullptr; + CountDownLatch* reject_latch_ = nullptr; + CountDownLatch* found_latch_ = nullptr; + CountDownLatch* lost_latch_ = nullptr; + CountDownLatch* payload_latch_ = nullptr; + CountDownLatch* disconnect_latch_ = nullptr; + Future* future_ = nullptr; + std::function predicate_; + std::string name_; + ConnectionOptions options_{.strategy = Strategy::kP2pCluster}; + ClientProxy client_; + OfflineServiceController ctrl_; +}; + +} // namespace connections +} // namespace nearby +} // namespace location + +#endif // CORE_V2_INTERNAL_OFFLINE_SIMULATION_USER_H_ diff --git a/cpp/core_v2/internal/payload_manager.cc b/cpp/core_v2/internal/payload_manager.cc index 947bbe37..ceab43e5 100644 --- a/cpp/core_v2/internal/payload_manager.cc +++ b/cpp/core_v2/internal/payload_manager.cc @@ -58,7 +58,7 @@ bool PayloadManager::SendPayloadLoop( for (const auto& endpoint : unavailable_endpoints) { HandleFinishedOutgoingPayload( client, {endpoint->id}, payload_header, next_chunk_offset, - EndpointInfoStatusToPayloadStatus(endpoint->status)); + EndpointInfoStatusToPayloadStatus(endpoint->status.Get())); } // Update the still-active recipients of this payload. @@ -156,8 +156,8 @@ PayloadManager::GetAvailableAndUnavailableEndpoints( Endpoints unavailable; for (auto* endpoint_info : pending_payload.GetEndpoints()) { NEARBY_LOG(INFO, "EndpointInfo: %p; id=%s; status=%d", endpoint_info, - endpoint_info->id.c_str(), endpoint_info->status); - if (endpoint_info->status == + endpoint_info->id.c_str(), endpoint_info->status.Get()); + if (endpoint_info->status.Get() == PayloadManager::EndpointInfo::Status::kAvailable) { available.push_back(endpoint_info); } else { @@ -253,12 +253,16 @@ void PayloadManager::CancelAllPayloads() { } } -PayloadManager::~PayloadManager() { - NEARBY_LOG(INFO, "PayloadManager: going down; self=%p", this); - shutdown_.Set(true); +void PayloadManager::DisconnectFromEndpointManager() { + if (shutdown_.Set(true)) return; // Unregister ourselves from the FrameProcessors. endpoint_manager_->UnregisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, handle_, true); +} + +PayloadManager::~PayloadManager() { + NEARBY_LOG(INFO, "PayloadManager: going down; self=%p", this); + DisconnectFromEndpointManager(); CancelAllPayloads(); NEARBY_LOG(INFO, "PayloadManager: turn down payload executors; self=%p", this); @@ -399,7 +403,11 @@ void PayloadManager::OnIncomingFrame( void PayloadManager::OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, CountDownLatch* barrier) { - RunOnStatusUpdateThread([this, client, endpoint_id, &barrier]() { + if (shutdown_.Get()) { + if (barrier) barrier->CountDown(); + return; + } + RunOnStatusUpdateThread([this, client, endpoint_id, barrier]() { // Iterate through all our payloads and look for payloads associated // with this endpoint. MutexLock lock(&mutex_); @@ -921,7 +929,7 @@ PayloadManager::EndpointInfo::ControlMessageEventToEndpointInfoStatus( void PayloadManager::EndpointInfo::SetStatusFromControlMessage( const PayloadTransferFrame::ControlMessage& control_message) { - status = ControlMessageEventToEndpointInfoStatus(control_message.event()); + status.Set(ControlMessageEventToEndpointInfoStatus(control_message.event())); } //////////////////////////////// PendingPayload //////////////////////////////// @@ -938,7 +946,7 @@ PayloadManager::PendingPayload::PendingPayload( for (const auto& id : endpoint_ids) { endpoints_.emplace(id, EndpointInfo{ .id = id, - .status = EndpointInfo::Status::kAvailable, + .status {EndpointInfo::Status::kAvailable}, }); } } diff --git a/cpp/core_v2/internal/payload_manager.h b/cpp/core_v2/internal/payload_manager.h index c7475dfb..0489dbff 100644 --- a/cpp/core_v2/internal/payload_manager.h +++ b/cpp/core_v2/internal/payload_manager.h @@ -29,6 +29,7 @@ #include "proto/connections/offline_wire_formats.pb.h" #include "platform_v2/base/byte_array.h" #include "platform_v2/public/atomic_boolean.h" +#include "platform_v2/public/atomic_reference.h" #include "platform_v2/public/count_down_latch.h" #include "platform_v2/public/mutex.h" #include "proto/connections_enums.pb.h" @@ -61,6 +62,8 @@ class PayloadManager : public EndpointManager::FrameProcessor { void OnEndpointDisconnect(ClientProxy* client, const std::string& endpoint_id, CountDownLatch* barrier) override; + void DisconnectFromEndpointManager(); + private: // Information about an endpoint for a particular payload. struct EndpointInfo { @@ -79,7 +82,7 @@ class PayloadManager : public EndpointManager::FrameProcessor { PayloadTransferFrame::ControlMessage::EventType event); std::string id; - Status status = Status::kUnknown; + AtomicReference status {Status::kUnknown}; std::int64_t offset = 0; }; diff --git a/cpp/core_v2/internal/payload_manager_test.cc b/cpp/core_v2/internal/payload_manager_test.cc index 2b64fe16..75239255 100644 --- a/cpp/core_v2/internal/payload_manager_test.cc +++ b/cpp/core_v2/internal/payload_manager_test.cc @@ -39,6 +39,7 @@ class PayloadSimulationUser : public SimulationUser { explicit PayloadSimulationUser(absl::string_view name) : SimulationUser(std::string(name)) {} ~PayloadSimulationUser() override { + NEARBY_LOGS(INFO) << "PayloadSimulationUser: [down] name=" << name_; // SystemClock::Sleep(kDefaultTimeout); } @@ -123,6 +124,8 @@ TEST_F(PayloadManagerTest, CanSendBytePayload) { EXPECT_EQ(user_a.GetPayload().AsBytes(), ByteArray(std::string(kMessage))); NEARBY_LOG(INFO, "Test completed."); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } @@ -171,6 +174,8 @@ TEST_F(PayloadManagerTest, CanSendStreamPayload) { rx.Close(); tx.Close(); NEARBY_LOG(INFO, "Test completed."); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } @@ -227,6 +232,8 @@ TEST_F(PayloadManagerTest, CanCancelPayloadOnReceiverSide) { rx.Close(); NEARBY_LOG(INFO, "Test completed."); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } @@ -283,6 +290,8 @@ TEST_F(PayloadManagerTest, CanCancelPayloadOnSenderSide) { rx.Close(); NEARBY_LOG(INFO, "Test completed."); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } diff --git a/cpp/core_v2/internal/pcp_manager.cc b/cpp/core_v2/internal/pcp_manager.cc index 296e2333..5961ce0f 100644 --- a/cpp/core_v2/internal/pcp_manager.cc +++ b/cpp/core_v2/internal/pcp_manager.cc @@ -35,6 +35,18 @@ PcpManager::PcpManager(Mediums& mediums, channel_manager); } +void PcpManager::DisconnectFromEndpointManager() { + if (shutdown_.Set(true)) return; + for (auto& item : handlers_) { + if (!item.second) continue; + item.second->DisconnectFromEndpointManager(); + } +} + +PcpManager::~PcpManager() { + DisconnectFromEndpointManager(); +} + Status PcpManager::StartAdvertising(ClientProxy* client, const string& service_id, const ConnectionOptions& options, diff --git a/cpp/core_v2/internal/pcp_manager.h b/cpp/core_v2/internal/pcp_manager.h index 5354335f..67cbeb5e 100644 --- a/cpp/core_v2/internal/pcp_manager.h +++ b/cpp/core_v2/internal/pcp_manager.h @@ -17,15 +17,16 @@ #include +#include "core_v2/internal/base_pcp_handler.h" #include "core_v2/internal/client_proxy.h" #include "core_v2/internal/endpoint_channel_manager.h" #include "core_v2/internal/endpoint_manager.h" #include "core_v2/internal/mediums/mediums.h" -#include "core_v2/internal/pcp_handler.h" #include "core_v2/listeners.h" #include "core_v2/options.h" #include "core_v2/status.h" #include "core_v2/strategy.h" +#include "platform_v2/public/atomic_boolean.h" #include "absl/container/flat_hash_map.h" namespace location { @@ -43,7 +44,7 @@ class PcpManager { public: PcpManager(Mediums& mediums, EndpointChannelManager& channel_manager, EndpointManager& endpoint_manager); - ~PcpManager() = default; + ~PcpManager(); Status StartAdvertising(ClientProxy* client_proxy, const string& service_id, const ConnectionOptions& options, @@ -62,13 +63,15 @@ class PcpManager { Status RejectConnection(ClientProxy* client_proxy, const string& endpoint_id); proto::connections::Medium GetBandwidthUpgradeMedium(); + void DisconnectFromEndpointManager(); private: bool SetCurrentPcpHandler(Strategy strategy); PcpHandler* GetPcpHandler(Pcp pcp) const; - absl::flat_hash_map> handlers_; - PcpHandler* current_; + AtomicBoolean shutdown_{false}; + absl::flat_hash_map> handlers_; + PcpHandler* current_ = nullptr; }; } // namespace connections diff --git a/cpp/core_v2/internal/pcp_manager_test.cc b/cpp/core_v2/internal/pcp_manager_test.cc index ef548cae..c42b7cc6 100644 --- a/cpp/core_v2/internal/pcp_manager_test.cc +++ b/cpp/core_v2/internal/pcp_manager_test.cc @@ -87,6 +87,8 @@ TEST_F(PcpManagerTest, CanConnect) { EXPECT_EQ(user_b.GetDiscovered().endpoint_name, user_a.GetName()); user_b.RequestConnection(&connection_latch); EXPECT_TRUE(connection_latch.Await(absl::Milliseconds(1000)).result()); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } @@ -107,6 +109,8 @@ TEST_F(PcpManagerTest, CanAccept) { user_a.AcceptConnection(&accept_latch); user_b.AcceptConnection(&accept_latch); EXPECT_TRUE(accept_latch.Await(absl::Milliseconds(1000)).result()); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } @@ -127,6 +131,8 @@ TEST_F(PcpManagerTest, CanReject) { user_b.ExpectRejectedConnection(reject_latch); user_a.RejectConnection(nullptr); EXPECT_TRUE(reject_latch.Await(absl::Milliseconds(1000)).result()); + user_a.Stop(); + user_b.Stop(); env_.Stop(); } diff --git a/cpp/core_v2/internal/service_controller.h b/cpp/core_v2/internal/service_controller.h index 137ca28d..0ac10be7 100644 --- a/cpp/core_v2/internal/service_controller.h +++ b/cpp/core_v2/internal/service_controller.h @@ -48,38 +48,38 @@ class ServiceController { ServiceController& operator=(const ServiceController&) = delete; // Starts advertising an endpoint for a local app. - virtual Status StartAdvertising(ClientProxy* client_proxy, + virtual Status StartAdvertising(ClientProxy* client, const std::string& service_id, const ConnectionOptions& options, const ConnectionRequestInfo& info) = 0; - virtual void StopAdvertising(ClientProxy* client_proxy) = 0; + virtual void StopAdvertising(ClientProxy* client) = 0; - virtual Status StartDiscovery(ClientProxy* client_proxy, + virtual Status StartDiscovery(ClientProxy* client, const std::string& service_id, const ConnectionOptions& options, const DiscoveryListener& listener) = 0; - virtual void StopDiscovery(ClientProxy* client_proxy) = 0; + virtual void StopDiscovery(ClientProxy* client) = 0; - virtual Status RequestConnection(ClientProxy* client_proxy, + virtual Status RequestConnection(ClientProxy* client, const std::string& endpoint_id, const ConnectionRequestInfo& info) = 0; - virtual Status AcceptConnection(ClientProxy* client_proxy, + virtual Status AcceptConnection(ClientProxy* client, const std::string& endpoint_id, const PayloadListener& listener) = 0; - virtual Status RejectConnection(ClientProxy* client_proxy, + virtual Status RejectConnection(ClientProxy* client, const std::string& endpoint_id) = 0; - virtual void InitiateBandwidthUpgrade(ClientProxy* client_proxy, + virtual void InitiateBandwidthUpgrade(ClientProxy* client, const std::string& endpoint_id) = 0; - virtual void SendPayload(ClientProxy* client_proxy, + virtual void SendPayload(ClientProxy* client, const std::vector& endpoint_ids, Payload payload) = 0; - virtual Status CancelPayload(ClientProxy* client_proxy, - std::int64_t payload_id) = 0; + virtual Status CancelPayload(ClientProxy* client, + Payload::Id payload_id) = 0; - virtual void DisconnectFromEndpoint(ClientProxy* client_proxy, + virtual void DisconnectFromEndpoint(ClientProxy* client, const std::string& endpoint_id) = 0; }; diff --git a/cpp/core_v2/internal/simulation_user.h b/cpp/core_v2/internal/simulation_user.h index b039e5fb..0f8ba8f7 100644 --- a/cpp/core_v2/internal/simulation_user.h +++ b/cpp/core_v2/internal/simulation_user.h @@ -50,7 +50,13 @@ class SimulationUser { explicit SimulationUser(const std::string& device_name) : name_(device_name) {} - virtual ~SimulationUser() = default; + virtual ~SimulationUser() { + Stop(); + } + void Stop() { + pm_.DisconnectFromEndpointManager(); + mgr_.DisconnectFromEndpointManager(); + } // Calls PcpManager::StartAdvertising. // If latch is provided, will call latch->CountDown() in the initiated_cb diff --git a/cpp/platform/impl/shared/BUILD b/cpp/platform/impl/shared/BUILD index 2cda10c8..79343f7f 100644 --- a/cpp/platform/impl/shared/BUILD +++ b/cpp/platform/impl/shared/BUILD @@ -83,5 +83,6 @@ cc_test( ":file", "//file/util:temp_path", "//testing/base/public:gunit_main", + "//absl/strings", ], ) diff --git a/cpp/platform/impl/shared/file_impl_test.cc b/cpp/platform/impl/shared/file_impl_test.cc index 6f85620b..40fe4a80 100644 --- a/cpp/platform/impl/shared/file_impl_test.cc +++ b/cpp/platform/impl/shared/file_impl_test.cc @@ -22,6 +22,7 @@ #include #include "gtest/gtest.h" +#include "absl/strings/string_view.h" namespace location { namespace nearby { @@ -35,7 +36,7 @@ class FileImplTest : public ::testing::Test { file_ = std::fstream(path_, std::fstream::in | std::fstream::out); } - void WriteToFile(const std::string& text) { + void WriteToFile(absl::string_view text) { file_ << text; file_.flush(); size_ += text.size(); diff --git a/cpp/platform_v2/api/BUILD b/cpp/platform_v2/api/BUILD index b3c1a458..7cbc51ad 100644 --- a/cpp/platform_v2/api/BUILD +++ b/cpp/platform_v2/api/BUILD @@ -86,6 +86,5 @@ cc_library( ":types", "//platform_v2/base", "//absl/strings", - "//absl/types:any", ], ) diff --git a/cpp/platform_v2/api/condition_variable.h b/cpp/platform_v2/api/condition_variable.h index 45b39ba4..04328767 100644 --- a/cpp/platform_v2/api/condition_variable.h +++ b/cpp/platform_v2/api/condition_variable.h @@ -41,7 +41,6 @@ class ConditionVariable { // Waits while timeout has not expired for Notify to be called. // May return prematurely in case of interrupt, if supported by platform. // Returns kSuccess, or kInterrupted on interrupt. - // If Timeout expired, and Notify was not called, returns kTimeout. virtual Exception Wait(absl::Duration timeout) = 0; }; diff --git a/cpp/platform_v2/api/executor.h b/cpp/platform_v2/api/executor.h index 64b4e016..0adf0676 100644 --- a/cpp/platform_v2/api/executor.h +++ b/cpp/platform_v2/api/executor.h @@ -21,6 +21,8 @@ namespace location { namespace nearby { namespace api { +int GetCurrentTid(); + // This abstract class is the superclass of all classes representing an // Executor. class Executor { @@ -33,6 +35,8 @@ class Executor { // https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html#shutdown-- virtual void Shutdown() = 0; + + virtual int GetTid(int index) const = 0; }; } // namespace api diff --git a/cpp/platform_v2/base/logging.h b/cpp/platform_v2/base/logging.h index c6f13001..53613f60 100644 --- a/cpp/platform_v2/base/logging.h +++ b/cpp/platform_v2/base/logging.h @@ -43,7 +43,12 @@ class LogMessageVoidify { location::nearby::api::LogMessage::Severity::kError #define NEARBY_SEVERITY_FATAL \ location::nearby::api::LogMessage::Severity::kFatal - +#if defined(_WIN32) +// wingdi.h defines ERROR to be 0. When we call LOG(ERROR), it gets substituted +// with 0, and it expands to NEARBY_SEVERITY_0. To allow us to keep using this +// syntax, we define this macro to do the same thing as NEARBY_SEVERITY_ERROR. +#define NEARBY_SEVERITY_0 location::nearby::api::LogMessage::Severity::kError +#endif // defined(_WIN32) #define NEARBY_SEVERITY(severity) NEARBY_SEVERITY_##severity // Log enabling diff --git a/cpp/platform_v2/impl/g3/BUILD b/cpp/platform_v2/impl/g3/BUILD index 2576157e..49342b51 100644 --- a/cpp/platform_v2/impl/g3/BUILD +++ b/cpp/platform_v2/impl/g3/BUILD @@ -45,7 +45,6 @@ cc_library( "//absl/base:core_headers", "//absl/synchronization", "//absl/time", - "//absl/types:any", "//thread", ], ) diff --git a/cpp/platform_v2/impl/g3/condition_variable.h b/cpp/platform_v2/impl/g3/condition_variable.h index f5e16a69..4c26392c 100644 --- a/cpp/platform_v2/impl/g3/condition_variable.h +++ b/cpp/platform_v2/impl/g3/condition_variable.h @@ -34,9 +34,8 @@ class ConditionVariable : public api::ConditionVariable { return {Exception::kSuccess}; } Exception Wait(absl::Duration timeout) override { - return cond_var_.WaitWithTimeout(mutex_, timeout) - ? Exception{Exception::kTimeout} - : Exception{Exception::kSuccess}; + cond_var_.WaitWithTimeout(mutex_, timeout); + return {Exception::kSuccess}; } void Notify() override { cond_var_.SignalAll(); } diff --git a/cpp/platform_v2/impl/g3/multi_thread_executor.h b/cpp/platform_v2/impl/g3/multi_thread_executor.h index c5cf331e..cc27b55e 100644 --- a/cpp/platform_v2/impl/g3/multi_thread_executor.h +++ b/cpp/platform_v2/impl/g3/multi_thread_executor.h @@ -47,6 +47,11 @@ class MultiThreadExecutor : public api::SubmittableExecutor { void Shutdown() override { DoShutdown(); } ~MultiThreadExecutor() override { DoShutdown(); } + int GetTid(int index) const override { + const auto* thread = thread_pool_.thread(index); + return thread ? thread->tid() : 0; + } + void ScheduleAfter(absl::Duration delay, Runnable&& runnable) { if (shutdown_) return; thread_pool_.ScheduleAt(absl::Now() + delay, std::move(runnable)); diff --git a/cpp/platform_v2/impl/g3/platform.cc b/cpp/platform_v2/impl/g3/platform.cc index d706a3bc..afa3f589 100644 --- a/cpp/platform_v2/impl/g3/platform.cc +++ b/cpp/platform_v2/impl/g3/platform.cc @@ -61,6 +61,11 @@ std::string GetPayloadPath(PayloadId payload_id) { } } // namespace +int GetCurrentTid() { + const LiveThread* my = Thread_GetMyLiveThread(); + return LiveThread_Pthread_TID(my); +} + std::unique_ptr ImplementationPlatform::CreateSingleThreadExecutor() { return absl::make_unique(); diff --git a/cpp/platform_v2/impl/g3/scheduled_executor.h b/cpp/platform_v2/impl/g3/scheduled_executor.h index 09553eaa..5a08dee5 100644 --- a/cpp/platform_v2/impl/g3/scheduled_executor.h +++ b/cpp/platform_v2/impl/g3/scheduled_executor.h @@ -45,6 +45,9 @@ class ScheduledExecutor final : public api::ScheduledExecutor { absl::Duration delay) override; void Shutdown() override { executor_.Shutdown(); } + int GetTid(int index) const override { + return executor_.GetTid(index); + } private: SingleThreadExecutor executor_; }; diff --git a/cpp/platform_v2/impl/shared/BUILD b/cpp/platform_v2/impl/shared/BUILD index eea5da52..0b657a6a 100644 --- a/cpp/platform_v2/impl/shared/BUILD +++ b/cpp/platform_v2/impl/shared/BUILD @@ -65,5 +65,6 @@ cc_test( "//file/util:temp_path", "//platform_v2/base", "//testing/base/public:gunit_main", + "//absl/strings", ], ) diff --git a/cpp/platform_v2/impl/shared/file_test.cc b/cpp/platform_v2/impl/shared/file_test.cc index a33ed5a3..50f05779 100644 --- a/cpp/platform_v2/impl/shared/file_test.cc +++ b/cpp/platform_v2/impl/shared/file_test.cc @@ -22,6 +22,7 @@ #include "file/util/temp_path.h" #include "platform_v2/base/byte_array.h" #include "gtest/gtest.h" +#include "absl/strings/string_view.h" namespace location { namespace nearby { @@ -36,7 +37,7 @@ class FileTest : public ::testing::Test { file_ = std::fstream(path_, std::fstream::in | std::fstream::out); } - void WriteToFile(const std::string& text) { + void WriteToFile(absl::string_view text) { file_ << text; file_.flush(); size_ += text.size(); diff --git a/cpp/platform_v2/public/BUILD b/cpp/platform_v2/public/BUILD index 6a119f23..254d49f6 100644 --- a/cpp/platform_v2/public/BUILD +++ b/cpp/platform_v2/public/BUILD @@ -53,7 +53,6 @@ cc_library( "//absl/base:core_headers", "//absl/container:flat_hash_map", "//absl/time", - "//absl/types:any", ], ) diff --git a/cpp/platform_v2/public/condition_variable_test.cc b/cpp/platform_v2/public/condition_variable_test.cc index d2b3805c..be798df8 100644 --- a/cpp/platform_v2/public/condition_variable_test.cc +++ b/cpp/platform_v2/public/condition_variable_test.cc @@ -17,6 +17,7 @@ #include "platform_v2/public/logging.h" #include "platform_v2/public/mutex.h" #include "platform_v2/public/single_thread_executor.h" +#include "platform_v2/public/system_clock.h" #include "gmock/gmock.h" #include "gtest/gtest.h" #include "absl/time/time.h" @@ -68,7 +69,12 @@ TEST(ConditionVariableTest, WaitTerminatesOnTimeoutWithoutNotify) { Mutex mutex; ConditionVariable cond{&mutex}; MutexLock lock(&mutex); - EXPECT_EQ(cond.Wait(absl::Milliseconds(100)), Exception{Exception::kTimeout}); + + const absl::Duration kWaitTime = absl::Milliseconds(100); + absl::Time start = SystemClock::ElapsedRealtime(); + cond.Wait(kWaitTime); + absl::Duration duration = SystemClock::ElapsedRealtime() - start; + EXPECT_GE(duration, kWaitTime); } } // namespace diff --git a/cpp/platform_v2/public/scheduled_executor.h b/cpp/platform_v2/public/scheduled_executor.h index 757cf232..c33528a4 100644 --- a/cpp/platform_v2/public/scheduled_executor.h +++ b/cpp/platform_v2/public/scheduled_executor.h @@ -64,6 +64,12 @@ class ScheduledExecutor final { DoShutdown(); } + int GetTid(int index) const { + MutexLock lock(&mutex_); + return impl_->GetTid(index); + } + int Tid() const { return GetTid(0); } + Cancelable Schedule(Runnable&& runnable, absl::Duration duration) ABSL_LOCKS_EXCLUDED(mutex_) { MutexLock lock(&mutex_); @@ -79,7 +85,7 @@ class ScheduledExecutor final { } } - Mutex mutex_; + mutable Mutex mutex_; std::unique_ptr ABSL_GUARDED_BY(mutex_) impl_; }; diff --git a/cpp/platform_v2/public/settable_future.h b/cpp/platform_v2/public/settable_future.h index a62cf263..20d96d4a 100644 --- a/cpp/platform_v2/public/settable_future.h +++ b/cpp/platform_v2/public/settable_future.h @@ -71,8 +71,8 @@ class SettableFuture : public api::SettableFuture { MutexLock lock(&mutex_); while (!done_) { absl::Time start_time = SystemClock::ElapsedRealtime(); - if (completed_.Wait(timeout).Raised(Exception::kTimeout)) { - SetExceptionLocked({Exception::kTimeout}); + if (completed_.Wait(timeout).Raised(Exception::kInterrupted)) { + SetExceptionLocked({Exception::kInterrupted}); break; } absl::Duration spent = SystemClock::ElapsedRealtime() - start_time; diff --git a/cpp/platform_v2/public/single_thread_executor.h b/cpp/platform_v2/public/single_thread_executor.h index 805d6efb..de8a531d 100644 --- a/cpp/platform_v2/public/single_thread_executor.h +++ b/cpp/platform_v2/public/single_thread_executor.h @@ -32,6 +32,7 @@ class SingleThreadExecutor final : public SubmittableExecutor { ~SingleThreadExecutor() override = default; SingleThreadExecutor(SingleThreadExecutor&&) = default; SingleThreadExecutor& operator=(SingleThreadExecutor&&) = default; + int Tid() const { return GetTid(0); } }; } // namespace nearby diff --git a/cpp/platform_v2/public/submittable_executor.h b/cpp/platform_v2/public/submittable_executor.h index 4751ef86..3fc703d0 100644 --- a/cpp/platform_v2/public/submittable_executor.h +++ b/cpp/platform_v2/public/submittable_executor.h @@ -31,6 +31,8 @@ namespace location { namespace nearby { +inline int GetCurrentTid() { return api::GetCurrentTid(); } + // Main interface to be used by platform as a base class for // - MultiThreadExecutor // - SingleThreadExecutor @@ -55,6 +57,11 @@ class SubmittableExecutor : public api::SubmittableExecutor { if (impl_) impl_->Execute(std::move(runnable)); } + int GetTid(int index) const ABSL_LOCKS_EXCLUDED(mutex_) override { + MutexLock lock(&mutex_); + return impl_ ? impl_->GetTid(index) : 0; + } + void Shutdown() ABSL_LOCKS_EXCLUDED(mutex_) override { MutexLock lock(&mutex_); DoShutdown(); @@ -100,7 +107,7 @@ class SubmittableExecutor : public api::SubmittableExecutor { ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) override { return impl_ ? impl_->DoSubmit(std::move(wrapped_callable)) : false; } - Mutex mutex_; + mutable Mutex mutex_; std::unique_ptr ABSL_GUARDED_BY(mutex_) impl_; };