Skip to content

Commit 850266b

Browse files
NickKhalowmikhail-dclAnsisMalins
authored
Opti: zero copy EngineAPI and SendBinary (#6456)
* avoid unnessasary buffer copying from Js * comment * extract UmanagedMemoryManager to util * add comments * update tests * update asmdef * remove obsolete method * PoolableByteArray.cs example of usage and avoid when async is not needed * zero copy invocation * last input for type casts * Update Explorer/Assets/DCL/Infrastructure/Utility/Memory/SingleUnmanagedMemoryManager.cs Co-authored-by: Mikhail Agapov <118179774+mikhail-dcl@users.noreply.github.com> Signed-off-by: Nick Khalow <71646502+NickKhalow@users.noreply.github.com> * apply format * format * benchmark * encoded message struct * native alloc Encoding.UTF8.GetBytes(data) * max size check * compile directives * array.InvokeWithDirectAccess alloc free * compilation fixes * add copying for buffer * remove boxing --------- Signed-off-by: Nick Khalow <71646502+NickKhalow@users.noreply.github.com> Co-authored-by: Mikhail Agapov <118179774+mikhail-dcl@users.noreply.github.com> Co-authored-by: Ansis Māliņš <ansis.malins@decentraland.org>
1 parent 71c4bb1 commit 850266b

29 files changed

Lines changed: 760 additions & 201 deletions

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/CRDT.ECS.Bridge.asmdef‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
],
1616
"includePlatforms": [],
1717
"excludePlatforms": [],
18-
"allowUnsafeCode": false,
18+
"allowUnsafeCode": true,
1919
"overrideReferences": false,
2020
"precompiledReferences": [],
2121
"autoReferenced": true,

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/ClientWebSocketApiImplementation.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ public async UniTask SendBinaryAsync(int websocketId, IArrayBuffer data, ulong s
6868

6969
if (size == 0) return;
7070

71+
// Usecase is justified.
72+
// InvokeWithDirectAccess<TArg, TResult>(Func<IntPtr, TArg, TResult>, TArg) doesn't support async.
73+
// it's unsafe to keep the pointer after its scope.
7174
using PoolableByteArray poolableArray = instancePoolsProvider.GetAPIRawDataPool((int)size);
7275

7376
data.ReadBytes(0, size, poolableArray.Array, 0);

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/Communications/CommunicationsControllerAPIImplementation.cs‎

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -30,22 +30,25 @@ protected override void OnMessageReceived(ISceneCommunicationPipe.DecodedMessage
3030
{
3131
var data = (IntPtr)dataPtr;
3232

33-
array.InvokeWithDirectAccess(buffer =>
34-
{
35-
var bufferPtr = (byte*)buffer;
33+
dataOffset = array.InvokeWithDirectAccess(static (buffer, args) =>
34+
{
35+
byte* bufferPtr = (byte*)buffer;
3636

37-
fixed (char* walletIdPtr = walletId)
38-
bufferPtr[0] = (byte)Encoding.UTF8.GetBytes(walletIdPtr, walletId.Length,
39-
bufferPtr + 1, byte.MaxValue);
37+
fixed (char* walletIdPtr = args.walletId)
38+
bufferPtr[0] = (byte)Encoding.UTF8.GetBytes(walletIdPtr, args.walletId.Length,
39+
bufferPtr + 1, byte.MaxValue);
4040

41-
dataOffset = bufferPtr[0] + 1;
41+
var dataOffsetScoped = bufferPtr[0] + 1;
4242

43-
if (dataOffset + dataLength > IJsOperations.LIVEKIT_MAX_SIZE)
44-
throw new InternalBufferOverflowException(
45-
"Received a message larger than LIVEKIT_MAX_SIZE");
43+
if (dataOffsetScoped + args.dataLength > IJsOperations.LIVEKIT_MAX_SIZE)
44+
throw new InternalBufferOverflowException(
45+
"Received a message larger than LIVEKIT_MAX_SIZE");
4646

47-
UnsafeUtility.MemCpy(bufferPtr + dataOffset, (byte*)data, dataLength);
48-
});
47+
UnsafeUtility.MemCpy(bufferPtr + dataOffsetScoped, (byte*)args.data, args.dataLength);
48+
return dataOffsetScoped;
49+
},
50+
(data, dataLength, walletId)
51+
);
4952
}
5053
}
5154

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/Communications/CommunicationsControllerAPIImplementationBase.cs‎

Lines changed: 103 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,11 @@
66
using SceneRuntime;
77
using SceneRuntime.Apis.Modules.CommunicationsControllerApi;
88
using System;
9+
using System.IO;
910
using System.Collections.Generic;
1011
using System.Threading;
1112
using Utility;
13+
using DCL.Diagnostics;
1214

1315
namespace CrdtEcsBridge.JsModulesImplementation.Communications
1416
{
@@ -57,40 +59,62 @@ public void Dispose()
5759
cancellationTokenSource.SafeCancelAndDispose();
5860
}
5961

60-
public void SendBinary(IReadOnlyList<PoolableByteArray> broadcastData, string? recipient = null)
62+
public void SendBinary<TEnumerable, TArray>(TEnumerable broadcastData, string? recipient = null)
63+
where TEnumerable : IEnumerable<TArray>
64+
where TArray : IPoolableByteArray
6165
{
6266
// Authoritative multiplayer enforces sending messages to the special peer
6367
if (sceneData.SceneEntityDefinition.metadata.authoritativeMultiplayer)
6468
recipient = "authoritative-server";
6569

66-
foreach (PoolableByteArray poolable in broadcastData)
67-
if (poolable.Length > 0)
70+
foreach (TArray data in broadcastData)
71+
if (data.Length > 0)
6872
{
69-
byte firstByte = poolable.Span[0];
73+
int length = (int) data.Length;
74+
var instance = this;
7075

71-
ISceneCommunicationPipe.ConnectivityAssertiveness assertiveness = firstByte == (int)CommsMessageType.REQ_CRDT_STATE
72-
? ISceneCommunicationPipe.ConnectivityAssertiveness.DELIVERY_ASSERTED
73-
: ISceneCommunicationPipe.ConnectivityAssertiveness.DROP_IF_NOT_CONNECTED;
74-
75-
// Filter CRDT messages before sending
76-
if (firstByte == (int)CommsMessageType.CRDT)
76+
int encodedLength = EncodedMessage.LengthWithReservedByte(length);
77+
if (encodedLength > IJsOperations.LIVEKIT_MAX_SIZE)
7778
{
78-
Span<byte> filtered = stackalloc byte[poolable.Memory.Span.Length];
79-
int filteredLength = FilterCRDTMessage(poolable.Memory.Span, filtered);
80-
EncodeAndSendMessage(ISceneCommunicationPipe.MsgType.Uint8Array, filtered.Slice(0, filteredLength), assertiveness, recipient);
79+
ReportHub.LogException(new InternalBufferOverflowException("Tried to encode a message larger than LIVEKIT_MAX_SIZE"), ReportCategory.CRDT_ECS_BRIDGE);
8180
continue;
8281
}
8382

84-
// Filter RES_CRDT_STATE messages before sending
85-
if (firstByte == (int)CommsMessageType.RES_CRDT_STATE)
86-
{
87-
Span<byte> filtered = stackalloc byte[poolable.Memory.Span.Length];
88-
int filteredLength = FilterCRDTStateMessage(poolable.Memory.Span, filtered);
89-
EncodeAndSendMessage(ISceneCommunicationPipe.MsgType.Uint8Array, filtered.Slice(0, filteredLength), assertiveness, recipient);
90-
continue;
91-
}
9283

93-
EncodeAndSendMessage(ISceneCommunicationPipe.MsgType.Uint8Array, poolable.Memory.Span, assertiveness, recipient);
84+
data.InvokeWithDirectAccess(
85+
static (ptr, args) => {
86+
Span<byte> span;
87+
unsafe
88+
{
89+
span = new Span<byte>(ptr.ToPointer(), args.length);
90+
}
91+
92+
byte firstByte = span[0];
93+
94+
ISceneCommunicationPipe.ConnectivityAssertiveness assertiveness = firstByte == (int)CommsMessageType.REQ_CRDT_STATE
95+
? ISceneCommunicationPipe.ConnectivityAssertiveness.DELIVERY_ASSERTED
96+
: ISceneCommunicationPipe.ConnectivityAssertiveness.DROP_IF_NOT_CONNECTED;
97+
98+
int length = EncodedMessage.LengthWithReservedByte(span.Length);
99+
// Considered save to stackalloc, it's checked the load cannot exceed LIVEKIT_MAX_SIZE
100+
Span<byte> contentAlloc = stackalloc byte[length];
101+
EncodedMessage encodedMessage = new EncodedMessage(contentAlloc);
102+
encodedMessage.AssignType(ISceneCommunicationPipe.MsgType.Uint8Array);
103+
Span<byte> contentPtr = encodedMessage.Content();
104+
UnityEngine.Assertions.Assert.AreEqual(span.Length, contentPtr.Length);
105+
span.CopyTo(contentPtr);
106+
107+
// Filter CRDT messages before sending
108+
if (firstByte == (int)CommsMessageType.CRDT)
109+
encodedMessage.FilterBy(CRDTFilter.FilterSceneMessageBatch, span);
110+
// Filter RES_CRDT_STATE messages before sending
111+
else if (firstByte == (int)CommsMessageType.RES_CRDT_STATE)
112+
encodedMessage.FilterBy(CRDTFilter.FilterCRDTState, span);
113+
114+
args.instance.EncodeAndSendMessage(encodedMessage, assertiveness, args.recipient);
115+
},
116+
(instance, recipient, length)
117+
);
94118
}
95119
}
96120

@@ -108,27 +132,69 @@ public ScriptObject GetResult()
108132
}
109133
}
110134

