From ae7c51e95d02a6102fec9506ddf30815406be798 Mon Sep 17 00:00:00 2001 From: Janusz Sobczak Date: Mon, 21 Jun 2021 03:53:33 -0700 Subject: [PATCH] Run workers on dedicated threads. 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 --- cpp/core/internal/endpoint_manager.cc | 137 +++++++++----------------- cpp/core/internal/endpoint_manager.h | 50 +++++----- cpp/platform/base/feature_flags.h | 3 - 3 files changed, 75 insertions(+), 115 deletions(-) diff --git a/cpp/core/internal/endpoint_manager.cc b/cpp/core/internal/endpoint_manager.cc index 1c0153cd..1079cec0 100644 --- a/cpp/core/internal/endpoint_manager.cc +++ b/cpp/core/internal/endpoint_manager.cc @@ -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 barrier, + const std::string& endpoint_id, std::function(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(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(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(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 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 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 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, diff --git a/cpp/core/internal/endpoint_manager.h b/cpp/core/internal/endpoint_manager.h index 8105c23e..c33f1956 100644 --- a/cpp/core/internal/endpoint_manager.h +++ b/cpp/core/internal/endpoint_manager.h @@ -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 barrier = - std::make_shared(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 barrier, + const std::string& endpoint_id, std::function(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 endpoints_; - MultiThreadExecutor keep_alive_executor_{kMaxConcurrentEndpoints}; - MultiThreadExecutor handlers_executor_{kMaxConcurrentEndpoints}; SingleThreadExecutor serial_executor_; }; diff --git a/cpp/platform/base/feature_flags.h b/cpp/platform/base/feature_flags.h index a745a912..914f4273 100644 --- a/cpp/platform/base/feature_flags.h +++ b/cpp/platform/base/feature_flags.h @@ -29,9 +29,6 @@ class FeatureFlags { struct Flags { bool enable_cancellation_flag = false; bool enable_async_bandwidth_upgrade = true; - // Let endpoint_manager erase deleted endpoint from endpoints_ inside - // function RemoveEndpoint. - bool endpoint_manager_ensure_workers_terminated_inside_remove = true; // If a scheduled runnable is already running, Cancel() will synchronously // wait for the task to complete. bool cancel_waits_for_running_tasks = true;