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

Internal change

Browse Source 
PiperOrigin-RevId: 371332999
This commit is contained in:
hai007
2021-04-30 07:57:08 -07:00
committed by Copybara-Service
parent 857958c87f
commit 54f30cd81d
17 changed files with 168 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/BUILD
+1
View File
@@ -26,6 +26,7 @@ cc_library(
deps = [
":core_types",
"//core/internal",
"//platform/base",
"//platform/public:logging",
"//platform/public:types",
"//absl/strings",
cpp/core/core.cc
+17
View File
@@ -18,6 +18,7 @@
#include <vector>
#include "core/options.h"
#include "platform/base/feature_flags.h"
#include "platform/public/count_down_latch.h"
#include "platform/public/logging.h"
#include "absl/time/clock.h"
@@ -78,6 +79,22 @@ void Core::RequestConnection(absl::string_view endpoint_id,
ResultCallback callback) {
assert(!endpoint_id.empty());
// Assign the default from feature flags for the keep-alive frame interval and
// timeout values if client don't mind them or has the unexpected ones.
if (options.keep_alive_interval_millis == 0 ||
options.keep_alive_timeout_millis == 0 ||
options.keep_alive_interval_millis >= options.keep_alive_timeout_millis) {
NEARBY_LOG(
WARNING,
"Client request connection with keep-alive frame as interval=%d, "
"timeout=%d, which is un-expected. Change to default.",
options.keep_alive_interval_millis, options.keep_alive_timeout_millis);
options.keep_alive_interval_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_interval_millis;
options.keep_alive_timeout_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_timeout_millis;
}
router_.RequestConnection(&client_, endpoint_id, info, options, callback);
}
cpp/core/internal/base_pcp_handler.cc
+39 -3
View File
@@ -364,6 +364,10 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
connection_info.options.remote_bluetooth_mac_address,
.fast_advertisement_service_uuid =
connection_info.options.fast_advertisement_service_uuid,
.keep_alive_interval_millis =
connection_info.options.keep_alive_interval_millis,
.keep_alive_timeout_millis =
connection_info.options.keep_alive_timeout_millis,
},
std::move(connection_info.channel), connection_info.listener);
@@ -498,7 +502,9 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client,
// endpoint about ourselves.
Exception write_exception = WriteConnectionRequestFrame(
channel.get(), client->GetLocalEndpointId(), info.endpoint_info,
nonce, GetSupportedConnectionMediumsByPriority(options));
nonce, GetSupportedConnectionMediumsByPriority(options),
options.keep_alive_interval_millis,
options.keep_alive_timeout_millis);
if (!write_exception.Ok()) {
NEARBY_LOG(INFO, "Failed to send connection request: endpoint_id=%s",
endpoint_id.c_str());
@@ -647,10 +653,13 @@ bool BasePcpHandler::CanReceiveIncomingConnection(ClientProxy* client) const {
Exception BasePcpHandler::WriteConnectionRequestFrame(
EndpointChannel* endpoint_channel, const std::string& local_endpoint_id,
const ByteArray& local_endpoint_info, std::int32_t nonce,
const std::vector<proto::connections::Medium>& supported_mediums) {
const std::vector<proto::connections::Medium>& supported_mediums,
std::int32_t keep_alive_interval_millis,
std::int32_t keep_alive_timeout_millis) {
return endpoint_channel->Write(parser::ForConnectionRequest(
local_endpoint_id, local_endpoint_info, nonce, /*supports_5_ghz =*/false,
/*bssid=*/std::string{}, supported_mediums));
/*bssid=*/std::string{}, supported_mediums, keep_alive_interval_millis,
keep_alive_timeout_millis));
}
void BasePcpHandler::ProcessPreConnectionInitiationFailure(
@@ -1076,6 +1085,32 @@ Exception BasePcpHandler::OnIncomingConnection(
? connection_request.endpoint_info()
: connection_request.endpoint_name()};
// Retrieve the keep-alive frame interval and timeout fields. If the frame
// doesn't have those fields, we need to get them as default from feature
// flags to prevent 0-values causing thread ill.
ConnectionOptions options = {.keep_alive_interval_millis = 0,
.keep_alive_timeout_millis = 0};
if (connection_request.has_keep_alive_interval_millis() &&
connection_request.has_keep_alive_timeout_millis()) {
options.keep_alive_interval_millis =
connection_request.keep_alive_interval_millis();
options.keep_alive_timeout_millis =
connection_request.keep_alive_timeout_millis();
}
if (options.keep_alive_interval_millis == 0 ||
options.keep_alive_timeout_millis == 0 ||
options.keep_alive_interval_millis >= options.keep_alive_timeout_millis) {
NEARBY_LOG(WARNING,
"Incoming connection has wrong keep-alive frame interval=%d, "
"timeout=%d values; correct them as default.",
options.keep_alive_interval_millis,
options.keep_alive_timeout_millis);
options.keep_alive_interval_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_interval_millis;
options.keep_alive_timeout_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_timeout_millis;
}
// We've successfully connected to the device, and are now about to jump on to
// the EncryptionRunner thread to start running our encryption protocol. We'll
// mark ourselves as pending in case we get another call to RequestConnection
@@ -1090,6 +1125,7 @@ Exception BasePcpHandler::OnIncomingConnection(
.is_incoming = true,
.start_time = start_time,
.listener = advertising_listener_,
.options = options,
.supported_mediums =
parser::ConnectionRequestMediumsToMediums(
connection_request),
cpp/core/internal/base_pcp_handler.h
+3 -1
View File
@@ -374,7 +374,9 @@ class BasePcpHandler : public PcpHandler,
static Exception WriteConnectionRequestFrame(
EndpointChannel* endpoint_channel, const std::string& local_endpoint_id,
const ByteArray& local_endpoint_info, std::int32_t nonce,
const std::vector<proto::connections::Medium>& supported_mediums);
const std::vector<proto::connections::Medium>& supported_mediums,
std::int32_t keep_alive_interval_millis,
std::int32_t keep_alive_timeout_millis);
static constexpr absl::Duration kConnectionRequestReadTimeout =
absl::Seconds(2);
cpp/core/internal/base_pcp_handler_test.cc
+4
View File
@@ -329,6 +329,10 @@ class BasePcpHandlerTest
ConnectionOptions options{
.remote_bluetooth_mac_address =
ByteArray{std::string("\x12\x34\x56\x78\x9a\xbc")},
.keep_alive_interval_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_interval_millis,
.keep_alive_timeout_millis =
FeatureFlags::GetInstance().GetFlags().keep_alive_timeout_millis,
};
EXPECT_CALL(mock_discovery_listener_.endpoint_found_cb, Call);
EXPECT_CALL(*pcp_handler, CanSendOutgoingConnection)
cpp/core/internal/endpoint_manager.cc
+28 -18
View File
@@ -20,7 +20,6 @@
#include "core/internal/endpoint_channel.h"
#include "core/internal/offline_frames.h"
#include "platform/base/exception.h"
#include "platform/base/feature_flags.h"
#include "platform/public/count_down_latch.h"
#include "platform/public/logging.h"
#include "platform/public/mutex_lock.h"
@@ -31,8 +30,6 @@ namespace connections {
using ::location::nearby::proto::connections::Medium;
constexpr absl::Duration EndpointManager::kKeepAliveWriteInterval;
constexpr absl::Duration EndpointManager::kKeepAliveReadTimeout;
constexpr absl::Duration EndpointManager::kProcessEndpointDisconnectionTimeout;
constexpr absl::Time EndpointManager::kInvalidTimestamp;
@@ -220,13 +217,13 @@ ExceptionOr<bool> EndpointManager::HandleData(
}
ExceptionOr<bool> EndpointManager::HandleKeepAlive(
EndpointChannel* endpoint_channel) {
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 + EndpointManager::kKeepAliveReadTimeout)) {
SystemClock::ElapsedRealtime() > (last_read_time + keep_alive_timeout)) {
NEARBY_LOG(INFO, "Receive timeout expired; aborting KeepAlive worker.");
return ExceptionOr<bool>(false);
}
@@ -244,8 +241,7 @@ ExceptionOr<bool> EndpointManager::HandleKeepAlive(
// 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(EndpointManager::kKeepAliveWriteInterval);
Exception sleep_exception = SystemClock::Sleep(keep_alive_interval);
if (!sleep_exception.Ok()) {
return ExceptionOr<bool>(sleep_exception);
}
@@ -408,6 +404,18 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
EnsureWorkersTerminated(endpoint_id);
}
}
absl::Duration keep_alive_interval =
absl::Milliseconds(options.keep_alive_interval_millis);
absl::Duration keep_alive_timeout =
absl::Milliseconds(options.keep_alive_timeout_millis);
NEARBY_LOG(INFO,
"Registering endpoint %s for client %d with keep-alive frame as "
"interval=%s, timeout=%s",
endpoint_id.c_str(), client->GetClientId(),
absl::FormatDuration(keep_alive_interval).c_str(),
absl::FormatDuration(keep_alive_timeout).c_str());
// Pass ownership of channel to EndpointChannelManager
NEARBY_LOG(INFO, "Registering endpoint with channel manager: id=%s",
endpoint_id.c_str());
@@ -443,8 +451,7 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
// running on the keep_alive_executor_ pool. 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 kKeepAliveReadTimeoutMillis milliseconds, it initiates a
// disconnection.
// within keep_alive_interval_, 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.
@@ -454,14 +461,17 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
//
// Using weak_ptr just in case the barrier is freed, to save the UAF crash
// in b/179800119.
StartEndpointKeepAliveManager([this, client, endpoint_id,
barrier = std::weak_ptr<CountDownLatch>(
endpoint_state.barrier)]() {
EndpointChannelLoopRunnable("KeepAliveManager", client, endpoint_id,
barrier, [this](EndpointChannel* channel) {
return HandleKeepAlive(channel);
});
});
StartEndpointKeepAliveManager(
[this, client, endpoint_id, keep_alive_interval, keep_alive_timeout,
barrier = std::weak_ptr<CountDownLatch>(endpoint_state.barrier)]() {
EndpointChannelLoopRunnable(
"KeepAliveManager", client, endpoint_id, barrier,
[this, keep_alive_interval,
keep_alive_timeout](EndpointChannel* channel) {
return HandleKeepAlive(channel, keep_alive_interval,
keep_alive_timeout);
});
});
NEARBY_LOG(INFO, "Workers started, notifying client; id=%s",
endpoint_id.c_str());
cpp/core/internal/endpoint_manager.h
+3 -5
View File
@@ -171,7 +171,9 @@ class EndpointManager {
ClientProxy* client_proxy,
EndpointChannel* endpoint_channel);
ExceptionOr<bool> HandleKeepAlive(EndpointChannel* endpoint_channel);
ExceptionOr<bool> HandleKeepAlive(EndpointChannel* endpoint_channel,
absl::Duration keep_alive_interval,
absl::Duration keep_alive_timeout);
// Waits for a given endpoint EndpointChannelLoopRunnable() workers to
// terminate.
@@ -190,10 +192,6 @@ class EndpointManager {
static void WaitForLatch(const std::string& method_name,
CountDownLatch* latch, std::int32_t timeout_millis);
static constexpr absl::Duration kKeepAliveWriteInterval =
absl::Milliseconds(5000);
static constexpr absl::Duration kKeepAliveReadTimeout =
absl::Milliseconds(30000);
static constexpr absl::Duration kProcessEndpointDisconnectionTimeout =
absl::Milliseconds(2000);
static constexpr std::int32_t kMaxConcurrentEndpoints = 50;
cpp/core/internal/endpoint_manager_test.cc
+3 -2
View File
@@ -176,8 +176,9 @@ TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) {
auto endpoint_channel = std::make_unique<MockEndpointChannel>();
auto connect_request = std::make_unique<MockFrameProcessor>();
ByteArray endpoint_info{"endpoint_name"};
auto read_data = parser::ForConnectionRequest(
"endpoint_id", endpoint_info, 1234, false, "", std::vector{Medium::BLE});
auto read_data =
parser::ForConnectionRequest("endpoint_id", endpoint_info, 1234, false,
"", std::vector{Medium::BLE}, 0, 0);
EXPECT_CALL(*connect_request, OnIncomingFrame);
EXPECT_CALL(*connect_request, OnEndpointDisconnect);
EXPECT_CALL(*endpoint_channel, Read())
cpp/core/internal/offline_frames.cc
+13 -2
View File
@@ -64,9 +64,12 @@ V1Frame::FrameType GetFrameType(const OfflineFrame& frame) {
ByteArray ForConnectionRequest(const std::string& endpoint_id,
const ByteArray& endpoint_info,
std::int32_t nonce, bool supports_5_ghz,
std::int32_t nonce,
bool supports_5_ghz,
const std::string& bssid,
const std::vector<Medium>& mediums) {
const std::vector<Medium>& mediums,
std::int32_t keep_alive_interval_millis,
std::int32_t keep_alive_timeout_millis) {
OfflineFrame frame;
frame.set_version(OfflineFrame::V1);
@@ -89,6 +92,14 @@ ByteArray ForConnectionRequest(const std::string& endpoint_id,
connection_request->add_mediums(MediumToConnectionRequestMedium(medium));
}
}
if (keep_alive_interval_millis > 0) {
connection_request->set_keep_alive_interval_millis(
keep_alive_interval_millis);
}
if (keep_alive_timeout_millis > 0) {
connection_request->set_keep_alive_timeout_millis(
keep_alive_timeout_millis);
}
return ToBytes(std::move(frame));
}
cpp/core/internal/offline_frames.h
+5 -2
View File
@@ -44,9 +44,12 @@ V1Frame::FrameType GetFrameType(const OfflineFrame& offline_frame);
// Builds Connection Request / Response messages.
ByteArray ForConnectionRequest(const std::string& endpoint_id,
const ByteArray& endpoint_info,
std::int32_t nonce, bool supports_5_ghz,
std::int32_t nonce,
bool supports_5_ghz,
const std::string& bssid,
const std::vector<Medium>& mediums);
const std::vector<Medium>& mediums,
std::int32_t keep_alive_interval_millis,
std::int32_t keep_alive_timeout_millis);
ByteArray ForConnectionResponse(std::int32_t status);
// Builds Payload transfer messages.
cpp/core/internal/offline_frames_test.cc
+8 -1
View File
@@ -43,6 +43,8 @@ constexpr std::array<Medium, 9> kMediums = {
Medium::BLE, Medium::WIFI_LAN, Medium::WIFI_AWARE,
Medium::NFC, Medium::WIFI_DIRECT, Medium::WEB_RTC,
};
constexpr int kKeepAliveIntervalMillis = 1000;
constexpr int kKeepAliveTimeoutMillis = 5000;
TEST(OfflineFramesTest, CanParseMessageFromBytes) {
OfflineFrame tx_message;
@@ -57,6 +59,8 @@ TEST(OfflineFramesTest, CanParseMessageFromBytes) {
sub_frame->set_endpoint_name(kEndpointName);
sub_frame->set_endpoint_info(kEndpointName);
sub_frame->set_nonce(kNonce);
sub_frame->set_keep_alive_interval_millis(kKeepAliveIntervalMillis);
sub_frame->set_keep_alive_timeout_millis(kKeepAliveTimeoutMillis);
auto* medium_metadata = sub_frame->mutable_medium_metadata();
medium_metadata->set_supports_5_ghz(kSupports5ghz);
@@ -101,12 +105,15 @@ TEST(OfflineFramesTest, CanGenerateConnectionRequest) {
mediums: NFC
mediums: WIFI_DIRECT
mediums: WEB_RTC
keep_alive_interval_millis: 1000
keep_alive_timeout_millis : 5000
>
>)pb";
ByteArray bytes = ForConnectionRequest(
std::string(kEndpointId), ByteArray{std::string(kEndpointName)}, kNonce,
kSupports5ghz, std::string(kBssid),
std::vector(kMediums.begin(), kMediums.end()));
std::vector(kMediums.begin(), kMediums.end()), kKeepAliveIntervalMillis,
kKeepAliveTimeoutMillis);
auto response = FromBytes(bytes);
ASSERT_TRUE(response.ok());
OfflineFrame message = FromBytes(bytes).result();
cpp/core/internal/offline_frames_validator_test.cc
+14 -7
View File
@@ -45,6 +45,8 @@ constexpr std::array<Medium, 9> kMediums = {
Medium::BLE, Medium::WIFI_LAN, Medium::WIFI_AWARE,
Medium::NFC, Medium::WIFI_DIRECT, Medium::WEB_RTC,
};
constexpr int kKeepAliveIntervalMillis = 1000;
constexpr int kKeepAliveTimeoutMillis = 5000;
TEST(OfflineFramesValidatorTest, ValidatesAsOkWithValidConnectionRequestFrame) {
OfflineFrame offline_frame;
@@ -52,7 +54,8 @@ TEST(OfflineFramesValidatorTest, ValidatesAsOkWithValidConnectionRequestFrame) {
ByteArray bytes = ForConnectionRequest(
std::string(kEndpointId), ByteArray{std::string(kEndpointName)}, kNonce,
kSupports5ghz, std::string(kBssid),
std::vector(kMediums.begin(), kMediums.end()));
std::vector(kMediums.begin(), kMediums.end()), kKeepAliveIntervalMillis,
kKeepAliveTimeoutMillis);
offline_frame.ParseFromString(std::string(bytes));
auto ret_value = EnsureValidOfflineFrame(offline_frame);
@@ -67,7 +70,8 @@ TEST(OfflineFramesValidatorTest,
ByteArray bytes = ForConnectionRequest(
std::string(kEndpointId), ByteArray{std::string(kEndpointName)}, kNonce,
kSupports5ghz, std::string(kBssid),
std::vector(kMediums.begin(), kMediums.end()));
std::vector(kMediums.begin(), kMediums.end()), kKeepAliveIntervalMillis,
kKeepAliveTimeoutMillis);
offline_frame.ParseFromString(std::string(bytes));
auto* v1_frame = offline_frame.mutable_v1();
@@ -86,7 +90,8 @@ TEST(OfflineFramesValidatorTest,
ByteArray bytes = ForConnectionRequest(
empty_enpoint_id, ByteArray{std::string(kEndpointName)}, kNonce,
kSupports5ghz, std::string(kBssid),
std::vector(kMediums.begin(), kMediums.end()));
std::vector(kMediums.begin(), kMediums.end()), kKeepAliveIntervalMillis,
kKeepAliveTimeoutMillis);
offline_frame.ParseFromString(std::string(bytes));
auto ret_value = EnsureValidOfflineFrame(offline_frame);
@@ -101,7 +106,8 @@ TEST(OfflineFramesValidatorTest,
ByteArray empty_endpoint_info;
ByteArray bytes = ForConnectionRequest(
std::string(kEndpointId), empty_endpoint_info, kNonce, kSupports5ghz,
std::string(kBssid), std::vector(kMediums.begin(), kMediums.end()));
std::string(kBssid), std::vector(kMediums.begin(), kMediums.end()),
kKeepAliveIntervalMillis, kKeepAliveTimeoutMillis);
offline_frame.ParseFromString(std::string(bytes));
auto ret_value = EnsureValidOfflineFrame(offline_frame);
@@ -116,8 +122,8 @@ TEST(OfflineFramesValidatorTest,
std::string empty_bssid;
ByteArray bytes = ForConnectionRequest(
std::string(kEndpointId), ByteArray{std::string(kEndpointName)}, kNonce,
kSupports5ghz, empty_bssid,
std::vector(kMediums.begin(), kMediums.end()));
kSupports5ghz, empty_bssid, std::vector(kMediums.begin(), kMediums.end()),
kKeepAliveIntervalMillis, kKeepAliveTimeoutMillis);
offline_frame.ParseFromString(std::string(bytes));
auto ret_value = EnsureValidOfflineFrame(offline_frame);
@@ -132,7 +138,8 @@ TEST(OfflineFramesValidatorTest,
std::vector<Medium> empty_mediums;
ByteArray bytes = ForConnectionRequest(
std::string(kEndpointId), ByteArray{std::string(kEndpointName)}, kNonce,
kSupports5ghz, std::string(kBssid), empty_mediums);
kSupports5ghz, std::string(kBssid), empty_mediums,
kKeepAliveIntervalMillis, kKeepAliveTimeoutMillis);
offline_frame.ParseFromString(std::string(bytes));
auto ret_value = EnsureValidOfflineFrame(offline_frame);
cpp/core/internal/offline_simulation_user.h
+9 -1
View File
@@ -50,7 +50,15 @@ class OfflineSimulationUser {
explicit OfflineSimulationUser(
absl::string_view device_name,
BooleanMediumSelector allowed = BooleanMediumSelector())
: info_{ByteArray{std::string(device_name)}},
: connection_options_{
.keep_alive_interval_millis = FeatureFlags::GetInstance()
.GetFlags()
.keep_alive_interval_millis,
.keep_alive_timeout_millis = FeatureFlags::GetInstance()
.GetFlags()
.keep_alive_timeout_millis,
},
info_{ByteArray{std::string(device_name)}},
options_{
.strategy = Strategy::kP2pCluster,
.allowed = allowed,
cpp/core/internal/simulation_user.h
+9 -1
View File
@@ -54,7 +54,15 @@ class SimulationUser {
explicit SimulationUser(
const std::string& device_name,
BooleanMediumSelector allowed = BooleanMediumSelector())
: info_{ByteArray{device_name}},
: connection_options_{
.keep_alive_interval_millis = FeatureFlags::GetInstance()
.GetFlags()
.keep_alive_interval_millis,
.keep_alive_timeout_millis = FeatureFlags::GetInstance()
.GetFlags()
.keep_alive_timeout_millis,
},
info_{ByteArray{device_name}},
options_{
.strategy = Strategy::kP2pCluster,
.allowed = allowed,
cpp/core/options.h
+2
View File
@@ -99,6 +99,8 @@ struct ConnectionOptions {
bool is_out_of_band_connection = false;
ByteArray remote_bluetooth_mac_address;
std::string fast_advertisement_service_uuid;
int keep_alive_interval_millis = 0;
int keep_alive_timeout_millis = 0;
// Verify if ConnectionOptions is in a not-initialized (Empty) state.
bool Empty() const { return strategy.IsNone(); }
// Bring ConnectionOptions to a not-initialized (Empty) state.