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: 366077078
This commit is contained in:
hai007
2021-03-31 11:24:52 -07:00
committed by Copybara-Service
parent 9c6b57b04e
commit e008c2edfc
2 changed files with 47 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/endpoint_manager.cc
+41 -49
View File
@@ -23,6 +23,7 @@
#include "platform/base/feature_flags.h"
#include "platform/public/count_down_latch.h"
#include "platform/public/logging.h"
#include "platform/public/mutex_lock.h"
#include "proto/connections_enums.pb.h"
namespace location {
namespace nearby {
@@ -304,22 +305,16 @@ EndpointManager::~EndpointManager() {
void EndpointManager::RegisterFrameProcessor(
V1Frame::FrameType frame_type, EndpointManager::FrameProcessor* processor) {
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, frame_type, &latch, processor]() {
auto it = frame_processors_.find(frame_type);
if (it != frame_processors_.end()) {
NEARBY_LOGS(INFO) << "Frame processor found: updated; type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
LockedFrameProcessor frame_processor_lock(&it->second);
frame_processor_lock.set(processor);
} else {
NEARBY_LOGS(INFO) << "Frame processor added; type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
frame_processors_.emplace(frame_type, processor);
}
latch.CountDown();
});
latch.Await();
if (auto frame_processor = GetFrameProcessor(frame_type)) {
NEARBY_LOGS(INFO) << "Frame processor found: updated; type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
frame_processor.set(processor);
} else {
MutexLock lock(&frame_processors_lock_);
NEARBY_LOGS(INFO) << "Frame processor added; type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
frame_processors_.emplace(frame_type, processor);
}
}
void EndpointManager::UnregisterFrameProcessor(
@@ -328,20 +323,9 @@ void EndpointManager::UnregisterFrameProcessor(
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [enter]: processor ="
<< processor;
if (processor == nullptr) return;
CountDownLatch latch(1);
RunOnEndpointManagerThread([this, frame_type, processor, &latch]() {
auto it = frame_processors_.find(frame_type);
if (it == frame_processors_.end()) {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: processor="
<< processor;
latch.CountDown();
return;
}
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [found]: processor="
<< processor;
LockedFrameProcessor frame_processor_lock(&it->second);
if (frame_processor_lock.get() == processor) {
frame_processor_lock.reset();
if (auto frame_processor = GetFrameProcessor(frame_type)) {
if (frame_processor.get() == processor) {
frame_processor.reset();
NEARBY_LOGS(INFO) << "Unregistered: type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
} else {
@@ -349,17 +333,17 @@ void EndpointManager::UnregisterFrameProcessor(
INFO,
"Failed to unregister: type=%d; processor mismatch: passed=%p, "
"expected=%p",
frame_type, processor, frame_processor_lock.get());
frame_type, processor, frame_processor.get());
}
latch.CountDown();
});
latch.Await();
NEARBY_LOGS(INFO) << "Unregistered [sync done]: type=" << frame_type
<< "; processor=" << processor << "; self=" << this;
} else {
NEARBY_LOGS(INFO) << "UnregisterFrameProcessor [not found]: processor="
<< processor;
}
}
EndpointManager::LockedFrameProcessor EndpointManager::GetFrameProcessor(
V1Frame::FrameType frame_type) {
MutexLock lock(&frame_processors_lock_);
auto it = frame_processors_.find(frame_type);
if (it != frame_processors_.end()) {
return LockedFrameProcessor(&it->second);
@@ -573,9 +557,28 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
void EndpointManager::WaitForEndpointDisconnectionProcessing(
ClientProxy* client, const std::string& endpoint_id) {
NEARBY_LOGS(INFO) << "Wait: client=" << client << "; id=" << endpoint_id;
CountDownLatch barrier =
NotifyFrameProcessorsOnEndpointDisconnect(client, endpoint_id);
NEARBY_LOGS(INFO) << "Waiting for 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;
}
}
CountDownLatch EndpointManager::NotifyFrameProcessorsOnEndpointDisconnect(
ClientProxy* client, const std::string& endpoint_id) {
NEARBY_LOGS(INFO) << "NotifyFrameProcessorsOnEndpointDisconnect: client="
<< client << "; id=" << endpoint_id;
MutexLock lock(&frame_processors_lock_);
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;
@@ -593,21 +596,10 @@ void EndpointManager::WaitForEndpointDisconnectionProcessing(
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;
}
return barrier;
}
std::vector<std::string> EndpointManager::SendTransferFrameBytes(
cpp/core/internal/endpoint_manager.h
+6 -1
View File
@@ -30,6 +30,7 @@
#include "platform/public/single_thread_executor.h"
#include "platform/public/system_clock.h"
#include "proto/connections_enums.pb.h"
#include "absl/base/thread_annotations.h"
#include "absl/container/flat_hash_map.h"
#include "absl/container/flat_hash_set.h"
#include "absl/time/time.h"
@@ -212,6 +213,9 @@ class EndpointManager {
void WaitForEndpointDisconnectionProcessing(ClientProxy* client,
const std::string& endpoint_id);
CountDownLatch NotifyFrameProcessorsOnEndpointDisconnect(
ClientProxy* client, const std::string& endpoint_id);
std::vector<std::string> SendTransferFrameBytes(
const std::vector<std::string>& endpoint_ids,
const ByteArray& payload_transfer_frame_bytes, std::int64_t payload_id,
@@ -234,8 +238,9 @@ class EndpointManager {
EndpointChannelManager* channel_manager_;
RecursiveMutex frame_processors_lock_;
absl::flat_hash_map<V1Frame::FrameType, FrameProcessorWithMutex>
frame_processors_;
frame_processors_ ABSL_GUARDED_BY(frame_processors_lock_);
// We keep track of all registered channel endpoints here.
absl::flat_hash_map<std::string, EndpointState> endpoints_;