NativeQueue Struggles

I am trying to queue items up in one script, and dequeue them while launching on separate threads in another script. Everytime my queue is over a set number, it should launch a bunch of separate jobs. I don't really care about waiting to see if they complete as they're bursting messages over UDP.

I've put together a small sample of what I'm trying to do, but I'm not successfully passing the information from one script into another. I regularly get errors on object references not set to an instance of an object.

I'd greatly appreciate some input!

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using UnityEngine;
4. using Unity.Collections;
5. using Unity.Jobs;
6. using UnityEngine.Profiling;
7. using Unity.Entities;
8.  
9. public class test_creator : MonoBehaviour
10. {
11.     public NativeQueue<float> myQueue = new NativeQueue<float>(Allocator.Persistent);
12.     public NativeArray<float> myArray;
13.  
14.     void FixedUpdate()
15.     {
16.         // Populate data
17.         myQueue.Enqueue(1.0f);
18.         Debug.Log("creator Queue size " + myQueue.Count);
19.     }
20.  
21.  
22. }

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using UnityEngine;
4. using Unity.Collections;
5. using Unity.Jobs;
6. using UnityEngine.Profiling;
7. using Unity.Entities;
8.  
9. public class test_doer : MonoBehaviour
10. {
11.     [HideInInspector]
12.     public test_creator thing;
13.     public NativeQueue<float> myQueue = new NativeQueue<float>(Allocator.Persistent);
14.  
15.     // Start is called before the first frame update
16.     void Start()
17.     {
18.        
19.     }
20.  
21.     public struct SerializeJob : IJob
22.     {
23.         public NativeArray<float> myArray;
24.  
25.         public void Execute()
26.         {
27.             // serialize data
28.             Debug.Log("myArray0" + myArray[0]);
29.         }
30.     }
31.  
32.     // Update is called once per frame
33.     void Update()
34.     {
35.         myQueue = thing.myQueue;
36.         while (myQueue.Count > 2)
37.         {
38.             Debug.Log("Queue count " + myQueue.Count);
39.             var myArray = new NativeArray<float>(2, Allocator.Temp);
40.             myArray[0] = myQueue.Dequeue();
41.             myArray[1] = myQueue.Dequeue();
42.             // launch paralell job
43.             SerializeJob jobData = new SerializeJob();
44.             jobData.myArray = myArray;
45.             // Schedule the job
46.             JobHandle handle = jobData.Schedule();
47.  
48.             // Wait for th ejob to complete
49.             handle.Complete();
50.  
51.             jobData.myArray.Dispose();
52.             myArray.Dispose();
53.         }
54.     }
55. }
56.  

 

Null Reference exceptions should be simple to track down. What's null in your code? Maybe you need to be initializing something in Awake?

 

I don't understand NativeQueue's well enough to figure it out.

I think

Code (CSharp):

1. thing.myQueue

is null.

 

Well, it should tell you the line number of the NRE, for starters, in the error. And you should be able to just step through the code in the debugger until you discover what's null.

Based on your earlier code, nothing is setting "thing" to a value. So, accessing any property off "thing" would give you an NRE. Or do you have some other code that sets "thing" to a value?

Anyway, attach the debugger and step through the code.

 

Are you really enqueuing inside the main thread, or are you enqueueing from another thread? You may not realize that you are enqueueing from another thread.

A common situation of receive null pointer exceptions in queueing code is using a non-thread-safe data structure. It should be clear that the C# job system is not thread-safe in the conventional sense of the term.

This code omits too many of the details to really identify the issue. Try to not use the C# job system to solve this problem. If you are enqueuing data from e.g. a network thread, use a true (not off-the-Internet gist that people do) ConcurrentQueue from .Net 4.0 or later, and dequeue from it in an Update on the main thread.

 

I have data that I'm generating in Unity. I want to enqueue data off the main thread. I don't particularly care about how it gets dequeued. Using a concurrent method is probably best for dequeueing. If there's a way to trigger an "event" of passing X number of items, that would work as well, instead of checking every update.

Current errors:

• A Native Collection has not been disposed, resulting in a memory leak. It was allocated at ...\Test_creator.cs:11.

• NullReferenceException: Object reference not set to an instance of an object. test_doer.Update() (at Assets/Scripts/test/test_doer.cs:30)[/code]

