Question A use case I'm unable to solve (ordered events system)

Looking for guidance on how to solve this problem I'm having. I made another thread that touched on this topic indirectly, but I figured it's better to make a new thread about the specific problem

What I want to do:

• I want a game where an ordered list of "events" is created, and then gets processed in order
• An "event" could do anything and should be easily user-implementable. Examples of events:
  • Apply damage to something
  • Spawn a VFX
  • Move a transform
  • Destroy an entity
  • Add a Buff to some unit
  • etc....
• An event can create another event which inserts itself at any point in the ordered events stack; not necessarily at the end. For instance, an event could create a new event that inserts itself directly after the original event, and must absolutely be processed before the rest of the events stack

The problems I'm having:

• Because the order of events is crucial, I need to keep some kind of list of all events in the right order somewhere. I can't just have one list per type of event and process those in batches, because that would break ordering
• Because events can modify ECS components in the world (like modify someone's HP for example), whatever Job we create to process those events must have a way to get passed the ComponentDataFromEntity arrays that those events need. And we can't know about those in advance.
  • What makes this even more complicated is that if two different event types require access to the same componentFromEntity, we need to make sure the componentFromEntity won't end up in the event processing job twice, because that would cause aliasing errors. Our strategy for easy modular event implementation needs to take this into account
• Because events could hold any type/size of data, I can't just have a global NativeList of events. I would need either:
  • A NativeStream of events, where each event starts with an id so we know how big its data is and we know how to read it
  • Entities as events. We have one NativeList<Entity> which keeps the order of events, and then the Entities referenced by that list hold the data of the events
• Because events could create other events that must be processed instantly before the rest of the events stack is processed....
  • I can't implement these events using NativeStream, because I can't write to a stream as I'm reading from it. And I can't "insert" data into a stream at a specific point either
  • I can't implement these events using Entities-as-events in a job, because event entity creation would need to be instantaneous

What my options look like, with all the restrictions mentioned above:

• Solution 1: Big monolithic event struct
  • I have one struct type that holds the data for every possible event type that could happen in the game.
  • I have a NativeList<MyMonolithicEvent> to keep order
  • I launch a job that is given all the possible ComponentDataFromEntity that these event could need, and process the events in order.
  • Since this is a list, I can insert new events into the list during the processing of events
• Solution 2: Main thread entities-as-events
  • My events exist as components on entities
  • I have an ordered list of structs that contain: event Entity + event Type
  • On the main thread, I go through that list, get the appropriate component type on that entity, and process that event
  • Since this runs on main thread, I can create a new event entity instantly and insert it into the events list

Both of these solutions seem pretty terrible to me, although I'd argue solution 1 is the least terrible. I'm not expecting this to be solvable in parallel, but I would at least want to find an elegant & modular single-thread bursted job approach to this. Or bursted main thread code, once that becomes possible

 

I would first ask if what you are trying to do is something you really should be doing. It kinda flows against the ECS architecture (at least this particular variant of ECS).

Some random thoughts:

1) If you want an event queue, but then the ability for an event to generate another event that is immediately processed, what you need is a queue and a stack. After each event in the queue, you go through and pop all the events in the stack before grabbing the next queue item.
2) Most likely for something as flexible as you want it to be, you may want to mimic what the EntityCommandBuffer does.
3) There is no way for an event that can "do anything" not be processed in the context of a sync point. Even if you could sometimes avoid the sync point, that makes your game loop volatile to hiccups.
4) If there was an API for ComponentDataFromEntityGeneric and BufferFromEntityGeneric, I could give you a pattern that would use those and function pointers to do some pretty cool main-thread operations in Burst. But unfortunately those don't exist, and you will likely have to asmref into the internals to get the tools you need.

Edit: Might be worth looking into how the Visual Scripting tool does events, since that's kinda what its purpose is.

 

I can definitely agree that it's not a problem that suits the ECS well, but I do think it's necessary for this to be relatively easily achievable within DOTS. I could have a game that would make good use of DOTS, but needs ordered events for a certain aspect of the game, etc....

But I think if a satisfactory solution that runs on the main thread and is burstable can be found, that would be plenty sufficient. I can't really imagine use cases where there would be 100k events that absolutely need to be in order, so it's not like this is a big performance-sensitive problem

Oh man, yes that sounds like it would solve that part of the problem. Awesome

I had no idea about this. Checking it out now, thanks

 

One thing that came to mind:

