From f7182159b4fef6d24dcf25f6c588d00a3c3ecb4b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 2 Jun 2026 01:13:08 +0100 Subject: [PATCH] Add C# CultMesh streaming surface --- CultLib.sln | 15 + README.md | 2 +- packages/cultmesh-ts/README.md | 20 +- src/GameCult.Mesh/CultMesh.cs | 8 + src/GameCult.Mesh/CultMeshStreaming.cs | 711 ++++++++++++++++++ src/GameCult.Mesh/README.md | 67 ++ .../CultMeshStreamingTests.cs | 162 ++++ .../GameCult.Mesh.Tests.csproj | 27 + 8 files changed, 1001 insertions(+), 11 deletions(-) create mode 100644 src/GameCult.Mesh/CultMeshStreaming.cs create mode 100644 tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs create mode 100644 tests/GameCult.Mesh.Tests/GameCult.Mesh.Tests.csproj diff --git a/CultLib.sln b/CultLib.sln index 506fe5f7..f6ae25e9 100644 --- a/CultLib.sln +++ b/CultLib.sln @@ -29,6 +29,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Geometry", "src\Ga EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Geometry.Tests", "tests\GameCult.Geometry.Tests\GameCult.Geometry.Tests.csproj", "{C95A9A21-34FA-4D82-BE5B-88FBDA836D31}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "GameCult.Mesh.Tests", "tests\GameCult.Mesh.Tests\GameCult.Mesh.Tests.csproj", "{F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -171,6 +173,18 @@ Global {C95A9A21-34FA-4D82-BE5B-88FBDA836D31}.Release|x64.Build.0 = Release|Any CPU {C95A9A21-34FA-4D82-BE5B-88FBDA836D31}.Release|x86.ActiveCfg = Release|Any CPU {C95A9A21-34FA-4D82-BE5B-88FBDA836D31}.Release|x86.Build.0 = Release|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Debug|Any CPU.Build.0 = Debug|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Debug|x64.ActiveCfg = Debug|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Debug|x64.Build.0 = Debug|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Debug|x86.ActiveCfg = Debug|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Debug|x86.Build.0 = Debug|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|Any CPU.ActiveCfg = Release|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|Any CPU.Build.0 = Release|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x64.ActiveCfg = Release|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x64.Build.0 = Release|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x86.ActiveCfg = Release|Any CPU + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -187,5 +201,6 @@ Global {EB15435B-7147-4BD9-BB9C-B0A14318D3D5} = {39D316AB-ABB5-49A2-8E83-CAF4C75B48F4} {94C086DA-3359-424B-9688-29A198806481} = {E733CBFF-42AE-4B6A-B5B9-0071ECFB45F1} {C95A9A21-34FA-4D82-BE5B-88FBDA836D31} = {39D316AB-ABB5-49A2-8E83-CAF4C75B48F4} + {F8A5060A-FB0B-4B8B-AE06-5147BB4CA127} = {39D316AB-ABB5-49A2-8E83-CAF4C75B48F4} EndGlobalSection EndGlobal diff --git a/README.md b/README.md index 821ff9e7..fa0d8cd2 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ and pretending the label makes it furniture. | Procedural geometry state | `GameCult.Geometry` | CultCache-native domain trees, LOD build requests, selected-cut diagnostics, and chunk artifact payloads | Rust, Unity, or remote workers need to share CSG/LOD geometry and graph metadata as typed state | Transport policy, peer discovery, or gameplay authority | | Network transport and database plumbing | `GameCult.Networking` / CultNet | LiteNetLib transport, authentication, schema-v0 wire contracts, shard authority, raw document replication, snapshots, and subscriptions | You need a client/server pipe, login/session flow, schema discovery, or a low-level distributed CultCache lane | Gameplay-facing mesh ergonomics, Verse policy, mod branches, or simulation consensus composition | | Distributed realtime gameplay state | `GameCult.Mesh` / CultMesh | Public mesh entrypoints, Verse discovery, peer exchange, shard replication defaults, authority leases, client prediction, and witness consensus | You want the game to treat clients and servers as one reactive database for persistent state, input state, and simulation facts | A tiny local-only tool, a bare transport client, or a storage format contract | -| Realtime media/frame streams | CultMesh streaming mode | Stream identity, authority, clock metadata, body transport negotiation, frame cursors, and backpressure state | Audio/video/tensor frames need to move between runtimes through shared memory, GPU handles, platform buffers, or CultCache page refs | Durable document mutation, mesh consensus facts, or pretending inline bytes are zero-copy | +| Realtime media/frame streams | `GameCult.Mesh` / CultMesh streaming mode | Stream identity, authority, clock metadata, body transport negotiation, frame cursors, and backpressure state | Audio/video/tensor frames need to move between runtimes through shared memory, GPU handles, platform buffers, or CultCache page refs | Durable document mutation, mesh consensus facts, or pretending inline bytes are zero-copy | Quick rule: diff --git a/packages/cultmesh-ts/README.md b/packages/cultmesh-ts/README.md index cf635466..2157e3d2 100644 --- a/packages/cultmesh-ts/README.md +++ b/packages/cultmesh-ts/README.md @@ -1,9 +1,10 @@ # CultMesh TS -`cultmesh-ts` is the TypeScript surface for local CultMesh nodes: typed -CultCache documents, peer catalogs, authority leases, and realtime stream -contracts for runtimes that need to share media without turning the mesh into a -base64 soup kitchen. +`cultmesh-ts` is the TypeScript projection of CultMesh contracts for local +tools and browser-adjacent runtimes. The C# `GameCult.Mesh` package owns the +native CultMesh stream implementation used by Mimir and Fensalir; this package +mirrors the typed declaration and negotiation shapes without owning the hot +frame body path. ## Streaming Mode @@ -21,7 +22,7 @@ The split is deliberate: - Producers report `unavoidableCopyCount`; zero-copy is a target and a measured property, not a vibe in a lab coat. -The first public API is `CultMeshStreamCatalog`: +The mirrored API shape is `CultMeshStreamCatalog`: ```ts import { CultMesh } from "cultmesh-ts"; @@ -68,8 +69,7 @@ streams.publishFrame({ }); ``` -Current implementation status: this package owns the typed stream declaration, -negotiation, and latest-frame cursor surface. Runtime-specific transports still -need platform adapters: named shared memory on local processes, D3D/Metal/Vulkan -handle import for GPU frames, and CultCache page append/readback for durable -recording or nonlocal fallbacks. +Current implementation status: this package mirrors typed stream declaration, +negotiation, and latest-frame cursor contracts for non-C# clients. The native +C# path owns shared-memory rings and GPU-handle integration points. CultCache +page append/readback remains the durable recording or nonlocal fallback. diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index a5b7ae3a..8cbeec76 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -212,6 +212,14 @@ namespace GameCult.Mesh return new CultMeshAuthorityLeaseCatalog(); } + /// + /// Creates a local stream catalog for zero-copy-oriented audio, video, tensor, and byte frame streams. + /// + public static CultMeshStreamCatalog CreateStreamCatalog() + { + return new CultMeshStreamCatalog(); + } + /// /// Creates a committer for writing quorum simulation facts to a mesh database. /// diff --git a/src/GameCult.Mesh/CultMeshStreaming.cs b/src/GameCult.Mesh/CultMeshStreaming.cs new file mode 100644 index 00000000..ba3dcfd6 --- /dev/null +++ b/src/GameCult.Mesh/CultMeshStreaming.cs @@ -0,0 +1,711 @@ +using System; +using System.Buffers; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using static GameCult.Mesh.CultMeshGuards; + +#pragma warning disable CS1591 + +namespace GameCult.Mesh +{ + public enum CultMeshStreamKind + { + Audio, + Video, + Tensor, + Bytes + } + + public enum CultMeshStreamBodyTransport + { + SharedMemory, + SharedD3D12Texture, + SharedD3D11Texture, + DmaBuf, + IOSurface, + AHardwareBuffer, + CultCachePage, + InlineBytes + } + + public enum CultMeshStreamAccess + { + Read, + Write, + ReadWrite + } + + public enum CultMeshStreamCopyBudget + { + ZeroCopyTarget, + OneCopyFallback, + OpaqueRuntime + } + + public sealed class CultMeshStreamClock + { + public CultMeshStreamClock( + string clockDomainId, + string? sourceId = null, + int sampleRate = 0, + long offsetToVerseTimeNs = 0, + double confidence = 0, + string? evidenceKind = null) + { + ClockDomainId = RequireNonEmpty(clockDomainId, nameof(clockDomainId)); + SourceId = sourceId; + SampleRate = sampleRate; + OffsetToVerseTimeNs = offsetToVerseTimeNs; + Confidence = confidence; + EvidenceKind = evidenceKind; + } + + public string ClockDomainId { get; } + public string? SourceId { get; } + public int SampleRate { get; } + public long OffsetToVerseTimeNs { get; } + public double Confidence { get; } + public string? EvidenceKind { get; } + } + + public sealed class CultMeshAudioStreamFormat + { + public CultMeshAudioStreamFormat(int sampleRate, int channels, string sampleFormat, int framesPerPacket = 0) + { + if (sampleRate <= 0) throw new ArgumentOutOfRangeException(nameof(sampleRate)); + if (channels <= 0) throw new ArgumentOutOfRangeException(nameof(channels)); + SampleRate = sampleRate; + Channels = channels; + SampleFormat = RequireNonEmpty(sampleFormat, nameof(sampleFormat)); + FramesPerPacket = framesPerPacket; + } + + public int SampleRate { get; } + public int Channels { get; } + public string SampleFormat { get; } + public int FramesPerPacket { get; } + } + + public sealed class CultMeshVideoStreamFormat + { + public CultMeshVideoStreamFormat(int width, int height, string pixelFormat, double framesPerSecond = 0, int planeCount = 1) + { + if (width <= 0) throw new ArgumentOutOfRangeException(nameof(width)); + if (height <= 0) throw new ArgumentOutOfRangeException(nameof(height)); + if (planeCount <= 0) throw new ArgumentOutOfRangeException(nameof(planeCount)); + Width = width; + Height = height; + PixelFormat = RequireNonEmpty(pixelFormat, nameof(pixelFormat)); + FramesPerSecond = framesPerSecond; + PlaneCount = planeCount; + } + + public int Width { get; } + public int Height { get; } + public string PixelFormat { get; } + public double FramesPerSecond { get; } + public int PlaneCount { get; } + } + + public sealed class CultMeshStreamDescriptor + { + public CultMeshStreamDescriptor( + string streamId, + string verseId, + string ownerPeerId, + CultMeshStreamKind kind, + CultMeshStreamClock clock, + IReadOnlyList preferredTransports, + string? label = null, + CultMeshAudioStreamFormat? audio = null, + CultMeshVideoStreamFormat? video = null, + CultMeshStreamAccess requiredAccess = CultMeshStreamAccess.Read, + int maxInFlightFrames = 0, + string? metadataSchemaId = null) + { + StreamId = RequireNonEmpty(streamId, nameof(streamId)); + VerseId = RequireNonEmpty(verseId, nameof(verseId)); + OwnerPeerId = RequireNonEmpty(ownerPeerId, nameof(ownerPeerId)); + Kind = kind; + Clock = clock ?? throw new ArgumentNullException(nameof(clock)); + PreferredTransports = preferredTransports?.ToArray() ?? throw new ArgumentNullException(nameof(preferredTransports)); + if (PreferredTransports.Count == 0) + { + throw new ArgumentException("At least one body transport must be preferred.", nameof(preferredTransports)); + } + + Label = label; + Audio = audio; + Video = video; + RequiredAccess = requiredAccess; + MaxInFlightFrames = maxInFlightFrames; + MetadataSchemaId = metadataSchemaId; + } + + public string StreamId { get; } + public string VerseId { get; } + public string OwnerPeerId { get; } + public CultMeshStreamKind Kind { get; } + public string? Label { get; } + public CultMeshStreamClock Clock { get; } + public CultMeshAudioStreamFormat? Audio { get; } + public CultMeshVideoStreamFormat? Video { get; } + public IReadOnlyList PreferredTransports { get; } + public CultMeshStreamAccess RequiredAccess { get; } + public int MaxInFlightFrames { get; } + public string? MetadataSchemaId { get; } + } + + public sealed class CultMeshStreamConsumerProfile + { + public CultMeshStreamConsumerProfile( + string peerId, + string verseId, + IReadOnlyList supportedTransports, + IReadOnlyList? acceptedKinds = null, + bool canImportGpuHandles = false, + bool canMapSharedMemory = false, + int maxInFlightFrames = 0) + { + PeerId = RequireNonEmpty(peerId, nameof(peerId)); + VerseId = RequireNonEmpty(verseId, nameof(verseId)); + SupportedTransports = supportedTransports?.ToArray() ?? throw new ArgumentNullException(nameof(supportedTransports)); + AcceptedKinds = acceptedKinds?.ToArray() ?? Array.Empty(); + CanImportGpuHandles = canImportGpuHandles; + CanMapSharedMemory = canMapSharedMemory; + MaxInFlightFrames = maxInFlightFrames; + } + + public string PeerId { get; } + public string VerseId { get; } + public IReadOnlyList SupportedTransports { get; } + public IReadOnlyList AcceptedKinds { get; } + public bool CanImportGpuHandles { get; } + public bool CanMapSharedMemory { get; } + public int MaxInFlightFrames { get; } + } + + public sealed class CultMeshStreamNegotiation + { + public CultMeshStreamNegotiation( + string streamId, + string producerPeerId, + string consumerPeerId, + CultMeshStreamBodyTransport transport, + CultMeshStreamAccess access, + int maxInFlightFrames, + CultMeshStreamCopyBudget copyBudget) + { + StreamId = streamId; + ProducerPeerId = producerPeerId; + ConsumerPeerId = consumerPeerId; + Transport = transport; + Access = access; + MaxInFlightFrames = maxInFlightFrames; + CopyBudget = copyBudget; + } + + public string StreamId { get; } + public string ProducerPeerId { get; } + public string ConsumerPeerId { get; } + public CultMeshStreamBodyTransport Transport { get; } + public CultMeshStreamAccess Access { get; } + public int MaxInFlightFrames { get; } + public CultMeshStreamCopyBudget CopyBudget { get; } + } + + public sealed class CultMeshStreamFrameHandle + { + public CultMeshStreamFrameHandle( + string streamId, + ulong sequence, + long timestampNs, + CultMeshStreamBodyTransport transport, + long durationNs = 0, + int byteLength = 0, + string? nativeHandle = null, + string? resourceKey = null, + string? pageRef = null, + string? fenceHandle = null, + ulong fenceValue = 0, + int unavoidableCopyCount = 0, + IReadOnlyDictionary? metadata = null) + { + StreamId = RequireNonEmpty(streamId, nameof(streamId)); + Sequence = sequence; + TimestampNs = timestampNs; + DurationNs = durationNs; + Transport = transport; + ByteLength = byteLength; + NativeHandle = nativeHandle; + ResourceKey = resourceKey; + PageRef = pageRef; + FenceHandle = fenceHandle; + FenceValue = fenceValue; + UnavoidableCopyCount = unavoidableCopyCount; + Metadata = metadata ?? new Dictionary(); + } + + public string StreamId { get; } + public ulong Sequence { get; } + public long TimestampNs { get; } + public long DurationNs { get; } + public CultMeshStreamBodyTransport Transport { get; } + public int ByteLength { get; } + public string? NativeHandle { get; } + public string? ResourceKey { get; } + public string? PageRef { get; } + public string? FenceHandle { get; } + public ulong FenceValue { get; } + public int UnavoidableCopyCount { get; } + public IReadOnlyDictionary Metadata { get; } + } + + public readonly struct CultMeshFrameWriteLease + { + internal CultMeshFrameWriteLease(int slotIndex, ulong sequence, Memory memory) + { + SlotIndex = slotIndex; + Sequence = sequence; + Memory = memory; + } + + public int SlotIndex { get; } + public ulong Sequence { get; } + public Memory Memory { get; } + public Span Span => Memory.Span; + } + + public readonly struct CultMeshFrameReadLease : IDisposable + { + private readonly CultMeshSharedMemoryFrameRing? _owner; + + internal CultMeshFrameReadLease( + CultMeshSharedMemoryFrameRing owner, + int slotIndex, + CultMeshStreamFrameHandle handle, + ReadOnlyMemory memory) + { + _owner = owner; + SlotIndex = slotIndex; + Handle = handle; + Memory = memory; + } + + public int SlotIndex { get; } + public CultMeshStreamFrameHandle Handle { get; } + public ReadOnlyMemory Memory { get; } + public ReadOnlySpan Span => Memory.Span; + + public void Dispose() + { + _owner?.ReleaseRead(SlotIndex); + } + } + + public sealed class CultMeshSharedMemoryFrameRingStats + { + public CultMeshSharedMemoryFrameRingStats( + string streamId, + int slotCount, + int slotByteLength, + ulong publishedFrames, + ulong droppedFrames, + ulong blockedWrites, + ulong latestSequence, + ulong unavoidableCopyCount) + { + StreamId = streamId; + SlotCount = slotCount; + SlotByteLength = slotByteLength; + PublishedFrames = publishedFrames; + DroppedFrames = droppedFrames; + BlockedWrites = blockedWrites; + LatestSequence = latestSequence; + UnavoidableCopyCount = unavoidableCopyCount; + } + + public string StreamId { get; } + public int SlotCount { get; } + public int SlotByteLength { get; } + public ulong PublishedFrames { get; } + public ulong DroppedFrames { get; } + public ulong BlockedWrites { get; } + public ulong LatestSequence { get; } + public ulong UnavoidableCopyCount { get; } + } + + public sealed class CultMeshSharedMemoryFrameRing : IDisposable + { + private readonly byte[] _storage; + private readonly int[] _readerCounts; + private readonly CultMeshStreamFrameHandle?[] _handles; + private int _writeCursor; + private int _latestSlot = -1; + private int _disposed; + private ulong _nextSequence; + private ulong _publishedFrames; + private ulong _droppedFrames; + private ulong _blockedWrites; + private ulong _unavoidableCopyCount; + + public CultMeshSharedMemoryFrameRing(string streamId, int slotCount, int slotByteLength) + { + StreamId = RequireNonEmpty(streamId, nameof(streamId)); + if (slotCount <= 0) throw new ArgumentOutOfRangeException(nameof(slotCount)); + if (slotByteLength <= 0) throw new ArgumentOutOfRangeException(nameof(slotByteLength)); + SlotByteLength = slotByteLength; + _storage = ArrayPool.Shared.Rent(checked(slotCount * slotByteLength)); + _readerCounts = new int[slotCount]; + _handles = new CultMeshStreamFrameHandle?[slotCount]; + } + + public string StreamId { get; } + public int SlotCount => _handles.Length; + public int SlotByteLength { get; } + + public bool TryAcquireWriteSlot(out CultMeshFrameWriteLease lease) + { + ThrowIfDisposed(); + + for (var attempt = 0; attempt < SlotCount; attempt++) + { + var slotIndex = (_writeCursor + attempt) % SlotCount; + if (Volatile.Read(ref _readerCounts[slotIndex]) != 0) + { + continue; + } + + if (_handles[slotIndex] != null) + { + _droppedFrames++; + } + + _writeCursor = (slotIndex + 1) % SlotCount; + lease = new CultMeshFrameWriteLease( + slotIndex, + _nextSequence, + new Memory(_storage, slotIndex * SlotByteLength, SlotByteLength)); + return true; + } + + _blockedWrites++; + lease = default; + return false; + } + + public CultMeshStreamFrameHandle CommitWriteSlot( + CultMeshFrameWriteLease lease, + long timestampNs, + int byteLength, + long durationNs = 0, + int unavoidableCopyCount = 0, + IReadOnlyDictionary? metadata = null) + { + ThrowIfDisposed(); + RequireSlot(lease.SlotIndex); + if (byteLength < 0 || byteLength > SlotByteLength) + { + throw new ArgumentOutOfRangeException(nameof(byteLength)); + } + + _unavoidableCopyCount += (ulong)Math.Max(0, unavoidableCopyCount); + var handle = new CultMeshStreamFrameHandle( + StreamId, + lease.Sequence, + timestampNs, + CultMeshStreamBodyTransport.SharedMemory, + durationNs, + byteLength, + resourceKey: $"{StreamId}:slot:{lease.SlotIndex}", + unavoidableCopyCount: unavoidableCopyCount, + metadata: metadata); + _handles[lease.SlotIndex] = handle; + _latestSlot = lease.SlotIndex; + _nextSequence = lease.Sequence + 1; + _publishedFrames++; + return handle; + } + + public bool TryPublishCopy( + ReadOnlySpan bytes, + long timestampNs, + long durationNs, + out CultMeshStreamFrameHandle handle) + { + ThrowIfDisposed(); + if (bytes.Length > SlotByteLength) + { + throw new ArgumentOutOfRangeException(nameof(bytes)); + } + + if (!TryAcquireWriteSlot(out var lease)) + { + handle = default!; + return false; + } + + bytes.CopyTo(lease.Span); + handle = CommitWriteSlot(lease, timestampNs, bytes.Length, durationNs, unavoidableCopyCount: 1); + return true; + } + + public bool TryAcquireLatestRead(out CultMeshFrameReadLease lease) + { + ThrowIfDisposed(); + var slotIndex = Volatile.Read(ref _latestSlot); + if (slotIndex < 0) + { + lease = default; + return false; + } + + var handle = _handles[slotIndex]; + if (handle == null) + { + lease = default; + return false; + } + + Interlocked.Increment(ref _readerCounts[slotIndex]); + lease = new CultMeshFrameReadLease( + this, + slotIndex, + handle, + new ReadOnlyMemory(_storage, slotIndex * SlotByteLength, handle.ByteLength)); + return true; + } + + public CultMeshSharedMemoryFrameRingStats Stats() + { + ThrowIfDisposed(); + var latest = _latestSlot < 0 ? 0 : _handles[_latestSlot]?.Sequence ?? 0; + return new CultMeshSharedMemoryFrameRingStats( + StreamId, + SlotCount, + SlotByteLength, + _publishedFrames, + _droppedFrames, + _blockedWrites, + latest, + _unavoidableCopyCount); + } + + internal void ReleaseRead(int slotIndex) + { + if (Volatile.Read(ref _disposed) != 0) + { + return; + } + + RequireSlot(slotIndex); + Interlocked.Decrement(ref _readerCounts[slotIndex]); + } + + public void Dispose() + { + if (Interlocked.Exchange(ref _disposed, 1) == 0) + { + ArrayPool.Shared.Return(_storage); + } + } + + private void ThrowIfDisposed() + { + if (Volatile.Read(ref _disposed) != 0) + { + throw new ObjectDisposedException(nameof(CultMeshSharedMemoryFrameRing)); + } + } + + private void RequireSlot(int slotIndex) + { + if (slotIndex < 0 || slotIndex >= SlotCount) + { + throw new ArgumentOutOfRangeException(nameof(slotIndex)); + } + } + } + + public sealed class CultMeshStreamCatalog + { + private readonly Dictionary _streams = new(StringComparer.Ordinal); + private readonly Dictionary _latestFrames = new(StringComparer.Ordinal); + private readonly Dictionary _rings = new(StringComparer.Ordinal); + private readonly object _gate = new(); + + public IReadOnlyList Streams + { + get + { + lock (_gate) + { + return _streams.Values.OrderBy(stream => stream.StreamId, StringComparer.Ordinal).ToArray(); + } + } + } + + public void Declare(CultMeshStreamDescriptor stream) + { + if (stream == null) throw new ArgumentNullException(nameof(stream)); + lock (_gate) + { + _streams[stream.StreamId] = stream; + } + } + + public CultMeshStreamDescriptor? Get(string streamId) + { + lock (_gate) + { + return _streams.TryGetValue(RequireNonEmpty(streamId, nameof(streamId)), out var stream) + ? stream + : null; + } + } + + public IReadOnlyList Find(string verseId, CultMeshStreamKind? kind = null) + { + verseId = RequireNonEmpty(verseId, nameof(verseId)); + lock (_gate) + { + return _streams.Values + .Where(stream => string.Equals(stream.VerseId, verseId, StringComparison.Ordinal) && + (!kind.HasValue || stream.Kind == kind.Value)) + .OrderBy(stream => stream.StreamId, StringComparer.Ordinal) + .ToArray(); + } + } + + public CultMeshStreamNegotiation Negotiate(string streamId, CultMeshStreamConsumerProfile consumer) + { + if (consumer == null) throw new ArgumentNullException(nameof(consumer)); + CultMeshStreamDescriptor stream; + lock (_gate) + { + stream = _streams.TryGetValue(RequireNonEmpty(streamId, nameof(streamId)), out var existing) + ? existing + : throw new InvalidOperationException($"Unknown CultMesh stream '{streamId}'."); + } + if (!string.Equals(stream.VerseId, consumer.VerseId, StringComparison.Ordinal)) + { + throw new InvalidOperationException("Stream and consumer must belong to the same Verse."); + } + + if (consumer.AcceptedKinds.Count > 0 && !consumer.AcceptedKinds.Contains(stream.Kind)) + { + throw new InvalidOperationException($"Consumer does not accept {stream.Kind} streams."); + } + + var transport = FirstCompatibleTransport(stream, consumer); + if (!transport.HasValue) + { + throw new InvalidOperationException("Stream and consumer have no compatible body transport."); + } + + var producerMax = stream.MaxInFlightFrames <= 0 ? int.MaxValue : stream.MaxInFlightFrames; + var consumerMax = consumer.MaxInFlightFrames <= 0 ? int.MaxValue : consumer.MaxInFlightFrames; + return new CultMeshStreamNegotiation( + stream.StreamId, + stream.OwnerPeerId, + consumer.PeerId, + transport.Value, + stream.RequiredAccess, + Math.Min(producerMax, consumerMax), + CopyBudgetFor(transport.Value)); + } + + public void PublishFrame(CultMeshStreamFrameHandle handle) + { + if (handle == null) throw new ArgumentNullException(nameof(handle)); + lock (_gate) + { + if (!_streams.ContainsKey(handle.StreamId)) + { + throw new InvalidOperationException($"Unknown CultMesh stream '{handle.StreamId}'."); + } + + _latestFrames[handle.StreamId] = handle; + } + } + + public CultMeshStreamFrameHandle? LatestFrame(string streamId) + { + lock (_gate) + { + return _latestFrames.TryGetValue(RequireNonEmpty(streamId, nameof(streamId)), out var handle) + ? handle + : null; + } + } + + public CultMeshSharedMemoryFrameRing CreateSharedMemoryRing(string streamId, int slotCount, int slotByteLength) + { + lock (_gate) + { + if (!_streams.ContainsKey(RequireNonEmpty(streamId, nameof(streamId)))) + { + throw new InvalidOperationException($"Unknown CultMesh stream '{streamId}'."); + } + + var ring = new CultMeshSharedMemoryFrameRing(streamId, slotCount, slotByteLength); + _rings[streamId] = ring; + return ring; + } + } + + public CultMeshSharedMemoryFrameRing? Ring(string streamId) + { + lock (_gate) + { + return _rings.TryGetValue(RequireNonEmpty(streamId, nameof(streamId)), out var ring) + ? ring + : null; + } + } + + private static CultMeshStreamCopyBudget CopyBudgetFor(CultMeshStreamBodyTransport transport) + { + return transport switch + { + CultMeshStreamBodyTransport.SharedMemory => CultMeshStreamCopyBudget.ZeroCopyTarget, + CultMeshStreamBodyTransport.SharedD3D12Texture => CultMeshStreamCopyBudget.ZeroCopyTarget, + CultMeshStreamBodyTransport.SharedD3D11Texture => CultMeshStreamCopyBudget.ZeroCopyTarget, + CultMeshStreamBodyTransport.DmaBuf => CultMeshStreamCopyBudget.ZeroCopyTarget, + CultMeshStreamBodyTransport.IOSurface => CultMeshStreamCopyBudget.ZeroCopyTarget, + CultMeshStreamBodyTransport.AHardwareBuffer => CultMeshStreamCopyBudget.ZeroCopyTarget, + CultMeshStreamBodyTransport.CultCachePage => CultMeshStreamCopyBudget.OneCopyFallback, + _ => CultMeshStreamCopyBudget.OpaqueRuntime + }; + } + + private static CultMeshStreamBodyTransport? FirstCompatibleTransport( + CultMeshStreamDescriptor stream, + CultMeshStreamConsumerProfile consumer) + { + foreach (var transport in stream.PreferredTransports) + { + if (consumer.SupportedTransports.Contains(transport)) + { + return transport; + } + } + + return null; + } + } + + internal static partial class CultMeshGuards + { + internal static string RequireNonEmpty(string value, string name) + { + if (string.IsNullOrWhiteSpace(value)) + { + throw new ArgumentException("Value must be non-empty.", name); + } + + return value; + } + } +} diff --git a/src/GameCult.Mesh/README.md b/src/GameCult.Mesh/README.md index 22170b89..c9b13f63 100644 --- a/src/GameCult.Mesh/README.md +++ b/src/GameCult.Mesh/README.md @@ -203,6 +203,73 @@ to be enough. The replicator can answer that by fetching a shard-bounded snapshot, replacing the local shard view, and advancing its cursor to the snapshot's represented log sequence. +## Streaming Mode + +CultMesh streaming mode is the low-latency body path for Mimir, Fensalir, Eve, +and other realtime runtimes. CultCache and CultNet own typed state, discovery, +stream descriptors, negotiations, and frame handles. The frame bodies stay in +runtime-native storage: shared D3D12 textures, shared memory rings, platform GPU +handles, DMA buffers, or paged CultCache fallbacks when a zero-copy transport is +not available. + +```csharp +using GameCult.Mesh; + +var streams = CultMesh.CreateStreamCatalog(); + +streams.Declare(new CultMeshStreamDescriptor( + "mimir:kiyo-pro:rgba", + "mimir-live", + "starfire", + CultMeshStreamKind.Video, + new CultMeshStreamClock("mimir:clock", "kiyo-pro", sampleRate: 90_000, confidence: 0.92), + new[] + { + CultMeshStreamBodyTransport.SharedD3D12Texture, + CultMeshStreamBodyTransport.SharedMemory, + CultMeshStreamBodyTransport.CultCachePage + }, + video: new CultMeshVideoStreamFormat(1920, 1080, "rgba8", framesPerSecond: 60), + maxInFlightFrames: 4)); + +var fensalir = new CultMeshStreamConsumerProfile( + "fensalir", + "mimir-live", + new[] { CultMeshStreamBodyTransport.SharedD3D12Texture }, + acceptedKinds: new[] { CultMeshStreamKind.Video }, + canImportGpuHandles: true); + +var negotiation = streams.Negotiate("mimir:kiyo-pro:rgba", fensalir); +``` + +For CPU-visible data, `CultMeshSharedMemoryFrameRing` preallocates fixed slots +and hands producers writable leases. Consumers hold read leases so the producer +does not overwrite a visible frame. `TryPublishCopy` exists for unavoidable +fallbacks and increments copy telemetry; the normal hot path writes directly +into a lease and commits a `CultMeshStreamFrameHandle`. + +```csharp +streams.Declare(new CultMeshStreamDescriptor( + "mimir:leap:depth", + "mimir-live", + "starfire", + CultMeshStreamKind.Tensor, + new CultMeshStreamClock("mimir:clock", "leap"), + new[] { CultMeshStreamBodyTransport.SharedMemory })); + +using var ring = streams.CreateSharedMemoryRing("mimir:leap:depth", slotCount: 4, slotByteLength: 640 * 480 * 2); + +if (ring.TryAcquireWriteSlot(out var write)) +{ + FillDepthFrame(write.Span); + var frame = ring.CommitWriteSlot(write, timestampNs: GetVerseTimeNs(), byteLength: 640 * 480 * 2); + streams.PublishFrame(frame); +} +``` + +The invariant is simple: typed mesh state describes and synchronizes streams; +hot frame bodies remain in the cheapest storage both endpoints can share. + ## Documentation - [Public API](docs/public-api.md) diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs new file mode 100644 index 00000000..f7f4abe4 --- /dev/null +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -0,0 +1,162 @@ +using System; +using FluentAssertions; +using NUnit.Framework; + +namespace GameCult.Mesh.Tests; + +public sealed class CultMeshStreamingTests +{ + [Test] + public void NegotiatesGpuTextureStreamsWithoutForcingCopies() + { + var catalog = CultMesh.CreateStreamCatalog(); + var stream = new CultMeshStreamDescriptor( + "mimir:kiyo-pro:rgba", + "mimir-live", + "starfire", + CultMeshStreamKind.Video, + new CultMeshStreamClock("mimir:clock", "kiyo-pro", sampleRate: 90_000, confidence: 0.92d), + new[] + { + CultMeshStreamBodyTransport.SharedD3D12Texture, + CultMeshStreamBodyTransport.SharedMemory, + CultMeshStreamBodyTransport.CultCachePage + }, + video: new CultMeshVideoStreamFormat(1920, 1080, "rgba8", framesPerSecond: 60), + maxInFlightFrames: 4); + + catalog.Declare(stream); + + var negotiation = catalog.Negotiate( + stream.StreamId, + new CultMeshStreamConsumerProfile( + "fensalir", + "mimir-live", + new[] { CultMeshStreamBodyTransport.SharedD3D12Texture, CultMeshStreamBodyTransport.CultCachePage }, + acceptedKinds: new[] { CultMeshStreamKind.Video }, + canImportGpuHandles: true, + maxInFlightFrames: 2)); + + negotiation.Transport.Should().Be(CultMeshStreamBodyTransport.SharedD3D12Texture); + negotiation.CopyBudget.Should().Be(CultMeshStreamCopyBudget.ZeroCopyTarget); + negotiation.MaxInFlightFrames.Should().Be(2); + + var handle = new CultMeshStreamFrameHandle( + stream.StreamId, + sequence: 42, + timestampNs: 123_456_789, + CultMeshStreamBodyTransport.SharedD3D12Texture, + nativeHandle: "shared-handle:0xfeedbeef", + fenceHandle: "fence:0x1234", + fenceValue: 7); + + catalog.PublishFrame(handle); + + catalog.LatestFrame(stream.StreamId).Should().BeSameAs(handle); + } + + [Test] + public void NegotiationRejectsStreamsWithoutACommonTransport() + { + var catalog = CultMesh.CreateStreamCatalog(); + catalog.Declare(new CultMeshStreamDescriptor( + "mimir:kiyo-pro:gpu-only", + "mimir-live", + "starfire", + CultMeshStreamKind.Video, + new CultMeshStreamClock("mimir:clock", "kiyo-pro"), + new[] { CultMeshStreamBodyTransport.SharedD3D12Texture })); + + var consumer = new CultMeshStreamConsumerProfile( + "cpu-recorder", + "mimir-live", + new[] { CultMeshStreamBodyTransport.SharedMemory }, + acceptedKinds: new[] { CultMeshStreamKind.Video }); + + Action act = () => catalog.Negotiate("mimir:kiyo-pro:gpu-only", consumer); + + act.Should().Throw() + .WithMessage("Stream and consumer have no compatible body transport."); + } + + [Test] + public void SharedMemoryRingPublishesWritableSlotsWithoutInternalCopies() + { + var catalog = CatalogWithByteStream(); + using var ring = catalog.CreateSharedMemoryRing("mimir:leap:depth", slotCount: 2, slotByteLength: 16); + + ring.TryAcquireWriteSlot(out var write).Should().BeTrue(); + ReadOnlySpan seed = stackalloc byte[] { 1, 2, 3, 4 }; + seed.CopyTo(write.Span); + + var handle = ring.CommitWriteSlot(write, timestampNs: 99, byteLength: 4); + catalog.PublishFrame(handle); + + ring.TryAcquireLatestRead(out var read).Should().BeTrue(); + using (read) + { + read.Handle.Sequence.Should().Be(0); + read.Handle.UnavoidableCopyCount.Should().Be(0); + read.Span.ToArray().Should().Equal(1, 2, 3, 4); + } + + var stats = ring.Stats(); + stats.PublishedFrames.Should().Be(1); + stats.UnavoidableCopyCount.Should().Be(0); + catalog.LatestFrame("mimir:leap:depth")!.ResourceKey.Should().Be("mimir:leap:depth:slot:0"); + } + + [Test] + public void SharedMemoryRingDoesNotOverwriteSlotsHeldByReaders() + { + var catalog = CatalogWithByteStream(); + using var ring = catalog.CreateSharedMemoryRing("mimir:leap:depth", slotCount: 1, slotByteLength: 8); + + ring.TryAcquireWriteSlot(out var firstWrite).Should().BeTrue(); + firstWrite.Span[0] = 11; + ring.CommitWriteSlot(firstWrite, timestampNs: 1, byteLength: 1); + + ring.TryAcquireLatestRead(out var read).Should().BeTrue(); + + ring.TryAcquireWriteSlot(out _).Should().BeFalse(); + ring.Stats().BlockedWrites.Should().Be(1); + + read.Dispose(); + + ring.TryAcquireWriteSlot(out var secondWrite).Should().BeTrue(); + secondWrite.Span[0] = 12; + ring.CommitWriteSlot(secondWrite, timestampNs: 2, byteLength: 1); + + var stats = ring.Stats(); + stats.PublishedFrames.Should().Be(2); + stats.DroppedFrames.Should().Be(1); + stats.LatestSequence.Should().Be(1); + } + + [Test] + public void CopyPublishMarksFallbackCopiesExplicitly() + { + var catalog = CatalogWithByteStream(); + using var ring = catalog.CreateSharedMemoryRing("mimir:leap:depth", slotCount: 2, slotByteLength: 8); + + ring.TryPublishCopy(stackalloc byte[] { 5, 6, 7 }, timestampNs: 10, durationNs: 2, out var handle) + .Should() + .BeTrue(); + + handle.UnavoidableCopyCount.Should().Be(1); + ring.Stats().UnavoidableCopyCount.Should().Be(1); + } + + private static CultMeshStreamCatalog CatalogWithByteStream() + { + var catalog = CultMesh.CreateStreamCatalog(); + catalog.Declare(new CultMeshStreamDescriptor( + "mimir:leap:depth", + "mimir-live", + "starfire", + CultMeshStreamKind.Tensor, + new CultMeshStreamClock("mimir:clock", "leap", confidence: 0.8d), + new[] { CultMeshStreamBodyTransport.SharedMemory, CultMeshStreamBodyTransport.CultCachePage })); + return catalog; + } +} diff --git a/tests/GameCult.Mesh.Tests/GameCult.Mesh.Tests.csproj b/tests/GameCult.Mesh.Tests/GameCult.Mesh.Tests.csproj new file mode 100644 index 00000000..14e99d63 --- /dev/null +++ b/tests/GameCult.Mesh.Tests/GameCult.Mesh.Tests.csproj @@ -0,0 +1,27 @@ + + + + net10.0 + false + GameCult.Mesh.Tests + + + + + + + + + runtime; build; native; contentfiles; analyzers; buildtransitive + all + + + + + + + all + + + +