Passing NativeContainers between systems dependency issue

I have two systems that share one NativeQueue. SystemA enqueues to the NativeQueue and SystemB dequeues from the NativeQueue. Upon entering play mode i receive the following error:

InvalidOperationException: The previously scheduled job SystemA:EnqueueJob writes to the NativeArray EnqueueJob.AnNativeQueue. You are trying to schedule a new job SystemB;DequeueJob, which writes to the same NativeArray (via DequeueJob.AnNativeQueue). To guarantee safety, you must include SystemA:EnqueueJob as a dependency of the newly scheduled job.

I understand that the error is telling me to pass a dependency or the appropriate inputDeps between the two systems. However, I am already returning both job's dependencies each OnUpdate and this does work fine between systems not sharing a NativeContainer.

So my question is: How do i properly let systems know that NativeContainers are used in jobs across multiple systems?

My two systems:

Spoiler: SystemA

Code (CSharp):

1. using Unity.Collections;
2. using Unity.Entities;
3. using Unity.Jobs;
4.  
5. class SystemA : JobComponentSystem
6. {
7.  
8.     public static NativeQueue<int> AnNativeQueue;
9.  
10.     public static JobHandle Handle_SystemA;
11.  
12.     public struct EnqueueJob : IJob
13.     {
14.         public NativeQueue<int> AnNativeQueue;
15.  
16.         public void Execute()
17.         {
18.             AnNativeQueue.Enqueue(5);
19.         }
20.     }
21.  
22.     protected override JobHandle OnUpdate(JobHandle inputDeps)
23.     {
24.  
25.         //inputDeps = JobHandle.CombineDependencies(inputDeps, SystemB.Handle_SystemB);
26.  
27.         var EnqueueJob = new EnqueueJob
28.         {
29.             AnNativeQueue = AnNativeQueue,
30.         };
31.         inputDeps = EnqueueJob.Schedule(inputDeps);
32.  
33.         //Handle_SystemA = inputDeps;
34.  
35.         return inputDeps;
36.     }
37.  
38.     protected override void OnCreateManager()
39.     {
40.         base.OnCreateManager();
41.  
42.         AnNativeQueue = new NativeQueue<int>(Allocator.Persistent);
43.  
44.     }
45.  
46.     protected override void OnDestroyManager()
47.     {
48.         base.OnDestroyManager();
49.  
50.         if (AnNativeQueue.IsCreated)
51.         {
52.             AnNativeQueue.Dispose();
53.         }
54.     }
55. }

Spoiler: SystemB

Code (CSharp):

1. using Unity.Collections;
2. using Unity.Entities;
3. using Unity.Jobs;
4.  
5. class SystemB : JobComponentSystem
6. {
7.  
8.     public static JobHandle Handle_SystemB;
9.  
10.     public struct DequeueJob : IJob
11.     {
12.         public NativeQueue<int> AnNativeQueue;
13.  
14.         public void Execute()
15.         {
16.             while (AnNativeQueue.TryDequeue(out int item))
17.             {
18.  
19.             }
20.         }
21.     }
22.  
23.     protected override JobHandle OnUpdate(JobHandle inputDeps)
24.     {
25.  
26.         //inputDeps = JobHandle.CombineDependencies(inputDeps, SystemA.Handle_SystemA);
27.  
28.         var DequeueJob = new DequeueJob
29.         {
30.             AnNativeQueue = SystemA.AnNativeQueue,
31.         };
32.         inputDeps = DequeueJob.Schedule(inputDeps);
33.  
34.         //Handle_SystemB = inputDeps;
35.  
36.         return inputDeps;
37.     }
38. }

When scheduling both jobs in just a single system there is no error. This let me to try to manually pass the dependencies around. Passing them both ways makes the error go away. However this seems like a non optimal solution and applying this method to my project with multiple systems introduces new errors.

 

This thread helped me work through a similar issue last night.

https://forum.unity.com/threads/sol...-injected-system-messes-up-dependency.601780/

 

You shouldn't use statics on the systems but rather inject one system into another (or alternative to inject since it is deprecated).
What you have to to there is store each system dependency and combine them.

Something like this should work:

Code (CSharp):

