Skip to content

Commit

Permalink
[common] Add support for ClientSocket on Linux (#97)
Browse files Browse the repository at this point in the history 
Uses the abstractions written for ServerSocket in ClientSocket, so
that it builds on Linux. Also adds a method to poll for connections
and uses that in cdc_rsync. Similar code will be used in cdc_fuse_fs
to wait for a connection in a future CL.
  • Loading branch information
@ljusten
ljusten authored May 27, 2023
1 parent 26ff934 commit 678ee0f
Show file tree
Hide file tree
Showing 6 changed files with 185 additions and 143 deletions.
21 changes: 3 additions & 18 deletions cdc_rsync/cdc_rsync_client.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -300,24 +300,9 @@ absl::Status CdcRsyncClient::StartServer(const ServerArch& arch) {
"Failed to start cdc_rsync_server process");
}

// Poll the connection to the socket with port |local_port| until the port
// forwarding connection is set up.
timeout_timer.Reset();
for (;;) {
assert(local_port != 0);
status = socket_.Connect(local_port);
if (status.ok()) {
break;
}

if (timeout_timer.ElapsedSeconds() > options_.connection_timeout_sec) {
return SetTag(
absl::DeadlineExceededError("Timeout while connecting to server"),
Tag::kConnectionTimeout);
}

Util::Sleep(10);
}
// Wait for port forwarding to be up.
RETURN_IF_ERROR(ClientSocket::WaitForConnection(
local_port, absl::Seconds(options_.connection_timeout_sec)));

