Lower remote Eve state over native QUIC
This commit is contained in:
Meta Vens
parent
6e19396f7f
commit
44e338bd9e
5 changed files with 277 additions and 39 deletions
|
|
@ -3,6 +3,7 @@ using System.Collections.Generic;
|
|||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using GameCult.Caching;
|
||||
using GameCult.Caching.MessagePack;
|
||||
|
|
@ -10,6 +11,7 @@ using GameCult.Eve.PluginFields;
|
|||
using GameCult.Eve.Surface;
|
||||
using GameCult.Eve.UnityScene.Fields;
|
||||
using GameCult.Mesh;
|
||||
using GameCult.Mesh.Quic.Native;
|
||||
using GameCult.Networking;
|
||||
using UnityEngine;
|
||||
using Stopwatch = System.Diagnostics.Stopwatch;
|
||||
|
|
@ -63,6 +65,11 @@ namespace GameCult.Eve.UnityScene
|
|||
private readonly List<ICultMeshBodyReadLease> _assetBodyLeases = new List<ICultMeshBodyReadLease>();
|
||||
private readonly Dictionary<string, int> _renderChannelLayers = new Dictionary<string, int>(StringComparer.Ordinal);
|
||||
private readonly ConcurrentQueue<object> _liveDocuments = new ConcurrentQueue<object>();
|
||||
private readonly object _realtimeFrameGate = new object();
|
||||
private readonly Dictionary<string, CultMeshRealtimeFrame> _latestRealtimeFrames =
|
||||
new Dictionary<string, CultMeshRealtimeFrame>(StringComparer.Ordinal);
|
||||
private readonly HashSet<string> _queuedRealtimeFrameBodies =
|
||||
new HashSet<string>(StringComparer.Ordinal);
|
||||
private readonly Dictionary<string, EveSurfaceDocument> _embeddedSurfaces =
|
||||
new Dictionary<string, EveSurfaceDocument>(StringComparer.Ordinal);
|
||||
private CultMeshNode? _node;
|
||||
|
|
@ -84,6 +91,9 @@ namespace GameCult.Eve.UnityScene
|
|||
private CultMeshBodyPublicationDocument? _latestEntityPublication;
|
||||
private CultMeshMappedFrameBodyCursor? _mappedEntityFrameCursor;
|
||||
private CultMeshBodyDescriptor? _mappedEntityFrameContract;
|
||||
private CancellationTokenSource? _realtimeLifetime;
|
||||
private Task? _realtimePump;
|
||||
private CultMeshRealtimeSession? _realtimeSession;
|
||||
private long _lastQueuedEntityViewEpoch = -1;
|
||||
private long _lastQueuedEntityViewSequence = -1;
|
||||
private long _lastPresentedEntityViewEpoch = -1;
|
||||
|
|
@ -221,6 +231,8 @@ namespace GameCult.Eve.UnityScene
|
|||
{
|
||||
if (document is PendingEntityGeneration entityGeneration)
|
||||
PublishEntityView(entityGeneration.View, entityGeneration.Publication);
|
||||
else if (document is PendingRealtimeEntityFrame realtimeFrame)
|
||||
PublishPendingRealtimeEntityFrame(realtimeFrame.BodyId);
|
||||
else if (document is EveFieldsSplatsDocument fields)
|
||||
FieldsSplatsAvailable?.Invoke(fields);
|
||||
else if (document is PendingSurfaceDocument surface)
|
||||
|
|
@ -398,6 +410,17 @@ namespace GameCult.Eve.UnityScene
|
|||
_mappedEntityFrameCursor?.Dispose();
|
||||
_mappedEntityFrameCursor = null;
|
||||
_mappedEntityFrameContract = null;
|
||||
_realtimeLifetime?.Cancel();
|
||||
_realtimeSession?.Dispose();
|
||||
_realtimeSession = null;
|
||||
if (_realtimePump != null)
|
||||
{
|
||||
try { _realtimePump.GetAwaiter().GetResult(); }
|
||||
catch (OperationCanceledException) { }
|
||||
}
|
||||
_realtimePump = null;
|
||||
_realtimeLifetime?.Dispose();
|
||||
_realtimeLifetime = null;
|
||||
_entityBody?.Dispose();
|
||||
_entityBody = null;
|
||||
_inputCapability?.Dispose();
|
||||
|
|
@ -502,7 +525,11 @@ namespace GameCult.Eve.UnityScene
|
|||
_networkRegistry);
|
||||
_meshClient = new CultMeshClient(new CultMeshClientOptions
|
||||
{
|
||||
RendezvousEndpoints = new[] { _endpoint }
|
||||
RendezvousEndpoints = new[] { _endpoint },
|
||||
RealtimeConnectors = new ICultMeshRealtimeTransportConnector[]
|
||||
{
|
||||
new CultMeshNativeQuicRealtimeTransportConnector()
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -595,35 +622,7 @@ namespace GameCult.Eve.UnityScene
|
|||
EnsureMappedEntityFrameCursor(publication);
|
||||
if (_mappedEntityFrameCursor != null)
|
||||
return;
|
||||
_bodyResolver ??= CreateBodyResolver();
|
||||
var lease = _bodyResolver.ResolveReadOnly(publication, new CultMeshBodyValidationRequest
|
||||
{
|
||||
BodyId = handle.BodyId,
|
||||
SchemaId = document.BodySchemaId,
|
||||
LayoutVersion = document.LayoutVersion,
|
||||
ProducerEpoch = document.ProducerEpoch,
|
||||
Sequence = document.Sequence,
|
||||
Capacity = document.Capacity,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly,
|
||||
NowUtc = now
|
||||
});
|
||||
var handler = EntityViewAvailable;
|
||||
if (handler == null)
|
||||
{
|
||||
lease.Dispose();
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
handler(document, lease);
|
||||
_lastPresentedEntityViewEpoch = document.ProducerEpoch;
|
||||
_lastPresentedEntityViewSequence = document.Sequence;
|
||||
}
|
||||
catch
|
||||
{
|
||||
lease.Dispose();
|
||||
throw;
|
||||
}
|
||||
EnsureRealtimeEntityState();
|
||||
}
|
||||
|
||||
private void EnsureMappedEntityFrameCursor(CultMeshBodyPublicationDocument publication)
|
||||
|
|
@ -651,10 +650,196 @@ namespace GameCult.Eve.UnityScene
|
|||
return;
|
||||
|
||||
_mappedEntityFrameCursor?.Dispose();
|
||||
_mappedEntityFrameCursor = new CultMeshMappedFrameBodyCursor(descriptor);
|
||||
_mappedEntityFrameContract = descriptor;
|
||||
TraceHotState(
|
||||
$"mapped frame cursor body={descriptor.BodyId} epoch={descriptor.ProducerEpoch} sequence={descriptor.Sequence}");
|
||||
try
|
||||
{
|
||||
_mappedEntityFrameCursor = new CultMeshMappedFrameBodyCursor(descriptor);
|
||||
_mappedEntityFrameContract = descriptor;
|
||||
TraceHotState(
|
||||
$"mapped frame cursor body={descriptor.BodyId} epoch={descriptor.ProducerEpoch} sequence={descriptor.Sequence}");
|
||||
}
|
||||
catch (Exception error) when (
|
||||
error is IOException ||
|
||||
error is UnauthorizedAccessException ||
|
||||
error is InvalidOperationException)
|
||||
{
|
||||
_mappedEntityFrameCursor = null;
|
||||
_mappedEntityFrameContract = null;
|
||||
TraceHotState($"mapped frame unavailable; selecting remote realtime plane: {error.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
private void EnsureRealtimeEntityState()
|
||||
{
|
||||
if (_realtimePump != null)
|
||||
return;
|
||||
if (_advertisement == null || string.IsNullOrWhiteSpace(_advertisement.VerseId))
|
||||
throw new InvalidOperationException(
|
||||
$"Eve provider '{_providerId}' does not advertise the Verse identity required for realtime state.");
|
||||
_realtimeLifetime = new CancellationTokenSource();
|
||||
_realtimePump = Task.Run(() => PumpRealtimeEntityStateAsync(
|
||||
_advertisement.VerseId,
|
||||
_realtimeLifetime.Token));
|
||||
}
|
||||
|
||||
private async Task PumpRealtimeEntityStateAsync(string verseId, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
_realtimeSession = await _meshClient!
|
||||
.ConnectRealtimeAsync(verseId, cancellationToken)
|
||||
.ConfigureAwait(false);
|
||||
TraceHotState($"realtime state connected transport={_realtimeSession.TransportId} verse={verseId}");
|
||||
while (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
var frame = await _realtimeSession.ReceiveAsync(cancellationToken).ConfigureAwait(false);
|
||||
EnqueueRealtimeEntityFrame(frame);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
}
|
||||
catch (ObjectDisposedException) when (_disposed)
|
||||
{
|
||||
}
|
||||
catch (Exception error)
|
||||
{
|
||||
_liveDocuments.Enqueue(error);
|
||||
}
|
||||
}
|
||||
|
||||
private void EnqueueRealtimeEntityFrame(CultMeshRealtimeFrame frame)
|
||||
{
|
||||
if (frame == null)
|
||||
throw new ArgumentNullException(nameof(frame));
|
||||
lock (_realtimeFrameGate)
|
||||
{
|
||||
if (_latestRealtimeFrames.TryGetValue(frame.BodyId, out var current) &&
|
||||
(frame.ProducerEpoch < current.ProducerEpoch ||
|
||||
(frame.ProducerEpoch == current.ProducerEpoch && frame.Sequence <= current.Sequence)))
|
||||
return;
|
||||
_latestRealtimeFrames[frame.BodyId] = frame;
|
||||
if (_queuedRealtimeFrameBodies.Add(frame.BodyId))
|
||||
_liveDocuments.Enqueue(new PendingRealtimeEntityFrame(frame.BodyId));
|
||||
}
|
||||
}
|
||||
|
||||
private void PublishPendingRealtimeEntityFrame(string bodyId)
|
||||
{
|
||||
CultMeshRealtimeFrame? frame;
|
||||
lock (_realtimeFrameGate)
|
||||
{
|
||||
_queuedRealtimeFrameBodies.Remove(bodyId);
|
||||
if (!_latestRealtimeFrames.TryGetValue(bodyId, out frame))
|
||||
return;
|
||||
_latestRealtimeFrames.Remove(bodyId);
|
||||
}
|
||||
PublishRealtimeEntityFrame(frame);
|
||||
}
|
||||
|
||||
private void PublishRealtimeEntityFrame(CultMeshRealtimeFrame frame)
|
||||
{
|
||||
var layout = _latestEntityLayout
|
||||
?? throw new InvalidOperationException("Realtime entity state arrived before its Eve SoA layout.");
|
||||
var publication = _latestEntityPublication
|
||||
?? throw new InvalidOperationException("Realtime entity state arrived before its body publication.");
|
||||
if (!string.Equals(frame.BodyId, publication.BodyId, StringComparison.Ordinal) ||
|
||||
!string.Equals(frame.SchemaId, layout.BodySchemaId, StringComparison.Ordinal))
|
||||
throw new UnauthorizedAccessException("Realtime entity state does not match the advertised Eve body contract.");
|
||||
if (frame.Delivery != CultMeshRealtimeDelivery.LatestOnly)
|
||||
throw new InvalidDataException("Playable entity state requires latest-only QUIC delivery.");
|
||||
if (frame.ProducerEpoch < _lastPresentedEntityViewEpoch ||
|
||||
(frame.ProducerEpoch == _lastPresentedEntityViewEpoch &&
|
||||
frame.Sequence <= _lastPresentedEntityViewSequence))
|
||||
return;
|
||||
var representation = (publication.Representations ?? Array.Empty<CultMeshBodyDescriptor>())
|
||||
.FirstOrDefault(value =>
|
||||
string.Equals(value.BodyId, frame.BodyId, StringComparison.Ordinal) &&
|
||||
string.Equals(value.SchemaId, frame.SchemaId, StringComparison.Ordinal))
|
||||
?? throw new InvalidDataException("Realtime entity state has no advertised body representation.");
|
||||
if (frame.Payload.Length != representation.ByteSize)
|
||||
throw new InvalidDataException(
|
||||
$"Realtime entity body size {frame.Payload.Length} does not match advertised size {representation.ByteSize}.");
|
||||
|
||||
var generation = MaterializeEntityGeneration(layout, frame.ProducerEpoch, frame.Sequence);
|
||||
var lease = new RealtimeEntityBodyLease(
|
||||
new CultMeshBodyDescriptor
|
||||
{
|
||||
BodyId = frame.BodyId,
|
||||
SchemaId = frame.SchemaId,
|
||||
LayoutVersion = layout.LayoutVersion,
|
||||
ByteSize = frame.Payload.Length,
|
||||
Capacity = layout.Capacity,
|
||||
ProducerEpoch = frame.ProducerEpoch,
|
||||
Sequence = frame.Sequence,
|
||||
AccessMode = CultMeshBodyAccessMode.ReadOnly,
|
||||
Synchronization = CultMeshBodySynchronization.EpochSequence,
|
||||
LeaseExpiresAtUnixMs = DateTimeOffset.UtcNow.AddSeconds(2).ToUnixTimeMilliseconds(),
|
||||
TransportKind = CultMeshBodyTransportKind.Network,
|
||||
CapabilityToken = "authenticated-quic",
|
||||
SemanticHash = ""
|
||||
},
|
||||
frame.Payload.ToArray());
|
||||
var handler = EntityViewAvailable;
|
||||
if (handler == null)
|
||||
{
|
||||
lease.Dispose();
|
||||
return;
|
||||
}
|
||||
try
|
||||
{
|
||||
handler(generation, lease);
|
||||
_lastPresentedEntityViewEpoch = frame.ProducerEpoch;
|
||||
_lastPresentedEntityViewSequence = frame.Sequence;
|
||||
}
|
||||
catch
|
||||
{
|
||||
lease.Dispose();
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class PendingRealtimeEntityFrame
|
||||
{
|
||||
public PendingRealtimeEntityFrame(string bodyId) =>
|
||||
BodyId = string.IsNullOrWhiteSpace(bodyId)
|
||||
? throw new ArgumentException("Body id must be non-empty.", nameof(bodyId))
|
||||
: bodyId;
|
||||
|
||||
public string BodyId { get; }
|
||||
}
|
||||
|
||||
private sealed class RealtimeEntityBodyLease : ICultMeshBodyReadLease
|
||||
{
|
||||
private byte[] _bytes;
|
||||
|
||||
public RealtimeEntityBodyLease(CultMeshBodyDescriptor descriptor, byte[] bytes)
|
||||
{
|
||||
Descriptor = descriptor;
|
||||
_bytes = bytes;
|
||||
}
|
||||
|
||||
public CultMeshBodyDescriptor Descriptor { get; }
|
||||
public CultMeshBodyTransportKind TransportKind => CultMeshBodyTransportKind.Network;
|
||||
public byte ReadByte(long offset) { Validate(offset, 1); return _bytes[(int)offset]; }
|
||||
public int ReadInt32(long offset) { Validate(offset, 4); return BitConverter.ToInt32(_bytes, (int)offset); }
|
||||
public long ReadInt64(long offset) { Validate(offset, 8); return BitConverter.ToInt64(_bytes, (int)offset); }
|
||||
public float ReadSingle(long offset) { Validate(offset, 4); return BitConverter.ToSingle(_bytes, (int)offset); }
|
||||
public double ReadDouble(long offset) { Validate(offset, 8); return BitConverter.ToDouble(_bytes, (int)offset); }
|
||||
|
||||
public int CopyTo(long offset, byte[] destination, int destinationOffset, int count)
|
||||
{
|
||||
Validate(offset, count);
|
||||
Buffer.BlockCopy(_bytes, (int)offset, destination, destinationOffset, count);
|
||||
return count;
|
||||
}
|
||||
|
||||
public void Dispose() => _bytes = Array.Empty<byte>();
|
||||
|
||||
private void Validate(long offset, int count)
|
||||
{
|
||||
if (offset < 0 || count < 0 || offset > _bytes.LongLength - count)
|
||||
throw new ArgumentOutOfRangeException(nameof(offset));
|
||||
}
|
||||
}
|
||||
|
||||
private void PumpMappedEntityFrame()
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
"GameCult.Caching.dll",
|
||||
"GameCult.Caching.MessagePack.dll",
|
||||
"GameCult.Mesh.dll",
|
||||
"GameCult.Mesh.Quic.Native.dll",
|
||||
"GameCult.Networking.dll"
|
||||
],
|
||||
"autoReferenced": true,
|
||||
|
|
|
|||
|
|
@ -115,6 +115,35 @@ namespace GameCult.Eve.UnityScene.Tests
|
|||
Assert.That(queued.GetType().GetProperty("Publication")!.GetValue(queued), Is.SameAs(publication));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void LiveTransportKeepsOnlyLatestPendingRealtimeGenerationPerBody()
|
||||
{
|
||||
using var transport = new EveUnityCultMeshLiveProviderTransport(
|
||||
"test-replica.cc",
|
||||
"cultnet://127.0.0.1:3075",
|
||||
"provider",
|
||||
"provider.pilot");
|
||||
var enqueue = typeof(EveUnityCultMeshLiveProviderTransport)
|
||||
.GetMethod("EnqueueRealtimeEntityFrame", BindingFlags.NonPublic | BindingFlags.Instance)!;
|
||||
var queue = (System.Collections.Concurrent.ConcurrentQueue<object>)
|
||||
typeof(EveUnityCultMeshLiveProviderTransport)
|
||||
.GetField("_liveDocuments", BindingFlags.NonPublic | BindingFlags.Instance)!
|
||||
.GetValue(transport)!;
|
||||
var latest = (Dictionary<string, CultMeshRealtimeFrame>)
|
||||
typeof(EveUnityCultMeshLiveProviderTransport)
|
||||
.GetField("_latestRealtimeFrames", BindingFlags.NonPublic | BindingFlags.Instance)!
|
||||
.GetValue(transport)!;
|
||||
|
||||
for (var sequence = 0; sequence < 100; sequence++)
|
||||
enqueue.Invoke(transport, new object[] { RealtimeFrame("entities", 7, sequence) });
|
||||
enqueue.Invoke(transport, new object[] { RealtimeFrame("entities", 7, 50) });
|
||||
|
||||
Assert.That(queue.Count, Is.EqualTo(1), "superseded frames must not become render-thread debt");
|
||||
Assert.That(latest, Has.Count.EqualTo(1));
|
||||
Assert.That(latest["entities"].ProducerEpoch, Is.EqualTo(7));
|
||||
Assert.That(latest["entities"].Sequence, Is.EqualTo(99));
|
||||
}
|
||||
|
||||
[Test]
|
||||
public void LiveTransportReadsLaterMappedFramesWithoutNewControlDocuments()
|
||||
{
|
||||
|
|
@ -2916,6 +2945,18 @@ namespace GameCult.Eve.UnityScene.Tests
|
|||
}
|
||||
};
|
||||
|
||||
private static CultMeshRealtimeFrame RealtimeFrame(string bodyId, long epoch, long sequence) =>
|
||||
new CultMeshRealtimeFrame
|
||||
{
|
||||
ChannelId = "state",
|
||||
BodyId = bodyId,
|
||||
SchemaId = "test.entity.slab.v1",
|
||||
ProducerEpoch = epoch,
|
||||
Sequence = sequence,
|
||||
Delivery = CultMeshRealtimeDelivery.LatestOnly,
|
||||
Payload = new byte[4]
|
||||
};
|
||||
|
||||
private static EveProviderAdvertisementDocument ProviderAdvertisement(
|
||||
string providerId,
|
||||
string serviceId,
|
||||
|
|
|
|||
|
|
@ -1,11 +1,11 @@
|
|||
{
|
||||
"name": "org.gamecult.eve.unity-scene",
|
||||
"displayName": "Eve Unity Scene Lowering",
|
||||
"version": "0.3.96",
|
||||
"version": "0.3.97",
|
||||
"unity": "6000.0",
|
||||
"description": "Provider-agnostic Unity scene shell for gamecult.eve.surface.v1 interactive world surfaces.",
|
||||
"dependencies": {
|
||||
"org.gamecult.cultlib": "1.0.35",
|
||||
"org.gamecult.cultlib": "1.0.36",
|
||||
"org.gamecult.eve.plugin-fields": "0.2.3",
|
||||
"org.gamecult.eve.surface": "0.2.4",
|
||||
"com.unity.mathematics": "1.3.2",
|
||||
|
|
|
|||
|
|
@ -39,10 +39,21 @@ $arguments = @(
|
|||
"-assemblyNames", "GameCult.Eve.UnityScene.Tests",
|
||||
"-testResults", $resultsPath, "-logFile", $logPath
|
||||
)
|
||||
$process = Start-Process -FilePath $UnityExe -ArgumentList $arguments -Wait -PassThru -WindowStyle Hidden
|
||||
if ($process.ExitCode -ne 0) {
|
||||
$process = Start-Process -FilePath $UnityExe -ArgumentList $arguments -PassThru -WindowStyle Hidden
|
||||
if (-not $process.WaitForExit([int][TimeSpan]::FromMinutes(10).TotalMilliseconds)) {
|
||||
Stop-Process -Id $process.Id -Force -ErrorAction SilentlyContinue
|
||||
if (Test-Path -LiteralPath $logPath) { Get-Content -LiteralPath $logPath -Tail 120 }
|
||||
throw "EveUnity package tests failed with exit code $($process.ExitCode)"
|
||||
throw "EveUnity package tests timed out after 10 minutes (Unity PID $($process.Id))."
|
||||
}
|
||||
$unityExitCode = $process.ExitCode
|
||||
if ($unityExitCode -ne 0) {
|
||||
if (Test-Path -LiteralPath $logPath) { Get-Content -LiteralPath $logPath -Tail 120 }
|
||||
throw "EveUnity package tests failed with exit code $unityExitCode"
|
||||
}
|
||||
|
||||
if (-not (Test-Path -LiteralPath $resultsPath)) {
|
||||
if (Test-Path -LiteralPath $logPath) { Get-Content -LiteralPath $logPath -Tail 120 }
|
||||
throw "EveUnity package tests produced no results: $resultsPath"
|
||||
}
|
||||
|
||||
[xml]$results = Get-Content -LiteralPath $resultsPath -Raw
|
||||
|
|
|
|||
Reference in a new issue