Event System

As the original post is a bit outdated now I've moved it to here

The project and documentation can be found here: https://gitlab.com/tertle/com.bovinelabs.event.git

Available on openupm https://openupm.com/packages/com.bovinelabs.event/

There's a 2.X branch that is a complete rewrite that supports ISystem (note it lacks documentation) https://gitlab.com/tertle/com.bovinelabs.event/-/tree/2.1.0

 

Original Post
Now a bit dated but kept here for reference as it's still relevant however usage has changed a bit. Read down page for latest details.

===========


Preface This is a follow up to the idea I discussed here and a continuation of my original event system I designed from the same thread.

What is this?
It’s really just an idea on one possible way someone might implement part of the architecture of an ECS based application. Fundamentally this particular solution is really just a method of automatically and safely sharing NativeStreams between multiple systems to create an event system.

Objectives

• same frame producing then consuming
• the option to share events between worlds
• support different loop update rates. e.g. produce in update, consume in fixed update
  
• any number of producers and consumers
• fully threaded
  
• no sync points
• no garbage
• easy dependency management
• negligible overhead

Where to get it?
The repo is available here
The best way to install it is to simply pull it into your project using the package manager. You can do this by adding

to your manifest.json. The package contains 3 directories

• Event
  The actual source code lives in here
• Event.Tests
  Unit tests for the solution. Good coverage but definitely not thorough yet (it didn't catch a bunch of issues I found building a sample). If you want to run these you’ll need to pull in the shared testing library I use between packages
  
• Samples~
  Contains importable samples. At this stage it is a 3 world system with 1 world being in fixed update. To import into project go to Window->Package Manager, Under BovineLabs Event select hit Import Samples.

When to use it?
~coming soon.
I will be working on a bunch of samples and use cases for ideas I have in my head in the coming days.

How to use?
To get the event system you simply request it like any other system.

this.eventSystem = this.World.GetOrCreateSystem<LateSimulationEventSystem>();

For now by default there is a LateSimulationEventSystem and PresentationEventSystem. It is important to use the system that is part of your loop. If you are using secondary worlds you'll need to implement your own system for that world (very easy, more below.)

The basics between being a producer and a consumer are pretty much the same. You have to get your writer/reader and then return your dependency handle. An understanding of NativeStreams is important as this is what events are written to and read from. You can read about them here

Producer (write)
To write events you first request a new NativeStream.

NativeStream.Writer writer = this.eventSystem.CreateEventWriter<TestEvent>(foreachCount);

The foreachCount is the foreachCount property of the NativeStream. I will be providing more examples on how to determine this in the coming days.
You can then pass this writer to a job or use it on the spot as you would any other native stream. It's important however to only write the event type requested to the stream (or a value that can be reinterpreted as the event type i.e. same memory layout).

Once you have passed this to the job, you must return the handle to the event system.

this.eventSystem.AddJobHandleForProducer<TestEvent>(handle);

An example of a IJobParallelFor use could be something like this

Code (CSharp):

1.     protected override JobHandle OnUpdate(JobHandle handle)
2.     {
3.         const int foreachCount = 100;
4.  
5.         NativeStream.Writer writer = this.eventSystem.CreateEventWriter<TestEvent>(foreachCount);
6.  
7.         handle = new ProduceJob
8.             {
9.                 Events = writer,
10.             }
11.             .Schedule(foreachCount, 16, handle);
12.  
13.         this.eventSystem.AddJobHandleForProducer<TestEvent>(handle);
14.  
15.         return handle;
16.     }

Consumer (read)
To read events you must request a collection of streams that have been registered since last update.

handle = this.eventSystem.GetEventReaders<TestEvent>(handle, out IReadOnlyList<Tuple<NativeStream.Reader, int>>readers);

Unlike CreateEventWriter this requires the input dependency of the system and returns an updated handle that you must use. If you are reading in a component system it is safe to pass in default(JobHandle) but you must call Complete() on the returned handle before reading the event streams.

The actual output is from the out field which returns a readonly collection of tuples, with Item1 being the NativeStream.Reader and item2 being the foreachCount. The included foreachCount is there as while it is actually safe for us to read the ForEachCount property of the stream, it unfortunately has a read access check which will throw an error (I believe this check is here because of the ScheduleConstruct option for streams.)

You can then iterate the streams, and create jobs from them. You must pass a handle back after you have scheduled your jobs.,

this.eventSystem.AddJobHandleForConsumer<TestEvent>(handle);

An example could be something like this

Code (CSharp):

1. protected override JobHandle OnUpdate(JobHandle handle)
2.         {
3.             handle = this.eventSystem.GetEventReaders<TestEvent>(handle, out var readers);
4.  
5.             for (var index = 0; index < readers.Count; index++)
6.             {
7.                 var reader = readers[index];
8.  
9.                 handle = new CountJob
10.                     {
11.                         Stream = reader.Item1,
12.                     }
13.                     .Schedule(reader.Item2, 8, handle);
14.             }
15.  
16.             this.eventSystem.AddJobHandleForConsumer<TestEvent>(handle);
17.  
18.             return handle;
19.         }

Things to note

• This is experimental.

I delayed posting until I at least had a test sample out to ensure it was somewhat stable however I do not currently have a project to test it on so the samples are being written to test the workflow to ensure viability as well as catch any bugs.

• You must balance CreateEventWriter<T> with AddJobHandleForProducer<T> and GetEventReaders<T> with AddJobHandleForConsumer<T>.

Safety checks are in place for this and an exception will be thrown if you call 2 CreateEventWriter in a row etc. This is important to know as it is technically safe to do this

Code (CSharp):

1. var event1 = this.eventSystem.CreateEventWriter<UpdateCountEvent>(foreachCount1);
2.  
3. handle = new TestJob
4.     {
5.         Event = event1,
6.     }
7.     .Schedule(foreachCount1, 8, handle);
8.  
9.     // Without this, while the system would actually work safely it will still throw an exception
10.     // this.eventSystem.AddJobHandleForProducer<UpdateCountEvent>(handle);
11.  
12.     var event2 = this.eventSystem.CreateEventWriter<UpdateCountEvent>(foreachCount2);
13.  
14. handle = new TestJob
15.     {
16.         Event = event2,
17.     }
18.     .Schedule(foreachCount2, 8, handle);
19.  
20. this.eventSystem.AddJobHandleForProducer<UpdateCountEvent>(handle);

yet it will still throw an exception. I decided the strict requirement to ensure safety and avoid errors slipping through is more important than the inconvenience of an extra line of code here or there. Let me know.

• NativeStreams don't optimize great with IJobForEach.

They do work and for small entity counts there are no issues but for large entity counts you want to optimize you should switch to IJobChunk.
I'll be posting samples on how to use these streams shortly for various job types.

Implementing your own Event System - Custom worlds or game loops
~coming soon
For now you can look at the existing sample. It demonstrates producing events in the default world, consuming events in a seperate world, and consuming events in a seperate world with a different (fixed) update rate. It will output a result like this.



49-51k events are being produced in the regular world.
The events are being consumed and counted every frame in both a world on Update as well as another World updating on FixedUpdate.
As fixed update only runs 50 times a second but the actual applications fps is nearly 3000 it is generating nearly 3millions events between fixed updates (a more realistic scenario would be producing events in the FixedUpdate and consuming in Update but the majority of Update frames would have 0 events being a dull example).
A thing to note the UI is also powered by sending events back after counting events!

Feedback
Would love to get some feedback on whether other developers think that they could find this useful and/or features they would like to see.

 

~Saved for later

 

I just discovered the Event System and I found it really interesting !!
correct me if im wrong.
1) this package allows us to use all CommandBuffer features in a more performant way and more important in burstable jobs.
2) allows passing events between worlds.
3) replace creation of Event Entities (small lifetime) by simply consuming events.
4) and all this in 10x faster!!

