Enable Windows workers (envoyproxy#17555)

Fixing an issue where Envoy workers are not picking up connections on Windows.

The root cause of the issue is in the Windows kernel.

There are two issues that we need to consider:

If we want to listen from a duplicated socket then we need to duplicate it after we call listen on the original socket.
If duplicated sockets try to accept at the same time, then one of the accept calls might block. Even if the sockets are non-blocking.
The best way to work around that issue is to only listen/accept connections from the first worker thread and then use ExactConnectionBalance to dispatch the connection to another worker thread. This is a temporary solution until the underlying issue is fixed on Windows.

Signed-off-by: Sotiris Nanopoulos <sonanopo@microsoft.com>

Author: Sotiris Nanopoulos

docs/root/intro/arch_overview/intro/threading_model.rst

@@ -24,3 +24,7 @@ to have Envoy forcibly balance connections between worker threads. To support th
 Envoy allows for different types of :ref:`connection balancing
 <envoy_v3_api_field_config.listener.v3.Listener.connection_balance_config>` to be configured on each :ref:`listener
 <arch_overview_listeners>`.
+
+On Windows the kernel is not able to balance the connections properly with the async IO model that Envoy is using.
+Until this is fixed by the platfrom, Envoy will enforce listener connection balancing on Windows. This allows us to
+balance connections between different worker threads. This behavior comes with a performance penalty.

docs/root/version_history/current.rst

@@ -9,6 +9,9 @@ Minor Behavior Changes
 
 Bug Fixes
 ---------
+*Changes expected to improve the state of the world and are unlikely to have negative effects*
+
+* listener: fixed an issue on Windows where connections are not handled by all worker threads.
 
 Removed Config or Runtime
 -------------------------

source/server/listener_impl.cc

@@ -523,6 +523,19 @@ void ListenerImpl::buildFilterChains() {
 void ListenerImpl::buildSocketOptions() {
   // TCP specific setup.
   if (connection_balancer_ == nullptr) {
+#ifdef WIN32
+    // On Windows we use the exact connection balancer to dispatch connections
+    // from worker 1 to all workers. This is a perf hit but it is the only way
+    // to make all the workers do work.
+    // TODO(davinci26): We can be faster here if we create a balancer implementation
+    // that dispatches the connection to a random thread.
+    ENVOY_LOG(warn,
+              "ExactBalance was forced enabled for TCP listener '{}' because "
+              "Envoy is running on Windows."
+              "ExactBalance is used to load balance connections between workers on Windows.",
+              config_.name());
+    connection_balancer_ = std::make_shared<Network::ExactConnectionBalancerImpl>();
+#else
     // Not in place listener update.
     if (config_.has_connection_balance_config()) {
       // Currently exact balance is the only supported type and there are no options.
@@ -531,6 +544,7 @@ void ListenerImpl::buildSocketOptions() {
     } else {
       connection_balancer_ = std::make_shared<Network::NopConnectionBalancerImpl>();
     }
+#endif
   }
 
   if (config_.has_tcp_fast_open_queue_length()) {

test/integration/integration_test.cc

@@ -143,6 +143,53 @@ TEST_P(IntegrationTest, PerWorkerStatsAndBalancing) {
   check_listener_stats(0, 1);
 }
 
+// Make sure all workers pick up connections
+TEST_P(IntegrationTest, AllWorkersAreHandlingLoad) {
+  concurrency_ = 2;
+  initialize();
+
+  std::string worker0_stat_name, worker1_stat_name;
+  if (GetParam() == Network::Address::IpVersion::v4) {
+    worker0_stat_name = "listener.127.0.0.1_0.worker_0.downstream_cx_total";
+    worker1_stat_name = "listener.127.0.0.1_0.worker_1.downstream_cx_total";
+  } else {
+    worker0_stat_name = "listener.[__1]_0.worker_0.downstream_cx_total";
+    worker1_stat_name = "listener.[__1]_0.worker_1.downstream_cx_total";
+  }
+
+  test_server_->waitForCounterEq(worker0_stat_name, 0);
+  test_server_->waitForCounterEq(worker1_stat_name, 0);
+
+  // We set the counters for the two workers to see how many connections each handles.
+  uint64_t w0_ctr = 0;
+  uint64_t w1_ctr = 0;
+  constexpr int loops = 5;
+  for (int i = 0; i < loops; i++) {
+    constexpr int requests_per_loop = 4;
+    std::array<IntegrationCodecClientPtr, requests_per_loop> connections;
+    for (int j = 0; j < requests_per_loop; j++) {
+      connections[j] = makeHttpConnection(lookupPort("http"));
+    }
+
+    auto worker0_ctr = test_server_->counter(worker0_stat_name);
+    auto worker1_ctr = test_server_->counter(worker1_stat_name);
+    auto target = w0_ctr + w1_ctr + requests_per_loop;
+    while (test_server_->counter(worker0_stat_name)->value() +
+               test_server_->counter(worker1_stat_name)->value() <
+           target) {
+      timeSystem().advanceTimeWait(std::chrono::milliseconds(10));
+    }
+    w0_ctr = test_server_->counter(worker0_stat_name)->value();
+    w1_ctr = test_server_->counter(worker1_stat_name)->value();
+    for (int j = 0; j < requests_per_loop; j++) {
+      connections[j]->close();
+    }
+  }
+
+  EXPECT_TRUE(w0_ctr > 1);
+  EXPECT_TRUE(w1_ctr > 1);
+}
+
 TEST_P(IntegrationTest, RouterDirectResponseWithBody) {
   const std::string body = "Response body";
   const std::string file_path = TestEnvironment::writeStringToFileForTest("test_envoy", body);