WiP admin: streaming response variant that uses lifecycle callbacks

source/exe/admin_response.cc

@@ -8,15 +8,24 @@
 namespace Envoy {
 
 AdminResponse::AdminResponse(Server::Instance& server, absl::string_view path,
-                             absl::string_view method, SharedPtrSet response_set)
-    : server_(server), opt_admin_(server.admin()), shared_response_set_(response_set) {
-  request_headers_->setMethod(method);
-  request_headers_->setPath(path);
+                             absl::string_view method /*, SharedPtrSet response_set*/)
+    : server_(server), opt_admin_(server.admin()),
+      // shared_response_set_(response_set)
+      lifecycle_notifier_(server.lifecycleNotifier().registerCallback(
+          Server::ServerLifecycleNotifier::Stage::ShutdownExit, [this] { terminate(); })) {
+  if (lifecycle_notifier_ == nullptr) {
+    terminate();
+  } else {
+    ENVOY_LOG_MISC(error, "AdminResponse(): {}", static_cast<void*>(this));
+    request_headers_->setMethod(method);
+    request_headers_->setPath(path);
+  }
 }
 
 AdminResponse::~AdminResponse() {
+  ENVOY_LOG_MISC(error, "~AdminResponse(): {}", static_cast<void*>(this));
   cancel();
-  shared_response_set_->detachResponse(this);
+  // shared_response_set_->detachResponse(this);
 }
 
 void AdminResponse::getHeaders(HeadersFn fn) {
@@ -87,10 +96,12 @@ bool AdminResponse::cancelled() const {
 // admin response, and so calls to getHeader and nextChunk remain valid,
 // resulting in 503 and an empty body.
 void AdminResponse::terminate() {
+  ENVOY_LOG_MISC(error, "terminate(): {}", static_cast<void*>(this));
   ASSERT_IS_MAIN_OR_TEST_THREAD();
   absl::MutexLock lock(&mutex_);
   if (!terminated_) {
     terminated_ = true;
+    lifecycle_notifier_.reset();
     sendErrorLockHeld();
     sendAbortChunkLockHeld();
   }
@@ -159,7 +170,7 @@ void AdminResponse::sendErrorLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_) {
   }
 }
 
-void AdminResponse::PtrSet::terminateAdminRequests() {
+/*void AdminResponse::PtrSet::terminateAdminRequests() {
   ASSERT_IS_MAIN_OR_TEST_THREAD();
 
   absl::MutexLock lock(&mutex_);
@@ -186,6 +197,6 @@ void AdminResponse::PtrSet::attachResponse(AdminResponse* response) {
 void AdminResponse::PtrSet::detachResponse(AdminResponse* response) {
   absl::MutexLock lock(&mutex_);
   response_set_.erase(response);
-}
+  }*/
 
 } // namespace Envoy

source/exe/admin_response.h

@@ -48,6 +48,7 @@ class AdminResponse : public std::enable_shared_from_this<AdminResponse> {
   // In summary:
   //  * MainCommonBase can outlive AdminResponse so we need detachResponse.
   //  * AdminResponse can outlive MainCommonBase, so we need shared_ptr.
+#if 0
   class PtrSet {
   public:
     /**
@@ -80,9 +81,10 @@ class AdminResponse : public std::enable_shared_from_this<AdminResponse> {
     bool accepting_admin_requests_ ABSL_GUARDED_BY(mutex_) = true;
   };
   using SharedPtrSet = std::shared_ptr<PtrSet>;
+#endif
 
-  AdminResponse(Server::Instance& server, absl::string_view path, absl::string_view method,
-                SharedPtrSet response_set);
+  AdminResponse(Server::Instance& server, absl::string_view path, absl::string_view method
+                /*, SharedPtrSet response_set*/);
   ~AdminResponse();
 
   /**
@@ -150,6 +152,7 @@ class AdminResponse : public std::enable_shared_from_this<AdminResponse> {
   void requestNextChunk();
   void sendAbortChunkLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
   void sendErrorLockHeld() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_);
+  // void clearLifecycleTerminator();
 
   Server::Instance& server_;
   OptRef<Server::Admin> opt_admin_;
@@ -181,7 +184,8 @@ class AdminResponse : public std::enable_shared_from_this<AdminResponse> {
   BodyFn body_fn_ ABSL_GUARDED_BY(mutex_);
   mutable absl::Mutex mutex_;
 
-  SharedPtrSet shared_response_set_;
+  // SharedPtrSet shared_response_set_;
+  Server::ServerLifecycleNotifier::HandlePtr lifecycle_notifier_ ABSL_GUARDED_BY(mutex_);
 };
 using AdminResponseSharedPtr = std::shared_ptr<AdminResponse>;
 

source/exe/main_common.cc

@@ -59,8 +59,8 @@ MainCommonBase::MainCommonBase(const Server::Options& options, Event::TimeSystem
                        std::move(platform_impl), std::move(random_generator),
                        std::move(process_context), createFunction())
 #ifdef ENVOY_ADMIN_FUNCTIONALITY
-      ,
-      shared_response_set_(std::make_shared<AdminResponse::PtrSet>())
+//,
+// shared_response_set_(std::make_shared<AdminResponse::PtrSet>())
 #endif
 {
 }
@@ -74,7 +74,7 @@ bool MainCommonBase::run() {
   case Server::Mode::Serve:
     runServer();
 #ifdef ENVOY_ADMIN_FUNCTIONALITY
-    shared_response_set_->terminateAdminRequests();
+    // shared_response_set_->terminateAdminRequests();
 #endif
     ret = true;
     break;
@@ -113,10 +113,7 @@ void MainCommonBase::adminRequest(absl::string_view path_and_query, absl::string
 
 AdminResponseSharedPtr MainCommonBase::adminRequest(absl::string_view path_and_query,
                                                     absl::string_view method) {
-  auto response =
-      std::make_shared<AdminResponse>(*server(), path_and_query, method, shared_response_set_);
-  shared_response_set_->attachResponse(response.get());
-  return response;
+  return std::make_shared<AdminResponse>(*server(), path_and_query, method);
 }
 #endif
 

source/exe/main_common.h

@@ -78,8 +78,10 @@ class MainCommonBase : public StrippedMainBase {
    */
   AdminResponseSharedPtr adminRequest(absl::string_view path_and_query, absl::string_view method);
 
+  using LifecycleHandleSharedPtr = std::shared_ptr<Server::ServerLifecycleNotifier::HandlePtr>;
+
 private:
-  AdminResponse::SharedPtrSet shared_response_set_;
+  // AdminResponse::SharedPtrSet shared_response_set_;
 #endif
 };
 

source/server/server.cc

@@ -85,7 +85,7 @@ InstanceBase::InstanceBase(Init::Manager& init_manager, const Options& options,
                            Filesystem::Instance& file_system,
                            std::unique_ptr<ProcessContext> process_context,
                            Buffer::WatermarkFactorySharedPtr watermark_factory)
-    : init_manager_(init_manager), live_(false), options_(options),
+    : init_manager_(init_manager), options_(options),
       validation_context_(options_.allowUnknownStaticFields(),
                           !options.rejectUnknownDynamicFields(),
                           options.ignoreUnknownDynamicFields()),
@@ -1054,7 +1054,10 @@ Runtime::Loader& InstanceBase::runtime() { return *runtime_; }
 
 void InstanceBase::shutdown() {
   ENVOY_LOG(info, "shutting down server instance");
-  shutdown_ = true;
+  {
+    absl::MutexLock lock(&stage_callbacks_mutex_);
+    shutdown_ = true;
+  }
   restarter_.sendParentTerminateRequest();
   notifyCallbacksForStage(Stage::ShutdownExit, [this] { dispatcher_->exit(); });
 }
@@ -1074,26 +1077,39 @@ void InstanceBase::shutdownAdmin() {
 
 ServerLifecycleNotifier::HandlePtr InstanceBase::registerCallback(Stage stage,
                                                                   StageCallback callback) {
+  absl::MutexLock lock(&stage_callbacks_mutex_);
+  if (shutdown_) {
+    return nullptr;
+  }
   auto& callbacks = stage_callbacks_[stage];
-  return std::make_unique<LifecycleCallbackHandle<StageCallback>>(callbacks, callback);
+  return std::make_unique<LifecycleCallbackHandle<StageCallback>>(callbacks, callback,
+                                                                  stage_callbacks_mutex_);
 }
 
 ServerLifecycleNotifier::HandlePtr
 InstanceBase::registerCallback(Stage stage, StageCallbackWithCompletion callback) {
   ASSERT(stage == Stage::ShutdownExit);
   auto& callbacks = stage_completable_callbacks_[stage];
-  return std::make_unique<LifecycleCallbackHandle<StageCallbackWithCompletion>>(callbacks,
-                                                                                callback);
+  return std::make_unique<LifecycleCallbackHandle<StageCallbackWithCompletion>>(
+      callbacks, callback, stage_callbacks_mutex_);
 }
 
 void InstanceBase::notifyCallbacksForStage(Stage stage, std::function<void()> completion_cb) {
   ASSERT_IS_MAIN_OR_TEST_THREAD();
-  const auto it = stage_callbacks_.find(stage);
-  if (it != stage_callbacks_.end()) {
-    for (const StageCallback& callback : it->second) {
-      callback();
+
+  std::vector<StageCallback> callbacks_to_run;
+  {
+    absl::MutexLock lock(&stage_callbacks_mutex_);
+    const auto it = stage_callbacks_.find(stage);
+    if (it != stage_callbacks_.end()) {
+      LifecycleNotifierCallbacks& callbacks = it->second;
+      callbacks_to_run.insert(callbacks_to_run.end(), callbacks.begin(), callbacks.end());
     }
   }
+  for (StageCallback callback : callbacks_to_run) {
+    callback();
+  }
+  callbacks_to_run.clear();
 
   // Wrap completion_cb so that it only gets invoked when all callbacks for this stage
   // have finished their work.

source/server/server.h

@@ -336,6 +336,7 @@ class InstanceBase : Logger::Loggable<Logger::Id::main>,
   void terminate();
   void notifyCallbacksForStage(
       Stage stage, std::function<void()> completion_cb = [] {});
+  void notifyCallbacksForStage(Stage stage, std::function<void()> completion_cb = [] {});
   void onRuntimeReady();
   void onClusterManagerPrimaryInitializationComplete();
 
@@ -353,8 +354,8 @@ class InstanceBase : Logger::Loggable<Logger::Id::main>,
   // - There may be active connections referencing it.
   std::unique_ptr<Secret::SecretManager> secret_manager_;
   bool workers_started_{false};
-  std::atomic<bool> live_;
-  bool shutdown_{false};
+  std::atomic<bool> live_{false};
+  std::atomic<bool> shutdown_{false};
   const Options& options_;
   ProtobufMessage::ProdValidationContextImpl validation_context_;
   TimeSource& time_source_;
@@ -385,7 +386,9 @@ class InstanceBase : Logger::Loggable<Logger::Id::main>,
   std::unique_ptr<Runtime::Loader> runtime_;
   ProdWorkerFactory worker_factory_;
   std::unique_ptr<ListenerManager> listener_manager_;
-  absl::node_hash_map<Stage, LifecycleNotifierCallbacks> stage_callbacks_;
+  absl::node_hash_map<Stage, LifecycleNotifierCallbacks>
+      stage_callbacks_ ABSL_GUARDED_BY(stage_callbacks_mutex_);
+  absl::Mutex stage_callbacks_mutex_;
   absl::node_hash_map<Stage, LifecycleNotifierCompletionCallbacks> stage_completable_callbacks_;
   Configuration::MainImpl config_;
   Network::DnsResolverSharedPtr dns_resolver_;
@@ -419,11 +422,22 @@ class InstanceBase : Logger::Loggable<Logger::Id::main>,
 
   bool stats_flush_in_progress_ : 1;
 
-  template <class T>
-  class LifecycleCallbackHandle : public ServerLifecycleNotifier::Handle, RaiiListElement<T> {
+  template <class T> class LifecycleCallbackHandle : public ServerLifecycleNotifier::Handle {
   public:
-    LifecycleCallbackHandle(std::list<T>& callbacks, T& callback)
-        : RaiiListElement<T>(callbacks, callback) {}
+    LifecycleCallbackHandle(std::list<T>& callbacks, T& callback, absl::Mutex& mutex)
+        : mutex_(mutex) {
+      // absl::MutexLock lock(&mutex_);
+      rail_list_element_ = std::make_unique<RaiiListElement<T>>(callbacks, callback);
+    }
+
+    ~LifecycleCallbackHandle() {
+      absl::MutexLock lock(&mutex_);
+      rail_list_element_.reset();
+    }
+
+  private:
+    absl::Mutex& mutex_;
+    std::unique_ptr<RaiiListElement<T>> rail_list_element_;
   };
 
 #ifdef ENVOY_PERFETTO