i wrote only the more intersting features for my case!

is it possible to add this package to my project using the Entities 0.1 Version ? (i cant go to the 0.2 for bugs reasons)
* No way to build player for other than Standalone Platform when using Subscenes.
* more garbage collection than the 0.1 version and performance waste.

 

1. It doesn't let you create or manipulate entities so I wouldn't really say this.
2. Yes
3. Yes
4. Yes

Just pull the commit before I updated to 0.2
https://gitlab.com/tertle/com.bovinelabs.event/tree/1b11a39fc8fbdee82ce7445520c013c9dde50e6a
I have added any functionality since 0.2 came out. The only minor changes were to how world messaging was done to be compatible with 0.2 changes.

 

im sorry i read this in the previous thread so i thought it does.

"And as I use quite a few short lived event entities in my project I was curious if I could write something that could generically replaced EntityCommandBuffer for these. So I did!
Firstly code, this is my first, not well tested and quickly written proof of concept.
-edit- dramatically updating code, will post it when i'm done.
-edit2- latested post here, now 3x faster than original version (10x faster than a barrier): https://forum.unity.com/threads/batch-entitycommandbuffer.593569/#post-3965998
How it works. It passes a NativeQueue<T> whenever requested.
Get the barrier with
this.batchBarrier = this.World.GetOrCreateManager<BatchBarrierSystem>();
The same barrier is used for all systems.
To create an entity use
NativeQueue<T>.Concurrent createQueue = this.batchBarrier.GetCreateQueue<T>().ToConcurrent();
where T is an IComponentData that will be added and set on the entity."

