diff --git a/Explorer/Assets/DCL/Infrastructure/CRDT/CRDTEntity.cs b/Explorer/Assets/DCL/Infrastructure/CRDT/CRDTEntity.cs index 7a874a2a30f..fe589541cd6 100644 --- a/Explorer/Assets/DCL/Infrastructure/CRDT/CRDTEntity.cs +++ b/Explorer/Assets/DCL/Infrastructure/CRDT/CRDTEntity.cs @@ -39,6 +39,14 @@ public int CompareTo(CRDTEntity other) => public bool Equals(CRDTEntity other) => Id.Equals(other.Id); + public override bool Equals(object obj) => + obj is CRDTEntity other && Equals(other); + + // Without the override the default ValueType.GetHashCode kicks in (boxing/reflection-based on Mono) + // when no explicit comparer is provided to a dictionary + public override int GetHashCode() => + Id; + public override string ToString() => $"E: number {EntityNumber} version {EntityVersion}"; diff --git a/Explorer/Assets/DCL/Infrastructure/CRDT/Deserializer/CRDTDeserializer.cs b/Explorer/Assets/DCL/Infrastructure/CRDT/Deserializer/CRDTDeserializer.cs index d56df1be60a..4d03036119e 100644 --- a/Explorer/Assets/DCL/Infrastructure/CRDT/Deserializer/CRDTDeserializer.cs +++ b/Explorer/Assets/DCL/Infrastructure/CRDT/Deserializer/CRDTDeserializer.cs @@ -1,6 +1,5 @@ using CRDT.Memory; using CRDT.Protocol; -using DCL.Diagnostics; using System; using System.Buffers; using System.Collections.Generic; @@ -149,10 +148,6 @@ public bool TryDeserializePutComponent(ref ReadOnlyMemory memory, CRDTMess memory = memory.Slice(shift + dataLength); crdtMessage = new CRDTMessage(messageType, entityId, componentId, timestamp, memoryOwner); - - if (messageType == CRDTMessageType.AUTHORITATIVE_PUT_COMPONENT) - ReportHub.Log(ReportCategory.CRDT, $"Deserialized {nameof(CRDTMessageType.AUTHORITATIVE_PUT_COMPONENT)} - Entity: {crdtMessage.EntityId}, Component: {crdtMessage.ComponentId}, Timestamp: {crdtMessage.Timestamp}"); - return true; } diff --git a/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/CRDTProtocol.cs b/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/CRDTProtocol.cs index 725db8d0bdb..8b19af7c762 100644 --- a/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/CRDTProtocol.cs +++ b/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/CRDTProtocol.cs @@ -1,5 +1,7 @@ using Collections.Pooled; +using CRDT.Memory; using CRDT.Protocol.Factory; +using CRDT.Serializer; using System; using System.Buffers; using System.Collections.Generic; @@ -11,6 +13,21 @@ public class CRDTProtocol : ICRDTProtocol, IComparer + /// Roughly the number of distinct SDK component types a scene touches: presizing avoids rehash cascades + /// during scene-load PUT storms (every rehash re-rents and copies the pooled backing arrays) + /// + private const int LWW_COMPONENTS_CAPACITY = 64; + + /// + /// Entities per component bucket; hot buckets (e.g. Transform) will still grow beyond it + /// + private const int LWW_ENTITIES_CAPACITY = 128; + + private const int APPEND_COMPONENTS_CAPACITY = 8; + private const int APPEND_ENTITIES_CAPACITY = 32; + private const int DELETED_ENTITIES_CAPACITY = 256; + private State crdtState; internal ref State CRDTState => ref crdtState; @@ -18,9 +35,9 @@ public class CRDTProtocol : ICRDTProtocol, IComparer(), - new PooledDictionary>(), - new PooledDictionary>>()); + new PooledDictionary(DELETED_ENTITIES_CAPACITY), + new PooledDictionary>(LWW_COMPONENTS_CAPACITY), + new PooledDictionary>>(APPEND_COMPONENTS_CAPACITY)); } public void Dispose() @@ -121,16 +138,38 @@ public int CreateMessagesFromTheCurrentState(ProcessedCRDTMessage[] preallocated public ProcessedCRDTMessage CreateAppendMessage(CRDTEntity entity, int componentId, int timestamp, in IMemoryOwner data) => CRDTMessagesFactory.CreateAppendMessage(entity, componentId, timestamp, data); - public ProcessedCRDTMessage CreatePutMessage(CRDTEntity entity, int componentId, in IMemoryOwner data) => - crdtState.CreatePutMessage(entity, componentId, data); + public ProcessedCRDTMessage CreateAndCommitPutMessage(CRDTEntity entity, int componentId, in IMemoryOwner data) => + CreateAndCommitLwwMessage(CRDTMessageType.PUT_COMPONENT, entity, componentId, data); - public ProcessedCRDTMessage CreateDeleteMessage(CRDTEntity entity, int componentId) => - crdtState.CreateDeleteMessage(entity, componentId); + public ProcessedCRDTMessage CreateAndCommitDeleteMessage(CRDTEntity entity, int componentId) => + CreateAndCommitLwwMessage(CRDTMessageType.DELETE_COMPONENT, entity, componentId, EmptyMemoryOwner.EMPTY); - public void EnforceLWWState(in CRDTMessage message) + /// + /// Creates the LWW message and writes it into the local CRDT state in a single probe of the state dictionaries. + /// The local message is by construction the newest state (its timestamp is the stored one + 1) so the full + /// reconciliation performed by would be redundant. + /// The CRDT state takes the ownership of . + /// + private ProcessedCRDTMessage CreateAndCommitLwwMessage(CRDTMessageType messageType, CRDTEntity entity, int componentId, in IMemoryOwner data) { - GetReconciliationResultFromLWWMessage(in message, out CRDTReconciliationEffect overrideEffect, out CRDTReconciliationEffect newComponentEffect); - UpdateLWWState(in message, overrideEffect, newComponentEffect); + if (!crdtState.lwwComponents.TryGetValue(componentId, out PooledDictionary inner)) + crdtState.lwwComponents[componentId] = inner = new PooledDictionary(LWW_ENTITIES_CAPACITY, CRDTEntityComparer.INSTANCE); + + var timestamp = 0; + + if (inner.TryGetValue(entity, out EntityComponentData storedData)) + { + timestamp = storedData.Timestamp + 1; + storedData.Data.Dispose(); + } + else + crdtState.messagesCount++; + + inner[entity] = new EntityComponentData(timestamp, data, messageType); + + return new ProcessedCRDTMessage( + new CRDTMessage(messageType, entity, componentId, timestamp, data), + CRDTMessageSerializationUtils.GetMessageDataLength(messageType, in data)); } private static void GetReconciliationResultFromLWWMessage(in CRDTMessage message, out CRDTReconciliationEffect overrideEffect, out CRDTReconciliationEffect newComponentEffect) @@ -154,7 +193,7 @@ private static void GetReconciliationResultFromLWWMessage(in CRDTMessage message private CRDTReconciliationResult UpdateLWWState(in CRDTMessage message, CRDTReconciliationEffect overrideEffect, CRDTReconciliationEffect newComponentEffect, bool ignoreTimestamp = false) { - bool innerSetExists = crdtState.TryGetLWWComponentState(message, out PooledDictionary inner, + bool innerSetExists = crdtState.TryGetLWWComponentState(in message, out PooledDictionary inner, out bool componentExists, out EntityComponentData storedData); bool componentWasDeleted = componentExists && storedData.isDeleted; @@ -201,7 +240,7 @@ private void UpdateLWWState(bool innerSetExists, bool componentExists, PooledDic in CRDTMessage crdtMessage, ref EntityComponentData componentData) { if (!innerSetExists) - crdtState.lwwComponents[crdtMessage.ComponentId] = inner = new PooledDictionary(CRDTEntityComparer.INSTANCE); + crdtState.lwwComponents[crdtMessage.ComponentId] = inner = new PooledDictionary(LWW_ENTITIES_CAPACITY, CRDTEntityComparer.INSTANCE); if (!componentExists) crdtState.messagesCount++; @@ -231,17 +270,17 @@ private void DeleteEntity(CRDTEntity entityId, int entityNumber, int entityVersi foreach (PooledDictionary componentsStorage in crdtState.lwwComponents.Values) { - if (componentsStorage.TryGetValue(entityId, out EntityComponentData componentData)) + // Remove with the out value probes the bucket once, unlike the TryGetValue + Remove pair + if (componentsStorage.Remove(entityId, out EntityComponentData componentData)) { componentData.Data.Dispose(); - componentsStorage.Remove(entityId); crdtState.messagesCount--; } } foreach (KeyValuePair>> componentsSet in crdtState.appendComponents) { - if (componentsSet.Value.TryGetValue(entityId, out PooledList list)) + if (componentsSet.Value.Remove(entityId, out PooledList list)) { crdtState.messagesCount -= list.Count; @@ -249,7 +288,6 @@ private void DeleteEntity(CRDTEntity entityId, int entityNumber, int entityVersi entityComponentData.Data.Dispose(); list.Dispose(); - componentsSet.Value.Remove(entityId); } } } @@ -300,7 +338,7 @@ private bool TryAppendComponent(in CRDTMessage message) // Pooled collection will take care of pooling an internal state, there will be a small garbage related to the class itself // We can tolerate it - crdtState.appendComponents[message.ComponentId] = outer = new PooledDictionary>(CRDTEntityComparer.INSTANCE); + crdtState.appendComponents[message.ComponentId] = outer = new PooledDictionary>(APPEND_ENTITIES_CAPACITY, CRDTEntityComparer.INSTANCE); outer[message.EntityId] = existingSet; return true; @@ -374,7 +412,7 @@ public State(PooledDictionary deletedEntities, PooledDictionary inner, out bool componentExists, + internal readonly bool TryGetLWWComponentState(in CRDTMessage message, out PooledDictionary inner, out bool componentExists, out EntityComponentData storedData) => TryGetLWWComponentState(message.EntityId, message.ComponentId, out inner, out componentExists, out storedData); diff --git a/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/Factory/CRDTMessagesFactory.cs b/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/Factory/CRDTMessagesFactory.cs index 01fd517f5cc..097001d1b33 100644 --- a/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/Factory/CRDTMessagesFactory.cs +++ b/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/Factory/CRDTMessagesFactory.cs @@ -2,7 +2,6 @@ using CRDT.Memory; using CRDT.Serializer; using System.Buffers; -using System.Runtime.CompilerServices; namespace CRDT.Protocol.Factory { @@ -15,12 +14,6 @@ internal static class CRDTMessagesFactory public static ProcessedCRDTMessage CreateAppendMessage(CRDTEntity entity, int componentId, int timestamp, in IMemoryOwner data) => new (new CRDTMessage(CRDTMessageType.APPEND_COMPONENT, entity, componentId, timestamp, data), CRDTMessageSerializationUtils.GetMessageDataLength(CRDTMessageType.APPEND_COMPONENT, in data)); - public static ProcessedCRDTMessage CreatePutMessage(this in CRDTProtocol.State state, CRDTEntity entity, int componentId, in IMemoryOwner data) => - CreateLwwMessage(in state, CRDTMessageType.PUT_COMPONENT, entity, componentId, data); - - public static ProcessedCRDTMessage CreateDeleteMessage(this in CRDTProtocol.State state, CRDTEntity entity, int componentId) => - CreateLwwMessage(in state, CRDTMessageType.DELETE_COMPONENT, entity, componentId, EmptyMemoryOwner.EMPTY); - /// /// Fills the array with messages corresponding to the current CRDT state /// Creates PUT, APPEND, DELETE COMPONENT and DELETE ENTITY messages @@ -85,14 +78,5 @@ private static int AddCRDTMessage(ProcessedCRDTMessage[] preallocatedArray, CRDT index++; return numberOfBytes; } - - [MethodImpl(MethodImplOptions.AggressiveInlining)] - private static ProcessedCRDTMessage CreateLwwMessage(in CRDTProtocol.State state, CRDTMessageType messageType, CRDTEntity entity, int componentId, in IMemoryOwner data) - { - var timestamp = 0; - - if (state.TryGetLWWComponentState(entity, componentId, out _, out _, out CRDTProtocol.EntityComponentData storedData)) timestamp = storedData.Timestamp + 1; - return new ProcessedCRDTMessage(new CRDTMessage(messageType, entity, componentId, timestamp, data), CRDTMessageSerializationUtils.GetMessageDataLength(messageType, in data)); - } } } diff --git a/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/ICRDTProtocol.cs b/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/ICRDTProtocol.cs index 0f951b7a894..fd6836202e8 100644 --- a/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/ICRDTProtocol.cs +++ b/Explorer/Assets/DCL/Infrastructure/CRDT/Protocol/ICRDTProtocol.cs @@ -17,11 +17,6 @@ public interface ICRDTProtocol : IDisposable CRDTReconciliationResult ProcessMessage(in CRDTMessage message); - /// - /// Enforce LWW state update, it must be guaranteed that the message is PUT_COMPONENT or DELETE_COMPONENT - /// - void EnforceLWWState(in CRDTMessage message); - /// /// /// @@ -36,13 +31,14 @@ public interface ICRDTProtocol : IDisposable ProcessedCRDTMessage CreateAppendMessage(CRDTEntity entity, int componentId, int timestamp, in IMemoryOwner data); /// - /// Creates an LWW PUT Message but does not process it + /// Creates an LWW PUT Message and commits it to the local CRDT state in a single pass. + /// The state takes the ownership of /// - ProcessedCRDTMessage CreatePutMessage(CRDTEntity entity, int componentId, in IMemoryOwner data); + ProcessedCRDTMessage CreateAndCommitPutMessage(CRDTEntity entity, int componentId, in IMemoryOwner data); /// - /// Creates an LWW DELETE Message but does not process it + /// Creates an LWW DELETE Message and commits it to the local CRDT state in a single pass /// - ProcessedCRDTMessage CreateDeleteMessage(CRDTEntity entity, int componentId); + ProcessedCRDTMessage CreateAndCommitDeleteMessage(CRDTEntity entity, int componentId); } } diff --git a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/EngineAPIImplementation.cs b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/EngineAPIImplementation.cs index 110ec1f41e0..5b37b2959fa 100644 --- a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/EngineAPIImplementation.cs +++ b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/EngineAPIImplementation.cs @@ -44,7 +44,11 @@ public class EngineAPIImplementation : IEngineApi private readonly CustomSampler worldSyncBufferSampler; private readonly SceneRuntimeMetrics metrics; - private readonly Action processPendingMessage; + /// + /// Invoked for every pending outgoing message before serialization. + /// Null by default so the serialization loop skips the delegate invocation entirely + /// + protected virtual Action? PendingMessageProcessor => null; public EngineAPIImplementation( ISharedPoolsProvider poolsProvider, @@ -78,8 +82,6 @@ public EngineAPIImplementation( outgoingMessagesSampler = CustomSampler.Create("OutgoingMessages"); crdtProcessMessagesSampler = CustomSampler.Create("CRDTProcessMessage"); applyBufferSampler = CustomSampler.Create(nameof(ApplySyncCommandBuffer)); - - processPendingMessage = ProcessPendingMessage; } public virtual void Dispose() { } @@ -154,9 +156,11 @@ public PoolableByteArray CrdtGetState() try { - // Apply outgoing messages straight-away so they are reflected in the current CRDT state + // Drain the outgoing messages so they are reflected in the current CRDT state: + // LWW messages are committed to the state by the provider on creation, + // the rest (APPEND) is not kept in the state so it must be released straight-away using (OutgoingCRDTMessagesSyncBlock outgoingMessagesSyncBlock = GetSerializationSyncBlock()) - SyncOutgoingCRDTMessages(outgoingMessagesSyncBlock.Messages); + DisposeMessagesNotOwnedByState(outgoingMessagesSyncBlock.Messages); // Create CRDT Messages from the current state // we know exactly how big the array should be @@ -190,9 +194,7 @@ public PoolableByteArray CrdtGetState() } private OutgoingCRDTMessagesSyncBlock GetSerializationSyncBlock() => - outgoingCrtdMessagesProvider.GetSerializationSyncBlock(processPendingMessage); - - protected virtual void ProcessPendingMessage(OutgoingCRDTMessagesProvider.PendingMessage pendingMessage) { } + outgoingCrtdMessagesProvider.GetSerializationSyncBlock(PendingMessageProcessor); private PoolableByteArray SerializeOutgoingCRDTMessages() { @@ -209,7 +211,7 @@ private PoolableByteArray SerializeOutgoingCRDTMessages() SerializeOutgoingCRDTMessages(outgoingMessagesSyncBlock.Messages, serializationBufferPoolable.Span); - SyncOutgoingCRDTMessages(outgoingMessagesSyncBlock.Messages); + DisposeMessagesNotOwnedByState(outgoingMessagesSyncBlock.Messages); metrics.MessagesToScene.Add(outgoingMessagesSyncBlock.Messages.Count); } @@ -225,29 +227,18 @@ private PoolableByteArray SerializeOutgoingCRDTMessages() } /// - /// If local messages are not written in the local CRDT, the final state - /// including timestamp is incorrect. Subsequent creation of propagated messages will result in a wrong timestamp + /// LWW messages (PUT/DELETE_COMPONENT) are committed to the local CRDT state when they are created + /// (see ) so their data is owned by the state. + /// The data of the remaining messages (APPEND) is not kept in so it must be released immediately /// - private void SyncOutgoingCRDTMessages(IReadOnlyList outgoingMessages) + private static void DisposeMessagesNotOwnedByState(IReadOnlyList outgoingMessages) { - for (var i = 0; i < outgoingMessages.Count; i++) { SyncCRDTMessage(outgoingMessages[i]); } - } - - private void SyncCRDTMessage(ProcessedCRDTMessage message) - { - // We are interested in LWW messages only, in AUTHORITATIVE_PUT_COMPONENT we are not interested as it can't be originated from the client - switch (message.message.Type) + for (var i = 0; i < outgoingMessages.Count; i++) { - case CRDTMessageType.DELETE_COMPONENT: - case CRDTMessageType.PUT_COMPONENT: - // instead of processing via CRDTProtocol.ProcessMessage - // we can skip part of the logic as we guarantee that the local message is the final valid state (see OutgoingCRDTMessagesProvider.AddLwwMessage) - crdtProtocol.EnforceLWWState(message.message); - break; - default: - // as this data is not kept in CRDTProtocol it must be released immediately - message.message.Data.Dispose(); - break; + CRDTMessage message = outgoingMessages[i].message; + + if (message.Type is not CRDTMessageType.PUT_COMPONENT and not CRDTMessageType.DELETE_COMPONENT) + message.Data.Dispose(); } } diff --git a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/SDKObservableEventsEngineAPIImplementation.cs b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/SDKObservableEventsEngineAPIImplementation.cs index fdf80620f65..ee78bb3a6fd 100644 --- a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/SDKObservableEventsEngineAPIImplementation.cs +++ b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/JsModulesImplementation/SDKObservableEventsEngineAPIImplementation.cs @@ -14,6 +14,7 @@ using SceneRunner.Scene.ExceptionsHandling; using SceneRuntime.Apis.Modules.EngineApi.SDKObservableEvents; using SceneRuntime.Apis.Modules.EngineApi.SDKObservableEvents.Events; +using System; using System.Collections.Generic; using Utility.Multithreading; @@ -24,15 +25,21 @@ public class SDKObservableEventsEngineAPIImplementation : EngineAPIImplementatio private readonly Dictionary userIdEntitiesMap = new (); private readonly List sdkObservableEvents = new (); private readonly HashSet sdkObservableEventSubscriptions = new (); + private readonly Action processPendingMessage; private bool enableSDKObservableMessagesDetection; private bool reportedSceneReady; + protected override Action? PendingMessageProcessor => processPendingMessage; + public SDKObservableEventsEngineAPIImplementation(ISharedPoolsProvider poolsProvider, IInstancePoolsProvider instancePoolsProvider, ICRDTProtocol crdtProtocol, ICRDTDeserializer crdtDeserializer, ICRDTSerializer crdtSerializer, ICRDTWorldSynchronizer crdtWorldSynchronizer, IOutgoingCRDTMessagesProvider outgoingCrtdMessagesProvider, ISystemGroupsUpdateGate systemGroupsUpdateGate, ISceneExceptionsHandler exceptionsHandler, MultiThreadSync multiThreadSync, MultiThreadSync.Owner syncOwner, SceneRuntimeMetrics metrics) : base(poolsProvider, instancePoolsProvider, crdtProtocol, crdtDeserializer, crdtSerializer, crdtWorldSynchronizer, outgoingCrtdMessagesProvider, - systemGroupsUpdateGate, exceptionsHandler, multiThreadSync, syncOwner, metrics) { } + systemGroupsUpdateGate, exceptionsHandler, multiThreadSync, syncOwner, metrics) + { + processPendingMessage = ProcessPendingMessage; + } public void TryAddSubscription(string eventId) { @@ -90,7 +97,7 @@ public override void Dispose() sdkObservableEventSubscriptions.Clear(); } - protected override void ProcessPendingMessage(OutgoingCRDTMessagesProvider.PendingMessage pendingMessage) + private void ProcessPendingMessage(OutgoingCRDTMessagesProvider.PendingMessage pendingMessage) { if (SDKObservableComponentIDs.Ids.Contains(pendingMessage.Bridge.Id)) DetectObservableEventsFromComponents(pendingMessage); diff --git a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingCRDTMessagesProvider.cs b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingCRDTMessagesProvider.cs index 59cb277ccd9..27c88680b21 100644 --- a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingCRDTMessagesProvider.cs +++ b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingCRDTMessagesProvider.cs @@ -15,13 +15,13 @@ namespace CrdtEcsBridge.OutgoingMessages { public class OutgoingCRDTMessagesProvider : IOutgoingCRDTMessagesProvider { - internal static readonly ThreadSafeDictionaryPool INDICES_SHARED_POOL = - new (64, PoolConstants.SCENES_COUNT, new EqualityComparer()); + internal static readonly ThreadSafeDictionaryPool INDICES_SHARED_POOL = + new (64, PoolConstants.SCENES_COUNT); internal static readonly ThreadSafeListPool MESSAGES_SHARED_POOL = new (64, PoolConstants.SCENES_COUNT); - internal readonly Dictionary lwwMessageIndices = INDICES_SHARED_POOL.Get(); + internal readonly Dictionary lwwMessageIndices = INDICES_SHARED_POOL.Get(); internal readonly List messages = MESSAGES_SHARED_POOL.Get(); private readonly ISDKComponentsRegistry componentsRegistry; @@ -74,7 +74,7 @@ private void AddLwwMessage(CRDTEntity entity, SDKComponentBridge componentBridge { lock (messages) { - var key = new OutgoingMessageKey(entity, componentBridge.Id); + long key = OutgoingMessageKey.Pack(entity, componentBridge.Id); if (lwwMessageIndices.TryGetValue(key, out int lwwIndex)) { @@ -143,7 +143,7 @@ public OutgoingCRDTMessagesSyncBlock GetSerializationSyncBlock(Action - { - public bool Equals(OutgoingMessageKey x, OutgoingMessageKey y) => - x.Entity.Equals(y.Entity) && x.ComponentId == y.ComponentId; - - public int GetHashCode(OutgoingMessageKey obj) => - HashCode.Combine(obj.Entity.Id, obj.ComponentId); - } } } diff --git a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingMessageKey.cs b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingMessageKey.cs index 8601e5678cf..e688ff7c892 100644 --- a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingMessageKey.cs +++ b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/OutgoingMessageKey.cs @@ -1,16 +1,15 @@ -using CRDT; +using CRDT; namespace CrdtEcsBridge.OutgoingMessages { - internal readonly struct OutgoingMessageKey + /// + /// (Entity, ComponentId) packed into a single long so the pending-messages dictionary + /// uses the devirtualized default comparer instead of a comparer-class dispatch with + /// per probe + /// + internal static class OutgoingMessageKey { - public readonly CRDTEntity Entity; - public readonly int ComponentId; - - public OutgoingMessageKey(CRDTEntity entity, int componentId) - { - Entity = entity; - ComponentId = componentId; - } + public static long Pack(CRDTEntity entity, int componentId) => + ((long)componentId << 32) | (uint)entity.Id; } } diff --git a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/Tests/OutgoingCRTDMessagesProviderShould.cs b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/Tests/OutgoingCRTDMessagesProviderShould.cs index 2b0233232e0..bd5d369ffe0 100644 --- a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/Tests/OutgoingCRTDMessagesProviderShould.cs +++ b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/OutgoingMessages/Tests/OutgoingCRTDMessagesProviderShould.cs @@ -43,7 +43,7 @@ public void AppendMessage() Assert.That(addedMessage.Bridge.Id, Is.EqualTo(ComponentID.POINTER_EVENTS_RESULT)); Assert.That(addedMessage.Timestamp, Is.EqualTo(100)); - Assert.That(provider.lwwMessageIndices, Does.Not.ContainKey(new OutgoingMessageKey(crdtEntity, ComponentID.POINTER_EVENTS_RESULT))); + Assert.That(provider.lwwMessageIndices, Does.Not.ContainKey(OutgoingMessageKey.Pack(crdtEntity, ComponentID.POINTER_EVENTS_RESULT))); } [Test] @@ -65,7 +65,7 @@ public void PutMessage() Assert.That(addedMessage.Bridge.Id, Is.EqualTo(ComponentID.POINTER_EVENTS_RESULT)); Assert.That(addedMessage.Timestamp, Is.EqualTo(0)); - Assert.That(provider.lwwMessageIndices, Contains.Key(new OutgoingMessageKey(crdtEntity, ComponentID.POINTER_EVENTS_RESULT))); + Assert.That(provider.lwwMessageIndices, Contains.Key(OutgoingMessageKey.Pack(crdtEntity, ComponentID.POINTER_EVENTS_RESULT))); } [Test] @@ -84,7 +84,7 @@ public void DeleteMessage() Assert.That(addedMessage.Bridge.Id, Is.EqualTo(ComponentID.POINTER_EVENTS_RESULT)); Assert.That(addedMessage.Timestamp, Is.EqualTo(0)); - Assert.That(provider.lwwMessageIndices, Contains.Key(new OutgoingMessageKey(crdtEntity, ComponentID.POINTER_EVENTS_RESULT))); + Assert.That(provider.lwwMessageIndices, Contains.Key(OutgoingMessageKey.Pack(crdtEntity, ComponentID.POINTER_EVENTS_RESULT))); } [Test] @@ -134,8 +134,8 @@ public void ProvideSerializationSyncBlock() using OutgoingCRDTMessagesSyncBlock syncBlock = provider.GetSerializationSyncBlock(null); - crdtProtocol.Received(1).CreatePutMessage(Arg.Is(c => c.Id == 1), ComponentID.POINTER_EVENTS_RESULT, Arg.Any>()); - crdtProtocol.Received(1).CreatePutMessage(Arg.Is(c => c.Id == 2), ComponentID.POINTER_EVENTS_RESULT, Arg.Any>()); + crdtProtocol.Received(1).CreateAndCommitPutMessage(Arg.Is(c => c.Id == 1), ComponentID.POINTER_EVENTS_RESULT, Arg.Any>()); + crdtProtocol.Received(1).CreateAndCommitPutMessage(Arg.Is(c => c.Id == 2), ComponentID.POINTER_EVENTS_RESULT, Arg.Any>()); crdtProtocol.Received(1).CreateAppendMessage(Arg.Is(c => c.Id == 3), ComponentID.POINTER_EVENTS_RESULT, 780, Arg.Any>()); } diff --git a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/WorldSynchronizer/WorldSyncCommandBuffer.cs b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/WorldSynchronizer/WorldSyncCommandBuffer.cs index 1f54ecd68ee..324a0dd1211 100644 --- a/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/WorldSynchronizer/WorldSyncCommandBuffer.cs +++ b/Explorer/Assets/DCL/Infrastructure/CrdtEcsBridge/WorldSynchronizer/WorldSyncCommandBuffer.cs @@ -16,33 +16,46 @@ namespace CrdtEcsBridge.WorldSynchronizer /// public class WorldSyncCommandBuffer : IWorldSyncCommandBuffer { + private const int MERGE_MATRIX_SIZE = 5; + /// - /// Represents the final state of the operation on (entity, component) + /// Represents the final state of the operation on (entity, component). + /// Flat array indexed by (first * size) + last: it is probed for every touched (entity, component) pair per batch + /// so it must stay cheaper than a dictionary lookup. Pairs never produced by resolve + /// to the default . /// - private static readonly Dictionary MERGE_MATRIX = - new () - { - // if the first op is "ComponentAdded" then the component didn't exists so it should be added if not deleted - { (CRDTReconciliationEffect.ComponentAdded, CRDTReconciliationEffect.ComponentAdded), CRDTReconciliationEffect.ComponentAdded }, - { (CRDTReconciliationEffect.ComponentAdded, CRDTReconciliationEffect.ComponentModified), CRDTReconciliationEffect.ComponentAdded }, + private static readonly CRDTReconciliationEffect[] MERGE_MATRIX = CreateMergeMatrix(); + + private static CRDTReconciliationEffect[] CreateMergeMatrix() + { + var matrix = new CRDTReconciliationEffect[MERGE_MATRIX_SIZE * MERGE_MATRIX_SIZE]; - // if deleted then nothing to do as it didn't exist before - { (CRDTReconciliationEffect.ComponentAdded, CRDTReconciliationEffect.ComponentDeleted), CRDTReconciliationEffect.NoChanges }, + static int Index(CRDTReconciliationEffect first, CRDTReconciliationEffect last) => + ((int)first * MERGE_MATRIX_SIZE) + (int)last; - // if component already existed then if the component still exists it should be modified - { (CRDTReconciliationEffect.ComponentModified, CRDTReconciliationEffect.ComponentAdded), CRDTReconciliationEffect.ComponentModified }, - { (CRDTReconciliationEffect.ComponentModified, CRDTReconciliationEffect.ComponentModified), CRDTReconciliationEffect.ComponentModified }, + // if the first op is "ComponentAdded" then the component didn't exists so it should be added if not deleted + matrix[Index(CRDTReconciliationEffect.ComponentAdded, CRDTReconciliationEffect.ComponentAdded)] = CRDTReconciliationEffect.ComponentAdded; + matrix[Index(CRDTReconciliationEffect.ComponentAdded, CRDTReconciliationEffect.ComponentModified)] = CRDTReconciliationEffect.ComponentAdded; - // if it was deleted then delete from World - { (CRDTReconciliationEffect.ComponentModified, CRDTReconciliationEffect.ComponentDeleted), CRDTReconciliationEffect.ComponentDeleted }, + // if deleted then nothing to do as it didn't exist before + matrix[Index(CRDTReconciliationEffect.ComponentAdded, CRDTReconciliationEffect.ComponentDeleted)] = CRDTReconciliationEffect.NoChanges; - // if component already existed then if the component still exists it should be modified - { (CRDTReconciliationEffect.ComponentDeleted, CRDTReconciliationEffect.ComponentModified), CRDTReconciliationEffect.ComponentModified }, - { (CRDTReconciliationEffect.ComponentDeleted, CRDTReconciliationEffect.ComponentAdded), CRDTReconciliationEffect.ComponentModified }, + // if component already existed then if the component still exists it should be modified + matrix[Index(CRDTReconciliationEffect.ComponentModified, CRDTReconciliationEffect.ComponentAdded)] = CRDTReconciliationEffect.ComponentModified; + matrix[Index(CRDTReconciliationEffect.ComponentModified, CRDTReconciliationEffect.ComponentModified)] = CRDTReconciliationEffect.ComponentModified; - // if it was deleted then delete from World - { (CRDTReconciliationEffect.ComponentDeleted, CRDTReconciliationEffect.ComponentDeleted), CRDTReconciliationEffect.ComponentDeleted }, - }; + // if it was deleted then delete from World + matrix[Index(CRDTReconciliationEffect.ComponentModified, CRDTReconciliationEffect.ComponentDeleted)] = CRDTReconciliationEffect.ComponentDeleted; + + // if component already existed then if the component still exists it should be modified + matrix[Index(CRDTReconciliationEffect.ComponentDeleted, CRDTReconciliationEffect.ComponentModified)] = CRDTReconciliationEffect.ComponentModified; + matrix[Index(CRDTReconciliationEffect.ComponentDeleted, CRDTReconciliationEffect.ComponentAdded)] = CRDTReconciliationEffect.ComponentModified; + + // if it was deleted then delete from World + matrix[Index(CRDTReconciliationEffect.ComponentDeleted, CRDTReconciliationEffect.ComponentDeleted)] = CRDTReconciliationEffect.ComponentDeleted; + + return matrix; + } private readonly Dictionary> batchStates; private readonly List deletedEntities; @@ -172,7 +185,8 @@ public void FinalizeAndDeserialize() foreach (BatchState batchState in componentsBatch.Values) { - CRDTReconciliationEffect finalState = MERGE_MATRIX[batchState.reconciliationState]; + ReconciliationState reconciliationState = batchState.reconciliationState; + CRDTReconciliationEffect finalState = MERGE_MATRIX[((int)reconciliationState.First * MERGE_MATRIX_SIZE) + (int)reconciliationState.Last]; // just override with the final state batchState.reconciliationState = new ReconciliationState(finalState, finalState);