server_process_ = std::move(srv_process);
port_forwarding_process_ = std::move(fwd_process);
Expand Down
11 changes: 8 additions & 3 deletions common/BUILD
View file
Original file line number Diff line number Diff line change
Expand Up @@ -62,15 +62,17 @@ cc_library(

cc_library(
name = "client_socket",
srcs = ["client_socket.cc"],
srcs = [
"client_socket.cc",
"socket_internal.h",
],
hdrs = ["client_socket.h"],
linkopts = select({
"//tools:windows": [
"/DEFAULTLIB:Ws2_32.lib", # Sockets, e.g. recv, send, WSA*.
],
"//conditions:default": [],
}),
target_compatible_with = ["@platforms//os:windows"],
deps = [
":log",
":socket",
Expand Down Expand Up @@ -473,7 +475,10 @@ cc_test(

cc_library(
name = "server_socket",
srcs = ["server_socket.cc"],
srcs = [
"server_socket.cc",
"socket_internal.h",
],
hdrs = ["server_socket.h"],
linkopts = select({
"//tools:windows": [
Expand Down
80 changes: 48 additions & 32 deletions common/client_socket.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,13 +14,12 @@

#include "common/client_socket.h"

#include <winsock2.h>
#include <ws2tcpip.h>

#include <cassert>

#include "common/log.h"
#include "common/socket_internal.h"
#include "common/status.h"
#include "common/stopwatch.h"
#include "common/util.h"

namespace cdc_ft {
Expand All @@ -29,9 +28,9 @@ namespace {
// Creates a status with the given |message| and the last WSA error.
// Assigns Tag::kSocketEof for WSAECONNRESET errors.
absl::Status MakeSocketStatus(const char* message) {
const int err = WSAGetLastError();
absl::Status status = MakeStatus("%s: %s", message, Util::GetWin32Error(err));
if (err == WSAECONNRESET) {
const int err = GetLastError();
absl::Status status = MakeStatus("%s: %s", message, GetErrorStr(err));
if (err == kErrConnReset) {
status = SetTag(status, Tag::kSocketEof);
}
return status;
Expand All @@ -40,18 +39,39 @@ absl::Status MakeSocketStatus(const char* message) {
} // namespace

struct ClientSocketInfo {
SOCKET socket;
SocketType socket;

ClientSocketInfo() : socket(INVALID_SOCKET) {}
ClientSocketInfo() : socket(kInvalidSocket) {}
};

ClientSocket::ClientSocket() = default;

ClientSocket::~ClientSocket() { Disconnect(); }

// static
absl::Status ClientSocket::WaitForConnection(int port, absl::Duration timeout) {
assert(port != 0);
Stopwatch timeout_timer;
ClientSocket socket;
for (;;) {
absl::Status status = socket.Connect(port);
if (status.ok()) {
return absl::OkStatus();
}

if (timeout_timer.Elapsed() > timeout) {
return SetTag(
absl::DeadlineExceededError("Timeout while connecting to server"),
Tag::kConnectionTimeout);
}

Util::Sleep(50);
}
}

absl::Status ClientSocket::Connect(int port) {
addrinfo hints;
ZeroMemory(&hints, sizeof(hints));
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_protocol = IPPROTO_TCP;
Expand All @@ -63,36 +83,34 @@ absl::Status ClientSocket::Connect(int port) {
if (result != 0) {
return MakeStatus("getaddrinfo() failed: %i", result);
}
AddrInfoReleaser releaser(addr_infos);

socket_info_ = std::make_unique<ClientSocketInfo>();
int count = 0;
for (addrinfo* curr = addr_infos; curr; curr = curr->ai_next, count++) {
socket_info_->socket =
socket(curr->ai_family, curr->ai_socktype, curr->ai_protocol);
if (socket_info_->socket == INVALID_SOCKET) {
if (socket_info_->socket == kInvalidSocket) {
LOG_DEBUG("socket() failed for addr_info %i: %s", count,
Util::GetWin32Error(WSAGetLastError()));
GetLastErrorStr());
continue;
}

// Connect to server.
result = connect(socket_info_->socket, curr->ai_addr,
static_cast<int>(curr->ai_addrlen));
if (result == SOCKET_ERROR) {
if (result == kSocketError) {
LOG_DEBUG("connect() failed for addr_info %i: %s", count,
Util::GetWin32Error(WSAGetLastError()));
closesocket(socket_info_->socket);
socket_info_->socket = INVALID_SOCKET;
GetLastErrorStr());
Close(&socket_info_->socket);
continue;
}

// Success!
break;
}

freeaddrinfo(addr_infos);

if (socket_info_->socket == INVALID_SOCKET) {
if (socket_info_->socket == kInvalidSocket) {
socket_info_.reset();
return MakeStatus("Unable to connect to port %i", port);
}
Expand All @@ -106,18 +124,15 @@ void ClientSocket::Disconnect() {
return;
}

if (socket_info_->socket != INVALID_SOCKET) {
closesocket(socket_info_->socket);
socket_info_->socket = INVALID_SOCKET;
}

Close(&socket_info_->socket);
socket_info_.reset();
}

absl::Status ClientSocket::Send(const void* buffer, size_t size) {
int result = send(socket_info_->socket, static_cast<const char*>(buffer),
static_cast<int>(size), /*flags */ 0);
if (result == SOCKET_ERROR) {
int result =
HANDLE_EINTR(send(socket_info_->socket, static_cast<const char*>(buffer),
static_cast<int>(size), /*flags */ 0));
if (result == kSocketError) {
return MakeSocketStatus("send() failed");
}

Expand All @@ -133,9 +148,10 @@ absl::Status ClientSocket::Receive(void* buffer, size_t size,
}

int flags = allow_partial_read ? 0 : MSG_WAITALL;
int bytes_read = recv(socket_info_->socket, static_cast<char*>(buffer),
static_cast<int>(size), flags);
if (bytes_read == SOCKET_ERROR) {
int bytes_read =
HANDLE_EINTR(recv(socket_info_->socket, static_cast<char*>(buffer),
static_cast<int>(size), flags));
if (bytes_read == kSocketError) {
return MakeSocketStatus("recv() failed");
}

Expand All @@ -154,9 +170,9 @@ absl::Status ClientSocket::Receive(void* buffer, size_t size,
}

absl::Status ClientSocket::ShutdownSendingEnd() {
int result = shutdown(socket_info_->socket, SD_SEND);
if (result == SOCKET_ERROR) {
return MakeSocketStatus("shutdown() failed");
int result = shutdown(socket_info_->socket, kSendingEnd);
if (result == kSocketError) {
return MakeStatus("Socket shutdown failed: %s", GetLastErrorStr());
}

return absl::OkStatus();
Expand Down
4 changes: 4 additions & 0 deletions common/client_socket.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@ class ClientSocket : public Socket {
ClientSocket();
~ClientSocket();

// Polls until a connection to |port| succeeds.
// Returns TimeoutError
static absl::Status WaitForConnection(int port, absl::Duration timeout);

// Connects to localhost on |port|.
absl::Status Connect(int port);

Expand Down
91 changes: 1 addition & 90 deletions common/server_socket.cc
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,99 +15,10 @@
#include "common/server_socket.h"

#include "common/log.h"
#include "common/platform.h"
#include "common/socket_internal.h"
#include "common/status.h"
#include "common/util.h"

#if PLATFORM_WINDOWS

#include <winsock2.h>
#include <ws2tcpip.h>

#elif PLATFORM_LINUX

#include <netdb.h>
#include <netinet/in.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

#include <cerrno>

#endif

namespace cdc_ft {
namespace {

#if PLATFORM_WINDOWS

using SocketType = SOCKET;
using SockAddrType = SOCKADDR;
constexpr SocketType kInvalidSocket = INVALID_SOCKET;
constexpr int kSocketError = SOCKET_ERROR;
constexpr int kSendingEnd = SD_SEND;

constexpr int kErrAgain = WSAEWOULDBLOCK; // There's no EAGAIN on Windows.
constexpr int kErrWouldBlock = WSAEWOULDBLOCK;
constexpr int kErrAddrInUse = WSAEADDRINUSE;

int GetLastError() { return WSAGetLastError(); }
std::string GetErrorStr(int err) { return Util::GetWin32Error(err); }
void Close(SocketType* socket) {
if (*socket != kInvalidSocket) {
closesocket(*socket);
*socket = kInvalidSocket;
}
}

// Not necessary on Windows.
#define HANDLE_EINTR(x) (x)

#elif PLATFORM_LINUX

using SocketType = int;
using SockAddrType = sockaddr;
constexpr SocketType kInvalidSocket = -1;
constexpr int kSocketError = -1;
constexpr int kSendingEnd = SHUT_WR;

constexpr int kErrAgain = EAGAIN;
constexpr int kErrWouldBlock = EWOULDBLOCK;
constexpr int kErrAddrInUse = EADDRINUSE;

int GetLastError() { return errno; }
std::string GetErrorStr(int err) { return strerror(err); }
void Close(SocketType* socket) {
if (*socket != kInvalidSocket) {
close(*socket);
*socket = kInvalidSocket;
}
}

// Keep re-evaluating the expression |x| while it returns EINTR.
#define HANDLE_EINTR(x) \
({ \
decltype(x) eintr_wrapper_result; \
do { \
eintr_wrapper_result = (x); \
} while (eintr_wrapper_result == -1 && errno == EINTR); \
eintr_wrapper_result; \
})

#endif

std::string GetLastErrorStr() { return GetErrorStr(GetLastError()); }

class AddrInfoReleaser {
public:
AddrInfoReleaser(addrinfo* addr_infos) : addr_infos_(addr_infos) {}
~AddrInfoReleaser() { freeaddrinfo(addr_infos_); }

private:
addrinfo* addr_infos_;
};

} // namespace

struct ServerSocketInfo {
// Listening socket file descriptor (where new connections are accepted).
Expand Down
Loading

0 comments on commit 678ee0f

Please sign in to comment.