I imagine that a Reliable Ordered network messages channel would be very close to the use case I'm facing. Maybe not for the "inserting new events in the stack" part, but just for the general implementation of generic user-implementable events that are all part of a same stream. I wonder if that's a problem the NetCode team has tackled?

 

I had a similar use case. NativeStreams were not working for me, dont remember exactly why but i think i couldnt reuse the same stream for reading/writing events.
I implementent such system by having nativequeue for each possible event + another event order queue which contained what event type should be processed next. I had two jobs, one consuming events which occured during the frame, sorting them and adding them to the queues. Other job was simply popping queues and executing commands. Was a quite big switch block, but did a job done.

Spoiler: Code

Code (CSharp):

1. public class CombatPresentationSystem : JobComponentSystem
2.     {
3.         private struct Exclude : IComponentData
4.         {
5.         }
6.         private struct PresentationHeader
7.         {
8.             public const int Move = 1;
9.             public const int AbilityCast = 2;
10.             public const int Death = 3;
11.             public const int HUDUpdate = 4;
12.  
13.             public int Type;
14.             public int Length;
15.         }
16.  
17.         private BeginInitializationEntityCommandBufferSystem _bufferSystem;
18.         private EntityQuery _pendingActionQuery;
19.         private EntityQuery _moveCommandQuery;
20.         private EntityQuery _deadUnitsQuery;
21.         private EntityQuery _liveUnitsQuery;
22.         private EntityQuery _abilityCastCommandQuery;
23.  
24.         private NativeQueue<PresentationHeader> _headersQueue;
25.         private NativeQueue<UnitArtMovementCommand> _unitMovedCommandsQueue;
26.         private NativeQueue<UnitArtAttackAnimationCommand> _unitAttackedCommandsQueue;
27.         private NativeQueue<UnitArtDeathCommand> _unitDeathCommandsQueue;
28.         private NativeQueue<UnitHUDUpdateCommand> _unitHUDUpdateCommandsQueue;
29.  
30.         protected override void OnCreate()
31.         {
32.             base.OnCreate();
33.  
34.             _bufferSystem = EntityManager.World.GetExistingSystem<BeginInitializationEntityCommandBufferSystem>();
35.             EntityManager.CreateEntity(typeof(PresentationState));
36.          
37.             _moveCommandQuery = GetEntityQuery(ComponentType.ReadOnly<GridMovementCommand>());
38.             _abilityCastCommandQuery = GetEntityQuery(ComponentType.ReadOnly<AbilityCastCommand>());
39.             _deadUnitsQuery = GetEntityQuery(ComponentType.ReadOnly<Unit>(), ComponentType.ReadOnly<Disabled>(),
40.                 ComponentType.ReadOnly<Dead>(), ComponentType.Exclude<Exclude>());
41.             _liveUnitsQuery = GetEntityQuery(ComponentType.ReadOnly<Unit>());
42.  
43.             _pendingActionQuery = GetEntityQuery(ComponentType.ReadOnly<PendingAction>(), typeof(Progress));
44.          
45.          
46.  
47.             _headersQueue = new NativeQueue<PresentationHeader>(Allocator.Persistent);
48.             _unitMovedCommandsQueue = new NativeQueue<UnitArtMovementCommand>(Allocator.Persistent);
49.             _unitAttackedCommandsQueue = new NativeQueue<UnitArtAttackAnimationCommand>(Allocator.Persistent);
50.             _unitDeathCommandsQueue = new NativeQueue<UnitArtDeathCommand>(Allocator.Persistent);
51.             _unitHUDUpdateCommandsQueue = new NativeQueue<UnitHUDUpdateCommand>(Allocator.Persistent);
52.             RequireSingletonForUpdate<CombatGameState>();
53.         }
54.  
55.         protected override void OnDestroy()
56.         {
57.             _headersQueue.Dispose();
58.             _unitMovedCommandsQueue.Dispose();
59.             _unitAttackedCommandsQueue.Dispose();
60.             _unitDeathCommandsQueue.Dispose();
61.             _unitHUDUpdateCommandsQueue.Dispose();
62.             base.OnDestroy();
63.         }
64.  
65.        // [BurstCompile]
66.         private struct PresentationQueueJob : IJob
67.         {
68.             public EntityCommandBuffer Buffer;
69.  
70.             [ReadOnly] public bool CombatUpdate;
71.             [ReadOnly] public ActiveUnit ActiveUnit;
72.             [ReadOnly] [DeallocateOnJobCompletion] public NativeArray<GridMovementCommand> UnitMovedEvents;
73.             [ReadOnly] [DeallocateOnJobCompletion] public NativeArray<AbilityCastCommand> UnitAttackedEvents;
74.             [ReadOnly] [DeallocateOnJobCompletion] public NativeArray<Entity> DeadUnits;
75.             [ReadOnly] [DeallocateOnJobCompletion] public NativeArray<Entity> LiveUnits;
76.  
77.             [WriteOnly] public NativeQueue<PresentationHeader> HeadersQueue;
78.             [WriteOnly] public NativeQueue<UnitArtMovementCommand> MoveCommandQueue;
79.             [WriteOnly] public NativeQueue<UnitArtAttackAnimationCommand> AttackCommandQueue;
80.             [WriteOnly] public NativeQueue<UnitArtDeathCommand> DeathCommandQueue;
81.             [WriteOnly] public NativeQueue<UnitHUDUpdateCommand> HUDCommandQueue;
82.  
83.             [ReadOnly] public ComponentDataFromEntity<Health> HealthFromEntity;
84.             [ReadOnly] public ComponentDataFromEntity<ActionPoints> ActionPointsFromEntity;
85.             [ReadOnly] public ComponentDataFromEntity<Stamina> StaminaFromEntity;
86.             [ReadOnly] public ComponentDataFromEntity<Initiative> InitiativeFromEntity;
87.             [ReadOnly] public ComponentDataFromEntity<Mana> ManaFromEntity;
88.          
89.             public void Execute()
90.             {
91.                 for (int i = 0; i < UnitMovedEvents.Length; i++)
92.                 {
93.                     var request = UnitMovedEvents[i];
94.                     if (request.Status == 0)
95.                     {
96.                         HeadersQueue.Enqueue(new PresentationHeader {Type = PresentationHeader.Move, Length = 1});
97.                         MoveCommandQueue.Enqueue(new UnitArtMovementCommand
98.                         {
99.                             MovingEntity = request.MovingEntity,
100.                             StartCellCoord = request.StartCellCoord,
101.                             TargetCellCoord = request.TargetCellCoord
102.                         });
103.                     }
104.                 }
105.  
106.                 for (int i = 0; i < UnitAttackedEvents.Length; i++)
107.                 {
108.                     var command = UnitAttackedEvents[i];
109.                     if (command.Status == 0)
110.                     {
111.                         HeadersQueue.Enqueue(new PresentationHeader {Type = PresentationHeader.AbilityCast, Length = 1});
112.                         AttackCommandQueue.Enqueue(new UnitArtAttackAnimationCommand
113.                         {
114.                             Ability = command.AbilityEntity,
115.                             Unit = command.UnitEntity
116.                         });
117.                     }
118.                 }
119.  
120.                 if (CombatUpdate && LiveUnits.Length > 0)
121.                 {
122.                     HeadersQueue.Enqueue(new PresentationHeader { Type = PresentationHeader.HUDUpdate, Length = LiveUnits.Length});
123.                     for (int i = 0; i < LiveUnits.Length; i++)
124.                     {
125.                         var liveUnit = LiveUnits[i];
126.                         HUDCommandQueue.Enqueue(new UnitHUDUpdateCommand
127.                         {
128.                             Unit = liveUnit,
129.                             IsActive = liveUnit.Equals(ActiveUnit.UnitEntity),
130.                             ActionPoints = ActionPointsFromEntity[liveUnit],
131.                             Health = HealthFromEntity[liveUnit],
132.                             Stamina = StaminaFromEntity[liveUnit],
133.                             Initiative = InitiativeFromEntity[liveUnit],
134.                             Mana = ManaFromEntity[liveUnit]
135.                         });
136.                     }
137.                 }
138.  
139.                 if (DeadUnits.Length > 0)
140.                 {
141.                     HeadersQueue.Enqueue(new PresentationHeader { Type = PresentationHeader.Death, Length = DeadUnits.Length});
142.                     for (int i = 0; i < DeadUnits.Length; i++)
143.                     {
144.                         var deadUnit = DeadUnits[i];
145.                         DeathCommandQueue.Enqueue(new UnitArtDeathCommand{Unit = deadUnit});
146.                      
147.                         Buffer.AddComponent<Exclude>(deadUnit);
148.                     }
149.                 }
150.             }
151.         }
152.  
153.         //[BurstCompile]
154.         private struct PresentationJob : IJob
155.         {
156.             public EntityCommandBuffer Buffer;
157.  
158.             public Entity PresentationState;
159.             public NativeQueue<PresentationHeader> HeadersQueue;
160.             public NativeQueue<UnitArtMovementCommand> MoveCommandsQueue;
161.             public NativeQueue<UnitArtAttackAnimationCommand> AttackCommandsQueue;
162.             public NativeQueue<UnitArtDeathCommand> DeathCommandsQueue;
163.             public NativeQueue<UnitHUDUpdateCommand> HUDCommandsQueue;
164.  
165.             [ReadOnly] [DeallocateOnJobCompletion] public NativeArray<Entity> PendingActions;
166.             [ReadOnly] [DeallocateOnJobCompletion] public NativeArray<Progress> PendingActionProgresses;
167.  
168.             public void Execute()
169.             {
170.                 if (HeadersQueue.Count > 0)
171.                 {
172.                     if (PendingActions.Length > 0)
173.                     {
174.                         if (AllPendingActionsCompleted())
175.                         {
176.                             DestroyPendingActions();
177.                             ProcessNext();
178.                         }
179.                     }
180.                     else
181.                     {
182.                         ProcessNext();
183.                     }
184.                 }
185.  
186.                 Buffer.SetComponent(PresentationState, new PresentationState
187.                 {
188.                     Commands = HeadersQueue.Count
189.                 });
190.             }
191.  
192.             private void ProcessNext()
193.             {
194.                 var header = HeadersQueue.Dequeue();
195.                 var type = header.Type;
196.                 var lenght = header.Length;
197.                 switch (type)
198.                 {
199.                     case PresentationHeader.Move:
200.                     {
201.                         CreatePendingActions(lenght, MoveCommandsQueue);
202.                         break;
203.                     }
204.                     case PresentationHeader.AbilityCast:
205.                     {
206.                         CreatePendingActions(lenght, AttackCommandsQueue);
207.                         break;
208.                     }
209.                     case PresentationHeader.Death:
210.                     {
211.                         CreatePendingActions(lenght, DeathCommandsQueue);
212.                         break;
213.                     }
214.                     case PresentationHeader.HUDUpdate:
215.                     {
216.                         CreateActions(lenght, HUDCommandsQueue);
217.                         break;
218.                     }
219.                 }
220.             }
221.  
222.             private void CreatePendingActions<T>(int count, NativeQueue<T> commandQueue) where T : struct, IComponentData
223.             {
224.                 for (int i = 0; i < count; i++)
225.                 {
226.                     var command = commandQueue.Dequeue();
227.                     var commandEntity = Buffer.CreateEntity();
228.                     Buffer.AddComponent<PendingAction>(commandEntity);
229.                     Buffer.AddComponent(commandEntity, command);
230.                     Buffer.AddComponent(commandEntity, new Progress {IsCompleted = false});
231.                 }
232.             }
233.  
234.             private void CreateActions<T>(int count, NativeQueue<T> commandQueue) where T : struct, IComponentData
235.             {
236.                 for (int i = 0; i < count; i++)
237.                 {
238.                     var command = commandQueue.Dequeue();
239.                     var commandEntity = Buffer.CreateEntity();
240.                     Buffer.AddComponent(commandEntity, command);
241.                 }
242.             }
243.          
244.             private void DestroyPendingActions()
245.             {
246.                 for (int i = 0; i < PendingActions.Length; i++)
247.                 {
248.                     Buffer.DestroyEntity(PendingActions[i]);
249.                 }
250.             }
251.  
252.             private bool AllPendingActionsCompleted()
253.             {
254.                 for (int i = 0; i < PendingActionProgresses.Length; i++)
255.                 {
256.                     if (!PendingActionProgresses[i].IsCompleted)
257.                     {
258.                         return false;
259.                     }
260.                 }
261.  
262.                 return true;
263.             }
264.         }
265.  
266.         protected override JobHandle OnUpdate(JobHandle inputDeps)
267.         {
268.             var buffer = _bufferSystem.CreateCommandBuffer();
269.             inputDeps = new PresentationQueueJob
270.             {
271.                 Buffer = buffer,
272.                 CombatUpdate = HasSingleton<CombatUpdate>(),
273.                 ActiveUnit = HasSingleton<ActiveUnit>() ? GetSingleton<ActiveUnit>() : new ActiveUnit { UnitEntity = Entity.Null},
274.                 HeadersQueue = _headersQueue,
275.                 MoveCommandQueue = _unitMovedCommandsQueue,
276.                 AttackCommandQueue = _unitAttackedCommandsQueue,
277.                 DeathCommandQueue = _unitDeathCommandsQueue,
278.                 HUDCommandQueue = _unitHUDUpdateCommandsQueue,
279.                 DeadUnits = _deadUnitsQuery.ToEntityArray(Allocator.TempJob),
280.                 LiveUnits = _liveUnitsQuery.ToEntityArray(Allocator.TempJob),
281.                 UnitMovedEvents = _moveCommandQuery.ToComponentDataArray<GridMovementCommand>(Allocator.TempJob),
282.                 UnitAttackedEvents =
283.                     _abilityCastCommandQuery.ToComponentDataArray<AbilityCastCommand>(Allocator.TempJob),
284.                 HealthFromEntity = GetComponentDataFromEntity<Health>(true),
285.                 StaminaFromEntity = GetComponentDataFromEntity<Stamina>(true),
286.                 ActionPointsFromEntity = GetComponentDataFromEntity<ActionPoints>(true),
287.                 InitiativeFromEntity = GetComponentDataFromEntity<Initiative>(true),
288.                 ManaFromEntity = GetComponentDataFromEntity<Mana>(true)
289.             }.Schedule(inputDeps);
290.  
291.             var pendingActions = _pendingActionQuery.ToEntityArray(Allocator.TempJob);
292.             var progresses = _pendingActionQuery.ToComponentDataArray<Progress>(Allocator.TempJob);
293.             inputDeps = new PresentationJob
294.             {
295.                 Buffer = buffer,
296.                 PresentationState = GetSingletonEntity<PresentationState>(),
297.                 HeadersQueue = _headersQueue,
298.                 MoveCommandsQueue = _unitMovedCommandsQueue,
299.                 AttackCommandsQueue = _unitAttackedCommandsQueue,
300.                 DeathCommandsQueue = _unitDeathCommandsQueue,
301.                 HUDCommandsQueue = _unitHUDUpdateCommandsQueue,
302.                 PendingActions = pendingActions,
303.                 PendingActionProgresses = progresses
304.             }.Schedule(inputDeps);
305.  
306.             _bufferSystem.AddJobHandleForProducer(inputDeps);
307.  
308.             return inputDeps;
309.         }
310.     }

 

