diff --git a/CMakeLists.txt b/CMakeLists.txt index 0caa8a7c..f0f5f610 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -14,6 +14,13 @@ option(MOBILEGL_ENABLE_TRACY "Enable tracy for profiling" option(MOBILEGL_BUILD_TRACE_REPLAY "Build desktop apitrace replay runner" OFF) option(MOBILEGL_TRACE_ANGLE_VARIANTS "Enable signed trace-APK ANGLE variant loading" OFF) option(MOBILEGL_IOS "Build MobileGL for iOS instead of macOS when APPLE is set" OFF) +# The disaggregated (two-process) shape. OFF is the shipping default and OFF +# must stay byte-comparable to a tree without MG_Remote at all: nothing under +# MobileGL/MG_Remote/ is compiled, no include path is added, and no library is +# linked, so `nm --defined-only libMobileGL.so | grep -i MG_Remote` is empty. +# That emptiness is one of the two byte-level equalities the plan's validation +# gates keep (section 10.3). +option(MOBILEGL_BUILD_DISAGGREGATED "Build the MG_Remote transport layer (two-process shape)" OFF) set(MOBILEGL_LOG_ACTIVE_LEVEL "MOBILEGL_LOG_LEVEL_INFO" CACHE STRING "MobileGL active log level macro") set(MOBILEGL_VULKAN_LIBRARY "" CACHE FILEPATH "Vulkan loader/MoltenVK library to link for iOS builds") @@ -419,6 +426,41 @@ set(SOURCE_FILES MobileGL/MG_State/GLState/RenderbufferState/RenderbufferState.cpp ) +# --------------------------------------------------------------------------- +# MG_Remote (disaggregated transport). Everything below is gated: with the +# option OFF not one file here is compiled and no include path is added. +# --------------------------------------------------------------------------- + +# FlatBuffers is a submodule and its runtime is header-only. Guard both ways: +# a checkout without the submodule must configure and build, just without the +# disaggregated shape, rather than fail with a missing-header error a hundred +# lines later. Note this only checks for the RUNTIME headers - flatc is never +# built here (see scripts/gen_protocol.py). +if (MOBILEGL_BUILD_DISAGGREGATED AND + NOT EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/3rdparty/flatbuffers/include/flatbuffers/flatbuffers.h") + message(WARNING + "MOBILEGL_BUILD_DISAGGREGATED=ON but 3rdparty/flatbuffers/include is missing. " + "Run `git submodule update --init 3rdparty/flatbuffers`. Forcing the option OFF.") + set(MOBILEGL_BUILD_DISAGGREGATED OFF CACHE BOOL + "Build the MG_Remote transport layer (two-process shape)" FORCE) +endif() + +if (MOBILEGL_BUILD_DISAGGREGATED) + message(STATUS "MobileGL: disaggregated transport ON, appending MG_Remote sources") + list(APPEND SOURCE_FILES + MobileGL/MG_Remote/Transport/Ring.cpp + MobileGL/MG_Remote/Transport/Doorbell.cpp + MobileGL/MG_Remote/Transport/ShmSegment.cpp + # Both platform halves are listed unconditionally and each is empty on + # the other OS, so neither can rot behind an `if (WIN32)` nobody + # configures. + MobileGL/MG_Remote/Transport/ShmSegmentPosix.cpp + MobileGL/MG_Remote/Transport/ShmSegmentWin32.cpp + MobileGL/MG_Remote/Transport/FdPassing.cpp + MobileGL/MG_Remote/Transport/InProcessTransport.cpp + ) +endif() + if (APPLE AND NOT MOBILEGL_IOS) list(APPEND SOURCE_FILES MobileGL/MG_Impl/CGLImpl/CGLImpl.cpp @@ -469,6 +511,10 @@ set(MOBILEGL_COMPILE_DEF -DASIO_NO_DEPRECATED ) +if (MOBILEGL_BUILD_DISAGGREGATED) + list(APPEND MOBILEGL_COMPILE_DEF -DMOBILEGL_BUILD_DISAGGREGATED=1) +endif() + message(STATUS "MOBILEGL_COMPILE_DEF=${MOBILEGL_COMPILE_DEF}") set(MOBILEGL_INCLUDE_DIR @@ -488,6 +534,13 @@ set(MOBILEGL_INCLUDE_DIR ${CMAKE_SOURCE_DIR}/3rdparty/asio/include ) +if (MOBILEGL_BUILD_DISAGGREGATED) + # Header-only runtime: an include path, no add_subdirectory, no link + # target, and above all no flatc in the build graph. protocol_generated.h + # is committed and regenerated by scripts/gen_protocol.py. + list(APPEND MOBILEGL_INCLUDE_DIR ${CMAKE_SOURCE_DIR}/3rdparty/flatbuffers/include) +endif() + add_library(${CMAKE_PROJECT_NAME} SHARED ${SOURCE_FILES} ) diff --git a/MobileGL/MG_Remote/Transport/Doorbell.cpp b/MobileGL/MG_Remote/Transport/Doorbell.cpp new file mode 100644 index 00000000..1d02da81 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/Doorbell.cpp @@ -0,0 +1,168 @@ +// MobileGL - MobileGL/MG_Remote/Transport/Doorbell.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +#include "Doorbell.h" + +#include + +#include +#include + +#if !defined(_WIN32) +#include +#include +#include +#include +#endif + +namespace MobileGL::MG_Remote::Transport { + + // ----------------------------------------------------------------------- + // CondVarDoorbell + // ----------------------------------------------------------------------- + + struct CondVarDoorbell::Impl { + std::mutex mutex; + std::condition_variable cv; + // Counted, not a flag: a wakeup that arrives while nobody is parked + // must still be observed by the next Park. + std::uint32_t signals = 0; + }; + + CondVarDoorbell::CondVarDoorbell() : m_impl(new Impl()) {} + + CondVarDoorbell::~CondVarDoorbell() { delete m_impl; } + + void CondVarDoorbell::Notify() { + { + std::lock_guard lock(m_impl->mutex); + ++m_impl->signals; + } + m_impl->cv.notify_one(); + } + + bool CondVarDoorbell::Park(std::uint32_t timeoutMs) { + std::unique_lock lock(m_impl->mutex); + if (m_impl->signals != 0) { + --m_impl->signals; + return true; + } + if (timeoutMs == 0) { + return false; + } + if (timeoutMs == kWaitForever) { + m_impl->cv.wait(lock, [this] { return m_impl->signals != 0; }); + } else if (!m_impl->cv.wait_for(lock, std::chrono::milliseconds(timeoutMs), + [this] { return m_impl->signals != 0; })) { + return false; + } + --m_impl->signals; + return true; + } + + void CondVarDoorbell::Reset() { + std::lock_guard lock(m_impl->mutex); + m_impl->signals = 0; + } + +#if !defined(_WIN32) + + // ----------------------------------------------------------------------- + // SocketDoorbell + // ----------------------------------------------------------------------- + + SocketDoorbell::SocketDoorbell(int fd, std::uint8_t code, bool ownsFd) + : m_fd(fd), m_code(code), m_ownsFd(ownsFd) {} + + SocketDoorbell::~SocketDoorbell() { + if (m_ownsFd && m_fd >= 0) { + ::close(m_fd); + } + } + + void SocketDoorbell::Notify() { + if (m_fd < 0) { + return; + } + const std::uint8_t byte = m_code; + for (;;) { + const ssize_t written = ::send(m_fd, &byte, 1, MSG_DONTWAIT | MSG_NOSIGNAL); + if (written == 1) { + return; + } + if (written < 0 && errno == EINTR) { + continue; + } + if (written < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + // The socket buffer already holds unread wakeups: the peer has + // one pending, which is all a doorbell promises. + return; + } + if (written < 0 && errno == EPIPE) { + return; // peer gone; the waiter learns it from its own read + } + MGLOG_D("MG_Remote doorbell: send failed (errno=%d)", errno); + return; + } + } + + bool SocketDoorbell::Park(std::uint32_t timeoutMs) { + if (m_fd < 0) { + return false; + } + const auto start = std::chrono::steady_clock::now(); + for (;;) { + int pollTimeout = -1; + if (timeoutMs != kWaitForever) { + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + const long long remaining = static_cast(timeoutMs) - elapsed; + pollTimeout = remaining <= 0 ? 0 : static_cast(remaining); + } + struct pollfd pfd{}; + pfd.fd = m_fd; + pfd.events = POLLIN; + const int ready = ::poll(&pfd, 1, pollTimeout); + if (ready < 0) { + if (errno == EINTR) { + continue; // a signal is not a wakeup; keep the deadline + } + MGLOG_D("MG_Remote doorbell: poll failed (errno=%d)", errno); + return false; + } + if (ready == 0) { + return false; // timed out + } + Reset(); + return true; + } + } + + void SocketDoorbell::Reset() { + if (m_fd < 0) { + return; + } + // Level-triggered to edge-triggered: swallow every queued byte so one + // stale wakeup cannot make later Parks return without an event. + std::uint8_t scratch[64]; + for (;;) { + const ssize_t got = ::recv(m_fd, scratch, sizeof(scratch), MSG_DONTWAIT); + if (got > 0) { + continue; + } + if (got < 0 && errno == EINTR) { + continue; + } + return; + } + } + +#endif // !_WIN32 + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/Doorbell.h b/MobileGL/MG_Remote/Transport/Doorbell.h new file mode 100644 index 00000000..78fc48c4 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/Doorbell.h @@ -0,0 +1,189 @@ +// MobileGL - MobileGL/MG_Remote/Transport/Doorbell.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// The bidirectional doorbell: spin briefly, then park. +// +// Both directions exist, and that is the point (inherited design, earlier plan +// section 6.2a): +// - client -> server: the consumer spins, sets consumerParked, then blocks; +// the producer rings only when consumerParked is set. +// - server -> client: the client spins MOBILEGL_IPC_SPIN_US (default 50us), +// sets producerParked, then blocks; the server rings after advancing any +// watermark, only when producerParked is set. +// Without the second direction every client wait - present credit, a blocking +// kNeedsAck request, a full ring - degenerates into a cross-process spin on +// one shared cache line: up to a whole frame of a big core at full clock on a +// phone, fighting the GPU and the game's JVM for it. MobileGL has no affinity +// control anywhere in the tree, so it cannot even be pushed to a little core. +// +// Two implementations, no platform-specific wakeup primitive (no futex, no +// eventfd, no named event): +// - CondVarDoorbell for `inproc` (one process, two threads), +// - SocketDoorbell for `spawn` (one byte on a socket; POSIX only). +// +// The lost-wakeup window is closed by ordering, not by luck: the waiter stores +// its park flag and THEN re-tests the condition, while the notifier publishes +// the watermark and THEN tests the park flag. Both use seq_cst on those two +// accesses, so at least one of the two sees the other. + +#pragma once + +#include +#include +#include + +#if defined(__x86_64__) || defined(__i386__) +#include +#endif + +namespace MobileGL::MG_Remote::Transport { + + // MOBILEGL_IPC_SPIN_US default. + inline constexpr std::uint32_t kDefaultSpinUs = 50; + + // Park with no deadline. + inline constexpr std::uint32_t kWaitForever = 0xFFFFFFFFu; + + // Wire codes, so a shared socket can carry both directions distinguishably. + inline constexpr std::uint8_t kDoorbellRingAdvanced = 0x01; // client -> server + inline constexpr std::uint8_t kDoorbellWatermarkAdvanced = 0x02; // server -> client + + inline void CpuRelax() { +#if defined(__x86_64__) || defined(__i386__) + _mm_pause(); +#elif defined(__aarch64__) || defined(__arm__) + __asm__ __volatile__("yield" ::: "memory"); +#else + std::atomic_signal_fence(std::memory_order_seq_cst); +#endif + } + + class Doorbell { + public: + virtual ~Doorbell() = default; + + Doorbell(const Doorbell&) = delete; + Doorbell& operator=(const Doorbell&) = delete; + + // Wakes a parked peer. Cheap and idempotent: a wakeup that arrives when + // nobody is parked is remembered, so the next Park returns immediately + // rather than sleeping through an event that already happened. + virtual void Notify() = 0; + + // Blocks until notified or the deadline passes. Returns true when a + // wakeup was consumed. timeoutMs == 0 polls; kWaitForever never times + // out. + virtual bool Park(std::uint32_t timeoutMs) = 0; + + // Drops pending wakeups. Used when a waiter gives up, so a stale byte + // does not make the next Park return spuriously forever. + virtual void Reset() = 0; + + // Spin `spinUs`, then park until `ready()` or the deadline. + // `parked` is the RingControl flag the peer tests before ringing. + template + bool Wait(std::atomic& parked, Ready&& ready, std::uint32_t spinUs, + std::uint32_t timeoutMs) { + if (ready()) { + return true; + } + const auto start = std::chrono::steady_clock::now(); + const auto deadline = timeoutMs == kWaitForever + ? std::chrono::steady_clock::time_point::max() + : start + std::chrono::milliseconds(timeoutMs); + + const auto spinEnd = start + std::chrono::microseconds(spinUs); + while (std::chrono::steady_clock::now() < spinEnd) { + if (ready()) { + return true; + } + CpuRelax(); + } + + for (;;) { + // Announce, THEN re-test: the notifier publishes and then reads + // this flag, so one of the two orderings always sees the other. + parked.store(1, std::memory_order_seq_cst); + if (ready()) { + parked.store(0, std::memory_order_seq_cst); + return true; + } + const auto now = std::chrono::steady_clock::now(); + if (now >= deadline) { + parked.store(0, std::memory_order_seq_cst); + return ready(); + } + std::uint32_t chunkMs = kWaitForever; + if (timeoutMs != kWaitForever) { + const auto remaining = + std::chrono::duration_cast(deadline - now).count(); + chunkMs = remaining <= 0 ? 0 : static_cast(remaining); + } + Park(chunkMs); + parked.store(0, std::memory_order_seq_cst); + if (ready()) { + return true; + } + if (timeoutMs != kWaitForever && std::chrono::steady_clock::now() >= deadline) { + return false; + } + } + } + + protected: + Doorbell() = default; + }; + + // Rings `bell` only when the peer said it is parked. The seq_cst load pairs + // with the waiter's seq_cst store of the same flag. + inline void NotifyIfParked(Doorbell& bell, std::atomic& parked) { + if (parked.load(std::memory_order_seq_cst) != 0) { + bell.Notify(); + } + } + + // `inproc`: one process, two threads. + class CondVarDoorbell final : public Doorbell { + public: + CondVarDoorbell(); + ~CondVarDoorbell() override; + + void Notify() override; + bool Park(std::uint32_t timeoutMs) override; + void Reset() override; + + private: + struct Impl; + Impl* m_impl; + }; + +#if !defined(_WIN32) + // `spawn`: one byte on a socket (one direction of a socketpair, or the aux + // socket). POSIX only; the Windows path will use an overlapped named pipe + // and is not part of this skeleton. + class SocketDoorbell final : public Doorbell { + public: + // `fd` must be a socket or pipe end. When `ownsFd` the descriptor is + // closed with this object. `code` is the byte written by Notify. + SocketDoorbell(int fd, std::uint8_t code, bool ownsFd); + ~SocketDoorbell() override; + + void Notify() override; + bool Park(std::uint32_t timeoutMs) override; + void Reset() override; + + int Fd() const { return m_fd; } + + private: + int m_fd; + std::uint8_t m_code; + bool m_ownsFd; + }; +#endif + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/FdPassing.cpp b/MobileGL/MG_Remote/Transport/FdPassing.cpp new file mode 100644 index 00000000..22991177 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/FdPassing.cpp @@ -0,0 +1,296 @@ +// MobileGL - MobileGL/MG_Remote/Transport/FdPassing.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +#include "FdPassing.h" + +#include + +#include +#include + +#if !defined(_WIN32) +#include +#include +#include +#include +#include +#endif + +namespace MobileGL::MG_Remote::Transport::FdPassing { + +#if defined(_WIN32) + + bool Supported() { return false; } + + MobileGLResult CreateSocketPair(int[2]) { return MOBILEGL_ERR_UNSUPPORTED; } + + MobileGLResult SendFd(int, int, MobileGLByteSpan) { return MOBILEGL_ERR_UNSUPPORTED; } + + MobileGLResult ReceiveFd(int, int*, MobileGLMutableByteSpan, std::uint64_t*, std::uint32_t) { + return MOBILEGL_ERR_UNSUPPORTED; + } + +#else + + namespace { + // Every datagram starts with this, so the sideband length is explicit + // and a stray datagram is recognisable. + struct SidebandHeader { + std::uint32_t magic; + std::uint32_t sidebandSize; + }; + constexpr std::uint32_t kSidebandMagic = 0x4446474Du; // 'MGFD' on the wire + + int WaitReadable(int socket, std::uint32_t timeoutMs) { + const auto start = std::chrono::steady_clock::now(); + for (;;) { + int pollTimeout = -1; + if (timeoutMs != 0xFFFFFFFFu) { + const auto elapsed = std::chrono::duration_cast( + std::chrono::steady_clock::now() - start) + .count(); + const long long remaining = static_cast(timeoutMs) - elapsed; + pollTimeout = remaining <= 0 ? 0 : static_cast(remaining); + } + struct pollfd pfd{}; + pfd.fd = socket; + pfd.events = POLLIN; + const int ready = ::poll(&pfd, 1, pollTimeout); + if (ready < 0 && errno == EINTR) { + continue; + } + return ready; + } + } + } // namespace + + bool Supported() { return true; } + + MobileGLResult CreateSocketPair(int outFds[2]) { + if (outFds == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + int fds[2] = {-1, -1}; + int type = SOCK_DGRAM; +#if defined(SOCK_CLOEXEC) + type |= SOCK_CLOEXEC; +#endif + if (::socketpair(AF_UNIX, type, 0, fds) != 0) { + MGLOG_E("MG_Remote fd passing: socketpair failed (errno=%d)", errno); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + outFds[0] = fds[0]; + outFds[1] = fds[1]; + return MOBILEGL_OK; + } + + MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband) { + if (socket < 0 || fd < 0) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + if (sideband.size > kMaxSidebandBytes || (sideband.size != 0 && sideband.data == nullptr)) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + + std::uint8_t payload[sizeof(SidebandHeader) + kMaxSidebandBytes]; + SidebandHeader header{}; + header.magic = kSidebandMagic; + header.sidebandSize = static_cast(sideband.size); + std::memcpy(payload, &header, sizeof(header)); + if (sideband.size != 0) { + std::memcpy(payload + sizeof(header), sideband.data, + static_cast(sideband.size)); + } + const std::size_t payloadSize = sizeof(header) + static_cast(sideband.size); + + struct iovec iov{}; + iov.iov_base = payload; + iov.iov_len = payloadSize; + + // CMSG_SPACE, not sizeof: the control buffer has to hold the aligned + // cmsghdr as well as the descriptor. + union { + struct cmsghdr align; + char bytes[CMSG_SPACE(sizeof(int))]; + } control{}; + std::memset(&control, 0, sizeof(control)); + + struct msghdr msg{}; + msg.msg_iov = &iov; + msg.msg_iovlen = 1; + msg.msg_control = control.bytes; + msg.msg_controllen = sizeof(control.bytes); + + struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg); + cmsg->cmsg_level = SOL_SOCKET; + cmsg->cmsg_type = SCM_RIGHTS; + cmsg->cmsg_len = CMSG_LEN(sizeof(int)); + std::memcpy(CMSG_DATA(cmsg), &fd, sizeof(fd)); + + for (;;) { + const ssize_t sent = ::sendmsg(socket, &msg, MSG_NOSIGNAL); + if (sent >= 0) { + if (static_cast(sent) != payloadSize) { + // A datagram socket sends all or nothing. + MGLOG_E("MG_Remote fd passing: short datagram (%zd of %zu bytes)", sent, + payloadSize); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + return MOBILEGL_OK; + } + if (errno == EINTR) { + continue; + } + if (errno == EPIPE || errno == ECONNRESET) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + MGLOG_E("MG_Remote fd passing: sendmsg failed (errno=%d)", errno); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + } + + MobileGLResult ReceiveFd(int socket, int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, std::uint32_t timeoutMs) { + if (socket < 0 || outFd == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + *outFd = -1; + if (outSidebandSize != nullptr) { + *outSidebandSize = 0; + } + // Checked before the recvmsg: a datagram cannot be partially consumed, + // so a too-small destination must never cost us the descriptor. + if (sideband.size < kMaxSidebandBytes) { + if (outSidebandSize != nullptr) { + *outSidebandSize = kMaxSidebandBytes; + } + return MOBILEGL_ERR_BUFFER_TOO_SMALL; + } + if (sideband.data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + + const int ready = WaitReadable(socket, timeoutMs); + if (ready < 0) { + MGLOG_E("MG_Remote fd passing: poll failed (errno=%d)", errno); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + if (ready == 0) { + return MOBILEGL_ERR_TIMEOUT; + } + + std::uint8_t payload[sizeof(SidebandHeader) + kMaxSidebandBytes]; + struct iovec iov{}; + iov.iov_base = payload; + iov.iov_len = sizeof(payload); + + union { + struct cmsghdr align; + char bytes[CMSG_SPACE(sizeof(int) * 4)]; + } control{}; + std::memset(&control, 0, sizeof(control)); + + struct msghdr msg{}; + msg.msg_iov = &iov; + msg.msg_iovlen = 1; + msg.msg_control = control.bytes; + msg.msg_controllen = sizeof(control.bytes); + + ssize_t got = 0; + for (;;) { + int flags = 0; +#if defined(MSG_CMSG_CLOEXEC) + flags |= MSG_CMSG_CLOEXEC; +#endif + got = ::recvmsg(socket, &msg, flags); + if (got >= 0) { + break; + } + if (errno == EINTR) { + continue; + } + if (errno == ECONNRESET) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + MGLOG_E("MG_Remote fd passing: recvmsg failed (errno=%d)", errno); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + if (got == 0) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + + // Collect every descriptor first, so an unexpected extra one is closed + // rather than leaked, whatever else is wrong with the message. + int received[4]; + int receivedCount = 0; + for (struct cmsghdr* cmsg = CMSG_FIRSTHDR(&msg); cmsg != nullptr; + cmsg = CMSG_NXTHDR(&msg, cmsg)) { + if (cmsg->cmsg_level != SOL_SOCKET || cmsg->cmsg_type != SCM_RIGHTS) { + continue; + } + const std::size_t bytes = cmsg->cmsg_len - CMSG_LEN(0); + const int count = static_cast(bytes / sizeof(int)); + for (int i = 0; i < count && receivedCount < 4; ++i) { + int fd = -1; + std::memcpy(&fd, CMSG_DATA(cmsg) + i * sizeof(int), sizeof(fd)); + received[receivedCount++] = fd; + } + } + const auto closeAll = [&](int keepIndex) { + for (int i = 0; i < receivedCount; ++i) { + if (i != keepIndex && received[i] >= 0) { + ::close(received[i]); + } + } + }; + + if ((msg.msg_flags & MSG_CTRUNC) != 0) { + // The kernel dropped ancillary data: whatever arrived is not a + // complete offer, and silently continuing would hand the caller a + // half-transferred segment. + MGLOG_E("MG_Remote fd passing: ancillary data truncated; the descriptor did not " + "arrive intact"); + closeAll(-1); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (receivedCount != 1) { + MGLOG_E("MG_Remote fd passing: expected exactly one descriptor, got %d", receivedCount); + closeAll(-1); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (static_cast(got) < sizeof(SidebandHeader)) { + MGLOG_E("MG_Remote fd passing: %zd byte datagram is shorter than the header", got); + closeAll(-1); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + + SidebandHeader header{}; + std::memcpy(&header, payload, sizeof(header)); + if (header.magic != kSidebandMagic || + header.sidebandSize > kMaxSidebandBytes || + sizeof(SidebandHeader) + header.sidebandSize != static_cast(got)) { + MGLOG_E("MG_Remote fd passing: bad sideband header (magic=0x%08X size=%u datagram=%zd)", + header.magic, header.sidebandSize, got); + closeAll(-1); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + + if (header.sidebandSize != 0) { + std::memcpy(sideband.data, payload + sizeof(SidebandHeader), header.sidebandSize); + } + if (outSidebandSize != nullptr) { + *outSidebandSize = header.sidebandSize; + } + *outFd = received[0]; + closeAll(0); + return MOBILEGL_OK; + } + +#endif // _WIN32 + +} // namespace MobileGL::MG_Remote::Transport::FdPassing diff --git a/MobileGL/MG_Remote/Transport/FdPassing.h b/MobileGL/MG_Remote/Transport/FdPassing.h new file mode 100644 index 00000000..d3ccb5c4 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/FdPassing.h @@ -0,0 +1,67 @@ +// MobileGL - MobileGL/MG_Remote/Transport/FdPassing.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// SCM_RIGHTS descriptor passing over an AF_UNIX socket pair. POSIX only. +// +// This is the FIRST transport commit, deliberately (inherited design, plan +// section 8.1, "SCM_RIGHTS must be implemented in the first transport +// commit"). The earlier branch pushed it to a later phase and hardcoded +// `out->fd = -1` in its offer poll, so on the only platform that matters its +// data plane could never move a byte: every segment announcement resolved to +// "no descriptor". A transport whose shm cannot cross the process boundary is +// not a transport. +// +// Channel shape: a dedicated AF_UNIX SOCK_DGRAM socketpair, NOT the control +// byte stream. Two reasons: +// - SOCK_DGRAM preserves message boundaries on every POSIX (SOCK_SEQPACKET +// does not exist on macOS), so one sendmsg is exactly one recvmsg and the +// ancillary data can never be split away from its payload; +// - ancillary data attached to a byte stream binds to whichever ordinary +// byte happens to be at the front of the reader's buffer, which is +// unmanageable once frames are being reassembled. + +#pragma once + +#include "../Protocol/mg_protocol_base.h" + +#include + +namespace MobileGL::MG_Remote::Transport::FdPassing { + + // Upper bound for the bytes that travel with a descriptor (a SegmentRef + // sized announcement, not payload). + inline constexpr std::uint64_t kMaxSidebandBytes = 256; + + // False on platforms without SCM_RIGHTS (Windows). + bool Supported(); + + // Creates the aux socket pair. Both descriptors are CLOEXEC and owned by + // the caller. outFds[0] is conventionally the client end, [1] the server's + // (the one that is inherited or passed to the spawned process). + MobileGLResult CreateSocketPair(int outFds[2]); + + // Sends `fd` with `sideband` attached. The caller keeps ownership of `fd` + // (the peer gets its own descriptor for the same open file description). + // sideband.size must be <= kMaxSidebandBytes. + MobileGLResult SendFd(int socket, int fd, MobileGLByteSpan sideband); + + // Receives one descriptor and its sideband bytes. + // + // `sideband` must be at least kMaxSidebandBytes: a datagram cannot be + // partially consumed, so the capacity is checked BEFORE anything is read. + // A short buffer returns MOBILEGL_ERR_BUFFER_TOO_SMALL with + // *outSidebandSize = kMaxSidebandBytes and consumes nothing, so no + // descriptor is ever dropped on the floor. + // + // On success *outFd owns a descriptor this process must close. + // MOBILEGL_ERR_TIMEOUT when nothing arrived (timeoutMs 0 = poll), + // MOBILEGL_ERR_TRANSPORT_CLOSED on peer close. + MobileGLResult ReceiveFd(int socket, int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, std::uint32_t timeoutMs); + +} // namespace MobileGL::MG_Remote::Transport::FdPassing diff --git a/MobileGL/MG_Remote/Transport/Framing.h b/MobileGL/MG_Remote/Transport/Framing.h new file mode 100644 index 00000000..9123f20c --- /dev/null +++ b/MobileGL/MG_Remote/Transport/Framing.h @@ -0,0 +1,206 @@ +// MobileGL - MobileGL/MG_Remote/Transport/Framing.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// Control-channel wire framing: [u32 magic 'MGLF'][u32 payloadLength][payload]. +// Length excludes the 8-byte header and is capped at 64 MiB. +// +// Two defects of the earlier branch's codec are fixed here, and both are the +// reason this file is not a copy of it: +// +// 1. Its Feed() unconditionally returned OK and its header peek merely +// returned false on a bad magic or an oversized length. A corrupt or +// desynchronized stream therefore turned into a silent, permanent hang - +// the reader kept waiting for a message that could never be parsed, with +// no error anywhere. Here a violation latches a failed state, is logged at +// ERROR, and every later call returns MOBILEGL_ERR_PROTOCOL_MISMATCH. +// +// 2. Its receive path failed the call and consumed the message when the +// caller's buffer was too small, wedging the stream. Here +// MOBILEGL_ERR_BUFFER_TOO_SMALL reports the required size and KEEPS the +// message queued. +// +// The reader is a plain byte-stream reassembler: it never assumes a read() +// returned a whole frame. + +#pragma once + +#include "../Protocol/mg_protocol_base.h" + +#include + +#include +#include +#include +#include + +namespace MobileGL::MG_Remote::Transport { + + // 'MGLF', little-endian on the wire (both ends are the same machine). + inline constexpr std::uint32_t kFrameMagic = 0x464C474Du; + inline constexpr std::uint64_t kFrameHeaderSize = 8; + inline constexpr std::uint64_t kMaxFramePayloadSize = 64ull * 1024 * 1024; + + // Compaction threshold: consumed bytes are dropped from the front once + // enough of them accumulate, so a long-lived reader neither memmoves per + // message nor grows without bound. + inline constexpr std::uint64_t kFrameReaderCompactThreshold = 64ull * 1024; + + // Appends one framed message to `out`. + inline MobileGLResult AppendFrame(std::vector& out, const void* payload, + std::uint64_t size) { + if (size > kMaxFramePayloadSize) { + MGLOG_E("MG_Remote framing: refusing to send a %llu byte payload (cap %llu); bulk " + "bytes belong in shm", + static_cast(size), + static_cast(kMaxFramePayloadSize)); + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + if (size != 0 && payload == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + std::uint8_t header[kFrameHeaderSize]; + const std::uint32_t magic = kFrameMagic; + const std::uint32_t length = static_cast(size); + std::memcpy(header + 0, &magic, sizeof(magic)); + std::memcpy(header + 4, &length, sizeof(length)); + out.insert(out.end(), header, header + kFrameHeaderSize); + const auto* bytes = static_cast(payload); + out.insert(out.end(), bytes, bytes + size); + return MOBILEGL_OK; + } + + // Incremental frame extractor over a raw byte stream. + class FrameReader { + public: + // Feeds raw stream bytes. Validates the frame header the moment enough + // bytes for one exist - a bad magic or an oversized length is reported + // here, not swallowed. + MobileGLResult Feed(const void* data, std::uint64_t size) { + if (m_failed) { + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (size != 0) { + if (data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + const auto* bytes = static_cast(data); + m_buffer.insert(m_buffer.end(), bytes, bytes + size); + } + return ParseHeader(); + } + + bool Failed() const { return m_failed; } + + bool HasMessage() const { + return !m_failed && m_haveHeader && Available() >= kFrameHeaderSize + m_pendingSize; + } + + // Size of the next complete message, or 0 when none is complete yet. + std::uint64_t PendingMessageSize() const { return HasMessage() ? m_pendingSize : 0; } + + std::uint64_t BufferedBytes() const { return Available(); } + + // Copies the next complete message out. + // MOBILEGL_OK - copied, *outSize set, message consumed + // MOBILEGL_ERR_BUFFER_TOO_SMALL - *outSize = required size, message KEPT + // MOBILEGL_ERR_TIMEOUT - no complete message buffered + // MOBILEGL_ERR_PROTOCOL_MISMATCH- the stream is latched failed + MobileGLResult TakeMessage(MobileGLMutableByteSpan buffer, std::uint64_t* outSize) { + if (m_failed) { + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (!HasMessage()) { + return MOBILEGL_ERR_TIMEOUT; + } + if (outSize != nullptr) { + *outSize = m_pendingSize; + } + if (buffer.size < m_pendingSize) { + // The message stays queued; the caller retries with a big + // enough buffer. + return MOBILEGL_ERR_BUFFER_TOO_SMALL; + } + if (m_pendingSize != 0) { + if (buffer.data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + std::memcpy(buffer.data, m_buffer.data() + m_readPos + kFrameHeaderSize, + static_cast(m_pendingSize)); + } + Consume(); + return MOBILEGL_OK; + } + + // Convenience overload that sizes the destination itself. + MobileGLResult TakeMessage(std::vector& out) { + if (m_failed) { + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (!HasMessage()) { + return MOBILEGL_ERR_TIMEOUT; + } + const auto* first = m_buffer.data() + m_readPos + kFrameHeaderSize; + out.assign(first, first + m_pendingSize); + Consume(); + return MOBILEGL_OK; + } + + private: + std::uint64_t Available() const { return m_buffer.size() - m_readPos; } + + MobileGLResult ParseHeader() { + if (m_haveHeader || Available() < kFrameHeaderSize) { + return MOBILEGL_OK; + } + std::uint32_t magic = 0; + std::uint32_t length = 0; + std::memcpy(&magic, m_buffer.data() + m_readPos, sizeof(magic)); + std::memcpy(&length, m_buffer.data() + m_readPos + 4, sizeof(length)); + if (magic != kFrameMagic) { + m_failed = true; + MGLOG_E("MG_Remote framing: bad frame magic 0x%08X (expected 0x%08X); the control " + "stream is desynchronized and this transport is now dead", + magic, kFrameMagic); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + if (length > kMaxFramePayloadSize) { + m_failed = true; + MGLOG_E("MG_Remote framing: frame length %u exceeds the %llu byte cap; refusing to " + "allocate on a peer-supplied length", + length, static_cast(kMaxFramePayloadSize)); + return MOBILEGL_ERR_PROTOCOL_MISMATCH; + } + m_pendingSize = length; + m_haveHeader = true; + return MOBILEGL_OK; + } + + void Consume() { + m_readPos += kFrameHeaderSize + m_pendingSize; + m_pendingSize = 0; + m_haveHeader = false; + if (m_readPos == m_buffer.size()) { + m_buffer.clear(); + m_readPos = 0; + } else if (m_readPos >= kFrameReaderCompactThreshold) { + m_buffer.erase(m_buffer.begin(), + m_buffer.begin() + static_cast(m_readPos)); + m_readPos = 0; + } + // Header of the next message may already be buffered. + (void)ParseHeader(); + } + + std::vector m_buffer; + std::uint64_t m_readPos = 0; + std::uint64_t m_pendingSize = 0; + bool m_haveHeader = false; + bool m_failed = false; + }; + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/ITransport.h b/MobileGL/MG_Remote/Transport/ITransport.h new file mode 100644 index 00000000..1a68bd24 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/ITransport.h @@ -0,0 +1,124 @@ +// MobileGL - MobileGL/MG_Remote/Transport/ITransport.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// The control-plane transport interface. +// +// It is deliberately dumb: complete messages in, complete messages out, plus +// the one thing shared memory cannot do without help - handing a file +// descriptor to the peer. No session routing, no seq accounting, no +// serialization; those live above, in the protocol layer. +// +// Everything on the hot path bypasses this interface entirely: records go into +// the SEG_CMD ring (Ring.h) and the peer is woken through a Doorbell +// (Doorbell.h). ITransport carries the handshake, surface ops, resync, aux +// requests and fatals - the rare, variable-length, must-evolve traffic that +// plan section 7.1 assigns to FlatBuffers tables. +// +// This header stays dependency-light on purpose (mg_protocol_base.h plus the +// standard library): it is included by both roles and by the eventual +// server-side binary, and nothing about a byte pipe needs the GL frontend's +// umbrella header. +// +// Threading: one instance is not internally synchronized for send; callers +// serialize sends. ReceiveFrame/ReceiveFd may be called from one dedicated +// reader thread concurrently with sends from another. + +#pragma once + +#include "../Protocol/mg_protocol_base.h" + +#include + +namespace MobileGL::MG_Remote::Transport { + + // Which end of the connection this instance is. + enum class TransportRole : std::uint32_t { + Server = 1, // accepts the client connection + Client = 2, // connects to the server endpoint + InProcess = 3, // same-process hand-off (CI / inproc delivery mode) + }; + + class ITransport { + public: + virtual ~ITransport() = default; + + ITransport(const ITransport&) = delete; + ITransport& operator=(const ITransport&) = delete; + + // ---- control plane ------------------------------------------------- + + // Sends one complete message. `bytes` is borrowed: the implementation + // either copies it or completes the underlying write before returning. + // A payload larger than Framing::kMaxFramePayloadSize is rejected with + // MOBILEGL_ERR_INVALID_ARGUMENT - bulk bytes belong in shm, never here. + virtual MobileGLResult SendFrame(MobileGLByteSpan bytes) = 0; + + // Receives the next complete message. + // + // MOBILEGL_OK - copied into `buffer`, *outSize is + // the message size, message consumed. + // MOBILEGL_ERR_BUFFER_TOO_SMALL - `buffer` is too small. *outSize is + // the size required and THE MESSAGE + // STAYS QUEUED: call again with a + // buffer of at least that size and it + // is still there. + // MOBILEGL_ERR_TIMEOUT - nothing arrived within timeoutMs + // (0 = non-blocking poll). + // MOBILEGL_ERR_TRANSPORT_CLOSED - peer gone, nothing left buffered. + // MOBILEGL_ERR_PROTOCOL_MISMATCH- framing violated; the transport is + // latched failed and never recovers. + // + // The buffer-too-small half of that contract is the whole point of + // having one: the earlier branch's transport failed the call AND + // dropped the message, which wedges the stream permanently the first + // time a message is bigger than the reader's guess. + virtual MobileGLResult ReceiveFrame(MobileGLMutableByteSpan buffer, std::uint64_t* outSize, + std::uint32_t timeoutMs) = 0; + + // Size of the next pending message, or 0 when none is buffered. Lets a + // caller size its buffer without a failed receive first. + virtual std::uint64_t PeekFrameSize() = 0; + + // ---- descriptor passing -------------------------------------------- + + // Hands `fd` to the peer. POSIX: SCM_RIGHTS over the aux socket (see + // FdPassing.h). Windows: not applicable, returns + // MOBILEGL_ERR_UNSUPPORTED - the section name travels inside SegmentRef + // instead. The caller keeps ownership of `fd` and closes it itself. + // + // This is a first-class member of the interface, not a later phase: the + // earlier branch deferred it and hardcoded `out->fd = -1` in its offer + // poll, so its data plane could not move a single byte on the only + // platform that matters. + virtual MobileGLResult ShareFd(int fd, MobileGLByteSpan sideband) = 0; + + // Receives one fd previously shared by the peer. On success *outFd owns + // a descriptor this process must close. `sideband` receives the bytes + // that travelled with it (may be empty) and must be at least + // FdPassing::kMaxSidebandBytes: an fd offer is one datagram and cannot + // be half-consumed, so the capacity is checked BEFORE anything is read + // and a short buffer returns MOBILEGL_ERR_BUFFER_TOO_SMALL with the + // required size, having consumed nothing and dropped no descriptor. + virtual MobileGLResult ReceiveFd(int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, std::uint32_t timeoutMs) = 0; + + // ---- lifecycle ------------------------------------------------------ + + // Idempotent. Unblocks every waiter with MOBILEGL_ERR_TRANSPORT_CLOSED + // and releases the endpoint. Messages already queued for this endpoint + // stay readable until drained, so a peer that shuts down after sending + // does not lose its last message. + virtual void Shutdown() = 0; + + virtual TransportRole Role() const = 0; + + protected: + ITransport() = default; + }; + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/InProcessTransport.cpp b/MobileGL/MG_Remote/Transport/InProcessTransport.cpp new file mode 100644 index 00000000..289ec3d8 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/InProcessTransport.cpp @@ -0,0 +1,275 @@ +// MobileGL - MobileGL/MG_Remote/Transport/InProcessTransport.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +#include "InProcessTransport.h" + +#include "FdPassing.h" +#include "Framing.h" + +#include + +#include +#include +#include +#include +#include +#include +#include + +#if !defined(_WIN32) +#include +#endif + +namespace MobileGL::MG_Remote::Transport { + + namespace { + struct FdOffer { + int fd = -1; + std::vector sideband; + }; + } // namespace + + // One direction of the channel: everything queued FOR one endpoint. + class InProcessChannel { + public: + struct Direction { + std::mutex mutex; + std::condition_variable cv; + std::deque> messages; + std::deque fdOffers; + bool closed = false; + }; + + ~InProcessChannel() { + for (Direction& dir : m_directions) { + for (FdOffer& offer : dir.fdOffers) { +#if !defined(_WIN32) + if (offer.fd >= 0) { + ::close(offer.fd); + } +#endif + } + dir.fdOffers.clear(); + } + } + + Direction& Inbox(int endpoint) { return m_directions[endpoint]; } + Direction& Outbox(int endpoint) { return m_directions[1 - endpoint]; } + CondVarDoorbell& Bell(int endpoint) { return m_bells[endpoint]; } + + void Close() { + for (Direction& dir : m_directions) { + { + std::lock_guard lock(dir.mutex); + dir.closed = true; + } + dir.cv.notify_all(); + } + // Anything parked on a ring doorbell has to come back too, or a + // shutdown mid-frame hangs the peer forever. + for (CondVarDoorbell& bell : m_bells) { + bell.Notify(); + } + } + + private: + Direction m_directions[2]; + CondVarDoorbell m_bells[2]; + }; + + InProcessTransport::InProcessTransport(std::shared_ptr channel, int endpoint) + : m_channel(std::move(channel)), m_endpoint(endpoint) {} + + InProcessTransport::~InProcessTransport() = default; + + void InProcessTransport::CreatePair(std::unique_ptr& outClient, + std::unique_ptr& outServer) { + auto channel = std::make_shared(); + outClient.reset(new InProcessTransport(channel, 0)); + outServer.reset(new InProcessTransport(channel, 1)); + } + + MobileGLResult InProcessTransport::SendFrame(MobileGLByteSpan bytes) { + if (bytes.size != 0 && bytes.data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + // Same cap as the byte-stream transports, so nothing legal here becomes + // illegal the day the delivery mode changes to `spawn`. + if (bytes.size > kMaxFramePayloadSize) { + MGLOG_E("MG_Remote inproc: refusing a %llu byte message (cap %llu)", + static_cast(bytes.size), + static_cast(kMaxFramePayloadSize)); + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + + InProcessChannel::Direction& dir = m_channel->Outbox(m_endpoint); + { + std::lock_guard lock(dir.mutex); + if (dir.closed) { + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + const auto* first = static_cast(bytes.data); + dir.messages.emplace_back(first, first + bytes.size); + } + dir.cv.notify_one(); + return MOBILEGL_OK; + } + + MobileGLResult InProcessTransport::ReceiveFrame(MobileGLMutableByteSpan buffer, + std::uint64_t* outSize, + std::uint32_t timeoutMs) { + if (outSize != nullptr) { + *outSize = 0; + } + InProcessChannel::Direction& dir = m_channel->Inbox(m_endpoint); + std::unique_lock lock(dir.mutex); + if (dir.messages.empty() && !dir.closed && timeoutMs != 0) { + const auto ready = [&dir] { return !dir.messages.empty() || dir.closed; }; + if (timeoutMs == kWaitForever) { + dir.cv.wait(lock, ready); + } else { + dir.cv.wait_for(lock, std::chrono::milliseconds(timeoutMs), ready); + } + } + if (dir.messages.empty()) { + // Queued messages outlive the peer's Shutdown; only an empty inbox + // is a closed one. + return dir.closed ? MOBILEGL_ERR_TRANSPORT_CLOSED : MOBILEGL_ERR_TIMEOUT; + } + + const std::vector& front = dir.messages.front(); + const std::uint64_t size = front.size(); + if (outSize != nullptr) { + *outSize = size; + } + if (buffer.size < size) { + // Contract: the message STAYS QUEUED. The earlier branch's + // transport failed the call and popped the message anyway, which + // wedges the stream permanently the first time a reader guesses the + // size wrong. + return MOBILEGL_ERR_BUFFER_TOO_SMALL; + } + if (size != 0) { + if (buffer.data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + std::memcpy(buffer.data, front.data(), static_cast(size)); + } + dir.messages.pop_front(); + return MOBILEGL_OK; + } + + std::uint64_t InProcessTransport::PeekFrameSize() { + InProcessChannel::Direction& dir = m_channel->Inbox(m_endpoint); + std::lock_guard lock(dir.mutex); + return dir.messages.empty() ? 0 : dir.messages.front().size(); + } + + MobileGLResult InProcessTransport::ShareFd(int fd, MobileGLByteSpan sideband) { +#if defined(_WIN32) + (void)fd; + (void)sideband; + return MOBILEGL_ERR_UNSUPPORTED; +#else + if (fd < 0) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + if (sideband.size > FdPassing::kMaxSidebandBytes || + (sideband.size != 0 && sideband.data == nullptr)) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + // Same ownership rule as SCM_RIGHTS: the peer gets its own descriptor + // for the same open file description and the caller keeps its own. + const int duplicate = ::dup(fd); + if (duplicate < 0) { + MGLOG_E("MG_Remote inproc: dup failed (errno=%d)", errno); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + + FdOffer offer; + offer.fd = duplicate; + if (sideband.size != 0) { + const auto* first = static_cast(sideband.data); + offer.sideband.assign(first, first + sideband.size); + } + + InProcessChannel::Direction& dir = m_channel->Outbox(m_endpoint); + { + std::lock_guard lock(dir.mutex); + if (dir.closed) { + ::close(duplicate); + return MOBILEGL_ERR_TRANSPORT_CLOSED; + } + dir.fdOffers.push_back(std::move(offer)); + } + dir.cv.notify_one(); + return MOBILEGL_OK; +#endif + } + + MobileGLResult InProcessTransport::ReceiveFd(int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, + std::uint32_t timeoutMs) { + if (outFd == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + *outFd = -1; + if (outSidebandSize != nullptr) { + *outSidebandSize = 0; + } +#if defined(_WIN32) + (void)sideband; + (void)timeoutMs; + return MOBILEGL_ERR_UNSUPPORTED; +#else + // Symmetric with FdPassing::ReceiveFd so callers behave identically in + // both delivery modes. + if (sideband.size < FdPassing::kMaxSidebandBytes) { + if (outSidebandSize != nullptr) { + *outSidebandSize = FdPassing::kMaxSidebandBytes; + } + return MOBILEGL_ERR_BUFFER_TOO_SMALL; + } + if (sideband.data == nullptr) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + + InProcessChannel::Direction& dir = m_channel->Inbox(m_endpoint); + std::unique_lock lock(dir.mutex); + if (dir.fdOffers.empty() && !dir.closed && timeoutMs != 0) { + const auto ready = [&dir] { return !dir.fdOffers.empty() || dir.closed; }; + if (timeoutMs == kWaitForever) { + dir.cv.wait(lock, ready); + } else { + dir.cv.wait_for(lock, std::chrono::milliseconds(timeoutMs), ready); + } + } + if (dir.fdOffers.empty()) { + return dir.closed ? MOBILEGL_ERR_TRANSPORT_CLOSED : MOBILEGL_ERR_TIMEOUT; + } + + FdOffer offer = std::move(dir.fdOffers.front()); + dir.fdOffers.pop_front(); + if (!offer.sideband.empty()) { + std::memcpy(sideband.data, offer.sideband.data(), offer.sideband.size()); + } + if (outSidebandSize != nullptr) { + *outSidebandSize = offer.sideband.size(); + } + *outFd = offer.fd; + return MOBILEGL_OK; +#endif + } + + void InProcessTransport::Shutdown() { m_channel->Close(); } + + Doorbell& InProcessTransport::PeerDoorbell() { return m_channel->Bell(1 - m_endpoint); } + + Doorbell& InProcessTransport::SelfDoorbell() { return m_channel->Bell(m_endpoint); } + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/InProcessTransport.h b/MobileGL/MG_Remote/Transport/InProcessTransport.h new file mode 100644 index 00000000..d3b1fe34 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/InProcessTransport.h @@ -0,0 +1,71 @@ +// MobileGL - MobileGL/MG_Remote/Transport/InProcessTransport.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// The `inproc` transport: two in-memory message queues and a pair of condvar +// doorbells, one connected endpoint at each end. +// +// It is not a test double. `inproc` is a delivery mode of its own (CMake +// option MOBILEGL_BUILD_DISAGGREGATED_INPROC): the server side is the +// monolith's own render thread, which is the single largest CPU lever this +// project has, and it is also the CI form of the split build. What it does NOT +// exercise is serialization of the byte stream, so the framing codec is +// covered separately by FramingTest. +// +// Messages are queued whole, so no framing bytes are involved; the size cap is +// still enforced so that a payload which would be illegal on a socket is +// illegal here too and does not pass CI only to fail after the switch to +// `spawn`. +// +// Descriptor passing is a plain dup(): both ends are the same process, so +// there is nothing to transfer, but the API stays identical so callers can be +// written once. + +#pragma once + +#include "Doorbell.h" +#include "ITransport.h" + +#include + +namespace MobileGL::MG_Remote::Transport { + + class InProcessChannel; + + class InProcessTransport final : public ITransport { + public: + ~InProcessTransport() override; + + // Creates one connected pair. Endpoint 0 is the client, endpoint 1 the + // server; both share one channel and either may be destroyed first. + static void CreatePair(std::unique_ptr& outClient, + std::unique_ptr& outServer); + + MobileGLResult SendFrame(MobileGLByteSpan bytes) override; + MobileGLResult ReceiveFrame(MobileGLMutableByteSpan buffer, std::uint64_t* outSize, + std::uint32_t timeoutMs) override; + std::uint64_t PeekFrameSize() override; + MobileGLResult ShareFd(int fd, MobileGLByteSpan sideband) override; + MobileGLResult ReceiveFd(int* outFd, MobileGLMutableByteSpan sideband, + std::uint64_t* outSidebandSize, std::uint32_t timeoutMs) override; + void Shutdown() override; + TransportRole Role() const override { return TransportRole::InProcess; } + + // The wake channel for the SEG_CMD/SEG_STAGE rings living beside this + // transport: ring the peer's bell after publishing a watermark (only + // when its park flag is set - see NotifyIfParked), park on your own. + Doorbell& PeerDoorbell(); + Doorbell& SelfDoorbell(); + + private: + InProcessTransport(std::shared_ptr channel, int endpoint); + + std::shared_ptr m_channel; + int m_endpoint = 0; + }; + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/Ring.cpp b/MobileGL/MG_Remote/Transport/Ring.cpp new file mode 100644 index 00000000..ee092645 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/Ring.cpp @@ -0,0 +1,286 @@ +// MobileGL - MobileGL/MG_Remote/Transport/Ring.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +#include "Ring.h" + +#include + +#include + +namespace MobileGL::MG_Remote::Transport { + + namespace { + constexpr std::uint64_t Align8(std::uint64_t value) { + return (value + (kRingRecordAlignment - 1)) & ~(kRingRecordAlignment - 1); + } + + bool IsPowerOfTwo(std::uint64_t value) { return value != 0 && (value & (value - 1)) == 0; } + + std::atomic& Head(RingControl& c, RingCursorSet which) { + return which == RingCursorSet::Cmd ? c.cmdHead : c.stageHead; + } + const std::atomic& Head(const RingControl& c, RingCursorSet which) { + return which == RingCursorSet::Cmd ? c.cmdHead : c.stageHead; + } + std::atomic& AppliedTail(RingControl& c, RingCursorSet which) { + return which == RingCursorSet::Cmd ? c.cmdAppliedTail : c.stageAppliedTail; + } + const std::atomic& AppliedTail(const RingControl& c, RingCursorSet which) { + return which == RingCursorSet::Cmd ? c.cmdAppliedTail : c.stageAppliedTail; + } + std::atomic& RetiredTail(RingControl& c, RingCursorSet which) { + return which == RingCursorSet::Cmd ? c.cmdRetiredTail : c.stageRetiredTail; + } + const std::atomic& RetiredTail(const RingControl& c, RingCursorSet which) { + return which == RingCursorSet::Cmd ? c.cmdRetiredTail : c.stageRetiredTail; + } + } // namespace + + void InitRingControl(RingControl& control) { + std::memset(static_cast(&control), 0, sizeof(RingControl)); + // 0 means "uninitialized" for both generations, so a peer that reads a + // zero page can tell it from a legal generation. + control.serverEpoch.store(1, std::memory_order_relaxed); + control.ringGeneration.store(1, std::memory_order_relaxed); + } + + bool RingCursorsValid(const RingControl& control, RingCursorSet cursors, + std::uint64_t capacityBytes) { + const std::uint64_t head = Head(control, cursors).load(std::memory_order_acquire); + const std::uint64_t applied = AppliedTail(control, cursors).load(std::memory_order_acquire); + const std::uint64_t retired = RetiredTail(control, cursors).load(std::memory_order_acquire); + if (applied > head || retired > applied) { + return false; + } + return head - retired <= capacityBytes; + } + + MobileGLResult HardDrainRing(RingControl& control, RingCursorSet cursors) { + const std::uint64_t head = Head(control, cursors).load(std::memory_order_acquire); + const std::uint64_t applied = AppliedTail(control, cursors).load(std::memory_order_acquire); + const std::uint64_t retired = RetiredTail(control, cursors).load(std::memory_order_acquire); + if (head != applied || applied != retired) { + MGLOG_E("MG_Remote ring: hard drain refused, ring is not quiesced " + "(head=%llu applied=%llu retired=%llu)", + static_cast(head), + static_cast(applied), + static_cast(retired)); + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + // Cursors stay monotonic across the drain - only the generation moves, + // so any offset either side cached is now recognisably stale. + control.ringGeneration.fetch_add(1, std::memory_order_acq_rel); + return MOBILEGL_OK; + } + + // ----------------------------------------------------------------------- + // Producer + // ----------------------------------------------------------------------- + + RingProducer::RingProducer(RingControl* control, void* base, std::uint64_t capacityBytes, + RingCursorSet cursors) + : m_control(control), m_base(static_cast(base)), m_capacity(capacityBytes), + m_mask(capacityBytes - 1), m_cursors(cursors) { + if (control == nullptr || base == nullptr || !IsPowerOfTwo(capacityBytes) || + capacityBytes < sizeof(RingRecordHeader)) { + MGLOG_E("MG_Remote ring: producer rejected, capacity %llu must be a power of two of at " + "least %zu bytes over a non-null mapping", + static_cast(capacityBytes), sizeof(RingRecordHeader)); + m_control = nullptr; + m_base = nullptr; + m_capacity = 0; + m_mask = 0; + return; + } + m_localHead = Head(*control, cursors).load(std::memory_order_acquire); + } + + std::uint64_t RingProducer::TailForReclaim() const { + // The conservative watermark: a slot borrowed into the GPU timeline is + // only free after retiredTail passes it. A consumer that never borrows + // publishes retired together with applied, so this costs nothing there. + return RetiredTail(*m_control, m_cursors).load(std::memory_order_acquire); + } + + std::uint64_t RingProducer::FreeBytes() const { + if (m_control == nullptr) { + return 0; + } + const std::uint64_t inFlight = m_localHead - TailForReclaim(); + return inFlight >= m_capacity ? 0 : m_capacity - inFlight; + } + + void* RingProducer::Reserve(std::uint16_t kind, std::uint16_t flags, + std::uint64_t payloadBytes) { + if (m_control == nullptr) { + return nullptr; + } + const std::uint64_t total = Align8(sizeof(RingRecordHeader) + payloadBytes); + if (total > m_capacity) { + // A single record larger than the whole ring is a caller bug: the + // record catalogue has to chunk oversized payloads (large subdata + // becomes several records) rather than emit one giant record. + MGLOG_E("MG_Remote ring: record kind %u of %llu bytes does not fit a %llu byte ring; " + "the emitter must chunk it", + static_cast(kind), static_cast(total), + static_cast(m_capacity)); + return nullptr; + } + + const std::uint64_t offset = m_localHead & m_mask; + const std::uint64_t spaceToEnd = m_capacity - offset; + // Every record is a multiple of 8, so the distance to the wrap boundary + // is too, and a pad header always fits. + const bool needsPad = spaceToEnd < total; + const std::uint64_t needed = needsPad ? spaceToEnd + total : total; + if (FreeBytes() < needed) { + MGLOG_D("MG_Remote ring: full, %llu bytes free, %llu needed", + static_cast(FreeBytes()), + static_cast(needed)); + return nullptr; + } + + if (needsPad) { + RingRecordHeader pad{}; + pad.kind = kRingPadRecordKind; + pad.flags = kRecPad; + pad.size = static_cast(spaceToEnd); + std::memcpy(SlotAt(m_localHead), &pad, sizeof(pad)); + m_localHead += spaceToEnd; + } + + RingRecordHeader header{}; + header.kind = kind; + header.flags = static_cast(flags & ~static_cast(kRecPad)); + header.size = static_cast(total); + std::uint8_t* slot = SlotAt(m_localHead); + std::memcpy(slot, &header, sizeof(header)); + m_localHead += total; + return slot + sizeof(RingRecordHeader); + } + + void RingProducer::Publish() { + if (m_control == nullptr) { + return; + } + // Release: everything written into the slots happens-before the peer's + // acquire load of the head. + Head(*m_control, m_cursors).store(m_localHead, std::memory_order_release); + } + + // ----------------------------------------------------------------------- + // Consumer + // ----------------------------------------------------------------------- + + RingConsumer::RingConsumer(RingControl* control, void* base, std::uint64_t capacityBytes, + RingCursorSet cursors) + : m_control(control), m_base(static_cast(base)), + m_capacity(capacityBytes), m_mask(capacityBytes - 1), m_cursors(cursors) { + if (control == nullptr || base == nullptr || !IsPowerOfTwo(capacityBytes) || + capacityBytes < sizeof(RingRecordHeader)) { + MGLOG_E("MG_Remote ring: consumer rejected, capacity %llu must be a power of two of at " + "least %zu bytes over a non-null mapping", + static_cast(capacityBytes), sizeof(RingRecordHeader)); + m_control = nullptr; + m_base = nullptr; + m_capacity = 0; + m_mask = 0; + return; + } + m_localTail = AppliedTail(*control, cursors).load(std::memory_order_acquire); + } + + bool RingConsumer::Pop(RingRecordView& out, bool* outCorrupt) { + if (outCorrupt != nullptr) { + *outCorrupt = false; + } + if (m_control == nullptr) { + return false; + } + const std::uint64_t head = Head(*m_control, m_cursors).load(std::memory_order_acquire); + while (m_localTail != head) { + const std::uint64_t available = head - m_localTail; + if (available < sizeof(RingRecordHeader) || available > m_capacity) { + MGLOG_E("MG_Remote ring: %llu bytes between tail and head is impossible for a %llu " + "byte ring", + static_cast(available), + static_cast(m_capacity)); + if (outCorrupt != nullptr) { + *outCorrupt = true; + } + return false; + } + const std::uint64_t offset = m_localTail & m_mask; + RingRecordHeader header{}; + std::memcpy(&header, m_base + offset, sizeof(header)); + + // SEG_CMD is written by the peer process: compile-time asserts on + // record sizes cannot see runtime corruption, so every dispatch is + // preceded by these bounds checks and a violation is fatal, never a + // retry (plan section 6.3, runtime bounds discipline). + const std::uint64_t size = header.size; + if (size < sizeof(RingRecordHeader) || (size % kRingRecordAlignment) != 0 || + size > available || offset + size > m_capacity) { + MGLOG_E("MG_Remote ring: corrupt record header at cursor %llu " + "(kind=%u flags=0x%04X size=%u available=%llu)", + static_cast(m_localTail), + static_cast(header.kind), static_cast(header.flags), + header.size, static_cast(available)); + if (outCorrupt != nullptr) { + *outCorrupt = true; + } + return false; + } + + if ((header.flags & kRecPad) != 0) { + m_localTail += size; + continue; + } + + out.kind = header.kind; + out.flags = header.flags; + out.payload = m_base + offset + sizeof(RingRecordHeader); + // Includes the alignment tail; the record catalogue knows the real + // payload length. + out.payloadSize = size - sizeof(RingRecordHeader); + out.cursor = m_localTail; + m_localTail += size; + return true; + } + return false; + } + + void RingConsumer::PublishApplied() { + if (m_control == nullptr) { + return; + } + AppliedTail(*m_control, m_cursors).store(m_localTail, std::memory_order_release); + } + + void RingConsumer::PublishRetired() { + if (m_control == nullptr) { + return; + } + // retiredTail must never overtake appliedTail, so publish both. + AppliedTail(*m_control, m_cursors).store(m_localTail, std::memory_order_release); + RetiredTail(*m_control, m_cursors).store(m_localTail, std::memory_order_release); + } + + void RingConsumer::PublishRetiredUpTo(std::uint64_t cursor) { + if (m_control == nullptr) { + return; + } + const std::uint64_t applied = AppliedTail(*m_control, m_cursors).load(std::memory_order_acquire); + const std::uint64_t clamped = cursor > applied ? applied : cursor; + const std::uint64_t current = RetiredTail(*m_control, m_cursors).load(std::memory_order_relaxed); + if (clamped > current) { + RetiredTail(*m_control, m_cursors).store(clamped, std::memory_order_release); + } + } + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/Ring.h b/MobileGL/MG_Remote/Transport/Ring.h new file mode 100644 index 00000000..7fc04205 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/Ring.h @@ -0,0 +1,225 @@ +// MobileGL - MobileGL/MG_Remote/Transport/Ring.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// SEG_CMD / SEG_STAGE ring control and the SPSC producer/consumer over it. +// +// RingControl is the shared page at the head of SEG_CMD, laid out exactly as +// the inherited transport design (plan section 8.1, referring the earlier +// plan's section 6.2) specifies: +// +// - TWO independent cursor triples, one for SEG_CMD and one for SEG_STAGE. +// The stage ring needs its own because "SEG_STAGE has less than a quarter +// left" is a publish trigger and that occupancy cannot be derived from the +// command ring's cursors, and because a stage slot retires on a different +// event than a command record does. +// - THREE separate sequence watermarks. Conflating them is the classic bug: +// appliedSeq releases *AppliedTail, submittedSeq releases staging, +// retiredSeq / completedFrameSerial release *RetiredTail and adopted +// stores. +// - TWO tails per ring, not one. Once the server borrows a ring slot into +// the GPU timeline instead of copying it out again, that slot can only be +// recycled after completedFrameSerial; a single tail would silently +// degrade to conservative reclaim the day borrowing lands. +// - Both park flags, because the doorbell is bidirectional: without the +// server->client direction every client wait degenerates into a +// cross-process spin on one shared cache line (a whole 16.6ms frame of a +// big core, on a phone, competing with the GPU and the game's JVM). +// +// Cursors are monotonically increasing byte counts; the ring is indexed with a +// power-of-two mask. They are never reset, so a torn read can never look like +// a valid earlier position. ringGeneration is bumped after a hard drain to +// invalidate every cached offset. +// +// Record framing inside the ring is the 8-byte header below, which is the +// layout the plan's RecHeader already fixes ({u16 kind, u16 flags, u32 size}, +// size including the header and a multiple of 8). The record CATALOGUE +// (Records.def / PipeCalls.def) is a separate deliverable; the ring itself +// only needs kind/flags/size, so it can carry the real records the day they +// land without changing shape. + +#pragma once + +#include "../Protocol/mg_protocol_base.h" + +#include +#include +#include + +namespace MobileGL::MG_Remote::Transport { + + // The shared control page. One 4 KiB page so it can be mapped alone, with + // each contended group on its own cache line. + struct alignas(4096) RingControl { + // ---- SEG_CMD cursors ------------------------------------------------ + alignas(64) std::atomic cmdHead; // producer: bytes written + alignas(64) std::atomic cmdAppliedTail; // consumer: bytes decoded/copied out + std::atomic cmdRetiredTail; // consumer: borrowed slots released + + // ---- SEG_STAGE cursors ---------------------------------------------- + alignas(64) std::atomic stageHead; + alignas(64) std::atomic stageAppliedTail; + std::atomic stageRetiredTail; + + // ---- sequence / frame watermarks ------------------------------------- + alignas(64) std::atomic appliedSeq; // records applied + std::atomic submittedSeq; // handed to the driver + std::atomic retiredSeq; // GPU finished + std::atomic completedFrameSerial; + std::atomic presentAckSerial; + + // ---- doorbell / generation ------------------------------------------- + alignas(64) std::atomic serverEpoch; // ++ on context loss / server restart + std::atomic ringGeneration; // ++ after a hard drain + std::atomic consumerParked; // server asleep, producer must ring + std::atomic producerParked; // client asleep, server must ring + std::atomic eventRingFull; // SEG_EVENT full, server stopped applying + std::atomic eventDropped; // dropped lossy events + }; + + static_assert(sizeof(RingControl) == 4096, "RingControl must be exactly one page"); + static_assert(alignof(RingControl) == 4096, "RingControl must be page aligned"); + static_assert(std::atomic::is_always_lock_free, + "the ring cursors are shared across processes: they must be lock-free"); + static_assert(std::atomic::is_always_lock_free, + "the doorbell flags are shared across processes: they must be lock-free"); + + // Per-record header. Prefix-identical to the plan's RecHeader so the + // generated record catalogue drops straight in. + struct RingRecordHeader { + std::uint16_t kind; + std::uint16_t flags; + std::uint32_t size; // header + payload + alignment padding, multiple of 8 + }; + static_assert(sizeof(RingRecordHeader) == 8, "RecHeader is 8 bytes on the wire"); + + enum RingRecordFlags : std::uint16_t { + kRecNone = 0, + kRecNeedsAck = 1u << 0, + kRecHasBlob = 1u << 1, + kRecPad = 1u << 2, // filler to the wrap boundary, no payload meaning + kRecBorrowSlot = 1u << 3, // slot is borrowed into the GPU timeline; retires late + kRecVarTail = 1u << 4, + }; + + // Reserved kind for the wrap filler. The catalogue starts at 1. + inline constexpr std::uint16_t kRingPadRecordKind = 0; + + inline constexpr std::uint64_t kRingRecordAlignment = 8; + + // Which cursor triple a producer/consumer pair drives. + enum class RingCursorSet : std::uint32_t { + Cmd = 0, + Stage = 1, + }; + + // Zeroes every cursor and starts serverEpoch / ringGeneration at 1, so that + // a zero read is always "uninitialized", never a legal generation. + void InitRingControl(RingControl& control); + + // head >= appliedTail >= retiredTail, and the ring never holds more than + // its capacity. False means the shared page is corrupt (or a peer is + // misbehaving), which is a Fatal{ProtocolCorruption}, never a retry. + bool RingCursorsValid(const RingControl& control, RingCursorSet cursors, + std::uint64_t capacityBytes); + + // Bumps ringGeneration, invalidating every offset either side has cached. + // Both sides must be quiesced and the ring fully drained + // (head == appliedTail == retiredTail); otherwise this returns + // MOBILEGL_ERR_INVALID_ARGUMENT and changes nothing. + MobileGLResult HardDrainRing(RingControl& control, RingCursorSet cursors); + + // A record as seen by the consumer. + struct RingRecordView { + std::uint16_t kind = 0; + std::uint16_t flags = 0; + const void* payload = nullptr; + std::uint64_t payloadSize = 0; + std::uint64_t cursor = 0; // producer cursor at the START of this record + }; + + // Single producer. Not thread-safe: one writer thread, by construction. + class RingProducer { + public: + RingProducer() = default; + // `base` is the ring's byte area (NOT the control page) and + // `capacityBytes` must be a power of two. + RingProducer(RingControl* control, void* base, std::uint64_t capacityBytes, + RingCursorSet cursors); + + bool Valid() const { return m_control != nullptr; } + + // Bytes still writable before the consumer has to catch up. + std::uint64_t FreeBytes() const; + + // Reserves room for one record and returns a pointer to its payload, + // or nullptr when the ring is full (or the record cannot fit at all). + // The payload is uninitialized; alignment padding at its tail is NOT + // zeroed. Emits a pad record automatically when the record would + // straddle the wrap boundary, so every record is contiguous. + void* Reserve(std::uint16_t kind, std::uint16_t flags, std::uint64_t payloadBytes); + + // Makes every reserved record visible to the consumer (release store on + // the head cursor). Cheap: publishing per record is fine, batching 8-16 + // only amortizes the doorbell store. + void Publish(); + + // Producer-local cursor including records not yet published. + std::uint64_t LocalHead() const { return m_localHead; } + std::uint64_t Capacity() const { return m_capacity; } + + private: + std::uint64_t TailForReclaim() const; + std::uint8_t* SlotAt(std::uint64_t cursor) const { + return m_base + static_cast(cursor & m_mask); + } + + RingControl* m_control = nullptr; + std::uint8_t* m_base = nullptr; + std::uint64_t m_capacity = 0; + std::uint64_t m_mask = 0; + std::uint64_t m_localHead = 0; + RingCursorSet m_cursors = RingCursorSet::Cmd; + }; + + // Single consumer. Not thread-safe: one reader thread, by construction. + class RingConsumer { + public: + RingConsumer() = default; + RingConsumer(RingControl* control, void* base, std::uint64_t capacityBytes, + RingCursorSet cursors); + + bool Valid() const { return m_control != nullptr; } + + // Pops the next record, skipping wrap fillers. Returns false when the + // ring is empty at this moment. A record whose header is impossible + // (size not 8-aligned, smaller than a header, or larger than what the + // producer has published) is refused: *outCorrupt is set, which the + // caller must escalate to Fatal{ProtocolCorruption} rather than retry. + bool Pop(RingRecordView& out, bool* outCorrupt = nullptr); + + // Publishes the applied cursor, releasing those bytes to the producer. + void PublishApplied(); + // Publishes the retired cursor. Records without kRecBorrowSlot retire + // as soon as they are applied; borrowed slots retire on + // completedFrameSerial, which is why this is a separate call. + void PublishRetired(); + void PublishRetiredUpTo(std::uint64_t cursor); + + std::uint64_t LocalTail() const { return m_localTail; } + std::uint64_t Capacity() const { return m_capacity; } + + private: + RingControl* m_control = nullptr; + const std::uint8_t* m_base = nullptr; + std::uint64_t m_capacity = 0; + std::uint64_t m_mask = 0; + std::uint64_t m_localTail = 0; + RingCursorSet m_cursors = RingCursorSet::Cmd; + }; + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/ShmSegment.cpp b/MobileGL/MG_Remote/Transport/ShmSegment.cpp new file mode 100644 index 00000000..298062dd --- /dev/null +++ b/MobileGL/MG_Remote/Transport/ShmSegment.cpp @@ -0,0 +1,49 @@ +// MobileGL - MobileGL/MG_Remote/Transport/ShmSegment.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// Platform-independent half of ShmSegment. The create/map/close bodies live in +// ShmSegmentPosix.cpp and ShmSegmentWin32.cpp. + +#include "ShmSegment.h" + +#include +#include + +namespace MobileGL::MG_Remote::Transport { + + ShmSegment::~ShmSegment() { Close(); } + + ShmSegment::ShmSegment(ShmSegment&& other) noexcept { Steal(std::move(other)); } + + ShmSegment& ShmSegment::operator=(ShmSegment&& other) noexcept { + if (this != &other) { + Close(); + Steal(std::move(other)); + } + return *this; + } + + void ShmSegment::Steal(ShmSegment&& other) noexcept { + std::memcpy(m_name, other.m_name, sizeof(m_name)); + m_mapping = other.m_mapping; + m_nativeHandle = other.m_nativeHandle; + m_size = other.m_size; + m_fd = other.m_fd; + m_readOnly = other.m_readOnly; + + std::memset(other.m_name, 0, sizeof(other.m_name)); + other.m_mapping = nullptr; + other.m_nativeHandle = nullptr; + other.m_size = 0; + other.m_fd = -1; + other.m_readOnly = false; + } + + bool ShmSegment::Valid() const { return m_size != 0 && (m_fd >= 0 || m_nativeHandle != nullptr); } + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/ShmSegment.h b/MobileGL/MG_Remote/Transport/ShmSegment.h new file mode 100644 index 00000000..c40b2cb9 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/ShmSegment.h @@ -0,0 +1,87 @@ +// MobileGL - MobileGL/MG_Remote/Transport/ShmSegment.h +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// One shared-memory segment: SEG_CMD, SEG_STAGE, SEG_REPLY, SEG_EVENT, a +// per-object SEG_SHADOW or a SEG_ADOPT store (inherited segment layout, plan +// section 8.1). +// +// Creation matrix (earlier plan section 6.1): +// - Android: ASharedMemory_create (API 26; libc's memfd_create wrapper +// only appears at API 30, which is above our floor) +// - desktop Linux: syscall(SYS_memfd_create, ...) directly, for the same +// reason - the glibc wrapper is recent and this file has to +// build against old sysroots +// - other POSIX: shm_open + immediate shm_unlink, the fd keeps it alive +// - Windows: CreateFileMappingW in the Local\ namespace +// +// Transfer is NOT done here. On POSIX the fd travels by SCM_RIGHTS +// (FdPassing.h / ITransport::ShareFd) and the name is only a debugging label; +// on Windows the section name travels inside the SegmentRef table. +// +// The Windows implementation is compile-guarded and untested at the time it +// was written: no Windows machine is a correctness gate for this project. + +#pragma once + +#include "../Protocol/mg_protocol_base.h" + +#include +#include + +namespace MobileGL::MG_Remote::Transport { + + inline constexpr std::size_t kShmNameMax = 128; + + class ShmSegment { + public: + ShmSegment() = default; + ~ShmSegment(); + + ShmSegment(const ShmSegment&) = delete; + ShmSegment& operator=(const ShmSegment&) = delete; + ShmSegment(ShmSegment&& other) noexcept; + ShmSegment& operator=(ShmSegment&& other) noexcept; + + // Creates a segment of `size` bytes owned by this process. `nameHint` + // is a short debug label (Windows: part of the section name peers + // resolve). The segment is NOT mapped yet. + static MobileGLResult Create(const char* nameHint, std::uint64_t size, ShmSegment& out); + + // POSIX only: adopts a descriptor received over SCM_RIGHTS. Takes + // ownership of `fd` on success; on failure the caller still owns it. + static MobileGLResult Adopt(int fd, std::uint64_t size, ShmSegment& out); + + // Windows only: opens a section the peer published by name. + static MobileGLResult OpenNamed(const char* name, std::uint64_t size, ShmSegment& out); + + // Maps the whole segment. Read-only mappings are what the peer gets for + // a segment it does not own (SEG_CMD/SEG_STAGE on the server side). + MobileGLResult Map(bool readOnly); + void Unmap(); + void Close(); // unmaps and releases the descriptor/handle + + bool Valid() const; + void* Data() const { return m_mapping; } + std::uint64_t Size() const { return m_size; } + bool MappedReadOnly() const { return m_readOnly; } + const char* Name() const { return m_name; } + // POSIX: the descriptor to hand to ShareFd. -1 on Windows. + int Fd() const { return m_fd; } + + private: + void Steal(ShmSegment&& other) noexcept; + + char m_name[kShmNameMax] = {}; + void* m_mapping = nullptr; + void* m_nativeHandle = nullptr; // Windows HANDLE; unused on POSIX + std::uint64_t m_size = 0; + int m_fd = -1; + bool m_readOnly = false; + }; + +} // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/ShmSegmentPosix.cpp b/MobileGL/MG_Remote/Transport/ShmSegmentPosix.cpp new file mode 100644 index 00000000..6c334d1d --- /dev/null +++ b/MobileGL/MG_Remote/Transport/ShmSegmentPosix.cpp @@ -0,0 +1,191 @@ +// MobileGL - MobileGL/MG_Remote/Transport/ShmSegmentPosix.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +#include "ShmSegment.h" + +#if !defined(_WIN32) + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#if defined(__ANDROID__) +#include +#elif defined(__linux__) +#include +#ifndef MFD_CLOEXEC +#define MFD_CLOEXEC 0x0001U +#endif +#endif + +namespace MobileGL::MG_Remote::Transport { + + namespace { + void CopyName(char (&dst)[kShmNameMax], const char* src) { + if (src == nullptr) { + dst[0] = '\0'; + return; + } + std::snprintf(dst, kShmNameMax, "%s", src); + } + +#if !defined(__ANDROID__) + // Unique per process; only used by the shm_open fallback, whose name + // must not collide with a concurrent creator's. + std::atomic g_shmCounter{0}; +#endif + } // namespace + + MobileGLResult ShmSegment::Create(const char* nameHint, std::uint64_t size, ShmSegment& out) { + if (size == 0) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + out.Close(); + + char label[kShmNameMax]; + std::snprintf(label, sizeof(label), "mgl-%s", nameHint != nullptr ? nameHint : "seg"); + + int fd = -1; +#if defined(__ANDROID__) + // API 26. libc's memfd_create wrapper is API 30, above MobileGL's floor. + fd = ASharedMemory_create(label, static_cast(size)); + if (fd < 0) { + MGLOG_W("MG_Remote shm: ASharedMemory_create(%s, %llu) failed (errno=%d)", label, + static_cast(size), errno); + } +#elif defined(__linux__) + // Raw syscall, not the glibc wrapper: the wrapper is too recent to rely + // on across the sysroots this builds against. + fd = static_cast(::syscall(SYS_memfd_create, label, MFD_CLOEXEC)); + if (fd >= 0 && ::ftruncate(fd, static_cast(size)) != 0) { + MGLOG_E("MG_Remote shm: ftruncate(%llu) failed (errno=%d)", + static_cast(size), errno); + ::close(fd); + fd = -1; + } +#endif + +#if !defined(__ANDROID__) + if (fd < 0) { + // Fallback: shm_open + immediate unlink. The name disappears at + // once; the descriptor is what keeps the object alive and what + // travels by SCM_RIGHTS. + char shmName[kShmNameMax]; + std::snprintf(shmName, sizeof(shmName), "/mgl-%d-%u-%s", static_cast(::getpid()), + g_shmCounter.fetch_add(1, std::memory_order_relaxed), + nameHint != nullptr ? nameHint : "seg"); + fd = ::shm_open(shmName, O_RDWR | O_CREAT | O_EXCL, 0600); + if (fd < 0) { + MGLOG_E("MG_Remote shm: shm_open(%s) failed (errno=%d)", shmName, errno); + return MOBILEGL_ERR_SHM_EXHAUSTED; + } + ::shm_unlink(shmName); + if (::ftruncate(fd, static_cast(size)) != 0) { + MGLOG_E("MG_Remote shm: ftruncate(%llu) failed (errno=%d)", + static_cast(size), errno); + ::close(fd); + return MOBILEGL_ERR_SHM_EXHAUSTED; + } + CopyName(out.m_name, shmName); + } else { + CopyName(out.m_name, label); + } +#else + if (fd < 0) { + return MOBILEGL_ERR_SHM_EXHAUSTED; + } + CopyName(out.m_name, label); +#endif + + out.m_fd = fd; + out.m_size = size; + out.m_nativeHandle = nullptr; + out.m_mapping = nullptr; + out.m_readOnly = false; + return MOBILEGL_OK; + } + + MobileGLResult ShmSegment::Adopt(int fd, std::uint64_t size, ShmSegment& out) { + if (fd < 0 || size == 0) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + // The peer's declared size is not trusted: a segment smaller than what + // the announcement claims would turn every later offset into an + // out-of-bounds map. + struct stat st{}; + if (::fstat(fd, &st) == 0 && st.st_size > 0 && + static_cast(st.st_size) < size) { + MGLOG_E("MG_Remote shm: peer announced %llu bytes but the descriptor is %lld", + static_cast(size), static_cast(st.st_size)); + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + out.Close(); + out.m_fd = fd; // ownership transferred + out.m_size = size; + out.m_nativeHandle = nullptr; + out.m_mapping = nullptr; + out.m_readOnly = false; + CopyName(out.m_name, "adopted"); + return MOBILEGL_OK; + } + + MobileGLResult ShmSegment::OpenNamed(const char*, std::uint64_t, ShmSegment&) { + // POSIX shares descriptors, not names. + return MOBILEGL_ERR_UNSUPPORTED; + } + + MobileGLResult ShmSegment::Map(bool readOnly) { + if (m_fd < 0 || m_size == 0) { + return MOBILEGL_ERR_NOT_INITIALIZED; + } + if (m_mapping != nullptr) { + if (m_readOnly == readOnly) { + return MOBILEGL_OK; + } + Unmap(); + } + const int prot = readOnly ? PROT_READ : (PROT_READ | PROT_WRITE); + void* addr = ::mmap(nullptr, static_cast(m_size), prot, MAP_SHARED, m_fd, 0); + if (addr == MAP_FAILED) { + MGLOG_E("MG_Remote shm: mmap of %llu bytes failed (errno=%d)", + static_cast(m_size), errno); + return MOBILEGL_ERR_OUT_OF_MEMORY; + } + m_mapping = addr; + m_readOnly = readOnly; + return MOBILEGL_OK; + } + + void ShmSegment::Unmap() { + if (m_mapping != nullptr) { + ::munmap(m_mapping, static_cast(m_size)); + m_mapping = nullptr; + } + } + + void ShmSegment::Close() { + Unmap(); + if (m_fd >= 0) { + ::close(m_fd); + m_fd = -1; + } + m_size = 0; + m_readOnly = false; + m_name[0] = '\0'; + } + +} // namespace MobileGL::MG_Remote::Transport + +#endif // !_WIN32 diff --git a/MobileGL/MG_Remote/Transport/ShmSegmentWin32.cpp b/MobileGL/MG_Remote/Transport/ShmSegmentWin32.cpp new file mode 100644 index 00000000..8fc07ea1 --- /dev/null +++ b/MobileGL/MG_Remote/Transport/ShmSegmentWin32.cpp @@ -0,0 +1,162 @@ +// MobileGL - MobileGL/MG_Remote/Transport/ShmSegmentWin32.cpp +// Copyright (c) 2025-2026 MobileGL-Dev +// Licensed under the GNU Lesser General Public License v3.0: +// https://www.gnu.org/licenses/gpl-3.0.txt +// https://www.gnu.org/licenses/lgpl-3.0.txt +// SPDX-License-Identifier: LGPL-3.0-only +// End of Source File Header + +// Windows half of ShmSegment: a named file-mapping section in the Local\ +// namespace, which the peer opens by the name carried in SegmentRef. +// +// UNTESTED. This project's Windows machine is not a correctness gate (its +// Vulkan lacks vkCreateHeadlessSurfaceEXT and accounts for most of its +// baseline integration failures), and the whole disaggregated build is gated +// behind MOBILEGL_BUILD_DISAGGREGATED, which is OFF by default. It is written +// now so the abstraction is shaped by two real platforms rather than one. + +#include "ShmSegment.h" + +#if defined(_WIN32) + +#include + +#include +#include +#include + +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif +#include + +namespace MobileGL::MG_Remote::Transport { + + namespace { + std::atomic g_sectionCounter{0}; + + bool ToWide(const char* utf8, wchar_t* out, int outChars) { + if (utf8 == nullptr || out == nullptr || outChars <= 0) { + return false; + } + const int written = ::MultiByteToWideChar(CP_UTF8, 0, utf8, -1, out, outChars); + return written > 0; + } + } // namespace + + MobileGLResult ShmSegment::Create(const char* nameHint, std::uint64_t size, ShmSegment& out) { + if (size == 0) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + out.Close(); + + char name[kShmNameMax]; + std::snprintf(name, sizeof(name), "Local\\mgl-%lu-%u-%s", + static_cast(::GetCurrentProcessId()), + g_sectionCounter.fetch_add(1, std::memory_order_relaxed), + nameHint != nullptr ? nameHint : "seg"); + + wchar_t wide[kShmNameMax]; + if (!ToWide(name, wide, static_cast(kShmNameMax))) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + + HANDLE section = ::CreateFileMappingW(INVALID_HANDLE_VALUE, nullptr, PAGE_READWRITE, + static_cast(size >> 32), + static_cast(size & 0xFFFFFFFFull), wide); + if (section == nullptr) { + MGLOG_E("MG_Remote shm: CreateFileMappingW(%s, %llu) failed (GetLastError=%lu)", name, + static_cast(size), + static_cast(::GetLastError())); + return MOBILEGL_ERR_SHM_EXHAUSTED; + } + if (::GetLastError() == ERROR_ALREADY_EXISTS) { + ::CloseHandle(section); + MGLOG_E("MG_Remote shm: section name %s already exists", name); + return MOBILEGL_ERR_SHM_EXHAUSTED; + } + + std::snprintf(out.m_name, kShmNameMax, "%s", name); + out.m_nativeHandle = section; + out.m_size = size; + out.m_fd = -1; + out.m_mapping = nullptr; + out.m_readOnly = false; + return MOBILEGL_OK; + } + + MobileGLResult ShmSegment::Adopt(int, std::uint64_t, ShmSegment&) { + // No SCM_RIGHTS here: Windows peers resolve the section by name. + return MOBILEGL_ERR_UNSUPPORTED; + } + + MobileGLResult ShmSegment::OpenNamed(const char* name, std::uint64_t size, ShmSegment& out) { + if (name == nullptr || name[0] == '\0' || size == 0) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + out.Close(); + + wchar_t wide[kShmNameMax]; + if (!ToWide(name, wide, static_cast(kShmNameMax))) { + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + HANDLE section = ::OpenFileMappingW(FILE_MAP_ALL_ACCESS, FALSE, wide); + if (section == nullptr) { + MGLOG_E("MG_Remote shm: OpenFileMappingW(%s) failed (GetLastError=%lu)", name, + static_cast(::GetLastError())); + return MOBILEGL_ERR_INVALID_ARGUMENT; + } + std::snprintf(out.m_name, kShmNameMax, "%s", name); + out.m_nativeHandle = section; + out.m_size = size; + out.m_fd = -1; + out.m_mapping = nullptr; + out.m_readOnly = false; + return MOBILEGL_OK; + } + + MobileGLResult ShmSegment::Map(bool readOnly) { + if (m_nativeHandle == nullptr || m_size == 0) { + return MOBILEGL_ERR_NOT_INITIALIZED; + } + if (m_mapping != nullptr) { + if (m_readOnly == readOnly) { + return MOBILEGL_OK; + } + Unmap(); + } + void* view = ::MapViewOfFile(static_cast(m_nativeHandle), + readOnly ? FILE_MAP_READ : FILE_MAP_ALL_ACCESS, 0, 0, + static_cast(m_size)); + if (view == nullptr) { + MGLOG_E("MG_Remote shm: MapViewOfFile of %llu bytes failed (GetLastError=%lu)", + static_cast(m_size), + static_cast(::GetLastError())); + return MOBILEGL_ERR_OUT_OF_MEMORY; + } + m_mapping = view; + m_readOnly = readOnly; + return MOBILEGL_OK; + } + + void ShmSegment::Unmap() { + if (m_mapping != nullptr) { + ::UnmapViewOfFile(m_mapping); + m_mapping = nullptr; + } + } + + void ShmSegment::Close() { + Unmap(); + if (m_nativeHandle != nullptr) { + ::CloseHandle(static_cast(m_nativeHandle)); + m_nativeHandle = nullptr; + } + m_size = 0; + m_readOnly = false; + m_name[0] = '\0'; + } + +} // namespace MobileGL::MG_Remote::Transport + +#endif // _WIN32