[Windows] Fixes Udp listener tests (#12635)

Fixes UDP listener tests on Windows by modernizing iovecToWSABUF and msghdrToWSAMSG from pointer arithmetic to c++ and makes message.msg_controllen the proper length.

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

source/common/api/win32/os_sys_calls_impl.cc

@@ -14,14 +14,15 @@ namespace Envoy {
 namespace Api {
 namespace {
 
-using WSABUFPtr = std::unique_ptr<WSABUF[]>;
+using WSAMSGPtr = std::unique_ptr<WSAMSG>;
 
-struct wsabufResult {
-  DWORD num_vec_;
-  WSABUFPtr wsabuf_;
+struct wsamsgResult {
+  WSAMSGPtr wsamsg_;
+  std::vector<WSABUF> buff_data_;
 };
 
-wsabufResult iovecToWSABUF(const iovec* vec, int in_vec) {
+std::vector<WSABUF> iovecToWSABUF(const iovec* vec, int in_vec) {
+
   DWORD num_vec = 0;
   for (int i = 0; i < in_vec; i++) {
     size_t cur_len = vec[i].iov_len;
@@ -32,26 +33,26 @@ wsabufResult iovecToWSABUF(const iovec* vec, int in_vec) {
     }
   }
 
-  WSABUFPtr wsa_buf(new WSABUF[num_vec]);
-
-  WSABUF* wsa_elt = wsa_buf.get();
-  for (int i = 0; i < in_vec; i++) {
-    CHAR* base = static_cast<CHAR*>(vec[i].iov_base);
-    size_t cur_len = vec[i].iov_len;
+  std::vector<WSABUF> buff(num_vec);
+  auto it = buff.begin();
+
+  std::vector<iovec> vecs(vec, vec + in_vec);
+  for (const auto& vec : vecs) {
+    auto chunk = (CHAR*)vec.iov_base;
+    size_t chunk_len = vec.iov_len;
+    // There is the case that the chunk does not fit into a single WSABUF buffer
+    // this is the case because sizeof(size_t) > sizeof(DWORD).
+    // In this case we split the chunk into multiple WSABUF buffers
+    auto remaining_data = chunk_len;
     do {
-      wsa_elt->buf = base;
-      if (cur_len > DWORD_MAX) {
-        wsa_elt->len = DWORD_MAX;
-      } else {
-        wsa_elt->len = static_cast<DWORD>(cur_len);
-      }
-      base += wsa_elt->len;
-      cur_len -= wsa_elt->len;
-      ++wsa_elt;
-    } while (cur_len > 0);
+      (*it).buf = chunk;
+      (*it).len = (remaining_data > DWORD_MAX) ? DWORD_MAX : static_cast<ULONG>(chunk_len);
+      remaining_data -= (*it).len;
+      chunk += (*it).len;
+      it++;
+    } while (remaining_data > 0);
   }
-
-  return {num_vec, std::move(wsa_buf)};
+  return buff;
 }
 
 LPFN_WSARECVMSG getFnPtrWSARecvMsg() {
@@ -73,23 +74,22 @@ LPFN_WSARECVMSG getFnPtrWSARecvMsg() {
   return recvmsg_fn_ptr;
 }
 
-using WSAMSGPtr = std::unique_ptr<WSAMSG>;
-
-WSAMSGPtr msghdrToWSAMSG(const msghdr* msg) {
+wsamsgResult msghdrToWSAMSG(const msghdr* msg) {
   WSAMSGPtr wsa_msg(new WSAMSG);
 
   wsa_msg->name = reinterpret_cast<SOCKADDR*>(msg->msg_name);
   wsa_msg->namelen = msg->msg_namelen;
-  wsabufResult wsabuf = iovecToWSABUF(msg->msg_iov, msg->msg_iovlen);
-  wsa_msg->lpBuffers = wsabuf.wsabuf_.get();
-  wsa_msg->dwBufferCount = wsabuf.num_vec_;
+  auto buffer = iovecToWSABUF(msg->msg_iov, msg->msg_iovlen);
+  wsa_msg->lpBuffers = buffer.data();
+  wsa_msg->dwBufferCount = buffer.size();
+
   WSABUF control;
   control.buf = reinterpret_cast<CHAR*>(msg->msg_control);
   control.len = msg->msg_controllen;
   wsa_msg->Control = control;
   wsa_msg->dwFlags = msg->msg_flags;
 
-  return wsa_msg;
+  return wsamsgResult{std::move(wsa_msg), std::move(buffer)};
 }
 
 } // namespace
@@ -116,10 +116,9 @@ SysCallIntResult OsSysCallsImpl::close(os_fd_t fd) {
 
 SysCallSizeResult OsSysCallsImpl::writev(os_fd_t fd, const iovec* iov, int num_iov) {
   DWORD bytes_sent;
-  wsabufResult wsabuf = iovecToWSABUF(iov, num_iov);
+  auto buffer = iovecToWSABUF(iov, num_iov);
 
-  const int rc =
-      ::WSASend(fd, wsabuf.wsabuf_.get(), wsabuf.num_vec_, &bytes_sent, 0, nullptr, nullptr);
+  const int rc = ::WSASend(fd, buffer.data(), buffer.size(), &bytes_sent, 0, nullptr, nullptr);
   if (SOCKET_FAILURE(rc)) {
     return {-1, ::WSAGetLastError()};
   }
@@ -129,10 +128,10 @@ SysCallSizeResult OsSysCallsImpl::writev(os_fd_t fd, const iovec* iov, int num_i
 SysCallSizeResult OsSysCallsImpl::readv(os_fd_t fd, const iovec* iov, int num_iov) {
   DWORD bytes_received;
   DWORD flags = 0;
-  wsabufResult wsabuf = iovecToWSABUF(iov, num_iov);
+  auto buffer = iovecToWSABUF(iov, num_iov);
 
-  const int rc = ::WSARecv(fd, wsabuf.wsabuf_.get(), wsabuf.num_vec_, &bytes_received, &flags,
-                           nullptr, nullptr);
+  const int rc =
+      ::WSARecv(fd, buffer.data(), buffer.size(), &bytes_received, &flags, nullptr, nullptr);
   if (SOCKET_FAILURE(rc)) {
     return {-1, ::WSAGetLastError()};
   }
@@ -147,16 +146,16 @@ SysCallSizeResult OsSysCallsImpl::recv(os_fd_t socket, void* buffer, size_t leng
 SysCallSizeResult OsSysCallsImpl::recvmsg(os_fd_t sockfd, msghdr* msg, int flags) {
   DWORD bytes_received;
   LPFN_WSARECVMSG recvmsg_fn_ptr = getFnPtrWSARecvMsg();
-  WSAMSGPtr wsa_msg = msghdrToWSAMSG(msg);
+  wsamsgResult wsamsg = msghdrToWSAMSG(msg);
   // Windows supports only a single flag on input to WSARecvMsg
-  wsa_msg->dwFlags = flags & MSG_PEEK;
-  const int rc = recvmsg_fn_ptr(sockfd, wsa_msg.get(), &bytes_received, nullptr, nullptr);
+  wsamsg.wsamsg_->dwFlags = flags & MSG_PEEK;
+  const int rc = recvmsg_fn_ptr(sockfd, wsamsg.wsamsg_.get(), &bytes_received, nullptr, nullptr);
   if (rc == SOCKET_ERROR) {
     return {-1, ::WSAGetLastError()};
   }
-  msg->msg_namelen = wsa_msg->namelen;
-  msg->msg_flags = wsa_msg->dwFlags;
-  msg->msg_controllen = wsa_msg->Control.len;
+  msg->msg_namelen = wsamsg.wsamsg_->namelen;
+  msg->msg_flags = wsamsg.wsamsg_->dwFlags;
+  msg->msg_controllen = wsamsg.wsamsg_->Control.len;
   return {bytes_received, 0};
 }
 
@@ -215,8 +214,9 @@ SysCallSocketResult OsSysCallsImpl::socket(int domain, int type, int protocol) {
 SysCallSizeResult OsSysCallsImpl::sendmsg(os_fd_t sockfd, const msghdr* msg, int flags) {
   DWORD bytes_received;
   // if overlapped and/or completion routines are supported adjust the arguments accordingly
+  wsamsgResult wsamsg = msghdrToWSAMSG(msg);
   const int rc =
-      ::WSASendMsg(sockfd, msghdrToWSAMSG(msg).get(), flags, &bytes_received, nullptr, nullptr);
+      ::WSASendMsg(sockfd, wsamsg.wsamsg_.get(), flags, &bytes_received, nullptr, nullptr);
   if (rc == SOCKET_ERROR) {
     return {-1, ::WSAGetLastError()};
   }

source/common/network/io_socket_handle_impl.cc

@@ -99,15 +99,16 @@ Api::IoCallUint64Result IoSocketHandleImpl::sendmsg(const Buffer::RawSlice* slic
     return sysCallResultToIoCallResult(result);
   } else {
     const size_t space_v6 = CMSG_SPACE(sizeof(in6_pktinfo));
-    // FreeBSD only needs in_addr size, but allocates more to unify code in two platforms.
     const size_t space_v4 = CMSG_SPACE(sizeof(in_pktinfo));
-    const size_t cmsg_space = (space_v4 < space_v6) ? space_v6 : space_v4;
+
+    // FreeBSD only needs in_addr size, but allocates more to unify code in two platforms.
+    const size_t cmsg_space = (self_ip->version() == Address::IpVersion::v4) ? space_v4 : space_v6;
     // kSpaceForIp should be big enough to hold both IPv4 and IPv6 packet info.
     absl::FixedArray<char> cbuf(cmsg_space);
     memset(cbuf.begin(), 0, cmsg_space);
 
     message.msg_control = cbuf.begin();
-    message.msg_controllen = cmsg_space * sizeof(char);
+    message.msg_controllen = cmsg_space;
     cmsghdr* const cmsg = CMSG_FIRSTHDR(&message);
     RELEASE_ASSERT(cmsg != nullptr, fmt::format("cbuf with size {} is not enough, cmsghdr size {}",
                                                 sizeof(cbuf), sizeof(cmsghdr)));

test/common/network/BUILD

@@ -218,7 +218,6 @@ envoy_cc_test_library(
 envoy_cc_test(
     name = "udp_listener_impl_test",
     srcs = ["udp_listener_impl_test.cc"],
-    tags = ["fails_on_windows"],
     deps = [
         ":udp_listener_impl_test_base_lib",
         "//source/common/event:dispatcher_lib",

test/common/network/udp_listener_impl_test_base.h

@@ -47,7 +47,7 @@ class UdpListenerImplTestBase : public ListenerImplTestBase {
 
   SocketSharedPtr createServerSocket(bool bind) {
     // Set IP_FREEBIND to allow sendmsg to send with non-local IPv6 source address.
-    return std::make_shared<UdpListenSocket>(Network::Test::getAnyAddress(version_),
+    return std::make_shared<UdpListenSocket>(Network::Test::getCanonicalLoopbackAddress(version_),
 #ifdef IP_FREEBIND
                                              SocketOptionFactory::buildIpFreebindOptions(),
 #else
@@ -64,12 +64,7 @@ class UdpListenerImplTestBase : public ListenerImplTestBase {
     if (version_ == Address::IpVersion::v4) {
       // Linux kernel regards any 127.x.x.x as local address. But Mac OS doesn't.
       send_from_addr = std::make_shared<Address::Ipv4Instance>(
-#ifndef __APPLE__
-          "127.1.2.3",
-#else
-          "127.0.0.1",
-#endif
-          server_socket_->localAddress()->ip()->port());
+          "127.0.0.1", server_socket_->localAddress()->ip()->port());
     } else {
       // Only use non-local v6 address if IP_FREEBIND is supported. Otherwise use
       // ::1 to avoid EINVAL error. Unfortunately this can't verify that sendmsg with

tools/spelling/spelling_dictionary.txt

@@ -70,6 +70,7 @@ DRYs
 DS
 DST
 DW
+DWORD
 EADDRINUSE
 EADDRNOTAVAIL
 EAGAIN
@@ -351,6 +352,7 @@ WRONGPASS
 WRR
 WS
 WSA
+WSABUF
 WSS
 Welford's
 Wi