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

Run workers on dedicated threads.

Browse Source 
Run reader and keep-alive workers on dedicated threads provided by
SingleThreadExecutor.
The thread handlers are stored in EndpointState, the workers are automatically
terminated when EndpointState is destroyed.

PiperOrigin-RevId: 380543888
This commit is contained in:
Janusz Sobczak
2021-06-21 03:54:01 -07:00
committed by Copybara-Service
parent 216a6d8ae0
commit ae7c51e95d
3 changed files with 75 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cpp/core/internal/endpoint_manager.cc
+49 -88
View File
@@ -76,15 +76,14 @@ class EndpointManager::LockedFrameProcessor {
// EndpointChannel until an Exception::IO is thrown. Once an
// Exception::IO occurs, a check will be performed to see if another
// EndpointChannel is available for the given endpoint and, if so,
// handler(EndpointChannel) will be called again. Return false to exit
// the loop.
// handler(EndpointChannel) will be called again.
void EndpointManager::EndpointChannelLoopRunnable(
const std::string& runnable_name, ClientProxy* client,
const std::string& endpoint_id, std::weak_ptr<CountDownLatch> barrier,
const std::string& endpoint_id,
std::function<ExceptionOr<bool>(EndpointChannel*)> handler) {
// EndpointChannelManager will not let multiple channels exist simultaneously
// for the same endpoint_id; it will be closing "old" channels as new ones
// come. (There will be a short overlap).
// come.
// Closed channel will return Exception::kIo for any Read, and loop (below)
// will retry and attempt to pick another channel.
// If channel is deleted (no mapping), or it is still the same channel
@@ -150,14 +149,6 @@ void EndpointManager::EndpointChannelLoopRunnable(
// if needed.
NEARBY_LOGS(INFO) << "Worker going down; worker name=" << runnable_name
<< "; endpoint_id=" << endpoint_id;
if (auto latch = barrier.lock()) {
latch->CountDown();
} else {
NEARBY_LOG(WARNING,
"Barrier already expired in worker name=%s, for endpoint %s",
runnable_name.c_str(), endpoint_id.c_str());
}
// Always clear out all state related to this endpoint before terminating
// this thread.
DiscardEndpoint(client, endpoint_id);
@@ -276,30 +267,11 @@ EndpointManager::~EndpointManager() {
CountDownLatch latch(1);
RunOnEndpointManagerThread("bring-down-endpoints", [this, &latch]() {
NEARBY_LOG(INFO, "Bringing down endpoints");
for (auto& item : endpoints_) {
const std::string& endpoint_id = item.first;
EndpointState& state = item.second;
// This will close the channel; all workers will sense that and
// terminate.
channel_manager_->UnregisterChannelForEndpoint(endpoint_id);
if (state.barrier) {
state.barrier->Await();
} else {
NEARBY_LOGS(WARNING)
<< "State barrier already freed before EM destructor for endpoint"
<< endpoint_id;
}
}
endpoints_.clear();
latch.CountDown();
});
latch.Await();
NEARBY_LOG(INFO, "Bringing down worker threads");
// Stop all the ongoing Runnables (as gracefully as possible).
// Order matters: bring worker pools down first; serial_executor_ thread
// should go last, since workers schedule jobs there even during shutdown.
handlers_executor_.Shutdown();
keep_alive_executor_.Shutdown();
NEARBY_LOG(INFO, "Bringing down control thread");
serial_executor_.Shutdown();
NEARBY_LOG(INFO, "EndpointManager is down");
@@ -361,26 +333,16 @@ EndpointManager::LockedFrameProcessor EndpointManager::GetFrameProcessor(
return LockedFrameProcessor();
}
void EndpointManager::EnsureWorkersTerminated(const std::string& endpoint_id) {
NEARBY_LOGS(ERROR) << "EnsureWorkersTerminated for endpoint " << endpoint_id;
void EndpointManager::RemoveEndpointState(const std::string& endpoint_id) {
NEARBY_LOGS(VERBOSE) << "EnsureWorkersTerminated for endpoint "
<< endpoint_id;
auto item = endpoints_.find(endpoint_id);
if (item != endpoints_.end()) {
NEARBY_LOGS(INFO) << "EndpointState found for endpoint " << endpoint_id;
// If another instance of data and keep-alive handlers is running, it will
// terminate soon; we should block until it happens.
EndpointState& endpoint_state = item->second;
NEARBY_LOGS(INFO) << "Waiting for workers to terminate for endpoint "
<< endpoint_id;
if (endpoint_state.barrier) {
endpoint_state.barrier->Await();
} else {
NEARBY_LOGS(WARNING)
<< "State barrier already freed before EnsureWorkersTerminated for "
"endpoint "
<< endpoint_id;
}
// terminate soon. Removing EndpointState waits for workers to complete.
endpoints_.erase(item);
NEARBY_LOGS(INFO) << "Workers terminated for endpoint " << endpoint_id;
NEARBY_LOGS(VERBOSE) << "Workers terminated for endpoint " << endpoint_id;
} else {
NEARBY_LOGS(INFO) << "EndpointState not found for endpoint " << endpoint_id;
}
@@ -406,12 +368,10 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
&options, &listener,
&latch]() {
if (endpoints_.contains(endpoint_id)) {
NEARBY_LOGS(WARNING) << "Registing duplicate endpoint " << endpoint_id;
if (!FeatureFlags::GetInstance()
.GetFlags()
.endpoint_manager_ensure_workers_terminated_inside_remove) {
EnsureWorkersTerminated(endpoint_id);
}
NEARBY_LOGS(WARNING) << "Registering duplicate endpoint " << endpoint_id;
// We must remove old endpoint state before registering a new one for the
// same endpoint_id.
RemoveEndpointState(endpoint_id);
}
absl::Duration keep_alive_interval =
@@ -432,32 +392,28 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
client, endpoint_id, std::unique_ptr<EndpointChannel>(channel));
EndpointState& endpoint_state =
endpoints_.emplace(endpoint_id, EndpointState()).first->second;
endpoint_state.client = client;
endpoints_
.emplace(endpoint_id, EndpointState(endpoint_id, channel_manager_))
.first->second;
NEARBY_LOGS(INFO) << "Starting workers: endpoint " << endpoint_id;
// For every endpoint, there's normally only one Read handler instance
// running on the handlers_executor_ pool. This instance reads data from the
// running on a dedicated thread. This instance reads data from the
// endpoint and delegates incoming frames to various FrameProcessors.
// Once the frame has been properly handled, it starts reading again for
// the next frame. If the handler fails its read and no other
// EndpointChannels are available for this endpoint, a disconnection
// will be initiated.
//
// Using weak_ptr just in case the barrier is freed, to save the UAF crash
// in b/179800119.
StartEndpointReader(
[this, client, endpoint_id,
barrier = std::weak_ptr<CountDownLatch>(endpoint_state.barrier)]() {
EndpointChannelLoopRunnable(
"Read", client, endpoint_id, barrier,
[this, client, endpoint_id](EndpointChannel* channel) {
return HandleData(endpoint_id, client, channel);
});
});
endpoint_state.StartEndpointReader([this, client, endpoint_id]() {
EndpointChannelLoopRunnable(
"Read", client, endpoint_id,
[this, client, endpoint_id](EndpointChannel* channel) {
return HandleData(endpoint_id, client, channel);
});
});
// For every endpoint, there's only one KeepAliveManager instance
// running on the keep_alive_executor_ pool. This instance will
// 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.
@@ -467,16 +423,12 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
// (**) Wifi Hotspots can fail to notice a connection has been lost, and
// they will happily keep writing to /dev/null. This is why we listen
// for the pong.
//
// Using weak_ptr just in case the barrier is freed, to save the UAF crash
// in b/179800119.
NEARBY_LOGS(VERBOSE) << "EndpointManager enabling KeepAlive for endpoint "
<< endpoint_id;
StartEndpointKeepAliveManager(
[this, client, endpoint_id, keep_alive_interval, keep_alive_timeout,
barrier = std::weak_ptr<CountDownLatch>(endpoint_state.barrier)]() {
endpoint_state.StartEndpointKeepAliveManager(
[this, client, endpoint_id, keep_alive_interval, keep_alive_timeout]() {
EndpointChannelLoopRunnable(
"KeepAliveManager", client, endpoint_id, barrier,
"KeepAliveManager", client, endpoint_id,
[this, keep_alive_interval,
keep_alive_timeout](EndpointChannel* channel) {
return HandleKeepAlive(channel, keep_alive_interval,
@@ -496,7 +448,7 @@ void EndpointManager::RegisterEndpoint(ClientProxy* client,
void EndpointManager::UnregisterEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
NEARBY_LOGS(ERROR) << "UnregisterEndpoint for endpoint " << endpoint_id;
NEARBY_LOGS(INFO) << "UnregisterEndpoint for endpoint " << endpoint_id;
CountDownLatch latch(1);
RunOnEndpointManagerThread(
"unregister-endpoint", [this, client, endpoint_id, &latch]() {
@@ -536,7 +488,7 @@ std::vector<std::string> EndpointManager::SendPayloadChunk(
// allow synchronous behavior here it will cause a live lock.
void EndpointManager::DiscardEndpoint(ClientProxy* client,
const std::string& endpoint_id) {
NEARBY_LOGS(ERROR) << "DiscardEndpoint for endpoint " << endpoint_id;
NEARBY_LOGS(VERBOSE) << "DiscardEndpoint for endpoint " << endpoint_id;
RunOnEndpointManagerThread("discard-endpoint", [this, client, endpoint_id]() {
RemoveEndpoint(client, endpoint_id,
/*notify=*/
@@ -561,7 +513,7 @@ std::vector<std::string> EndpointManager::SendControlMessage(
void EndpointManager::RemoveEndpoint(ClientProxy* client,
const std::string& endpoint_id,
bool notify) {
NEARBY_LOGS(ERROR) << "RemoveEndpoint for endpoint " << endpoint_id;
NEARBY_LOGS(INFO) << "RemoveEndpoint for endpoint " << endpoint_id;
// Unregistering from channel_manager_ will also serve to terminate
// the dedicated handler and KeepAlive threads we started when we registered
// this endpoint.
@@ -576,11 +528,7 @@ void EndpointManager::RemoveEndpoint(ClientProxy* client,
client->OnDisconnected(endpoint_id, notify);
NEARBY_LOGS(INFO) << "Removed endpoint for endpoint " << endpoint_id;
}
if (FeatureFlags::GetInstance()
.GetFlags()
.endpoint_manager_ensure_workers_terminated_inside_remove) {
EnsureWorkersTerminated(endpoint_id);
}
RemoveEndpointState(endpoint_id);
}
// @EndpointManagerThread
@@ -667,12 +615,25 @@ std::vector<std::string> EndpointManager::SendTransferFrameBytes(
return failed_endpoint_ids;
}
void EndpointManager::StartEndpointReader(Runnable runnable) {
handlers_executor_.Execute("reader", std::move(runnable));
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) {
NEARBY_LOG(VERBOSE, "EndpointState destructor %s", endpoint_id_.c_str());
channel_manager_->UnregisterChannelForEndpoint(endpoint_id_);
}
}
void EndpointManager::StartEndpointKeepAliveManager(Runnable runnable) {
keep_alive_executor_.Execute("keep-alive", std::move(runnable));
void EndpointManager::EndpointState::StartEndpointReader(Runnable&& runnable) {
reader_thread_.Execute("reader", std::move(runnable));
}
void EndpointManager::EndpointState::StartEndpointKeepAliveManager(
Runnable&& runnable) {
keep_alive_thread_.Execute("keep-alive", std::move(runnable));
}
void EndpointManager::RunOnEndpointManagerThread(const std::string& name,
cpp/core/internal/endpoint_manager.h
+26 -24
View File
@@ -140,13 +140,30 @@ class EndpointManager {
void DiscardEndpoint(ClientProxy* client, const std::string& endpoint_id);
private:
struct EndpointState {
// ClientProxy object associated with this endpoint.
ClientProxy* client;
// Execution barrier, used to ensure that all workers associated with an
// endpoint on handlers_executor_ and keep_alive_executor_ are terminated.
std::shared_ptr<CountDownLatch> barrier =
std::make_shared<CountDownLatch>(2);
class EndpointState {
public:
EndpointState(const std::string& endpoint_id,
EndpointChannelManager* channel_manager)
: endpoint_id_{endpoint_id}, channel_manager_{channel_manager} {}
EndpointState(const EndpointState&) = delete;
// default move constructor would not reset |channel_manager_|
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_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);
private:
const std::string endpoint_id_;
EndpointChannelManager* channel_manager_;
SingleThreadExecutor reader_thread_;
SingleThreadExecutor keep_alive_thread_;
};
// RAII accessor for FrameProcessor
@@ -180,11 +197,11 @@ class EndpointManager {
// Is called from RegisterEndpoint to avoid races; also called from
// RemoveEndpoint as part of proper endpoint shutdown sequence.
// @EndpointManagerThread
void EnsureWorkersTerminated(const std::string& endpoint_id);
void RemoveEndpointState(const std::string& endpoint_id);
void EndpointChannelLoopRunnable(
const std::string& runnable_name, ClientProxy* client_proxy,
const std::string& endpoint_id, std::weak_ptr<CountDownLatch> barrier,
const std::string& endpoint_id,
std::function<ExceptionOr<bool>(EndpointChannel*)> handler);
static void WaitForLatch(const std::string& method_name,
@@ -194,7 +211,6 @@ class EndpointManager {
static constexpr absl::Duration kProcessEndpointDisconnectionTimeout =
absl::Milliseconds(2000);
static constexpr std::int32_t kMaxConcurrentEndpoints = 50;
static constexpr absl::Time kInvalidTimestamp = absl::InfinitePast();
// It should be noted that this method may be called multiple times (because
@@ -217,18 +233,6 @@ class EndpointManager {
const ByteArray& payload_transfer_frame_bytes, std::int64_t payload_id,
std::int64_t offset, const std::string& packet_type);
// Executes data-handing jobs on a separate thread for each endpoint, on a
// handlers_executor_.
// If amount of concurrent connections is less the pool capacity, it is
// possible that while a channel is being replaced, two jobs are trying to
// run for the same endpoint (for a short time).
// TODO (apolyudov): do not let extra job start.
void StartEndpointReader(Runnable runnable);
// Executes keep-alive jobs on a separate thread for each endpoint on a
// keep_alive_executor_.
void StartEndpointKeepAliveManager(Runnable runnable);
// Executes all jobs sequentially, on a serial_executor_.
void RunOnEndpointManagerThread(const std::string& name, Runnable runnable);
@@ -241,8 +245,6 @@ class EndpointManager {
// We keep track of all registered channel endpoints here.
absl::flat_hash_map<std::string, EndpointState> endpoints_;
MultiThreadExecutor keep_alive_executor_{kMaxConcurrentEndpoints};
MultiThreadExecutor handlers_executor_{kMaxConcurrentEndpoints};
SingleThreadExecutor serial_executor_;
};