111-
private static int FilterCRDTMessage(ReadOnlySpan<byte> message, Span<byte> output)
135+
protected void EncodeAndSendMessage(EncodedMessage encodedMessage, ISceneCommunicationPipe.ConnectivityAssertiveness assertivenes, string? specialRecipient)
112136
{
113-
CRDTFilter.FilterSceneMessageBatch(message, output, out int totalWrite);
114-
return totalWrite;
137+
sceneCommunicationPipe.SendMessage(encodedMessage.ContentWithHeader(), sceneId, assertivenes, cancellationTokenSource.Token, specialRecipient);
115138
}
116139

117-
private static int FilterCRDTStateMessage(ReadOnlySpan<byte> message, Span<byte> output)
118-
{
119-
CRDTFilter.FilterCRDTState(message, output, out int totalWrite);
120-
return totalWrite;
121-
}
140+
protected abstract void OnMessageReceived(ISceneCommunicationPipe.DecodedMessage decodedMessage);
141+
142+
143+
public delegate void SpanFilter(ReadOnlySpan<byte> src, Span<byte> dst, out int writtenBytes);
122144

123-
protected void EncodeAndSendMessage(ISceneCommunicationPipe.MsgType msgType, ReadOnlySpan<byte> message, ISceneCommunicationPipe.ConnectivityAssertiveness assertivenes, string? specialRecipient)
145+
protected ref struct EncodedMessage
124146
{
125-
Span<byte> encodedMessage = stackalloc byte[message.Length + 1];
126-
encodedMessage[0] = (byte)msgType;
127-
message.CopyTo(encodedMessage[1..]);
147+
public const int RESERVED_SIZE = 1;
128148

129-
sceneCommunicationPipe.SendMessage(encodedMessage, sceneId, assertivenes, cancellationTokenSource.Token, specialRecipient);
130-
}
149+
private Span<byte> data;
150+
private int contentLength;
131151

