Internal change

PiperOrigin-RevId: 363243327
This commit is contained in:
hai007
2021-03-16 12:27:41 -07:00
committed by Copybara-Service
parent 053833d103
commit bb5043b95a
7 changed files with 40 additions and 49 deletions
+4 -6
View File
@@ -54,7 +54,7 @@ BasePcpHandler::BasePcpHandler(Mediums* mediums,
BasePcpHandler::~BasePcpHandler() {
NEARBY_LOGS(INFO) << "BasePcpHandler: going down; strategy="
<< strategy_.GetName() << "; handle=" << handle_;
<< strategy_.GetName();
DisconnectFromEndpointManager();
// Stop all the ongoing Runnables (as gracefully as possible).
NEARBY_LOGS(INFO) << "BasePcpHandler: bringing down executors; strategy="
@@ -68,10 +68,10 @@ BasePcpHandler::~BasePcpHandler() {
void BasePcpHandler::DisconnectFromEndpointManager() {
if (stop_.Set(true)) return;
NEARBY_LOGS(INFO) << "BasePcpHandler: Unregister from EPM; strategy="
<< strategy_.GetName() << "; handle=" << handle_;
<< strategy_.GetName();
// Unregister ourselves from EPM message dispatcher.
endpoint_manager_->UnregisterFrameProcessor(V1Frame::CONNECTION_RESPONSE,
handle_, true);
this);
}
Status BasePcpHandler::StartAdvertising(ClientProxy* client,
@@ -284,9 +284,7 @@ void BasePcpHandler::OnEncryptionSuccessRunnable(
endpoint_id.c_str());
// Set ourselves up so that we receive all acceptance/rejection messages
handle_ = endpoint_manager_->RegisterFrameProcessor(
V1Frame::CONNECTION_RESPONSE,
static_cast<EndpointManager::FrameProcessor*>(this));
endpoint_manager_->RegisterFrameProcessor(V1Frame::CONNECTION_RESPONSE, this);
// Now we register our endpoint so that we can listen for both sides to
// accept.
-1
View File
@@ -511,7 +511,6 @@ class BasePcpHandler : public PcpHandler,
Prng prng_;
EncryptionRunner encryption_runner_;
BwuManager* bwu_manager_;
EndpointManager::FrameProcessor::Handle handle_ = nullptr;
};
} // namespace connections
+24 -24
View File
@@ -269,9 +269,8 @@ EndpointManager::~EndpointManager() {
NEARBY_LOG(INFO, "EndpointManager is down");
}
EndpointManager::FrameProcessor::Handle EndpointManager::RegisterFrameProcessor(
void EndpointManager::RegisterFrameProcessor(
V1Frame::FrameType frame_type, EndpointManager::FrameProcessor* processor) {
const FrameProcessor::Handle handle = processor;
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, frame_type, &latch, processor]() {
auto it = frame_processors_.find(frame_type);
@@ -287,40 +286,41 @@ EndpointManager::FrameProcessor::Handle EndpointManager::RegisterFrameProcessor(
latch.CountDown();
});
latch.Await();
return handle;
}
void EndpointManager::UnregisterFrameProcessor(V1Frame::FrameType frame_type,
const void* handle, bool sync) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: handle=" << handle;
if (handle == nullptr) return;
void EndpointManager::UnregisterFrameProcessor(
V1Frame::FrameType frame_type,
const EndpointManager::FrameProcessor* processor) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: processor ="
<< processor;
if (processor == nullptr) return;
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, frame_type, handle, &latch, sync]() {
RunOnEndpointManagerThread([this, frame_type, processor, &latch]() {
auto it = frame_processors_.find(frame_type);
if (it == frame_processors_.end()) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: handle="
<< handle;
if (sync) latch.CountDown();
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: processor="
<< processor;
latch.CountDown();
return;
}
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: handle=" << handle;
if (it->second == handle) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: processor="
<< processor;
if (it->second == processor) {
frame_processors_.erase(it);
NEARBY_LOGS(INFO) << "Unregistered: type=" << frame_type
<< "; processor=" << handle << "; self=" << this;
<< "; processor=" << processor << "; self=" << this;
} else {
NEARBY_LOG(INFO,
"Failed to unregister: type=%d; handle mismatch: passed=%p, "
"expected=%p",
frame_type, handle, it->second);
NEARBY_LOG(
INFO,
"Failed to unregister: type=%d; processor mismatch: passed=%p, "
"expected=%p",
frame_type, processor, it->second);
}
if (sync) latch.CountDown();
latch.CountDown();
});
if (sync) {
latch.Await();
NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type
<< "; processor=" << handle << "; self=" << this;
}
latch.Await();
NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
}
EndpointManager::FrameProcessor* EndpointManager::GetFrameProcessor(
+3 -6
View File
@@ -60,8 +60,6 @@ class EndpointManager {
public:
class FrameProcessor {
public:
using Handle = void*;
virtual ~FrameProcessor() = default;
// @EndpointManagerReaderThread
@@ -93,12 +91,11 @@ class EndpointManager {
// Invoked from the constructors of the various *Manager components that make
// up the OfflineServiceController implementation.
// FrameProcessor* instances are of dynamic duration and survive all sessions.
// returns unique handle to be used for unregistering.
// Blocks until registration is complete.
FrameProcessor::Handle RegisterFrameProcessor(V1Frame::FrameType frame_type,
FrameProcessor* processor);
void RegisterFrameProcessor(V1Frame::FrameType frame_type,
FrameProcessor* processor);
void UnregisterFrameProcessor(V1Frame::FrameType frame_type,
const void* handle, bool sync = false);
const FrameProcessor* processor);
// Invoked from the different PcpHandler implementations (of which there can
// be only one at a time).
+7 -7
View File
@@ -188,10 +188,9 @@ TEST_F(EndpointManagerTest, RegisterFrameProcessorWorks) {
// Register frame processor, then register endpoint.
// Endpoint will read one frame, then fail to read more and terminate.
// On disconnection, it will notify frame processor and we verify that.
const void* handle = em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST,
connect_request.get());
em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST,
connect_request.get());
processors_.emplace_back(std::move(connect_request));
EXPECT_NE(handle, nullptr);
RegisterEndpoint(std::move(endpoint_channel));
}
@@ -206,11 +205,12 @@ TEST_F(EndpointManagerTest, UnregisterFrameProcessorWorks) {
auto connect_request = std::make_unique<StrictMock<MockFrameProcessor>>();
// Register frame processor and immediately unregister it.
const void* handle = em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST,
connect_request.get());
em_.RegisterFrameProcessor(V1Frame::CONNECTION_REQUEST,
connect_request.get());
em_.UnregisterFrameProcessor(V1Frame::CONNECTION_REQUEST,
connect_request.get());
processors_.emplace_back(std::move(connect_request));
EXPECT_NE(handle, nullptr);
em_.UnregisterFrameProcessor(V1Frame::CONNECTION_REQUEST, handle);
// Endpoint will not send OnDisconnect notification to frame processor.
RegisterEndpoint(std::move(endpoint_channel), false);
em_.UnregisterEndpoint(&client_, endpoint_id_);
+2 -4
View File
@@ -217,8 +217,7 @@ Payload::Id PayloadManager::CreateOutgoingPayload(
PayloadManager::PayloadManager(EndpointManager& endpoint_manager)
: endpoint_manager_(&endpoint_manager) {
handle_ = endpoint_manager_->RegisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER,
this);
endpoint_manager_->RegisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, this);
}
void PayloadManager::CancelAllPayloads() {
@@ -249,8 +248,7 @@ void PayloadManager::CancelAllPayloads() {
void PayloadManager::DisconnectFromEndpointManager() {
if (shutdown_.Set(true)) return;
// Unregister ourselves from the FrameProcessors.
endpoint_manager_->UnregisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER,
handle_, true);
endpoint_manager_->UnregisterFrameProcessor(V1Frame::PAYLOAD_TRANSFER, this);
}
PayloadManager::~PayloadManager() {
-1
View File
@@ -281,7 +281,6 @@ class PayloadManager : public EndpointManager::FrameProcessor {
void CancelAllPayloads() ABSL_LOCKS_EXCLUDED(mutex_);
mutable Mutex mutex_;
EndpointManager::FrameProcessor::Handle handle_;
AtomicBoolean shutdown_{false};
std::unique_ptr<CountDownLatch> shutdown_barrier_;
int send_payload_count_ = 0;