ill add it and give you my feedback.
Thanks!

 

One thing I do often in my very basic Entities-based Event system is attach buffers to the event entities. Since that's not possible here, is the idea that the system is so fast that I would just create individual events for each item that I would have put in a buffer?

 

You can pass anything to a native stream, including an array of data, as long as you know what to expect.
Quick example.

Code (CSharp):

1. [BurstCompile]
2. private struct ProduceJob : IJobParallelFor
3. {
4.     public NativeStream.Writer Events;
5.  
6.     public Random Random;
7.  
8.     public void Execute(int index)
9.     {
10.         this.Random.state = (uint)(this.Random.state + index);
11.  
12.         var eventCount = this.Random.NextInt(1, 1000);
13.  
14.         this.Events.BeginForEachIndex(index);
15.      
16.         this.Events.Write(default(TestEventEmpty));
17.      
18.         for (var i = 0; i < eventCount; i++)
19.         {
20.             this.Events.Write(i);
21.         }
22.  
23.         this.Events.EndForEachIndex();
24.     }
25. }
26.  
27. [BurstCompile]
28. public struct ConsumeJob : IJobParallelFor
29. {
30.     public NativeStream.Writer EventCount;
31.     public NativeStream.Reader Stream;
32.  
33.     public void Execute(int index)
34.     {
35.         var count = this.Stream.BeginForEachIndex(index);
36.      
37.         if (count == 0)
38.         {
39.             return;
40.         }
41.      
42.         // First element is TestEventEmpty
43.         var eventData = this.Stream.Read<TestEventEmpty>();
44.      
45.         // Rest of elements are integers
46.         for(var i = 1; i < count; i++)
47.         {
48.             var v = this.Stream.Read<int>();
49.         }
50.      
51.         this.Stream.EndForEachIndex();
52.     }
53. }

 

@tertle there is a small mistake in your manifest line.

"com.bovinelabs.events": "[URL]https://gitlab.com/tertle/com.bovinelabs.event.git[/URL]",

com.bovinelabs.events should be com.bovinelabs.event otherwise the package manager will complain.

 

Cheers fixed.

 

That was the original version which actually created events using entities. This new version ditches entities for performance and a lot more flexibility.

 

Thanks for your event system! I get it from Package Manager with git url, and import the sample folder, but there are a few problems stop me run the sample. How to fix it?

Spoiler

Assets/Samples/BovineLabs Event/0.1.0/Samples/CustomBootstrap.cs(16,36): error CS0535: 'CustomBootstrap' does not implement interface member 'ICustomBootstrap.Initialize(string)'

