mirror of https://github.com/grpc/grpc.git
[EventEngine] WindowsEndpoint (#31735)
* [EventEngine] WindowsEndpoint Initial sketch, all tests passing * Port fix from #28432 * GPR_WINDOWS guard * use MemoryAllocator::MakeReservation for allocated buffers * better logging (respect slice length) * Automated change: Fix sanity tests * improvements * Automated change: Fix sanity tests * InlinedVector<WSABUF, kMaxWSABUFCount> * initial attempt at socket util reunification * posix fixes + local run of sanitize.sh * posix socket includes * fix * Automated change: Fix sanity tests * remove unused include (breaks windows) * remove stale comment Co-authored-by: drfloob <drfloob@users.noreply.github.com>pull/31768/head
AJ Heller
2 years ago
committed by
GitHub
parent
808347ffe8
commit
557e558825
39 changed files with 1613 additions and 559 deletions
@ -0,0 +1,320 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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 <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/event_engine/tcp_socket_utils.h" |
||||
|
||||
#include "src/core/lib/iomgr/port.h" |
||||
|
||||
#ifdef GRPC_POSIX_SOCKET_UTILS_COMMON |
||||
#include <arpa/inet.h> // IWYU pragma: keep |
||||
|
||||
#ifdef GRPC_LINUX_TCP_H |
||||
#include <linux/tcp.h> |
||||
#else |
||||
#include <netinet/in.h> // IWYU pragma: keep |
||||
#include <netinet/tcp.h> |
||||
#endif |
||||
#include <fcntl.h> |
||||
#include <sys/socket.h> |
||||
#include <unistd.h> |
||||
#endif // GRPC_POSIX_SOCKET_UTILS_COMMON
|
||||
|
||||
#ifdef GRPC_HAVE_UNIX_SOCKET |
||||
#include <sys/stat.h> // IWYU pragma: keep |
||||
#include <sys/un.h> |
||||
#endif |
||||
|
||||
#include <errno.h> |
||||
#include <inttypes.h> |
||||
#include <stdlib.h> |
||||
#include <string.h> |
||||
|
||||
#include <utility> |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "absl/strings/str_cat.h" |
||||
#include "absl/strings/str_format.h" |
||||
|
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/gprpp/host_port.h" |
||||
#include "src/core/lib/gprpp/status_helper.h" |
||||
#include "src/core/lib/iomgr/sockaddr.h" |
||||
#include "src/core/lib/uri/uri_parser.h" |
||||
|
||||
namespace grpc_event_engine { |
||||
namespace experimental { |
||||
|
||||
namespace { |
||||
constexpr uint8_t kV4MappedPrefix[] = {0, 0, 0, 0, 0, 0, |
||||
0, 0, 0, 0, 0xff, 0xff}; |
||||
absl::StatusOr<std::string> GetScheme( |
||||
const EventEngine::ResolvedAddress& resolved_address) { |
||||
switch (resolved_address.address()->sa_family) { |
||||
case AF_INET: |
||||
return "ipv4"; |
||||
case AF_INET6: |
||||
return "ipv6"; |
||||
case AF_UNIX: |
||||
return "unix"; |
||||
default: |
||||
return absl::InvalidArgumentError(absl::StrFormat( |
||||
"Unknown scheme: %d", resolved_address.address()->sa_family)); |
||||
} |
||||
} |
||||
} // namespace
|
||||
|
||||
bool ResolvedAddressIsV4Mapped( |
||||
const EventEngine::ResolvedAddress& resolved_addr, |
||||
EventEngine::ResolvedAddress* resolved_addr4_out) { |
||||
const sockaddr* addr = resolved_addr.address(); |
||||
if (addr->sa_family == AF_INET6) { |
||||
const sockaddr_in6* addr6 = reinterpret_cast<const sockaddr_in6*>(addr); |
||||
sockaddr_in* addr4_out = |
||||
resolved_addr4_out == nullptr |
||||
? nullptr |
||||
: reinterpret_cast<sockaddr_in*>( |
||||
const_cast<sockaddr*>(resolved_addr4_out->address())); |
||||
|
||||
if (memcmp(addr6->sin6_addr.s6_addr, kV4MappedPrefix, |
||||
sizeof(kV4MappedPrefix)) == 0) { |
||||
if (resolved_addr4_out != nullptr) { |
||||
// Normalize ::ffff:0.0.0.0/96 to IPv4.
|
||||
memset(addr4_out, 0, sizeof(sockaddr_in)); |
||||
addr4_out->sin_family = AF_INET; |
||||
// s6_addr32 would be nice, but it's non-standard.
|
||||
memcpy(&addr4_out->sin_addr, &addr6->sin6_addr.s6_addr[12], 4); |
||||
addr4_out->sin_port = addr6->sin6_port; |
||||
*resolved_addr4_out = EventEngine::ResolvedAddress( |
||||
reinterpret_cast<sockaddr*>(addr4_out), |
||||
static_cast<socklen_t>(sizeof(sockaddr_in))); |
||||
} |
||||
return true; |
||||
} |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
bool ResolvedAddressToV4Mapped( |
||||
const EventEngine::ResolvedAddress& resolved_addr, |
||||
EventEngine::ResolvedAddress* resolved_addr6_out) { |
||||
GPR_ASSERT(&resolved_addr != resolved_addr6_out); |
||||
const sockaddr* addr = resolved_addr.address(); |
||||
sockaddr_in6* addr6_out = const_cast<sockaddr_in6*>( |
||||
reinterpret_cast<const sockaddr_in6*>(resolved_addr6_out->address())); |
||||
if (addr->sa_family == AF_INET) { |
||||
const sockaddr_in* addr4 = reinterpret_cast<const sockaddr_in*>(addr); |
||||
memset(resolved_addr6_out, 0, sizeof(*resolved_addr6_out)); |
||||
addr6_out->sin6_family = AF_INET6; |
||||
memcpy(&addr6_out->sin6_addr.s6_addr[0], kV4MappedPrefix, 12); |
||||
memcpy(&addr6_out->sin6_addr.s6_addr[12], &addr4->sin_addr, 4); |
||||
addr6_out->sin6_port = addr4->sin_port; |
||||
*resolved_addr6_out = EventEngine::ResolvedAddress( |
||||
reinterpret_cast<sockaddr*>(addr6_out), |
||||
static_cast<socklen_t>(sizeof(sockaddr_in6))); |
||||
return true; |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
EventEngine::ResolvedAddress ResolvedAddressMakeWild6(int port) { |
||||
EventEngine::ResolvedAddress resolved_wild_out; |
||||
sockaddr_in6* wild_out = reinterpret_cast<sockaddr_in6*>( |
||||
const_cast<sockaddr*>(resolved_wild_out.address())); |
||||
GPR_ASSERT(port >= 0 && port < 65536); |
||||
memset(wild_out, 0, sizeof(sockaddr_in6)); |
||||
wild_out->sin6_family = AF_INET6; |
||||
wild_out->sin6_port = htons(static_cast<uint16_t>(port)); |
||||
return EventEngine::ResolvedAddress( |
||||
reinterpret_cast<sockaddr*>(wild_out), |
||||
static_cast<socklen_t>(sizeof(sockaddr_in6))); |
||||
} |
||||
|
||||
EventEngine::ResolvedAddress ResolvedAddressMakeWild4(int port) { |
||||
EventEngine::ResolvedAddress resolved_wild_out; |
||||
sockaddr_in* wild_out = reinterpret_cast<sockaddr_in*>( |
||||
const_cast<sockaddr*>(resolved_wild_out.address())); |
||||
GPR_ASSERT(port >= 0 && port < 65536); |
||||
memset(wild_out, 0, sizeof(sockaddr_in)); |
||||
wild_out->sin_family = AF_INET; |
||||
wild_out->sin_port = htons(static_cast<uint16_t>(port)); |
||||
return EventEngine::ResolvedAddress( |
||||
reinterpret_cast<sockaddr*>(wild_out), |
||||
static_cast<socklen_t>(sizeof(sockaddr_in))); |
||||
} |
||||
|
||||
int ResolvedAddressGetPort(const EventEngine::ResolvedAddress& resolved_addr) { |
||||
const sockaddr* addr = resolved_addr.address(); |
||||
switch (addr->sa_family) { |
||||
case AF_INET: |
||||
return ntohs((reinterpret_cast<const sockaddr_in*>(addr))->sin_port); |
||||
case AF_INET6: |
||||
return ntohs((reinterpret_cast<const sockaddr_in6*>(addr))->sin6_port); |
||||
#ifdef GRPC_HAVE_UNIX_SOCKET |
||||
case AF_UNIX: |
||||
return 1; |
||||
#endif |
||||
default: |
||||
gpr_log(GPR_ERROR, "Unknown socket family %d in ResolvedAddressGetPort", |
||||
addr->sa_family); |
||||
abort(); |
||||
} |
||||
} |
||||
|
||||
void ResolvedAddressSetPort(EventEngine::ResolvedAddress& resolved_addr, |
||||
int port) { |
||||
sockaddr* addr = const_cast<sockaddr*>(resolved_addr.address()); |
||||
switch (addr->sa_family) { |
||||
case AF_INET: |
||||
GPR_ASSERT(port >= 0 && port < 65536); |
||||
(reinterpret_cast<sockaddr_in*>(addr))->sin_port = |
||||
htons(static_cast<uint16_t>(port)); |
||||
return; |
||||
case AF_INET6: |
||||
GPR_ASSERT(port >= 0 && port < 65536); |
||||
(reinterpret_cast<sockaddr_in6*>(addr))->sin6_port = |
||||
htons(static_cast<uint16_t>(port)); |
||||
return; |
||||
default: |
||||
gpr_log(GPR_ERROR, "Unknown socket family %d in grpc_sockaddr_set_port", |
||||
addr->sa_family); |
||||
abort(); |
||||
} |
||||
} |
||||
|
||||
absl::optional<int> ResolvedAddressIsWildcard( |
||||
const EventEngine::ResolvedAddress& addr) { |
||||
const EventEngine::ResolvedAddress* resolved_addr = &addr; |
||||
EventEngine::ResolvedAddress addr4_normalized; |
||||
if (ResolvedAddressIsV4Mapped(addr, &addr4_normalized)) { |
||||
resolved_addr = &addr4_normalized; |
||||
} |
||||
if (resolved_addr->address()->sa_family == AF_INET) { |
||||
// Check for 0.0.0.0
|
||||
const sockaddr_in* addr4 = |
||||
reinterpret_cast<const sockaddr_in*>(resolved_addr->address()); |
||||
if (addr4->sin_addr.s_addr != 0) { |
||||
return absl::nullopt; |
||||
} |
||||
return static_cast<int>(ntohs(addr4->sin_port)); |
||||
} else if (resolved_addr->address()->sa_family == AF_INET6) { |
||||
// Check for ::
|
||||
const sockaddr_in6* addr6 = |
||||
reinterpret_cast<const sockaddr_in6*>(resolved_addr->address()); |
||||
int i; |
||||
for (i = 0; i < 16; i++) { |
||||
if (addr6->sin6_addr.s6_addr[i] != 0) { |
||||
return absl::nullopt; |
||||
} |
||||
} |
||||
return static_cast<int>(ntohs(addr6->sin6_port)); |
||||
} else { |
||||
return absl::nullopt; |
||||
} |
||||
} |
||||
|
||||
absl::StatusOr<std::string> ResolvedAddressToNormalizedString( |
||||
const EventEngine::ResolvedAddress& resolved_addr) { |
||||
EventEngine::ResolvedAddress addr_normalized; |
||||
if (ResolvedAddressIsV4Mapped(resolved_addr, &addr_normalized)) { |
||||
return ResolvedAddressToString(addr_normalized); |
||||
} |
||||
return ResolvedAddressToString(resolved_addr); |
||||
} |
||||
|
||||
absl::StatusOr<std::string> ResolvedAddressToString( |
||||
const EventEngine::ResolvedAddress& resolved_addr) { |
||||
const int save_errno = errno; |
||||
const sockaddr* addr = resolved_addr.address(); |
||||
std::string out; |
||||
#ifdef GRPC_HAVE_UNIX_SOCKET |
||||
if (addr->sa_family == AF_UNIX) { |
||||
const sockaddr_un* addr_un = reinterpret_cast<const sockaddr_un*>(addr); |
||||
bool abstract = addr_un->sun_path[0] == '\0'; |
||||
if (abstract) { |
||||
#ifdef GPR_APPLE |
||||
int len = resolved_addr.size() - sizeof(addr_un->sun_family) - |
||||
sizeof(addr_un->sun_len); |
||||
#else |
||||
int len = resolved_addr.size() - sizeof(addr_un->sun_family); |
||||
#endif |
||||
if (len <= 0) { |
||||
return absl::InvalidArgumentError("Empty UDS abstract path"); |
||||
} |
||||
out = std::string(addr_un->sun_path, len); |
||||
} else { |
||||
size_t maxlen = sizeof(addr_un->sun_path); |
||||
if (strnlen(addr_un->sun_path, maxlen) == maxlen) { |
||||
return absl::InvalidArgumentError("UDS path is not null-terminated"); |
||||
} |
||||
out = std::string(addr_un->sun_path); |
||||
} |
||||
return out; |
||||
} |
||||
#endif // GRPC_HAVE_UNIX_SOCKET
|
||||
|
||||
const void* ip = nullptr; |
||||
int port = 0; |
||||
uint32_t sin6_scope_id = 0; |
||||
if (addr->sa_family == AF_INET) { |
||||
const sockaddr_in* addr4 = reinterpret_cast<const sockaddr_in*>(addr); |
||||
ip = &addr4->sin_addr; |
||||
port = ntohs(addr4->sin_port); |
||||
} else if (addr->sa_family == AF_INET6) { |
||||
const sockaddr_in6* addr6 = reinterpret_cast<const sockaddr_in6*>(addr); |
||||
ip = &addr6->sin6_addr; |
||||
port = ntohs(addr6->sin6_port); |
||||
sin6_scope_id = addr6->sin6_scope_id; |
||||
} |
||||
char ntop_buf[INET6_ADDRSTRLEN]; |
||||
if (ip != nullptr && |
||||
inet_ntop(addr->sa_family, ip, ntop_buf, sizeof(ntop_buf)) != nullptr) { |
||||
if (sin6_scope_id != 0) { |
||||
// Enclose sin6_scope_id with the format defined in RFC 6874
|
||||
// section 2.
|
||||
std::string host_with_scope = |
||||
absl::StrFormat("%s%%%" PRIu32, ntop_buf, sin6_scope_id); |
||||
out = grpc_core::JoinHostPort(host_with_scope, port); |
||||
} else { |
||||
out = grpc_core::JoinHostPort(ntop_buf, port); |
||||
} |
||||
} else { |
||||
return absl::InvalidArgumentError( |
||||
absl::StrCat("Unknown sockaddr family: ", addr->sa_family)); |
||||
} |
||||
// This is probably redundant, but we wouldn't want to log the wrong
|
||||
// error.
|
||||
errno = save_errno; |
||||
return out; |
||||
} |
||||
|
||||
absl::StatusOr<std::string> ResolvedAddressToURI( |
||||
const EventEngine::ResolvedAddress& resolved_address) { |
||||
if (resolved_address.size() == 0) { |
||||
return absl::InvalidArgumentError("Empty address"); |
||||
} |
||||
auto scheme = GetScheme(resolved_address); |
||||
GRPC_RETURN_IF_ERROR(scheme.status()); |
||||
auto path = ResolvedAddressToString(resolved_address); |
||||
GRPC_RETURN_IF_ERROR(path.status()); |
||||
absl::StatusOr<grpc_core::URI> uri = |
||||
grpc_core::URI::Create(*scheme, /*authority=*/"", std::move(path.value()), |
||||
/*query_parameter_pairs=*/{}, /*fragment=*/""); |
||||
if (!uri.ok()) return uri.status(); |
||||
return uri->ToString(); |
||||
} |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc_event_engine
|
||||
@ -0,0 +1,85 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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.
|
||||
#ifndef GRPC_CORE_LIB_EVENT_ENGINE_TCP_SOCKET_UTILS_H |
||||
#define GRPC_CORE_LIB_EVENT_ENGINE_TCP_SOCKET_UTILS_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <string> |
||||
|
||||
#include "absl/status/statusor.h" |
||||
#include "absl/types/optional.h" |
||||
|
||||
#include <grpc/event_engine/event_engine.h> |
||||
|
||||
namespace grpc_event_engine { |
||||
namespace experimental { |
||||
|
||||
// Returns true if resolved_addr is an IPv4-mapped IPv6 address within the
|
||||
// ::ffff:0.0.0.0/96 range, or false otherwise.
|
||||
// If resolved_addr4_out is non-NULL, the inner IPv4 address will be copied
|
||||
// here when returning true.
|
||||
bool ResolvedAddressIsV4Mapped( |
||||
const EventEngine::ResolvedAddress& resolved_addr, |
||||
EventEngine::ResolvedAddress* resolved_addr4_out); |
||||
|
||||
// If resolved_addr is an AF_INET address, writes the corresponding
|
||||
// ::ffff:0.0.0.0/96 address to resolved_addr6_out and returns true. Otherwise
|
||||
// returns false.
|
||||
bool ResolvedAddressToV4Mapped( |
||||
const EventEngine::ResolvedAddress& resolved_addr, |
||||
EventEngine::ResolvedAddress* resolved_addr6_out); |
||||
|
||||
// Make wild card IPv6 address with specified port.
|
||||
EventEngine::ResolvedAddress ResolvedAddressMakeWild6(int port); |
||||
|
||||
// Make wild card IPv4 address with specified port.
|
||||
EventEngine::ResolvedAddress ResolvedAddressMakeWild4(int port); |
||||
|
||||
// Given a resolved address, return the port number in the address.
|
||||
int ResolvedAddressGetPort(const EventEngine::ResolvedAddress& resolved_addr); |
||||
|
||||
// Modifies the address, setting the specified port number.
|
||||
// The operation would only succeed if the passed address is an IPv4 or Ipv6
|
||||
// address. Otherwise the function call would abort fail.
|
||||
void ResolvedAddressSetPort(EventEngine::ResolvedAddress& resolved_addr, |
||||
int port); |
||||
|
||||
// Returns the port number associated with the address if the given address is
|
||||
// not a wildcard ipv6 or ipv6 address. Otherwise returns absl::nullopt
|
||||
absl::optional<int> ResolvedAddressIsWildcard( |
||||
const EventEngine::ResolvedAddress& addr); |
||||
|
||||
// Converts a EventEngine::ResolvedAddress into a newly-allocated
|
||||
// human-readable string.
|
||||
// Currently, only the AF_INET, AF_INET6, and AF_UNIX families are
|
||||
// recognized.
|
||||
absl::StatusOr<std::string> ResolvedAddressToString( |
||||
const EventEngine::ResolvedAddress& resolved_addr); |
||||
|
||||
// Converts a EventEngine::ResolvedAddress into a newly-allocated
|
||||
// human-readable string. See ResolvedAddressToString.
|
||||
// This functional normalizes, so for example: ::ffff:0.0.0.0/96 IPv6
|
||||
// addresses are displayed as plain IPv4.
|
||||
absl::StatusOr<std::string> ResolvedAddressToNormalizedString( |
||||
const EventEngine::ResolvedAddress& resolved_addr); |
||||
|
||||
// Returns the URI string corresponding to the resolved_address
|
||||
absl::StatusOr<std::string> ResolvedAddressToURI( |
||||
const EventEngine::ResolvedAddress& resolved_address); |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc_event_engine
|
||||
|
||||
#endif // GRPC_CORE_LIB_EVENT_ENGINE_TCP_SOCKET_UTILS_H
|
||||
@ -0,0 +1,273 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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 <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_WINDOWS |
||||
|
||||
#include "absl/cleanup/cleanup.h" |
||||
#include "absl/functional/any_invocable.h" |
||||
#include "absl/status/status.h" |
||||
#include "absl/strings/str_format.h" |
||||
|
||||
#include <grpc/event_engine/memory_allocator.h> |
||||
#include <grpc/support/log_windows.h> |
||||
|
||||
#include "src/core/lib/event_engine/tcp_socket_utils.h" |
||||
#include "src/core/lib/event_engine/trace.h" |
||||
#include "src/core/lib/event_engine/windows/windows_endpoint.h" |
||||
#include "src/core/lib/gprpp/debug_location.h" |
||||
#include "src/core/lib/gprpp/status_helper.h" |
||||
#include "src/core/lib/iomgr/error.h" |
||||
|
||||
namespace grpc_event_engine { |
||||
namespace experimental { |
||||
|
||||
// TODO(hork): The previous implementation required internal ref counting. Add
|
||||
// this when it becomes necessary.
|
||||
// TODO(hork): The previous implementation required a 2-phase shutdown. Add this
|
||||
// when it becomes necessary.
|
||||
|
||||
namespace { |
||||
constexpr int64_t kDefaultTargetReadSize = 8192; |
||||
constexpr int kMaxWSABUFCount = 16; |
||||
|
||||
void AbortOnEvent(absl::Status) { |
||||
GPR_ASSERT(false && |
||||
"INTERNAL ERROR: Asked to handle read/write event with an invalid " |
||||
"callback"); |
||||
} |
||||
|
||||
} // namespace
|
||||
|
||||
WindowsEndpoint::WindowsEndpoint( |
||||
const EventEngine::ResolvedAddress& peer_address, |
||||
std::unique_ptr<WinSocket> socket, MemoryAllocator&& allocator, |
||||
const EndpointConfig& /* config */, Executor* executor) |
||||
: peer_address_(peer_address), |
||||
socket_(std::move(socket)), |
||||
allocator_(std::move(allocator)), |
||||
handle_read_event_(this), |
||||
handle_write_event_(this), |
||||
executor_(executor) { |
||||
sockaddr addr; |
||||
int addr_len = sizeof(addr); |
||||
if (getsockname(socket_->socket(), &addr, &addr_len) < 0) { |
||||
gpr_log(GPR_ERROR, "Unrecoverable error: Failed to get local socket name."); |
||||
abort(); |
||||
} |
||||
local_address_ = EventEngine::ResolvedAddress(&addr, addr_len); |
||||
local_address_string_ = *ResolvedAddressToURI(local_address_); |
||||
peer_address_string_ = *ResolvedAddressToURI(peer_address_); |
||||
} |
||||
|
||||
WindowsEndpoint::~WindowsEndpoint() { |
||||
socket_->MaybeShutdown(absl::OkStatus()); |
||||
} |
||||
|
||||
void WindowsEndpoint::Read(absl::AnyInvocable<void(absl::Status)> on_read, |
||||
SliceBuffer* buffer, const ReadArgs* args) { |
||||
// TODO(hork): last_read_buffer from iomgr: Is it only garbage, or optimized?
|
||||
GRPC_EVENT_ENGINE_TRACE("WindowsEndpoint::%p reading", this); |
||||
// Prepare the WSABUF struct
|
||||
WSABUF wsa_buffers[kMaxWSABUFCount]; |
||||
int min_read_size = kDefaultTargetReadSize; |
||||
if (args != nullptr && args->read_hint_bytes > 0) { |
||||
min_read_size = args->read_hint_bytes; |
||||
} |
||||
if (buffer->Length() < min_read_size && buffer->Count() < kMaxWSABUFCount) { |
||||
buffer->AppendIndexed(Slice(allocator_.MakeSlice(min_read_size))); |
||||
} |
||||
GPR_ASSERT(buffer->Count() <= kMaxWSABUFCount); |
||||
for (int i = 0; i < buffer->Count(); i++) { |
||||
Slice tmp = buffer->RefSlice(i); |
||||
wsa_buffers[i].buf = (char*)tmp.begin(); |
||||
wsa_buffers[i].len = tmp.size(); |
||||
} |
||||
DWORD bytes_read = 0; |
||||
DWORD flags = 0; |
||||
// First let's try a synchronous, non-blocking read.
|
||||
int status = WSARecv(socket_->socket(), wsa_buffers, (DWORD)buffer->Count(), |
||||
&bytes_read, &flags, nullptr, nullptr); |
||||
int wsa_error = status == 0 ? 0 : WSAGetLastError(); |
||||
// Did we get data immediately ? Yay.
|
||||
if (wsa_error != WSAEWOULDBLOCK) { |
||||
// prune slicebuffer
|
||||
if (bytes_read != buffer->Length()) { |
||||
buffer->RemoveLastNBytes(buffer->Length() - bytes_read); |
||||
} |
||||
executor_->Run([on_read = std::move(on_read)]() mutable { |
||||
on_read(absl::OkStatus()); |
||||
}); |
||||
return; |
||||
} |
||||
// Otherwise, let's retry, by queuing a read.
|
||||
memset(socket_->read_info()->overlapped(), 0, sizeof(OVERLAPPED)); |
||||
status = |
||||
WSARecv(socket_->socket(), wsa_buffers, (DWORD)buffer->Count(), |
||||
&bytes_read, &flags, socket_->read_info()->overlapped(), nullptr); |
||||
wsa_error = status == 0 ? 0 : WSAGetLastError(); |
||||
if (wsa_error != 0 && wsa_error != WSA_IO_PENDING) { |
||||
// Async read returned immediately with an error
|
||||
executor_->Run([this, on_read = std::move(on_read), wsa_error]() mutable { |
||||
on_read(GRPC_WSA_ERROR( |
||||
wsa_error, |
||||
absl::StrFormat("WindowsEndpont::%p Read failed", this).c_str())); |
||||
}); |
||||
return; |
||||
} |
||||
|
||||
handle_read_event_.Prime(buffer, std::move(on_read)); |
||||
socket_->NotifyOnRead(&handle_read_event_); |
||||
} |
||||
|
||||
void WindowsEndpoint::Write(absl::AnyInvocable<void(absl::Status)> on_writable, |
||||
SliceBuffer* data, const WriteArgs* /* args */) { |
||||
if (grpc_event_engine_trace.enabled()) { |
||||
for (int i = 0; i < data->Count(); i++) { |
||||
auto str = data->RefSlice(i).as_string_view(); |
||||
gpr_log(GPR_INFO, "WindowsEndpoint::%p WRITE (peer=%s): %.*s", this, |
||||
peer_address_string_.c_str(), str.length(), str.data()); |
||||
} |
||||
} |
||||
GPR_ASSERT(data->Count() <= UINT_MAX); |
||||
absl::InlinedVector<WSABUF, kMaxWSABUFCount> buffers(data->Count()); |
||||
for (int i = 0; i < data->Count(); i++) { |
||||
auto slice = data->RefSlice(i); |
||||
GPR_ASSERT(slice.size() <= ULONG_MAX); |
||||
buffers[i].len = slice.size(); |
||||
buffers[i].buf = (char*)slice.begin(); |
||||
} |
||||
// First, let's try a synchronous, non-blocking write.
|
||||
DWORD bytes_sent; |
||||
int status = WSASend(socket_->socket(), buffers.data(), (DWORD)buffers.size(), |
||||
&bytes_sent, 0, nullptr, nullptr); |
||||
size_t async_buffers_offset; |
||||
if (status == 0) { |
||||
if (bytes_sent == data->Length()) { |
||||
// Write completed, exiting early
|
||||
executor_->Run( |
||||
[cb = std::move(on_writable)]() mutable { cb(absl::OkStatus()); }); |
||||
return; |
||||
} |
||||
// The data was not completely delivered, we should send the rest of it by
|
||||
// doing an async write operation.
|
||||
for (int i = 0; i < data->Count(); i++) { |
||||
if (buffers[i].len > bytes_sent) { |
||||
buffers[i].buf += bytes_sent; |
||||
buffers[i].len -= bytes_sent; |
||||
break; |
||||
} |
||||
bytes_sent -= buffers[i].len; |
||||
async_buffers_offset++; |
||||
} |
||||
} else { |
||||
// We would kind of expect to get a WSAEWOULDBLOCK here, especially on a
|
||||
// busy connection that has its send queue filled up. But if we don't,
|
||||
// then we can avoid doing an async write operation at all.
|
||||
int wsa_error = WSAGetLastError(); |
||||
if (wsa_error != WSAEWOULDBLOCK) { |
||||
executor_->Run([cb = std::move(on_writable), wsa_error]() mutable { |
||||
cb(GRPC_WSA_ERROR(wsa_error, "WSASend")); |
||||
}); |
||||
return; |
||||
} |
||||
} |
||||
auto write_info = socket_->write_info(); |
||||
memset(write_info->overlapped(), 0, sizeof(OVERLAPPED)); |
||||
status = WSASend(socket_->socket(), &buffers[async_buffers_offset], |
||||
(DWORD)(data->Count() - async_buffers_offset), nullptr, 0, |
||||
write_info->overlapped(), nullptr); |
||||
|
||||
if (status != 0) { |
||||
int wsa_error = WSAGetLastError(); |
||||
if (wsa_error != WSA_IO_PENDING) { |
||||
executor_->Run([cb = std::move(on_writable), wsa_error]() mutable { |
||||
cb(GRPC_WSA_ERROR(wsa_error, "WSASend")); |
||||
}); |
||||
return; |
||||
} |
||||
} |
||||
// As all is now setup, we can now ask for the IOCP notification. It may
|
||||
// trigger the callback immediately however, but no matter.
|
||||
handle_write_event_.Prime(data, std::move(on_writable)); |
||||
socket_->NotifyOnWrite(&handle_write_event_); |
||||
} |
||||
const EventEngine::ResolvedAddress& WindowsEndpoint::GetPeerAddress() const { |
||||
return peer_address_; |
||||
} |
||||
const EventEngine::ResolvedAddress& WindowsEndpoint::GetLocalAddress() const { |
||||
return local_address_; |
||||
} |
||||
|
||||
// ---- Handle{Read|Write}Closure
|
||||
|
||||
WindowsEndpoint::BaseEventClosure::BaseEventClosure(WindowsEndpoint* endpoint) |
||||
: endpoint_(endpoint), cb_(&AbortOnEvent) {} |
||||
|
||||
void WindowsEndpoint::HandleReadClosure::Run() { |
||||
GRPC_EVENT_ENGINE_TRACE("WindowsEndpoint::%p Handling Read Event", endpoint_); |
||||
absl::Status status; |
||||
auto* read_info = endpoint_->socket_->read_info(); |
||||
auto cb_cleanup = absl::MakeCleanup([this, &status]() { |
||||
auto cb = std::move(cb_); |
||||
cb_ = &AbortOnEvent; |
||||
cb(status); |
||||
}); |
||||
if (read_info->wsa_error() != 0) { |
||||
status = GRPC_WSA_ERROR(read_info->wsa_error(), "Async Read Error"); |
||||
buffer_->Clear(); |
||||
return; |
||||
} |
||||
if (read_info->bytes_transferred() > 0) { |
||||
GPR_ASSERT(read_info->bytes_transferred() <= buffer_->Length()); |
||||
if (read_info->bytes_transferred() != buffer_->Length()) { |
||||
buffer_->RemoveLastNBytes(buffer_->Length() - |
||||
read_info->bytes_transferred()); |
||||
} |
||||
GPR_ASSERT(read_info->bytes_transferred() == buffer_->Length()); |
||||
if (grpc_event_engine_trace.enabled()) { |
||||
for (int i = 0; i < buffer_->Count(); i++) { |
||||
auto str = buffer_->RefSlice(i).as_string_view(); |
||||
gpr_log(GPR_INFO, "WindowsEndpoint::%p READ (peer=%s): %.*s", this, |
||||
endpoint_->peer_address_string_.c_str(), str.length(), |
||||
str.data()); |
||||
} |
||||
} |
||||
return; |
||||
} |
||||
// Either the endpoint is shut down or we've seen the end of the stream
|
||||
buffer_->Clear(); |
||||
// TODO(hork): different error message if shut down
|
||||
status = absl::UnavailableError("End of TCP stream"); |
||||
} |
||||
|
||||
void WindowsEndpoint::HandleWriteClosure::Run() { |
||||
GRPC_EVENT_ENGINE_TRACE("WindowsEndpoint::%p Handling Write Event", |
||||
endpoint_); |
||||
auto* write_info = endpoint_->socket_->write_info(); |
||||
auto cb = std::move(cb_); |
||||
cb_ = &AbortOnEvent; |
||||
absl::Status status; |
||||
if (write_info->wsa_error() != 0) { |
||||
status = GRPC_WSA_ERROR(write_info->wsa_error(), "WSASend"); |
||||
} else { |
||||
GPR_ASSERT(write_info->bytes_transferred() == buffer_->Length()); |
||||
} |
||||
cb(status); |
||||
} |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc_event_engine
|
||||
|
||||
#endif // GPR_WINDOWS
|
||||
@ -0,0 +1,94 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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.
|
||||
#ifndef GRPC_CORE_LIB_EVENT_ENGINE_WINDOWS_WINDOWS_ENDPOINT_H |
||||
#define GRPC_CORE_LIB_EVENT_ENGINE_WINDOWS_WINDOWS_ENDPOINT_H |
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_WINDOWS |
||||
|
||||
#include <grpc/event_engine/event_engine.h> |
||||
|
||||
#include "src/core/lib/event_engine/windows/win_socket.h" |
||||
|
||||
namespace grpc_event_engine { |
||||
namespace experimental { |
||||
|
||||
class WindowsEndpoint : public EventEngine::Endpoint { |
||||
public: |
||||
WindowsEndpoint(const EventEngine::ResolvedAddress& peer_address, |
||||
std::unique_ptr<WinSocket> socket, |
||||
MemoryAllocator&& allocator, const EndpointConfig& config, |
||||
Executor* Executor); |
||||
~WindowsEndpoint() override; |
||||
void Read(absl::AnyInvocable<void(absl::Status)> on_read, SliceBuffer* buffer, |
||||
const ReadArgs* args) override; |
||||
void Write(absl::AnyInvocable<void(absl::Status)> on_writable, |
||||
SliceBuffer* data, const WriteArgs* args) override; |
||||
const EventEngine::ResolvedAddress& GetPeerAddress() const override; |
||||
const EventEngine::ResolvedAddress& GetLocalAddress() const override; |
||||
|
||||
private: |
||||
// Base class for the Read- and Write-specific event handler callbacks
|
||||
class BaseEventClosure : public EventEngine::Closure { |
||||
public: |
||||
explicit BaseEventClosure(WindowsEndpoint* endpoint); |
||||
// Calls the bound application callback, inline.
|
||||
// If called through IOCP, this will be run from within an Executor.
|
||||
virtual void Run() = 0; |
||||
|
||||
// Prepare the closure by setting the application callback and SliceBuffer
|
||||
void Prime(SliceBuffer* buffer, absl::AnyInvocable<void(absl::Status)> cb) { |
||||
cb_ = std::move(cb); |
||||
buffer_ = buffer; |
||||
} |
||||
|
||||
protected: |
||||
absl::AnyInvocable<void(absl::Status)> cb_; |
||||
SliceBuffer* buffer_; |
||||
WindowsEndpoint* endpoint_; |
||||
}; |
||||
|
||||
// Permanent closure type for Read callbacks
|
||||
class HandleReadClosure : public BaseEventClosure { |
||||
public: |
||||
explicit HandleReadClosure(WindowsEndpoint* endpoint) |
||||
: BaseEventClosure(endpoint) {} |
||||
void Run() override; |
||||
}; |
||||
|
||||
// Permanent closure type for Write callbacks
|
||||
class HandleWriteClosure : public BaseEventClosure { |
||||
public: |
||||
explicit HandleWriteClosure(WindowsEndpoint* endpoint) |
||||
: BaseEventClosure(endpoint) {} |
||||
void Run() override; |
||||
}; |
||||
|
||||
EventEngine::ResolvedAddress peer_address_; |
||||
std::string peer_address_string_; |
||||
EventEngine::ResolvedAddress local_address_; |
||||
std::string local_address_string_; |
||||
std::unique_ptr<WinSocket> socket_; |
||||
MemoryAllocator allocator_; |
||||
HandleReadClosure handle_read_event_; |
||||
HandleWriteClosure handle_write_event_; |
||||
Executor* executor_; |
||||
}; |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc_event_engine
|
||||
|
||||
#endif |
||||
|
||||
#endif // GRPC_CORE_LIB_EVENT_ENGINE_WINDOWS_WINDOWS_ENDPOINT_H
|
||||
@ -0,0 +1,283 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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 <grpc/support/port_platform.h> |
||||
|
||||
#include "src/core/lib/event_engine/tcp_socket_utils.h" |
||||
|
||||
#include <errno.h> |
||||
#include <stdint.h> |
||||
#include <string.h> |
||||
|
||||
// IWYU pragma: no_include <arpa/inet.h>
|
||||
|
||||
#include <string> |
||||
|
||||
#include "src/core/lib/iomgr/port.h" |
||||
|
||||
#ifdef GRPC_HAVE_UNIX_SOCKET |
||||
#include <sys/un.h> |
||||
#endif |
||||
|
||||
#include "absl/status/status.h" |
||||
#include "absl/status/statusor.h" |
||||
#include "gtest/gtest.h" |
||||
|
||||
#include <grpc/event_engine/event_engine.h> |
||||
#include <grpc/support/log.h> |
||||
|
||||
#include "src/core/lib/iomgr/sockaddr.h" |
||||
|
||||
namespace grpc_event_engine { |
||||
namespace experimental { |
||||
|
||||
namespace { |
||||
using ::grpc_event_engine::experimental::EventEngine; |
||||
|
||||
const uint8_t kMapped[] = {0, 0, 0, 0, 0, 0, 0, 0, |
||||
0, 0, 0xff, 0xff, 192, 0, 2, 1}; |
||||
const uint8_t kNotQuiteMapped[] = {0, 0, 0, 0, 0, 0, 0, 0, |
||||
0, 0, 0xff, 0xfe, 192, 0, 2, 99}; |
||||
const uint8_t kIPv4[] = {192, 0, 2, 1}; |
||||
const uint8_t kIPv6[] = {0x20, 0x01, 0x0d, 0xb8, 0, 0, 0, 0, |
||||
0, 0, 0, 0, 0, 0, 0, 1}; |
||||
|
||||
EventEngine::ResolvedAddress MakeAddr4(const uint8_t* data, size_t data_len) { |
||||
EventEngine::ResolvedAddress resolved_addr4; |
||||
sockaddr_in* addr4 = reinterpret_cast<sockaddr_in*>( |
||||
const_cast<sockaddr*>(resolved_addr4.address())); |
||||
memset(&resolved_addr4, 0, sizeof(resolved_addr4)); |
||||
addr4->sin_family = AF_INET; |
||||
GPR_ASSERT(data_len == sizeof(addr4->sin_addr.s_addr)); |
||||
memcpy(&addr4->sin_addr.s_addr, data, data_len); |
||||
addr4->sin_port = htons(12345); |
||||
return EventEngine::ResolvedAddress( |
||||
reinterpret_cast<sockaddr*>(addr4), |
||||
static_cast<socklen_t>(sizeof(sockaddr_in))); |
||||
} |
||||
|
||||
EventEngine::ResolvedAddress MakeAddr6(const uint8_t* data, size_t data_len) { |
||||
EventEngine::ResolvedAddress resolved_addr6; |
||||
sockaddr_in6* addr6 = reinterpret_cast<sockaddr_in6*>( |
||||
const_cast<sockaddr*>(resolved_addr6.address())); |
||||
memset(&resolved_addr6, 0, sizeof(resolved_addr6)); |
||||
addr6->sin6_family = AF_INET6; |
||||
GPR_ASSERT(data_len == sizeof(addr6->sin6_addr.s6_addr)); |
||||
memcpy(&addr6->sin6_addr.s6_addr, data, data_len); |
||||
addr6->sin6_port = htons(12345); |
||||
return EventEngine::ResolvedAddress( |
||||
reinterpret_cast<sockaddr*>(addr6), |
||||
static_cast<socklen_t>(sizeof(sockaddr_in6))); |
||||
} |
||||
|
||||
void SetIPv6ScopeId(EventEngine::ResolvedAddress& addr, uint32_t scope_id) { |
||||
sockaddr_in6* addr6 = |
||||
reinterpret_cast<sockaddr_in6*>(const_cast<sockaddr*>(addr.address())); |
||||
ASSERT_EQ(addr6->sin6_family, AF_INET6); |
||||
addr6->sin6_scope_id = scope_id; |
||||
} |
||||
|
||||
#ifdef GRPC_HAVE_UNIX_SOCKET |
||||
absl::StatusOr<EventEngine::ResolvedAddress> UnixSockaddrPopulate( |
||||
absl::string_view path) { |
||||
EventEngine::ResolvedAddress resolved_addr; |
||||
memset(const_cast<sockaddr*>(resolved_addr.address()), 0, |
||||
resolved_addr.size()); |
||||
struct sockaddr_un* un = reinterpret_cast<struct sockaddr_un*>( |
||||
const_cast<sockaddr*>(resolved_addr.address())); |
||||
const size_t maxlen = sizeof(un->sun_path) - 1; |
||||
if (path.size() > maxlen) { |
||||
return absl::InternalError(absl::StrCat( |
||||
"Path name should not have more than ", maxlen, " characters")); |
||||
} |
||||
un->sun_family = AF_UNIX; |
||||
path.copy(un->sun_path, path.size()); |
||||
un->sun_path[path.size()] = '\0'; |
||||
return EventEngine::ResolvedAddress(reinterpret_cast<sockaddr*>(un), |
||||
static_cast<socklen_t>(sizeof(*un))); |
||||
} |
||||
|
||||
absl::StatusOr<EventEngine::ResolvedAddress> UnixAbstractSockaddrPopulate( |
||||
absl::string_view path) { |
||||
EventEngine::ResolvedAddress resolved_addr; |
||||
memset(const_cast<sockaddr*>(resolved_addr.address()), 0, |
||||
resolved_addr.size()); |
||||
struct sockaddr* addr = const_cast<sockaddr*>(resolved_addr.address()); |
||||
struct sockaddr_un* un = reinterpret_cast<struct sockaddr_un*>(addr); |
||||
const size_t maxlen = sizeof(un->sun_path) - 1; |
||||
if (path.size() > maxlen) { |
||||
return absl::InternalError(absl::StrCat( |
||||
"Path name should not have more than ", maxlen, " characters")); |
||||
} |
||||
un->sun_family = AF_UNIX; |
||||
un->sun_path[0] = '\0'; |
||||
path.copy(un->sun_path + 1, path.size()); |
||||
#ifdef GPR_APPLE |
||||
return EventEngine::ResolvedAddress( |
||||
addr, static_cast<socklen_t>(sizeof(un->sun_len) + |
||||
sizeof(un->sun_family) + path.size() + 1)); |
||||
#else |
||||
return EventEngine::ResolvedAddress( |
||||
addr, static_cast<socklen_t>(sizeof(un->sun_family) + path.size() + 1)); |
||||
#endif |
||||
} |
||||
#endif // GRPC_HAVE_UNIX_SOCKET
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(TcpSocketUtilsTest, ResolvedAddressIsV4MappedTest) { |
||||
// v4mapped input should succeed.
|
||||
EventEngine::ResolvedAddress input6 = MakeAddr6(kMapped, sizeof(kMapped)); |
||||
ASSERT_TRUE(ResolvedAddressIsV4Mapped(input6, nullptr)); |
||||
EventEngine::ResolvedAddress output4; |
||||
ASSERT_TRUE(ResolvedAddressIsV4Mapped(input6, &output4)); |
||||
EventEngine::ResolvedAddress expect4 = MakeAddr4(kIPv4, sizeof(kIPv4)); |
||||
ASSERT_EQ(memcmp(expect4.address(), output4.address(), expect4.size()), 0); |
||||
|
||||
// Non-v4mapped input should fail.
|
||||
input6 = MakeAddr6(kNotQuiteMapped, sizeof(kNotQuiteMapped)); |
||||
ASSERT_FALSE(ResolvedAddressIsV4Mapped(input6, nullptr)); |
||||
ASSERT_FALSE(ResolvedAddressIsV4Mapped(input6, &output4)); |
||||
// Output is unchanged.
|
||||
ASSERT_EQ(memcmp(expect4.address(), output4.address(), expect4.size()), 0); |
||||
|
||||
// Plain IPv4 input should also fail.
|
||||
EventEngine::ResolvedAddress input4 = MakeAddr4(kIPv4, sizeof(kIPv4)); |
||||
ASSERT_FALSE(ResolvedAddressIsV4Mapped(input4, nullptr)); |
||||
} |
||||
|
||||
TEST(TcpSocketUtilsTest, ResolvedAddressToV4MappedTest) { |
||||
// IPv4 input should succeed.
|
||||
EventEngine::ResolvedAddress input4 = MakeAddr4(kIPv4, sizeof(kIPv4)); |
||||
EventEngine::ResolvedAddress output6; |
||||
ASSERT_TRUE(ResolvedAddressToV4Mapped(input4, &output6)); |
||||
EventEngine::ResolvedAddress expect6 = MakeAddr6(kMapped, sizeof(kMapped)); |
||||
ASSERT_EQ(memcmp(expect6.address(), output6.address(), output6.size()), 0); |
||||
|
||||
// IPv6 input should fail.
|
||||
EventEngine::ResolvedAddress input6 = MakeAddr6(kIPv6, sizeof(kIPv6)); |
||||
ASSERT_TRUE(!ResolvedAddressToV4Mapped(input6, &output6)); |
||||
// Output is unchanged.
|
||||
ASSERT_EQ(memcmp(expect6.address(), output6.address(), output6.size()), 0); |
||||
|
||||
// Already-v4mapped input should also fail.
|
||||
input6 = MakeAddr6(kMapped, sizeof(kMapped)); |
||||
ASSERT_TRUE(!ResolvedAddressToV4Mapped(input6, &output6)); |
||||
} |
||||
|
||||
TEST(TcpSocketUtilsTest, ResolvedAddressToStringTest) { |
||||
errno = 0x7EADBEEF; |
||||
|
||||
EventEngine::ResolvedAddress input4 = MakeAddr4(kIPv4, sizeof(kIPv4)); |
||||
EXPECT_EQ(ResolvedAddressToString(input4).value(), "192.0.2.1:12345"); |
||||
EventEngine::ResolvedAddress input6 = MakeAddr6(kIPv6, sizeof(kIPv6)); |
||||
EXPECT_EQ(ResolvedAddressToString(input6).value(), "[2001:db8::1]:12345"); |
||||
SetIPv6ScopeId(input6, 2); |
||||
EXPECT_EQ(ResolvedAddressToString(input6).value(), "[2001:db8::1%2]:12345"); |
||||
SetIPv6ScopeId(input6, 101); |
||||
EXPECT_EQ(ResolvedAddressToString(input6).value(), "[2001:db8::1%101]:12345"); |
||||
EventEngine::ResolvedAddress input6x = MakeAddr6(kMapped, sizeof(kMapped)); |
||||
EXPECT_EQ(ResolvedAddressToString(input6x).value(), |
||||
"[::ffff:192.0.2.1]:12345"); |
||||
EventEngine::ResolvedAddress input6y = |
||||
MakeAddr6(kNotQuiteMapped, sizeof(kNotQuiteMapped)); |
||||
EXPECT_EQ(ResolvedAddressToString(input6y).value(), |
||||
"[::fffe:c000:263]:12345"); |
||||
EventEngine::ResolvedAddress phony; |
||||
memset(const_cast<sockaddr*>(phony.address()), 0, phony.size()); |
||||
sockaddr* phony_addr = const_cast<sockaddr*>(phony.address()); |
||||
phony_addr->sa_family = 123; |
||||
EXPECT_EQ(ResolvedAddressToString(phony).status(), |
||||
absl::InvalidArgumentError("Unknown sockaddr family: 123")); |
||||
} |
||||
|
||||
TEST(TcpSocketUtilsTest, ResolvedAddressToNormalizedStringTest) { |
||||
errno = 0x7EADBEEF; |
||||
|
||||
EventEngine::ResolvedAddress input4 = MakeAddr4(kIPv4, sizeof(kIPv4)); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(input4).value(), |
||||
"192.0.2.1:12345"); |
||||
EventEngine::ResolvedAddress input6 = MakeAddr6(kIPv6, sizeof(kIPv6)); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(input6).value(), |
||||
"[2001:db8::1]:12345"); |
||||
SetIPv6ScopeId(input6, 2); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(input6).value(), |
||||
"[2001:db8::1%2]:12345"); |
||||
SetIPv6ScopeId(input6, 101); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(input6).value(), |
||||
"[2001:db8::1%101]:12345"); |
||||
EventEngine::ResolvedAddress input6x = MakeAddr6(kMapped, sizeof(kMapped)); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(input6x).value(), |
||||
"192.0.2.1:12345"); |
||||
EventEngine::ResolvedAddress input6y = |
||||
MakeAddr6(kNotQuiteMapped, sizeof(kNotQuiteMapped)); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(input6y).value(), |
||||
"[::fffe:c000:263]:12345"); |
||||
EventEngine::ResolvedAddress phony; |
||||
memset(const_cast<sockaddr*>(phony.address()), 0, phony.size()); |
||||
sockaddr* phony_addr = const_cast<sockaddr*>(phony.address()); |
||||
phony_addr->sa_family = 123; |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(phony).status(), |
||||
absl::InvalidArgumentError("Unknown sockaddr family: 123")); |
||||
|
||||
#ifdef GRPC_HAVE_UNIX_SOCKET |
||||
EventEngine::ResolvedAddress inputun = |
||||
*UnixSockaddrPopulate("/some/unix/path"); |
||||
struct sockaddr_un* sock_un = reinterpret_cast<struct sockaddr_un*>( |
||||
const_cast<sockaddr*>(inputun.address())); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(inputun).value(), |
||||
"/some/unix/path"); |
||||
std::string max_filepath(sizeof(sock_un->sun_path) - 1, 'x'); |
||||
inputun = *UnixSockaddrPopulate(max_filepath); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(inputun).value(), max_filepath); |
||||
inputun = *UnixSockaddrPopulate(max_filepath); |
||||
sock_un->sun_path[sizeof(sockaddr_un::sun_path) - 1] = 'x'; |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(inputun).status(), |
||||
absl::InvalidArgumentError("UDS path is not null-terminated")); |
||||
EventEngine::ResolvedAddress inputun2 = |
||||
*UnixAbstractSockaddrPopulate("some_unix_path"); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(inputun2).value(), |
||||
absl::StrCat(std::string(1, '\0'), "some_unix_path")); |
||||
std::string max_abspath(sizeof(sock_un->sun_path) - 1, '\0'); |
||||
EventEngine::ResolvedAddress inputun3 = |
||||
*UnixAbstractSockaddrPopulate(max_abspath); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(inputun3).value(), |
||||
absl::StrCat(std::string(1, '\0'), max_abspath)); |
||||
#endif |
||||
} |
||||
|
||||
TEST(TcpSocketUtilsTest, SockAddrPortTest) { |
||||
EventEngine::ResolvedAddress wild6 = ResolvedAddressMakeWild6(20); |
||||
EventEngine::ResolvedAddress wild4 = ResolvedAddressMakeWild4(20); |
||||
// Verify the string description matches the expected wildcard address with
|
||||
// correct port number.
|
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(wild6).value(), "[::]:20"); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(wild4).value(), "0.0.0.0:20"); |
||||
// Update the port values.
|
||||
ResolvedAddressSetPort(wild4, 21); |
||||
ResolvedAddressSetPort(wild6, 22); |
||||
// Read back the port values.
|
||||
EXPECT_EQ(ResolvedAddressGetPort(wild4), 21); |
||||
EXPECT_EQ(ResolvedAddressGetPort(wild6), 22); |
||||
// Ensure the string description reflects the updated port values.
|
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(wild4).value(), "0.0.0.0:21"); |
||||
EXPECT_EQ(ResolvedAddressToNormalizedString(wild6).value(), "[::]:22"); |
||||
} |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc_event_engine
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
@ -0,0 +1,170 @@ |
||||
// Copyright 2022 gRPC authors.
|
||||
//
|
||||
// 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 <grpc/support/port_platform.h> |
||||
|
||||
#ifdef GPR_WINDOWS |
||||
|
||||
#include <gtest/gtest.h> |
||||
|
||||
#include "absl/status/status.h" |
||||
|
||||
#include <grpc/event_engine/event_engine.h> |
||||
#include <grpc/grpc.h> |
||||
|
||||
#include "src/core/lib/event_engine/channel_args_endpoint_config.h" |
||||
#include "src/core/lib/event_engine/thread_pool.h" |
||||
#include "src/core/lib/event_engine/windows/iocp.h" |
||||
#include "src/core/lib/event_engine/windows/windows_endpoint.h" |
||||
#include "src/core/lib/gprpp/notification.h" |
||||
#include "src/core/lib/resource_quota/memory_quota.h" |
||||
#include "test/core/event_engine/windows/create_sockpair.h" |
||||
|
||||
namespace grpc_event_engine { |
||||
namespace experimental { |
||||
|
||||
using namespace std::chrono_literals; |
||||
|
||||
class WindowsEndpointTest : public testing::Test {}; |
||||
|
||||
TEST_F(WindowsEndpointTest, BasicCommunication) { |
||||
// TODO(hork): deduplicate against winsocket and iocp tests
|
||||
// Setup
|
||||
ThreadPool executor; |
||||
IOCP iocp(&executor); |
||||
grpc_core::MemoryQuota quota("endpoint_test"); |
||||
SOCKET sockpair[2]; |
||||
CreateSockpair(sockpair, IOCP::GetDefaultSocketFlags()); |
||||
auto wrapped_client_socket = iocp.Watch(sockpair[0]); |
||||
auto wrapped_server_socket = iocp.Watch(sockpair[1]); |
||||
sockaddr_in loopback_addr = GetSomeIpv4LoopbackAddress(); |
||||
EventEngine::ResolvedAddress addr((sockaddr*)&loopback_addr, |
||||
sizeof(loopback_addr)); |
||||
WindowsEndpoint client(addr, std::move(wrapped_client_socket), |
||||
quota.CreateMemoryAllocator("client"), |
||||
ChannelArgsEndpointConfig(), &executor); |
||||
WindowsEndpoint server(addr, std::move(wrapped_server_socket), |
||||
quota.CreateMemoryAllocator("server"), |
||||
ChannelArgsEndpointConfig(), &executor); |
||||
// Test
|
||||
std::string message = "0xDEADBEEF"; |
||||
grpc_core::Notification read_done; |
||||
SliceBuffer read_buffer; |
||||
server.Read( |
||||
[&read_done, &message, &read_buffer](absl::Status status) { |
||||
ASSERT_EQ(read_buffer.Count(), 1); |
||||
auto slice = read_buffer.TakeFirst(); |
||||
EXPECT_EQ(slice.as_string_view(), message); |
||||
read_done.Notify(); |
||||
}, |
||||
&read_buffer, nullptr); |
||||
grpc_core::Notification write_done; |
||||
SliceBuffer write_buffer; |
||||
write_buffer.Append(Slice::FromCopiedString(message)); |
||||
client.Write([&write_done](absl::Status status) { write_done.Notify(); }, |
||||
&write_buffer, nullptr); |
||||
iocp.Work(5s, []() {}); |
||||
// Cleanup
|
||||
write_done.WaitForNotification(); |
||||
read_done.WaitForNotification(); |
||||
executor.Quiesce(); |
||||
} |
||||
|
||||
TEST_F(WindowsEndpointTest, Conversation) { |
||||
// Setup
|
||||
ThreadPool executor; |
||||
IOCP iocp(&executor); |
||||
grpc_core::MemoryQuota quota("endpoint_test"); |
||||
SOCKET sockpair[2]; |
||||
CreateSockpair(sockpair, IOCP::GetDefaultSocketFlags()); |
||||
sockaddr_in loopback_addr = GetSomeIpv4LoopbackAddress(); |
||||
EventEngine::ResolvedAddress addr((sockaddr*)&loopback_addr, |
||||
sizeof(loopback_addr)); |
||||
// Test
|
||||
struct AppState { |
||||
AppState(const EventEngine::ResolvedAddress& addr, |
||||
std::unique_ptr<WinSocket> client, |
||||
std::unique_ptr<WinSocket> server, grpc_core::MemoryQuota& quota, |
||||
Executor& executor) |
||||
: client(addr, std::move(client), quota.CreateMemoryAllocator("client"), |
||||
ChannelArgsEndpointConfig(), &executor), |
||||
server(addr, std::move(server), quota.CreateMemoryAllocator("server"), |
||||
ChannelArgsEndpointConfig(), &executor) {} |
||||
grpc_core::Notification done; |
||||
WindowsEndpoint client; |
||||
WindowsEndpoint server; |
||||
SliceBuffer read_buffer; |
||||
SliceBuffer write_buffer; |
||||
const std::vector<std::string> messages{ |
||||
"Java is to Javascript what car is to carpet. -Heilmann", |
||||
"Make it work, make it right, make it fast. -Beck", |
||||
"First, solve the problem. Then write the code. -Johnson", |
||||
"It works on my machine."}; |
||||
// incremented after a corresponding read of a previous write
|
||||
// if exchange%2 == 0, client -> server
|
||||
// if exchange%2 == 1, server -> client
|
||||
// if exchange == messages.length, done
|
||||
std::atomic<int> exchange{0}; |
||||
|
||||
// Initiates a Write and corresponding Read on two endpoints.
|
||||
void WriteAndQueueReader(WindowsEndpoint* writer, WindowsEndpoint* reader) { |
||||
write_buffer.Clear(); |
||||
write_buffer.Append(Slice::FromCopiedString(messages[exchange])); |
||||
writer->Write([](absl::Status) {}, &write_buffer, /*args=*/nullptr); |
||||
auto cb = [this](absl::Status status) { ReadCB(status); }; |
||||
read_buffer.Clear(); |
||||
reader->Read(cb, &read_buffer, /*args=*/nullptr); |
||||
} |
||||
|
||||
// Asserts that the received string matches, then queues the next Write/Read
|
||||
// pair
|
||||
void ReadCB(absl::Status status) { |
||||
ASSERT_EQ(read_buffer.Count(), 1); |
||||
ASSERT_EQ(read_buffer.TakeFirst().as_string_view(), messages[exchange]); |
||||
if (++exchange == messages.size()) { |
||||
done.Notify(); |
||||
return; |
||||
} |
||||
if (exchange % 2 == 0) { |
||||
WriteAndQueueReader(/*writer=*/&client, /*reader=*/&server); |
||||
} else { |
||||
WriteAndQueueReader(/*writer=*/&server, /*reader=*/&client); |
||||
} |
||||
} |
||||
}; |
||||
AppState state(addr, /*client=*/iocp.Watch(sockpair[0]), |
||||
/*server=*/iocp.Watch(sockpair[1]), quota, executor); |
||||
state.WriteAndQueueReader(/*writer=*/&state.client, /*reader=*/&state.server); |
||||
while (iocp.Work(100ms, []() {}) == Poller::WorkResult::kOk || |
||||
!state.done.HasBeenNotified()) { |
||||
} |
||||
// Cleanup
|
||||
state.done.WaitForNotification(); |
||||
executor.Quiesce(); |
||||
} |
||||
|
||||
} // namespace experimental
|
||||
} // namespace grpc_event_engine
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
grpc_init(); |
||||
int status = RUN_ALL_TESTS(); |
||||
grpc_shutdown(); |
||||
return status; |
||||
} |
||||
|
||||
#else // not GPR_WINDOWS
|
||||
int main(int /* argc */, char** /* argv */) { return 0; } |
||||
#endif |
||||
Loading…
Reference in new issue