test dummy duplication in tests

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

Author: Sotiris Nanopoulos

source/common/api/win32/os_sys_calls_impl.cc

@@ -367,14 +367,8 @@ SysCallSizeResult OsSysCallsImpl::write(os_fd_t sockfd, const void* buffer, size
 }
 
 SysCallSocketResult OsSysCallsImpl::duplicate(os_fd_t oldfd) {
-  WSAPROTOCOL_INFO info;
-  auto currentProcess = ::GetCurrentProcessId();
-  auto rc = WSADuplicateSocket(oldfd, currentProcess, &info);
-  if (rc == SOCKET_ERROR) {
-    return {(SOCKET)-1, ::WSAGetLastError()};
-  }
-  auto new_socket = ::WSASocket(info.iAddressFamily, info.iSocketType, info.iProtocol, &info, 0, 0);
-  return {new_socket, SOCKET_VALID(new_socket) ? 0 : ::WSAGetLastError()};
+  // Do a swallow copy.
+  return {oldfd, ::WSAGetLastError()};
 }
 
 SysCallSocketResult OsSysCallsImpl::accept(os_fd_t sockfd, sockaddr* addr, socklen_t* addrlen) {

source/server/listener_impl.cc

@@ -535,6 +535,11 @@ void ListenerImpl::buildFilterChains() {
 }
 
 void ListenerImpl::buildSocketOptions() {
+#ifdef WIN32
+  // On Windows we use the exact connection balancer in along with reuse_port to
+  // balance connections between workers.
+  connection_balancer_ = std::make_shared<Network::ExactConnectionBalancerImpl>();
+#else
   // TCP specific setup.
   if (connection_balancer_ == nullptr) {
     // Not in place listener update.
@@ -546,7 +551,7 @@ void ListenerImpl::buildSocketOptions() {
       connection_balancer_ = std::make_shared<Network::NopConnectionBalancerImpl>();
     }
   }
-
+#endif
   if (config_.has_tcp_fast_open_queue_length()) {
     addListenSocketOptions(Network::SocketOptionFactory::buildTcpFastOpenOptions(
         config_.tcp_fast_open_queue_length().value()));

test/integration/integration_test.cc

@@ -157,6 +157,55 @@ TEST_P(IntegrationTest, PerWorkerStatsAndBalancing) {
   check_listener_stats(0, 1);
 }
 
+// Make sure we have correctly specified per-worker performance stats.
+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;
+    ENVOY_LOG_MISC(info, "current values are {} {} with target {}", worker0_ctr->value(),
+                   worker1_ctr->value(), target);
+    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);