Spoiler

Assets/Samples/BovineLabs Event/0.1.0/Samples/MultiWorld/UpdateEventSystem.cs(17,34): error CS0115: 'UpdateEventSystem.CustomWorld': no suitable method found to override

 

Thank you for sharing Tertle!

 

If you haven't updated to the latest entities package (0.2) you'll need the version posted earlier in the thread.

 

my fault. I use latest 0.2.0 entities package & 2019.3.0f1 unity.

 

I keep getting:

ArgumentException: NativeStream.Writer must be passed by ref once it is in use.

when using it in an IJobChunk.

When using the example of the IJobParallelFor it works without any issues.

OnUpdate (simplified):

Spoiler

Code (CSharp):

1.   protected override JobHandle OnUpdate(JobHandle inputDeps)
2.   {
3.     EntityQuery jobSeekersQuery = GetEntityQuery(typeof(Citizen), typeof(Translation));
4.  
5.     int numberOfRequests = jobSeekersQuery.CalculateChunkCount();
6.     if (numberOfRequests == 0) return inputDeps;
7.  
8.     var writer = eventSystem.CreateEventWriter<JobFoundEvent>(numberOfRequests);
9.  
10.     inputDeps = new FindJobForCitizen
11.     {
12.       jobDataArray = jobDataArray,
13.       writer = writer,
14.       translationComponentType = GetArchetypeChunkComponentType<Translation>(false),
15.       entitiesEntityType = GetArchetypeChunkEntityType()
16.     }.Schedule(jobSeekersQuery, inputDeps);
17.  
18.     eventSystem.AddJobHandleForProducer<JobFoundEvent>(inputDeps);
19.  
20.     return inputDeps;
21.   }

The Job (simplified):

Spoiler

Code (CSharp):

1.   [BurstCompile]
2.   struct FindJobForCitizen : IJobChunk
3.   {
4.     [DeallocateOnJobCompletion] [ReadOnly] public NativeArray<JobData> jobDataArray;
5.     public NativeStream.Writer writer;
6.  
7.     public void Execute(ArchetypeChunk chunk, int chunkIndex, int firstEntityIndex)
8.     {
9.       for (int i = chunkIndex; i < chunk.Count; i++)
10.       {
11.           // Do stuff
12.  
13.         if (closestJob != Entity.Null)
14.         {
15.           writer.Write(new JobFoundEvent { citizen = entities[i], job = closestJob });
16.         }
17.       }
18.     }
19.   }

 

NativeStream requires
writer.BeginForEachIndex(index);
and
writer.EndForEachIndex();

So your job should look something like this

Code (CSharp):

1. writer.BeginForEachIndex(chunkIndex);
2.  
3. for (int i = chunkIndex; i < chunk.Count; i++)
4. {
5.     // Do stuff
6.  
7.     if (closestJob != Entity.Null)
8.     {
9.         writer.Write(new JobFoundEvent { citizen = entities[i], job = closestJob });
10.     }
11. }
12.  
13. writer.EndForEachIndex();

 

Is it just me being daft or are there currently compatibility issues with the latest Entities (0.3.0 p4)

 

Haven't actually tested with 0.3 yet as I just assumed it'd be fine with 0.2 changes as it didn't seem to change much. I'll take a look shortly.

 

Pushed an update. I think just the samples had an issue. I haven't thoroughly tested it but my sample scene seems to be working as expected again.

 

@tertle just curious, have you tried compare with current ECB state? When we can use it in burst job fully, and structural changes now bursted too (entities 0.4.0)?

 

Nope, was going to look at 0.4 today though. However this has nearly 0 overhead cost so I expect it's still going to be faster regardless.

Entities as events do have benefit of driving system updates but it has some limitations this solution can improve on. If you don't need it though, then entity events are much more promising now after the changes in 0.2 and 0.4

 

@tertle Does your event system automatically order systems that use event system so that all read-write dependencies execute in order in the same frame? For example, if the dependency is A->B->C, would the event system update A->B->C, or is it possible for C->B->A to be the order, resulting in changes taking multiple frames to propagate?

 