test_creator.cs:

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using UnityEngine;
4. using Unity.Collections;
5. using Unity.Jobs;
6. using UnityEngine.Profiling;
7. using Unity.Entities;
8.  
9. public class test_creator : MonoBehaviour
10. {
11.     public NativeQueue<float> myQueue = new NativeQueue<float>(Allocator.Persistent);
12.     //public NativeArray<float> myArray;
13.  
14.     void FixedUpdate()
15.     {
16.         // Populate data
17.         myQueue.Enqueue(1.0f);
18.         Debug.Log("creator Queue size " + myQueue.Count);
19.     }
20.  
21.     private void OnDestroy()
22.     {
23.         myQueue.Dispose();
24.     }
25. }

test_doer.cs

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using UnityEngine;
4. using Unity.Collections;
5. using Unity.Jobs;
6. using UnityEngine.Profiling;
7. using Unity.Entities;
8.  
9. public class test_doer : MonoBehaviour
10. {
11.     [HideInInspector]
12.     public test_creator thing;
13.     //public NativeQueue<float> myQueue;// = thing.myQueue; //new NativeQueue<float>(Allocator.Persistent);
14.  
15.     public struct SerializeJob : IJob
16.     {
17.         public NativeArray<float> myArray;
18.  
19.         public void Execute()
20.         {
21.             // serialize data
22.             Debug.Log("myArray0" + myArray[0]);
23.         }
24.     }
25.  
26.     // Update is called once per frame
27.     void Update()
28.     {
29.         //myQueue = thing.myQueue;
30.         while (thing.myQueue.Count > 2)
31.         {
32.             Debug.Log("Queue count " + thing.myQueue.Count);
33.             var myArray = new NativeArray<float>(2, Allocator.Temp);
34.             myArray[0] = thing.myQueue.Dequeue();
35.             myArray[1] = thing.myQueue.Dequeue();
36.             // launch paralell job
37.             SerializeJob jobData = new SerializeJob();
38.             jobData.myArray = myArray;
39.             // Schedule the job
40.             JobHandle handle = jobData.Schedule();
41.  
42.             // Wait for th ejob to complete
43.             handle.Complete();
44.  
45.             jobData.myArray.Dispose();
46.             myArray.Dispose();
47.         }
48.     }
49. }
50.  

 

First, where are you referencing test_creator in test_doer? The reference has a [HideInInspector] and no code to set the parameter.

Second, avoid initialising Native Containers on declaration. It'll cause problems on prefabs. Move the declaration to Awake.

Third, NativeQueue doesn't support parallel dequeue, only enqueue.

[]'s

 

The following code seems to work for anyone else that might come across a similar issue.
I moved the variable definition to Awake() per elcionap's suggestions. By making the variable static, I'm able to reference it from the script's name instead.

test_creator.cs

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using UnityEngine;
4. using Unity.Collections;
5. using Unity.Jobs;
6. using UnityEngine.Profiling;
7. using Unity.Entities;
8.  
9. public class test_creator : MonoBehaviour
10. {
11.     public static NativeQueue<float> myQueue;
12.     void Awake()
13.     {
14.     myQueue = new NativeQueue<float>(Allocator.Persistent);
15.     }
16.  
17.     void FixedUpdate()
18.     {
19.         // Populate data
20.         myQueue.Enqueue(1.0f);
21.         Debug.Log("creator Queue size " + myQueue.Count);
22.     }
23.  
24.     private void OnDestroy()
25.     {
26.         myQueue.Dispose();
27.     }
28. }

test_doer.cs

Code (CSharp):

1. using System.Collections;
2. using System.Collections.Generic;
3. using UnityEngine;
4. using Unity.Collections;
5. using Unity.Jobs;
6. using UnityEngine.Profiling;
7. using Unity.Entities;
8.  
9. public class test_doer : MonoBehaviour
10. {
11.     public struct SerializeJob : IJob
12.     {
13.         public NativeArray<float> myArray;
14.  
15.         public void Execute()
16.         {
17.             // serialize data
18.             Debug.Log("myArray0" + myArray[0]);
19.         }
20.     }
21.  
22.     // Update is called once per frame
23.     void Update()
24.     {
25.         while (test_creator.myQueue.Count >= 2)
26.         {
27.             Debug.Log("Queue count " + test_creator.myQueue.Count);
28.             var myArray = new NativeArray<float>(2, Allocator.TempJob);
29.             myArray[0] = test_creator.myQueue.Dequeue();
30.             myArray[1] = test_creator.myQueue.Dequeue();
31.             // launch paralell job
32.             SerializeJob jobData = new SerializeJob();
33.             jobData.myArray = myArray;
34.             // Schedule the job
35.             JobHandle handle = jobData.Schedule();
36.  
37.             // Wait for the job to complete
38.             handle.Complete();
39.  
40.             jobData.myArray.Dispose();
41.         }
42.     }
43. }