eds: decrease computational complexity of updates

source/common/common/utility.h

@@ -709,4 +709,32 @@ class InlineString : public InlineStorage {
   char data_[];
 };
 
+/**
+ * Utilities for working with containers.
+ */
+class Containers {
+public:
+  /**
+   * Remove elements that match the predicate from the list by swapping them to
+   * the end and truncating. Iterates over c once, order is not preserved.
+   *
+   * @param c The container on which to operate
+   * @param predicate A function which will be called once for every entry in c,
+   * which returns true if that entry should be removed.
+   */
+  template <class Container, typename T>
+  static void removeMatchingElements(Container& c, std::function<bool(const T&)> predicate) {
+    auto erase_from = c.end();
+    for (auto i = c.begin(); i != erase_from;) {
+      if (predicate(*i)) {
+        *i = *(--erase_from);
+      } else {
+        ++i;
+      }
+    }
+
+    c.erase(erase_from, c.end());
+  }
+};
+
 } // namespace Envoy

source/common/upstream/upstream_impl.cc

@@ -1327,9 +1327,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.
 
@@ -1432,16 +1430,18 @@ 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());
+  std::function<bool(const HostSharedPtr&)> predicate =
+      [&existing_hosts_for_current_priority](const HostSharedPtr& p) mutable {
+        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;
+      };
+  Containers::removeMatchingElements(current_priority_hosts, predicate);
 
   // 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.
@@ -1459,22 +1459,22 @@ 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();
+    predicate = [&updated_hosts, &final_hosts, &max_host_weight](const HostSharedPtr& p) mutable {
+      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(*i);
-        updated_hosts[(*i)->address()->asString()] = *i;
-        (*i)->healthFlagSet(Host::HealthFlag::PENDING_DYNAMIC_REMOVAL);
-        i = current_priority_hosts.erase(i);
-      } else {
-        i++;
+        final_hosts.push_back(p);
+        updated_hosts[p->address()->asString()] = p;
+        p->healthFlagSet(Host::HealthFlag::PENDING_DYNAMIC_REMOVAL);
+        return true;
       }
-    }
+      return false;
+    };
   }
+  Containers::removeMatchingElements(current_priority_hosts, predicate);
 
   // At this point we've accounted for all the new hosts as well the hosts that previously
   // existed in this priority.

test/common/common/utility_test.cc

@@ -875,4 +875,29 @@ TEST(InlineStorageTest, InlineString) {
   EXPECT_EQ("Hello, world!", hello->toString());
 }
 
+TEST(ContainerRemoveElementsTest, Containers) {
+  auto l = StringUtil::splitToken("one,two,three", ",");
+  std::function<bool(const absl::string_view&)> onep = [](const absl::string_view& s) {
+    return "one" == s;
+  };
+  std::function<bool(const absl::string_view&)> truep =
+      [](ABSL_ATTRIBUTE_UNUSED const absl::string_view& s) { return true; };
+  std::function<bool(const absl::string_view&)> falsep =
+      [](ABSL_ATTRIBUTE_UNUSED const absl::string_view& s) { return false; };
+
+  Containers::removeMatchingElements(l, falsep);
+  // nothing is removed:
+  EXPECT_EQ(3, l.size());
+
+  Containers::removeMatchingElements(l, onep);
+  // one element is removed:
+  EXPECT_EQ(2, l.size());
+  // and the last element is now first:
+  EXPECT_EQ("three", l[0]);
+
+  Containers::removeMatchingElements(l, truep);
+  // everything is removed:
+  EXPECT_EQ(0, l.size());
+}
+
 } // namespace Envoy

test/common/upstream/eds_speed_test.cc

@@ -43,7 +43,24 @@ 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; });
+  }
 
   void resetCluster(const std::string& yaml_config, Cluster::InitializePhase initialize_phase) {
     local_info_.node_.mutable_locality()->set_zone("us-east-1a");
@@ -63,30 +80,13 @@ 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 +99,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,11 +124,11 @@ class EdsSpeedTest {
                      "");
       resource->set_type_url("type.googleapis.com/envoy.api.v2.ClusterLoadAssignment");
     }
-    EXPECT_CALL(*async_client_, startRaw(_, _, _, _)).WillOnce(testing::Return(&async_stream_));
+    EXPECT_CALL(*async_client_, startRaw(_, _, _, _))
+        .WillRepeatedly(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);
   }
@@ -165,8 +168,34 @@ static void priorityAndLocalityWeighted(benchmark::State& state) {
                                        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));
+    speed_test.priorityAndLocalityWeightedHelper(state.range(1), state.range(2), true);
   }
 }
 
 BENCHMARK(priorityAndLocalityWeighted)->Ranges({{false, true}, {false, true}, {2000, 100000}});
+
+static void duplicateUpdate(benchmark::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);
+    speed_test.priorityAndLocalityWeightedHelper(true, state.range(0), true);
+    speed_test.priorityAndLocalityWeightedHelper(true, state.range(0), true);
+  }
+}
+
+BENCHMARK(duplicateUpdate)->Range(2000, 100000);
+
+static void healthOnlyUpdate(benchmark::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);
+    speed_test.priorityAndLocalityWeightedHelper(true, state.range(0), true);
+    speed_test.priorityAndLocalityWeightedHelper(true, state.range(0), false);
+  }
+}
+
+BENCHMARK(healthOnlyUpdate)->Range(2000, 100000);