diff --git a/.gitignore b/.gitignore index c3b8764..0dafaae 100644 --- a/.gitignore +++ b/.gitignore @@ -16,3 +16,7 @@ /packages/ !DotNet~/**/*.csproj !Godot~/**/*.csproj + +/Adapters~/**/bin/ +/Adapters~/**/obj/ +!Adapters~/**/*.csproj diff --git a/Adapters~/MessagePack/README.md b/Adapters~/MessagePack/README.md new file mode 100644 index 0000000..39ed62b --- /dev/null +++ b/Adapters~/MessagePack/README.md @@ -0,0 +1,9 @@ +# Generated MessagePack adapter + +独立可选包。Unity 先安装 MessagePack-CSharp 3.1.8（含其生成器及依赖），再通过 UPM 引用本目录。此目录放在 `Adapters~` 下，不让未安装 MessagePack 的项目产生缺失程序集错误。.NET 使用 `ShrinkSDK.Network.MessagePack`。 + +消息使用 `[MessagePackObject]`、稳定 `[Key(n)]` 和生成的 resolver。构造 `ShrinkNetwork.MessagePack.ShrinkMessagePackNetworkSerializer` 时显式传入 resolver，再为每个消息调用 `Register()`。未注册或没有 formatter 的类型明确报错；不回退到 Contractless／动态 IL／反射调用。 + +在绑定 transport 前注册；停止使用相关消息后才能撤回 codec。共享合同不能依赖 Unity 对象或引擎专属类型。双方必须使用相同合同与消息编码。 + +可编译示例与生成式互通测试：Workspace `Tools/AgentSupport/RuntimeTests/NetworkCodecTests.cs`。 diff --git a/Adapters~/MessagePack/README.md.meta b/Adapters~/MessagePack/README.md.meta new file mode 100644 index 0000000..888d05d --- /dev/null +++ b/Adapters~/MessagePack/README.md.meta @@ -0,0 +1,7 @@ +fileFormatVersion: 2 +guid: 2ad6cd0d17394435bfb2d7bc04e78c0a +DefaultImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Adapters~/MessagePack/Runtime.meta b/Adapters~/MessagePack/Runtime.meta new file mode 100644 index 0000000..20380e6 --- /dev/null +++ b/Adapters~/MessagePack/Runtime.meta @@ -0,0 +1,8 @@ +fileFormatVersion: 2 +guid: b29018215be842d782c6e5b4fc53cf7e +folderAsset: yes +DefaultImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Adapters~/MessagePack/Runtime/ShrinkMessagePackNetworkSerializer.cs b/Adapters~/MessagePack/Runtime/ShrinkMessagePackNetworkSerializer.cs new file mode 100644 index 0000000..4fe3279 --- /dev/null +++ b/Adapters~/MessagePack/Runtime/ShrinkMessagePackNetworkSerializer.cs @@ -0,0 +1,47 @@ +#nullable enable +using System; +using System.Buffers; +using MessagePack; +using MessagePack.Formatters; + +namespace ShrinkNetwork.MessagePack +{ + /// Pass a generated-only resolver; register all message types before starting transports. + public sealed class ShrinkMessagePackNetworkSerializer : ShrinkRegisteredNetworkSerializer + { + private readonly MessagePackSerializerOptions _options; + public ShrinkMessagePackNetworkSerializer(IFormatterResolver generatedResolver) + { + _options = MessagePackSerializerOptions.Standard + .WithResolver(global::MessagePack.Resolvers.CompositeResolver.Create( + generatedResolver ?? throw new ArgumentNullException(nameof(generatedResolver)), + global::MessagePack.Resolvers.BuiltinResolver.Instance)) + .WithSecurity(MessagePackSecurity.UntrustedData); + } + public void Register() => Register(new FormatterCodec(_options)); + + private sealed class FormatterCodec : IShrinkMessageCodec + { + private readonly IMessagePackFormatter _formatter; + private readonly MessagePackSerializerOptions _options; + public FormatterCodec(MessagePackSerializerOptions options) + { + _options = options; + _formatter = options.Resolver.GetFormatter() ?? throw new InvalidOperationException($"No generated MessagePack formatter for {typeof(T).FullName}."); + } + public void Write(IBufferWriter buffer, T value) + { + var writer = new MessagePackWriter(buffer); + _formatter.Serialize(ref writer, value, _options); + writer.Flush(); + } + public T Read(ReadOnlyMemory payload) + { + var reader = new MessagePackReader(payload); + var result = _formatter.Deserialize(ref reader, _options); + if (!reader.End) throw new System.IO.InvalidDataException("Trailing MessagePack payload."); + return result; + } + } + } +} diff --git a/Runtime/Serialization/ShrinkMessagePackNetworkSerializer.cs.meta b/Adapters~/MessagePack/Runtime/ShrinkMessagePackNetworkSerializer.cs.meta similarity index 83% rename from Runtime/Serialization/ShrinkMessagePackNetworkSerializer.cs.meta rename to Adapters~/MessagePack/Runtime/ShrinkMessagePackNetworkSerializer.cs.meta index cfd6e06..fa3326b 100644 --- a/Runtime/Serialization/ShrinkMessagePackNetworkSerializer.cs.meta +++ b/Adapters~/MessagePack/Runtime/ShrinkMessagePackNetworkSerializer.cs.meta @@ -1,5 +1,5 @@ fileFormatVersion: 2 -guid: 88c6239e9fc049f43b2999d7311eec1d +guid: c8394ccb85a14b23b66d1c9ec3a40dc3 MonoImporter: externalObjects: {} serializedVersion: 2 diff --git a/Adapters~/MessagePack/Runtime/ShrinkNetwork.MessagePack.asmdef b/Adapters~/MessagePack/Runtime/ShrinkNetwork.MessagePack.asmdef new file mode 100644 index 0000000..99a2e9c --- /dev/null +++ b/Adapters~/MessagePack/Runtime/ShrinkNetwork.MessagePack.asmdef @@ -0,0 +1,6 @@ +{ + "name": "ShrinkNetwork.MessagePack", + "references": ["ShrinkNetwork.Runtime"], + "autoReferenced": true, + "noEngineReferences": true +} diff --git a/Adapters~/MessagePack/Runtime/ShrinkNetwork.MessagePack.asmdef.meta b/Adapters~/MessagePack/Runtime/ShrinkNetwork.MessagePack.asmdef.meta new file mode 100644 index 0000000..8b5c47f --- /dev/null +++ b/Adapters~/MessagePack/Runtime/ShrinkNetwork.MessagePack.asmdef.meta @@ -0,0 +1,7 @@ +fileFormatVersion: 2 +guid: 74898e5de85247f6b0228c06ee3bdf6e +AssemblyDefinitionImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Adapters~/MessagePack/ShrinkSDK.Network.MessagePack.csproj b/Adapters~/MessagePack/ShrinkSDK.Network.MessagePack.csproj new file mode 100644 index 0000000..6b038b7 --- /dev/null +++ b/Adapters~/MessagePack/ShrinkSDK.Network.MessagePack.csproj @@ -0,0 +1,15 @@ + + + netstandard2.1 + latest + enable + ShrinkSDK.Network.MessagePack + 0.1.0 + false + + + + + + + diff --git a/Adapters~/MessagePack/package.json b/Adapters~/MessagePack/package.json new file mode 100644 index 0000000..83e0b65 --- /dev/null +++ b/Adapters~/MessagePack/package.json @@ -0,0 +1,10 @@ +{ + "name": "com.cneicy.shrink-network-messagepack", + "version": "0.1.0", + "displayName": "Shrink Network MessagePack", + "description": "Explicit generated MessagePack codecs for ShrinkNetwork protocol v2. Requires MessagePack-CSharp 3.1.8.", + "unity": "2022.3", + "dependencies": { + "com.cneicy.shrink-network": "0.4.2" + } +} diff --git a/Adapters~/MessagePack/package.json.meta b/Adapters~/MessagePack/package.json.meta new file mode 100644 index 0000000..bc430df --- /dev/null +++ b/Adapters~/MessagePack/package.json.meta @@ -0,0 +1,7 @@ +fileFormatVersion: 2 +guid: 7e1dd2e113fe4a74b6afe398b315db86 +DefaultImporter: + externalObjects: {} + userData: + assetBundleName: + assetBundleVariant: diff --git a/DotNet~/ShrinkSDK.Network.csproj b/DotNet~/ShrinkSDK.Network.csproj index 540e6a2..ea72056 100644 --- a/DotNet~/ShrinkSDK.Network.csproj +++ b/DotNet~/ShrinkSDK.Network.csproj @@ -5,7 +5,7 @@ ShrinkNetwork.Runtime ShrinkNetwork ShrinkSDK.Network - 0.3.0 + 0.4.2 ShrinkSDK messaging, RPC, permission and transport runtime. true false @@ -13,10 +13,9 @@ - - + diff --git a/Editor/Scaffolding/ServerProjectTemplate/ShrinkNetwork.ServerHost.csproj.txt b/Editor/Scaffolding/ServerProjectTemplate/ShrinkNetwork.ServerHost.csproj.txt index 0272ffc..5d19fb4 100644 --- a/Editor/Scaffolding/ServerProjectTemplate/ShrinkNetwork.ServerHost.csproj.txt +++ b/Editor/Scaffolding/ServerProjectTemplate/ShrinkNetwork.ServerHost.csproj.txt @@ -11,33 +11,10 @@ - - - - - - - - - - - - - - - - - - - - - - - - + diff --git a/README.md b/README.md index da6526b..728069e 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ public static class PingHandlers ```csharp var service = new ShrinkNetworkService( - new ShrinkMessagePackNetworkSerializer(), + new ShrinkJsonNetworkSerializer(), new ShrinkNetworkMessageRegistry(), new ShrinkNetworkRouter()); @@ -285,9 +285,10 @@ new ShrinkRpcCallOptions ```csharp new ShrinkJsonNetworkSerializer() -new ShrinkMessagePackNetworkSerializer() ``` +生成式 MessagePack 通过可选包 `Adapters~/MessagePack` 接入。使用 `ShrinkNetwork.MessagePack.ShrinkMessagePackNetworkSerializer(GeneratedResolver.Instance)` 并显式 `Register()`；旧无参反射适配器已移除。消息体只编码一次，外层统一使用协议 v2 的 SHK2 二进制封装。 + ## ✅ 最佳实践 - 所有消息统一用 `[ShrinkNetworkMessage]` 声明，不要只靠手写 `RegisterMessage()` @@ -367,3 +368,13 @@ var serverTransport = new ShrinkTcpServerTransport( ## 📄 License [MIT](LICENSE) + +## 可复用能力 + + +Capability: 消息合同、RPC、权限、TCP/KCP/Loopback 传输与有预算的串行排队 +Aliases: 联机 网络 远程 rpc 广播 拥塞 背压 同步 +Limits: 协议 v2 拒绝旧封装；TCP/KCP 均为可靠通道；可替换状态仅合并排队项，不用于快照分片 +Extension: RegisterMessage/RegisterHandler；IShrinkNetworkBufferSerializer；ShrinkNetworkWorkQueue 状态键 +Evidence: [ShrinkNetworkService](Runtime/Core/ShrinkNetworkService.cs); [ShrinkPacketCodec](Runtime/Serialization/ShrinkPacketCodec.cs); [ShrinkNetworkWorkQueue](Runtime/Core/ShrinkNetworkWorkQueue.cs) + diff --git a/Runtime/Core/ShrinkNetworkService.cs b/Runtime/Core/ShrinkNetworkService.cs index 40e65ad..6dbd43b 100644 --- a/Runtime/Core/ShrinkNetworkService.cs +++ b/Runtime/Core/ShrinkNetworkService.cs @@ -54,7 +54,7 @@ namespace ShrinkNetwork Serializer = serializer ?? throw new ArgumentNullException(nameof(serializer)); MessageRegistry = messageRegistry ?? throw new ArgumentNullException(nameof(messageRegistry)); Router = router ?? throw new ArgumentNullException(nameof(router)); - _dispatchScheduler = dispatchScheduler ?? ShrinkNetworkDispatchSchedulers.Inline; + _dispatchScheduler = dispatchScheduler ?? new ShrinkNetworkInlineDispatchScheduler(); } public IShrinkNetworkSerializer Serializer { get; } @@ -74,6 +74,8 @@ namespace ShrinkNetwork set => _dispatchScheduler = value ?? throw new ArgumentNullException(nameof(value)); } + /// Reliable receive work could not be processed. Session-control transports also disconnect the peer. + public event Action? OnDispatchRejected; public event Action? OnSessionConnected; public event Action? OnSessionDisconnected; @@ -273,11 +275,11 @@ namespace ShrinkNetwork if (messageType == null) throw new ArgumentNullException(nameof(messageType)); - return SendPacketAsync(session, messageType, kind, requestToken, route, Serializer.Serialize(message)); + return SendPacketAsync(session, messageType, kind, requestToken, route, Array.Empty(), message); } private async UniTask SendPacketAsync(ShrinkNetworkSession session, Type messageType, - ShrinkNetworkPacketKind kind, ShrinkRequestToken requestToken, string? route, byte[] payload) + ShrinkNetworkPacketKind kind, ShrinkRequestToken requestToken, string? route, byte[] payload, object? message = null) { if (session == null) throw new ArgumentNullException(nameof(session)); @@ -299,16 +301,22 @@ namespace ShrinkNetwork Payload = payload }; - var packetData = Serializer.Serialize(packet); + using var encoded = ShrinkPacketCodec.Encode(packet, Serializer, message); + var packetData = encoded.WrittenMemory; Interlocked.Increment(ref _packetsSent); Interlocked.Add(ref _bytesSent, packetData.Length); + if (transport is IShrinkNetworkMemoryTransport memoryTransport) + { + await memoryTransport.SendAsync(session.SessionId, packetData); + return; + } if (transport is IShrinkNetworkAsyncTransport asyncTransport) { - await asyncTransport.SendAsync(session.SessionId, packetData); + await asyncTransport.SendAsync(session.SessionId, packetData.ToArray()); return; } - transport.Send(session.SessionId, packetData); + transport.Send(session.SessionId, packetData.ToArray()); } private void OnTransportEvent(ShrinkNetworkTransportEvent evt) @@ -330,6 +338,12 @@ namespace ShrinkNetwork return; } + // Responses complete pending RPCs independently of the serial handler queue, including nested calls. + if (evt.Type == ShrinkNetworkTransportEventType.Packet && ShrinkPacketCodec.IsResponse(evt.PacketData)) + { + HandlePacketAsync(evt).Forget(); + return; + } ScheduleTransportEventAsync(evt).Forget(); } @@ -337,9 +351,23 @@ namespace ShrinkNetwork { try { - var scheduled = await DispatchScheduler.ScheduleAsync(() => HandleTransportEventAsync(evt)); + var scheduler = DispatchScheduler; + _sessions.TryGetValue(evt.SessionId, out var receivedSession); + UniTask DispatchCurrent() + { + return receivedSession != null && _sessions.TryGetValue(evt.SessionId, out var current) && ReferenceEquals(current, receivedSession) + ? HandleTransportEventAsync(evt) : UniTask.CompletedTask; + } + var scheduled = scheduler is IShrinkNetworkPacketDispatchScheduler queue + ? await queue.ScheduleAsync(evt.SessionId, evt.PacketData?.Length ?? 0, DispatchCurrent) + : await scheduler.ScheduleAsync(DispatchCurrent); if (!scheduled) + { Interlocked.Increment(ref _dispatchQueueRejectedCount); + OnDispatchRejected?.Invoke(evt.SessionId); + if (_transport is IShrinkNetworkSessionControlTransport control) + control.DisconnectSession(evt.SessionId, "SHRINK-NET-CONGESTION: reliable receive queue rejected work."); + } } catch (Exception ex) { @@ -395,12 +423,12 @@ namespace ShrinkNetwork { Interlocked.Increment(ref _packetsReceived); Interlocked.Add(ref _bytesReceived, evt.PacketData.Length); - var packet = Serializer.Deserialize(evt.PacketData); + var packet = ShrinkPacketCodec.Decode(evt.PacketData); if (!_sessions.TryGetValue(evt.SessionId, out var session)) { - session = _sessions.GetOrAdd(evt.SessionId, - id => new ShrinkNetworkSession(id, evt.RemoteAddress, this)); + // Queued packets from a disconnected session cannot resurrect its state. + return; } if (!ValidatePacketCompatibility(packet, evt.SessionId)) @@ -410,7 +438,7 @@ namespace ShrinkNetwork if (packet.Kind == ShrinkNetworkPacketKind.Response) { - HandleResponse(packet); + HandleResponse(session, packet); return; } @@ -425,7 +453,7 @@ namespace ShrinkNetwork } var resolvedMeta = meta!; - var message = Serializer.Deserialize(packet.Payload, resolvedMeta.MessageType); + var message = DeserializePayload(packet.Payload, resolvedMeta.MessageType); if (message == null) { ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Failed to deserialize message for opcode {packet.Opcode}."); @@ -442,6 +470,12 @@ namespace ShrinkNetwork ShrinkNetworkLogger.Warn($"[ShrinkNetwork] No handler found for {resolvedMeta.MessageType.FullName}"); } } + catch (ShrinkProtocolException ex) + { + Interlocked.Increment(ref _protocolViolations); + if (DisconnectOnProtocolViolation && _transport is IShrinkNetworkSessionControlTransport control) + control.DisconnectSession(evt.SessionId, ex.Message); + } catch (Exception ex) { Interlocked.Increment(ref _serializationErrorCount); @@ -450,9 +484,10 @@ namespace ShrinkNetwork } } - private void HandleResponse(ShrinkNetworkPacket packet) + private void HandleResponse(ShrinkNetworkSession session, ShrinkNetworkPacket packet) { - if (!_pendingRequests.TryRemove(packet.RequestToken, out var pending)) + if (!_pendingRequests.TryGetValue(packet.RequestToken, out var expected) || expected.SessionId != session.SessionId || + !_pendingRequests.TryRemove(packet.RequestToken, out var pending)) { ShrinkNetworkLogger.Warn($"[ShrinkNetwork] Pending request not found. RequestToken={packet.RequestToken}"); return; @@ -460,7 +495,7 @@ namespace ShrinkNetwork try { - var response = Serializer.Deserialize(packet.Payload, pending.ResponseType); + var response = DeserializePayload(packet.Payload, pending.ResponseType); pending.CompletionSource.TrySetResult(response); } catch (Exception ex) @@ -469,6 +504,10 @@ namespace ShrinkNetwork } } + private object DeserializePayload(ReadOnlyMemory payload, Type type) => + Serializer is IShrinkNetworkBufferSerializer buffered + ? buffered.Deserialize(payload, type) : Serializer.Deserialize(payload.ToArray(), type); + private async UniTask WaitForPendingResponse(ShrinkRequestToken requestToken, PendingRequest pending, ShrinkRpcCallOptions? options) where TResponse : class, IShrinkNetworkResponse @@ -627,20 +666,42 @@ namespace ShrinkNetwork public static class ShrinkNetworkDispatchSchedulers { - public static IShrinkNetworkDispatchScheduler Inline { get; } = - new ShrinkNetworkInlineDispatchScheduler(); + public static IShrinkNetworkDispatchScheduler Inline => new ShrinkNetworkInlineDispatchScheduler(); } - public sealed class ShrinkNetworkInlineDispatchScheduler : IShrinkNetworkDispatchScheduler + public interface IShrinkNetworkPacketDispatchScheduler : IShrinkNetworkDispatchScheduler { - public async UniTask ScheduleAsync(Func callback) - { - if (callback == null) - throw new ArgumentNullException(nameof(callback)); + UniTask ScheduleAsync(long sessionId, int bytes, Func callback); + } - await callback(); - return true; + /// Automatically drains a bounded serial queue. Callbacks start on the initiating transport/continuation thread. + public sealed class ShrinkNetworkInlineDispatchScheduler : IShrinkNetworkPacketDispatchScheduler, IDisposable + { + private readonly object _gate = new(); + private readonly ShrinkNetworkWorkQueue _queue = new(4096, 64 * 1024 * 1024); + private bool _draining; + public ShrinkNetworkQueueDiagnostics CaptureDiagnostics() => _queue.CaptureDiagnostics(); + public UniTask ScheduleAsync(Func callback) => ScheduleAsync(0, 0, callback); + public async UniTask ScheduleAsync(long sessionId, int bytes, Func callback) + { + var completion = _queue.EnqueueAsync(sessionId, "receive", bytes, callback); + var start = false; + lock (_gate) { if (!_draining) { _draining = true; start = true; } } + if (start) DrainAsync().Forget(); + return await completion == ShrinkNetworkQueueResult.Completed; } + private async UniTask DrainAsync() + { + while (true) + { + await _queue.PumpAsync(int.MaxValue); + lock (_gate) + { + if (_queue.PendingCount == 0) { _draining = false; return; } + } + } + } + public void Dispose() => _queue.Dispose(); } public enum ShrinkNetworkDispatchOverflowPolicy @@ -650,163 +711,30 @@ namespace ShrinkNetwork DropOldest = 2 } - /// - /// A caller-pumped, bounded dispatch queue. Unity can pump it from Update - /// while a dedicated server can keep the default inline scheduler. - /// - public sealed class ShrinkNetworkDispatchQueue : IShrinkNetworkDispatchScheduler, IDisposable + /// Caller-pumped serial receive queue; rejects overflow without silently dropping reliable packets. + public sealed class ShrinkNetworkDispatchQueue : IShrinkNetworkPacketDispatchScheduler, IDisposable { - private sealed class WorkItem - { - public Func Callback = null!; - public UniTaskCompletionSource Completion = null!; - } - - private readonly ConcurrentQueue _queue = new(); - private readonly object _lifecycleLock = new(); - private readonly int _capacity; - private readonly ShrinkNetworkDispatchOverflowPolicy _overflowPolicy; - private int _queuedCount; - private int _pumping; - private int _disposed; - private long _rejectedCount; - private long _droppedCount; - + private readonly ShrinkNetworkWorkQueue _queue; public ShrinkNetworkDispatchQueue(int capacity, - ShrinkNetworkDispatchOverflowPolicy overflowPolicy = ShrinkNetworkDispatchOverflowPolicy.Reject) + ShrinkNetworkDispatchOverflowPolicy overflowPolicy = ShrinkNetworkDispatchOverflowPolicy.Reject, + long byteCapacity = 64 * 1024 * 1024, int perSessionCapacity = int.MaxValue) { - if (capacity <= 0) - throw new ArgumentOutOfRangeException(nameof(capacity)); - - _capacity = capacity; - _overflowPolicy = overflowPolicy; - } - - public int Capacity => _capacity; - public int PendingCount => Volatile.Read(ref _queuedCount); - public long RejectedCount => Volatile.Read(ref _rejectedCount); - public long DroppedCount => Volatile.Read(ref _droppedCount); - - public UniTask ScheduleAsync(Func callback) - { - if (callback == null) - throw new ArgumentNullException(nameof(callback)); - if (Volatile.Read(ref _disposed) != 0) - return UniTask.FromException(new ObjectDisposedException(nameof(ShrinkNetworkDispatchQueue))); - - var item = new WorkItem - { - Callback = callback, - Completion = new UniTaskCompletionSource() - }; - - while (true) - { - if (Volatile.Read(ref _queuedCount) >= _capacity) - { - switch (_overflowPolicy) - { - case ShrinkNetworkDispatchOverflowPolicy.Reject: - Interlocked.Increment(ref _rejectedCount); - return UniTask.FromResult(false); - case ShrinkNetworkDispatchOverflowPolicy.DropNewest: - Interlocked.Increment(ref _droppedCount); - return UniTask.FromResult(false); - case ShrinkNetworkDispatchOverflowPolicy.DropOldest: - if (_queue.TryDequeue(out var dropped)) - { - Interlocked.Decrement(ref _queuedCount); - Interlocked.Increment(ref _droppedCount); - dropped.Completion.TrySetResult(false); - continue; - } - - Thread.Yield(); - continue; - default: - throw new ArgumentOutOfRangeException(); - } - } - - var currentCount = Volatile.Read(ref _queuedCount); - if (currentCount >= _capacity || - Interlocked.CompareExchange(ref _queuedCount, currentCount + 1, currentCount) != currentCount) - { - continue; - } - - lock (_lifecycleLock) - { - if (Volatile.Read(ref _disposed) != 0) - { - Interlocked.Decrement(ref _queuedCount); - item.Completion.TrySetResult(false); - return item.Completion.Task; - } - - _queue.Enqueue(item); - return item.Completion.Task; - } - } - } - - public UniTask PumpAsync(int maxItems) - { - if (maxItems <= 0) - throw new ArgumentOutOfRangeException(nameof(maxItems)); - if (Interlocked.Exchange(ref _pumping, 1) == 1) - return UniTask.FromResult(0); - - return PumpCoreAsync(maxItems); - } - - public void Dispose() - { - lock (_lifecycleLock) - { - if (Interlocked.Exchange(ref _disposed, 1) != 0) - return; - - while (_queue.TryDequeue(out var item)) - { - Interlocked.Decrement(ref _queuedCount); - item.Completion.TrySetResult(false); - } - } - } - - private async UniTask PumpCoreAsync(int maxItems) - { - var processed = 0; - try - { - while (processed < maxItems && _queue.TryDequeue(out var item)) - { - Interlocked.Decrement(ref _queuedCount); - await ExecuteItemAsync(item); - processed++; - } - - return processed; - } - finally - { - Volatile.Write(ref _pumping, 0); - } - } - - private static async UniTask ExecuteItemAsync(WorkItem item) - { - try - { - await item.Callback(); - item.Completion.TrySetResult(true); - } - catch (Exception ex) - { - item.Completion.TrySetException(ex); - } + if (overflowPolicy != ShrinkNetworkDispatchOverflowPolicy.Reject) + throw new ArgumentException("Protocol v2 reliable dispatch requires Reject. Use an explicit state key on ShrinkNetworkWorkQueue for replaceable state.", nameof(overflowPolicy)); + Capacity = capacity; + _queue = new ShrinkNetworkWorkQueue(capacity, byteCapacity, perSessionCapacity); } + public int Capacity { get; } + public int PendingCount => _queue.PendingCount; + public long RejectedCount => CaptureDiagnostics().Rejected; + public long DroppedCount => 0; + public ShrinkNetworkQueueDiagnostics CaptureDiagnostics() => _queue.CaptureDiagnostics(); + public UniTask ScheduleAsync(Func callback) => ScheduleAsync(0, 0, callback); + public async UniTask ScheduleAsync(long sessionId, int bytes, Func callback) => + await _queue.EnqueueAsync(sessionId, "receive", bytes, callback) == ShrinkNetworkQueueResult.Completed; + public UniTask PumpAsync(int maxItems) => _queue.PumpAsync(maxItems); + public UniTask PumpAsync(int maxItems, long maxBytes, TimeSpan timeBudget) => _queue.PumpAsync(maxItems, maxBytes, timeBudget); + public void Dispose() => _queue.Dispose(); } public sealed class ShrinkNetworkServiceDiagnosticsSnapshot diff --git a/Runtime/Core/ShrinkNetworkWorkQueue.cs b/Runtime/Core/ShrinkNetworkWorkQueue.cs new file mode 100644 index 0000000..d652ed7 --- /dev/null +++ b/Runtime/Core/ShrinkNetworkWorkQueue.cs @@ -0,0 +1,159 @@ +#nullable enable +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Threading; +using Cysharp.Threading.Tasks; + +namespace ShrinkNetwork +{ + public enum ShrinkNetworkQueueResult { Completed, Rejected, Replaced, Canceled } + + public sealed class ShrinkNetworkQueueDiagnostics + { + public int PendingCount { get; internal set; } + public long PendingBytes { get; internal set; } + public long Rejected { get; internal set; } + public long Replaced { get; internal set; } + public long Completed { get; internal set; } + public double OldestWaitMilliseconds { get; internal set; } + public double LastWaitMilliseconds { get; internal set; } + } + + /// Caller-pumped serial execution, fair across session/channel partitions. Only queued state is replaceable. + public sealed class ShrinkNetworkWorkQueue : IDisposable + { + private sealed class Work + { + public Func Callback = null!; + public UniTaskCompletionSource Completion = new(); + public string? StateKey; + public int Bytes; + public long Enqueued = Stopwatch.GetTimestamp(); + public CancellationToken Cancellation; + } + private sealed class Partition + { + public readonly LinkedList Items = new(); + public readonly Dictionary> States = new(StringComparer.Ordinal); + } + private readonly object _gate = new(); + private readonly Dictionary<(long, string), Partition> _partitions = new(); + private readonly Queue<(long, string)> _ready = new(); + private readonly int _capacity; + private readonly long _byteCapacity; + private readonly int _partitionCapacity; + private int _count, _pumping; + private long _bytes, _rejected, _replaced, _completed; + private double _lastWait; + private bool _disposed; + + public int PendingCount { get { lock (_gate) return _count; } } + + public ShrinkNetworkWorkQueue(int capacity, long byteCapacity, int perPartitionCapacity = int.MaxValue) + { + if (capacity <= 0 || byteCapacity <= 0 || perPartitionCapacity <= 0) throw new ArgumentOutOfRangeException(nameof(capacity)); + _capacity = capacity; _byteCapacity = byteCapacity; _partitionCapacity = perPartitionCapacity; + } + + /// A nonempty stateKey explicitly permits replacing an unsent state in this session/channel. Never use for RPC or snapshot fragments. + public UniTask EnqueueAsync(long sessionId, string channel, int byteCount, + Func callback, string? stateKey = null, CancellationToken cancellationToken = default) + { + if (callback == null) throw new ArgumentNullException(nameof(callback)); + if (channel == null) throw new ArgumentNullException(nameof(channel)); + if (byteCount < 0) throw new ArgumentOutOfRangeException(nameof(byteCount)); + Work? replaced = null; + Work item; + lock (_gate) + { + if (_disposed || cancellationToken.IsCancellationRequested) return UniTask.FromResult(ShrinkNetworkQueueResult.Canceled); + var key = (sessionId, channel); + _partitions.TryGetValue(key, out var partition); + LinkedListNode? old = null; + if (!string.IsNullOrEmpty(stateKey)) partition?.States.TryGetValue(stateKey!, out old); + var nextBytes = _bytes - (old?.Value.Bytes ?? 0) + byteCount; + if (nextBytes > _byteCapacity || (old == null && (_count >= _capacity || (partition?.Items.Count ?? 0) >= _partitionCapacity))) + { _rejected++; return UniTask.FromResult(ShrinkNetworkQueueResult.Rejected); } + item = new Work { Callback = callback, Bytes = byteCount, StateKey = string.IsNullOrEmpty(stateKey) ? null : stateKey, Cancellation = cancellationToken }; + if (partition == null) { partition = new Partition(); _partitions.Add(key, partition); _ready.Enqueue(key); } + if (old != null) + { + replaced = old.Value; + // Move a replacement to the tail: later state must not jump ahead of intervening reliable operations. + partition.Items.Remove(old); _replaced++; + } + else _count++; + var node = partition.Items.AddLast(item); + if (item.StateKey != null) partition.States[item.StateKey] = node; + _bytes = nextBytes; + } + replaced?.Completion.TrySetResult(ShrinkNetworkQueueResult.Replaced); + return item.Completion.Task; + } + + public async UniTask PumpAsync(int maxItems, long maxBytes = long.MaxValue, TimeSpan? timeBudget = null) + { + if (maxItems <= 0 || maxBytes <= 0 || (timeBudget.HasValue && timeBudget.Value <= TimeSpan.Zero)) throw new ArgumentOutOfRangeException(nameof(maxItems)); + if (Interlocked.Exchange(ref _pumping, 1) != 0) return 0; + var started = Stopwatch.GetTimestamp(); + var processed = 0; + long bytes = 0; + try + { + while (processed < maxItems && (!timeBudget.HasValue || Elapsed(started) < timeBudget.Value.TotalMilliseconds)) + { + Work item; + lock (_gate) + { + if (_disposed || _ready.Count == 0) break; + var key = _ready.Peek(); + var partition = _partitions[key]; + item = partition.Items.First!.Value; + // Allow one oversized item so a byte budget cannot permanently starve a valid packet. + if (processed > 0 && item.Bytes > maxBytes - bytes) break; + _ready.Dequeue(); partition.Items.RemoveFirst(); + if (item.StateKey != null) partition.States.Remove(item.StateKey); + if (partition.Items.Count == 0) _partitions.Remove(key); else _ready.Enqueue(key); + _count--; _bytes -= item.Bytes; _lastWait = Elapsed(item.Enqueued); + } + try + { + if (item.Cancellation.IsCancellationRequested) item.Completion.TrySetResult(ShrinkNetworkQueueResult.Canceled); + else { await item.Callback(); item.Completion.TrySetResult(ShrinkNetworkQueueResult.Completed); lock (_gate) _completed++; } + } + catch (OperationCanceledException) { item.Completion.TrySetResult(ShrinkNetworkQueueResult.Canceled); } + catch (Exception ex) { item.Completion.TrySetException(ex); } + processed++; bytes += item.Bytes; + } + return processed; + } + finally { Volatile.Write(ref _pumping, 0); } + } + + public ShrinkNetworkQueueDiagnostics CaptureDiagnostics() + { + lock (_gate) + { + long oldest = Stopwatch.GetTimestamp(); + foreach (var partition in _partitions.Values) + if (partition.Items.First != null) oldest = Math.Min(oldest, partition.Items.First.Value.Enqueued); + return new ShrinkNetworkQueueDiagnostics { PendingCount = _count, PendingBytes = _bytes, Rejected = _rejected, Replaced = _replaced, + Completed = _completed, LastWaitMilliseconds = _lastWait, OldestWaitMilliseconds = _count == 0 ? 0 : Elapsed(oldest) }; + } + } + private static double Elapsed(long start) => (Stopwatch.GetTimestamp() - start) * 1000d / Stopwatch.Frequency; + public void Dispose() + { + List canceled = new(); + lock (_gate) + { + if (_disposed) return; + _disposed = true; + foreach (var partition in _partitions.Values) canceled.AddRange(partition.Items); + _partitions.Clear(); _ready.Clear(); _count = 0; _bytes = 0; + } + foreach (var item in canceled) item.Completion.TrySetResult(ShrinkNetworkQueueResult.Canceled); + } + } +} diff --git a/Runtime/Core/ShrinkNetworkWorkQueue.cs.meta b/Runtime/Core/ShrinkNetworkWorkQueue.cs.meta new file mode 100644 index 0000000..8e0b7ef --- /dev/null +++ b/Runtime/Core/ShrinkNetworkWorkQueue.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 3b35a44b897396549b5a1dec58ae05c0 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Runtime/Metadata/ShrinkNetworkPacket.cs b/Runtime/Metadata/ShrinkNetworkPacket.cs index 01299cd..9d60094 100644 --- a/Runtime/Metadata/ShrinkNetworkPacket.cs +++ b/Runtime/Metadata/ShrinkNetworkPacket.cs @@ -6,7 +6,7 @@ namespace ShrinkNetwork { public static class ShrinkNetworkProtocol { - public const int CurrentProtocolVersion = 1; + public const int CurrentProtocolVersion = 2; public const int CurrentSchemaVersion = 1; } @@ -28,6 +28,6 @@ namespace ShrinkNetwork public long SessionTokenExpiresAtUnixTimeSeconds; public string? Route; public ShrinkNetworkPacketKind Kind; - public byte[] Payload = Array.Empty(); + public ReadOnlyMemory Payload = ReadOnlyMemory.Empty; } } diff --git a/Runtime/Routing/ShrinkNetworkMessageRegistry.cs b/Runtime/Routing/ShrinkNetworkMessageRegistry.cs index 31ebdf3..fe882e0 100644 --- a/Runtime/Routing/ShrinkNetworkMessageRegistry.cs +++ b/Runtime/Routing/ShrinkNetworkMessageRegistry.cs @@ -9,6 +9,8 @@ namespace ShrinkNetwork { private readonly Dictionary _opcodeToMeta = new(); private readonly Dictionary _typeToMeta = new(); + private readonly HashSet _routes = new(StringComparer.Ordinal); + public IReadOnlyCollection Registrations => _typeToMeta.Values; public void Register(int opcode, string? route = null) where TMessage : IShrinkNetworkMessage => Register(typeof(TMessage), opcode, route); @@ -24,6 +26,8 @@ namespace ShrinkNetwork if (_typeToMeta.ContainsKey(messageType)) throw new InvalidOperationException($"Message type {messageType.FullName} is already registered."); + route = string.IsNullOrWhiteSpace(route) ? null : route.Trim(); + if (route != null && !_routes.Add(route)) throw new InvalidOperationException($"SHRINK002 Route '{route}' is already registered."); var meta = new ShrinkNetworkMessageMeta(opcode, messageType, route); _opcodeToMeta.Add(opcode, meta); _typeToMeta.Add(messageType, meta); diff --git a/Runtime/Routing/ShrinkNetworkRouter.cs b/Runtime/Routing/ShrinkNetworkRouter.cs index 52f8a4f..e7ff7c2 100644 --- a/Runtime/Routing/ShrinkNetworkRouter.cs +++ b/Runtime/Routing/ShrinkNetworkRouter.cs @@ -2,6 +2,7 @@ using System; using System.Collections.Generic; +using System.Reflection; using Cysharp.Threading.Tasks; namespace ShrinkNetwork @@ -21,6 +22,8 @@ namespace ShrinkNetwork public Func> Handler = null!; } + private readonly Dictionary _bindingMethods = new(); + public IReadOnlyDictionary CaptureBindings() => new Dictionary(_bindingMethods); private readonly Dictionary _messageHandlers = new(); private readonly Dictionary _requestHandlers = new(); @@ -28,7 +31,9 @@ namespace ShrinkNetwork ShrinkNetworkPermissionRequirement requirement = default) where TMessage : IShrinkNetworkMessage { + if (handler == null) throw new ArgumentNullException(nameof(handler)); RegisterHandler(typeof(TMessage), (context, message) => handler(context, (TMessage)message), requirement); + _bindingMethods[typeof(TMessage)] = handler.Method; } public void RegisterHandler(Type messageType, Func handler, @@ -41,6 +46,7 @@ namespace ShrinkNetwork if (_messageHandlers.ContainsKey(messageType) || _requestHandlers.ContainsKey(messageType)) throw new InvalidOperationException($"Handler already exists for {messageType.FullName}."); + _bindingMethods[messageType] = handler.Method; _messageHandlers.Add(messageType, new MessageHandlerRegistration { Requirement = requirement, @@ -53,8 +59,10 @@ namespace ShrinkNetwork where TRequest : IShrinkNetworkRequest where TResponse : class, IShrinkNetworkResponse { + if (handler == null) throw new ArgumentNullException(nameof(handler)); RegisterRequestHandler(typeof(TRequest), typeof(TResponse), async (context, message) => await handler(context, (TRequest)message), requirement); + _bindingMethods[typeof(TRequest)] = handler.Method; } public void RegisterRequestHandler(Type requestType, Type responseType, @@ -70,6 +78,7 @@ namespace ShrinkNetwork if (_messageHandlers.ContainsKey(requestType) || _requestHandlers.ContainsKey(requestType)) throw new InvalidOperationException($"Handler already exists for {requestType.FullName}."); + _bindingMethods[requestType] = handler.Method; _requestHandlers.Add(requestType, new RequestHandlerRegistration { ResponseType = responseType, diff --git a/Runtime/Serialization/IShrinkNetworkSerializer.cs b/Runtime/Serialization/IShrinkNetworkSerializer.cs index 9b2a032..6b4394f 100644 --- a/Runtime/Serialization/IShrinkNetworkSerializer.cs +++ b/Runtime/Serialization/IShrinkNetworkSerializer.cs @@ -1,4 +1,5 @@ using System; +using System.Buffers; namespace ShrinkNetwork { @@ -8,4 +9,10 @@ namespace ShrinkNetwork object Deserialize(byte[] payload, Type type); T Deserialize(byte[] payload); } + + public interface IShrinkNetworkBufferSerializer : IShrinkNetworkSerializer + { + void Serialize(IBufferWriter writer, object value); + object Deserialize(ReadOnlyMemory payload, Type type); + } } diff --git a/Runtime/Serialization/ShrinkBufferWriter.cs b/Runtime/Serialization/ShrinkBufferWriter.cs new file mode 100644 index 0000000..535362c --- /dev/null +++ b/Runtime/Serialization/ShrinkBufferWriter.cs @@ -0,0 +1,40 @@ +#nullable enable +using System; +using System.Buffers; + +namespace ShrinkNetwork +{ + /// Owns rented memory until Dispose. A sender must await completion before disposing. + public sealed class ShrinkBufferWriter : IBufferWriter, IDisposable + { + private byte[]? _buffer; + public ShrinkBufferWriter(int initialCapacity = 256) => _buffer = ArrayPool.Shared.Rent(Math.Max(1, initialCapacity)); + public int WrittenCount { get; private set; } + public ReadOnlyMemory WrittenMemory => Buffer.AsMemory(0, WrittenCount); + internal Span WrittenSpan => Buffer.AsSpan(0, WrittenCount); + private byte[] Buffer => _buffer ?? throw new ObjectDisposedException(nameof(ShrinkBufferWriter)); + public void Advance(int count) + { + if (count < 0 || count > Buffer.Length - WrittenCount) throw new ArgumentOutOfRangeException(nameof(count)); + WrittenCount += count; + } + public Memory GetMemory(int sizeHint = 0) { Ensure(sizeHint); return Buffer.AsMemory(WrittenCount); } + public Span GetSpan(int sizeHint = 0) { Ensure(sizeHint); return Buffer.AsSpan(WrittenCount); } + private void Ensure(int sizeHint) + { + if (sizeHint < 0) throw new ArgumentOutOfRangeException(nameof(sizeHint)); + sizeHint = Math.Max(1, sizeHint); + if (sizeHint <= Buffer.Length - WrittenCount) return; + var next = ArrayPool.Shared.Rent(checked(Math.Max(Buffer.Length * 2, WrittenCount + sizeHint))); + Buffer.AsSpan(0, WrittenCount).CopyTo(next); + ArrayPool.Shared.Return(Buffer, clearArray: true); + _buffer = next; + } + public void Dispose() + { + var buffer = _buffer; + _buffer = null; + if (buffer != null) ArrayPool.Shared.Return(buffer, clearArray: true); + } + } +} diff --git a/Runtime/Serialization/ShrinkBufferWriter.cs.meta b/Runtime/Serialization/ShrinkBufferWriter.cs.meta new file mode 100644 index 0000000..71a2da2 --- /dev/null +++ b/Runtime/Serialization/ShrinkBufferWriter.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 9ae3dee10fcf29f4da1b7ace3063bf92 +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Runtime/Serialization/ShrinkJsonNetworkSerializer.cs b/Runtime/Serialization/ShrinkJsonNetworkSerializer.cs index 5561e0d..c8cf451 100644 --- a/Runtime/Serialization/ShrinkJsonNetworkSerializer.cs +++ b/Runtime/Serialization/ShrinkJsonNetworkSerializer.cs @@ -2,11 +2,13 @@ using System; using System.Text; +using System.Buffers; +using System.IO; using Newtonsoft.Json; namespace ShrinkNetwork { - public sealed class ShrinkJsonNetworkSerializer : IShrinkNetworkSerializer + public sealed class ShrinkJsonNetworkSerializer : IShrinkNetworkBufferSerializer { private static readonly JsonSerializerSettings Settings = new() { @@ -25,5 +27,33 @@ namespace ShrinkNetwork public T Deserialize(byte[] payload) => JsonConvert.DeserializeObject(Encoding.UTF8.GetString(payload), Settings) ?? throw new JsonSerializationException($"Failed to deserialize payload into {typeof(T).FullName}."); + + public void Serialize(IBufferWriter writer, object value) + { + using var stream = new BufferStream(writer); + using var text = new StreamWriter(stream, new UTF8Encoding(false), 1024, leaveOpen: true); + using var json = new JsonTextWriter(text); + JsonSerializer.Create(Settings).Serialize(json, value); + } + + public object Deserialize(ReadOnlyMemory payload, Type type) => + JsonConvert.DeserializeObject(Encoding.UTF8.GetString(payload.Span), type, Settings) + ?? throw new JsonSerializationException($"Failed to deserialize payload into {type.FullName}."); + + private sealed class BufferStream : Stream + { + private readonly IBufferWriter _writer; + public BufferStream(IBufferWriter writer) => _writer = writer; + public override bool CanRead => false; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public override void Flush() { } + public override void Write(byte[] buffer, int offset, int count) { buffer.AsSpan(offset, count).CopyTo(_writer.GetSpan(count)); _writer.Advance(count); } + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + } } } diff --git a/Runtime/Serialization/ShrinkMessagePackNetworkSerializer.cs b/Runtime/Serialization/ShrinkMessagePackNetworkSerializer.cs deleted file mode 100644 index 4a7bbdb..0000000 --- a/Runtime/Serialization/ShrinkMessagePackNetworkSerializer.cs +++ /dev/null @@ -1,150 +0,0 @@ -#nullable enable - -using System; -using System.Linq; -using System.Reflection; - -namespace ShrinkNetwork -{ - public sealed class ShrinkMessagePackNetworkSerializer : IShrinkNetworkSerializer - { - private readonly MethodInfo _serializeMethod; - private readonly MethodInfo _deserializeMethod; - private readonly object? _serializerOptions; - - public ShrinkMessagePackNetworkSerializer() - { - var serializerType = Type.GetType("MessagePack.MessagePackSerializer, MessagePack"); - if (serializerType == null) - { - throw new InvalidOperationException( - "MessagePack assembly was not found. Please install MessagePack-CSharp before using ShrinkMessagePackNetworkSerializer."); - } - - _serializerOptions = ResolveSerializerOptions(serializerType.Assembly); - - var serializeMethod = serializerType - .GetMethods(BindingFlags.Public | BindingFlags.Static) - .FirstOrDefault(m => - { - if (m.Name != "Serialize") - return false; - var parameters = m.GetParameters(); - return parameters.Length >= 2 && - parameters[0].ParameterType == typeof(Type) && - parameters[1].ParameterType == typeof(object); - }); - - var deserializeMethod = serializerType - .GetMethods(BindingFlags.Public | BindingFlags.Static) - .FirstOrDefault(m => - { - if (m.Name != "Deserialize") - return false; - var parameters = m.GetParameters(); - return parameters.Length >= 2 && - parameters[0].ParameterType == typeof(Type) && - (parameters[1].ParameterType == typeof(byte[]) || - parameters[1].ParameterType == typeof(ReadOnlyMemory)); - }); - - if (serializeMethod == null || deserializeMethod == null) - throw new MissingMethodException("MessagePack serialize/deserialize API not found."); - - _serializeMethod = serializeMethod; - _deserializeMethod = deserializeMethod; - } - - public byte[] Serialize(object value) - { - if (value == null) - return Array.Empty(); - - var parameters = BuildParameters(_serializeMethod, value.GetType(), value, _serializerOptions); - return (byte[])_serializeMethod.Invoke(null, parameters)!; - } - - public object Deserialize(byte[] payload, Type type) - { - payload ??= Array.Empty(); - var parameters = BuildParameters(_deserializeMethod, type, payload, _serializerOptions); - return _deserializeMethod.Invoke(null, parameters) - ?? throw new InvalidOperationException($"MessagePack returned null for type {type.FullName}."); - } - - public T Deserialize(byte[] payload) - { - return (T)Deserialize(payload, typeof(T)); - } - - private static object?[] BuildParameters(MethodInfo method, Type type, object valueOrBytes, object? serializerOptions) - { - var parameters = method.GetParameters(); - var args = new object?[parameters.Length]; - - if (parameters.Length > 0) - args[0] = type; - if (parameters.Length > 1) - args[1] = ConvertPrimaryArgument(parameters[1].ParameterType, valueOrBytes); - - for (var i = 2; i < parameters.Length; i++) - { - args[i] = ResolveAdditionalArgument(parameters[i], serializerOptions); - } - - return args; - } - - private static object? ResolveAdditionalArgument(ParameterInfo parameter, object? serializerOptions) - { - if (serializerOptions != null && parameter.ParameterType.IsInstanceOfType(serializerOptions)) - return serializerOptions; - - return parameter.HasDefaultValue - ? parameter.DefaultValue - : GetDefault(parameter.ParameterType); - } - - private static object ConvertPrimaryArgument(Type parameterType, object value) - { - if (parameterType == typeof(ReadOnlyMemory) && value is byte[] bytes) - return new ReadOnlyMemory(bytes); - - return value; - } - - private static object? ResolveSerializerOptions(Assembly serializerAssembly) - { - var contractlessResolverType = serializerAssembly.GetType("MessagePack.Resolvers.ContractlessStandardResolver"); - if (contractlessResolverType != null) - { - var optionsField = contractlessResolverType.GetField("Options", - BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Static); - var options = optionsField?.GetValue(null); - if (options != null) - return options; - - var instanceField = contractlessResolverType.GetField("Instance", - BindingFlags.Public | BindingFlags.NonPublic | BindingFlags.Static); - var instance = instanceField?.GetValue(null); - if (instance != null) - { - var optionsType = serializerAssembly.GetType("MessagePack.MessagePackSerializerOptions"); - var standardProperty = optionsType?.GetProperty("Standard", BindingFlags.Public | BindingFlags.Static); - var standardOptions = standardProperty?.GetValue(null); - var withResolverMethod = optionsType?.GetMethod("WithResolver", BindingFlags.Public | BindingFlags.Instance); - var resolvedOptions = withResolverMethod?.Invoke(standardOptions, new[] { instance }); - if (resolvedOptions != null) - return resolvedOptions; - } - } - - return null; - } - - private static object? GetDefault(Type type) - { - return type.IsValueType ? Activator.CreateInstance(type) : null; - } - } -} diff --git a/Runtime/Serialization/ShrinkPacketCodec.cs b/Runtime/Serialization/ShrinkPacketCodec.cs new file mode 100644 index 0000000..7668ca0 --- /dev/null +++ b/Runtime/Serialization/ShrinkPacketCodec.cs @@ -0,0 +1,105 @@ +#nullable enable +using System; +using System.Buffers.Binary; +using System.IO; +using System.Text; + +namespace ShrinkNetwork +{ + public sealed class ShrinkProtocolException : IOException + { + public ShrinkProtocolException(string message) : base(message) { } + public ShrinkProtocolException(string message, Exception inner) : base(message, inner) { } + } + + /// V2 little-endian framing, independent of the payload serializer. Decode borrows input memory. + public static class ShrinkPacketCodec + { + private const uint Magic = 0x324B4853; // SHK2 + public const int HeaderSize = 33; + public const int MaximumPacketBytes = 16 * 1024 * 1024; + private static readonly UTF8Encoding Utf8 = new(false, true); + + public static ShrinkBufferWriter Encode(ShrinkNetworkPacket packet, IShrinkNetworkSerializer? serializer = null, object? message = null) + { + var route = packet.Route ?? string.Empty; + var token = packet.SessionToken ?? string.Empty; + var routeBytes = Utf8.GetByteCount(route); + var tokenBytes = Utf8.GetByteCount(token); + if (routeBytes > ushort.MaxValue || tokenBytes > ushort.MaxValue) throw new ShrinkProtocolException("Route or session token exceeds framing limit."); + if (packet.ProtocolVersion != ShrinkNetworkProtocol.CurrentProtocolVersion || packet.SchemaVersion is < 0 or > ushort.MaxValue || packet.Kind < 0 || packet.Kind > ShrinkNetworkPacketKind.Response) + throw new ShrinkProtocolException("Invalid packet header."); + var writer = new ShrinkBufferWriter(HeaderSize + routeBytes + tokenBytes); + try + { + var header = writer.GetSpan(HeaderSize + routeBytes + tokenBytes); + BinaryPrimitives.WriteUInt32LittleEndian(header, Magic); + BinaryPrimitives.WriteUInt16LittleEndian(header.Slice(4), (ushort)packet.ProtocolVersion); + BinaryPrimitives.WriteUInt16LittleEndian(header.Slice(6), (ushort)packet.SchemaVersion); + BinaryPrimitives.WriteInt32LittleEndian(header.Slice(8), packet.Opcode); + BinaryPrimitives.WriteInt32LittleEndian(header.Slice(12), packet.RequestToken.Value); + header[16] = (byte)packet.Kind; + BinaryPrimitives.WriteInt64LittleEndian(header.Slice(17), packet.SessionTokenExpiresAtUnixTimeSeconds); + BinaryPrimitives.WriteUInt16LittleEndian(header.Slice(25), (ushort)routeBytes); + BinaryPrimitives.WriteUInt16LittleEndian(header.Slice(27), (ushort)tokenBytes); + Utf8.GetBytes(route.AsSpan(), header.Slice(HeaderSize, routeBytes)); + Utf8.GetBytes(token.AsSpan(), header.Slice(HeaderSize + routeBytes, tokenBytes)); + writer.Advance(HeaderSize + routeBytes + tokenBytes); + var payloadStart = writer.WrittenCount; + if (message != null) + { + if (serializer is IShrinkNetworkBufferSerializer buffered) buffered.Serialize(writer, message); + else + { + var bytes = (serializer ?? throw new ArgumentNullException(nameof(serializer))).Serialize(message); + bytes.CopyTo(writer.GetSpan(bytes.Length)); writer.Advance(bytes.Length); + } + } + else + { + packet.Payload.Span.CopyTo(writer.GetSpan(packet.Payload.Length)); + writer.Advance(packet.Payload.Length); + } + if (writer.WrittenCount > MaximumPacketBytes) throw new ShrinkProtocolException("Packet exceeds framing limit."); + BinaryPrimitives.WriteInt32LittleEndian(writer.WrittenSpan.Slice(29), writer.WrittenCount - payloadStart); + return writer; + } + catch { writer.Dispose(); throw; } + } + + internal static bool IsResponse(ReadOnlyMemory memory) => memory.Length >= HeaderSize && + BinaryPrimitives.ReadUInt32LittleEndian(memory.Span) == Magic && memory.Span[16] == (byte)ShrinkNetworkPacketKind.Response; + + public static ShrinkNetworkPacket Decode(ReadOnlyMemory memory) + { + var span = memory.Span; + if (span.Length < HeaderSize || span.Length > MaximumPacketBytes || BinaryPrimitives.ReadUInt32LittleEndian(span) != Magic) + throw new ShrinkProtocolException("Expected ShrinkNetwork protocol v2 binary envelope. Legacy JSON/MessagePack envelopes are unsupported."); + var protocol = BinaryPrimitives.ReadUInt16LittleEndian(span.Slice(4)); + if (protocol != ShrinkNetworkProtocol.CurrentProtocolVersion) throw new ShrinkProtocolException("Unsupported protocol version " + protocol); + var routeLength = BinaryPrimitives.ReadUInt16LittleEndian(span.Slice(25)); + var tokenLength = BinaryPrimitives.ReadUInt16LittleEndian(span.Slice(27)); + var payloadLength = BinaryPrimitives.ReadInt32LittleEndian(span.Slice(29)); + var start = HeaderSize + routeLength + tokenLength; + if (payloadLength < 0 || start > span.Length || payloadLength != span.Length - start || span[16] > (byte)ShrinkNetworkPacketKind.Response) + throw new ShrinkProtocolException("Invalid packet lengths or kind."); + string route, token; + try + { + route = Utf8.GetString(span.Slice(HeaderSize, routeLength)); + token = Utf8.GetString(span.Slice(HeaderSize + routeLength, tokenLength)); + } + catch (DecoderFallbackException exception) + { + throw new ShrinkProtocolException("Invalid UTF-8 in packet header.", exception); + } + return new ShrinkNetworkPacket { + ProtocolVersion = protocol, SchemaVersion = BinaryPrimitives.ReadUInt16LittleEndian(span.Slice(6)), + Opcode = BinaryPrimitives.ReadInt32LittleEndian(span.Slice(8)), RequestToken = new ShrinkRequestToken(BinaryPrimitives.ReadInt32LittleEndian(span.Slice(12))), + Kind = (ShrinkNetworkPacketKind)span[16], SessionTokenExpiresAtUnixTimeSeconds = BinaryPrimitives.ReadInt64LittleEndian(span.Slice(17)), + Route = route, SessionToken = token, + Payload = memory.Slice(start, payloadLength) + }; + } + } +} diff --git a/Runtime/Serialization/ShrinkPacketCodec.cs.meta b/Runtime/Serialization/ShrinkPacketCodec.cs.meta new file mode 100644 index 0000000..0e95599 --- /dev/null +++ b/Runtime/Serialization/ShrinkPacketCodec.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: 8a01225f12864f142b0d0c6395ef381b +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Runtime/Serialization/ShrinkRegisteredNetworkSerializer.cs b/Runtime/Serialization/ShrinkRegisteredNetworkSerializer.cs new file mode 100644 index 0000000..008215c --- /dev/null +++ b/Runtime/Serialization/ShrinkRegisteredNetworkSerializer.cs @@ -0,0 +1,49 @@ +#nullable enable +using System; +using System.Buffers; +using System.Collections.Generic; + +namespace ShrinkNetwork +{ + public interface IShrinkMessageCodec + { + void Write(IBufferWriter writer, T value); + T Read(ReadOnlyMemory payload); + } + + /// Explicit codecs work with AOT and external modules without runtime generic reflection. + public class ShrinkRegisteredNetworkSerializer : IShrinkNetworkBufferSerializer + { + private interface ICodec + { + void Write(IBufferWriter writer, object value); + object Read(ReadOnlyMemory payload); + } + private sealed class Codec : ICodec + { + private readonly IShrinkMessageCodec _codec; + public Codec(IShrinkMessageCodec codec) => _codec = codec; + public void Write(IBufferWriter writer, object value) => _codec.Write(writer, (T)value); + public object Read(ReadOnlyMemory payload) => _codec.Read(payload)!; + } + private readonly Dictionary _codecs = new(); + public void Register(IShrinkMessageCodec codec) + { + if (codec == null) throw new ArgumentNullException(nameof(codec)); + _codecs.Add(typeof(T), new Codec(codec)); + } + public bool Unregister() => _codecs.Remove(typeof(T)); + private ICodec Resolve(Type type) => _codecs.TryGetValue(type, out var codec) ? codec : + throw new InvalidOperationException($"SHRINK-NET-CODEC: No codec registered for {type.FullName}. Register a generated formatter before binding the transport."); + public void Serialize(IBufferWriter writer, object value) => Resolve(value.GetType()).Write(writer, value); + public object Deserialize(ReadOnlyMemory payload, Type type) => Resolve(type).Read(payload); + public byte[] Serialize(object value) + { + using var writer = new ShrinkBufferWriter(); + Serialize(writer, value); + return writer.WrittenMemory.ToArray(); + } + public object Deserialize(byte[] payload, Type type) => Deserialize((ReadOnlyMemory)payload, type); + public T Deserialize(byte[] payload) => (T)Deserialize(payload, typeof(T)); + } +} diff --git a/Runtime/Serialization/ShrinkRegisteredNetworkSerializer.cs.meta b/Runtime/Serialization/ShrinkRegisteredNetworkSerializer.cs.meta new file mode 100644 index 0000000..4e5a353 --- /dev/null +++ b/Runtime/Serialization/ShrinkRegisteredNetworkSerializer.cs.meta @@ -0,0 +1,11 @@ +fileFormatVersion: 2 +guid: b0225bad560778f43827ddefb85fd61c +MonoImporter: + externalObjects: {} + serializedVersion: 2 + defaultReferences: [] + executionOrder: 0 + icon: {instanceID: 0} + userData: + assetBundleName: + assetBundleVariant: diff --git a/Runtime/Transport/Abstractions/IShrinkNetworkAsyncTransport.cs b/Runtime/Transport/Abstractions/IShrinkNetworkAsyncTransport.cs index f10e203..bba1f3f 100644 --- a/Runtime/Transport/Abstractions/IShrinkNetworkAsyncTransport.cs +++ b/Runtime/Transport/Abstractions/IShrinkNetworkAsyncTransport.cs @@ -1,4 +1,6 @@ using Cysharp.Threading.Tasks; +using System; +using System.Threading; namespace ShrinkNetwork { @@ -6,4 +8,10 @@ namespace ShrinkNetwork { UniTask SendAsync(long sessionId, byte[] packetData); } + + /// Borrowed memory remains owned by caller until SendAsync completes, faults or cancels. + public interface IShrinkNetworkMemoryTransport : IShrinkNetworkAsyncTransport + { + UniTask SendAsync(long sessionId, ReadOnlyMemory packetData, CancellationToken cancellationToken = default); + } } diff --git a/Runtime/Transport/Kcp/ShrinkKcpServerTransport.cs b/Runtime/Transport/Kcp/ShrinkKcpServerTransport.cs index 87c482e..4f96926 100644 --- a/Runtime/Transport/Kcp/ShrinkKcpServerTransport.cs +++ b/Runtime/Transport/Kcp/ShrinkKcpServerTransport.cs @@ -163,9 +163,9 @@ namespace ShrinkNetwork { while (!cancellationToken.IsCancellationRequested) { - var sessions = _sessions.Values.ToArray(); - foreach (var session in sessions) + foreach (var pair in _sessions) { + var session = pair.Value; var shouldDisconnect = false; lock (session.SyncRoot) { diff --git a/Runtime/Transport/Loopback/ShrinkLoopbackTransport.cs b/Runtime/Transport/Loopback/ShrinkLoopbackTransport.cs index 256b70b..15ee990 100644 --- a/Runtime/Transport/Loopback/ShrinkLoopbackTransport.cs +++ b/Runtime/Transport/Loopback/ShrinkLoopbackTransport.cs @@ -1,11 +1,12 @@ #nullable enable using System; using System.Collections.Generic; +using System.Threading; using Cysharp.Threading.Tasks; namespace ShrinkNetwork { - public sealed class ShrinkLoopbackTransport : IShrinkNetworkAsyncTransport, IShrinkNetworkSessionControlTransport + public sealed class ShrinkLoopbackTransport : IShrinkNetworkMemoryTransport, IShrinkNetworkSessionControlTransport { private readonly HashSet _openedSessions = new(); private ShrinkLoopbackTransport? _peer; @@ -73,5 +74,12 @@ namespace ShrinkNetwork _peer.OnEvent?.Invoke(ShrinkNetworkTransportEvent.Packet(sessionId, packetData)); return UniTask.CompletedTask; } + + public UniTask SendAsync(long sessionId, ReadOnlyMemory packetData, CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + // Receivers may enqueue the event beyond this call; transfer a dedicated copy. + return SendAsync(sessionId, packetData.ToArray()); + } } } diff --git a/Runtime/Transport/Tcp/ShrinkTcpClientTransport.cs b/Runtime/Transport/Tcp/ShrinkTcpClientTransport.cs index 896eb0e..1eb1e34 100644 --- a/Runtime/Transport/Tcp/ShrinkTcpClientTransport.cs +++ b/Runtime/Transport/Tcp/ShrinkTcpClientTransport.cs @@ -1,6 +1,7 @@ #nullable enable using System; using System.Buffers.Binary; +using System.Buffers; using System.IO; using System.Net.Sockets; using System.Net.Security; @@ -10,7 +11,7 @@ using Cysharp.Threading.Tasks; namespace ShrinkNetwork { - public sealed class ShrinkTcpClientTransport : IShrinkNetworkAsyncTransport + public sealed class ShrinkTcpClientTransport : IShrinkNetworkMemoryTransport { private readonly string _host; private readonly int _port; @@ -86,7 +87,9 @@ namespace ShrinkNetwork SendAsync(sessionId, packetData).Forget(); } - public UniTask SendAsync(long sessionId, byte[] packetData) + public UniTask SendAsync(long sessionId, byte[] packetData) => SendAsync(sessionId, (ReadOnlyMemory)(packetData ?? Array.Empty())); + + public UniTask SendAsync(long sessionId, ReadOnlyMemory packetData, CancellationToken cancellationToken = default) { if (!IsStarted) throw new InvalidOperationException("Transport is not started."); @@ -94,11 +97,11 @@ namespace ShrinkNetwork throw new InvalidOperationException($"Unsupported session id {sessionId}. This transport only supports {_sessionId}."); if (_client == null || !_client.Connected || _stream == null) throw new InvalidOperationException("TCP client is not connected."); - if ((packetData?.Length ?? 0) > _maxPacketSize) - throw new InvalidOperationException($"TCP packet is too large. Size={(packetData?.Length ?? 0)}, Limit={_maxPacketSize}."); + if (packetData.Length > _maxPacketSize) + throw new InvalidOperationException($"TCP packet is too large. Size={packetData.Length}, Limit={_maxPacketSize}."); - return SendInternalAsync(_stream, _sendLock, packetData ?? Array.Empty(), - _cts?.Token ?? CancellationToken.None); + return SendInternalAsync(_stream, _sendLock, packetData, + cancellationToken.CanBeCanceled ? cancellationToken : _cts?.Token ?? CancellationToken.None); } private async UniTaskVoid ConnectAsync(CancellationToken cancellationToken) @@ -209,18 +212,20 @@ namespace ShrinkNetwork ex.SocketErrorCode == SocketError.Shutdown; } - private static async UniTask SendInternalAsync(Stream stream, SemaphoreSlim sendLock, byte[] packetData, + private static async UniTask SendInternalAsync(Stream stream, SemaphoreSlim sendLock, ReadOnlyMemory packetData, CancellationToken cancellationToken) { await sendLock.WaitAsync(cancellationToken); try { - var header = new byte[4]; - BinaryPrimitives.WriteInt32LittleEndian(header, packetData.Length); - await stream.WriteAsync(header, cancellationToken); - if (packetData.Length > 0) - await stream.WriteAsync(packetData, cancellationToken); - await stream.FlushAsync(cancellationToken); + var frame = ArrayPool.Shared.Rent(checked(4 + packetData.Length)); + try + { + BinaryPrimitives.WriteInt32LittleEndian(frame, packetData.Length); + packetData.CopyTo(frame.AsMemory(4)); + await stream.WriteAsync(frame.AsMemory(0, 4 + packetData.Length), cancellationToken); + } + finally { ArrayPool.Shared.Return(frame, clearArray: true); } } finally { diff --git a/Runtime/Transport/Tcp/ShrinkTcpServerTransport.cs b/Runtime/Transport/Tcp/ShrinkTcpServerTransport.cs index 8ca183a..3b1bfa2 100644 --- a/Runtime/Transport/Tcp/ShrinkTcpServerTransport.cs +++ b/Runtime/Transport/Tcp/ShrinkTcpServerTransport.cs @@ -1,6 +1,7 @@ #nullable enable using System; using System.Buffers.Binary; +using System.Buffers; using System.Collections.Concurrent; using System.IO; using System.Net; @@ -13,7 +14,7 @@ using Cysharp.Threading.Tasks; namespace ShrinkNetwork { - public sealed class ShrinkTcpServerTransport : IShrinkNetworkAsyncTransport, IShrinkNetworkSessionControlTransport + public sealed class ShrinkTcpServerTransport : IShrinkNetworkMemoryTransport, IShrinkNetworkSessionControlTransport { private readonly ConcurrentDictionary _clients = new(); private readonly ConcurrentDictionary _sendLocks = new(); @@ -92,11 +93,8 @@ namespace ShrinkNetwork _streams.Clear(); - foreach (var pair in _sendLocks) - { - pair.Value.Dispose(); - } - + // In-flight writers still release these managed semaphores after the stream closes. + // No WaitHandle is allocated; let their final users release and then collect them. _sendLocks.Clear(); } @@ -125,7 +123,9 @@ namespace ShrinkNetwork return true; } - public async UniTask SendAsync(long sessionId, byte[] packetData) + public UniTask SendAsync(long sessionId, byte[] packetData) => SendAsync(sessionId, (ReadOnlyMemory)(packetData ?? Array.Empty())); + + public async UniTask SendAsync(long sessionId, ReadOnlyMemory packetData, CancellationToken cancellationToken = default) { if (!_clients.TryGetValue(sessionId, out var client)) throw new InvalidOperationException($"Session {sessionId} is not connected."); @@ -134,10 +134,10 @@ namespace ShrinkNetwork if (!_sendLocks.TryGetValue(sessionId, out var sendLock)) throw new InvalidOperationException($"Session {sessionId} send lock is not initialized."); - if ((packetData?.Length ?? 0) > _maxPacketSize) - throw new InvalidOperationException($"TCP packet is too large. Size={(packetData?.Length ?? 0)}, Limit={_maxPacketSize}."); + if (packetData.Length > _maxPacketSize) + throw new InvalidOperationException($"TCP packet is too large. Size={packetData.Length}, Limit={_maxPacketSize}."); - await SendInternalAsync(stream, sendLock, packetData ?? Array.Empty(), CancellationToken.None); + await SendInternalAsync(stream, sendLock, packetData, cancellationToken); } private async Task AcceptLoopAsync(CancellationToken cancellationToken) @@ -244,8 +244,7 @@ namespace ShrinkNetwork { if (_streams.TryRemove(sessionId, out var ownedStream)) ownedStream.Dispose(); - if (_sendLocks.TryRemove(sessionId, out var sendLock)) - sendLock.Dispose(); + _sendLocks.TryRemove(sessionId, out _); var remoteAddress = removed.Client.RemoteEndPoint?.ToString() ?? "unknown"; try @@ -261,18 +260,20 @@ namespace ShrinkNetwork } } - private static async Task SendInternalAsync(Stream stream, SemaphoreSlim sendLock, byte[] packetData, + private static async Task SendInternalAsync(Stream stream, SemaphoreSlim sendLock, ReadOnlyMemory packetData, CancellationToken cancellationToken) { await sendLock.WaitAsync(cancellationToken); try { - var header = new byte[4]; - BinaryPrimitives.WriteInt32LittleEndian(header, packetData.Length); - await stream.WriteAsync(header, cancellationToken); - if (packetData.Length > 0) - await stream.WriteAsync(packetData, cancellationToken); - await stream.FlushAsync(cancellationToken); + var frame = ArrayPool.Shared.Rent(checked(4 + packetData.Length)); + try + { + BinaryPrimitives.WriteInt32LittleEndian(frame, packetData.Length); + packetData.CopyTo(frame.AsMemory(4)); + await stream.WriteAsync(frame.AsMemory(0, 4 + packetData.Length), cancellationToken); + } + finally { ArrayPool.Shared.Return(frame, clearArray: true); } } finally { diff --git a/Samples/PingClient/ShrinkNetworkPingClientExample.cs b/Samples/PingClient/ShrinkNetworkPingClientExample.cs index 020476f..fefc1f6 100644 --- a/Samples/PingClient/ShrinkNetworkPingClientExample.cs +++ b/Samples/PingClient/ShrinkNetworkPingClientExample.cs @@ -59,7 +59,7 @@ public sealed class ShrinkNetworkPingClientExample : MonoBehaviour Disconnect(); _service = new ShrinkNetworkService( - new ShrinkMessagePackNetworkSerializer(), + new ShrinkJsonNetworkSerializer(), new ShrinkNetworkMessageRegistry(), new ShrinkNetworkRouter()); _service.AutoRegisterAttributedMessages(); diff --git a/package.json b/package.json index c7a4575..c0a8a24 100644 --- a/package.json +++ b/package.json @@ -1,11 +1,11 @@ { "name": "com.cneicy.shrink-network", - "version": "0.3.0", + "version": "0.4.2", "displayName": "ShrinkNetwork", "description": "面向 Unity 的轻量网络框架，提供会话、消息注册、RPC、权限控制，以及 TCP/KCP 传输抽象。", "unity": "2022.3", "dependencies": { - "com.cneicy.shrink-shared-codegen": "0.1.1", + "com.cneicy.shrink-shared-codegen": "0.2.1", "com.cysharp.unitask": "2.5.10", "com.unity.nuget.newtonsoft-json": "3.2.2" },