diff --git a/cpp/core/internal/base_endpoint_channel.cc b/cpp/core/internal/base_endpoint_channel.cc index 23e7e476..bdc0d743 100644 --- a/cpp/core/internal/base_endpoint_channel.cc +++ b/cpp/core/internal/base_endpoint_channel.cc @@ -219,6 +219,10 @@ Exception BaseEndpointChannel::Write(const ByteArray& data) { } } + { + MutexLock lock(&last_write_mutex_); + last_write_timestamp_ = SystemClock::ElapsedRealtime(); + } return {Exception::kSuccess}; } @@ -326,6 +330,11 @@ absl::Time BaseEndpointChannel::GetLastReadTimestamp() const { return last_read_timestamp_; } +absl::Time BaseEndpointChannel::GetLastWriteTimestamp() const { + MutexLock lock(&last_write_mutex_); + return last_write_timestamp_; +} + bool BaseEndpointChannel::IsEncryptionEnabledLocked() const { return crypto_context_ != nullptr; } diff --git a/cpp/core/internal/base_endpoint_channel.h b/cpp/core/internal/base_endpoint_channel.h index aaab6760..c8dc81d7 100644 --- a/cpp/core/internal/base_endpoint_channel.h +++ b/cpp/core/internal/base_endpoint_channel.h @@ -89,6 +89,11 @@ class BaseEndpointChannel : public EndpointChannel { absl::Time GetLastReadTimestamp() const ABSL_LOCKS_EXCLUDED(last_read_mutex_) override; + // Returns the timestamp (returned by ElapsedRealtime) of the last write to + // this endpoint, or -1 if no writes have occurred. + absl::Time GetLastWriteTimestamp() const + ABSL_LOCKS_EXCLUDED(last_write_mutex_) override; + void SetAnalyticsRecorder(analytics::AnalyticsRecorder* analytics_recorder, const std::string& endpoint_id) override; @@ -108,11 +113,18 @@ class BaseEndpointChannel : public EndpointChannel { void BlockUntilUnpaused() ABSL_EXCLUSIVE_LOCKS_REQUIRED(is_paused_mutex_); void CloseIo() ABSL_NO_THREAD_SAFETY_ANALYSIS; - // We need a separate mutex to pritect read timestamp, because if a read + // We need a separate mutex to protect read timestamp, because if a read // blocks on IO, we don't want timestamp read access to block too. mutable Mutex last_read_mutex_; absl::Time last_read_timestamp_ ABSL_GUARDED_BY(last_read_mutex_) = absl::InfinitePast(); + + // We need a separate mutex to protect write timestamp, because if a write + // blocks on IO, we don't want timestamp write access to block too. + mutable Mutex last_write_mutex_; + absl::Time last_write_timestamp_ ABSL_GUARDED_BY(last_write_mutex_) = + absl::InfinitePast(); + const std::string channel_name_; // The reader and writer are synchronized independently since we can't have diff --git a/cpp/core/internal/encryption_runner_test.cc b/cpp/core/internal/encryption_runner_test.cc index 7945c0a5..06720e33 100644 --- a/cpp/core/internal/encryption_runner_test.cc +++ b/cpp/core/internal/encryption_runner_test.cc @@ -42,6 +42,7 @@ class FakeEndpointChannel : public EndpointChannel { : ExceptionOr{Exception::kIo}; } Exception Write(const ByteArray& data) override { + write_timestamp_ = SystemClock::ElapsedRealtime(); return out_ ? out_->Write(data) : Exception{Exception::kIo}; } void Close() override { @@ -61,6 +62,7 @@ class FakeEndpointChannel : public EndpointChannel { void Pause() override {} void Resume() override {} absl::Time GetLastReadTimestamp() const override { return read_timestamp_; } + absl::Time GetLastWriteTimestamp() const override { return write_timestamp_; } void SetAnalyticsRecorder(analytics::AnalyticsRecorder* analytics_recorder, const std::string& endpoint_id) override {} @@ -68,6 +70,7 @@ class FakeEndpointChannel : public EndpointChannel { InputStream* in_ = nullptr; OutputStream* out_ = nullptr; absl::Time read_timestamp_ = absl::InfinitePast(); + absl::Time write_timestamp_ = absl::InfinitePast(); }; struct User { diff --git a/cpp/core/internal/endpoint_channel.h b/cpp/core/internal/endpoint_channel.h index 7ff8602a..43b219e3 100644 --- a/cpp/core/internal/endpoint_channel.h +++ b/cpp/core/internal/endpoint_channel.h @@ -81,6 +81,10 @@ class EndpointChannel { // reads have occurred. virtual absl::Time GetLastReadTimestamp() const = 0; + // Returns the timestamp of the last write to this endpoint, or -1 if no + // writes have occurred. + virtual absl::Time GetLastWriteTimestamp() const = 0; + // Sets the AnalyticsRecorder instance for analytics. virtual void SetAnalyticsRecorder( analytics::AnalyticsRecorder* analytics_recorder, diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc index 987c79ff..e050b9f4 100644 --- a/cpp/core/internal/endpoint_manager.cc +++ b/cpp/core/internal/endpoint_manager.cc @@ -214,32 +214,45 @@ ExceptionOr EndpointManager::HandleData( ExceptionOr EndpointManager::HandleKeepAlive( EndpointChannel* endpoint_channel, absl::Duration keep_alive_interval, - absl::Duration keep_alive_timeout) { - // Check if it has been too long since we received a frame from our - // endpoint. - auto last_read_time = endpoint_channel->GetLastReadTimestamp(); - if (last_read_time != kInvalidTimestamp && - SystemClock::ElapsedRealtime() > (last_read_time + keep_alive_timeout)) { - NEARBY_LOG(INFO, "Receive timeout expired; aborting KeepAlive worker."); + absl::Duration keep_alive_timeout, Mutex* keep_alive_waiter_mutex, + ConditionVariable* keep_alive_waiter) { + // Check if it has been too long since we received a frame from our endpoint. + absl::Time last_read_time = endpoint_channel->GetLastReadTimestamp(); + absl::Duration duration_until_timeout = + last_read_time == kInvalidTimestamp + ? keep_alive_timeout + : last_read_time + keep_alive_timeout - + SystemClock::ElapsedRealtime(); + if (duration_until_timeout <= absl::ZeroDuration()) { return ExceptionOr(false); } - // Attempt to send the KeepAlive frame over the endpoint channel - if the - // write fails, our super class will loop back around and try our luck again - // in case there's been a replacement for this endpoint. - Exception write_exception = endpoint_channel->Write(parser::ForKeepAlive()); - if (!write_exception.Ok()) { - return ExceptionOr(write_exception); + // If we haven't written anything to the endpoint for a while, attempt to send + // the KeepAlive frame over the endpoint channel. If the write fails, our + // super class will loop back around and try our luck again in case there's + // been a replacement for this endpoint. + absl::Time last_write_time = endpoint_channel->GetLastWriteTimestamp(); + absl::Duration duration_until_write_keep_alive = + last_write_time == kInvalidTimestamp + ? keep_alive_interval + : last_write_time + keep_alive_interval - + SystemClock::ElapsedRealtime(); + if (duration_until_write_keep_alive <= absl::ZeroDuration()) { + Exception write_exception = endpoint_channel->Write(parser::ForKeepAlive()); + if (!write_exception.Ok()) { + return ExceptionOr(write_exception); + } + duration_until_write_keep_alive = keep_alive_interval; } - // We sleep as the very last step because we want to minimize the caching of - // the EndpointChannel. If we do hold on to the EndpointChannel, and it's - // switched out from under us in BandwidthUpgradeManager, our write will - // trigger an erroneous write to the encryption context that will cascade - // into all our remote endpoint's future reads failing. - Exception sleep_exception = SystemClock::Sleep(keep_alive_interval); - if (!sleep_exception.Ok()) { - return ExceptionOr(sleep_exception); + absl::Duration wait_for = + std::min(duration_until_timeout, duration_until_write_keep_alive); + { + MutexLock lock(keep_alive_waiter_mutex); + Exception wait_exception = keep_alive_waiter->Wait(wait_for); + if (!wait_exception.Ok()) { + return ExceptionOr(wait_exception); + } } return ExceptionOr(true); @@ -414,11 +427,11 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client, }); }); - // For every endpoint, there's only one KeepAliveManager instance - // running on a dedicated thread. This instance will - // periodically send out a ping* to the endpoint while listening for an - // incoming pong**. If it fails to send the ping, or if no pong is heard - // within keep_alive_interval_, it initiates a disconnection. + // For every endpoint, there's only one KeepAliveManager instance running on + // a dedicated thread. This instance will periodically send out a ping* to + // the endpoint while listening for an incoming pong**. If it fails to send + // the ping, or if no pong is heard within keep_alive_timeout, it initiates + // a disconnection. // // (*) Bluetooth requires a constant outgoing stream of messages. If // there's silence, Android will break the socket. This is why we ping. @@ -428,13 +441,17 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client, NEARBY_LOGS(VERBOSE) << "EndpointManager enabling KeepAlive for endpoint " << endpoint_id; endpoint_state.StartEndpointKeepAliveManager( - [this, client, endpoint_id, keep_alive_interval, keep_alive_timeout]() { + [this, client, endpoint_id, keep_alive_interval, keep_alive_timeout]( + Mutex* keep_alive_waiter_mutex, + ConditionVariable* keep_alive_waiter) { EndpointChannelLoopRunnable( "KeepAliveManager", client, endpoint_id, - [this, keep_alive_interval, - keep_alive_timeout](EndpointChannel* channel) { - return HandleKeepAlive(channel, keep_alive_interval, - keep_alive_timeout); + [this, keep_alive_interval, keep_alive_timeout, + keep_alive_waiter_mutex, + keep_alive_waiter](EndpointChannel* channel) { + return HandleKeepAlive( + channel, keep_alive_interval, keep_alive_timeout, + keep_alive_waiter_mutex, keep_alive_waiter); }); }); NEARBY_LOGS(INFO) << "Registering endpoint " << endpoint_id @@ -456,7 +473,7 @@ void EndpointManager::UnregisterEndpoint(ClientProxy* client, RunOnEndpointManagerThread( "unregister-endpoint", [this, client, endpoint_id, &latch]() { RemoveEndpoint(client, endpoint_id, - client->IsConnectedToEndpoint(endpoint_id)); + /*notify=*/client->IsConnectedToEndpoint(endpoint_id)); latch.CountDown(); }); latch.Await(); @@ -494,8 +511,7 @@ void EndpointManager::DiscardEndpoint(ClientProxy* client, NEARBY_LOGS(VERBOSE) << "DiscardEndpoint for endpoint " << endpoint_id; RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id]() { RemoveEndpoint(client, endpoint_id, - /*notify=*/ - client->IsConnectedToEndpoint(endpoint_id)); + /*notify=*/client->IsConnectedToEndpoint(endpoint_id)); }); } @@ -619,15 +635,20 @@ std::vector EndpointManager::SendTransferFrameBytes( } EndpointManager::EndpointState::~EndpointState() { - // We must unregister the endpoint first to signal the runnables - // that they should exit their loops. SingleThreadExecutor destructors will - // wait for the workers to finish. - // |channel_manager_| is null when we moved from this object (in move - // constructor) which prevents unregistering the channel prematurely. - if (channel_manager_ != nullptr) { + // We must unregister the endpoint first to signal the runnables that they + // should exit their loops. SingleThreadExecutor destructors will wait for the + // workers to finish. |channel_manager_| is null after moved from this object + // (in move constructor) which prevents unregistering the channel prematurely. + if (channel_manager_) { NEARBY_LOG(VERBOSE, "EndpointState destructor %s", endpoint_id_.c_str()); channel_manager_->UnregisterChannelForEndpoint(endpoint_id_); } + + // Make sure the KeepAlive thread isn't blocking shutdown. + if (keep_alive_waiter_mutex_ && keep_alive_waiter_) { + MutexLock lock(keep_alive_waiter_mutex_.get()); + keep_alive_waiter_->Notify(); + } } void EndpointManager::EndpointState::StartEndpointReader(Runnable&& runnable) { @@ -635,8 +656,13 @@ void EndpointManager::EndpointState::StartEndpointReader(Runnable&& runnable) { } void EndpointManager::EndpointState::StartEndpointKeepAliveManager( - Runnable&& runnable) { - keep_alive_thread_.Execute("keep-alive", std::move(runnable)); + std::function runnable) { + keep_alive_thread_.Execute( + "keep-alive", + [runnable, keep_alive_waiter_mutex = keep_alive_waiter_mutex_.get(), + keep_alive_waiter = keep_alive_waiter_.get()]() { + runnable(keep_alive_waiter_mutex, keep_alive_waiter); + }); } void EndpointManager::RunOnEndpointManagerThread(const std::string& name, diff --git a/cpp/core/internal/endpoint_manager.h b/cpp/core/internal/endpoint_manager.h index 36646e90..ec1a1619 100644 --- a/cpp/core/internal/endpoint_manager.h +++ b/cpp/core/internal/endpoint_manager.h @@ -28,6 +28,7 @@ #include "core/listeners.h" #include "platform/base/byte_array.h" #include "platform/base/runnable.h" +#include "platform/public/condition_variable.h" #include "platform/public/count_down_latch.h" #include "platform/public/multi_thread_executor.h" #include "platform/public/single_thread_executor.h" @@ -145,25 +146,44 @@ class EndpointManager { public: EndpointState(const std::string& endpoint_id, EndpointChannelManager* channel_manager) - : endpoint_id_{endpoint_id}, channel_manager_{channel_manager} {} + : endpoint_id_{endpoint_id}, + channel_manager_{channel_manager}, + keep_alive_waiter_mutex_{std::make_unique()}, + keep_alive_waiter_{std::make_unique( + keep_alive_waiter_mutex_.get())} {} + EndpointState(const EndpointState&) = delete; - // default move constructor would not reset |channel_manager_| + // The default move constructor would not reset |channel_manager_|, for + // example. This needs to be nullified so the destructor shutdown logic is + // bypassed when objects are moved. EndpointState(EndpointState&& other) : endpoint_id_{std::move(other.endpoint_id_)}, channel_manager_{std::exchange(other.channel_manager_, nullptr)}, reader_thread_{std::move(other.reader_thread_)}, + keep_alive_waiter_mutex_{ + std::exchange(other.keep_alive_waiter_mutex_, nullptr)}, + keep_alive_waiter_{std::exchange(other.keep_alive_waiter_, nullptr)}, keep_alive_thread_{std::move(other.keep_alive_thread_)} {} EndpointState& operator=(const EndpointState&) = delete; EndpointState&& operator=(EndpointState&&) = delete; ~EndpointState(); void StartEndpointReader(Runnable&& runnable); - void StartEndpointKeepAliveManager(Runnable&& runnable); + void StartEndpointKeepAliveManager( + std::function runnable); private: const std::string endpoint_id_; EndpointChannelManager* channel_manager_; SingleThreadExecutor reader_thread_; + + // Use a condition variable so we can wait on the thread but still be able + // to wake it up before shutting down. We don't want to just sleep and risk + // blocking shutdown. Note: Create the mutex/condition variable on the heap + // so raw pointers sent to HandleKeepAlive() aren't invalidated during + // std::move operations. + mutable std::unique_ptr keep_alive_waiter_mutex_; + std::unique_ptr keep_alive_waiter_; SingleThreadExecutor keep_alive_thread_; }; @@ -191,7 +211,9 @@ class EndpointManager { ExceptionOr HandleKeepAlive(EndpointChannel* endpoint_channel, absl::Duration keep_alive_interval, - absl::Duration keep_alive_timeout); + absl::Duration keep_alive_timeout, + Mutex* keep_alive_waiter_mutex, + ConditionVariable* keep_alive_waiter); // Waits for a given endpoint EndpointChannelLoopRunnable() workers to // terminate. diff --git a/cpp/core/internal/endpoint_manager_test.cc b/cpp/core/internal/endpoint_manager_test.cc index 04d2392e..dae1057b 100644 --- a/cpp/core/internal/endpoint_manager_test.cc +++ b/cpp/core/internal/endpoint_manager_test.cc @@ -64,6 +64,7 @@ class MockEndpointChannel : public EndpointChannel { MOCK_METHOD(void, Pause, (), (override)); MOCK_METHOD(void, Resume, (), (override)); MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const override)); + MOCK_METHOD(absl::Time, GetLastWriteTimestamp, (), (const override)); MOCK_METHOD(void, SetAnalyticsRecorder, (analytics::AnalyticsRecorder*, const std::string&), (override)); @@ -108,6 +109,8 @@ class EndpointManagerTest : public ::testing::Test { EXPECT_CALL(*channel, GetMedium()).WillRepeatedly(Return(Medium::BLE)); EXPECT_CALL(*channel, GetLastReadTimestamp()) .WillRepeatedly(Return(start_time_)); + EXPECT_CALL(*channel, GetLastWriteTimestamp()) + .WillRepeatedly(Return(start_time_)); EXPECT_CALL(mock_listener_.initiated_cb, Call).Times(1); em_.RegisterEndpoint(&client_, endpoint_id_, info_, options_, std::move(channel), listener_, connection_token);