It forces systems to be correctly ordered and will throw an exception if they aren't (but will not order them for you.)

The rule is simple : all writers have to come before all readers of the same event type in the same group.

 

Hey. Great work! =)
I am seriously thinking that such or similar system should be part of the Unity DOTS by default.
Definitely want to utilize this solution in my project, so I’ll leave some feedback here in general and on the topic of samples in particular, if I may.

Brief background - I have been quietly developing ZPG (Kind of self-playing game) for quite some time. I made a prototype of a simple ECS vegetation spreading system separately, as It seemed to me a good way to determine a more successful project architecture in terms of entities at the same time with an easily scalable stress test function.
Take a look. At 0:36

Here trees are entities that check environment and also change it.
So there is a "NativeArray<byte>" based grid and the tree processing system that has read access to it.

In the system, the tree grows to its maximum size along the way adding to it's spreading counter. As soon as it is filled in, a random neighboring cell is checked and a new tree is created via CommandBuffer. New entity also receives a temporal tag signaling the need to mark the cell on the grid as forest one/occupied.
The max growth size of the trees and the appropriate cell check in vid is also determined by the common "Unity.Mathematics.noise.snoise" using tree position and constant scaling(0.07f).

With CommandBuffer, even if it is not heavily involved, the consumption of performance seem to be gradually increase along with the number of entities - up to very significant values.
So the first thing I did was separate the tree simulation into a separate world, and I also made my own custom system based on NativeQueue.Concurent where the component with the necessary data is simply transferred, which is then processed and executed to create planned "saplings" or change the grid where it was a requested. Of course, this is much faster than adding / removing components / other stuff with CommandBuffer.
Now, having stumbled upon your com.bovinelabs.entities, I understand that this is about resolving ~same thing.

Why don't you use a similar example with trees for samples if it's possible? The size of the grid determines the scale of the simulation, and the tree growth rate is its intensity. Although I’m not sure that someone will do trees as entities like this - it’s grounded example, easy to do, there is no fancy algorithms and it might be as visual as you would like to even with primitives. If event system can handle something like that - anyone would be convinced that you can build the whole game on this. I mean even if event system does not completely replace command buffer, this would show at least to what extent.

Among other things, you can expand it with the fire system or a basic custom entities culling, since HybridRender does not do it in the most efficient way when there really a lot of entities.

Thank you.

 

If anyone is interested, I've finally got around to thoroughly testing this as I've been using this in a large production project for past month without issue.

 

Super interested! Just grab the latest from git, or have you branched it off somewhere?

 

I'm currently exploring your events system and at this point I"m unsure what the limitations are or if this might be over-kill for some situations. To gain a better understanding of when and how to use the event system, would you be willing to describe some scenarios in which you have utilized the event system in production development?

 

I'm getting a "cannot be called in read mode" on the second Update(), but only if I have my consumer code running. I'm using SystemBase so I'm wondering if I'm doing the dependencies wrong. But it seems nicer to use than ECBs if I can learn how to use it

 

So there is a strict order required for each event.

Writing needs to execute before reading
Reading needs to execute before the event system updates.

(Usually this is just done by job handles and should not need any completes.)

 

Ah okay, I was using UpdateEventSystem from the samples, which I guess gets created for a different World, looking at CustomBootstrap.cs. So combined with [DisableAutoCreation], it was only being created in my default world via my producer/consumer GetOrCreateSystem(), but not added to to update player loop maybe? My mistake, in any case. Using LateSimulationEventSystem instead is what I should have been doing, for anyone messing with the Samples and not trying to do a custom world

 

Had a long weekend and decided to do some work on this and made a range of changes.

• Ordering is no longer strict and you will no longer see errors if you're producers execute after your consumers. Instead read mode is active (first consumer reads) then all producers that request a write stream will write the next frame

Assuming this order
ProducerA -> ConsumerA -> ProducerB -> ConsumerB -> EventSystem

