Is There a Way to Iterate Keys of NativeMultiHashMap in a Job?

I have a NativeMultiHashMap<Entity, number> which is passed into an IJobForEach, which will concurrently write K:V pairs into the hash map. I want to schedule a second job which will iterate for each key and sum all of the values that each key maps to.

IJobNativeMultiHashMapVisitKeyValue iterates each K:V pair, but I don't think there's any way to prevent concurrency issues from visiting two pairs with the same key at the same time. IJobParallelFor(Batch/Defer) with GetKeyArray() has similar issues, and additionally needs some way to defer getting the array. But if you could get the key array and filter it to only have unique elements, ignore duplicates, or ensure that duplicates are always in the same thread, it would work.

I understand that there is an
IJobNativeMultiHashMapMergedSharedKeyIndices, which will ensure that a given key is only visited in a single thread, but this applies only to MultiHashMap<int, int>. I suspect there's a way to use this by putting the Entity, number info into NativeArrays and indexing into them, but I can't quite wrap my head around how that would work.

Any thoughts? This problem comes up several times in my project. It seems like it should be a simple producer/consumer sort of pattern but I can't figure out a good way to approach it.

Something like this would be nice:

Code (CSharp):

1. IterateKeysJob : IJobNativeMultiHashMapVisitEachKey<TKey, TValue>
2.         {
3.             NativeMultiMashMap<TKey, TValue> hashmap;
4.  
5.             public void Execute(TKey key)
6.             {
7.                 // can use hsahmap.TryGetFirstValue/NextValue
8.             }
9.  
10.         }

Or a generic version of IJobNativeMultiHashMapMergedSharedKeyIndices. Why isn't there a generic version of that?

 

That's not true as far as I'm aware. Due to how NativeMultiHashMaps works each key will only exist in a single thread. You will never get a KVP that shares a key on a different thread. If you only have 1 key and a lot of values, you will only use 1 thread.

This is very useful to know and a great pattern for certain things. For example this is safe

Code (CSharp):

1.     public struct ReduceHealthJob : IJobNativeMultiHashMapVisitKeyValue<Entity, int>
2.     {
3.         [NativeDisableParallelForRestriction]
4.         public ComponentDataFromEntity<Health> Healths;
5.  
6.         public void ExecuteNext(Entity key, Entity value)
7.         {
8.             var health = Healths[key];
9.             health.value -= int;
10.             Healths[key] = health;
11.         }
12.     }

 

That's pretty much exactly what I needed, thanks! Is the thread-safe behavior mentioned in the docs?

 

Not that I'm aware of.

I only realized this when I wrote my own hashmap jobs a while ago and looked at code (before Unity had them which they ended up implementing nearly identically so should hold)

It would be good to get official word on it to confirm because I have used this quite heavily and never had an issue but would kind of suck if behaviour changed one day.

You have got me curious enough to check again, so might have a quick look

 

Yep my test seems to confirm what I understood.

Here's my test if you're curious, 1 million elements and tested at 1,10 and 1000 keys.

Code (CSharp):

1. public class MapTest
2.     {
3.         [TestCase(1)]
4.         [TestCase(10)]
5.         [TestCase(1000)]
6.         [Test]
7.         public void CheckThreadSafety(int keys)
8.         {
9.             int count = 1000000;
10.  
11.             var data = new NativeMultiHashMap<int, byte>(count, Allocator.TempJob);
12.  
13.             for (var i = 0; i < count; i++)
14.             {
15.                 data.Add(Random.Range(0, keys), 0);
16.             }
17.  
18.             var output = new NativeMultiHashMap<int, int>(count, Allocator.TempJob);
19.  
20.             new JobTest
21.                 {
22.                     Output = output.AsParallelWriter(),
23.                 }
24.                 .Schedule(data, 1).Complete();
25.  
26.             for (var i = 0; i < keys; i++)
27.             {
28.                 if (!output.TryGetFirstValue(i, out var threadIndex, out var it))
29.                 {
30.                     continue;
31.                 }
32.              
33.                 // Debug.Log(threadIndex); // quick visual test to make sure we're actually running on multiple threads
34.  
35.                 while (output.TryGetNextValue(out var index, ref it))
36.                 {
37.                     Assert.AreEqual(threadIndex, index);
38.                 }
39.             }
40.  
41.             data.Dispose();
42.             output.Dispose();
43.         }
44.  
45.         //[BurstCompile]
46.         private struct JobTest : IJobNativeMultiHashMapVisitKeyValue<int, byte>
47.         {
48.             [NativeSetThreadIndex]
49.             public int ThreadIndex;
50.  
51.             public NativeMultiHashMap<int, int>.ParallelWriter Output;
52.  
53.             public void ExecuteNext(int key, byte value)
54.             {
55.                 Output.Add(key, this.ThreadIndex);
56.  
57.                 // Busy work to make sure we use multiple threads
58.                 FindPrimeNumber(10);
59.             }
60.  
61.             public long FindPrimeNumber(int n)
62.             {
63.                 int count=0;
64.                 long a = 2;
65.                 while(count<n)
66.                 {
67.                     long b = 2;
68.                     int prime = 1;// to check if found a prime
69.                     while(b * b <= a)
70.                     {
71.                         if(a % b == 0)
72.                         {
73.                             prime = 0;
74.                             break;
75.                         }
76.                         b++;
77.                     }
78.                     if(prime > 0)
79.                     {
80.                         count++;
81.                     }
82.                     a++;
83.                 }
84.                 return (--a);
85.             }
86.         }
87.     }

All keys execute on same thread.

With a single key

Code (CSharp):

1. CheckThreadSafety(1) (1.234s)
2. ---
3. 25

Only executes on 1 thread

With multiple keys executes over multiple threads

Code (CSharp):

1. CheckThreadSafety(10) (0.597s)
2. ---
3. 25
4. 25
5. 1
6. 1
7. 2
8. 2
9. 2
10. 3
11. 1
12. 3

But the key is only ever seen on the same thread.

result seems still valid even if I mix up a bunch of data like this

Code (CSharp):

1. for (var i = 0; i < keys / 2; i++)
2.             {
3.                 data.Remove(i);
4.             }
5.  
6.             var newMax = keys + keys / 2;
7.             for (var i = 0; i < count / 2; i++)
8.             {
9.                 data.Add(Random.Range(0, newMax), 0);
10.             }