132-
protected abstract void OnMessageReceived(ISceneCommunicationPipe.DecodedMessage decodedMessage);
152+
// Be sure to reserve 1 byte for the msg type
153+
public EncodedMessage(Span<byte> data)
154+
{
155+
this.data = data; // Extra byte for message type
156+
contentLength = data.Length - RESERVED_SIZE;
157+
}
158+
159+
public static int LengthWithReservedByte(int length)
160+
{
161+
return length + RESERVED_SIZE;
162+
}
163+
164+
public Span<byte> Content()
165+
{
166+
return data.Slice(RESERVED_SIZE, contentLength); // first byte is msg type
167+
}
168+
169+
public Span<byte> ContentWithHeader()
170+
{
171+
return data.Slice(0, contentLength + RESERVED_SIZE); // first byte is msg type
172+
}
173+
174+
public void ResizeContent(int length)
175+
{
176+
int targetTotalLength = length + RESERVED_SIZE;
177+
UnityEngine.Assertions.Assert.IsFalse(
178+
targetTotalLength > data.Length,
179+
"Cannot resize to target length greater than the origin span"
180+
);
181+
182+
contentLength = length;
183+
}
184+
185+
public void FilterBy(SpanFilter filter, ReadOnlySpan<byte> src)
186+
{
187+
Span<byte> content = Content();
188+
filter(src, content, out int newContentLength);
189+
ResizeContent(newContentLength);
190+
}
191+
192+
public void AssignType(ISceneCommunicationPipe.MsgType msgType)
193+
{
194+
data[0] = (byte)msgType;
195+
}
196+
}
133197
}
134198
}
199+
200+

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/Communications/SDKMessageBus/SDKMessageBusCommsAPIImplementation.cs‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
using System;
66
using System.Collections.Generic;
77
using System.Text;
8+
using Unity.Collections;
89

