Add higher-level TCP client socket/factory support to QUICHE Added some basic interfaces that could be used to abstract away implementation details of socket operations, including a SocketFactory that allows initiation details like QuicEventLoop to be hidden away from code interacting with it and creating/running sockets. Especially helpful to pass factory objects to code that doesn't need to know details around the event/async libraries in use or need to stay agnostic between them to build on multiple plaforms/environments. Interfaces support my current needs (TCP client sockets), but left them deliberately open to future expansion to server sockets or UDP, etc. Created a QuicEventLoop-based implementation of the interfaces. PiperOrigin-RevId: 466366913
diff --git a/build/source_list.bzl b/build/source_list.bzl index 30d068a..a928fe6 100644 --- a/build/source_list.bzl +++ b/build/source_list.bzl
@@ -947,10 +947,14 @@ "quic/core/batch_writer/quic_batch_writer_test.h", "quic/core/batch_writer/quic_gso_batch_writer.h", "quic/core/batch_writer/quic_sendmmsg_batch_writer.h", + "quic/core/io/event_loop_socket_factory.h", + "quic/core/io/event_loop_tcp_client_socket.h", "quic/core/io/quic_default_event_loop.h", "quic/core/io/quic_event_loop.h", "quic/core/io/quic_poll_event_loop.h", "quic/core/io/socket.h", + "quic/core/io/socket_factory.h", + "quic/core/io/stream_client_socket.h", "quic/core/quic_default_packet_writer.h", "quic/core/quic_epoll_alarm_factory.h", "quic/core/quic_epoll_clock.h", @@ -981,6 +985,8 @@ "quic/core/batch_writer/quic_batch_writer_buffer.cc", "quic/core/batch_writer/quic_gso_batch_writer.cc", "quic/core/batch_writer/quic_sendmmsg_batch_writer.cc", + "quic/core/io/event_loop_socket_factory.cc", + "quic/core/io/event_loop_tcp_client_socket.cc", "quic/core/io/quic_default_event_loop.cc", "quic/core/io/quic_poll_event_loop.cc", "quic/core/io/socket_posix.cc", @@ -1302,6 +1308,7 @@ "quic/core/http/quic_spdy_client_session_test.cc", "quic/core/http/quic_spdy_client_stream_test.cc", "quic/core/http/quic_spdy_server_stream_base_test.cc", + "quic/core/io/event_loop_tcp_client_socket_test.cc", "quic/core/io/quic_all_event_loops_test.cc", "quic/core/io/quic_poll_event_loop_test.cc", "quic/core/io/socket_test.cc",
diff --git a/build/source_list.gni b/build/source_list.gni index daadaf6..2fd7580 100644 --- a/build/source_list.gni +++ b/build/source_list.gni
@@ -947,10 +947,14 @@ "src/quiche/quic/core/batch_writer/quic_batch_writer_test.h", "src/quiche/quic/core/batch_writer/quic_gso_batch_writer.h", "src/quiche/quic/core/batch_writer/quic_sendmmsg_batch_writer.h", + "src/quiche/quic/core/io/event_loop_socket_factory.h", + "src/quiche/quic/core/io/event_loop_tcp_client_socket.h", "src/quiche/quic/core/io/quic_default_event_loop.h", "src/quiche/quic/core/io/quic_event_loop.h", "src/quiche/quic/core/io/quic_poll_event_loop.h", "src/quiche/quic/core/io/socket.h", + "src/quiche/quic/core/io/socket_factory.h", + "src/quiche/quic/core/io/stream_client_socket.h", "src/quiche/quic/core/quic_default_packet_writer.h", "src/quiche/quic/core/quic_epoll_alarm_factory.h", "src/quiche/quic/core/quic_epoll_clock.h", @@ -981,6 +985,8 @@ "src/quiche/quic/core/batch_writer/quic_batch_writer_buffer.cc", "src/quiche/quic/core/batch_writer/quic_gso_batch_writer.cc", "src/quiche/quic/core/batch_writer/quic_sendmmsg_batch_writer.cc", + "src/quiche/quic/core/io/event_loop_socket_factory.cc", + "src/quiche/quic/core/io/event_loop_tcp_client_socket.cc", "src/quiche/quic/core/io/quic_default_event_loop.cc", "src/quiche/quic/core/io/quic_poll_event_loop.cc", "src/quiche/quic/core/io/socket_posix.cc", @@ -1302,6 +1308,7 @@ "src/quiche/quic/core/http/quic_spdy_client_session_test.cc", "src/quiche/quic/core/http/quic_spdy_client_stream_test.cc", "src/quiche/quic/core/http/quic_spdy_server_stream_base_test.cc", + "src/quiche/quic/core/io/event_loop_tcp_client_socket_test.cc", "src/quiche/quic/core/io/quic_all_event_loops_test.cc", "src/quiche/quic/core/io/quic_poll_event_loop_test.cc", "src/quiche/quic/core/io/socket_test.cc",
diff --git a/build/source_list.json b/build/source_list.json index ca31469..02f608b 100644 --- a/build/source_list.json +++ b/build/source_list.json
@@ -946,10 +946,14 @@ "quiche/quic/core/batch_writer/quic_batch_writer_test.h", "quiche/quic/core/batch_writer/quic_gso_batch_writer.h", "quiche/quic/core/batch_writer/quic_sendmmsg_batch_writer.h", + "quiche/quic/core/io/event_loop_socket_factory.h", + "quiche/quic/core/io/event_loop_tcp_client_socket.h", "quiche/quic/core/io/quic_default_event_loop.h", "quiche/quic/core/io/quic_event_loop.h", "quiche/quic/core/io/quic_poll_event_loop.h", "quiche/quic/core/io/socket.h", + "quiche/quic/core/io/socket_factory.h", + "quiche/quic/core/io/stream_client_socket.h", "quiche/quic/core/quic_default_packet_writer.h", "quiche/quic/core/quic_epoll_alarm_factory.h", "quiche/quic/core/quic_epoll_clock.h", @@ -980,6 +984,8 @@ "quiche/quic/core/batch_writer/quic_batch_writer_buffer.cc", "quiche/quic/core/batch_writer/quic_gso_batch_writer.cc", "quiche/quic/core/batch_writer/quic_sendmmsg_batch_writer.cc", + "quiche/quic/core/io/event_loop_socket_factory.cc", + "quiche/quic/core/io/event_loop_tcp_client_socket.cc", "quiche/quic/core/io/quic_default_event_loop.cc", "quiche/quic/core/io/quic_poll_event_loop.cc", "quiche/quic/core/io/socket_posix.cc", @@ -1301,6 +1307,7 @@ "quiche/quic/core/http/quic_spdy_client_session_test.cc", "quiche/quic/core/http/quic_spdy_client_stream_test.cc", "quiche/quic/core/http/quic_spdy_server_stream_base_test.cc", + "quiche/quic/core/io/event_loop_tcp_client_socket_test.cc", "quiche/quic/core/io/quic_all_event_loops_test.cc", "quiche/quic/core/io/quic_poll_event_loop_test.cc", "quiche/quic/core/io/socket_test.cc",
diff --git a/quiche/quic/core/io/event_loop_socket_factory.cc b/quiche/quic/core/io/event_loop_socket_factory.cc new file mode 100644 index 0000000..4d2508e --- /dev/null +++ b/quiche/quic/core/io/event_loop_socket_factory.cc
@@ -0,0 +1,36 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/core/io/event_loop_socket_factory.h" + +#include <memory> + +#include "quiche/quic/core/io/event_loop_tcp_client_socket.h" +#include "quiche/quic/core/io/quic_event_loop.h" +#include "quiche/quic/core/io/stream_client_socket.h" +#include "quiche/quic/core/quic_types.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/quiche_buffer_allocator.h" + +namespace quic { + +EventLoopSocketFactory::EventLoopSocketFactory( + QuicEventLoop* event_loop, quiche::QuicheBufferAllocator* buffer_allocator) + : event_loop_(event_loop), buffer_allocator_(buffer_allocator) { + QUICHE_DCHECK(event_loop_); + QUICHE_DCHECK(buffer_allocator_); +} + +std::unique_ptr<StreamClientSocket> +EventLoopSocketFactory::CreateTcpClientSocket( + const quic::QuicSocketAddress& peer_address, + QuicByteCount receive_buffer_size, QuicByteCount send_buffer_size, + StreamClientSocket::AsyncVisitor* async_visitor) { + return std::make_unique<EventLoopTcpClientSocket>( + peer_address, receive_buffer_size, send_buffer_size, event_loop_, + buffer_allocator_, async_visitor); +} + +} // namespace quic
diff --git a/quiche/quic/core/io/event_loop_socket_factory.h b/quiche/quic/core/io/event_loop_socket_factory.h new file mode 100644 index 0000000..6882e80 --- /dev/null +++ b/quiche/quic/core/io/event_loop_socket_factory.h
@@ -0,0 +1,41 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_CORE_IO_EVENT_LOOP_SOCKET_FACTORY_H_ +#define QUICHE_QUIC_CORE_IO_EVENT_LOOP_SOCKET_FACTORY_H_ + +#include <memory> + +#include "quiche/quic/core/io/quic_event_loop.h" +#include "quiche/quic/core/io/socket_factory.h" +#include "quiche/quic/core/io/stream_client_socket.h" +#include "quiche/quic/core/quic_types.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/common/platform/api/quiche_export.h" +#include "quiche/common/quiche_buffer_allocator.h" + +namespace quic { + +// A socket factory that creates sockets implemented using an underlying +// QuicEventLoop. +class QUICHE_EXPORT_PRIVATE EventLoopSocketFactory : public SocketFactory { + public: + // `event_loop` and `buffer_allocator` must outlive the created factory. + EventLoopSocketFactory(QuicEventLoop* event_loop, + quiche::QuicheBufferAllocator* buffer_allocator); + + // SocketFactory: + std::unique_ptr<StreamClientSocket> CreateTcpClientSocket( + const quic::QuicSocketAddress& peer_address, + QuicByteCount receive_buffer_size, QuicByteCount send_buffer_size, + StreamClientSocket::AsyncVisitor* async_visitor) override; + + private: + QuicEventLoop* const event_loop_; // unowned + quiche::QuicheBufferAllocator* buffer_allocator_; // unowned +}; + +} // namespace quic + +#endif // QUICHE_QUIC_CORE_IO_EVENT_LOOP_SOCKET_FACTORY_H_
diff --git a/quiche/quic/core/io/event_loop_tcp_client_socket.cc b/quiche/quic/core/io/event_loop_tcp_client_socket.cc new file mode 100644 index 0000000..84263bc --- /dev/null +++ b/quiche/quic/core/io/event_loop_tcp_client_socket.cc
@@ -0,0 +1,610 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/core/io/event_loop_tcp_client_socket.h" + +#include <limits> +#include <string> +#include <utility> + +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/string_view.h" +#include "absl/types/span.h" +#include "absl/types/variant.h" +#include "quiche/quic/core/io/quic_event_loop.h" +#include "quiche/quic/core/io/socket.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/platform/api/quiche_mem_slice.h" + +namespace quic { + +EventLoopTcpClientSocket::EventLoopTcpClientSocket( + const quic::QuicSocketAddress& peer_address, + QuicByteCount receive_buffer_size, QuicByteCount send_buffer_size, + QuicEventLoop* event_loop, quiche::QuicheBufferAllocator* buffer_allocator, + AsyncVisitor* async_visitor) + : peer_address_(peer_address), + receive_buffer_size_(receive_buffer_size), + send_buffer_size_(send_buffer_size), + event_loop_(event_loop), + buffer_allocator_(buffer_allocator), + async_visitor_(async_visitor) { + QUICHE_DCHECK(event_loop_); + QUICHE_DCHECK(buffer_allocator_); +} + +EventLoopTcpClientSocket::~EventLoopTcpClientSocket() { + // Connected socket must be closed via Disconnect() before destruction. Cannot + // safely recover if state indicates caller may be expecting async callbacks. + QUICHE_DCHECK(connect_status_ != ConnectStatus::kConnecting); + QUICHE_DCHECK(!receive_max_size_.has_value()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + if (descriptor_ != kInvalidSocketFd) { + QUICHE_BUG(quic_event_loop_tcp_socket_invalid_destruction) + << "Must call Disconnect() on connected TCP socket before destruction."; + Close(); + } + + QUICHE_DCHECK(connect_status_ == ConnectStatus::kNotConnected); + QUICHE_DCHECK(send_remaining_.empty()); +} + +absl::Status EventLoopTcpClientSocket::ConnectBlocking() { + QUICHE_DCHECK_EQ(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kNotConnected); + QUICHE_DCHECK(!receive_max_size_.has_value()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + absl::Status status = Open(); + if (!status.ok()) { + return status; + } + + status = socket_api::SetSocketBlocking(descriptor_, /*blocking=*/true); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to set socket to address: " << peer_address_.ToString() + << " as blocking for connect with error: " << status; + Close(); + return status; + } + + status = DoInitialConnect(); + + if (absl::IsUnavailable(status)) { + QUICHE_LOG_FIRST_N(ERROR, 100) + << "Non-blocking connect to should-be blocking socket to address:" + << peer_address_.ToString() << "."; + Close(); + connect_status_ = ConnectStatus::kNotConnected; + return status; + } else if (!status.ok()) { + // DoInitialConnect() closes the socket on failures. + QUICHE_DCHECK_EQ(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kNotConnected); + return status; + } + + status = socket_api::SetSocketBlocking(descriptor_, /*blocking=*/false); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to return socket to address: " << peer_address_.ToString() + << " to non-blocking after connect with error: " << status; + Close(); + connect_status_ = ConnectStatus::kNotConnected; + } + + QUICHE_DCHECK(connect_status_ != ConnectStatus::kConnecting); + return status; +} + +void EventLoopTcpClientSocket::ConnectAsync() { + QUICHE_DCHECK(async_visitor_); + QUICHE_DCHECK_EQ(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kNotConnected); + QUICHE_DCHECK(!receive_max_size_.has_value()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + absl::Status status = Open(); + if (!status.ok()) { + async_visitor_->ConnectComplete(status); + return; + } + + FinishOrRearmAsyncConnect(DoInitialConnect()); +} + +void EventLoopTcpClientSocket::Disconnect() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ != ConnectStatus::kNotConnected); + + Close(); + QUICHE_DCHECK_EQ(descriptor_, kInvalidSocketFd); + + // Reset all state before invoking any callbacks. + bool require_connect_callback = connect_status_ == ConnectStatus::kConnecting; + connect_status_ = ConnectStatus::kNotConnected; + bool require_receive_callback = receive_max_size_.has_value(); + receive_max_size_.reset(); + bool require_send_callback = + !absl::holds_alternative<absl::monostate>(send_data_); + send_data_ = absl::monostate(); + send_remaining_ = ""; + + if (require_connect_callback) { + QUICHE_DCHECK(async_visitor_); + async_visitor_->ConnectComplete(absl::CancelledError()); + } + if (require_receive_callback) { + QUICHE_DCHECK(async_visitor_); + async_visitor_->ReceiveComplete(absl::CancelledError()); + } + if (require_send_callback) { + QUICHE_DCHECK(async_visitor_); + async_visitor_->SendComplete(absl::CancelledError()); + } +} + +absl::StatusOr<quiche::QuicheMemSlice> +EventLoopTcpClientSocket::ReceiveBlocking(QuicByteCount max_size) { + QUICHE_DCHECK_GT(max_size, 0u); + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + QUICHE_DCHECK(!receive_max_size_.has_value()); + + absl::Status status = + socket_api::SetSocketBlocking(descriptor_, /*blocking=*/true); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to set socket to address: " << peer_address_.ToString() + << " as blocking for receive with error: " << status; + return status; + } + + receive_max_size_ = max_size; + absl::StatusOr<quiche::QuicheMemSlice> buffer = ReceiveInternal(); + + if (!buffer.ok() && absl::IsUnavailable(buffer.status())) { + QUICHE_LOG_FIRST_N(ERROR, 100) + << "Non-blocking receive from should-be blocking socket to address:" + << peer_address_.ToString() << "."; + receive_max_size_.reset(); + } else { + QUICHE_DCHECK(!receive_max_size_.has_value()); + } + + absl::Status set_non_blocking_status = + socket_api::SetSocketBlocking(descriptor_, /*blocking=*/false); + if (!set_non_blocking_status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to return socket to address: " << peer_address_.ToString() + << " to non-blocking after receive with error: " + << set_non_blocking_status; + return set_non_blocking_status; + } + + return buffer; +} + +void EventLoopTcpClientSocket::ReceiveAsync(QuicByteCount max_size) { + QUICHE_DCHECK(async_visitor_); + QUICHE_DCHECK_GT(max_size, 0u); + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + QUICHE_DCHECK(!receive_max_size_.has_value()); + + receive_max_size_ = max_size; + + FinishOrRearmAsyncReceive(ReceiveInternal()); +} + +absl::Status EventLoopTcpClientSocket::SendBlocking(std::string data) { + QUICHE_DCHECK(!data.empty()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + send_data_ = std::move(data); + return SendBlockingInternal(); +} + +absl::Status EventLoopTcpClientSocket::SendBlocking( + quiche::QuicheMemSlice data) { + QUICHE_DCHECK(!data.empty()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + send_data_ = std::move(data); + return SendBlockingInternal(); +} + +void EventLoopTcpClientSocket::SendAsync(std::string data) { + QUICHE_DCHECK(!data.empty()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + send_data_ = std::move(data); + send_remaining_ = absl::get<std::string>(send_data_); + + FinishOrRearmAsyncSend(SendInternal()); +} + +void EventLoopTcpClientSocket::SendAsync(quiche::QuicheMemSlice data) { + QUICHE_DCHECK(!data.empty()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + send_data_ = std::move(data); + send_remaining_ = + absl::get<quiche::QuicheMemSlice>(send_data_).AsStringView(); + + FinishOrRearmAsyncSend(SendInternal()); +} + +void EventLoopTcpClientSocket::OnSocketEvent(QuicEventLoop* event_loop, + SocketFd fd, + QuicSocketEventMask events) { + QUICHE_DCHECK_EQ(event_loop, event_loop_); + QUICHE_DCHECK_EQ(fd, descriptor_); + + if (connect_status_ == ConnectStatus::kConnecting && + (events & (kSocketEventWritable | kSocketEventError))) { + FinishOrRearmAsyncConnect(GetConnectResult()); + return; + } + + if (receive_max_size_.has_value() && + (events & (kSocketEventReadable | kSocketEventError))) { + FinishOrRearmAsyncReceive(ReceiveInternal()); + } + if (!send_remaining_.empty() && + (events & (kSocketEventWritable | kSocketEventError))) { + FinishOrRearmAsyncSend(SendInternal()); + } +} + +absl::Status EventLoopTcpClientSocket::Open() { + QUICHE_DCHECK_EQ(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kNotConnected); + QUICHE_DCHECK(!receive_max_size_.has_value()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + QUICHE_DCHECK(send_remaining_.empty()); + + absl::StatusOr<SocketFd> descriptor = socket_api::CreateSocket( + peer_address_.host().address_family(), socket_api::SocketProtocol::kTcp, + /*blocking=*/false); + if (!descriptor.ok()) { + QUICHE_DVLOG(1) << "Failed to open socket for connection to address: " + << peer_address_.ToString() + << " with error: " << descriptor.status(); + return descriptor.status(); + } + QUICHE_DCHECK_NE(descriptor.value(), kInvalidSocketFd); + + descriptor_ = descriptor.value(); + + if (async_visitor_) { + bool registered; + if (event_loop_->SupportsEdgeTriggered()) { + registered = event_loop_->RegisterSocket( + descriptor_, + kSocketEventReadable | kSocketEventWritable | kSocketEventError, + this); + } else { + // Just register the socket without any armed events for now. Will rearm + // with specific events as needed. Registering now before events are + // needed makes it easier to ensure the socket is registered only once + // and can always be unregistered on socket close. + registered = event_loop_->RegisterSocket(descriptor_, /*events=*/0, this); + } + QUICHE_DCHECK(registered); + } + + if (receive_buffer_size_ != 0) { + absl::Status status = + socket_api::SetReceiveBufferSize(descriptor_, receive_buffer_size_); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to set receive buffer size to: " << receive_buffer_size_ + << " for socket to address: " << peer_address_.ToString() + << " with error: " << status; + Close(); + return status; + } + } + + if (send_buffer_size_ != 0) { + absl::Status status = + socket_api::SetSendBufferSize(descriptor_, send_buffer_size_); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to set send buffer size to: " << send_buffer_size_ + << " for socket to address: " << peer_address_.ToString() + << " with error: " << status; + Close(); + return status; + } + } + + return absl::OkStatus(); +} + +void EventLoopTcpClientSocket::Close() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + + bool unregistered = event_loop_->UnregisterSocket(descriptor_); + QUICHE_DCHECK_EQ(unregistered, !!async_visitor_); + + absl::Status status = socket_api::Close(descriptor_); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Could not close socket to address: " << peer_address_.ToString() + << " with error: " << status; + } + + descriptor_ = kInvalidSocketFd; +} + +absl::Status EventLoopTcpClientSocket::DoInitialConnect() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kNotConnected); + QUICHE_DCHECK(!receive_max_size_.has_value()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + absl::Status connect_result = socket_api::Connect(descriptor_, peer_address_); + + if (connect_result.ok()) { + connect_status_ = ConnectStatus::kConnected; + } else if (absl::IsUnavailable(connect_result)) { + connect_status_ = ConnectStatus::kConnecting; + } else { + QUICHE_DVLOG(1) << "Synchronously failed to connect socket to address: " + << peer_address_.ToString() + << " with error: " << connect_result; + Close(); + connect_status_ = ConnectStatus::kNotConnected; + } + + return connect_result; +} + +absl::Status EventLoopTcpClientSocket::GetConnectResult() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnecting); + QUICHE_DCHECK(!receive_max_size_.has_value()); + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + + absl::Status error = socket_api::GetSocketError(descriptor_); + + if (!error.ok()) { + QUICHE_DVLOG(1) << "Asynchronously failed to connect socket to address: " + << peer_address_.ToString() << " with error: " << error; + Close(); + connect_status_ = ConnectStatus::kNotConnected; + return error; + } + + // Peek at one byte to confirm the connection is actually alive. Motivation: + // 1) Plausibly could have a lot of cases where the connection operation + // itself technically succeeds but the socket then quickly fails. Don't + // want to claim connection success here if, by the time this code is + // running after event triggers and such, the socket has already failed. + // Lot of undefined room around whether or not such errors would be saved + // into SO_ERROR and returned by socket_api::GetSocketError(). + // 2) With the various platforms and event systems involved, less than 100% + // trust that it's impossible to end up in this method before the async + // connect has completed/errored. Given that Connect() and GetSocketError() + // does not difinitevely differentiate between success and + // still-in-progress, and given that there's a very simple and performant + // way to positively confirm the socket is connected (peek), do that here. + // (Could consider making the not-connected case a QUIC_BUG if a way is + // found to differentiate it from (1).) + absl::StatusOr<bool> peek_data = OneBytePeek(); + if (peek_data.ok() || absl::IsUnavailable(peek_data.status())) { + connect_status_ = ConnectStatus::kConnected; + } else { + error = peek_data.status(); + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Socket to address: " << peer_address_.ToString() + << " signalled writable after connect and no connect error found, " + "but socket does not appear connected with error: " + << error; + Close(); + connect_status_ = ConnectStatus::kNotConnected; + } + + return error; +} + +void EventLoopTcpClientSocket::FinishOrRearmAsyncConnect(absl::Status status) { + if (absl::IsUnavailable(status)) { + if (!event_loop_->SupportsEdgeTriggered()) { + bool result = event_loop_->RearmSocket( + descriptor_, kSocketEventWritable | kSocketEventError); + QUICHE_DCHECK(result); + } + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnecting); + } else { + QUICHE_DCHECK(connect_status_ != ConnectStatus::kConnecting); + async_visitor_->ConnectComplete(status); + } +} + +absl::StatusOr<quiche::QuicheMemSlice> +EventLoopTcpClientSocket::ReceiveInternal() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + QUICHE_CHECK(receive_max_size_.has_value()); + QUICHE_DCHECK_GE(receive_max_size_.value(), 1u); + QUICHE_DCHECK_LE(receive_max_size_.value(), + std::numeric_limits<size_t>::max()); + + // Before allocating a buffer, do a 1-byte peek to determine if needed. + if (receive_max_size_.value() > 1) { + absl::StatusOr<bool> peek_data = OneBytePeek(); + if (!peek_data.ok()) { + if (!absl::IsUnavailable(peek_data.status())) { + receive_max_size_.reset(); + } + return peek_data.status(); + } else if (!peek_data.value()) { + receive_max_size_.reset(); + return quiche::QuicheMemSlice(); + } + } + + quiche::QuicheBuffer buffer(buffer_allocator_, receive_max_size_.value()); + absl::StatusOr<absl::Span<char>> received = socket_api::Receive( + descriptor_, absl::MakeSpan(buffer.data(), buffer.size())); + + if (received.ok()) { + QUICHE_DCHECK_LE(received.value().size(), buffer.size()); + QUICHE_DCHECK_EQ(received.value().data(), buffer.data()); + + receive_max_size_.reset(); + return quiche::QuicheMemSlice( + quiche::QuicheBuffer(buffer.Release(), received.value().size())); + } else { + if (!absl::IsUnavailable(received.status())) { + QUICHE_DVLOG(1) << "Failed to receive from socket to address: " + << peer_address_.ToString() + << " with error: " << received.status(); + receive_max_size_.reset(); + } + return received.status(); + } +} + +void EventLoopTcpClientSocket::FinishOrRearmAsyncReceive( + absl::StatusOr<quiche::QuicheMemSlice> buffer) { + QUICHE_DCHECK(async_visitor_); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + + if (!buffer.ok() && absl::IsUnavailable(buffer.status())) { + if (!event_loop_->SupportsEdgeTriggered()) { + bool result = event_loop_->RearmSocket( + descriptor_, kSocketEventReadable | kSocketEventError); + QUICHE_DCHECK(result); + } + QUICHE_DCHECK(receive_max_size_.has_value()); + } else { + QUICHE_DCHECK(!receive_max_size_.has_value()); + async_visitor_->ReceiveComplete(std::move(buffer)); + } +} + +absl::StatusOr<bool> EventLoopTcpClientSocket::OneBytePeek() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + + char peek_buffer; + absl::StatusOr<absl::Span<char>> peek_received = socket_api::Receive( + descriptor_, absl::MakeSpan(&peek_buffer, /*size=*/1), /*peek=*/true); + if (!peek_received.ok()) { + return peek_received.status(); + } else { + return !peek_received.value().empty(); + } +} + +absl::Status EventLoopTcpClientSocket::SendBlockingInternal() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + QUICHE_DCHECK(!absl::holds_alternative<absl::monostate>(send_data_)); + QUICHE_DCHECK(send_remaining_.empty()); + + absl::Status status = + socket_api::SetSocketBlocking(descriptor_, /*blocking=*/true); + if (!status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to set socket to address: " << peer_address_.ToString() + << " as blocking for send with error: " << status; + send_data_ = absl::monostate(); + return status; + } + + if (absl::holds_alternative<std::string>(send_data_)) { + send_remaining_ = absl::get<std::string>(send_data_); + } else { + send_remaining_ = + absl::get<quiche::QuicheMemSlice>(send_data_).AsStringView(); + } + + status = SendInternal(); + if (absl::IsUnavailable(status)) { + QUICHE_LOG_FIRST_N(ERROR, 100) + << "Non-blocking send for should-be blocking socket to address:" + << peer_address_.ToString(); + send_data_ = absl::monostate(); + send_remaining_ = ""; + } else { + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + QUICHE_DCHECK(send_remaining_.empty()); + } + + absl::Status set_non_blocking_status = + socket_api::SetSocketBlocking(descriptor_, /*blocking=*/false); + if (!set_non_blocking_status.ok()) { + QUICHE_LOG_FIRST_N(WARNING, 100) + << "Failed to return socket to address: " << peer_address_.ToString() + << " to non-blocking after send with error: " + << set_non_blocking_status; + return set_non_blocking_status; + } + + return status; +} + +absl::Status EventLoopTcpClientSocket::SendInternal() { + QUICHE_DCHECK_NE(descriptor_, kInvalidSocketFd); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + QUICHE_DCHECK(!absl::holds_alternative<absl::monostate>(send_data_)); + QUICHE_DCHECK(!send_remaining_.empty()); + + // Repeat send until all data sent, unavailable, or error. + while (!send_remaining_.empty()) { + absl::StatusOr<absl::string_view> remainder = + socket_api::Send(descriptor_, send_remaining_); + + if (remainder.ok()) { + QUICHE_DCHECK(remainder.value().empty() || + (remainder.value().data() >= send_remaining_.data() && + remainder.value().data() < + send_remaining_.data() + send_remaining_.size())); + QUICHE_DCHECK(remainder.value().empty() || + (remainder.value().data() + remainder.value().size() == + send_remaining_.data() + send_remaining_.size())); + send_remaining_ = remainder.value(); + } else { + if (!absl::IsUnavailable(remainder.status())) { + QUICHE_DVLOG(1) << "Failed to send to socket to address: " + << peer_address_.ToString() + << " with error: " << remainder.status(); + send_data_ = absl::monostate(); + send_remaining_ = ""; + } + return remainder.status(); + } + } + + send_data_ = absl::monostate(); + return absl::OkStatus(); +} + +void EventLoopTcpClientSocket::FinishOrRearmAsyncSend(absl::Status status) { + QUICHE_DCHECK(async_visitor_); + QUICHE_DCHECK(connect_status_ == ConnectStatus::kConnected); + + if (absl::IsUnavailable(status)) { + if (!event_loop_->SupportsEdgeTriggered()) { + bool result = event_loop_->RearmSocket( + descriptor_, kSocketEventWritable | kSocketEventError); + QUICHE_DCHECK(result); + } + QUICHE_DCHECK(!absl::holds_alternative<absl::monostate>(send_data_)); + QUICHE_DCHECK(!send_remaining_.empty()); + } else { + QUICHE_DCHECK(absl::holds_alternative<absl::monostate>(send_data_)); + QUICHE_DCHECK(send_remaining_.empty()); + async_visitor_->SendComplete(status); + } +} + +} // namespace quic
diff --git a/quiche/quic/core/io/event_loop_tcp_client_socket.h b/quiche/quic/core/io/event_loop_tcp_client_socket.h new file mode 100644 index 0000000..2d50087 --- /dev/null +++ b/quiche/quic/core/io/event_loop_tcp_client_socket.h
@@ -0,0 +1,103 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_CORE_IO_EVENT_LOOP_TCP_CLIENT_SOCKET_H_ +#define QUICHE_QUIC_CORE_IO_EVENT_LOOP_TCP_CLIENT_SOCKET_H_ + +#include <string> + +#include "absl/status/status.h" +#include "absl/strings/string_view.h" +#include "absl/types/optional.h" +#include "absl/types/variant.h" +#include "quiche/quic/core/io/quic_event_loop.h" +#include "quiche/quic/core/io/stream_client_socket.h" +#include "quiche/quic/core/quic_types.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/common/platform/api/quiche_export.h" +#include "quiche/common/quiche_buffer_allocator.h" + +namespace quic { + +// A TCP client socket implemented using an underlying QuicEventLoop. +class QUICHE_EXPORT_PRIVATE EventLoopTcpClientSocket + : public StreamClientSocket, + public QuicSocketEventListener { + public: + // Will use platform default buffer size if `receive_buffer_size` or + // `send_buffer_size` is zero. `async_visitor` may be null if no async + // operations will be requested. `event_loop`, `buffer_allocator`, and + // `async_visitor` (if non-null) must outlive the created socket. + EventLoopTcpClientSocket(const quic::QuicSocketAddress& peer_address, + QuicByteCount receive_buffer_size, + QuicByteCount send_buffer_size, + QuicEventLoop* event_loop, + quiche::QuicheBufferAllocator* buffer_allocator, + AsyncVisitor* async_visitor); + + ~EventLoopTcpClientSocket() override; + + // StreamClientSocket: + absl::Status ConnectBlocking() override; + void ConnectAsync() override; + void Disconnect() override; + + // Socket: + absl::StatusOr<quiche::QuicheMemSlice> ReceiveBlocking( + QuicByteCount max_size) override; + void ReceiveAsync(QuicByteCount max_size) override; + absl::Status SendBlocking(std::string data) override; + absl::Status SendBlocking(quiche::QuicheMemSlice data) override; + void SendAsync(std::string data) override; + void SendAsync(quiche::QuicheMemSlice data) override; + + // QuicSocketEventListener: + void OnSocketEvent(QuicEventLoop* event_loop, SocketFd fd, + QuicSocketEventMask events) override; + + private: + enum class ConnectStatus { + kNotConnected, + kConnecting, + kConnected, + }; + + absl::Status Open(); + void Close(); + absl::Status DoInitialConnect(); + absl::Status GetConnectResult(); + void FinishOrRearmAsyncConnect(absl::Status status); + absl::StatusOr<quiche::QuicheMemSlice> ReceiveInternal(); + void FinishOrRearmAsyncReceive(absl::StatusOr<quiche::QuicheMemSlice> buffer); + // Returns `true` if a byte received, or `false` if successfully received + // empty data. + absl::StatusOr<bool> OneBytePeek(); + absl::Status SendBlockingInternal(); + absl::Status SendInternal(); + void FinishOrRearmAsyncSend(absl::Status status); + + const QuicSocketAddress peer_address_; + const QuicByteCount receive_buffer_size_; + const QuicByteCount send_buffer_size_; + QuicEventLoop* const event_loop_; // unowned + quiche::QuicheBufferAllocator* buffer_allocator_; // unowned + AsyncVisitor* const async_visitor_; // unowned, potentially null + + SocketFd descriptor_ = kInvalidSocketFd; + ConnectStatus connect_status_ = ConnectStatus::kNotConnected; + + // Only set while receive in progress or pending, otherwise nullopt. + absl::optional<QuicByteCount> receive_max_size_; + + // Only contains data while send in progress or pending, otherwise monostate. + absl::variant<absl::monostate, std::string, quiche::QuicheMemSlice> + send_data_; + // Points to the unsent portion of `send_data_` while send in progress or + // pending, otherwise empty. + absl::string_view send_remaining_; +}; + +} // namespace quic + +#endif // QUICHE_QUIC_CORE_IO_EVENT_LOOP_TCP_CLIENT_SOCKET_H_
diff --git a/quiche/quic/core/io/event_loop_tcp_client_socket_test.cc b/quiche/quic/core/io/event_loop_tcp_client_socket_test.cc new file mode 100644 index 0000000..f3c0800 --- /dev/null +++ b/quiche/quic/core/io/event_loop_tcp_client_socket_test.cc
@@ -0,0 +1,523 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#include "quiche/quic/core/io/event_loop_tcp_client_socket.h" + +#include <functional> +#include <memory> +#include <utility> +#include <vector> + +#include "absl/functional/bind_front.h" +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/string_view.h" +#include "absl/types/optional.h" +#include "absl/types/span.h" +#include "quiche/quic/core/io/event_loop_socket_factory.h" +#include "quiche/quic/core/io/quic_default_event_loop.h" +#include "quiche/quic/core/io/quic_event_loop.h" +#include "quiche/quic/core/io/socket.h" +#include "quiche/quic/core/io/stream_client_socket.h" +#include "quiche/quic/core/quic_time.h" +#include "quiche/quic/platform/api/quic_ip_address_family.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/quic/test_tools/mock_clock.h" +#include "quiche/quic/test_tools/quic_test_utils.h" +#include "quiche/common/platform/api/quiche_logging.h" +#include "quiche/common/platform/api/quiche_mem_slice.h" +#include "quiche/common/platform/api/quiche_mutex.h" +#include "quiche/common/platform/api/quiche_test.h" +#include "quiche/common/platform/api/quiche_test_loopback.h" +#include "quiche/common/platform/api/quiche_thread.h" +#include "quiche/common/simple_buffer_allocator.h" + +namespace quic::test { +namespace { + +bool CreateListeningServerSocket(SocketFd* out_socket_descriptor, + QuicSocketAddress* out_socket_address) { + QUICHE_CHECK(out_socket_descriptor); + QUICHE_CHECK(out_socket_address); + + absl::StatusOr<SocketFd> socket = socket_api::CreateSocket( + quiche::TestLoopback().address_family(), socket_api::SocketProtocol::kTcp, + /*blocking=*/true); + QUICHE_CHECK(socket.ok()); + + // Set an extremely small receive buffer size to increase the odds of buffers + // filling up when testing asynchronous writes. + static const QuicByteCount kReceiveBufferSize = 2; + absl::Status result = + socket_api::SetReceiveBufferSize(socket.value(), kReceiveBufferSize); + QUICHE_CHECK(result.ok()); + + QuicSocketAddress bind_address(quiche::TestLoopback(), /*port=*/0); + result = socket_api::Bind(socket.value(), bind_address); + QUICHE_CHECK(result.ok()); + + absl::StatusOr<QuicSocketAddress> socket_address = + socket_api::GetSocketAddress(socket.value()); + QUICHE_CHECK(socket_address.ok()); + + result = socket_api::Listen(socket.value(), /*backlog=*/1); + QUICHE_CHECK(result.ok()); + + *out_socket_descriptor = socket.value(); + *out_socket_address = std::move(socket_address).value(); + return true; +} + +class TestTcpServerSocketRunner : public quiche::QuicheThread { + public: + using SocketBehavior = std::function<void(SocketFd connected_socket)>; + + // On construction, spins a separate thread to accept a connection from + // `server_socket_descriptor`, runs `behavior` with that connection, and then + // closes the accepted connection socket. If `allow_accept_failure` is true, + // will silently stop if an error is encountered accepting the connection. + TestTcpServerSocketRunner(SocketFd server_socket_descriptor, + SocketBehavior behavior, + bool allow_accept_failure = false) + : QuicheThread("TestTcpServerSocketRunner"), + server_socket_descriptor_(server_socket_descriptor), + behavior_(std::move(behavior)), + allow_accept_failure_(allow_accept_failure) { + Start(); + } + + ~TestTcpServerSocketRunner() override { WaitForCompletion(); } + + void WaitForCompletion() { completion_notification_.WaitForNotification(); } + + protected: + void Run() override { + if (AcceptSocket()) { + behavior_(connection_socket_descriptor_); + CloseSocket(); + } else { + QUICHE_CHECK(allow_accept_failure_); + } + + completion_notification_.Notify(); + } + + private: + bool AcceptSocket() { + absl::StatusOr<socket_api::AcceptResult> connection_socket = + socket_api::Accept(server_socket_descriptor_, /*blocking=*/true); + if (connection_socket.ok()) { + connection_socket_descriptor_ = connection_socket.value().fd; + } + return connection_socket.ok(); + } + + void CloseSocket() { + QUICHE_CHECK(socket_api::Close(connection_socket_descriptor_).ok()); + } + + const SocketFd server_socket_descriptor_; + const SocketBehavior behavior_; + const bool allow_accept_failure_; + + SocketFd connection_socket_descriptor_; + + quiche::QuicheNotification completion_notification_; +}; + +class EventLoopTcpClientSocketTest + : public quiche::test::QuicheTestWithParam<QuicEventLoopFactory*>, + public StreamClientSocket::AsyncVisitor { + public: + void SetUp() override { + QUICHE_CHECK(CreateListeningServerSocket(&server_socket_descriptor_, + &server_socket_address_)); + } + + void TearDown() override { + if (server_socket_descriptor_ != kInvalidSocketFd) { + QUICHE_CHECK(socket_api::Close(server_socket_descriptor_).ok()); + } + } + + void ConnectComplete(absl::Status status) override { + QUICHE_CHECK(!connect_result_.has_value()); + connect_result_ = std::move(status); + } + + void ReceiveComplete(absl::StatusOr<quiche::QuicheMemSlice> data) override { + QUICHE_CHECK(!receive_result_.has_value()); + receive_result_ = std::move(data); + } + + void SendComplete(absl::Status status) override { + QUICHE_CHECK(!send_result_.has_value()); + send_result_ = std::move(status); + } + + protected: + SocketFd server_socket_descriptor_ = kInvalidSocketFd; + QuicSocketAddress server_socket_address_; + + MockClock clock_; + std::unique_ptr<QuicEventLoop> event_loop_ = GetParam()->Create(&clock_); + EventLoopSocketFactory socket_factory_{event_loop_.get(), + quiche::SimpleBufferAllocator::Get()}; + + absl::optional<absl::Status> connect_result_; + absl::optional<absl::StatusOr<quiche::QuicheMemSlice>> receive_result_; + absl::optional<absl::Status> send_result_; +}; + +std::string GetTestParamName( + ::testing::TestParamInfo<QuicEventLoopFactory*> info) { + return EscapeTestParamName(info.param->GetName()); +} + +INSTANTIATE_TEST_SUITE_P(EventLoopTcpClientSocketTests, + EventLoopTcpClientSocketTest, + ::testing::ValuesIn(GetAllSupportedEventLoops()), + &GetTestParamName); + +TEST_P(EventLoopTcpClientSocketTest, Connect) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/nullptr); + + // No socket runner to accept the connection for the server, but that is not + // expected to be necessary for the connection to complete from the client. + EXPECT_TRUE(socket->ConnectBlocking().ok()); + + socket->Disconnect(); +} + +TEST_P(EventLoopTcpClientSocketTest, ConnectAsync) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/this); + + socket->ConnectAsync(); + + // Synchronous completion not normally expected, but since there is no known + // way to delay the server side of the connection (the OS does not wait for + // an accept() call), cannot be gauranteed that the connection will always + // complete asynchronously. If connecting asynchronously (normal behavior), + // expect completion once signalled by the event loop. + if (!connect_result_.has_value()) { + event_loop_->RunEventLoopOnce(QuicTime::Delta::FromSeconds(1)); + ASSERT_TRUE(connect_result_.has_value()); + } + EXPECT_TRUE(connect_result_.value().ok()); + + connect_result_.reset(); + socket->Disconnect(); + EXPECT_FALSE(connect_result_.has_value()); +} + +TEST_P(EventLoopTcpClientSocketTest, ErrorBeforeConnectAsync) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/this); + + // Close the server socket. + EXPECT_TRUE(socket_api::Close(server_socket_descriptor_).ok()); + server_socket_descriptor_ = kInvalidSocketFd; + + socket->ConnectAsync(); + if (!connect_result_.has_value()) { + event_loop_->RunEventLoopOnce(QuicTime::Delta::FromSeconds(1)); + ASSERT_TRUE(connect_result_.has_value()); + } + + // Expect an error because server socket was closed before connection. + EXPECT_FALSE(connect_result_.value().ok()); +} + +TEST_P(EventLoopTcpClientSocketTest, ErrorDuringConnectAsync) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/this); + + socket->ConnectAsync(); + + if (connect_result_.has_value()) { + // Not typical, but theoretically nothing to stop the connection from + // completing before the server socket is closed to trigger the error. + EXPECT_TRUE(connect_result_.value().ok()); + return; + } + + // Close the server socket. + EXPECT_TRUE(socket_api::Close(server_socket_descriptor_).ok()); + server_socket_descriptor_ = kInvalidSocketFd; + + // Expect an error once signalled. + EXPECT_FALSE(connect_result_.has_value()); + event_loop_->RunEventLoopOnce(QuicTime::Delta::FromSeconds(1)); + ASSERT_TRUE(connect_result_.has_value()); + EXPECT_FALSE(connect_result_.value().ok()); +} + +TEST_P(EventLoopTcpClientSocketTest, Disconnect) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/nullptr); + + ASSERT_TRUE(socket->ConnectBlocking().ok()); + socket->Disconnect(); +} + +TEST_P(EventLoopTcpClientSocketTest, DisconnectCancelsConnectAsync) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/this); + + socket->ConnectAsync(); + + if (connect_result_.has_value()) { + // Not typical, but theoretically nothing to stop the connection from + // completing before the server socket is closed to trigger the error. + EXPECT_TRUE(connect_result_.value().ok()); + return; + } + + socket->Disconnect(); + + // Expect immediate cancelled error. + ASSERT_TRUE(connect_result_.has_value()); + EXPECT_TRUE(absl::IsCancelled(connect_result_.value())); +} + +TEST_P(EventLoopTcpClientSocketTest, ConnectAndReconnect) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/nullptr); + + ASSERT_TRUE(socket->ConnectBlocking().ok()); + socket->Disconnect(); + + // Expect `socket` can reconnect now that it has been disconnected. + EXPECT_TRUE(socket->ConnectBlocking().ok()); + socket->Disconnect(); +} + +void SendDataOnSocket(absl::string_view data, SocketFd connected_socket) { + while (!data.empty()) { + absl::StatusOr<absl::string_view> remainder = + socket_api::Send(connected_socket, data); + if (!remainder.ok()) { + return; + } + data = remainder.value(); + } +} + +TEST_P(EventLoopTcpClientSocketTest, Receive) { + std::string expected = {1, 2, 3, 4, 5, 6, 7, 8}; + TestTcpServerSocketRunner runner( + server_socket_descriptor_, absl::bind_front(&SendDataOnSocket, expected)); + + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/nullptr); + ASSERT_TRUE(socket->ConnectBlocking().ok()); + + std::string received; + absl::StatusOr<quiche::QuicheMemSlice> data; + do { + data = socket->ReceiveBlocking(100); + ASSERT_TRUE(data.ok()); + received.append(data.value().data(), data.value().length()); + } while (!data.value().empty()); + EXPECT_EQ(received, expected); + + socket->Disconnect(); +} + +TEST_P(EventLoopTcpClientSocketTest, ReceiveAsync) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/this); + ASSERT_TRUE(socket->ConnectBlocking().ok()); + + // Start an async receive. Expect no immediate results because runner not yet + // setup to accept and send. + socket->ReceiveAsync(100); + EXPECT_FALSE(receive_result_.has_value()); + + // Send data from server. + std::string expected = {1, 2, 3, 4, 5, 6, 7, 8}; + TestTcpServerSocketRunner runner( + server_socket_descriptor_, absl::bind_front(&SendDataOnSocket, expected)); + EXPECT_FALSE(receive_result_.has_value()); + for (int i = 0; i < 5 && !receive_result_.has_value(); ++i) { + event_loop_->RunEventLoopOnce(QuicTime::Delta::FromSeconds(1)); + } + + // Expect to receive at least some of the sent data. + ASSERT_TRUE(receive_result_.has_value()); + ASSERT_TRUE(receive_result_.value().ok()); + EXPECT_FALSE(receive_result_.value().value().empty()); + std::string received(receive_result_.value().value().data(), + receive_result_.value().value().length()); + + // Get any remaining data via blocking calls. + absl::StatusOr<quiche::QuicheMemSlice> data; + do { + data = socket->ReceiveBlocking(100); + ASSERT_TRUE(data.ok()); + received.append(data.value().data(), data.value().length()); + } while (!data.value().empty()); + + EXPECT_EQ(received, expected); + + receive_result_.reset(); + socket->Disconnect(); + EXPECT_FALSE(receive_result_.has_value()); +} + +TEST_P(EventLoopTcpClientSocketTest, DisconnectCancelsReceiveAsync) { + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/this); + + ASSERT_TRUE(socket->ConnectBlocking().ok()); + + // Start an asynchronous read, expecting no completion because server never + // sends any data. + socket->ReceiveAsync(100); + EXPECT_FALSE(receive_result_.has_value()); + + // Disconnect and expect an immediate cancelled error. + socket->Disconnect(); + ASSERT_TRUE(receive_result_.has_value()); + ASSERT_FALSE(receive_result_.value().ok()); + EXPECT_TRUE(absl::IsCancelled(receive_result_.value().status())); +} + +// Receive from `connected_socket` until connection is closed, writing received +// data to `out_received`. +void ReceiveDataFromSocket(std::string* out_received, + SocketFd connected_socket) { + out_received->clear(); + + std::string buffer(100, 0); + absl::StatusOr<absl::Span<char>> received; + do { + received = socket_api::Receive(connected_socket, absl::MakeSpan(buffer)); + QUICHE_CHECK(received.ok()); + out_received->insert(out_received->end(), received.value().begin(), + received.value().end()); + } while (!received.value().empty()); +} + +TEST_P(EventLoopTcpClientSocketTest, Send) { + std::string sent; + TestTcpServerSocketRunner runner( + server_socket_descriptor_, + absl::bind_front(&ReceiveDataFromSocket, &sent)); + + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/0, + /*async_visitor=*/nullptr); + ASSERT_TRUE(socket->ConnectBlocking().ok()); + + std::string expected = {1, 2, 3, 4, 5, 6, 7, 8}; + EXPECT_TRUE(socket->SendBlocking(expected).ok()); + socket->Disconnect(); + + runner.WaitForCompletion(); + EXPECT_EQ(sent, expected); +} + +TEST_P(EventLoopTcpClientSocketTest, SendAsync) { + // Use a small send buffer to improve chances of a send needing to be + // asynchronous. + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/4, + /*async_visitor=*/this); + ASSERT_TRUE(socket->ConnectBlocking().ok()); + + std::string data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; + std::string expected; + + // Repeatedly write to socket until it does not complete synchronously. + do { + expected.insert(expected.end(), data.begin(), data.end()); + send_result_.reset(); + socket->SendAsync(data); + ASSERT_TRUE(!send_result_.has_value() || send_result_.value().ok()); + } while (send_result_.has_value()); + + // Begin receiving from server and expect more data to send. + std::string sent; + TestTcpServerSocketRunner runner( + server_socket_descriptor_, + absl::bind_front(&ReceiveDataFromSocket, &sent)); + EXPECT_FALSE(send_result_.has_value()); + for (int i = 0; i < 5 && !send_result_.has_value(); ++i) { + event_loop_->RunEventLoopOnce(QuicTime::Delta::FromSeconds(1)); + } + ASSERT_TRUE(send_result_.has_value()); + EXPECT_TRUE(send_result_.value().ok()); + + send_result_.reset(); + socket->Disconnect(); + EXPECT_FALSE(send_result_.has_value()); + + runner.WaitForCompletion(); + EXPECT_EQ(sent, expected); +} + +TEST_P(EventLoopTcpClientSocketTest, DisconnectCancelsSendAsync) { + // Use a small send buffer to improve chances of a send needing to be + // asynchronous. + std::unique_ptr<StreamClientSocket> socket = + socket_factory_.CreateTcpClientSocket(server_socket_address_, + /*receive_buffer_size=*/0, + /*send_buffer_size=*/4, + /*async_visitor=*/this); + ASSERT_TRUE(socket->ConnectBlocking().ok()); + + std::string data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; + + // Repeatedly write to socket until it does not complete synchronously. + do { + send_result_.reset(); + socket->SendAsync(data); + ASSERT_TRUE(!send_result_.has_value() || send_result_.value().ok()); + } while (send_result_.has_value()); + + // Disconnect and expect immediate cancelled error. + socket->Disconnect(); + ASSERT_TRUE(send_result_.has_value()); + EXPECT_TRUE(absl::IsCancelled(send_result_.value())); +} + +} // namespace +} // namespace quic::test
diff --git a/quiche/quic/core/io/socket.h b/quiche/quic/core/io/socket.h index 748de4d..fcc6418 100644 --- a/quiche/quic/core/io/socket.h +++ b/quiche/quic/core/io/socket.h
@@ -5,6 +5,9 @@ #ifndef QUICHE_QUIC_CORE_IO_SOCKET_H_ #define QUICHE_QUIC_CORE_IO_SOCKET_H_ +#include <functional> +#include <string> + #include "absl/status/status.h" #include "absl/status/statusor.h" #include "absl/strings/string_view.h" @@ -13,6 +16,7 @@ #include "quiche/quic/platform/api/quic_ip_address_family.h" #include "quiche/quic/platform/api/quic_socket_address.h" #include "quiche/common/platform/api/quiche_export.h" +#include "quiche/common/platform/api/quiche_mem_slice.h" #if defined(_WIN32) #include <winsock2.h> @@ -28,6 +32,57 @@ inline constexpr SocketFd kInvalidSocketFd = -1; #endif +// A read/write socket. +// +// Warning regarding blocking calls: Code in the QUICHE library typically +// handles IO on a single thread, so if making calls from that typical +// environment, it would be problematic to make a blocking call and block that +// single thread. +class QUICHE_EXPORT_PRIVATE Socket { + public: + class AsyncVisitor { + public: + virtual ~AsyncVisitor() = default; + + // If the operation completed without error, `data` is set to the received + // data. + virtual void ReceiveComplete( + absl::StatusOr<quiche::QuicheMemSlice> data) = 0; + + virtual void SendComplete(absl::Status status) = 0; + }; + + virtual ~Socket() = default; + + // Blocking read. Receives and returns a buffer of up to `max_size` bytes from + // socket. Returns status on error. + virtual absl::StatusOr<quiche::QuicheMemSlice> ReceiveBlocking( + QuicByteCount max_size) = 0; + + // Asynchronous read. Receives up to `max_size` bytes from socket. If + // no data is synchronously available to be read, waits until some data is + // available or the socket is closed. On completion, calls ReceiveComplete() + // on the visitor, potentially before return from ReceiveAsync(). + // + // After calling, the socket must not be destroyed until ReceiveComplete() is + // called. + virtual void ReceiveAsync(QuicByteCount max_size) = 0; + + // Blocking write. Sends all of `data` (potentially via multiple underlying + // socket sends). + virtual absl::Status SendBlocking(std::string data) = 0; + virtual absl::Status SendBlocking(quiche::QuicheMemSlice data) = 0; + + // Asynchronous write. Sends all of `data` (potentially via multiple + // underlying socket sends). On completion, calls SendComplete() on the + // visitor, potentially before return from SendAsync(). + // + // After calling, the socket must not be destroyed until SendComplete() is + // called. + virtual void SendAsync(std::string data) = 0; + virtual void SendAsync(quiche::QuicheMemSlice data) = 0; +}; + // Low-level platform-agnostic socket operations. Closely follows the behavior // of basic POSIX socket APIs, diverging mostly only to convert to/from cleaner // and platform-agnostic types.
diff --git a/quiche/quic/core/io/socket_factory.h b/quiche/quic/core/io/socket_factory.h new file mode 100644 index 0000000..39102f1 --- /dev/null +++ b/quiche/quic/core/io/socket_factory.h
@@ -0,0 +1,34 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_CORE_IO_SOCKET_FACTORY_H_ +#define QUICHE_QUIC_CORE_IO_SOCKET_FACTORY_H_ + +#include <memory> + +#include "quiche/quic/core/io/stream_client_socket.h" +#include "quiche/quic/core/quic_types.h" +#include "quiche/quic/platform/api/quic_socket_address.h" +#include "quiche/common/platform/api/quiche_export.h" + +namespace quic { + +// A factory to create objects of type Socket and derived interfaces. +class QUICHE_EXPORT_PRIVATE SocketFactory { + public: + virtual ~SocketFactory() = default; + + // Will use platform default buffer size if `receive_buffer_size` or + // `send_buffer_size` is zero. If `async_visitor` is null, async operations + // must not be called on the created socket. If `async_visitor` is non-null, + // it must outlive the created socket. + virtual std::unique_ptr<StreamClientSocket> CreateTcpClientSocket( + const quic::QuicSocketAddress& peer_address, + QuicByteCount receive_buffer_size, QuicByteCount send_buffer_size, + StreamClientSocket::AsyncVisitor* async_visitor) = 0; +}; + +} // namespace quic + +#endif // QUICHE_QUIC_CORE_IO_SOCKET_FACTORY_H_
diff --git a/quiche/quic/core/io/stream_client_socket.h b/quiche/quic/core/io/stream_client_socket.h new file mode 100644 index 0000000..00f9a42 --- /dev/null +++ b/quiche/quic/core/io/stream_client_socket.h
@@ -0,0 +1,62 @@ +// Copyright 2022 The Chromium Authors. All rights reserved. +// Use of this source code is governed by a BSD-style license that can be +// found in the LICENSE file. + +#ifndef QUICHE_QUIC_CORE_IO_STREAM_CLIENT_SOCKET_H_ +#define QUICHE_QUIC_CORE_IO_STREAM_CLIENT_SOCKET_H_ + +#include "absl/status/status.h" +#include "quiche/quic/core/io/socket.h" +#include "quiche/common/platform/api/quiche_export.h" + +namespace quic { + +// A client socket using a protocol (typically TCP) that provides +// connection-based streams. +// +// Must not destroy a connected/connecting socket. If connected or connecting, +// must call Disconnect() to disconnect or cancel the connection before +// destruction. +// +// Warning regarding blocking calls: Code in the QUICHE library typically +// handles IO on a single thread, so if making calls from that typical +// environment, it would be problematic to make a blocking call and block that +// single thread. +class QUICHE_EXPORT_PRIVATE StreamClientSocket : public Socket { + public: + class AsyncVisitor : public Socket::AsyncVisitor { + public: + virtual void ConnectComplete(absl::Status status) = 0; + }; + + ~StreamClientSocket() override = default; + + // Establishes a connection synchronously. Should not be called if socket has + // already been successfully connected without first calling Disconnect(). + virtual absl::Status ConnectBlocking() = 0; + + // Establishes a connection asynchronously. On completion, calls + // ConnectComplete() on the visitor, potentially before return from + // ConnectAsync(). Should not be called if socket has already been + // successfully connected without first calling Disconnect(). + // + // After calling, the socket must not be destroyed until Disconnect() is + // called. + virtual void ConnectAsync() = 0; + + // Disconnects a connected socket or cancels an in-progress ConnectAsync(), + // invoking the `ConnectComplete(absl::CancelledError())` on the visitor. + // After success, it is possible to call ConnectBlocking() or ConnectAsync() + // again to establish a new connection. Cancels any pending read or write + // operations, calling visitor completion methods with + // `absl::CancelledError()`. + // + // Typically implemented via a call to ::close(), which for TCP can result in + // either FIN or RST, depending on socket/platform state and undefined + // platform behavior. + virtual void Disconnect() = 0; +}; + +} // namespace quic + +#endif // QUICHE_QUIC_CORE_IO_STREAM_CLIENT_SOCKET_H_