eds: decrease computational complexity of updates

bazel/test_for_benchmark_wrapper.sh

@@ -1,4 +1,6 @@
 #!/bin/bash
 
-# Set the benchmark time to 0 to just verify that the benchmark runs to completion.
-"${TEST_SRCDIR}/envoy/$@" --benchmark_min_time=0
+# Set the benchmark time to 0 to just verify that the benchmark runs to
+# completion.  We're interacting with two different flag parsers, so the order
+# of flags and the -- matters.
+"${TEST_SRCDIR}/envoy/$@" --skip_expensive_benchmarks -- --benchmark_min_time=0

source/common/upstream/upstream_impl.cc

@@ -1332,9 +1332,7 @@ bool BaseDynamicClusterImpl::updateDynamicHostList(const HostVector& new_hosts,
   bool hosts_changed = false;
 
   // Go through and see if the list we have is different from what we just got. If it is, we make a
-  // new host list and raise a change notification. This uses an N^2 search given that this does not
-  // happen very often and the list sizes should be small (see
-  // https://github.com/envoyproxy/envoy/issues/2874). We also check for duplicates here. It's
+  // new host list and raise a change notification. We also check for duplicates here. It's
   // possible for DNS to return the same address multiple times, and a bad EDS implementation could
   // do the same thing.
 
@@ -1437,16 +1435,20 @@ bool BaseDynamicClusterImpl::updateDynamicHostList(const HostVector& new_hosts,
 
   // Remove hosts from current_priority_hosts that were matched to an existing host in the previous
   // loop.
-  for (auto itr = current_priority_hosts.begin(); itr != current_priority_hosts.end();) {
-    auto existing_itr = existing_hosts_for_current_priority.find((*itr)->address()->asString());
+  auto erase_from =
+      std::remove_if(current_priority_hosts.begin(), current_priority_hosts.end(),
+                     [&existing_hosts_for_current_priority](const HostSharedPtr& p) {
+                       auto existing_itr =
+                           existing_hosts_for_current_priority.find(p->address()->asString());
 
-    if (existing_itr != existing_hosts_for_current_priority.end()) {
-      existing_hosts_for_current_priority.erase(existing_itr);
-      itr = current_priority_hosts.erase(itr);
-    } else {
-      itr++;
-    }
-  }
+                       if (existing_itr != existing_hosts_for_current_priority.end()) {
+                         existing_hosts_for_current_priority.erase(existing_itr);
+                         return true;
+                       }
+
+                       return false;
+                     });
+  current_priority_hosts.erase(erase_from, current_priority_hosts.end());
 
   // If we saw existing hosts during this iteration from a different priority, then we've moved
   // a host from another priority into this one, so we should mark the priority as having changed.
@@ -1464,21 +1466,23 @@ bool BaseDynamicClusterImpl::updateDynamicHostList(const HostVector& new_hosts,
   const bool dont_remove_healthy_hosts =
       health_checker_ != nullptr && !info()->drainConnectionsOnHostRemoval();
   if (!current_priority_hosts.empty() && dont_remove_healthy_hosts) {
-    for (auto i = current_priority_hosts.begin(); i != current_priority_hosts.end();) {
-      if (!((*i)->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC) ||
-            (*i)->healthFlagGet(Host::HealthFlag::FAILED_EDS_HEALTH))) {
-        if ((*i)->weight() > max_host_weight) {
-          max_host_weight = (*i)->weight();
-        }
-
-        final_hosts.push_back(*i);
-        updated_hosts[(*i)->address()->asString()] = *i;
-        (*i)->healthFlagSet(Host::HealthFlag::PENDING_DYNAMIC_REMOVAL);
-        i = current_priority_hosts.erase(i);
-      } else {
-        i++;
-      }
-    }
+    erase_from =
+        std::remove_if(current_priority_hosts.begin(), current_priority_hosts.end(),
+                       [&updated_hosts, &final_hosts, &max_host_weight](const HostSharedPtr& p) {
+                         if (!(p->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC) ||
+                               p->healthFlagGet(Host::HealthFlag::FAILED_EDS_HEALTH))) {
+                           if (p->weight() > max_host_weight) {
+                             max_host_weight = p->weight();
+                           }
+
+                           final_hosts.push_back(p);
+                           updated_hosts[p->address()->asString()] = p;
+                           p->healthFlagSet(Host::HealthFlag::PENDING_DYNAMIC_REMOVAL);
+                           return true;
+                         }
+                         return false;
+                       });
+    current_priority_hosts.erase(erase_from, current_priority_hosts.end());
   }
 
   // At this point we've accounted for all the new hosts as well the hosts that previously

test/benchmark/BUILD

@@ -11,8 +11,10 @@ envoy_package()
 envoy_cc_test_library(
     name = "main",
     srcs = ["main.cc"],
+    hdrs = ["main.h"],
     external_deps = [
         "benchmark",
+        "tclap",
     ],
     deps = [
         "//test/test_common:environment_lib",

test/benchmark/main.cc

@@ -1,16 +1,40 @@
 // NOLINT(namespace-envoy)
 // This is an Envoy driver for benchmarks.
+#include "test/benchmark/main.h"
+
 #include "test/test_common/environment.h"
 
 #include "benchmark/benchmark.h"
+#include "tclap/CmdLine.h"
+
+static bool skip_expensive_benchmarks = false;
 
-// Boilerplate main(), which discovers benchmarks and runs them.
+// Boilerplate main(), which discovers benchmarks and runs them. This uses two
+// different flag parsers, so the order of flags matters: flags defined here
+// must be passed first, and flags defined in benchmark::Initialize second,
+// separated by --.
+// TODO(pgenera): convert this to abseil/flags/ when benchmark also adopts abseil.
 int main(int argc, char** argv) {
   Envoy::TestEnvironment::initializeTestMain(argv[0]);
 
-  benchmark::Initialize(&argc, argv);
-  if (benchmark::ReportUnrecognizedArguments(argc, argv)) {
-    return 1;
+  // NOLINTNEXTLINE(clang-analyzer-optin.cplusplus.VirtualCall)
+  TCLAP::CmdLine cmd("envoy-benchmark-test", ' ', "0.1");
+  TCLAP::SwitchArg skip_switch("s", "skip_expensive_benchmarks",
+                               "skip or minimize expensive benchmarks", cmd, false);
+
+  cmd.setExceptionHandling(false);
+  try {
+    cmd.parse(argc, argv);
+  } catch (const TCLAP::ExitException& e) {
+    // parse() throws an ExitException with status 0 after printing the output
+    // for --help and --version.
+    return 0;
   }
+
+  skip_expensive_benchmarks = skip_switch.getValue();
+
+  benchmark::Initialize(&argc, argv);
   benchmark::RunSpecifiedBenchmarks();
 }
+
+bool Envoy::benchmark::skipExpensiveBenchmarks() { return skip_expensive_benchmarks; }

test/benchmark/main.h

@@ -0,0 +1,13 @@
+#pragma once
+
+/**
+ * Benchmarks can use this to skip or hurry through long-running tests in CI.
+ */
+
+namespace Envoy {
+namespace benchmark {
+
+bool skipExpensiveBenchmarks();
+
+}
+} // namespace Envoy

test/common/upstream/eds_speed_test.cc

@@ -16,6 +16,7 @@
 
 #include "server/transport_socket_config_impl.h"
 
+#include "test/benchmark/main.h"
 #include "test/common/upstream/utility.h"
 #include "test/mocks/local_info/mocks.h"
 #include "test/mocks/protobuf/mocks.h"
@@ -27,12 +28,15 @@
 
 #include "benchmark/benchmark.h"
 
+using ::benchmark::State;
+using Envoy::benchmark::skipExpensiveBenchmarks;
+
 namespace Envoy {
 namespace Upstream {
 
 class EdsSpeedTest {
 public:
-  EdsSpeedTest(benchmark::State& state, bool v2_config)
+  EdsSpeedTest(State& state, bool v2_config)
       : state_(state), v2_config_(v2_config),
         type_url_(v2_config_
                       ? "type.googleapis.com/envoy.api.v2.ClusterLoadAssignment"
@@ -43,7 +47,26 @@ class EdsSpeedTest {
             local_info_, std::unique_ptr<Grpc::MockAsyncClient>(async_client_), dispatcher_,
             *Protobuf::DescriptorPool::generated_pool()->FindMethodByName(
                 "envoy.service.endpoint.v3.EndpointDiscoveryService.StreamEndpoints"),
-            envoy::config::core::v3::ApiVersion::AUTO, random_, stats_, {}, true)) {}
+            envoy::config::core::v3::ApiVersion::AUTO, random_, stats_, {}, true)) {
+    resetCluster(R"EOF(
+      name: name
+      connect_timeout: 0.25s
+      type: EDS
+      eds_cluster_config:
+        service_name: fare
+        eds_config:
+          api_config_source:
+            cluster_names:
+            - eds
+            refresh_delay: 1s
+    )EOF",
+                 Envoy::Upstream::Cluster::InitializePhase::Secondary);
+
+    EXPECT_CALL(*cm_.subscription_factory_.subscription_, start(_));
+    cluster_->initialize([this] { initialized_ = true; });
+    EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(testing::Return(&async_stream_));
+    subscription_->start({"fare"});
+  }
 
   void resetCluster(const std::string& yaml_config, Cluster::InitializePhase initialize_phase) {
     local_info_.node_.mutable_locality()->set_zone("us-east-1a");
@@ -63,30 +86,14 @@ class EdsSpeedTest {
         std::chrono::milliseconds(), false);
   }
 
-  void initialize() {
-    EXPECT_CALL(*cm_.subscription_factory_.subscription_, start(_));
-    cluster_->initialize([this] { initialized_ = true; });
-  }
-
   // Set up an EDS config with multiple priorities, localities, weights and make sure
-  // they are loaded and reloaded as expected.
-  void priorityAndLocalityWeightedHelper(bool ignore_unknown_dynamic_fields, size_t num_hosts) {
+  // they are loaded as expected.
+  void priorityAndLocalityWeightedHelper(bool ignore_unknown_dynamic_fields, size_t num_hosts,
+                                         bool healthy) {
     state_.PauseTiming();
+
     envoy::config::endpoint::v3::ClusterLoadAssignment cluster_load_assignment;
     cluster_load_assignment.set_cluster_name("fare");
-    resetCluster(R"EOF(
-      name: name
-      connect_timeout: 0.25s
-      type: EDS
-      eds_cluster_config:
-        service_name: fare
-        eds_config:
-          api_config_source:
-            cluster_names:
-            - eds
-            refresh_delay: 1s
-    )EOF",
-                 Envoy::Upstream::Cluster::InitializePhase::Secondary);
 
     // Add a whole bunch of hosts in a single place:
     auto* endpoints = cluster_load_assignment.add_endpoints();
@@ -99,18 +106,21 @@ class EdsSpeedTest {
 
     uint32_t port = 1000;
     for (size_t i = 0; i < num_hosts; ++i) {
-      auto* socket_address = endpoints->add_lb_endpoints()
-                                 ->mutable_endpoint()
-                                 ->mutable_address()
-                                 ->mutable_socket_address();
+      auto* lb_endpoint = endpoints->add_lb_endpoints();
+      if (healthy) {
+        lb_endpoint->set_health_status(envoy::config::core::v3::HEALTHY);
+      } else {
+        lb_endpoint->set_health_status(envoy::config::core::v3::UNHEALTHY);
+      }
+      auto* socket_address =
+          lb_endpoint->mutable_endpoint()->mutable_address()->mutable_socket_address();
       socket_address->set_address("10.0.1." + std::to_string(i / 60000));
       socket_address->set_port_value((port + i) % 60000);
     }
 
     // this is what we're actually testing:
     validation_visitor_.setSkipValidation(ignore_unknown_dynamic_fields);
 
-    initialize();
     auto response = std::make_unique<envoy::service::discovery::v3::DiscoveryResponse>();
     response->set_type_url(type_url_);
     auto* resource = response->mutable_resources()->Add();
@@ -121,16 +131,13 @@ class EdsSpeedTest {
                      "");
       resource->set_type_url("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment");
     }
-    EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(testing::Return(&async_stream_));
-    subscription_->start({"fare"});
     state_.ResumeTiming();
     grpc_mux_->grpcStreamForTest().onReceiveMessage(std::move(response));
-    ASSERT(initialized_);
     ASSERT(cluster_->prioritySet().hostSetsPerPriority()[1]->hostsPerLocality().get()[0].size() ==
            num_hosts);
   }
 
-  benchmark::State& state_;
+  State& state_;
   const bool v2_config_;
   const std::string type_url_;
   bool initialized_{};
@@ -161,14 +168,50 @@ class EdsSpeedTest {
 } // namespace Upstream
 } // namespace Envoy
 
-static void priorityAndLocalityWeighted(benchmark::State& state) {
+static void priorityAndLocalityWeighted(State& state) {
   Envoy::Thread::MutexBasicLockable lock;
   Envoy::Logger::Context logging_state(spdlog::level::warn,
                                        Envoy::Logger::Logger::DEFAULT_LOG_FORMAT, lock, false);
   for (auto _ : state) {
     Envoy::Upstream::EdsSpeedTest speed_test(state, state.range(0));
-    speed_test.priorityAndLocalityWeightedHelper(state.range(1), state.range(2));
+    // if we've been instructed to skip tests, only run once no matter the argument:
+    uint32_t endpoints = skipExpensiveBenchmarks() ? 1 : state.range(2);
+
+    speed_test.priorityAndLocalityWeightedHelper(state.range(1), endpoints, true);
+  }
+}
+
+BENCHMARK(priorityAndLocalityWeighted)
+    ->Ranges({{false, true}, {false, true}, {1, 100000}})
+    ->Unit(benchmark::kMillisecond);
+
+static void duplicateUpdate(State& state) {
+  Envoy::Thread::MutexBasicLockable lock;
+  Envoy::Logger::Context logging_state(spdlog::level::warn,
+                                       Envoy::Logger::Logger::DEFAULT_LOG_FORMAT, lock, false);
+
+  for (auto _ : state) {
+    Envoy::Upstream::EdsSpeedTest speed_test(state, false);
+    uint32_t endpoints = skipExpensiveBenchmarks() ? 1 : state.range(0);
+
+    speed_test.priorityAndLocalityWeightedHelper(true, endpoints, true);
+    speed_test.priorityAndLocalityWeightedHelper(true, endpoints, true);
+  }
+}
+
+BENCHMARK(duplicateUpdate)->Range(1, 100000)->Unit(benchmark::kMillisecond);
+
+static void healthOnlyUpdate(State& state) {
+  Envoy::Thread::MutexBasicLockable lock;
+  Envoy::Logger::Context logging_state(spdlog::level::warn,
+                                       Envoy::Logger::Logger::DEFAULT_LOG_FORMAT, lock, false);
+  for (auto _ : state) {
+    Envoy::Upstream::EdsSpeedTest speed_test(state, false);
+    uint32_t endpoints = skipExpensiveBenchmarks() ? 1 : state.range(0);
+
+    speed_test.priorityAndLocalityWeightedHelper(true, endpoints, true);
+    speed_test.priorityAndLocalityWeightedHelper(true, endpoints, false);
   }
 }
 
-BENCHMARK(priorityAndLocalityWeighted)->Ranges({{false, true}, {false, true}, {2000, 100000}});
+BENCHMARK(healthOnlyUpdate)->Range(1, 100000)->Unit(benchmark::kMillisecond);