[cdc_rsync] Add support for ServerSocket on Windows (#48)

Makes ServerSocket multi-platform, mainly by working around some small
API differences. The code is largely the same, there should be no
differences on Linux.

Also moves WSAStartup() and WSACleanup() up to the Socket level as
static methods because it's used by both ClientSocket and ServerSocket,
and because it doesn't make sense to do that in the socket class as
that would prevent one from using several sockets.

all_files.vcxitems

@@ -15,6 +15,7 @@
   </ItemGroup>
   <ItemGroup>
     <ClCompile Include="$(MSBuildThisFileDirectory)absl_helper\jedec_size_flag.cc" />
+    <ClCompile Include="$(MSBuildThisFileDirectory)cdc_rsync\base\socket.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\asset_stream_config.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\asset_stream_server.cc" />
     <ClCompile Include="$(MSBuildThisFileDirectory)cdc_stream\background_service_impl.cc" />

cdc_rsync/base/BUILD

@@ -80,7 +80,15 @@ cc_library(
 
 cc_library(
     name = "socket",
+    srcs = ["socket.cc"],
     hdrs = ["socket.h"],
+    deps = [
+        "//common:log",
+        "//common:platform",
+        "//common:status",
+        "//common:util",
+        "@com_google_absl//absl/status",
+    ],
 )
 
 filegroup(

cdc_rsync/base/socket.cc

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "cdc_rsync/base/socket.h"
+
+#include "common/log.h"
+#include "common/platform.h"
+#include "common/status.h"
+#include "common/util.h"
+
+#if PLATFORM_WINDOWS
+#include <winsock2.h>
+#endif
+
+namespace cdc_ft {
+
+// static
+absl::Status Socket::Initialize() {
+#if PLATFORM_WINDOWS
+  WSADATA wsaData;
+  const int result = WSAStartup(MAKEWORD(2, 2), &wsaData);
+  if (result != 0) {
+    return MakeStatus("WSAStartup() failed: %s", Util::GetWin32Error(result));
+  }
+  return absl::OkStatus();
+#elif PLATFORM_LINUX
+  return absl::OkStatus();
+#endif
+}
+
+// static
+absl::Status Socket::Shutdown() {
+#if PLATFORM_WINDOWS
+  const int result = WSACleanup();
+  if (result == SOCKET_ERROR) {
+    return MakeStatus("WSACleanup() failed: %s",
+                      Util::GetWin32Error(WSAGetLastError()));
+  }
+  return absl::OkStatus();
+#elif PLATFORM_LINUX
+  return absl::OkStatus();
+#endif
+}
+
+SocketFinalizer::~SocketFinalizer() {
+  absl::Status status = Socket::Shutdown();
+  if (!status.ok()) {
+    LOG_ERROR("Socket shutdown failed: %s", status.message())
+  }
+};
+
+}  // namespace cdc_ft

cdc_rsync/base/socket.h

@@ -26,6 +26,14 @@ class Socket {
   Socket() = default;
   virtual ~Socket() = default;
 
+  // Calls WSAStartup() on Windows, no-op on Linux.
+  // Must be called before using sockets.
+  static absl::Status Initialize();
+
+  // Calls WSACleanup() on Windows, no-op on Linux.
+  // Must be called after using sockets.
+  static absl::Status Shutdown();
+
   // Send data to the socket.
   virtual absl::Status Send(const void* buffer, size_t size) = 0;
 
@@ -40,6 +48,12 @@ class Socket {
                                size_t* bytes_received) = 0;
 };
 
+// Convenience class that calls Shutdown() on destruction. Logs on errors.
+class SocketFinalizer {
+ public:
+  ~SocketFinalizer();
+};
+
 }  // namespace cdc_ft
 
 #endif  // CDC_RSYNC_BASE_SOCKET_H_

cdc_rsync/cdc_rsync_client.cc

@@ -263,6 +263,12 @@ absl::Status CdcRsyncClient::StartServer() {
     return SetTag(MakeStatus("Redeploy server"), Tag::kDeployServer);
   }
 
+  status = Socket::Initialize();
+  if (!status.ok()) {
+    return WrapStatus(status, "Failed to initialize sockets");
+  }
+  socket_finalizer_ = std::make_unique<SocketFinalizer>();
+
   assert(is_server_listening_);
   status = socket_.Connect(port);
   if (!status.ok()) {

cdc_rsync/cdc_rsync_client.h

@@ -123,6 +123,7 @@ class CdcRsyncClient {
   WinProcessFactory process_factory_;
   RemoteUtil remote_util_;
   PortManager port_manager_;
+  std::unique_ptr<SocketFinalizer> socket_finalizer_;
   ClientSocket socket_;
   MessagePump message_pump_{&socket_, MessagePump::PacketReceivedDelegate()};
   ConsoleProgressPrinter printer_;

cdc_rsync/client_socket.cc

@@ -39,23 +39,17 @@ absl::Status MakeSocketStatus(const char* message) {
 
 }  // namespace
 
-struct SocketInfo {
+struct ClientSocketInfo {
   SOCKET socket;
 
-  SocketInfo() : socket(INVALID_SOCKET) {}
+  ClientSocketInfo() : socket(INVALID_SOCKET) {}
 };
 
 ClientSocket::ClientSocket() = default;
 
 ClientSocket::~ClientSocket() { Disconnect(); }
 
 absl::Status ClientSocket::Connect(int port) {
-  WSADATA wsaData;
-  int result = WSAStartup(MAKEWORD(2, 2), &wsaData);
-  if (result != 0) {
-    return MakeStatus("WSAStartup() failed: %i", result);
-  }
-
   addrinfo hints;
   ZeroMemory(&hints, sizeof(hints));
   hints.ai_family = AF_INET;
@@ -64,14 +58,13 @@ absl::Status ClientSocket::Connect(int port) {
 
   // Resolve the server address and port.
   addrinfo* addr_infos = nullptr;
-  result = getaddrinfo("localhost", std::to_string(port).c_str(), &hints,
-                       &addr_infos);
+  int result = getaddrinfo("localhost", std::to_string(port).c_str(), &hints,
+                           &addr_infos);
   if (result != 0) {
-    WSACleanup();
     return MakeStatus("getaddrinfo() failed: %i", result);
   }
 
-  socket_info_ = std::make_unique<SocketInfo>();
+  socket_info_ = std::make_unique<ClientSocketInfo>();
   int count = 0;
   for (addrinfo* curr = addr_infos; curr; curr = curr->ai_next, count++) {
     socket_info_->socket =
@@ -101,7 +94,6 @@ absl::Status ClientSocket::Connect(int port) {
 
   if (socket_info_->socket == INVALID_SOCKET) {
     socket_info_.reset();
-    WSACleanup();
     return MakeStatus("Unable to connect to port %i", port);
   }
 
@@ -120,7 +112,6 @@ void ClientSocket::Disconnect() {
   }
 
   socket_info_.reset();
-  WSACleanup();
 }
 
 absl::Status ClientSocket::Send(const void* buffer, size_t size) {

cdc_rsync/client_socket.h

@@ -45,7 +45,7 @@ class ClientSocket : public Socket {
                        size_t* bytes_received) override;
 
  private:
-  std::unique_ptr<struct SocketInfo> socket_info_;
+  std::unique_ptr<struct ClientSocketInfo> socket_info_;
 };
 
 }  // namespace cdc_ft

cdc_rsync_server/BUILD

@@ -127,11 +127,17 @@ cc_library(
     name = "server_socket",
     srcs = ["server_socket.cc"],
     hdrs = ["server_socket.h"],
-    target_compatible_with = ["@platforms//os:linux"],
+    linkopts = select({
+        "//tools:windows": [
+            "/DEFAULTLIB:Ws2_32.lib",  # Sockets, e.g. recv, send, WSA*.
+        ],
+        "//conditions:default": [],
+    }),
     deps = [
         "//cdc_rsync/base:socket",
         "//common:log",
         "//common:status",
+        "//common:util",
         "@com_google_absl//absl/status",
     ],
 )

cdc_rsync_server/cdc_rsync_server.cc

@@ -148,10 +148,7 @@ PathFilter::Rule::Type ToInternalType(
 
 CdcRsyncServer::CdcRsyncServer() = default;
 
-CdcRsyncServer::~CdcRsyncServer() {
-  message_pump_.reset();
-  socket_.reset();
-}
+CdcRsyncServer::~CdcRsyncServer() = default;
 
 bool CdcRsyncServer::CheckComponents(
     const std::vector<GameletComponent>& components) {
@@ -173,8 +170,14 @@ bool CdcRsyncServer::CheckComponents(
 }
 
 absl::Status CdcRsyncServer::Run(int port) {
+  absl::Status status = Socket::Initialize();
+  if (!status.ok()) {
+    return WrapStatus(status, "Failed to initialize sockets");
+  }
+  socket_finalizer_ = std::make_unique<SocketFinalizer>();
+
   socket_ = std::make_unique<ServerSocket>();
-  absl::Status status = socket_->StartListening(port);
+  status = socket_->StartListening(port);
   if (!status.ok()) {
     return WrapStatus(status, "Failed to start listening on port %i", port);
   }
@@ -563,7 +566,7 @@ absl::Status CdcRsyncServer::HandleSendMissingFileData() {
     // Verify that there is no directory existing with the same name.
     if (path::Exists(filepath) && path::DirExists(filepath)) {
       assert(!diff_.extraneous_dirs.empty());
-      absl::Status status = path::RemoveFile(filepath);
+      status = path::RemoveFile(filepath);
       if (!status.ok()) {
         return WrapStatus(
             status, "Failed to remove folder '%s' before creating file '%s'",

cdc_rsync_server/cdc_rsync_server.h

@@ -32,6 +32,7 @@ namespace cdc_ft {
 
 class MessagePump;
 class ServerSocket;
+class SocketFinalizer;
 
 class CdcRsyncServer {
  public:
@@ -90,6 +91,8 @@ class CdcRsyncServer {
   // Used to toggle decompression.
   void Thread_OnPackageReceived(PacketType type);
 
+  // The order determines the correct destruction order, so keep it!
+  std::unique_ptr<SocketFinalizer> socket_finalizer_;
   std::unique_ptr<ServerSocket> socket_;
   std::unique_ptr<MessagePump> message_pump_;