mirror of
https://github.com/MobileGL-Dev/MobileGL
synced 2026-09-08 04:08:32 +09:00
[Test] (MG_Remote, CI): cover the wire layer with five suites and gate the generated header with flatc-check
- MobileGL/MG_Test/Wire, 42 gtest cases in five binaries, ctest label `unit`, registered only when MOBILEGL_BUILD_DISAGGREGATED is ON so the default configuration is untouched. - FdPassingTest is the one the earlier branch could not have had: it forks a child that creates a shared segment, fills 64 KiB with a pattern and hands the descriptor over SCM_RIGHTS; the parent adopts it, maps it read-only and compares every byte. Everything in Feat/CS-Delta-IPC ran in one process, which is why `out->fd = -1` survived unnoticed. It also pins that a too-small sideband buffer is refused before the datagram is consumed (so no descriptor is dropped), that an empty sideband still carries its fd, and that a receive with no offer times out. The `spawn` SocketDoorbell is covered in the same file because it rides the same kind of socket: a parked waiter woken through NotifyIfParked, a clean timeout, and a wakeup that arrives before anyone parks being remembered and then consumed exactly once. - RingTest: control-page size/alignment/cache-line layout, a non-power-of-two capacity refused, payload alignment, 200 records driven through a 256-byte ring so the wrap filler path runs repeatedly and no record ever straddles the boundary, backpressure (full ring refuses, applied alone frees nothing, retired frees), a record larger than the ring refused, the generation bump refused while records are in flight and accepted once quiesced, a corrupt header refused instead of dispatched, and a 20000-record two-thread producer/consumer run checking order, content and the final cursor equality. - FramingTest: byte-at-a-time reassembly of two frames, the magic reading "MGLF" on the wire, empty payloads, a bad magic and an oversized length each latching the reader dead (the old code hung silently instead), the send-side cap, and the buffer-too-small contract keeping the message so the retry still finds it. - InProcessTransportTest: both directions, ordering, buffer-too-small, poll and timeout, shutdown draining queued messages before it closes, a blocked receiver woken by a send and by a shutdown, the frame cap, descriptor hand-off with its sideband, and three condvar doorbell cases - a parked waiter woken through NotifyIfParked, an already-true condition that must never park, and a clean timeout. - ProtocolSmokeTest: a Hello built and read back through the committed generated header, a Welcome carrying the four segment announcements, the same buffer travelling across the transport unchanged, a truncated buffer failing verification rather than being read, and the CtrlMsg / SegmentKind / LogLevel / FatalCode tag values frozen - they are wire numbers, so if that test has to be edited the schema change was a wire break. - CI: a `flatc-check` job in test.yml that checks out only the flatbuffers submodule, builds the pinned flatc through scripts/gen_protocol.py into the runner temp directory, regenerates and runs `git diff --exit-code` on protocol_generated.h. It needs no MobileGL build, so it does not depend on build-linux (plan B section 8.1 / earlier section 7.1: the committed header plus a CI regeneration check, and no flatc in the default build graph). - Verified: `ctest -L unit` is green in both configurations - 1382 cases with the option OFF and 1424 with it ON (the same 1382 plus these 42); the nm gate is 0 matches OFF and 93 ON; regeneration of protocol_generated.h is byte-identical, and the flatc-check gate goes red on a hand-edited header and green again after a clean regeneration.
This commit is contained in:
@@ -295,6 +295,33 @@ jobs:
|
||||
path: /tmp/core.*
|
||||
if-no-files-found: ignore
|
||||
|
||||
# MobileGL/MG_Remote/Protocol/generated/protocol_generated.h is COMMITTED, and
|
||||
# flatc is deliberately absent from the default build graph (a codegen step in
|
||||
# the graph is how the earlier branch ended up cross-compiling an arm64 flatc
|
||||
# and trying to run it on the host). This job is what keeps the committed
|
||||
# header honest: build the pinned flatc, regenerate, and fail on any diff.
|
||||
# It needs no MobileGL build, so it does not depend on build-linux.
|
||||
flatc-check:
|
||||
runs-on: ubuntu-latest
|
||||
|
||||
steps:
|
||||
- name: Checkout repo
|
||||
uses: actions/checkout@v6
|
||||
|
||||
- name: Get CMake
|
||||
uses: lukka/get-cmake@v4.3.3
|
||||
|
||||
- name: Check out the FlatBuffers submodule only
|
||||
# Just this one: the schema check has nothing to do with glslang,
|
||||
# SPIRV-Cross or the trace fixtures.
|
||||
run: git submodule update --init 3rdparty/flatbuffers
|
||||
|
||||
- name: Regenerate protocol_generated.h
|
||||
run: python3 scripts/gen_protocol.py --build-dir "${{ runner.temp }}/flatc-build"
|
||||
|
||||
- name: Fail if the committed header is stale
|
||||
run: git diff --exit-code -- MobileGL/MG_Remote/Protocol/generated/protocol_generated.h
|
||||
|
||||
benchmark:
|
||||
runs-on: ubuntu-latest
|
||||
needs: build-linux
|
||||
|
||||
@@ -88,3 +88,8 @@ add_subdirectory(Backend/DirectGLES)
|
||||
if (ENABLE_INTEGRATION_TESTS)
|
||||
add_subdirectory(Backend/DirectVulkan)
|
||||
endif()
|
||||
# The wire layer only exists in the disaggregated configuration, so its suite
|
||||
# is only registered there. Nothing under MG_Remote is compiled otherwise.
|
||||
if (MOBILEGL_BUILD_DISAGGREGATED)
|
||||
add_subdirectory(Wire)
|
||||
endif()
|
||||
|
||||
@@ -0,0 +1,36 @@
|
||||
cmake_minimum_required(VERSION 3.14)
|
||||
|
||||
# The MG_Remote wire layer: framing, the SPSC ring, the in-process transport,
|
||||
# SCM_RIGHTS descriptor passing and the generated control-plane schema. Only
|
||||
# reachable with MOBILEGL_BUILD_DISAGGREGATED=ON (see MG_Test/CMakeLists.txt).
|
||||
|
||||
set(MOBILEGL_WIRE_TESTS
|
||||
FramingTest
|
||||
RingTest
|
||||
InProcessTransportTest
|
||||
ProtocolSmokeTest
|
||||
)
|
||||
|
||||
if (NOT WIN32)
|
||||
# SCM_RIGHTS and fork(): POSIX only.
|
||||
list(APPEND MOBILEGL_WIRE_TESTS FdPassingTest)
|
||||
endif()
|
||||
|
||||
include(GoogleTest)
|
||||
|
||||
foreach (test IN LISTS MOBILEGL_WIRE_TESTS)
|
||||
add_executable(${test} ${test}.cpp)
|
||||
|
||||
target_include_directories(${test} PRIVATE
|
||||
${MGL_ROOT}/include
|
||||
${MGL_ROOT}/MobileGL
|
||||
${MGL_ROOT}/3rdparty/flatbuffers/include
|
||||
)
|
||||
|
||||
target_link_libraries(${test} PRIVATE
|
||||
GTest::gtest_main
|
||||
${LINK_LIBRARIES}
|
||||
)
|
||||
|
||||
gtest_discover_tests(${test} DISCOVERY_TIMEOUT 30 PROPERTIES LABELS unit)
|
||||
endforeach ()
|
||||
@@ -0,0 +1,250 @@
|
||||
// MobileGL - MobileGL/MG_Test/Wire/FdPassingTest.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
|
||||
|
||||
// SCM_RIGHTS across a real process boundary: a forked child creates a shared
|
||||
// segment, fills it, and hands the descriptor over the aux socket; the parent
|
||||
// adopts it, maps it read-only and compares every byte.
|
||||
//
|
||||
// This is the test the earlier branch never had. Its transport hardcoded
|
||||
// `out->fd = -1` in the offer poll, so its data plane could not move a byte
|
||||
// between processes - and nothing in its suite noticed, because everything ran
|
||||
// in one process.
|
||||
|
||||
#include <MG_Remote/Transport/Doorbell.h>
|
||||
#include <MG_Remote/Transport/FdPassing.h>
|
||||
#include <MG_Remote/Transport/ShmSegment.h>
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstdint>
|
||||
#include <string>
|
||||
#include <sys/wait.h>
|
||||
#include <thread>
|
||||
#include <unistd.h>
|
||||
#include <vector>
|
||||
|
||||
using namespace MobileGL::MG_Remote::Transport;
|
||||
|
||||
namespace {
|
||||
|
||||
constexpr std::uint64_t kSegmentSize = 64 * 1024;
|
||||
|
||||
std::uint8_t ByteAt(std::uint64_t index) {
|
||||
return static_cast<std::uint8_t>((index * 31u + 7u) & 0xFFu);
|
||||
}
|
||||
|
||||
// Child-side exit codes, so a failure says where it happened.
|
||||
enum ChildStatus : int {
|
||||
kChildOk = 0,
|
||||
kChildCreateFailed = 2,
|
||||
kChildMapFailed = 3,
|
||||
kChildSendFailed = 4,
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(FdPassingTest, IsSupportedOnThisPlatform) { EXPECT_TRUE(FdPassing::Supported()); }
|
||||
|
||||
TEST(FdPassingTest, ChildSharesASegmentThatTheParentMapsAndVerifies) {
|
||||
int sockets[2] = {-1, -1};
|
||||
ASSERT_EQ(FdPassing::CreateSocketPair(sockets), MOBILEGL_OK);
|
||||
|
||||
const std::string sideband = "SegmentRef{id=7,kind=Stage}";
|
||||
|
||||
const pid_t pid = ::fork();
|
||||
ASSERT_GE(pid, 0);
|
||||
if (pid == 0) {
|
||||
// Child. No gtest assertions here: a failed expectation in a forked
|
||||
// child would report into a copy of the parent's test state.
|
||||
::close(sockets[0]);
|
||||
int status = kChildOk;
|
||||
ShmSegment segment;
|
||||
if (ShmSegment::Create("fdpass", kSegmentSize, segment) != MOBILEGL_OK) {
|
||||
status = kChildCreateFailed;
|
||||
} else if (segment.Map(false) != MOBILEGL_OK) {
|
||||
status = kChildMapFailed;
|
||||
} else {
|
||||
auto* bytes = static_cast<std::uint8_t*>(segment.Data());
|
||||
for (std::uint64_t i = 0; i < kSegmentSize; ++i) {
|
||||
bytes[i] = ByteAt(i);
|
||||
}
|
||||
const MobileGLByteSpan span{sideband.data(), sideband.size()};
|
||||
if (FdPassing::SendFd(sockets[1], segment.Fd(), span) != MOBILEGL_OK) {
|
||||
status = kChildSendFailed;
|
||||
}
|
||||
}
|
||||
::close(sockets[1]);
|
||||
::_exit(status);
|
||||
}
|
||||
|
||||
// Parent.
|
||||
::close(sockets[1]);
|
||||
|
||||
// A destination smaller than kMaxSidebandBytes is refused BEFORE the
|
||||
// datagram is consumed, so the descriptor is not lost by a caller that
|
||||
// guessed the size wrong.
|
||||
std::vector<std::uint8_t> small(8);
|
||||
int fd = -1;
|
||||
std::uint64_t required = 0;
|
||||
MobileGLMutableByteSpan smallSpan{small.data(), small.size()};
|
||||
EXPECT_EQ(FdPassing::ReceiveFd(sockets[0], &fd, smallSpan, &required, 5000),
|
||||
MOBILEGL_ERR_BUFFER_TOO_SMALL);
|
||||
EXPECT_EQ(required, FdPassing::kMaxSidebandBytes);
|
||||
EXPECT_EQ(fd, -1);
|
||||
|
||||
std::vector<std::uint8_t> sidebandBuffer(FdPassing::kMaxSidebandBytes);
|
||||
std::uint64_t sidebandSize = 0;
|
||||
MobileGLMutableByteSpan sidebandSpan{sidebandBuffer.data(), sidebandBuffer.size()};
|
||||
ASSERT_EQ(FdPassing::ReceiveFd(sockets[0], &fd, sidebandSpan, &sidebandSize, 5000),
|
||||
MOBILEGL_OK);
|
||||
ASSERT_GE(fd, 0);
|
||||
EXPECT_EQ(std::string(reinterpret_cast<const char*>(sidebandBuffer.data()),
|
||||
static_cast<std::size_t>(sidebandSize)),
|
||||
sideband);
|
||||
|
||||
ShmSegment adopted;
|
||||
ASSERT_EQ(ShmSegment::Adopt(fd, kSegmentSize, adopted), MOBILEGL_OK);
|
||||
EXPECT_TRUE(adopted.Valid());
|
||||
ASSERT_EQ(adopted.Map(true), MOBILEGL_OK);
|
||||
EXPECT_TRUE(adopted.MappedReadOnly());
|
||||
|
||||
const auto* bytes = static_cast<const std::uint8_t*>(adopted.Data());
|
||||
ASSERT_NE(bytes, nullptr);
|
||||
std::uint64_t mismatches = 0;
|
||||
for (std::uint64_t i = 0; i < kSegmentSize; ++i) {
|
||||
if (bytes[i] != ByteAt(i)) {
|
||||
++mismatches;
|
||||
}
|
||||
}
|
||||
EXPECT_EQ(mismatches, 0u);
|
||||
|
||||
int childStatus = 0;
|
||||
ASSERT_EQ(::waitpid(pid, &childStatus, 0), pid);
|
||||
ASSERT_TRUE(WIFEXITED(childStatus));
|
||||
EXPECT_EQ(WEXITSTATUS(childStatus), kChildOk);
|
||||
|
||||
::close(sockets[0]);
|
||||
}
|
||||
|
||||
TEST(FdPassingTest, ReceiveTimesOutWithNoOffer) {
|
||||
int sockets[2] = {-1, -1};
|
||||
ASSERT_EQ(FdPassing::CreateSocketPair(sockets), MOBILEGL_OK);
|
||||
|
||||
std::vector<std::uint8_t> sidebandBuffer(FdPassing::kMaxSidebandBytes);
|
||||
int fd = -1;
|
||||
std::uint64_t sidebandSize = 0;
|
||||
MobileGLMutableByteSpan span{sidebandBuffer.data(), sidebandBuffer.size()};
|
||||
EXPECT_EQ(FdPassing::ReceiveFd(sockets[0], &fd, span, &sidebandSize, 20), MOBILEGL_ERR_TIMEOUT);
|
||||
EXPECT_EQ(fd, -1);
|
||||
|
||||
::close(sockets[0]);
|
||||
::close(sockets[1]);
|
||||
}
|
||||
|
||||
TEST(FdPassingTest, RejectsBadArguments) {
|
||||
int sockets[2] = {-1, -1};
|
||||
ASSERT_EQ(FdPassing::CreateSocketPair(sockets), MOBILEGL_OK);
|
||||
|
||||
const std::vector<std::uint8_t> tooBig(FdPassing::kMaxSidebandBytes + 1, 0);
|
||||
const MobileGLByteSpan oversized{tooBig.data(), tooBig.size()};
|
||||
EXPECT_EQ(FdPassing::SendFd(sockets[1], sockets[0], oversized), MOBILEGL_ERR_INVALID_ARGUMENT);
|
||||
EXPECT_EQ(FdPassing::SendFd(sockets[1], -1, MobileGLByteSpan{nullptr, 0}),
|
||||
MOBILEGL_ERR_INVALID_ARGUMENT);
|
||||
|
||||
::close(sockets[0]);
|
||||
::close(sockets[1]);
|
||||
}
|
||||
|
||||
TEST(FdPassingTest, SegmentWithoutASidebandStillCarriesItsDescriptor) {
|
||||
int sockets[2] = {-1, -1};
|
||||
ASSERT_EQ(FdPassing::CreateSocketPair(sockets), MOBILEGL_OK);
|
||||
|
||||
ShmSegment segment;
|
||||
ASSERT_EQ(ShmSegment::Create("nosideband", 4096, segment), MOBILEGL_OK);
|
||||
ASSERT_EQ(segment.Map(false), MOBILEGL_OK);
|
||||
static_cast<std::uint8_t*>(segment.Data())[0] = 0xA5;
|
||||
|
||||
ASSERT_EQ(FdPassing::SendFd(sockets[1], segment.Fd(), MobileGLByteSpan{nullptr, 0}),
|
||||
MOBILEGL_OK);
|
||||
|
||||
std::vector<std::uint8_t> sidebandBuffer(FdPassing::kMaxSidebandBytes);
|
||||
int fd = -1;
|
||||
std::uint64_t sidebandSize = 123;
|
||||
MobileGLMutableByteSpan span{sidebandBuffer.data(), sidebandBuffer.size()};
|
||||
ASSERT_EQ(FdPassing::ReceiveFd(sockets[0], &fd, span, &sidebandSize, 5000), MOBILEGL_OK);
|
||||
EXPECT_EQ(sidebandSize, 0u);
|
||||
ASSERT_GE(fd, 0);
|
||||
|
||||
ShmSegment adopted;
|
||||
ASSERT_EQ(ShmSegment::Adopt(fd, 4096, adopted), MOBILEGL_OK);
|
||||
ASSERT_EQ(adopted.Map(true), MOBILEGL_OK);
|
||||
EXPECT_EQ(static_cast<const std::uint8_t*>(adopted.Data())[0], 0xA5);
|
||||
|
||||
::close(sockets[0]);
|
||||
::close(sockets[1]);
|
||||
}
|
||||
|
||||
// The `spawn` doorbell rides the same kind of socket as the fd channel, so it
|
||||
// is covered here rather than beside the in-process one.
|
||||
TEST(FdPassingTest, SocketDoorbellWakesAParkedWaiter) {
|
||||
int sockets[2] = {-1, -1};
|
||||
ASSERT_EQ(FdPassing::CreateSocketPair(sockets), MOBILEGL_OK);
|
||||
|
||||
// One end each: the waiter reads its own end, the notifier writes the
|
||||
// other, exactly as the two processes will.
|
||||
SocketDoorbell waiterBell(sockets[0], kDoorbellWatermarkAdvanced, /*ownsFd=*/false);
|
||||
SocketDoorbell notifierBell(sockets[1], kDoorbellWatermarkAdvanced, /*ownsFd=*/false);
|
||||
|
||||
std::atomic<std::uint32_t> parked{0};
|
||||
std::atomic<bool> ready{false};
|
||||
std::atomic<bool> woke{false};
|
||||
|
||||
std::thread waiter([&] {
|
||||
woke.store(waiterBell.Wait(
|
||||
parked, [&] { return ready.load(std::memory_order_acquire); }, kDefaultSpinUs, 5000));
|
||||
});
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
ready.store(true, std::memory_order_release);
|
||||
NotifyIfParked(notifierBell, parked);
|
||||
|
||||
waiter.join();
|
||||
EXPECT_TRUE(woke.load());
|
||||
EXPECT_EQ(parked.load(), 0u);
|
||||
|
||||
::close(sockets[0]);
|
||||
::close(sockets[1]);
|
||||
}
|
||||
|
||||
TEST(FdPassingTest, SocketDoorbellTimesOutAndRemembersAnEarlyWakeup) {
|
||||
int sockets[2] = {-1, -1};
|
||||
ASSERT_EQ(FdPassing::CreateSocketPair(sockets), MOBILEGL_OK);
|
||||
|
||||
SocketDoorbell waiterBell(sockets[0], kDoorbellRingAdvanced, /*ownsFd=*/false);
|
||||
SocketDoorbell notifierBell(sockets[1], kDoorbellRingAdvanced, /*ownsFd=*/false);
|
||||
|
||||
// Nothing rings: the park has to end on its deadline, not hang.
|
||||
const auto start = std::chrono::steady_clock::now();
|
||||
EXPECT_FALSE(waiterBell.Park(30));
|
||||
EXPECT_GE(std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::steady_clock::now() - start)
|
||||
.count(),
|
||||
20);
|
||||
|
||||
// A wakeup that arrives before anyone parks is not lost - it is sitting in
|
||||
// the socket buffer, so the next Park returns at once.
|
||||
notifierBell.Notify();
|
||||
EXPECT_TRUE(waiterBell.Park(1000));
|
||||
// ...and it was consumed, so the one after that times out again.
|
||||
EXPECT_FALSE(waiterBell.Park(10));
|
||||
|
||||
::close(sockets[0]);
|
||||
::close(sockets[1]);
|
||||
}
|
||||
@@ -0,0 +1,197 @@
|
||||
// MobileGL - MobileGL/MG_Test/Wire/FramingTest.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
|
||||
|
||||
// The control-channel frame codec, and specifically the two contracts the
|
||||
// earlier branch's codec got wrong: a bad header must be REPORTED (it used to
|
||||
// turn into a silent permanent hang) and a too-small destination buffer must
|
||||
// KEEP the message (it used to fail the call and drop it, wedging the stream).
|
||||
|
||||
#include <MG_Remote/Transport/Framing.h>
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <cstring>
|
||||
#include <numeric>
|
||||
#include <vector>
|
||||
|
||||
using namespace MobileGL::MG_Remote::Transport;
|
||||
|
||||
namespace {
|
||||
|
||||
std::vector<std::uint8_t> Pattern(std::size_t size, std::uint8_t seed) {
|
||||
std::vector<std::uint8_t> out(size);
|
||||
for (std::size_t i = 0; i < size; ++i) {
|
||||
out[i] = static_cast<std::uint8_t>(seed + i * 7u);
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(FramingTest, RoundTripsTwoMessagesFedOneByteAtATime) {
|
||||
const std::vector<std::uint8_t> first = Pattern(37, 0x11);
|
||||
const std::vector<std::uint8_t> second = Pattern(120, 0x83);
|
||||
|
||||
std::vector<std::uint8_t> stream;
|
||||
ASSERT_EQ(AppendFrame(stream, first.data(), first.size()), MOBILEGL_OK);
|
||||
ASSERT_EQ(AppendFrame(stream, second.data(), second.size()), MOBILEGL_OK);
|
||||
EXPECT_EQ(stream.size(), 2 * kFrameHeaderSize + first.size() + second.size());
|
||||
|
||||
// A stream transport hands over arbitrary fragments; one byte at a time is
|
||||
// the worst case and must work.
|
||||
FrameReader reader;
|
||||
std::vector<std::vector<std::uint8_t>> received;
|
||||
for (std::uint8_t byte : stream) {
|
||||
ASSERT_EQ(reader.Feed(&byte, 1), MOBILEGL_OK);
|
||||
while (reader.HasMessage()) {
|
||||
std::vector<std::uint8_t> message;
|
||||
ASSERT_EQ(reader.TakeMessage(message), MOBILEGL_OK);
|
||||
received.push_back(std::move(message));
|
||||
}
|
||||
}
|
||||
|
||||
ASSERT_EQ(received.size(), 2u);
|
||||
EXPECT_EQ(received[0], first);
|
||||
EXPECT_EQ(received[1], second);
|
||||
EXPECT_FALSE(reader.Failed());
|
||||
EXPECT_EQ(reader.BufferedBytes(), 0u);
|
||||
}
|
||||
|
||||
TEST(FramingTest, MagicIsOnTheWireAsMGLF) {
|
||||
std::vector<std::uint8_t> stream;
|
||||
const std::uint8_t payload = 0xAB;
|
||||
ASSERT_EQ(AppendFrame(stream, &payload, 1), MOBILEGL_OK);
|
||||
ASSERT_GE(stream.size(), 4u);
|
||||
EXPECT_EQ(stream[0], 'M');
|
||||
EXPECT_EQ(stream[1], 'G');
|
||||
EXPECT_EQ(stream[2], 'L');
|
||||
EXPECT_EQ(stream[3], 'F');
|
||||
}
|
||||
|
||||
TEST(FramingTest, EmptyPayloadRoundTrips) {
|
||||
std::vector<std::uint8_t> stream;
|
||||
ASSERT_EQ(AppendFrame(stream, nullptr, 0), MOBILEGL_OK);
|
||||
|
||||
FrameReader reader;
|
||||
ASSERT_EQ(reader.Feed(stream.data(), stream.size()), MOBILEGL_OK);
|
||||
ASSERT_TRUE(reader.HasMessage());
|
||||
EXPECT_EQ(reader.PendingMessageSize(), 0u);
|
||||
|
||||
std::vector<std::uint8_t> message{0xFF};
|
||||
ASSERT_EQ(reader.TakeMessage(message), MOBILEGL_OK);
|
||||
EXPECT_TRUE(message.empty());
|
||||
}
|
||||
|
||||
TEST(FramingTest, BadMagicIsReportedAndLatchesTheReaderDead) {
|
||||
std::uint8_t header[8] = {};
|
||||
const std::uint32_t wrongMagic = 0xDEADBEEF;
|
||||
const std::uint32_t length = 4;
|
||||
std::memcpy(header + 0, &wrongMagic, sizeof(wrongMagic));
|
||||
std::memcpy(header + 4, &length, sizeof(length));
|
||||
|
||||
FrameReader reader;
|
||||
// The failure surfaces at Feed time, not as a message that never arrives.
|
||||
EXPECT_EQ(reader.Feed(header, sizeof(header)), MOBILEGL_ERR_PROTOCOL_MISMATCH);
|
||||
EXPECT_TRUE(reader.Failed());
|
||||
EXPECT_FALSE(reader.HasMessage());
|
||||
|
||||
// And it stays dead: a desynchronized stream is never re-synchronized by
|
||||
// feeding it more bytes.
|
||||
const std::uint8_t more[4] = {1, 2, 3, 4};
|
||||
EXPECT_EQ(reader.Feed(more, sizeof(more)), MOBILEGL_ERR_PROTOCOL_MISMATCH);
|
||||
std::vector<std::uint8_t> message;
|
||||
EXPECT_EQ(reader.TakeMessage(message), MOBILEGL_ERR_PROTOCOL_MISMATCH);
|
||||
}
|
||||
|
||||
TEST(FramingTest, OversizedLengthIsRejectedBeforeAnyAllocation) {
|
||||
std::uint8_t header[8] = {};
|
||||
const std::uint32_t magic = kFrameMagic;
|
||||
const std::uint32_t length = static_cast<std::uint32_t>(kMaxFramePayloadSize) + 1;
|
||||
std::memcpy(header + 0, &magic, sizeof(magic));
|
||||
std::memcpy(header + 4, &length, sizeof(length));
|
||||
|
||||
FrameReader reader;
|
||||
EXPECT_EQ(reader.Feed(header, sizeof(header)), MOBILEGL_ERR_PROTOCOL_MISMATCH);
|
||||
EXPECT_TRUE(reader.Failed());
|
||||
}
|
||||
|
||||
TEST(FramingTest, SendRefusesAPayloadOverTheCap) {
|
||||
std::vector<std::uint8_t> stream;
|
||||
const std::uint8_t dummy = 0;
|
||||
// The size check happens before the payload is touched, so no 64MiB
|
||||
// allocation is needed to cover it.
|
||||
EXPECT_EQ(AppendFrame(stream, &dummy, kMaxFramePayloadSize + 1), MOBILEGL_ERR_INVALID_ARGUMENT);
|
||||
EXPECT_TRUE(stream.empty());
|
||||
}
|
||||
|
||||
TEST(FramingTest, BufferTooSmallReportsTheSizeAndKeepsTheMessage) {
|
||||
const std::vector<std::uint8_t> payload = Pattern(200, 0x5A);
|
||||
std::vector<std::uint8_t> stream;
|
||||
ASSERT_EQ(AppendFrame(stream, payload.data(), payload.size()), MOBILEGL_OK);
|
||||
|
||||
FrameReader reader;
|
||||
ASSERT_EQ(reader.Feed(stream.data(), stream.size()), MOBILEGL_OK);
|
||||
ASSERT_TRUE(reader.HasMessage());
|
||||
|
||||
std::vector<std::uint8_t> small(8);
|
||||
std::uint64_t required = 0;
|
||||
MobileGLMutableByteSpan smallSpan{small.data(), small.size()};
|
||||
EXPECT_EQ(reader.TakeMessage(smallSpan, &required), MOBILEGL_ERR_BUFFER_TOO_SMALL);
|
||||
EXPECT_EQ(required, payload.size());
|
||||
|
||||
// Still there. This is the whole point: the old transport dropped it here
|
||||
// and the stream never recovered.
|
||||
ASSERT_TRUE(reader.HasMessage());
|
||||
|
||||
std::vector<std::uint8_t> big(required);
|
||||
std::uint64_t got = 0;
|
||||
MobileGLMutableByteSpan bigSpan{big.data(), big.size()};
|
||||
ASSERT_EQ(reader.TakeMessage(bigSpan, &got), MOBILEGL_OK);
|
||||
EXPECT_EQ(got, payload.size());
|
||||
EXPECT_EQ(big, payload);
|
||||
EXPECT_FALSE(reader.HasMessage());
|
||||
}
|
||||
|
||||
TEST(FramingTest, TakeWithNoCompleteMessageDoesNotBlockOrCorrupt) {
|
||||
const std::vector<std::uint8_t> payload = Pattern(64, 0x22);
|
||||
std::vector<std::uint8_t> stream;
|
||||
ASSERT_EQ(AppendFrame(stream, payload.data(), payload.size()), MOBILEGL_OK);
|
||||
|
||||
FrameReader reader;
|
||||
// Header plus half the payload.
|
||||
ASSERT_EQ(reader.Feed(stream.data(), kFrameHeaderSize + 32), MOBILEGL_OK);
|
||||
EXPECT_FALSE(reader.HasMessage());
|
||||
EXPECT_EQ(reader.PendingMessageSize(), 0u);
|
||||
|
||||
std::vector<std::uint8_t> message;
|
||||
EXPECT_EQ(reader.TakeMessage(message), MOBILEGL_ERR_TIMEOUT);
|
||||
|
||||
ASSERT_EQ(reader.Feed(stream.data() + kFrameHeaderSize + 32,
|
||||
stream.size() - kFrameHeaderSize - 32),
|
||||
MOBILEGL_OK);
|
||||
ASSERT_TRUE(reader.HasMessage());
|
||||
ASSERT_EQ(reader.TakeMessage(message), MOBILEGL_OK);
|
||||
EXPECT_EQ(message, payload);
|
||||
}
|
||||
|
||||
TEST(FramingTest, ManyMessagesCompactTheBufferInsteadOfGrowing) {
|
||||
// Drives the reader past its compaction threshold so the "consumed bytes
|
||||
// are reclaimed" path is actually taken.
|
||||
const std::vector<std::uint8_t> payload = Pattern(1024, 0x07);
|
||||
FrameReader reader;
|
||||
for (int i = 0; i < 300; ++i) {
|
||||
std::vector<std::uint8_t> stream;
|
||||
ASSERT_EQ(AppendFrame(stream, payload.data(), payload.size()), MOBILEGL_OK);
|
||||
ASSERT_EQ(reader.Feed(stream.data(), stream.size()), MOBILEGL_OK);
|
||||
ASSERT_TRUE(reader.HasMessage());
|
||||
std::vector<std::uint8_t> message;
|
||||
ASSERT_EQ(reader.TakeMessage(message), MOBILEGL_OK);
|
||||
ASSERT_EQ(message, payload);
|
||||
}
|
||||
EXPECT_EQ(reader.BufferedBytes(), 0u);
|
||||
}
|
||||
@@ -0,0 +1,276 @@
|
||||
// MobileGL - MobileGL/MG_Test/Wire/InProcessTransportTest.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
|
||||
|
||||
// The `inproc` transport: message queues in both directions, the
|
||||
// buffer-too-small contract, shutdown semantics, descriptor hand-off, and the
|
||||
// condvar doorbells the rings park on.
|
||||
|
||||
#include <MG_Remote/Transport/FdPassing.h>
|
||||
#include <MG_Remote/Transport/InProcessTransport.h>
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <cstring>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#if !defined(_WIN32)
|
||||
#include <unistd.h>
|
||||
#endif
|
||||
|
||||
using namespace MobileGL::MG_Remote::Transport;
|
||||
|
||||
namespace {
|
||||
|
||||
MobileGLByteSpan Span(const std::string& text) {
|
||||
return MobileGLByteSpan{text.data(), text.size()};
|
||||
}
|
||||
|
||||
std::string Receive(ITransport& transport, std::uint32_t timeoutMs = 1000) {
|
||||
std::vector<std::uint8_t> buffer(4096);
|
||||
std::uint64_t size = 0;
|
||||
MobileGLMutableByteSpan span{buffer.data(), buffer.size()};
|
||||
const MobileGLResult result = transport.ReceiveFrame(span, &size, timeoutMs);
|
||||
if (result != MOBILEGL_OK) {
|
||||
return std::string("<result=") + std::to_string(static_cast<int>(result)) + ">";
|
||||
}
|
||||
return std::string(reinterpret_cast<const char*>(buffer.data()),
|
||||
static_cast<std::size_t>(size));
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(InProcessTransportTest, CarriesFramesInBothDirections) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
ASSERT_TRUE(client && server);
|
||||
EXPECT_EQ(client->Role(), TransportRole::InProcess);
|
||||
|
||||
const std::string hello = "Hello{abiMajor=1}";
|
||||
const std::string welcome = "Welcome{serverPid=42}";
|
||||
ASSERT_EQ(client->SendFrame(Span(hello)), MOBILEGL_OK);
|
||||
EXPECT_EQ(server->PeekFrameSize(), hello.size());
|
||||
// A message goes to the PEER's inbox, never back to the sender.
|
||||
EXPECT_EQ(client->PeekFrameSize(), 0u);
|
||||
EXPECT_EQ(Receive(*server), hello);
|
||||
|
||||
ASSERT_EQ(server->SendFrame(Span(welcome)), MOBILEGL_OK);
|
||||
EXPECT_EQ(Receive(*client), welcome);
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, PreservesOrder) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
for (int i = 0; i < 64; ++i) {
|
||||
const std::string message = "msg-" + std::to_string(i);
|
||||
ASSERT_EQ(client->SendFrame(Span(message)), MOBILEGL_OK);
|
||||
}
|
||||
for (int i = 0; i < 64; ++i) {
|
||||
EXPECT_EQ(Receive(*server), "msg-" + std::to_string(i));
|
||||
}
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, BufferTooSmallKeepsTheMessage) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
const std::string message(300, 'x');
|
||||
ASSERT_EQ(client->SendFrame(Span(message)), MOBILEGL_OK);
|
||||
|
||||
std::vector<std::uint8_t> small(16);
|
||||
std::uint64_t required = 0;
|
||||
MobileGLMutableByteSpan smallSpan{small.data(), small.size()};
|
||||
EXPECT_EQ(server->ReceiveFrame(smallSpan, &required, 0), MOBILEGL_ERR_BUFFER_TOO_SMALL);
|
||||
EXPECT_EQ(required, message.size());
|
||||
// Still queued - the caller just retries with the size it was told.
|
||||
EXPECT_EQ(server->PeekFrameSize(), message.size());
|
||||
EXPECT_EQ(Receive(*server), message);
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, PollAndTimeoutDoNotBlockForever) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
std::vector<std::uint8_t> buffer(64);
|
||||
std::uint64_t size = 0;
|
||||
MobileGLMutableByteSpan span{buffer.data(), buffer.size()};
|
||||
EXPECT_EQ(server->ReceiveFrame(span, &size, 0), MOBILEGL_ERR_TIMEOUT);
|
||||
|
||||
const auto start = std::chrono::steady_clock::now();
|
||||
EXPECT_EQ(server->ReceiveFrame(span, &size, 30), MOBILEGL_ERR_TIMEOUT);
|
||||
EXPECT_GE(std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::steady_clock::now() - start)
|
||||
.count(),
|
||||
20);
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, ShutdownDrainsBeforeItCloses) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
const std::string last = "Fatal{code=DeviceLost}";
|
||||
ASSERT_EQ(client->SendFrame(Span(last)), MOBILEGL_OK);
|
||||
client->Shutdown();
|
||||
|
||||
// A peer that shuts down right after sending must not lose its last
|
||||
// message - that is usually the one that says why it is going away.
|
||||
EXPECT_EQ(Receive(*server), last);
|
||||
|
||||
std::vector<std::uint8_t> buffer(64);
|
||||
std::uint64_t size = 0;
|
||||
MobileGLMutableByteSpan span{buffer.data(), buffer.size()};
|
||||
EXPECT_EQ(server->ReceiveFrame(span, &size, 100), MOBILEGL_ERR_TRANSPORT_CLOSED);
|
||||
EXPECT_EQ(server->SendFrame(Span(last)), MOBILEGL_ERR_TRANSPORT_CLOSED);
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, BlockedReceiverWakesOnSendAndOnShutdown) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
std::atomic<bool> got{false};
|
||||
std::thread reader([&] {
|
||||
got.store(Receive(*server, kWaitForever) == "wake");
|
||||
});
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
ASSERT_EQ(client->SendFrame(Span(std::string("wake"))), MOBILEGL_OK);
|
||||
reader.join();
|
||||
EXPECT_TRUE(got.load());
|
||||
|
||||
std::thread closer([&] {
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
client->Shutdown();
|
||||
});
|
||||
std::vector<std::uint8_t> buffer(64);
|
||||
std::uint64_t size = 0;
|
||||
MobileGLMutableByteSpan span{buffer.data(), buffer.size()};
|
||||
EXPECT_EQ(server->ReceiveFrame(span, &size, kWaitForever), MOBILEGL_ERR_TRANSPORT_CLOSED);
|
||||
closer.join();
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, RefusesAPayloadOverTheFrameCap) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
// Not allocated: the cap is checked before the bytes are touched. Keeping
|
||||
// the same limit as the socket transports means nothing passes CI here and
|
||||
// then fails after the switch to `spawn`.
|
||||
const std::uint8_t dummy = 0;
|
||||
MobileGLByteSpan huge{&dummy, 64ull * 1024 * 1024 + 1};
|
||||
EXPECT_EQ(client->SendFrame(huge), MOBILEGL_ERR_INVALID_ARGUMENT);
|
||||
}
|
||||
|
||||
#if !defined(_WIN32)
|
||||
TEST(InProcessTransportTest, HandsOverADescriptorAndItsSideband) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
int pipeFds[2] = {-1, -1};
|
||||
ASSERT_EQ(::pipe(pipeFds), 0);
|
||||
|
||||
const std::string sideband = "SegmentRef{id=1,kind=Cmd}";
|
||||
ASSERT_EQ(client->ShareFd(pipeFds[0], Span(sideband)), MOBILEGL_OK);
|
||||
|
||||
// Symmetric with the SCM_RIGHTS path: a short sideband buffer is refused
|
||||
// before anything is consumed, so the descriptor is never dropped.
|
||||
std::vector<std::uint8_t> small(8);
|
||||
int fd = -1;
|
||||
std::uint64_t required = 0;
|
||||
MobileGLMutableByteSpan smallSpan{small.data(), small.size()};
|
||||
EXPECT_EQ(server->ReceiveFd(&fd, smallSpan, &required, 0), MOBILEGL_ERR_BUFFER_TOO_SMALL);
|
||||
EXPECT_EQ(required, FdPassing::kMaxSidebandBytes);
|
||||
EXPECT_EQ(fd, -1);
|
||||
|
||||
std::vector<std::uint8_t> big(FdPassing::kMaxSidebandBytes);
|
||||
std::uint64_t sidebandSize = 0;
|
||||
MobileGLMutableByteSpan bigSpan{big.data(), big.size()};
|
||||
ASSERT_EQ(server->ReceiveFd(&fd, bigSpan, &sidebandSize, 100), MOBILEGL_OK);
|
||||
ASSERT_GE(fd, 0);
|
||||
EXPECT_EQ(std::string(reinterpret_cast<const char*>(big.data()),
|
||||
static_cast<std::size_t>(sidebandSize)),
|
||||
sideband);
|
||||
|
||||
// Same open file description, independent descriptor.
|
||||
const char payload[] = "bytes";
|
||||
ASSERT_EQ(::write(pipeFds[1], payload, sizeof(payload)), static_cast<ssize_t>(sizeof(payload)));
|
||||
char readBack[sizeof(payload)] = {};
|
||||
ASSERT_EQ(::read(fd, readBack, sizeof(readBack)), static_cast<ssize_t>(sizeof(payload)));
|
||||
EXPECT_STREQ(readBack, payload);
|
||||
|
||||
::close(fd);
|
||||
::close(pipeFds[0]);
|
||||
::close(pipeFds[1]);
|
||||
}
|
||||
#endif
|
||||
|
||||
TEST(InProcessTransportTest, DoorbellWakesAParkedWaiter) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
// producerParked / consumerParked live in RingControl; here a standalone
|
||||
// flag stands in for one.
|
||||
std::atomic<std::uint32_t> parked{0};
|
||||
std::atomic<bool> ready{false};
|
||||
std::atomic<bool> woke{false};
|
||||
|
||||
std::thread waiter([&] {
|
||||
woke.store(client->SelfDoorbell().Wait(
|
||||
parked, [&] { return ready.load(std::memory_order_acquire); }, kDefaultSpinUs, 5000));
|
||||
});
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(20));
|
||||
ready.store(true, std::memory_order_release);
|
||||
// The peer only rings when the waiter says it parked, which is what makes
|
||||
// the common (spin-only) case free.
|
||||
NotifyIfParked(server->PeerDoorbell(), parked);
|
||||
|
||||
waiter.join();
|
||||
EXPECT_TRUE(woke.load());
|
||||
EXPECT_EQ(parked.load(), 0u);
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, DoorbellReturnsImmediatelyWhenAlreadyReady) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
std::atomic<std::uint32_t> parked{0};
|
||||
// No notification is sent at all: a condition that is already true must
|
||||
// never park, or the lost-wakeup window would be reachable.
|
||||
EXPECT_TRUE(client->SelfDoorbell().Wait(
|
||||
parked, [] { return true; }, kDefaultSpinUs, 0));
|
||||
EXPECT_EQ(parked.load(), 0u);
|
||||
}
|
||||
|
||||
TEST(InProcessTransportTest, DoorbellTimesOutWhenNothingHappens) {
|
||||
std::unique_ptr<InProcessTransport> client;
|
||||
std::unique_ptr<InProcessTransport> server;
|
||||
InProcessTransport::CreatePair(client, server);
|
||||
|
||||
std::atomic<std::uint32_t> parked{0};
|
||||
const auto start = std::chrono::steady_clock::now();
|
||||
EXPECT_FALSE(client->SelfDoorbell().Wait(
|
||||
parked, [] { return false; }, kDefaultSpinUs, 30));
|
||||
EXPECT_GE(std::chrono::duration_cast<std::chrono::milliseconds>(
|
||||
std::chrono::steady_clock::now() - start)
|
||||
.count(),
|
||||
20);
|
||||
EXPECT_EQ(parked.load(), 0u);
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
// MobileGL - MobileGL/MG_Test/Wire/ProtocolSmokeTest.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
|
||||
|
||||
// The committed control-plane schema: encode/decode a handshake through the
|
||||
// generated header, and pin the union tag values, which are wire numbers that
|
||||
// may only ever be appended to.
|
||||
|
||||
#include <MG_Remote/Protocol/generated/protocol_generated.h>
|
||||
#include <MG_Remote/Transport/Framing.h>
|
||||
#include <MG_Remote/Transport/InProcessTransport.h>
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <cstdint>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <vector>
|
||||
|
||||
using namespace MobileGL::Wire;
|
||||
namespace Transport = MobileGL::MG_Remote::Transport;
|
||||
|
||||
namespace {
|
||||
|
||||
std::vector<std::uint8_t> BuildHello() {
|
||||
::flatbuffers::FlatBufferBuilder builder(1024);
|
||||
const std::vector<std::uint8_t> config{1, 2, 3, 4};
|
||||
auto hello = CreateHelloDirect(builder, MOBILEGL_PROTOCOL_ABI_MAJOR,
|
||||
MOBILEGL_PROTOCOL_ABI_MINOR, "mobilegl-test-build",
|
||||
/*backendType=*/2, /*pid=*/4242, &config);
|
||||
auto envelope = CreateCtrlEnvelope(builder, CtrlMsg::Hello, hello.Union());
|
||||
FinishCtrlEnvelopeBuffer(builder, envelope);
|
||||
const std::uint8_t* begin = builder.GetBufferPointer();
|
||||
return std::vector<std::uint8_t>(begin, begin + builder.GetSize());
|
||||
}
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(ProtocolSmokeTest, HelloRoundTrips) {
|
||||
const std::vector<std::uint8_t> buffer = BuildHello();
|
||||
|
||||
// Every message from the peer is verified before a single field is read:
|
||||
// the control plane is parsed from another process's memory.
|
||||
::flatbuffers::Verifier verifier(buffer.data(), buffer.size());
|
||||
ASSERT_TRUE(VerifyCtrlEnvelopeBuffer(verifier));
|
||||
ASSERT_TRUE(CtrlEnvelopeBufferHasIdentifier(buffer.data()));
|
||||
|
||||
const CtrlEnvelope* envelope = GetCtrlEnvelope(buffer.data());
|
||||
ASSERT_NE(envelope, nullptr);
|
||||
ASSERT_EQ(envelope->msg_type(), CtrlMsg::Hello);
|
||||
|
||||
const Hello* hello = envelope->msg_as_Hello();
|
||||
ASSERT_NE(hello, nullptr);
|
||||
EXPECT_EQ(hello->abiMajor(), static_cast<std::uint32_t>(MOBILEGL_PROTOCOL_ABI_MAJOR));
|
||||
EXPECT_EQ(hello->abiMinor(), static_cast<std::uint32_t>(MOBILEGL_PROTOCOL_ABI_MINOR));
|
||||
ASSERT_NE(hello->buildFingerprint(), nullptr);
|
||||
EXPECT_EQ(hello->buildFingerprint()->str(), "mobilegl-test-build");
|
||||
EXPECT_EQ(hello->backendType(), 2u);
|
||||
EXPECT_EQ(hello->pid(), 4242u);
|
||||
ASSERT_NE(hello->configBlob(), nullptr);
|
||||
ASSERT_EQ(hello->configBlob()->size(), 4u);
|
||||
EXPECT_EQ(hello->configBlob()->Get(3), 4u);
|
||||
|
||||
// A message of the wrong kind reads back as null rather than as garbage.
|
||||
EXPECT_EQ(envelope->msg_as_Welcome(), nullptr);
|
||||
}
|
||||
|
||||
TEST(ProtocolSmokeTest, WelcomeCarriesTheFourSegmentAnnouncements) {
|
||||
::flatbuffers::FlatBufferBuilder builder(1024);
|
||||
auto cmd = CreateSegmentRefDirect(builder, 1, SegmentKind::Cmd, 8ull * 1024 * 1024, "cmd");
|
||||
auto stage = CreateSegmentRefDirect(builder, 2, SegmentKind::Stage, 32ull * 1024 * 1024, "stage");
|
||||
auto reply = CreateSegmentRefDirect(builder, 3, SegmentKind::Reply, 8ull * 1024 * 1024, "reply");
|
||||
auto event = CreateSegmentRefDirect(builder, 4, SegmentKind::Event, 256ull * 1024, "event");
|
||||
auto welcome = CreateWelcome(builder, MOBILEGL_PROTOCOL_ABI_MAJOR, MOBILEGL_PROTOCOL_ABI_MINOR,
|
||||
/*serverPid=*/99, cmd, stage, reply, event);
|
||||
auto envelope = CreateCtrlEnvelope(builder, CtrlMsg::Welcome, welcome.Union());
|
||||
FinishCtrlEnvelopeBuffer(builder, envelope);
|
||||
|
||||
::flatbuffers::Verifier verifier(builder.GetBufferPointer(), builder.GetSize());
|
||||
ASSERT_TRUE(VerifyCtrlEnvelopeBuffer(verifier));
|
||||
|
||||
const Welcome* parsed = GetCtrlEnvelope(builder.GetBufferPointer())->msg_as_Welcome();
|
||||
ASSERT_NE(parsed, nullptr);
|
||||
EXPECT_EQ(parsed->serverPid(), 99u);
|
||||
ASSERT_NE(parsed->cmdRing(), nullptr);
|
||||
EXPECT_EQ(parsed->cmdRing()->kind(), SegmentKind::Cmd);
|
||||
EXPECT_EQ(parsed->cmdRing()->sizeBytes(), 8ull * 1024 * 1024);
|
||||
ASSERT_NE(parsed->stageRing(), nullptr);
|
||||
EXPECT_EQ(parsed->stageRing()->sizeBytes(), 32ull * 1024 * 1024);
|
||||
ASSERT_NE(parsed->eventRing(), nullptr);
|
||||
EXPECT_EQ(parsed->eventRing()->sizeBytes(), 256ull * 1024);
|
||||
}
|
||||
|
||||
TEST(ProtocolSmokeTest, UnionTagsAreFrozenWireValues) {
|
||||
// Appending to CtrlMsg is a compatible change; reordering it is not. If
|
||||
// this test has to be edited, the schema change was a wire break.
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::NONE), 0);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::Hello), 1);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::Welcome), 2);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::CapsSnapshot), 3);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::SurfaceOp), 4);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::SurfaceReply), 5);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::ResyncRequest), 6);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::ResyncDone), 7);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::AuxRequest), 8);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::Fatal), 9);
|
||||
EXPECT_EQ(static_cast<int>(CtrlMsg::LogLine), 10);
|
||||
|
||||
EXPECT_EQ(static_cast<int>(SegmentKind::Cmd), 1);
|
||||
EXPECT_EQ(static_cast<int>(SegmentKind::Adopt), 6);
|
||||
EXPECT_EQ(static_cast<int>(LogLevel::Error), 3);
|
||||
EXPECT_EQ(static_cast<int>(FatalCode::ProtocolCorruption), 1);
|
||||
}
|
||||
|
||||
TEST(ProtocolSmokeTest, TruncatedMessageFailsVerificationInsteadOfReadingGarbage) {
|
||||
std::vector<std::uint8_t> buffer = BuildHello();
|
||||
ASSERT_GT(buffer.size(), 8u);
|
||||
buffer.resize(buffer.size() / 2);
|
||||
|
||||
::flatbuffers::Verifier verifier(buffer.data(), buffer.size());
|
||||
EXPECT_FALSE(VerifyCtrlEnvelopeBuffer(verifier));
|
||||
}
|
||||
|
||||
TEST(ProtocolSmokeTest, TravelsAcrossTheTransportUnchanged) {
|
||||
std::unique_ptr<Transport::InProcessTransport> client;
|
||||
std::unique_ptr<Transport::InProcessTransport> server;
|
||||
Transport::InProcessTransport::CreatePair(client, server);
|
||||
|
||||
const std::vector<std::uint8_t> sent = BuildHello();
|
||||
ASSERT_EQ(client->SendFrame(MobileGLByteSpan{sent.data(), sent.size()}), MOBILEGL_OK);
|
||||
|
||||
const std::uint64_t pending = server->PeekFrameSize();
|
||||
ASSERT_EQ(pending, sent.size());
|
||||
std::vector<std::uint8_t> received(pending);
|
||||
std::uint64_t size = 0;
|
||||
MobileGLMutableByteSpan span{received.data(), received.size()};
|
||||
ASSERT_EQ(server->ReceiveFrame(span, &size, 1000), MOBILEGL_OK);
|
||||
ASSERT_EQ(size, sent.size());
|
||||
|
||||
::flatbuffers::Verifier verifier(received.data(), received.size());
|
||||
ASSERT_TRUE(VerifyCtrlEnvelopeBuffer(verifier));
|
||||
const Hello* hello = GetCtrlEnvelope(received.data())->msg_as_Hello();
|
||||
ASSERT_NE(hello, nullptr);
|
||||
EXPECT_EQ(hello->pid(), 4242u);
|
||||
}
|
||||
@@ -0,0 +1,325 @@
|
||||
// MobileGL - MobileGL/MG_Test/Wire/RingTest.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
|
||||
|
||||
// The SEG_CMD/SEG_STAGE SPSC ring: layout of the shared control page, cursor
|
||||
// invariants, wrap-around, backpressure, the generation bump after a hard
|
||||
// drain, and a real two-thread producer/consumer run.
|
||||
|
||||
#include <MG_Remote/Transport/Ring.h>
|
||||
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <cstring>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
using namespace MobileGL::MG_Remote::Transport;
|
||||
|
||||
namespace {
|
||||
|
||||
// A ring plus its control page, sized like a small SEG_CMD.
|
||||
class RingFixture {
|
||||
public:
|
||||
explicit RingFixture(std::uint64_t capacity, RingCursorSet cursors = RingCursorSet::Cmd)
|
||||
: m_bytes(static_cast<std::size_t>(capacity)), m_capacity(capacity) {
|
||||
InitRingControl(m_control);
|
||||
m_producer = RingProducer(&m_control, m_bytes.data(), capacity, cursors);
|
||||
m_consumer = RingConsumer(&m_control, m_bytes.data(), capacity, cursors);
|
||||
m_cursors = cursors;
|
||||
}
|
||||
|
||||
RingControl& Control() { return m_control; }
|
||||
RingProducer& Producer() { return m_producer; }
|
||||
RingConsumer& Consumer() { return m_consumer; }
|
||||
std::uint64_t Capacity() const { return m_capacity; }
|
||||
bool Invariants() const { return RingCursorsValid(m_control, m_cursors, m_capacity); }
|
||||
|
||||
// Writes one record whose payload is `size` bytes of a recognisable
|
||||
// pattern seeded by `seed`.
|
||||
bool WriteRecord(std::uint16_t kind, std::uint64_t size, std::uint8_t seed) {
|
||||
void* payload = m_producer.Reserve(kind, kRecNone, size);
|
||||
if (payload == nullptr) {
|
||||
return false;
|
||||
}
|
||||
auto* bytes = static_cast<std::uint8_t*>(payload);
|
||||
for (std::uint64_t i = 0; i < size; ++i) {
|
||||
bytes[i] = static_cast<std::uint8_t>(seed + i);
|
||||
}
|
||||
m_producer.Publish();
|
||||
return true;
|
||||
}
|
||||
|
||||
static bool CheckPattern(const RingRecordView& view, std::uint64_t size, std::uint8_t seed) {
|
||||
const auto* bytes = static_cast<const std::uint8_t*>(view.payload);
|
||||
for (std::uint64_t i = 0; i < size; ++i) {
|
||||
if (bytes[i] != static_cast<std::uint8_t>(seed + i)) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private:
|
||||
alignas(4096) RingControl m_control{};
|
||||
std::vector<std::uint8_t> m_bytes;
|
||||
RingProducer m_producer;
|
||||
RingConsumer m_consumer;
|
||||
std::uint64_t m_capacity;
|
||||
RingCursorSet m_cursors = RingCursorSet::Cmd;
|
||||
};
|
||||
|
||||
} // namespace
|
||||
|
||||
TEST(RingTest, ControlPageLayoutIsTheSharedContract) {
|
||||
// The page is mapped by two processes; its size and alignment are wire
|
||||
// contract, not an implementation detail.
|
||||
EXPECT_EQ(sizeof(RingControl), 4096u);
|
||||
EXPECT_EQ(alignof(RingControl), 4096u);
|
||||
EXPECT_EQ(sizeof(RingRecordHeader), 8u);
|
||||
|
||||
alignas(4096) RingControl control{};
|
||||
InitRingControl(control);
|
||||
// Zero is reserved for "uninitialized" on both generations.
|
||||
EXPECT_EQ(control.serverEpoch.load(), 1u);
|
||||
EXPECT_EQ(control.ringGeneration.load(), 1u);
|
||||
EXPECT_EQ(control.cmdHead.load(), 0u);
|
||||
EXPECT_EQ(control.stageHead.load(), 0u);
|
||||
EXPECT_EQ(control.consumerParked.load(), 0u);
|
||||
EXPECT_EQ(control.producerParked.load(), 0u);
|
||||
EXPECT_EQ(control.eventRingFull.load(), 0u);
|
||||
EXPECT_EQ(control.eventDropped.load(), 0u);
|
||||
|
||||
// Each contended group on its own cache line.
|
||||
const auto offset = [&control](const void* member) {
|
||||
return reinterpret_cast<const std::uint8_t*>(member) -
|
||||
reinterpret_cast<const std::uint8_t*>(&control);
|
||||
};
|
||||
EXPECT_EQ(offset(&control.cmdHead) % 64, 0);
|
||||
EXPECT_EQ(offset(&control.cmdAppliedTail) % 64, 0);
|
||||
EXPECT_EQ(offset(&control.stageHead) % 64, 0);
|
||||
EXPECT_EQ(offset(&control.stageAppliedTail) % 64, 0);
|
||||
EXPECT_EQ(offset(&control.appliedSeq) % 64, 0);
|
||||
EXPECT_EQ(offset(&control.serverEpoch) % 64, 0);
|
||||
// cmdHead and cmdAppliedTail are written by different processes: they must
|
||||
// not share a line.
|
||||
EXPECT_NE(offset(&control.cmdHead) / 64, offset(&control.cmdAppliedTail) / 64);
|
||||
}
|
||||
|
||||
TEST(RingTest, RejectsANonPowerOfTwoCapacity) {
|
||||
alignas(4096) RingControl control{};
|
||||
InitRingControl(control);
|
||||
std::vector<std::uint8_t> bytes(1000);
|
||||
RingProducer producer(&control, bytes.data(), 1000, RingCursorSet::Cmd);
|
||||
EXPECT_FALSE(producer.Valid());
|
||||
EXPECT_EQ(producer.Reserve(1, kRecNone, 8), nullptr);
|
||||
}
|
||||
|
||||
TEST(RingTest, RoundTripsRecordsInOrder) {
|
||||
RingFixture ring(4096);
|
||||
ASSERT_TRUE(ring.WriteRecord(1, 16, 0x10));
|
||||
ASSERT_TRUE(ring.WriteRecord(2, 24, 0x20));
|
||||
EXPECT_TRUE(ring.Invariants());
|
||||
|
||||
RingRecordView view{};
|
||||
bool corrupt = false;
|
||||
ASSERT_TRUE(ring.Consumer().Pop(view, &corrupt));
|
||||
EXPECT_FALSE(corrupt);
|
||||
EXPECT_EQ(view.kind, 1u);
|
||||
EXPECT_EQ(view.payloadSize, 16u);
|
||||
EXPECT_TRUE(RingFixture::CheckPattern(view, 16, 0x10));
|
||||
|
||||
ASSERT_TRUE(ring.Consumer().Pop(view, &corrupt));
|
||||
EXPECT_EQ(view.kind, 2u);
|
||||
EXPECT_EQ(view.payloadSize, 24u);
|
||||
EXPECT_TRUE(RingFixture::CheckPattern(view, 24, 0x20));
|
||||
|
||||
EXPECT_FALSE(ring.Consumer().Pop(view, &corrupt));
|
||||
ring.Consumer().PublishRetired();
|
||||
EXPECT_TRUE(ring.Invariants());
|
||||
EXPECT_EQ(ring.Control().cmdAppliedTail.load(), ring.Control().cmdHead.load());
|
||||
EXPECT_EQ(ring.Control().cmdRetiredTail.load(), ring.Control().cmdHead.load());
|
||||
}
|
||||
|
||||
TEST(RingTest, PayloadIsPaddedToTheRecordAlignment) {
|
||||
RingFixture ring(4096);
|
||||
ASSERT_TRUE(ring.WriteRecord(7, 3, 0x77));
|
||||
RingRecordView view{};
|
||||
ASSERT_TRUE(ring.Consumer().Pop(view));
|
||||
// 8 (header) + 3 rounded up to 16 -> 8 bytes of payload space.
|
||||
EXPECT_EQ(view.payloadSize, 8u);
|
||||
EXPECT_TRUE(RingFixture::CheckPattern(view, 3, 0x77));
|
||||
}
|
||||
|
||||
TEST(RingTest, WrapsWithoutSplittingARecord) {
|
||||
// Small ring, records that do not divide it evenly, so the wrap boundary
|
||||
// lands mid-record and the pad path is exercised many times.
|
||||
RingFixture ring(256);
|
||||
std::uint8_t seed = 0;
|
||||
for (int i = 0; i < 200; ++i) {
|
||||
const std::uint64_t size = 24 + (i % 5) * 8;
|
||||
ASSERT_TRUE(ring.WriteRecord(static_cast<std::uint16_t>(1 + (i % 3)), size, seed))
|
||||
<< "record " << i;
|
||||
RingRecordView view{};
|
||||
bool corrupt = false;
|
||||
ASSERT_TRUE(ring.Consumer().Pop(view, &corrupt)) << "record " << i;
|
||||
ASSERT_FALSE(corrupt);
|
||||
EXPECT_EQ(view.kind, static_cast<std::uint16_t>(1 + (i % 3)));
|
||||
// Contiguity: the payload never straddles the end of the mapping.
|
||||
EXPECT_TRUE(RingFixture::CheckPattern(view, size, seed)) << "record " << i;
|
||||
ring.Consumer().PublishRetired();
|
||||
ASSERT_TRUE(ring.Invariants());
|
||||
seed = static_cast<std::uint8_t>(seed + 13);
|
||||
}
|
||||
// Cursors are monotonic byte counts, so they are far past the capacity.
|
||||
EXPECT_GT(ring.Control().cmdHead.load(), ring.Capacity());
|
||||
}
|
||||
|
||||
TEST(RingTest, FullRingRefusesAndRecoversWhenTheConsumerRetires) {
|
||||
RingFixture ring(256);
|
||||
int written = 0;
|
||||
while (ring.WriteRecord(1, 24, static_cast<std::uint8_t>(written))) {
|
||||
++written;
|
||||
ASSERT_LT(written, 100);
|
||||
}
|
||||
EXPECT_GT(written, 0);
|
||||
// Backpressure, not corruption.
|
||||
EXPECT_TRUE(ring.Invariants());
|
||||
EXPECT_LT(ring.Producer().FreeBytes(), 32u);
|
||||
|
||||
RingRecordView view{};
|
||||
ASSERT_TRUE(ring.Consumer().Pop(view));
|
||||
// Applied alone does not free a slot that may still be borrowed by the GPU
|
||||
// timeline: reclaim follows the retired cursor.
|
||||
ring.Consumer().PublishApplied();
|
||||
EXPECT_EQ(ring.Producer().FreeBytes(), 0u);
|
||||
ring.Consumer().PublishRetired();
|
||||
EXPECT_GT(ring.Producer().FreeBytes(), 0u);
|
||||
EXPECT_TRUE(ring.WriteRecord(1, 24, 0xEE));
|
||||
}
|
||||
|
||||
TEST(RingTest, RecordLargerThanTheRingIsRefused) {
|
||||
RingFixture ring(256);
|
||||
EXPECT_EQ(ring.Producer().Reserve(1, kRecNone, 4096), nullptr);
|
||||
EXPECT_TRUE(ring.Invariants());
|
||||
}
|
||||
|
||||
TEST(RingTest, HardDrainBumpsTheGenerationOnlyWhenQuiesced) {
|
||||
RingFixture ring(256);
|
||||
ASSERT_TRUE(ring.WriteRecord(1, 32, 0x01));
|
||||
const std::uint32_t before = ring.Control().ringGeneration.load();
|
||||
|
||||
// Records still in flight: the drain is refused and nothing changes.
|
||||
EXPECT_EQ(HardDrainRing(ring.Control(), RingCursorSet::Cmd), MOBILEGL_ERR_INVALID_ARGUMENT);
|
||||
EXPECT_EQ(ring.Control().ringGeneration.load(), before);
|
||||
|
||||
RingRecordView view{};
|
||||
ASSERT_TRUE(ring.Consumer().Pop(view));
|
||||
ring.Consumer().PublishRetired();
|
||||
EXPECT_EQ(HardDrainRing(ring.Control(), RingCursorSet::Cmd), MOBILEGL_OK);
|
||||
EXPECT_EQ(ring.Control().ringGeneration.load(), before + 1);
|
||||
// Cursors stay monotonic across the drain - only the generation moves.
|
||||
EXPECT_EQ(ring.Control().cmdHead.load(), ring.Control().cmdAppliedTail.load());
|
||||
EXPECT_GT(ring.Control().cmdHead.load(), 0u);
|
||||
}
|
||||
|
||||
TEST(RingTest, CorruptHeaderIsRefusedRatherThanDispatched) {
|
||||
// SEG_CMD is written by the peer process, so a compile-time size assert on
|
||||
// the record catalogue proves nothing about what is actually in the
|
||||
// mapping. Hand-build a ring whose first header is impossible (a size that
|
||||
// is not a multiple of 8) and check the consumer refuses it instead of
|
||||
// dispatching into undefined behaviour.
|
||||
alignas(4096) RingControl control{};
|
||||
InitRingControl(control);
|
||||
std::vector<std::uint8_t> bytes(256, 0);
|
||||
RingRecordHeader bad{};
|
||||
bad.kind = 5;
|
||||
bad.flags = kRecNone;
|
||||
bad.size = 13; // not 8-aligned
|
||||
std::memcpy(bytes.data(), &bad, sizeof(bad));
|
||||
control.cmdHead.store(64, std::memory_order_release);
|
||||
|
||||
RingConsumer consumer(&control, bytes.data(), bytes.size(), RingCursorSet::Cmd);
|
||||
RingRecordView view{};
|
||||
bool corrupt = false;
|
||||
EXPECT_FALSE(consumer.Pop(view, &corrupt));
|
||||
EXPECT_TRUE(corrupt);
|
||||
|
||||
// A record claiming more bytes than the producer has published is the same
|
||||
// class of violation and is refused the same way.
|
||||
bad.size = 128;
|
||||
std::memcpy(bytes.data(), &bad, sizeof(bad));
|
||||
RingConsumer second(&control, bytes.data(), bytes.size(), RingCursorSet::Cmd);
|
||||
corrupt = false;
|
||||
EXPECT_FALSE(second.Pop(view, &corrupt));
|
||||
EXPECT_TRUE(corrupt);
|
||||
}
|
||||
|
||||
TEST(RingTest, SpscProducerConsumerThreadsAgreeOnEveryRecord) {
|
||||
constexpr int kRecords = 20000;
|
||||
RingFixture ring(4096);
|
||||
|
||||
std::atomic<bool> failed{false};
|
||||
std::atomic<int> consumed{0};
|
||||
|
||||
std::thread consumer([&] {
|
||||
int next = 0;
|
||||
while (next < kRecords) {
|
||||
RingRecordView view{};
|
||||
bool corrupt = false;
|
||||
if (!ring.Consumer().Pop(view, &corrupt)) {
|
||||
if (corrupt) {
|
||||
failed.store(true);
|
||||
return;
|
||||
}
|
||||
std::this_thread::yield();
|
||||
continue;
|
||||
}
|
||||
const std::uint32_t expectedKind = static_cast<std::uint16_t>(1 + (next % 7));
|
||||
if (view.kind != expectedKind || view.payloadSize < sizeof(std::uint32_t)) {
|
||||
failed.store(true);
|
||||
return;
|
||||
}
|
||||
std::uint32_t value = 0;
|
||||
std::memcpy(&value, view.payload, sizeof(value));
|
||||
if (value != static_cast<std::uint32_t>(next)) {
|
||||
failed.store(true);
|
||||
return;
|
||||
}
|
||||
++next;
|
||||
consumed.store(next, std::memory_order_relaxed);
|
||||
// Retire as we go; a consumer that never retires would deadlock the
|
||||
// producer, which is exactly the contract being pinned.
|
||||
ring.Consumer().PublishRetired();
|
||||
}
|
||||
});
|
||||
|
||||
for (int i = 0; i < kRecords; ++i) {
|
||||
const std::uint64_t payloadSize = sizeof(std::uint32_t) + (i % 4) * 8;
|
||||
void* payload = nullptr;
|
||||
while ((payload = ring.Producer().Reserve(static_cast<std::uint16_t>(1 + (i % 7)),
|
||||
kRecNone, payloadSize)) == nullptr) {
|
||||
if (failed.load()) {
|
||||
break;
|
||||
}
|
||||
std::this_thread::yield();
|
||||
}
|
||||
if (payload == nullptr) {
|
||||
break;
|
||||
}
|
||||
const std::uint32_t value = static_cast<std::uint32_t>(i);
|
||||
std::memcpy(payload, &value, sizeof(value));
|
||||
ring.Producer().Publish();
|
||||
}
|
||||
|
||||
consumer.join();
|
||||
EXPECT_FALSE(failed.load());
|
||||
EXPECT_EQ(consumed.load(), kRecords);
|
||||
EXPECT_TRUE(ring.Invariants());
|
||||
EXPECT_EQ(ring.Control().cmdRetiredTail.load(), ring.Control().cmdHead.load());
|
||||
}
|
||||
Reference in New Issue
Block a user