ProducerA events will be read same frame by ConsumerA, ConsumerB
ProducerB events will not be read by any consumer on frame it is sent, instead they will be read next frame.

The primary reason for this change was to allow 2 systems to communicate forward and back. For example

SystemA -> SystemB -> EventSystem

SystemA produces Event1, consumesEvent2
SystemB consumes Event1, produces Event2

This will now work (with SystemA receiving SystemB Event2 the next frame.

• Added extensions, and can be accessed via EventSystem.Ex<T>() where T is the event type. Currently only 1 extension exists but more common patterns will be added in the future.

Code (csharp):

1. EnsureHashMapCapacity<TK, TV>(JobHandle handle, NativeHashMap<TK, TV> hashMap)

This extension ensures a hashmap capacity is large enough to handle storing every event from every writer.
Example usage.

Code (csharp):

1. var handle = es.Ex<TestEvent>().EnsureHashMapCapacity(handle, hashmap);

A quick note on why these extensions are part of the Extension<T> struct is to reduce verbose generic fields. If it was just an extension method (or method in the actual class) it would have to look like this var handle = es.EnsureHashMapCapacity<TestEvent, Entity, RequestNewPathfindingRequest>(handle, hashmap);

• The biggest and probably best change, custom jobs for reading

Was finding it very verbose and tedious so wrote a couple of jobs to handle this in a much more fluid way. No longer have to iterate a bunch of streams and/or worry about handles etc as it's all handled automatically.

1. IJobEvent<T> will just iterate every event, from every writer and pass those to the Execute method. Internally this actually creates 1 job per NativeStream in series, with each ForEachIndex put on a separate thread.

Code (CSharp):

1.  
2.     /// <summary> Job that visits each event. </summary>
3.     /// <typeparam name="T"> Type of event. </typeparam>
4.     public interface IJobEvent<T>
5.         where T : struct
6.     {
7.         /// <summary> Executes the next event. </summary>
8.         /// <param name="e"> The event. </param>
9.         void Execute(T e);
10.     }

Code (CSharp):

1. public static unsafe JobHandle ScheduleParallel<TJob, T>(
2.             this TJob jobData, EventSystem eventSystem, int minIndicesPerJobCount, JobHandle dependsOn = default)

Currently you need to be specify the generic params to schedule this job, but I'd like to avoid this so it'll probably change in the future.

Example of this job in action

Code (CSharp):

1. [BurstCompile]
2.         private struct TestJob : IJobEvent<TestEvent>
3.         {
4.             public NativeQueue<int>.ParallelWriter Counter;
5.  
6.             public void Execute(TestEvent e)
7.             {
8.                 this.Counter.Enqueue(e.Value);
9.             }
10.         }

Code (CSharp):

1.                 var es = this.World.GetOrCreateSystem<TestEventSystem>();
2.  
3.                 var finalHandle = new TestJob
4.                     {
5.                         Counter = counter.AsParallelWriter(),
6.                     }
7.                     .ScheduleParallel<TestJob, TestEvent>(es, 64);

2. IJobEventStream<T> More of a specialized job that creates a job per native stream. Won't mention much except to say it can be executed in series, 1 stream after the other or in parallel, all streams simultaneously as separate jobs. If you find a need for this, check out IJobEventStream and JobStreamTests (or the EnsureHashMapCapacity implementation.)

• A bunch of optimizations, updated to Entities 0.7 and other stuff that I can't remember.

 

Hello, thanks for this great asset^^

I am having an issue, i am trying to write to an event from TriggerJob but since there is no way to get an unique index for TriggerJob(afaik), it gives error ArgumentException: BeginForEachIndex can only be called once for the same index (0).

Is there a 'fix' for this or should i look for a workaround ?

 

I'm working on a significant change at the moment that would make this issue vanish. I've been trying to switch to a custom event container instead of nativestream that no longer requires indices so it works a lot nicer with Entities.ForEach. It was ever so close to working over weekend but last minute found a very rare timing issue.

 

Can't wait for the new version. I'm still using your old event system version (without stream), but would like to try the new version

 

Hey tertle, I'm doing pretty much the same thing.Will send my code to github later. It's in my private gitlab.
I'm using NativeStream for my event system and working on a new custom container as NativeStream has some limitations. In my version, Event data type is fully customizable, you can have any number of any kinds of unmanaged struct Added to and event.

 

Actually it's been released for a week I just haven't got around to writing any documentation for it which is why I haven't mentioned it yet.

But TLDR; no longer need BeginForEach/EndForEach on the writer and can just use it in Entities.ForEach

If you're interested in the container can be viewed here: https://gitlab.com/tertle/com.bovinelabs.event/-/tree/master/BovineLabs.Event/Containers

The reader is near identical to NativeStream, the writer has a fixed size (where size is the max thread count)

at high counts, it's 7.75x faster than native queue to read but only 10% slower to write



and it's 1.7x faster than NativeStream at both reading/writing if writing is done from an Entities.Foreach



As for the event system, I can write something like 5 billion events/second/60fps on a 3900x (yes it's a lot of cores and not a fair cpu to test on)

When I get some time I'll provide more details.

-edit- noticed a unit test failing though not sure if that's just my local copy because i'm pretty certain version i put online was passing. investigating.
-edit2- ah just for different update systems, pretty sure it's also a false positive due to how i changed safety.

 

Hi @tertle,

That a great work you have done here ! love it ! I wish unity implements this kind of system natively.

I have a question for you, do you have plan for have an IJobEvent runnable in mono-thread ? because now, I need to use a nativequeue to merge my datas and then begin the work on it, it's not optimal :/

Thank for you work

 

Fixed the unit tests (and 1 issue.)

I'll do a write up this weekend if I find time.

I'm not certain what you are asking for and I'd need to understand what you're trying to do, but I'd say no at this stage?

The IJobEvent interface was implemented to try avoid the need to for grouping.

 

I ask a Schedule method for IJobEvent.
Actually I do ScheduleParallel->NativeQueue->Filter and I want do Schedule->Filter

My goal is to filter events because they can be duplicate .

I don't know if I'm understandable here :/

 

Hi @tertle ,

Sorry to bother, I see you moved away from the native stream provited by unity to make your own native container but you first post mentionned coming explainations about the foreachcount in the unity native stream implementation.

I can't seem to find any explaination of what it is and how to determine the most suitable value for it.

As you seemed to know what it is would you care to give a bit more details ?

Thanks

 

Hello @tertle . Thanks for sharing again.

I'm have a hard time using the Readers. After some experimentation, I used IJobEvent.
The only thing is that I want to use a ConcurentCommandBuffer inside, but I don't have access to the Job Index in the Execute function.

Also, is there any way to use the Readers with Job.WithCode ? I'm trying to use the "new" syntax of DOTS

EDIT: Not sure if it's the right way, but seems to work, I passed the jobIndex in
IJobEvent.cs line 142:
fullData.JobData.Execute(e,jobIndex);

 

I'd usually just inject thread index for my command buffer but I can see it making sense to pass in a unique event key. Will look at doing that.

As for Job.WithCode, yes you can read back in this but you need to queue them sequential/parallel yourself. I found this very tedious which is why I wrote a custom job for it.

Something like (wrote in notepad, might have syntax error)

Code (CSharp):

1. this.Dependency = this.eventSystem.GetEventReaders<YourEventType>(this.Dependency, out var events);
2.  
3. for (var i = 0; i < events.Count; i++)
4. {
5.     var reader = events[i];
6.  
7.     this.Job.WithCode(() => {
8.  
9.         for(var i = 0; i < reader.ForEachCount; i++)
10.         {
11.             var count = reader.BeginForEachIndex(begin);
12.  
13.             for (var j = 0; j < count; j++)
14.             {
15.                 var e = reader.Read<YourEventType>();
16.            
17.                 // Do stuff
18.             }
19.  
20.             reader.EndForEachIndex();
21.         }
22.     }).Schedule();
23.  
24.     this.eventSystem.AddJobHandleForConsumer<T>(this.Dependency);
25. }

You also lose all your threading potential. This would more palatable if there was a IJobParallalFor equivalent of Job.WithCode

Think of NativeStream as an array of queues (it's not the same but it's good for explaining)
The ForEach index is the size of the array.
If you're working with entities, ideally this would be your chunk count and that's how you grouped your entities.
However this doesn't work with Entities.ForEach it needs to be your entity count which is inefficient which is why I wrote my own version.

 

Thanks for your answer.

I'm working on a set of system where I want to add the same type of struct to a native container in parallel and from different jobs.

I don't know in advance how many of that struct I'll add to the native container because they are part of an IBufferElement that can change from entity to entity. So I have no clue has to how I can set a propoer value for the foreach count. Not to mention that I assume I'll need it also for reading the stream and that it needs to be the same values.

For now I'm using a nativequeue wich support parrallel write and is very convinient to use.

However it also means I'm stuck with single threaded read.

Do you think native stream or your native container could solve my problem?

Edit : just re read your reply, if native stream are to be thougth of as array of queues, then does it mean that for my use case I can just go with a value of 1? Because I only have one type of data to push in.

Or is it more accurate to think of it as a list of array with the foreachcount being the size of the array?

 

ForEach count isn't how much data you're writing, it's how wide you're writing the data.
I think what you're trying to do is this?

Code (CSharp):

1.             var nativeStream = new NativeStream(this.query.CalculateEntityCount(), Allocator.TempJob);
2.             var writer = nativeStream.AsWriter();
3.  
4.             this.Entities.ForEach((int entityIndexInQuery, in DynamicBuffer<Element> buffer) =>
5.                 {
6.                     writer.BeginForEachIndex(entityIndexInQuery);
7.  
8.                     for (var i = 0; i < buffer.Length; i++)
9.                     {
10.                         writer.Write(buffer[i].Value);
11.                     }
12.  
13.                     writer.EndForEachIndex();
14.                 })
15.                 .WithStoreEntityQueryInField(ref this.query)
16.                 .ScheduleParallel();

But you can't do this from different jobs.

 

I'm actually in a context of an ITriggerEvent job from the physics package. So I understand I would need to get the number of trigger event beeing processed.

And as far as I can see your are never sharing the same nativestream between several systems.

 

Thanks for the answer.
I guess I would need to do the same thing if I wanted to iterate on the events on the MainThread ?
I'm using this Event to instantiate and setup entity which currently, with my setup, is way easier to do on MainThread with the EntityManager .

I also did another modification and I'm pretty sure it might not be thread safe but if you know what you're doing, it can be pretty usefull:

Code (CSharp):

1. fullData.JobData.Execute(ref fullData.Reader.Read<T>(),jobIndex);

It's passing the Reader as "ref", like that you can use the simple IJobEvent in a job that can modify the event and then schedule another job with the same readers with data modified.
But again, and you can tell me, I guess that would be pretty bad if you don't handle your dependency correctly

 

You can read from multiple systems fine. The system even supports reading in different update rates (one of main features.) You can write in some world running at 50fps and read it in another world running 10fps and you won't get duplicate events, they might just be up to 5 frames delayed obviously.

You use a different native stream for writing for each system. This is why a list of streams is returned from the reader - each stream is from a different system.

Safe only if you are only reading these events from 1 system ever. If you have 2 different systems reacting to events this is definitely not safe. 2 systems can read simultaneously. Because of this I would not recommend it as one of the perks is being able to react in multiple places - for example can have a stat tracking or achievement system running at same time.

 

@tertle Thanks for all your explainations. I think I understand a bit better know.
I managed to get my systems to work with native queues, I'll stick with that for know and see if the performances encourgae me to revisit the matter.

 

Hey @tertle did you have any issue when using .asmdef ?
https://forum.unity.com/threads/burst-error-when-using-asmdef.869428/#post-5722135

I have Burst errors and I only have them when events a triggered.

Maybe you could help ?

Thanks !