Publish cross-process CultMesh frame bodies
This commit is contained in:
Meta Vens
parent
8472330f90
commit
dbe6a3f9de
3 changed files with 298 additions and 2 deletions
|
|
@ -429,6 +429,8 @@ namespace GameCult.Mesh
|
|||
CultMeshBodyDescriptorValidator.Validate(descriptor, request);
|
||||
if (string.IsNullOrWhiteSpace(descriptor.CapabilityToken))
|
||||
throw new InvalidOperationException("Shared-memory CultMesh body descriptor has no transport capability token.");
|
||||
if (CultMeshFrameBodyPublisher.IsFrameCapability(descriptor.CapabilityToken))
|
||||
return CultMeshFrameBodyPublisher.OpenReadOnly(descriptor);
|
||||
MemoryMappedFile mapping;
|
||||
try
|
||||
{
|
||||
|
|
@ -461,13 +463,20 @@ namespace GameCult.Mesh
|
|||
{
|
||||
}
|
||||
|
||||
internal CultMeshMappedBodyLease(CultMeshBodyDescriptor descriptor, MemoryMappedFile mapping)
|
||||
internal CultMeshMappedBodyLease(
|
||||
CultMeshBodyDescriptor descriptor,
|
||||
MemoryMappedFile mapping,
|
||||
long offset = 0,
|
||||
Action? release = null)
|
||||
{
|
||||
Descriptor = descriptor;
|
||||
_mapping = mapping;
|
||||
_view = _mapping.CreateViewAccessor(0, descriptor.ByteSize, MemoryMappedFileAccess.Read);
|
||||
_release = release;
|
||||
_view = _mapping.CreateViewAccessor(offset, descriptor.ByteSize, MemoryMappedFileAccess.Read);
|
||||
}
|
||||
|
||||
private readonly Action? _release;
|
||||
|
||||
public CultMeshBodyDescriptor Descriptor { get; }
|
||||
public CultMeshBodyTransportKind TransportKind => Descriptor.TransportKind;
|
||||
public byte ReadByte(long offset) { ValidateRange(offset, 1); return _view.ReadByte(offset); }
|
||||
|
|
@ -488,6 +497,7 @@ namespace GameCult.Mesh
|
|||
_disposed = true;
|
||||
_view.Dispose();
|
||||
_mapping.Dispose();
|
||||
_release?.Invoke();
|
||||
}
|
||||
|
||||
private void ValidateRange(long offset, long length)
|
||||
|
|
|
|||
|
|
@ -1,11 +1,219 @@
|
|||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.IO.MemoryMappedFiles;
|
||||
using System.Security.Cryptography;
|
||||
using System.Text;
|
||||
using System.Threading;
|
||||
|
||||
#nullable enable
|
||||
|
||||
namespace GameCult.Mesh
|
||||
{
|
||||
// Owner: this publisher owns mapping lifetime, logical generation, and slot commit.
|
||||
// Consumers own only read leases; mappings, capabilities, and slot selection remain CultMesh state.
|
||||
public sealed class CultMeshFrameBodyPublisher : IDisposable
|
||||
{
|
||||
private const string CapabilityPrefix = "frame-v1-";
|
||||
private const int HeaderSize = 128;
|
||||
private const int SequenceOffset = 8;
|
||||
private const int ReaderCountOffset = 32;
|
||||
private const int ContractHashOffset = 64;
|
||||
private readonly object _gate = new();
|
||||
private readonly MemoryMappedFile _mapping;
|
||||
private readonly MemoryMappedViewAccessor _control;
|
||||
private readonly Mutex _mutex;
|
||||
private readonly TimeSpan _leaseDuration;
|
||||
private readonly string _token;
|
||||
private int _cursor;
|
||||
private long _nextSequence;
|
||||
private bool _disposed;
|
||||
|
||||
public CultMeshFrameBodyPublisher(
|
||||
string bodyId,
|
||||
string schemaId,
|
||||
int layoutVersion,
|
||||
int capacity,
|
||||
long producerEpoch,
|
||||
int slotByteLength,
|
||||
TimeSpan? leaseDuration = null)
|
||||
{
|
||||
BodyId = Require(bodyId, nameof(bodyId));
|
||||
SchemaId = Require(schemaId, nameof(schemaId));
|
||||
if (layoutVersion < 0) throw new ArgumentOutOfRangeException(nameof(layoutVersion));
|
||||
if (capacity < 0) throw new ArgumentOutOfRangeException(nameof(capacity));
|
||||
if (producerEpoch < 0) throw new ArgumentOutOfRangeException(nameof(producerEpoch));
|
||||
if (slotByteLength <= 0) throw new ArgumentOutOfRangeException(nameof(slotByteLength));
|
||||
LayoutVersion = layoutVersion;
|
||||
Capacity = capacity;
|
||||
ProducerEpoch = producerEpoch;
|
||||
SlotByteLength = slotByteLength;
|
||||
_leaseDuration = leaseDuration ?? TimeSpan.FromSeconds(5);
|
||||
if (_leaseDuration <= TimeSpan.Zero) throw new ArgumentOutOfRangeException(nameof(leaseDuration));
|
||||
_token = CapabilityPrefix + RandomToken();
|
||||
_mapping = MemoryMappedFile.CreateNew(MappingName(_token), checked(HeaderSize + CultMeshFrameRegion.SlotCount * slotByteLength));
|
||||
_control = _mapping.CreateViewAccessor(0, HeaderSize, MemoryMappedFileAccess.ReadWrite);
|
||||
_control.Write(0, slotByteLength);
|
||||
_control.Write(4, CultMeshFrameRegion.SlotCount);
|
||||
var contractHash = Encoding.ASCII.GetBytes(ContractHash(BodyId, SchemaId, LayoutVersion, Capacity, ProducerEpoch, SlotByteLength));
|
||||
_control.WriteArray(ContractHashOffset, contractHash, 0, contractHash.Length);
|
||||
_mutex = new Mutex(false, MutexName(_token));
|
||||
}
|
||||
|
||||
public string BodyId { get; }
|
||||
public string SchemaId { get; }
|
||||
public int LayoutVersion { get; }
|
||||
public int Capacity { get; }
|
||||
public long ProducerEpoch { get; }
|
||||
public int SlotByteLength { get; }
|
||||
|
||||
public bool TryPublish(ReadOnlySpan<byte> body, DateTimeOffset nowUtc, out CultMeshBodyDescriptor descriptor)
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
ThrowIfDisposed();
|
||||
if (body.Length > SlotByteLength) throw new ArgumentOutOfRangeException(nameof(body));
|
||||
var publishedBytes = body.ToArray();
|
||||
CultMeshBodyDescriptor? published = null;
|
||||
WithMutex(() =>
|
||||
{
|
||||
for (var attempt = 0; attempt < CultMeshFrameRegion.SlotCount; attempt++)
|
||||
{
|
||||
var slot = (_cursor + attempt) % CultMeshFrameRegion.SlotCount;
|
||||
if (_control.ReadInt32(ReaderCountOffset + slot * sizeof(int)) != 0) continue;
|
||||
using (var view = _mapping.CreateViewAccessor(SlotOffset(slot, SlotByteLength), SlotByteLength, MemoryMappedFileAccess.Write))
|
||||
view.WriteArray(0, publishedBytes, 0, publishedBytes.Length);
|
||||
_control.Write(SequenceOffset + slot * sizeof(long), _nextSequence);
|
||||
_cursor = (slot + 1) % CultMeshFrameRegion.SlotCount;
|
||||
published = Descriptor(slot, publishedBytes, nowUtc);
|
||||
_nextSequence++;
|
||||
return;
|
||||
}
|
||||
});
|
||||
descriptor = published!;
|
||||
return published != null;
|
||||
}
|
||||
}
|
||||
|
||||
private CultMeshBodyDescriptor Descriptor(int slot, ReadOnlySpan<byte> body, DateTimeOffset nowUtc) => new()
|
||||
{
|
||||
BodyId = BodyId, SchemaId = SchemaId, LayoutVersion = LayoutVersion, ByteSize = body.Length,
|
||||
Capacity = Capacity, ProducerEpoch = ProducerEpoch, Sequence = _nextSequence,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly, Synchronization = CultMeshBodySynchronization.TripleBuffer,
|
||||
LeaseExpiresAtUnixMs = nowUtc.Add(_leaseDuration).ToUnixTimeMilliseconds(),
|
||||
TransportKind = CultMeshBodyTransportKind.SharedMemory,
|
||||
CapabilityToken = _token + "." + slot,
|
||||
SemanticHash = CultMeshBodyDescriptorValidator.ComputeSemanticHash(body)
|
||||
};
|
||||
|
||||
internal static bool IsFrameCapability(string capability) => capability.StartsWith(CapabilityPrefix, StringComparison.Ordinal);
|
||||
|
||||
internal static ICultMeshBodyReadLease OpenReadOnly(CultMeshBodyDescriptor descriptor)
|
||||
{
|
||||
ParseCapability(descriptor.CapabilityToken, out var token, out var slot);
|
||||
MemoryMappedFile mapping;
|
||||
Mutex mutex;
|
||||
try
|
||||
{
|
||||
mapping = MemoryMappedFile.OpenExisting(MappingName(token), MemoryMappedFileRights.ReadWrite);
|
||||
mutex = Mutex.OpenExisting(MutexName(token));
|
||||
}
|
||||
catch (Exception error) when (error is FileNotFoundException || error is WaitHandleCannotBeOpenedException)
|
||||
{
|
||||
throw new InvalidDataException("CultMesh frame capability is unavailable or revoked.", error);
|
||||
}
|
||||
try
|
||||
{
|
||||
int slotByteLength = 0;
|
||||
WithMutex(mutex, () =>
|
||||
{
|
||||
using var control = mapping.CreateViewAccessor(0, HeaderSize, MemoryMappedFileAccess.ReadWrite);
|
||||
slotByteLength = control.ReadInt32(0);
|
||||
if (slot < 0 || slot >= control.ReadInt32(4) || descriptor.ByteSize > slotByteLength)
|
||||
throw new InvalidDataException("CultMesh frame capability does not match its mapping layout.");
|
||||
var contractHash = new byte[64];
|
||||
control.ReadArray(ContractHashOffset, contractHash, 0, contractHash.Length);
|
||||
var expectedContract = ContractHash(
|
||||
descriptor.BodyId, descriptor.SchemaId, descriptor.LayoutVersion,
|
||||
descriptor.Capacity, descriptor.ProducerEpoch, slotByteLength);
|
||||
if (!string.Equals(Encoding.ASCII.GetString(contractHash), expectedContract, StringComparison.Ordinal) ||
|
||||
control.ReadInt64(SequenceOffset + slot * sizeof(long)) != descriptor.Sequence)
|
||||
throw new InvalidOperationException("CultMesh frame generation is stale or belongs to a different body contract.");
|
||||
var offset = ReaderCountOffset + slot * sizeof(int);
|
||||
control.Write(offset, checked(control.ReadInt32(offset) + 1));
|
||||
});
|
||||
return new CultMeshMappedBodyLease(
|
||||
descriptor, mapping, SlotOffset(slot, slotByteLength),
|
||||
() => ReleaseReader(token, slot));
|
||||
}
|
||||
catch
|
||||
{
|
||||
mapping.Dispose();
|
||||
mutex.Dispose();
|
||||
throw;
|
||||
}
|
||||
finally { mutex.Dispose(); }
|
||||
}
|
||||
|
||||
private static void ReleaseReader(string token, int slot)
|
||||
{
|
||||
try
|
||||
{
|
||||
using var mapping = MemoryMappedFile.OpenExisting(MappingName(token), MemoryMappedFileRights.ReadWrite);
|
||||
using var mutex = Mutex.OpenExisting(MutexName(token));
|
||||
WithMutex(mutex, () =>
|
||||
{
|
||||
using var control = mapping.CreateViewAccessor(0, HeaderSize, MemoryMappedFileAccess.ReadWrite);
|
||||
var offset = ReaderCountOffset + slot * sizeof(int);
|
||||
var readers = control.ReadInt32(offset);
|
||||
if (readers > 0) control.Write(offset, readers - 1);
|
||||
});
|
||||
}
|
||||
catch (Exception error) when (error is FileNotFoundException || error is WaitHandleCannotBeOpenedException) { }
|
||||
}
|
||||
|
||||
public void Dispose()
|
||||
{
|
||||
lock (_gate)
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
_control.Dispose();
|
||||
_mapping.Dispose();
|
||||
_mutex.Dispose();
|
||||
}
|
||||
}
|
||||
|
||||
private void WithMutex(Action action) => WithMutex(_mutex, action);
|
||||
private static void WithMutex(Mutex mutex, Action action)
|
||||
{
|
||||
mutex.WaitOne();
|
||||
try { action(); }
|
||||
finally { mutex.ReleaseMutex(); }
|
||||
}
|
||||
private static long SlotOffset(int slot, int slotByteLength) => HeaderSize + (long)slot * slotByteLength;
|
||||
private static string MappingName(string token) => "cultmesh-map-" + token;
|
||||
private static string MutexName(string token) => "cultmesh-lock-" + token;
|
||||
private static void ParseCapability(string capability, out string token, out int slot)
|
||||
{
|
||||
var separator = capability.LastIndexOf('.');
|
||||
if (separator <= CapabilityPrefix.Length || !int.TryParse(capability.Substring(separator + 1), out slot))
|
||||
throw new UnauthorizedAccessException("Invalid CultMesh frame capability.");
|
||||
token = capability.Substring(0, separator);
|
||||
}
|
||||
private static string RandomToken()
|
||||
{
|
||||
var bytes = new byte[24];
|
||||
using (var random = RandomNumberGenerator.Create()) random.GetBytes(bytes);
|
||||
return BitConverter.ToString(bytes).Replace("-", string.Empty).ToLowerInvariant();
|
||||
}
|
||||
private static string ContractHash(string bodyId, string schemaId, int layoutVersion, int capacity, long epoch, int slotByteLength) =>
|
||||
CultMeshBodyDescriptorValidator.ComputeSemanticHash(
|
||||
Encoding.UTF8.GetBytes($"{bodyId}\n{schemaId}\n{layoutVersion}\n{capacity}\n{epoch}\n{slotByteLength}"));
|
||||
private void ThrowIfDisposed() { if (_disposed) throw new ObjectDisposedException(nameof(CultMeshFrameBodyPublisher)); }
|
||||
private static string Require(string value, string name) => string.IsNullOrWhiteSpace(value) ? throw new ArgumentException("Value is required.", name) : value;
|
||||
}
|
||||
|
||||
public sealed class CultMeshFrameGeneration
|
||||
{
|
||||
internal CultMeshFrameGeneration(
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Threading.Tasks;
|
||||
using FluentAssertions;
|
||||
|
|
@ -11,6 +12,69 @@ namespace GameCult.Mesh.Tests;
|
|||
[TestFixture]
|
||||
public sealed class CultMeshFrameBodyTests
|
||||
{
|
||||
[Test]
|
||||
public void CrossProcessPublisherOpensReadOnlyBytesAndProtectsLeasedSlot()
|
||||
{
|
||||
using var publisher = Publisher();
|
||||
publisher.TryPublish(new byte[] { 1, 2, 3 }, DateTimeOffset.UtcNow, out var first).Should().BeTrue();
|
||||
var adapter = new CultMeshSharedMemoryBodyAdapter();
|
||||
using var lease = adapter.OpenReadOnly(first, Request(first));
|
||||
var bytes = new byte[3];
|
||||
lease.CopyTo(0, bytes, 0, bytes.Length).Should().Be(3);
|
||||
bytes.Should().Equal(1, 2, 3);
|
||||
|
||||
publisher.TryPublish(new byte[] { 4 }, DateTimeOffset.UtcNow, out var second).Should().BeTrue();
|
||||
publisher.TryPublish(new byte[] { 5 }, DateTimeOffset.UtcNow, out var third).Should().BeTrue();
|
||||
publisher.TryPublish(new byte[] { 6 }, DateTimeOffset.UtcNow, out var fourth).Should().BeTrue();
|
||||
fourth.CapabilityToken.Should().NotBe(first.CapabilityToken);
|
||||
lease.ReadByte(0).Should().Be(1);
|
||||
}
|
||||
|
||||
[TestCase("wrong", 2, 9)]
|
||||
[TestCase("tests.frame.v2", 1, 9)]
|
||||
[TestCase("tests.frame.v2", 2, 8)]
|
||||
public void CrossProcessOpenRejectsStaleSchemaLayoutOrEpoch(string schema, int layout, long epoch)
|
||||
{
|
||||
using var publisher = Publisher();
|
||||
publisher.TryPublish(new byte[] { 42 }, DateTimeOffset.UtcNow, out var descriptor).Should().BeTrue();
|
||||
var request = Request(descriptor);
|
||||
request.SchemaId = schema;
|
||||
request.LayoutVersion = layout;
|
||||
request.ProducerEpoch = epoch;
|
||||
|
||||
Action open = () => new CultMeshSharedMemoryBodyAdapter().OpenReadOnly(descriptor, request);
|
||||
open.Should().Throw<InvalidOperationException>();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void ReusedSlotRejectsItsStaleGenerationDescriptor()
|
||||
{
|
||||
using var publisher = Publisher();
|
||||
publisher.TryPublish(new byte[] { 1 }, DateTimeOffset.UtcNow, out var stale).Should().BeTrue();
|
||||
publisher.TryPublish(new byte[] { 2 }, DateTimeOffset.UtcNow, out _).Should().BeTrue();
|
||||
publisher.TryPublish(new byte[] { 3 }, DateTimeOffset.UtcNow, out _).Should().BeTrue();
|
||||
publisher.TryPublish(new byte[] { 4 }, DateTimeOffset.UtcNow, out _).Should().BeTrue();
|
||||
|
||||
Action open = () => new CultMeshSharedMemoryBodyAdapter().OpenReadOnly(stale, Request(stale));
|
||||
open.Should().Throw<InvalidOperationException>().WithMessage("*stale*");
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void CleanupRevokesNewOpensAndDisposedLeaseIsInvalid()
|
||||
{
|
||||
var publisher = Publisher();
|
||||
publisher.TryPublish(new byte[] { 42 }, DateTimeOffset.UtcNow, out var descriptor).Should().BeTrue();
|
||||
var adapter = new CultMeshSharedMemoryBodyAdapter();
|
||||
var lease = adapter.OpenReadOnly(descriptor, Request(descriptor));
|
||||
lease.Dispose();
|
||||
Action readDisposed = () => lease.ReadByte(0);
|
||||
readDisposed.Should().Throw<ObjectDisposedException>();
|
||||
|
||||
publisher.Dispose();
|
||||
Action reopen = () => adapter.OpenReadOnly(descriptor, Request(descriptor));
|
||||
reopen.Should().Throw<InvalidDataException>();
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void PublishesOneAtomicSchemaVersionedGenerationPerCommit()
|
||||
{
|
||||
|
|
@ -178,6 +242,20 @@ public sealed class CultMeshFrameBodyTests
|
|||
private static CultMeshFrameRegion Region() =>
|
||||
new("body", "tests.frame.v2", layoutVersion: 2, capacity: 64, producerEpoch: 9, slotByteLength: 64);
|
||||
|
||||
private static CultMeshFrameBodyPublisher Publisher() =>
|
||||
new("body", "tests.frame.v2", layoutVersion: 2, capacity: 64, producerEpoch: 9, slotByteLength: 64);
|
||||
|
||||
private static CultMeshBodyValidationRequest Request(CultMeshBodyDescriptor descriptor) => new()
|
||||
{
|
||||
BodyId = descriptor.BodyId,
|
||||
SchemaId = descriptor.SchemaId,
|
||||
LayoutVersion = descriptor.LayoutVersion,
|
||||
ProducerEpoch = descriptor.ProducerEpoch,
|
||||
Capacity = descriptor.Capacity,
|
||||
Sequence = descriptor.Sequence,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly
|
||||
};
|
||||
|
||||
private static CultMeshBodyValidationRequest Request(long? sequence = null) => new()
|
||||
{
|
||||
BodyId = "body",
|
||||
|
|
|
|||
Loading…
Reference in a new issue