MobileGL/MobileGL/MG_Test/Wire/StreamLinkTest.cpp

// SPDX-License-Identifier: LGPL-3.0-only
#include <MG_Remote/Transport/StreamLink.h>
#include <MG_Remote/Transport/SessionRings.h>
#include <gtest/gtest.h>
#include <array>
#include <chrono>
#include <cstring>
#include <thread>
#include <sys/socket.h>
#include <unistd.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <cstdio>

using namespace MobileGL::MG_Remote::Transport;
namespace {
    class StreamPair : public testing::Test {
    protected:
        SessionSegments clientMemory, serverMemory;
        StreamLink client, server;
        RingProducer commands;
        RingConsumer consumer;
        SessionProducer publish;
        SessionConsumer apply;
        void SetUp() override {
            SessionSegmentSizes sizes;
            sizes.CmdRingBytes = 1024;
            sizes.StageBytes = 4096;
            sizes.ReplyBytes = 4096;
            sizes.EventRingBytes = 1024;
            ASSERT_EQ(clientMemory.CreatePrivate(sizes, MemoryRole::Client), MOBILEGL_OK);
            ASSERT_EQ(serverMemory.CreatePrivate(sizes, MemoryRole::Server), MOBILEGL_OK);
            int sockets[2];
            ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, sockets), 0);
            ASSERT_EQ(client.Attach(sockets[0], clientMemory, TransportRoleTag::ClientProducer), MOBILEGL_OK);
            ASSERT_EQ(server.Attach(sockets[1], serverMemory, TransportRoleTag::ServerConsumer), MOBILEGL_OK);
            commands = RingProducer(clientMemory.CmdControl(), clientMemory.CmdRingBase(), 1024, RingCursorSet::Cmd);
            consumer = RingConsumer(serverMemory.CmdControl(), serverMemory.CmdRingBase(), 1024, RingCursorSet::Cmd);
            publish.Attach(clientMemory.CmdControl(), &commands, &client.ConsumerBell(), &client.ProducerBell(), 0);
            publish.SetLink(&client);
            apply.Attach(serverMemory.CmdControl(), &consumer, &server.ProducerBell(), &server.ConsumerBell(), 0);
            apply.SetLink(&server);
        }
        void TearDown() override {
            client.Detach();
            server.Detach();
        }
        void Queue(std::uint64_t seq, std::uint64_t value) {
            auto* p = commands.Reserve(1, 0, sizeof value);
            ASSERT_NE(p, nullptr);
            std::memcpy(p, &value, sizeof value);
            publish.PublishAndNotify(seq);
        }
        template <class F>
        bool Eventually(F ready) {
            const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(2);
            while (!ready() && std::chrono::steady_clock::now() < deadline)
                std::this_thread::yield();
            return ready();
        }
    };

    TEST_F(StreamPair, StagingArrivesBeforeItsCommandAndLastRecordProgressFlushesOnIdle) {
        const std::array<unsigned char, 8> bytes{1, 2, 3, 4, 5, 6, 7, 8};
        std::memcpy(static_cast<unsigned char*>(clientMemory.StageBase()) + 24, bytes.data(), bytes.size());
        client.NoteStage({24, bytes.size()});
        Queue(1, 24);
        ASSERT_EQ(client.Flush(), MOBILEGL_OK);
        ASSERT_EQ(apply.WaitForWork(2000), SessionWait::Reached);
        RingRecordView view;
        ASSERT_TRUE(consumer.Pop(view));
        [&](const RingRecordView& v) {
            std::uint64_t offset;
            std::memcpy(&offset, v.payload, sizeof offset);
            const void* p = nullptr;
            ASSERT_EQ(server.ResolveSpan(LinkSegment::Stage, {offset, bytes.size()}, &p), MOBILEGL_OK);
            EXPECT_EQ(std::memcmp(p, bytes.data(), bytes.size()), 0);
        }(view);
        // Advance locally without a periodic-progress callback: this isolates
        // flush-on-idle so removing it cannot be masked by the one-ms timer.
        Watermark::AdvanceApplied(*serverMemory.CmdControl(), 1);
        Watermark::AdvanceRetired(*serverMemory.CmdControl(), 1);
        consumer.PublishRetired();
        // This is the production idle discipline, not a polling sender thread.
        EXPECT_EQ(apply.WaitForWork(1), SessionWait::TimedOut);
        EXPECT_EQ(publish.WaitForApplied(1, 2000), SessionWait::Reached);
        // applied and retired are independent watermarks. The reader publishes
        // them in dependency order, and an applied waiter can run between stores.
        EXPECT_TRUE(client.ProducerBell().Wait(
            clientMemory.CmdControl()->producerParked,
            [&] { return client.Progress()->retiredSeq.load(std::memory_order_acquire) >= 1; }, 0, 2000));
    }

    TEST_F(StreamPair, RetiredByteCursorCrossesManyWrapsWithoutStageRingCursors) {
        for (std::uint64_t seq = 1; seq <= 160; ++seq) {
            Queue(seq, seq);
            ASSERT_EQ(client.Flush(), MOBILEGL_OK);
            ASSERT_EQ(apply.WaitForWork(2000), SessionWait::Reached);
            ASSERT_TRUE(apply.ApplyOne([&](const RingRecordView& v) {
                std::uint64_t value;
                std::memcpy(&value, v.payload, sizeof value);
                EXPECT_EQ(value, seq);
            }));
            apply.RetireThrough(seq);
            ASSERT_EQ(server.FlushProgress(), MOBILEGL_OK);
            ASSERT_EQ(publish.WaitForApplied(seq, 2000), SessionWait::Reached);
        }
        EXPECT_GT(commands.LocalHead(), 1024u);
        EXPECT_EQ(commands.FreeBytes(), 1024u);
        EXPECT_EQ(clientMemory.CmdControl()->stageHead.load(), 0u);
        EXPECT_EQ(clientMemory.CmdControl()->stageRetiredTail.load(), 0u);
    }

    TEST_F(StreamPair, UnwantedReplyDoesNotBlockLaterReplyAndAppliedPublication) {
        Queue(1, 1);
        Queue(2, 2);
        ASSERT_EQ(client.Flush(), MOBILEGL_OK);
        ASSERT_EQ(apply.WaitForWork(2000), SessionWait::Reached);
        ASSERT_TRUE(apply.ApplyOne([&](const RingRecordView&) {
            const std::uint64_t value = 11;
            EXPECT_EQ(server.PostReply(1, 0, &value, sizeof value), MOBILEGL_OK);
        }));
        ASSERT_TRUE(apply.ApplyOne([&](const RingRecordView&) {
            const std::uint64_t value = 22;
            EXPECT_EQ(server.PostReply(2, 0, &value, sizeof value), MOBILEGL_OK);
        }));
        apply.RetireThrough(2);
        ASSERT_EQ(server.FlushProgress(), MOBILEGL_OK);
        ASSERT_EQ(publish.WaitForApplied(2, 2000), SessionWait::Reached);
        std::int32_t status;
        const void* p = nullptr;
        std::uint64_t size = 0;
        ASSERT_EQ(client.ReadReply(2, &status, &p, &size), MOBILEGL_OK);
        ASSERT_EQ(size, sizeof(std::uint64_t));
        std::uint64_t value;
        std::memcpy(&value, p, sizeof value);
        EXPECT_EQ(value, 22u);
    }

    TEST_F(StreamPair, EventDrainReturnsCreditAndClearsRemoteFullLatch) {
        EventRingProducer events(serverMemory.EventControl(), serverMemory.CmdControl(), serverMemory.EventRingBase(),
                                 1024);
        EventRingConsumer read(clientMemory.EventControl(), clientMemory.CmdControl(), clientMemory.EventRingBase(),
                               1024, clientMemory.EventSegmentBase());
        read.SetLink(&client);
        auto* p = events.Reserve(kEventGlError, 64);
        ASSERT_NE(p, nullptr);
        std::memset(p, 0, 64);
        events.Ring().Publish();
        serverMemory.CmdControl()->eventRingFull.store(1);
        ASSERT_EQ(server.FlushProgress(), MOBILEGL_OK);
        ASSERT_TRUE(Eventually([&] { return clientMemory.CmdControl()->eventRingFull.load() != 0; }));
        RingRecordView record;
        ASSERT_TRUE(read.Pop(record));
        read.Drained();
        ASSERT_TRUE(Eventually([&] { return serverMemory.CmdControl()->eventRingFull.load() == 0; }));
        EXPECT_EQ(events.Ring().FreeBytes(), 1024u);
    }

    TEST_F(StreamPair, PeerEofKillsBothLocalBellsAndWakesAWaiter) {
        server.Detach();
        ASSERT_TRUE(Eventually([&] { return client.PeerHungUp(); }));
        EXPECT_TRUE(client.ProducerBell().Dead());
        EXPECT_TRUE(client.ConsumerBell().Dead());
        EXPECT_EQ(publish.WaitForApplied(10, 2000), SessionWait::ShutDown);
    }

    TEST_F(StreamPair, PeerEofReleasesAnAlreadyWaitingReplyReader) {
        std::atomic<bool> finished{false};
        MobileGLResult result = MOBILEGL_OK;
        std::thread waiting([&] {
            std::int32_t status;
            const void* payload;
            std::uint64_t size;
            result = client.ReadReply(1, &status, &payload, &size);
            finished.store(true, std::memory_order_release);
        });
        server.Detach();
        EXPECT_TRUE(Eventually([&] { return finished.load(std::memory_order_acquire); }));
        waiting.join();
        EXPECT_EQ(result, MOBILEGL_ERR_TRANSPORT_CLOSED);
    }

    TEST_F(StreamPair, AppliedWithoutItsReplyIsRefusedWithoutAnUnboundedWait) {
        Queue(1, 1);
        ASSERT_EQ(client.Flush(), MOBILEGL_OK);
        ASSERT_EQ(apply.WaitForWork(2000), SessionWait::Reached);
        ASSERT_TRUE(apply.ApplyOne([](const RingRecordView&) {}));
        apply.RetireThrough(1);
        ASSERT_EQ(server.FlushProgress(), MOBILEGL_OK);
        ASSERT_EQ(publish.WaitForApplied(1, 2000), SessionWait::Reached);
        std::int32_t status;
        const void* p;
        std::uint64_t size;
        EXPECT_EQ(client.ReadReply(1, &status, &p, &size), MOBILEGL_ERR_PROTOCOL_MISMATCH);
    }

    TEST_F(StreamPair, ControlFenceWaitsForEventBytesOnTheIndependentDataConnection) {
        EventRingProducer events(serverMemory.EventControl(), serverMemory.CmdControl(), serverMemory.EventRingBase(),
                                 1024);
        auto* payload = events.Reserve(kEventGlError, 8);
        ASSERT_NE(payload, nullptr);
        std::memset(payload, 0, 8);
        events.Ring().Publish();
        const auto fence = server.EventPublishedHead();
        EXPECT_EQ(client.WaitForEventDelivery(fence, 1), MOBILEGL_ERR_TIMEOUT);
        ASSERT_EQ(server.Flush(), MOBILEGL_OK);
        EXPECT_EQ(client.WaitForEventDelivery(fence, 2000), MOBILEGL_OK);
        EXPECT_GE(client.EventPublishedHead(), fence);
    }

    TEST(StreamLinkTcp, RecordsSixtyFourMiBStageBurstThroughputOnRealLoopbackTcp) {
        int listener = ::socket(AF_INET, SOCK_STREAM, 0);
        ASSERT_GE(listener, 0);
        sockaddr_in address{};
        address.sin_family = AF_INET;
        address.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
        ASSERT_EQ(::bind(listener, reinterpret_cast<sockaddr*>(&address), sizeof address), 0);
        ASSERT_EQ(::listen(listener, 1), 0);
        socklen_t addressBytes = sizeof address;
        ASSERT_EQ(::getsockname(listener, reinterpret_cast<sockaddr*>(&address), &addressBytes), 0);
        int clientFd = ::socket(AF_INET, SOCK_STREAM, 0);
        ASSERT_GE(clientFd, 0);
        ASSERT_EQ(::connect(clientFd, reinterpret_cast<sockaddr*>(&address), sizeof address), 0);
        int serverFd = ::accept(listener, nullptr, nullptr);
        ASSERT_GE(serverFd, 0);
        ::close(listener);
        int enabled = 1;
        ASSERT_EQ(::setsockopt(clientFd, IPPROTO_TCP, TCP_NODELAY, &enabled, sizeof enabled), 0);
        ASSERT_EQ(::setsockopt(serverFd, IPPROTO_TCP, TCP_NODELAY, &enabled, sizeof enabled), 0);
        SessionSegmentSizes sizes;
        sizes.CmdRingBytes = 1024;
        sizes.EventRingBytes = 1024;
        sizes.StageBytes = 64ull * 1024 * 1024;
        sizes.ReplyBytes = 4096;
        std::unique_ptr<ILink> client, server;
        ASSERT_EQ(CreateStreamLink(clientFd, sizes, TransportRoleTag::ClientProducer, client), MOBILEGL_OK);
        ASSERT_EQ(CreateStreamLink(serverFd, sizes, TransportRoleTag::ServerConsumer, server), MOBILEGL_OK);
        client->InitializeEndpoints();
        server->InitializeEndpoints();
        std::memset(client->Memory().StageBase(), 0x5a, static_cast<std::size_t>(sizes.StageBytes));
        client->NoteStage({0, sizes.StageBytes});
        SessionProducer producer;
        producer.Attach(*client, 0);
        SessionConsumer consumer;
        consumer.Attach(*server, 0);
        auto* record = client->CommandsOut().Reserve(1, 0, sizeof(std::uint64_t));
        ASSERT_NE(record, nullptr);
        std::memcpy(record, &sizes.StageBytes, sizeof(std::uint64_t));
        producer.PublishAndNotify(1);
        const auto start = std::chrono::steady_clock::now();
        ASSERT_EQ(client->Flush(), MOBILEGL_OK);
        ASSERT_EQ(consumer.WaitForWork(10000), SessionWait::Reached);
        ASSERT_TRUE(consumer.ApplyOne([&](const RingRecordView&) {
            const auto* bytes = static_cast<const unsigned char*>(server->Memory().StageBase());
            for (std::uint64_t offset = 0; offset < sizes.StageBytes; offset += 4096)
                EXPECT_EQ(bytes[offset], 0x5a);
            EXPECT_EQ(bytes[sizes.StageBytes - 1], 0x5a);
        }));
        consumer.RetireThrough(1);
        ASSERT_EQ(server->FlushProgress(), MOBILEGL_OK);
        ASSERT_EQ(producer.WaitForApplied(1, 10000), SessionWait::Reached);
        const double elapsed = std::chrono::duration<double>(std::chrono::steady_clock::now() - start).count();
        const double mibPerSecond = 64.0 / elapsed;
        RecordProperty("tcp_stage_bytes", std::to_string(sizes.StageBytes));
        RecordProperty("tcp_stage_mib_per_second", std::to_string(mibPerSecond));
        std::printf("P6.5 Stage burst control=none data=tcp-loopback bytes=%llu seconds=%.6f MiB/s=%.3f\n",
                    static_cast<unsigned long long>(sizes.StageBytes), elapsed, mibPerSecond);
    }

    void WriteFrame(int fd, unsigned kind, std::uint64_t a, const void* payload, std::uint32_t size) {
        unsigned char h[28]{};
        auto put = [&](unsigned at, std::uint64_t value, unsigned width) {
            for (unsigned i = 0; i < width; ++i)
                h[at + i] = value >> (8 * i);
        };
        put(0, 0x444c474d, 4);
        put(4, size + 20, 4);
        h[8] = kind;
        put(12, a, 8);
        ASSERT_EQ(::send(fd, h, sizeof h, MSG_NOSIGNAL), sizeof h);
        if (size) ASSERT_EQ(::send(fd, payload, size, MSG_NOSIGNAL), size);
    }

    TEST(StreamLinkDeath, StageOutsideNegotiatedWindowIsNamedProtocolCorruption) {
        EXPECT_DEATH(
            {
                SessionSegments memory;
                SessionSegmentSizes sizes;
                sizes.StageBytes = 64;
                memory.CreatePrivate(sizes, MemoryRole::Server);
                int sockets[2];
                socketpair(AF_UNIX, SOCK_STREAM, 0, sockets);
                StreamLink link;
                link.Attach(sockets[0], memory, TransportRoleTag::ServerConsumer);
                unsigned char byte = 1;
                WriteFrame(sockets[1], 2, 64, &byte, 1);
                std::this_thread::sleep_for(std::chrono::seconds(1));
            },
            "ProtocolCorruption.*window");
    }

    TEST(StreamLinkDeath, RegressingProgressIsNamedProtocolCorruption) {
        EXPECT_DEATH(
            {
                SessionSegments memory;
                SessionSegmentSizes sizes;
                memory.CreatePrivate(sizes, MemoryRole::Client);
                memory.CmdControl()->submittedSeq.store(3);
                int sockets[2];
                socketpair(AF_UNIX, SOCK_STREAM, 0, sockets);
                StreamLink link;
                link.Attach(sockets[0], memory, TransportRoleTag::ClientProducer);
                unsigned char progress[64]{};
                progress[0] = 2;
                WriteFrame(sockets[1], 4, 0, progress, 64);
                progress[0] = 1;
                WriteFrame(sockets[1], 4, 0, progress, 64);
                std::this_thread::sleep_for(std::chrono::seconds(1));
            },
            "ProtocolCorruption.*backwards");
    }
} // namespace