diff --git a/MobileGL/MG_Remote/Server/ServerSession.cpp b/MobileGL/MG_Remote/Server/ServerSession.cpp index c860d166..715aca79 100644 --- a/MobileGL/MG_Remote/Server/ServerSession.cpp +++ b/MobileGL/MG_Remote/Server/ServerSession.cpp @@ -9,7 +9,6 @@ // P5 package s1: the server half of a session. #include "../Transport/SocketTransport.h" -#include "../Transport/StreamLink.h" #include "ServerSession.h" #include @@ -564,7 +563,7 @@ namespace MobileGL::MG_Remote::Server { // its descriptor: Welcome has to announce the sizes of memory the connection it names // will fill. Uint8 dataNonce[Transport::kDataNonceBytes] = {}; - Transport::StreamLink* pendingStream = nullptr; + Transport::ILink* pendingStream = nullptr; if (stream) { auto* socket = dynamic_cast(&transport); if (socket == nullptr || !socket->IsTcp() || !m_dataSource) @@ -581,8 +580,8 @@ namespace MobileGL::MG_Remote::Server { "stream data plane unavailable: the server could not mint a data nonce"); return created; } - auto link = std::make_unique(); - created = link->AttachOwnedDeferred(m_sizes, Transport::TransportRoleTag::ServerConsumer); + std::unique_ptr link; + created = Transport::CreateDeferredStreamLink(m_sizes, Transport::TransportRoleTag::ServerConsumer, link); if (created == MOBILEGL_OK) { pendingStream = link.get(); AttachDataLink(std::move(link)); @@ -801,7 +800,7 @@ namespace MobileGL::MG_Remote::Server { // control peer going away; the deadline is refused on the control connection, where the // client is listening. MobileGLResult ServerSession::BindDataConnection(Transport::ITransport& control, - Transport::StreamLink& link, const Uint8* nonce) { + Transport::ILink& link, const Uint8* nonce) { const auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(kHandshakeTimeoutMs); Uint32 refused = 0; for (;;) { @@ -823,7 +822,7 @@ namespace MobileGL::MG_Remote::Server { return offered; } if (Transport::ConstantTimeNonceMatch(nonce, presented)) { - const MobileGLResult bound = link.BindDataFd(fd); + const MobileGLResult bound = Transport::BindStreamLinkDataFd(link, fd); #if !defined(_WIN32) if (bound != MOBILEGL_OK) ::close(fd); #endif @@ -837,14 +836,6 @@ namespace MobileGL::MG_Remote::Server { } } - MobileGLResult ServerSession::AttachStreamLink(int fd, const Transport::SessionSegmentSizes& sizes) { - std::unique_ptr link; - const auto attached = Transport::CreateStreamLink(fd, sizes, Transport::TransportRoleTag::ServerConsumer, link); - if (attached != MOBILEGL_OK) return attached; - AttachDataLink(std::move(link)); - return MOBILEGL_OK; - } - void ServerSession::AttachDataLink(std::unique_ptr link) { m_link = std::move(link); m_commands = &m_link->CommandsIn(); m_events = &m_link->EventsOut(); diff --git a/MobileGL/MG_Remote/Server/ServerSession.h b/MobileGL/MG_Remote/Server/ServerSession.h index 715de479..e5c4e612 100644 --- a/MobileGL/MG_Remote/Server/ServerSession.h +++ b/MobileGL/MG_Remote/Server/ServerSession.h @@ -58,10 +58,6 @@ #include "../Wire/PipeWireCodec.h" #include "PipeApplier.h" -namespace MobileGL::MG_Remote::Transport { - class StreamLink; -} - namespace MobileGL::MG_Remote::Server { class ServerSession { @@ -70,7 +66,6 @@ namespace MobileGL::MG_Remote::Server { ServerSession(); ~ServerSession(); - MobileGLResult AttachStreamLink(int dataFd, const Transport::SessionSegmentSizes& sizes); void AttachDataLink(std::unique_ptr link); Transport::ILink* DataLink() const { return m_link.get(); } void FlushDataProgress(); @@ -227,7 +222,7 @@ namespace MobileGL::MG_Remote::Server { void LogMemory(const char* phase) const; private: - MobileGLResult BindDataConnection(Transport::ITransport& control, Transport::StreamLink& link, + MobileGLResult BindDataConnection(Transport::ITransport& control, Transport::ILink& link, const Uint8* nonce); DataConnectionSource m_dataSource; diff --git a/MobileGL/MG_Remote/Transport/ILink.h b/MobileGL/MG_Remote/Transport/ILink.h index 913d76b5..1a9639e1 100644 --- a/MobileGL/MG_Remote/Transport/ILink.h +++ b/MobileGL/MG_Remote/Transport/ILink.h @@ -343,5 +343,15 @@ namespace MobileGL::MG_Remote::Transport { std::unique_ptr CreateSharedLink(TransportRoleTag role); MobileGLResult CreateStreamLink(int fd, const SessionSegmentSizes& sizes, TransportRoleTag role, std::unique_ptr& out); + // PH-7 (4), ID-P7-3. The server's nonce-bound stream link, in the two steps + // StreamLink::AttachOwnedDeferred / BindDataFd describe: created with its owned memory and + // no descriptor (Welcome announces those sizes before the data connection exists), then + // bound to the one data connection that presented the session's nonce. Both halves live + // here so session code never names the concrete link (scripts/ci/link_seam_purity.py). + // Binding answers INVALID_ARGUMENT for a link these factories did not make as a stream, + // and leaves `fd` to the caller whenever it does not answer OK. + MobileGLResult CreateDeferredStreamLink(const SessionSegmentSizes& sizes, TransportRoleTag role, + std::unique_ptr& out); + MobileGLResult BindStreamLinkDataFd(ILink& link, int fd); } // namespace MobileGL::MG_Remote::Transport diff --git a/MobileGL/MG_Remote/Transport/StreamLink.cpp b/MobileGL/MG_Remote/Transport/StreamLink.cpp index 1d77a0ed..ea093746 100644 --- a/MobileGL/MG_Remote/Transport/StreamLink.cpp +++ b/MobileGL/MG_Remote/Transport/StreamLink.cpp @@ -399,6 +399,18 @@ namespace MobileGL::MG_Remote::Transport { #endif return result; } + MobileGLResult CreateDeferredStreamLink(const SessionSegmentSizes& sizes, TransportRoleTag role, + std::unique_ptr& out) { + auto link = std::make_unique(); + const auto result = link->AttachOwnedDeferred(sizes, role); + if (result == MOBILEGL_OK) out = std::move(link); + return result; + } + MobileGLResult BindStreamLinkDataFd(ILink& link, int fd) { + auto* stream = dynamic_cast(&link); + if (stream == nullptr) return MOBILEGL_ERR_INVALID_ARGUMENT; + return stream->BindDataFd(fd); + } void StreamLink::InitializeEndpoints() { m_impl->owned->SetRole(m_impl->role); m_impl->owned->InitializeEndpoints(); diff --git a/MobileGL/MG_Test/Wire/StreamLinkTest.cpp b/MobileGL/MG_Test/Wire/StreamLinkTest.cpp index 32b13ff4..7dab5059 100644 --- a/MobileGL/MG_Test/Wire/StreamLinkTest.cpp +++ b/MobileGL/MG_Test/Wire/StreamLinkTest.cpp @@ -268,6 +268,60 @@ namespace { static_cast(sizes.StageBytes), elapsed, mibPerSecond); } + // PH-7 (4), ID-P7-3, through the Transport factory pair ServerSession::Accept uses so that + // session code never names the concrete link (scripts/ci/link_seam_purity.py). In the + // production order: created without a descriptor, endpoints initialised, then bound once. + // A second bind, and a bind onto a link the factory did not make as a stream, are refused + // and leave the descriptor with the caller; the bound link then carries a record. + TEST(StreamLinkFactory, DeferredServerLinkBindsOnceAndThenCarriesARecord) { + SessionSegmentSizes sizes; + sizes.CmdRingBytes = 1024; + sizes.EventRingBytes = 1024; + sizes.StageBytes = 4096; + sizes.ReplyBytes = 4096; + std::unique_ptr server; + ASSERT_EQ(CreateDeferredStreamLink(sizes, TransportRoleTag::ServerConsumer, server), MOBILEGL_OK); + ASSERT_NE(server, nullptr); + EXPECT_FALSE(server->Capabilities().PublishIsDelivery); + server->InitializeEndpoints(); + + int sockets[2]; + ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, sockets), 0); + std::unique_ptr client; + ASSERT_EQ(CreateStreamLink(sockets[0], sizes, TransportRoleTag::ClientProducer, client), MOBILEGL_OK); + client->InitializeEndpoints(); + ASSERT_EQ(BindStreamLinkDataFd(*server, sockets[1]), MOBILEGL_OK); + + int spare[2]; + ASSERT_EQ(::socketpair(AF_UNIX, SOCK_STREAM, 0, spare), 0); + EXPECT_EQ(BindStreamLinkDataFd(*server, spare[0]), MOBILEGL_ERR_INVALID_ARGUMENT); + auto shared = CreateSharedLink(TransportRoleTag::ServerConsumer); + ASSERT_NE(shared, nullptr); + EXPECT_EQ(BindStreamLinkDataFd(*shared, spare[1]), MOBILEGL_ERR_INVALID_ARGUMENT); + EXPECT_EQ(::close(spare[0]), 0); + EXPECT_EQ(::close(spare[1]), 0); + + SessionProducer producer; + producer.Attach(*client, 0); + SessionConsumer consumer; + consumer.Attach(*server, 0); + const std::uint64_t value = 0x5a5a5a5aull; + auto* record = client->CommandsOut().Reserve(1, 0, sizeof value); + ASSERT_NE(record, nullptr); + std::memcpy(record, &value, sizeof value); + producer.PublishAndNotify(1); + ASSERT_EQ(client->Flush(), MOBILEGL_OK); + ASSERT_EQ(consumer.WaitForWork(2000), SessionWait::Reached); + ASSERT_TRUE(consumer.ApplyOne([&](const RingRecordView& v) { + std::uint64_t seen = 0; + std::memcpy(&seen, v.payload, sizeof seen); + EXPECT_EQ(seen, value); + })); + consumer.RetireThrough(1); + ASSERT_EQ(server->FlushProgress(), MOBILEGL_OK); + EXPECT_EQ(producer.WaitForApplied(1, 2000), SessionWait::Reached); + } + 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) {