CLI - cancel execution upon signal reception

source/client/process_impl.cc

@@ -123,11 +123,15 @@ void ProcessImpl::shutdown() {
   // Before we shut down the worker threads, stop threading.
   tls_.shutdownGlobalThreading();
   store_root_.shutdownThreading();
-  // Before shutting down the cluster manager, stop the workers.
-  for (auto& worker : workers_) {
-    worker->shutdown();
+
+  {
+    auto guard = std::make_unique<Envoy::Thread::LockGuard>(workers_lock_);
+    // Before shutting down the cluster manager, stop the workers.
+    for (auto& worker : workers_) {
+      worker->shutdown();
+    }
+    workers_.clear();
   }
-  workers_.clear();
   if (cluster_manager_ != nullptr) {
     cluster_manager_->shutdown();
   }
@@ -137,13 +141,15 @@ void ProcessImpl::shutdown() {
 
 bool ProcessImpl::requestExecutionCancellation() {
   ENVOY_LOG(debug, "Requesting workers to cancel execution");
+  auto guard = std::make_unique<Envoy::Thread::LockGuard>(workers_lock_);
   for (auto& worker : workers_) {
     worker->requestExecutionCancellation();
   }
+  cancelled_ = true;
   return true;
 }
 
-const std::vector<ClientWorkerPtr>& ProcessImpl::createWorkers(const uint32_t concurrency) {
+void ProcessImpl::createWorkers(const uint32_t concurrency) {
   // TODO(oschaaf): Expose kMinimalDelay in configuration.
   const std::chrono::milliseconds kMinimalWorkerDelay = 500ms + (concurrency * 50ms);
   ASSERT(workers_.empty());
@@ -173,7 +179,6 @@ const std::vector<ClientWorkerPtr>& ProcessImpl::createWorkers(const uint32_t co
                                 : ClientWorkerImpl::HardCodedWarmupStyle::OFF));
     worker_number++;
   }
-  return workers_;
 }
 
 void ProcessImpl::configureComponentLogLevels(spdlog::level::level_enum level) {
@@ -385,44 +390,50 @@ void ProcessImpl::addRequestSourceCluster(
 
 bool ProcessImpl::runInternal(OutputCollector& collector, const std::vector<UriPtr>& uris,
                               const UriPtr& request_source_uri, const UriPtr& tracing_uri) {
-  int number_of_workers = determineConcurrency();
-  shutdown_ = false;
-  const std::vector<ClientWorkerPtr>& workers = createWorkers(number_of_workers);
-  tls_.registerThread(*dispatcher_, true);
-  store_root_.initializeThreading(*dispatcher_, tls_);
-  runtime_singleton_ = std::make_unique<Envoy::Runtime::ScopedLoaderSingleton>(
-      Envoy::Runtime::LoaderPtr{new Envoy::Runtime::LoaderImpl(
-          *dispatcher_, tls_, {}, *local_info_, store_root_, generator_,
-          Envoy::ProtobufMessage::getStrictValidationVisitor(), *api_)});
-  ssl_context_manager_ =
-      std::make_unique<Extensions::TransportSockets::Tls::ContextManagerImpl>(time_system_);
-  cluster_manager_factory_ = std::make_unique<ClusterManagerFactory>(
-      admin_, Envoy::Runtime::LoaderSingleton::get(), store_root_, tls_, generator_,
-      dispatcher_->createDnsResolver({}, false), *ssl_context_manager_, *dispatcher_, *local_info_,
-      secret_manager_, validation_context_, *api_, http_context_, grpc_context_,
-      access_log_manager_, *singleton_manager_);
-  cluster_manager_factory_->setConnectionReuseStrategy(
-      options_.h1ConnectionReuseStrategy() == nighthawk::client::H1ConnectionReuseStrategy::LRU
-          ? Http1PoolImpl::ConnectionReuseStrategy::LRU
-          : Http1PoolImpl::ConnectionReuseStrategy::MRU);
-  cluster_manager_factory_->setPrefetchConnections(options_.prefetchConnections());
-  envoy::config::bootstrap::v3::Bootstrap bootstrap;
-  createBootstrapConfiguration(bootstrap, uris, request_source_uri, number_of_workers);
-  if (tracing_uri != nullptr) {
-    setupTracingImplementation(bootstrap, *tracing_uri);
-    addTracingCluster(bootstrap, *tracing_uri);
-  }
-  ENVOY_LOG(debug, "Computed configuration: {}", bootstrap.DebugString());
-  cluster_manager_ = cluster_manager_factory_->clusterManagerFromProto(bootstrap);
-  maybeCreateTracingDriver(bootstrap.tracing());
-  cluster_manager_->setInitializedCb([this]() -> void { init_manager_.initialize(init_watcher_); });
+  {
+    auto guard = std::make_unique<Envoy::Thread::LockGuard>(workers_lock_);
+    if (cancelled_) {
+      return true;
+    }
+    int number_of_workers = determineConcurrency();
+    shutdown_ = false;
+    createWorkers(number_of_workers);
+    tls_.registerThread(*dispatcher_, true);
+    store_root_.initializeThreading(*dispatcher_, tls_);
+    runtime_singleton_ = std::make_unique<Envoy::Runtime::ScopedLoaderSingleton>(
+        Envoy::Runtime::LoaderPtr{new Envoy::Runtime::LoaderImpl(
+            *dispatcher_, tls_, {}, *local_info_, store_root_, generator_,
+            Envoy::ProtobufMessage::getStrictValidationVisitor(), *api_)});
+    ssl_context_manager_ =
+        std::make_unique<Extensions::TransportSockets::Tls::ContextManagerImpl>(time_system_);
+    cluster_manager_factory_ = std::make_unique<ClusterManagerFactory>(
+        admin_, Envoy::Runtime::LoaderSingleton::get(), store_root_, tls_, generator_,
+        dispatcher_->createDnsResolver({}, false), *ssl_context_manager_, *dispatcher_,
+        *local_info_, secret_manager_, validation_context_, *api_, http_context_, grpc_context_,
+        access_log_manager_, *singleton_manager_);
+    cluster_manager_factory_->setConnectionReuseStrategy(
+        options_.h1ConnectionReuseStrategy() == nighthawk::client::H1ConnectionReuseStrategy::LRU
+            ? Http1PoolImpl::ConnectionReuseStrategy::LRU
+            : Http1PoolImpl::ConnectionReuseStrategy::MRU);
+    cluster_manager_factory_->setPrefetchConnections(options_.prefetchConnections());
+    envoy::config::bootstrap::v3::Bootstrap bootstrap;
+    createBootstrapConfiguration(bootstrap, uris, request_source_uri, number_of_workers);
+    if (tracing_uri != nullptr) {
+      setupTracingImplementation(bootstrap, *tracing_uri);
+      addTracingCluster(bootstrap, *tracing_uri);
+    }
+    ENVOY_LOG(debug, "Computed configuration: {}", bootstrap.DebugString());
+    cluster_manager_ = cluster_manager_factory_->clusterManagerFromProto(bootstrap);
+    maybeCreateTracingDriver(bootstrap.tracing());
+    cluster_manager_->setInitializedCb(
+        [this]() -> void { init_manager_.initialize(init_watcher_); });
 
-  Runtime::LoaderSingleton::get().initialize(*cluster_manager_);
+    Runtime::LoaderSingleton::get().initialize(*cluster_manager_);
 
-  for (auto& w : workers_) {
-    w->start();
+    for (auto& w : workers_) {
+      w->start();
+    }
   }
-
   for (auto& w : workers_) {
     w->waitForCompletion();
   }
@@ -451,7 +462,7 @@ bool ProcessImpl::runInternal(OutputCollector& collector, const std::vector<UriP
   const auto& counters = Utility().mapCountersFromStore(
       store_root_, [](absl::string_view, uint64_t value) { return value > 0; });
   StatisticFactoryImpl statistic_factory(options_);
-  collector.addResult("global", mergeWorkerStatistics(workers), counters,
+  collector.addResult("global", mergeWorkerStatistics(workers_), counters,
                       total_execution_duration / workers_.size());
   return counters.find("sequencer.failed_terminations") == counters.end();
 }

source/client/process_impl.h

@@ -90,7 +90,7 @@ class ProcessImpl : public Process, public Envoy::Logger::Loggable<Envoy::Logger
   void maybeCreateTracingDriver(const envoy::config::trace::v3::Tracing& configuration);
 
   void configureComponentLogLevels(spdlog::level::level_enum level);
-  const std::vector<ClientWorkerPtr>& createWorkers(const uint32_t concurrency);
+  void createWorkers(const uint32_t concurrency);
   std::vector<StatisticPtr> vectorizeStatisticPtrMap(const StatisticPtrMap& statistics) const;
   std::vector<StatisticPtr>
   mergeWorkerStatistics(const std::vector<ClientWorkerPtr>& workers) const;
@@ -136,6 +136,8 @@ class ProcessImpl : public Process, public Envoy::Logger::Loggable<Envoy::Logger
   Envoy::Server::ValidationAdmin admin_;
   Envoy::ProtobufMessage::ProdValidationContextImpl validation_context_;
   bool shutdown_{true};
+  Envoy::Thread::MutexBasicLockable workers_lock_;
+  bool cancelled_{false};
 };
 
 } // namespace Client

source/client/remote_process_impl.cc

@@ -56,6 +56,7 @@ bool RemoteProcessImpl::run(OutputCollector& collector) {
 
 bool RemoteProcessImpl::requestExecutionCancellation() {
   ENVOY_LOG(error, "Remote process cancellation not supported yet");
+  // TODO(XXX): Send a cancel request to the gRPC service.
   return false;
 }
 

test/integration/test_integration_basics.py

@@ -659,7 +659,9 @@ def test_http_request_release_timing(http_test_server_fixture, qps_parameterizat
 
 
 def send_sigterm(p):
-  time.sleep(2)
+  # Sleep for a while, under tsan the client needs a lot of time
+  # to start up.
+  time.sleep(10)
   p.terminate()
 
 

test/process_test.cc

@@ -1,3 +1,4 @@
+#include <thread>
 #include <vector>
 
 #include "nighthawk/common/exception.h"
@@ -32,12 +33,35 @@ class ProcessTest : public TestWithParam<Envoy::Network::Address::IpVersion> {
       : loopback_address_(Envoy::Network::Test::getLoopbackAddressUrlString(GetParam())),
         options_(TestUtility::createOptionsImpl(
             fmt::format("foo --duration 1 -v error --rps 10 https://{}/", loopback_address_))){};
-  void runProcess(RunExpectation expectation) {
+  void runProcess(RunExpectation expectation, bool do_cancel = false) {
     ProcessPtr process = std::make_unique<ProcessImpl>(*options_, time_system_);
     OutputCollectorImpl collector(time_system_, *options_);
+    std::thread cancel_thread;
+    if (do_cancel) {
+      cancel_thread = std::thread([&process] {
+        sleep(5);
+        std::cerr << "request cancel" << std::endl;
+        process->requestExecutionCancellation();
+      });
+    }
     const auto result =
         process->run(collector) ? RunExpectation::EXPECT_SUCCESS : RunExpectation::EXPECT_FAILURE;
     EXPECT_EQ(result, expectation);
+    if (cancel_thread.joinable()) {
+      cancel_thread.join();
+    }
+    if (do_cancel) {
+      auto proto = collector.toProto();
+      int cancel_requests = 0;
+      for (const auto& result : proto.results()) {
+        for (const auto& counter : result.counters()) {
+          if (counter.name() == "cancel_requests") {
+            cancel_requests++;
+          }
+        }
+      }
+      EXPECT_EQ(3, cancel_requests); // global results + two workers
+    }
     process->shutdown();
   }
 
@@ -64,5 +88,12 @@ TEST_P(ProcessTest, BadTracerSpec) {
   runProcess(RunExpectation::EXPECT_FAILURE);
 }
 
+TEST_P(ProcessTest, CancelExecution) {
+  options_ = TestUtility::createOptionsImpl(
+      fmt::format("foo --duration 10 --failure-predicate foo:0 --concurrency 2 https://{}/",
+                  loopback_address_));
+  runProcess(RunExpectation::EXPECT_SUCCESS, true);
+}
+
 } // namespace Client
 } // namespace Nighthawk