1.  
2. class SystemA : JobComponentSystem {
3.  
4.   public JobHandle InputDeps {
5.     get;
6.     private set;
7.   }
8.  
9.   SystemB m_systemB;
10.  
11.   public NativeQueue<int> AnNativeQueue;
12.  
13.   protected override void OnCreateManager() {
14.     base.OnCreateManager();
15.     AnNativeQueue = new NativeQueue<int>(Allocator.Persistent);
16.     m_systemB = World.Active.GetOrCreateManager<SystemB>();
17.   }
18.  
19.   protected override void OnDestroyManager() {
20.     base.OnDestroyManager();
21.  
22.     if (AnNativeQueue.IsCreated) {
23.       AnNativeQueue.Dispose();
24.     }
25.   }
26.  
27.   public struct EnqueueJob : IJob {
28.     [WriteOnly]
29.     public NativeQueue<int> AnNativeQueue;
30.  
31.     public void Execute() {
32.       AnNativeQueue.Enqueue(5);
33.     }
34.   }
35.  
36.   protected override JobHandle OnUpdate(JobHandle inputDeps) {
37.     inputDeps = JobHandle.CombineDependencies(inputDeps, m_systemB.InputDeps);
38.     inputDeps = new EnqueueJob {
39.       AnNativeQueue = AnNativeQueue,
40.     }.Schedule(inputDeps);
41.     InputDeps = inputDeps;
42.     return inputDeps;
43.   }
44. }
45.  
46. class SystemB : JobComponentSystem {
47.  
48.   public JobHandle InputDeps {
49.     get;
50.     private set;
51.   }
52.  
53.   SystemA m_systemA;
54.  
55.   protected override void OnCreateManager() {
56.     base.OnCreateManager();
57.     m_systemA = World.Active.GetOrCreateManager<SystemA>();
58.   }
59.  
60.   public struct DequeueJob : IJob {
61.     public NativeQueue<int> AnNativeQueue;
62.  
63.     public void Execute() {
64.       while (AnNativeQueue.TryDequeue(out int item)) {
65.  
66.       }
67.     }
68.   }
69.  
70.   protected override JobHandle OnUpdate(JobHandle inputDeps) {
71.     inputDeps = JobHandle.CombineDependencies(inputDeps, m_systemA.InputDeps);
72.     inputDeps = new DequeueJob {
73.       AnNativeQueue = m_systemA.AnNativeQueue,
74.     }.Schedule(inputDeps);
75.  
76.     InputDeps = inputDeps;
77.     return inputDeps;
78.   }
79. }

 

I tried it the way you listed here and I still have the same issue. Is there supposed to be a UpdateBefore or After on either of those (or an AlwaysUpdate for queues? I saw someone mention that it was important somewhere for some reason) or does doing the dependencies that way make it so that is not needed? I thought I had it working, well, I sort of did. I did it similar to the one in my link above where I manually called complete on the first job and just passed over the dependencies into the second job.

Doing it that way, I was able to Dequeue my item into the second job but then it would start throwing that same error he had above. As of now, trying it the way you have listed above, it never gets to the point of dequeuing before the error starts popping.

 

No need to set those. I don't have any error with that exact code above.

 

Are you doing concurrent Enqueue?
If so you must storea version of the concurrent container when you create the system otherwise you will be reading on the main thread.
The system that enqueues should then look like this:

Code (CSharp):

1. class SystemA : JobComponentSystem {
2.  
3.   public JobHandle InputDeps {
4.     get;
5.     private set;
6.   }
7.  
8.   SystemB m_systemB;
9.  
10.   public NativeQueue<int> AnNativeQueue;
11.   private NativeQueue<int>.Concurrent m_concurrentNativeQueue;
12.  
13.   protected override void OnCreateManager() {
14.     base.OnCreateManager();
15.     AnNativeQueue = new NativeQueue<int>(Allocator.Persistent);
16.     m_concurrentNativeQueue = AnNativeQueue.ToConcurrent();
17.     m_systemB = World.Active.GetOrCreateManager<SystemB>();
18.   }
19.  
20.   protected override void OnDestroyManager() {
21.     base.OnDestroyManager();
22.  
23.     if (AnNativeQueue.IsCreated) {
24.       AnNativeQueue.Dispose();
25.     }
26.   }
27.  
28.   public struct EnqueueJob : IJobParallelFor {
29.     [WriteOnly]
30.     public NativeQueue<int>.Concurrent AnNativeQueue;
31.  
32.     public void Execute(int index) {
33.       AnNativeQueue.Enqueue(index);
34.     }
35.   }
36.  
37.   protected override JobHandle OnUpdate(JobHandle inputDeps) {
38.     inputDeps = JobHandle.CombineDependencies(m_systemB.InputDeps, inputDeps);
39.     inputDeps = new EnqueueJob {
40.       AnNativeQueue = m_concurrentNativeQueue,
41.     }.Schedule(100, 8, inputDeps);
42.     InputDeps = inputDeps;
43.     return inputDeps;
44.   }
45. }

 

Thank you both for your suggestions, also for attending me to use GetOrCreateManager<System> instead of static variables.

Ill be using the manually passing around dependencies method between the two systems for now. Though by using this method I can already see myself forgetting to pass one and debug for hours trying to find the source of error.

Hopefully in the future there will be a more concise and automated method of notifying jobs about the dependency on NativeContainers.

 

Ah, ok. l understand. I was not storing a separate concurrent queue. I was just passing in the original with ToConcurrent() I will try that after work. Thanks!

 

I agree. I don't use such approach myself.
Why don't you use a more suitable container where you just need read only access to get the values? If you use a NativeHashMap/NativeMultiHashMap or NativeList (if no need for concurrent write) you could get rind of those dependencies

EDIT: Seems like you will end up with the same problem. Guess that for now the only way is to pass those dependencies