910
namespace CrdtEcsBridge.JsModulesImplementation.Communications.SDKMessageBus
1011
{
@@ -24,8 +25,17 @@ public void ClearMessages()
2425

2526
public void Send(string data)
2627
{
27-
byte[] dataBytes = Encoding.UTF8.GetBytes(data);
28-
EncodeAndSendMessage(ISceneCommunicationPipe.MsgType.String, dataBytes, ISceneCommunicationPipe.ConnectivityAssertiveness.DROP_IF_NOT_CONNECTED, null);
28+
int byteCount = Encoding.UTF8.GetByteCount(data);
29+
int length = EncodedMessage.LengthWithReservedByte(byteCount);
30+
using NativeArray<byte> dataBytes = new NativeArray<byte>(length, Allocator.Temp, NativeArrayOptions.UninitializedMemory);
31+
EncodedMessage encodedMessage = new EncodedMessage(dataBytes.AsSpan());
32+
33+
Span<byte> contentSpan = encodedMessage.Content();
34+
Encoding.UTF8.GetBytes(data, contentSpan);
35+
36+
encodedMessage.AssignType(ISceneCommunicationPipe.MsgType.String);
37+
38+
EncodeAndSendMessage(encodedMessage, ISceneCommunicationPipe.ConnectivityAssertiveness.DROP_IF_NOT_CONNECTED, null);
2939
}
3040

3141
protected override void OnMessageReceived(ISceneCommunicationPipe.DecodedMessage message)

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/Tests/CommunicationControllerAPIImplementationShould.cs‎

Lines changed: 17 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -6,17 +6,20 @@
66
using DCL.Multiplayer.Connections.Messaging.Pipe;
77
using ECS;
88
using Microsoft.ClearScript;
9+
using Microsoft.ClearScript.JavaScript;
910
using Microsoft.ClearScript.V8;
1011
using NSubstitute;
1112
using NUnit.Framework;
1213
using SceneRunner.Scene;
1314
using SceneRuntime;
1415
using System;
16+
using System.Collections;
1517
using System.Collections.Generic;
1618
using System.Linq;
1719
using System.Text;
1820
using System.Threading;
1921
using Utility;
22+
using SceneRuntime.Apis.Modules.EngineApi;
2023

2124
namespace CrdtEcsBridge.JsModulesImplementation.Tests
2225
{
@@ -55,24 +58,25 @@ public void SetUp()
5558
jsOperations);
5659
}
5760

