TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 22:56:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

[Nearby Connections][C++] Wake up keep-alive thread to unblock destruction

Browse Source 
PiperOrigin-RevId: 399693890
This commit is contained in:
nohle
2021-09-29 08:39:37 -07:00
committed by Copybara-Service
parent 1f0ec837b6
commit bc8769c64e
7 changed files with 126 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/base_endpoint_channel.cc
+9
View File
@@ -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;
}
cpp/core/internal/base_endpoint_channel.h
+13 -1
View File
@@ -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
cpp/core/internal/encryption_runner_test.cc
+3
View File
@@ -42,6 +42,7 @@ class FakeEndpointChannel : public EndpointChannel {
: ExceptionOr<ByteArray>{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 {
cpp/core/internal/endpoint_channel.h
+4
View File
@@ -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,
cpp/core/internal/endpoint_manager.cc
+68 -42
View File
@@ -214,32 +214,45 @@ ExceptionOr<bool> EndpointManager::HandleData(
ExceptionOr<bool> 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<bool>(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<bool>(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<bool>(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<bool>(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<bool>(wait_exception);
}
}
return ExceptionOr<bool>(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<std::string> 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<void(Mutex*, ConditionVariable*)> 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,
cpp/core/internal/endpoint_manager.h
+26 -4
View File
@@ -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<Mutex>()},
keep_alive_waiter_{std::make_unique<ConditionVariable>(
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<void(Mutex*, ConditionVariable*)> 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<Mutex> keep_alive_waiter_mutex_;
std::unique_ptr<ConditionVariable> keep_alive_waiter_;
SingleThreadExecutor keep_alive_thread_;
};
@@ -191,7 +211,9 @@ class EndpointManager {
ExceptionOr<bool> 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.
cpp/core/internal/endpoint_manager_test.cc
+3
View File
@@ -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);