diff --git a/src/libs/network.cpp b/src/libs/network.cpp index 8ab8e431f..4bc30e0b6 100644 --- a/src/libs/network.cpp +++ b/src/libs/network.cpp @@ -849,20 +849,38 @@ using SocketIoLength = size_t; #endif struct SocketTransport { - explicit SocketTransport(NativeSocket value): socket(value) {} + SocketTransport(NativeSocket value, int socket_type): socket(value), type(socket_type) {} ~SocketTransport() { Close(); } + bool HasPendingReceive() const { +#if defined(_WIN32) + return buffered_bytes.load(std::memory_order_relaxed) != 0 || + receive_shutdown.load(std::memory_order_relaxed); +#else + return false; +#endif + } int Close() { - if (socket == INVALID_NATIVE_SOCKET) { +#if defined(_WIN32) + receive_shutdown.store(true, std::memory_order_relaxed); +#endif + const auto value = socket.exchange(INVALID_NATIVE_SOCKET); + if (value == INVALID_NATIVE_SOCKET) { return 0; } - const auto value = std::exchange(socket, INVALID_NATIVE_SOCKET); #if defined(_WIN32) return closesocket(value); #else return ::close(value); #endif } - NativeSocket socket; + std::atomic socket; + const int type; +#if defined(_WIN32) + std::mutex receive_mutex; + std::vector receive_buffer; + std::atomic_size_t buffered_bytes = 0; + std::atomic_bool receive_shutdown = false; +#endif }; struct P2pEndpoint { @@ -1164,7 +1182,8 @@ static int ConvertHostSockaddr(const sockaddr_storage* addr, SocketLength addrle return 0; } -static bool GetSocketBackend(int guest_fd, NativeSocket* out, SocketSlot* state = nullptr) { +static bool GetSocketBackend(int guest_fd, NativeSocket* out, SocketSlot* state = nullptr, + std::shared_ptr* transport = nullptr) { EXIT_IF(out == nullptr); if (guest_fd < 0 || guest_fd >= SOCKET_FD_MAX) { @@ -1188,6 +1207,9 @@ static bool GetSocketBackend(int guest_fd, NativeSocket* out, SocketSlot* state if (state != nullptr) { *state = slot; } + if (transport != nullptr) { + *transport = slot.transport; + } return true; } @@ -1234,6 +1256,63 @@ static int AllocSocketFd(const std::shared_ptr& transport, bool return -1; } +static int SelectSockets(int nfds, fd_set* readfds, fd_set* writefds, fd_set* exceptfds, + timeval* timeout, + const std::vector>& stream_reads) { +#if defined(_WIN32) + if (!stream_reads.empty()) { + const auto requested_read = *readfds; + const auto requested_write = writefds != nullptr ? *writefds : fd_set {}; + const auto requested_except = exceptfds != nullptr ? *exceptfds : fd_set {}; + std::optional deadline; + if (timeout != nullptr) { + deadline = std::chrono::steady_clock::now() + std::chrono::seconds(timeout->tv_sec) + + std::chrono::microseconds(timeout->tv_usec); + } + for (;;) { + *readfds = requested_read; + if (writefds != nullptr) { + *writefds = requested_write; + } + if (exceptfds != nullptr) { + *exceptfds = requested_except; + } + const bool ready = std::any_of(stream_reads.begin(), stream_reads.end(), + [](const auto& transport) { return transport->HasPendingReceive(); }); + // A concurrent peek can move native bytes into the guest buffer after this check. + // Bound native waits so both readiness sources are rechecked without holding receive locks. + auto wait = std::chrono::microseconds(ready ? 0 : 10'000); + if (deadline) { + const auto remaining = std::chrono::duration_cast( + *deadline - std::chrono::steady_clock::now()); + wait = std::min(wait, std::max(remaining, std::chrono::microseconds::zero())); + } + timeval slice {}; + slice.tv_usec = static_cast(wait.count()); + int result = ::select(nfds, readfds, writefds, exceptfds, &slice); + if (result < 0) { + return result; + } + for (const auto& transport: stream_reads) { + const auto socket = transport->socket.load(); + if (socket == INVALID_NATIVE_SOCKET) { + WSASetLastError(WSAENOTSOCK); + return SOCKET_ERROR; + } + if (transport->HasPendingReceive() && !FD_ISSET(socket, readfds)) { + FD_SET(socket, readfds); + result++; + } + } + if (result != 0 || (deadline && std::chrono::steady_clock::now() >= *deadline)) { + return result; + } + } + } +#endif + return ::select(nfds, readfds, writefds, exceptfds, timeout); +} + static bool GuestFdIsSet(const void* fds, int fd) { if (fds == nullptr || fd < 0 || fd >= SOCKET_FD_MAX) { return false; @@ -1621,6 +1700,7 @@ int KYTY_SYSV_ABI EpollWait(int eid, NetEpollEvent* events, int maxevents, int t FD_ZERO(&host_except); std::vector host_registrations; + std::vector> stream_reads; host_registrations.reserve(registrations.size()); int host_nfds = 0; for (const auto& registration: registrations) { @@ -1641,6 +1721,11 @@ int KYTY_SYSV_ABI EpollWait(int eid, NetEpollEvent* events, int maxevents, int t #endif if ((registration.event.events & EPOLL_IN) != 0) { FD_SET(socket, &host_read); +#if defined(_WIN32) + if (state.transport->type == SOCK_STREAM) { + stream_reads.push_back(state.transport); + } +#endif } if ((registration.event.events & EPOLL_OUT) != 0) { FD_SET(socket, &host_write); @@ -1660,8 +1745,8 @@ int KYTY_SYSV_ABI EpollWait(int eid, NetEpollEvent* events, int maxevents, int t host_timeout.tv_usec = timeout % 1000000; host_timeout_ptr = &host_timeout; } - - const int result = ::select(host_nfds, &host_read, &host_write, &host_except, host_timeout_ptr); + const int result = SelectSockets(host_nfds, &host_read, &host_write, &host_except, + host_timeout_ptr, stream_reads); if (result < 0) { return SetHostSocketError(); } @@ -1727,7 +1812,12 @@ int KYTY_SYSV_ABI SocketClose(int s) { } RemoveSocketFromEpolls(s); - if (slot.transport.use_count() == 1 && slot.transport->Close() != 0) { + bool close_transport = slot.transport.use_count() == 1; +#if defined(_WIN32) + // Retained receive/readiness state must not defer native TCP close cancellation. + close_transport |= !slot.p2p && slot.transport->type == SOCK_STREAM; +#endif + if (close_transport && slot.transport->Close() != 0) { return NET_ERROR_EBADF; } @@ -1762,7 +1852,7 @@ int KYTY_SYSV_ABI Socket(int family, int type, int protocol) { return SetHostSocketError(); } - auto transport = std::make_shared(socket); + auto transport = std::make_shared(socket, p2p ? SOCK_DGRAM : type); if (p2p) { // Logical sockets keep their own blocking mode; the shared transport never blocks. #if defined(_WIN32) @@ -1945,7 +2035,7 @@ int KYTY_SYSV_ABI Accept(int s, void* addr, uint32_t* addrlen) { return SetHostSocketError(); } - auto transport = std::make_shared(accepted); + auto transport = std::make_shared(accepted, SOCK_STREAM); const int fd = AllocSocketFd(transport); if (fd < 0) { *Posix::GetErrorAddr() = Posix::POSIX_EMFILE; @@ -1974,8 +2064,9 @@ int KYTY_SYSV_ABI Shutdown(int s, int how) { "\t how = %d\n", s, how); - NativeSocket socket = INVALID_NATIVE_SOCKET; - if (!GetSocketBackend(s, &socket)) { + NativeSocket socket = INVALID_NATIVE_SOCKET; + std::shared_ptr transport; + if (!GetSocketBackend(s, &socket, nullptr, &transport)) { return -1; } @@ -1988,6 +2079,9 @@ int KYTY_SYSV_ABI Shutdown(int s, int how) { if (::shutdown(socket, how) == SOCKET_ERROR) { return SetHostSocketError(); } + if (how == SD_RECEIVE || how == SD_BOTH) { + transport->receive_shutdown.store(true, std::memory_order_relaxed); + } return 0; #else @@ -2205,6 +2299,77 @@ int64_t KYTY_SYSV_ABI Sendto(int s, const void* buf, uint64_t len, int flags, co return result; } +#if defined(_WIN32) +static int ReceiveStream(SocketTransport& transport, char* output, int length, int flags, + bool dontwait) { + std::unique_lock lock(transport.receive_mutex, std::defer_lock); + if (dontwait) { + if (!lock.try_lock()) { + return SetGuestSocketError(Posix::POSIX_EWOULDBLOCK); + } + } else { + lock.lock(); + } + if (transport.receive_shutdown.load(std::memory_order_relaxed)) { + return SetGuestSocketError(Posix::POSIX_ESHUTDOWN); + } + auto& buffered = transport.receive_buffer; + const bool peek = (flags & MSG_PEEK) != 0; + const bool waitall = (flags & MSG_WAITALL) != 0; + int copied = static_cast(std::min(buffered.size(), length)); + if (copied != 0) { + std::memcpy(output, buffered.data(), copied); + if (!peek) { + buffered.erase(buffered.begin(), buffered.begin() + copied); + transport.buffered_bytes.store(buffered.size(), std::memory_order_relaxed); + } + } + if (copied == length || (copied != 0 && !waitall)) { + return copied; + } + + // Winsock rejects PEEK | WAITALL. Consume into transport-owned storage when peeking, + // and let ordinary recv report EOF, timeouts, or nonblocking partial progress. + flags &= ~(MSG_PEEK | MSG_WAITALL); + do { + if (transport.receive_shutdown.load(std::memory_order_relaxed)) { + return copied != 0 ? copied : SetGuestSocketError(Posix::POSIX_ESHUTDOWN); + } + const auto socket = transport.socket.load(); + if (dontwait) { + fd_set readable {}; + FD_SET(socket, &readable); + timeval immediate {}; + const int ready = ::select(0, &readable, nullptr, nullptr, &immediate); + if (ready <= 0) { + return copied != 0 ? copied : (ready < 0 ? SetHostSocketError() : + SetGuestSocketError(Posix::POSIX_EWOULDBLOCK)); + } + } + const auto previous_size = buffered.size(); + const int requested = peek ? std::min(length - copied, 64 * 1024) : length - copied; + char* destination = output + copied; + if (peek) { + buffered.resize(previous_size + requested); + destination = buffered.data() + previous_size; + } + const int received = ::recv(socket, destination, requested, flags); + if (peek) { + if (received > 0) { + std::memcpy(output + copied, destination, received); + } + buffered.resize(previous_size + std::max(received, 0)); + transport.buffered_bytes.store(buffered.size(), std::memory_order_relaxed); + } + if (received <= 0) { + return copied != 0 || received == 0 ? copied : SetHostSocketError(); + } + copied += received; + } while (waitall && copied < length); + return copied; +} +#endif + int64_t KYTY_SYSV_ABI Recv(int s, void* buf, uint64_t len, int flags) { return Recvfrom(s, buf, len, flags, nullptr, nullptr); } @@ -2227,8 +2392,9 @@ int64_t KYTY_SYSV_ABI Recvfrom(int s, void* buf, uint64_t len, int flags, void* return -1; } - NativeSocket socket = INVALID_NATIVE_SOCKET; - if (!GetSocketBackend(s, &socket)) { + NativeSocket socket = INVALID_NATIVE_SOCKET; + std::shared_ptr transport; + if (!GetSocketBackend(s, &socket, nullptr, &transport)) { return -1; } @@ -2242,11 +2408,27 @@ int64_t KYTY_SYSV_ABI Recvfrom(int s, void* buf, uint64_t len, int flags, void* sockaddr_storage host_addr {}; SocketLength host_addrlen = sizeof(host_addr); int64_t result = 0; - if (addr == nullptr) { - result = ::recv(socket, static_cast(buf), host_len, host_flags); - } else { - result = ::recvfrom(socket, static_cast(buf), host_len, host_flags, - reinterpret_cast(&host_addr), &host_addrlen); +#if defined(_WIN32) + if (transport->type == SOCK_STREAM) { + if (addr != nullptr && + ::getpeername(socket, reinterpret_cast(&host_addr), &host_addrlen) != 0) { + return SetHostSocketError(); + } + result = ReceiveStream(*transport, static_cast(buf), host_len, host_flags, + (flags & 0x80) != 0); + if (result < 0) { + return result; + } + } else +#endif + { + transport.reset(); + if (addr == nullptr) { + result = ::recv(socket, static_cast(buf), host_len, host_flags); + } else { + result = ::recvfrom(socket, static_cast(buf), host_len, host_flags, + reinterpret_cast(&host_addr), &host_addrlen); + } } #if defined(_WIN32) if (result < 0 && WSAGetLastError() == WSAEMSGSIZE) { @@ -2303,6 +2485,7 @@ int KYTY_SYSV_ABI Select(int nfds, void* readfds, void* writefds, void* exceptfd FD_ZERO(&host_except); std::vector> descriptors; + std::vector> stream_reads; int host_nfds = 0; for (int fd = 0; fd < nfds; fd++) { const bool read = GuestFdIsSet(readfds, fd); @@ -2311,13 +2494,14 @@ int KYTY_SYSV_ABI Select(int nfds, void* readfds, void* writefds, void* exceptfd if (!read && !write && !except) { continue; } - NativeSocket socket = INVALID_NATIVE_SOCKET; + NativeSocket socket = INVALID_NATIVE_SOCKET; + std::shared_ptr transport; #if !defined(_WIN32) if (fd < 3) { socket = fd; } else #endif - if (!GetSocketBackend(fd, &socket)) { + if (!GetSocketBackend(fd, &socket, nullptr, &transport)) { return -1; } #if defined(_WIN32) @@ -2335,6 +2519,11 @@ int KYTY_SYSV_ABI Select(int nfds, void* readfds, void* writefds, void* exceptfd descriptors.emplace_back(fd, socket); if (read) { FD_SET(socket, &host_read); +#if defined(_WIN32) + if (transport->type == SOCK_STREAM) { + stream_reads.push_back(std::move(transport)); + } +#endif } if (write) { FD_SET(socket, &host_write); @@ -2365,9 +2554,10 @@ int KYTY_SYSV_ABI Select(int nfds, void* readfds, void* writefds, void* exceptfd } else #endif { - result = ::select(host_nfds, readfds != nullptr ? &host_read : nullptr, - writefds != nullptr ? &host_write : nullptr, - exceptfds != nullptr ? &host_except : nullptr, host_timeout_ptr); + result = SelectSockets(host_nfds, readfds != nullptr ? &host_read : nullptr, + writefds != nullptr ? &host_write : nullptr, + exceptfds != nullptr ? &host_except : nullptr, host_timeout_ptr, + stream_reads); if (result < 0) { return SetHostSocketError(); } @@ -2376,15 +2566,19 @@ int KYTY_SYSV_ABI Select(int nfds, void* readfds, void* writefds, void* exceptfd GuestFdZero(readfds, nfds); GuestFdZero(writefds, nfds); GuestFdZero(exceptfds, nfds); + result = 0; for (const auto& [fd, socket]: descriptors) { if (FD_ISSET(socket, &host_read)) { GuestFdSet(readfds, fd); + result++; } if (FD_ISSET(socket, &host_write)) { GuestFdSet(writefds, fd); + result++; } if (FD_ISSET(socket, &host_except)) { GuestFdSet(exceptfds, fd); + result++; } } return result; diff --git a/tests/KernelFileSystemTests.cpp b/tests/KernelFileSystemTests.cpp index 1faf2ecd9..9253c519f 100644 --- a/tests/KernelFileSystemTests.cpp +++ b/tests/KernelFileSystemTests.cpp @@ -25,10 +25,12 @@ #include #include #include +#include #include #include #include #include +#include #include namespace Libs::LibKernelApr { @@ -811,6 +813,210 @@ void CheckAprPaths(const std::filesystem::path &root) { FileSystem::Umount("/app0"); } +std::array CreateTcpPair() { + namespace Net = Libs::Network::Net; + // Guest sockaddr_in: length, family, network-order port/address, padding. + std::array address {16, 2, 0, 0, 127, 0, 0, 1}; + const int listener = Net::Socket(2, 1, 0); + Check(listener >= 0, "create loopback listener"); + Check(Net::Bind(listener, address.data(), address.size()) == 0, "bind loopback"); + Check(Net::Listen(listener, 1) == 0, "listen on loopback"); + uint32_t address_size = address.size(); + Check(Net::Getsockname(listener, address.data(), &address_size) == 0, + "get assigned loopback port"); + const int writer = Net::Socket(2, 1, 0); + Check(writer >= 0 && Net::Connect(writer, address.data(), address_size) == 0, + "connect wake socket"); + const int reader = Net::Accept(listener, nullptr, nullptr); + Check(reader >= 0, "accept wake socket"); + Check(Net::SocketClose(listener) == 0, "close listener"); + return {reader, writer}; +} + +#if defined(_WIN32) +void CheckSocketReceiveConcurrency() { + namespace Net = Libs::Network::Net; + const auto [reader, writer] = CreateTcpPair(); + const int timeout_ms = 3000; + Check(Net::Setsockopt(reader, 0xffff, 0x1006, &timeout_ms, sizeof(timeout_ms)) == 0, + "bound the blocked receive concurrency fixture"); + std::promise entered; + std::promise completed; + auto entered_future = entered.get_future(); + auto completed_future = completed.get_future(); + std::jthread receiver([&] { + char byte = 0; + entered.set_value(); + completed.set_value(Net::Recv(reader, &byte, 1, 0x42)); + }); + Check(entered_future.wait_for(std::chrono::seconds(2)) == std::future_status::ready, + "start blocked reader"); + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + char byte = 0; + const auto before_nonblocking = std::chrono::steady_clock::now(); + Check(Net::Recv(reader, &byte, 1, 0x80) == -1 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_EWOULDBLOCK && + std::chrono::steady_clock::now() - before_nonblocking < std::chrono::milliseconds(500) && + completed_future.wait_for(std::chrono::milliseconds(0)) == std::future_status::timeout, + "DONTWAIT does not wait behind another blocked reader"); + const auto before_close = std::chrono::steady_clock::now(); + Check(Net::SocketClose(reader) == 0, "close succeeds while another thread receives"); + Check(std::chrono::steady_clock::now() - before_close < std::chrono::seconds(1), + "close does not wait for the blocked receive timeout"); + Check(completed_future.wait_for(std::chrono::seconds(1)) == std::future_status::ready, + "close wakes the blocked receive within one second"); + Check(completed_future.get() <= 0, "closed receive reports EOF or error"); + Check(std::chrono::steady_clock::now() - before_close < std::chrono::seconds(1), + "closing a socket wakes its blocked receive before the receive timeout"); + receiver.join(); + Check(Net::SocketClose(writer) == 0, "close blocked receive writer"); + + const auto [shutdown_reader, shutdown_writer] = CreateTcpPair(); + Check(Net::Setsockopt(shutdown_reader, 0xffff, 0x1006, &timeout_ms, sizeof(timeout_ms)) == 0 && + Net::Send(shutdown_writer, "x", 1, 0) == 1 && + Net::Recv(shutdown_reader, &byte, 1, 0x42) == 1 && byte == 'x' && + Net::Shutdown(shutdown_reader, 0) == 0 && + Net::Recv(shutdown_reader, &byte, 1, 0) == -1 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_ESHUTDOWN, + "receive shutdown rejects bytes retained by an earlier peek"); + Check(Net::SocketClose(shutdown_reader) == 0 && Net::SocketClose(shutdown_writer) == 0, + "close receive shutdown fixture"); +} + +void CheckSocketReceiveBuffer(int reader, int writer) { + namespace Net = Libs::Network::Net; + constexpr int PeekWaitAll = 0x42; + constexpr char Payload[] = "abcdef"; + const int timeout_ms = 2000; + Check(Net::Setsockopt(reader, 0xffff, 0x1006, &timeout_ms, sizeof(timeout_ms)) == 0, + "bound blocking receive regressions with SO_RCVTIMEO"); + Check(Net::Send(writer, Payload, 2, 0) == 2, "send first TCP fragment"); + std::jthread sender([&] { + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + Check(Net::Send(writer, Payload + 2, 4, 0) == 4, "send second TCP fragment"); + }); + std::array received {}; + Check(Net::Recv(reader, received.data(), 6, PeekWaitAll) == 6 && + std::memcmp(received.data(), Payload, 6) == 0, + "combined peek waits for fragmented TCP bytes"); + sender.join(); + received.fill(0); + Check(Net::Recv(reader, received.data(), 6, PeekWaitAll) == 6 && + std::memcmp(received.data(), Payload, 6) == 0, + "repeated combined peek preserves all bytes"); + + const int epoll = Net::EpollCreate("buffered receive", 0); + Net::NetEpollEvent registration {}; + registration.events = 1; + registration.data.u64 = 0x12345678; + Check(epoll >= 0 && Net::EpollControl(epoll, 1, reader, ®istration) == 0, + "register buffered socket readability"); + const auto check_readable = [&](bool expected) { + std::array readable {}; + const auto bit = uint64_t {1} << (reader % 64); + readable[reader / 64] = bit; + const std::array timeout {expected ? 1 : 0, 0}; + Check(Net::Select(reader + 1, readable.data(), nullptr, nullptr, timeout.data()) == + (expected ? 1 : 0) && readable[reader / 64] == (expected ? bit : 0), + "select reflects buffered bytes and their removal"); + if (expected) { + auto writable = readable; + Check(Net::Select(reader + 1, readable.data(), writable.data(), nullptr, + timeout.data()) == 2 && readable[reader / 64] == bit && + writable[reader / 64] == bit, + "select counts readable and writable bits once each"); + } + Net::NetEpollEvent event {}; + Check(Net::EpollWait(epoll, &event, 1, expected ? 1000000 : 0) == + (expected ? 1 : 0) && + (!expected || (event.events == 1 && event.data.u64 == registration.data.u64)), + "epoll reflects buffered bytes and preserves registration data"); + }; + check_readable(true); + std::array peer {}, expected_peer {}; + uint32_t peer_size = peer.size(), expected_peer_size = expected_peer.size(); + Check(Net::Getsockname(writer, expected_peer.data(), &expected_peer_size) == 0 && + Net::Recv(reader, received.data(), 0, PeekWaitAll) == 0 && + Net::Recvfrom(reader, received.data(), 2, 0, peer.data(), &peer_size) == 2 && + peer_size == expected_peer_size && peer == expected_peer && + std::memcmp(received.data(), Payload, 2) == 0 && + FileSystem::KernelRead(reader, received.data(), 2) == 2 && + std::memcmp(received.data(), Payload + 2, 2) == 0, + "zero-length peek preserves bytes; recvfrom returns peer and shares kernel read buffer"); + check_readable(true); + Check(Net::Send(writer, "gh", 2, 0) == 2 && + Net::Recv(reader, received.data(), 4, 0x40) == 4 && + std::memcmp(received.data(), "efgh", 4) == 0, + "waitall joins buffered bytes with newly received bytes"); + check_readable(false); + Check(Net::Send(writer, "mn", 2, 0) == 2, "send plain peek fixture"); + const auto before_peek = std::chrono::steady_clock::now(); + Check(Net::Recv(reader, received.data(), 4, 2) == 2 && + std::memcmp(received.data(), "mn", 2) == 0 && + std::chrono::steady_clock::now() - before_peek < std::chrono::milliseconds(500) && + Net::Recv(reader, received.data(), 4, 0) == 2 && + std::memcmp(received.data(), "mn", 2) == 0, + "plain peek returns available bytes promptly and preserves them for ordinary receive"); + + const int enabled = 1, disabled = 0; + Check(Net::Setsockopt(reader, 0xffff, 0x1200, &enabled, sizeof(enabled)) == 0 && + Net::Recv(reader, received.data(), 4, PeekWaitAll) == -1 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_EWOULDBLOCK, + "empty nonblocking combined peek returns would-block"); + Check(Net::Send(writer, "ij", 2, 0) == 2, "send nonblocking receive fixture"); + check_readable(true); + *Libs::Posix::GetErrorAddr() = Libs::Posix::POSIX_EINVAL; + for (int i = 0; i < 2; ++i) { + Check(Net::Recv(reader, received.data(), 4, PeekWaitAll) == 2 && + std::memcmp(received.data(), "ij", 2) == 0 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_EINVAL, + "nonblocking combined peek preserves a short result and successful errno"); + } + Check(Net::Recv(reader, received.data(), 4, 0) == 2 && + std::memcmp(received.data(), "ij", 2) == 0, + "ordinary receive consumes a short buffered result"); + check_readable(false); + Check(Net::Setsockopt(reader, 0xffff, 0x1200, &disabled, sizeof(disabled)) == 0 && + Net::EpollDestroy(epoll) == 0, + "restore blocking receives and destroy epoll"); + Check(Net::Recv(reader, received.data(), 4, PeekWaitAll | 0x80) == -1 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_EWOULDBLOCK, + "per-call nonblocking combined peek returns would-block on a blocking socket"); + Check(Net::Send(writer, "kl", 2, 0) == 2 && Net::Shutdown(writer, 1) == 0, + "send a final partial payload and EOF"); + for (int i = 0; i < 2; ++i) { + Check(Net::Recv(reader, received.data(), 4, PeekWaitAll) == 2 && + std::memcmp(received.data(), "kl", 2) == 0, + "EOF ends waitall while repeated peeks preserve the short payload"); + } + Check(Net::Recv(reader, received.data(), 4, 0x40) == 2 && + std::memcmp(received.data(), "kl", 2) == 0 && + Net::Recv(reader, received.data(), 4, PeekWaitAll) == 0, + "draining the final buffered payload exposes EOF"); + + const int datagram = Net::Socket(2, 2, 0); + const int datagram_writer = Net::Socket(2, 2, 0); + std::array address {16, 2, 0, 0, 127, 0, 0, 1}; + uint32_t address_size = address.size(); + Check(datagram >= 0 && datagram_writer >= 0 && + Net::Bind(datagram, address.data(), address.size()) == 0 && + Net::Getsockname(datagram, address.data(), &address_size) == 0 && + Net::Setsockopt(datagram, 0xffff, 0x1006, &timeout_ms, sizeof(timeout_ms)) == 0 && + Net::Sendto(datagram_writer, Payload, 6, 0, address.data(), address_size) == 6, + "create bounded UDP truncation fixture"); + *Libs::Posix::GetErrorAddr() = Libs::Posix::POSIX_EINVAL; + Check(Net::Recvfrom(datagram, received.data(), 2, 0, nullptr, nullptr) == 2 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_EINVAL && + std::memcmp(received.data(), Payload, 2) == 0 && + Net::Setsockopt(datagram, 0xffff, 0x1200, &enabled, sizeof(enabled)) == 0 && + Net::Recv(datagram, received.data(), received.size(), 0) == -1 && + *Libs::Posix::GetErrorAddr() == Libs::Posix::POSIX_EWOULDBLOCK, + "UDP truncation returns its prefix and discards the rest of the datagram"); + Check(Net::SocketClose(datagram) == 0 && Net::SocketClose(datagram_writer) == 0, + "close UDP truncation fixture"); +} +#endif + void CheckSocketWakeup() { namespace Net = Libs::Network::Net; Loader::SymbolDatabase symbols; @@ -829,21 +1035,7 @@ void CheckSocketWakeup() { const auto net_send = reinterpret_cast(send_symbol->vaddr); const auto net_recv = reinterpret_cast(recv_symbol->vaddr); auto *net_errno = reinterpret_cast(errno_symbol->vaddr)(); - // Guest sockaddr_in: length, family, network-order port/address, padding. - std::array address {16, 2, 0, 0, 127, 0, 0, 1}; - const int listener = Net::Socket(2, 1, 0); - Check(listener >= 0, "create loopback listener"); - Check(Net::Bind(listener, address.data(), address.size()) == 0, "bind loopback"); - Check(Net::Listen(listener, 1) == 0, "listen on loopback"); - uint32_t address_size = address.size(); - Check(Net::Getsockname(listener, address.data(), &address_size) == 0, - "get assigned loopback port"); - const int writer = Net::Socket(2, 1, 0); - Check(writer >= 0 && Net::Connect(writer, address.data(), address_size) == 0, - "connect wake socket"); - const int reader = Net::Accept(listener, nullptr, nullptr); - Check(reader >= 0, "accept wake socket"); - Check(Net::SocketClose(listener) == 0, "close listener"); + const auto [reader, writer] = CreateTcpPair(); const int enabled = 1; Check(Net::Setsockopt(writer, 6, 1, &enabled, sizeof(enabled)) == 0, "enable TCP_NODELAY"); @@ -911,6 +1103,10 @@ void CheckSocketWakeup() { *net_errno == Libs::Posix::POSIX_EPIPE, "Net send reports a broken pipe without raising host SIGPIPE"); Check(Net::SocketClose(disconnected) == 0, "close unconnected socket"); +#endif +#if defined(_WIN32) + CheckSocketReceiveBuffer(reader, writer); + CheckSocketReceiveConcurrency(); #endif Check(Net::SocketClose(writer) == 0, "close wake writer"); *net_errno = Libs::Posix::POSIX_EINVAL;