61+
5862
[Test]
5963
public void SendBinary([Range(0, 5)] int outerArraySize, [Range(1, 50)] int innerArrayMessagesCount)
6064
{
6165
// Generate random array of arrays
6266

63-
var outerArray = new PoolableByteArray[outerArraySize];
67+
var outerArray = new List<PoolableByteArray>(outerArraySize);
6468

6569
for (var i = 0; i < outerArraySize; i++)
6670
{
6771
byte[] messages = GetRandomMessagesSequence(innerArrayMessagesCount);
68-
outerArray[i] = new PoolableByteArray(messages, messages.Length, null);
72+
outerArray.Add(new PoolableByteArray(messages, messages.Length, null));
6973
}
7074

71-
api.SendBinary(outerArray);
75+
api.SendBinary<List<PoolableByteArray>, PoolableByteArray>(outerArray);
7276
api.GetResult();
7377

7478
var expectedCalls = outerArray
75-
.Select(o => o.Array
79+
.Select(o => o.CloneAsArray()
7680
.Prepend((byte)ISceneCommunicationPipe.MsgType.Uint8Array)
7781
.Take(o.Length + 1))
7882
.ToList();
@@ -167,12 +171,11 @@ public void ApplyFilterToCRDTMessages()
167171
crdtBody.Write(contentLength); // content length
168172
crdtBody = crdtBody.Slice(contentLength);
169173

170-
var inputs = new PoolableByteArray[]
171-
{
172-
new PoolableByteArray(crdtMessage, crdtMessage.Length, null),
173-
};
174+
var inputs = new List<PoolableByteArray>();
175+
var s = new PoolableByteArray(crdtMessage, crdtMessage.Length, null);
176+
inputs.Add(s);
174177

175-
api.SendBinary(inputs);
178+
api.SendBinary<List<PoolableByteArray>, PoolableByteArray>(inputs);
176179
api.GetResult();
177180

178181
// Expected: CRDT message should be filtered
@@ -259,12 +262,12 @@ public void ApplyFilterToCRDTStateResponseMessages()
259262
addressBytes.CopyTo(resSpan.Slice(2));
260263
crdtData.CopyTo(resSpan.Slice(2 + addressLength));
261264

262-
var inputs = new PoolableByteArray[]
263-
{
264-
new PoolableByteArray(resMessage, resMessage.Length, null),
265-
};
265+
var inputs = new List<PoolableByteArray>();
266+
var s = new PoolableByteArray(resMessage, resMessage.Length, null);
267+
inputs.Add(s);
268+
266269

267-
api.SendBinary(inputs);
270+
api.SendBinary<List<PoolableByteArray>, PoolableByteArray>(inputs);
268271
api.GetResult();
269272

270273
// Expected: RES_CRDT_STATE should be filtered

‎Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/Tests/Tests.asmdef‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727
"Editor"
2828
],
2929
"excludePlatforms": [],
30-
"allowUnsafeCode": false,
30+
"allowUnsafeCode": true,
3131
"overrideReferences": true,
3232
"precompiledReferences": [
3333
"nunit.framework.dll",
@@ -40,4 +40,4 @@
4040
],
4141
"versionDefines": [],
4242
"noEngineReferences": false
43-
}
43+
}

0 commit comments

Comments
 (0)