I'm also interested for the solution of this problem. Mine is a similar but for AI (ordered actions). Is there an example usage of NativeStream?

 

@PhilSA I’ve appreciated your posts here, and I’m glad for a chance to add something which *might* be helpful. I had a similar problem is my work, and I solved it this way:

In my case, I had several entities with several children each. Each of the children could raise events which affected any number of its siblings. Those events had to be processed in order, but would need to be raised from parallel, Bursted jobs.

before writing a custom collection type, I wanted to see if a solution was possible using the base ECS tools already available.

here’s what I did:

1. Bound the problem. I set a hard rule that an event raised by a child entity could only affect its sibling entities.

2. Bound the number of events per frame: Originally, I set a static readonly constant for this. Later, I switched to an “EventProducerAttribute” which would decorate a job from which events were fired, and contained an int for the max number of events which could be fired from that job, per sibling. In an initialization step, I would calculate the max number of possible events that could be fired each frame.

3. Gave each of the child entities a buffer to hold the events. The capacity = the number of the entity’s siblings * the max number of events per frame.

4. Chunked/partitioned each event DB: I assigned each sibling an index. That sibling could only ever assign events to its siblings' event DBs from the range of it’s own index, up to it’s index + maxEventsPerFrame.

5. Shared a single BufferFromEntity<EventType> across all job types which needed to raise events. I also marked that BFE with attributes to disable safety checks. This way, those jobs could write to those DBs in parallel.

6. Included ordering data inside each event struct: When an event-raising job is scheduled, I request an integer from a global counter, which increments after each request. That integer will identify the order in which my event-raising jobs were scheduled (which I treat as the logical order in which those events should be processed). I also include an eventCount (how many events have been raised this frame, from *this* job). <-- crucial to this last part is that all of my event-raising jobs are IJobChunks. I tightly control how that eventCount int in incremented, so it's safe from multiple threads.

7. After all events have been raised, I sort the events for each event DB in a followup system, based on the data from step 6.

I was concerned about the sort speed, but since the situation as bounded in several ways, it ended up being quite speedy. I can have tens of thousands of children raising events for itself and its siblings each frame, without the events becoming a bottleneck.

Every situation is different, and maybe this approach wouldn't map to yours. But maybe there is something here that will offer some ideas. GL - I'll be watching this thread.