Error when trying to access NativeMultiHashmap from two jobs that should run sequentially

Entity Component System (ECS)
, ,
1

Hey guys,

I have a system that schedules two jobs with the second one depending on the first one. The first one, writes to a NativeMultiHashMap using the ParallelWriter while the second one reads from that map and does some other stuff based on it. I was assuming that when I pass the JobHandle get from scheduling the first job to the schedule function of the second one, that there shouldn’t be any conflicts regarding access to the map. Still, I get the following error:

InvalidOperationException: The previously scheduled job EventHandlerSystem:OnUpdate_LambdaJob0 writes to the Unity.Collections.NativeMultiHashMap`2[Unity.Entities.Entity,System.Int32] OnUpdate_LambdaJob0.JobData.constantIrrigationChangesPV. You must call JobHandle.Complete() on the job EventHandlerSystem:OnUpdate_LambdaJob0, before you can read from the Unity.Collections.NativeMultiHashMap`2[Unity.Entities.Entity,System.Int32] safely.

So, either I have a fundamental misunderstanding of how the dependencies work or I’m missing something here. I’d appreciate it if you could point out to me what might be going wrong here. When I force the first job to Complete(), everything works as expected but as mentioned I would assume that this is the case usig the dependencies. Anyway, here is the complete code for my system. I’m aware that I don’t really need to do all the JobHandle combines since some of that is done implicitly as I understand it (for the Entities.ForEach at least) but I just wanted to make it super explicit and see if that changes anything (well, it doesn’t):

[UpdateInGroup(typeof(SimulationSystemGroup))]
public class EventHandlerSystem : SystemBase
{
    EndSimulationEntityCommandBufferSystem commandBufferSystem;
    private EntityQuery eventEntitiesQuery;

    protected override void OnCreate()
    {
        // Acquire/create command buffer
        commandBufferSystem = World.GetOrCreateSystem<EndSimulationEntityCommandBufferSystem>();
    }

    protected override void OnUpdate()
    {
        EntityCommandBuffer.ParallelWriter commandBuffer = commandBufferSystem.CreateCommandBuffer().AsParallelWriter();

        int eventCount = eventEntitiesQuery.CalculateEntityCount();

        NativeMultiHashMap<Entity, int> constantIrrigationChanges = new NativeMultiHashMap<Entity, int>(eventCount, Allocator.TempJob);
        NativeMultiHashMap<Entity, int>.ParallelWriter constantIrrigationChangesPV = constantIrrigationChanges.AsParallelWriter();

        JobHandle accumulateJobDependency = Entities
            .WithStoreEntityQueryInField(ref eventEntitiesQuery)
            .ForEach((in ConstantIrrigationChangeEvent constantIrrigationChange, in int entityInQueryIndex, in Entity entity) =>
            {
                constantIrrigationChangesPV.Add(constantIrrigationChange.targetEntity, constantIrrigationChange.irrigationChange);
                commandBuffer.DestroyEntity(entityInQueryIndex, entity);

            }).WithBurst()
              .ScheduleParallel(this.Dependency);

        this.Dependency = JobHandle.CombineDependencies(this.Dependency, accumulateJobDependency);

        UpdateConstantIrrigationJob updateJob = new UpdateConstantIrrigationJob()
        {
            mapKeys = constantIrrigationChanges.GetKeyArray(Allocator.TempJob),
            map = constantIrrigationChanges,
            constantIrrigationData = GetComponentDataFromEntity<ConstantIrrigation>(false),
            ecb = commandBuffer
        };

        JobHandle updateJobDependency = updateJob.Schedule(constantIrrigationChanges.Count(), 10, this.Dependency);

        this.Dependency = JobHandle.CombineDependencies(this.Dependency, updateJobDependency);

        constantIrrigationChanges.Dispose(this.Dependency);

        commandBufferSystem.AddJobHandleForProducer(this.Dependency);
    }

    protected override void OnDestroy()
    {
    }

    [BurstCompatible]
    public struct UpdateConstantIrrigationJob : IJobParallelFor
    {
        [ReadOnly]
        [DeallocateOnJobCompletion]
        public NativeArray<Entity> mapKeys;

        [ReadOnly]
        public NativeMultiHashMap<Entity, int> map;

        [NativeDisableParallelForRestriction]
        public ComponentDataFromEntity<ConstantIrrigation> constantIrrigationData;

        public EntityCommandBuffer.ParallelWriter ecb;

        public void Execute(int index)
        {
            for (int i = 0; i < map.Count(); ++i)
            {
                Entity entity = mapKeys[i];
                int sum = 0;

                if (map.TryGetFirstValue(entity, out int value, out NativeMultiHashMapIterator<Entity> iterator))
                {

                    sum = value;
                    while (map.TryGetNextValue(out value, ref iterator))
                    {
                        sum += value;
                    }

                    if (constantIrrigationData.HasComponent(entity))
                    {
                        ConstantIrrigation constIrrigation = constantIrrigationData[entity];
                        constIrrigation.value = math.max(constIrrigation.value + sum, 0);
                        constantIrrigationData[entity] = constIrrigation;

                        if(constIrrigation.value == 0)
                        {
                            ecb.RemoveComponent<ConstantIrrigation>(index, entity);
                        }
                    } else
                    {
                        ecb.AddComponent(index, entity, new ConstantIrrigation()
                        {
                            value = sum
                        });
                    }
                }
            }
        }
    }
}

Thanks a lot for your effort, hope the Dots veterans here can spot the issue!

2

You’re using constantIrrigationChanges.GetKeyArray at schedule time, but the job hasn’t run yet, so that’s empty array. Other than that if you build up job handles on local variables and assign this.Dependency at last it would be easier to see what’s wrong

3

You’re correct, the key array is empty when copied to the job struct and thus also stays empty. Thanks for pointing that out, I do have trouble fixing it though. I found the function AsDeferredJobArray() on NativeList (Struct NativeList<T> | Package Manager UI website) with which one should be able to pass an array that can be populated by one job and will be updated for usage by another one. However, I couldn’t get this to work when trying to fill a NativeList with the keys of the Multihashmap that is filled in the first job. This is the relevant code (I changed some stuff and now use an IJobEntityBatch instead of the Entities.ForEach):

NativeMultiHashMap<Entity, int> constantIrrigationChanges = new NativeMultiHashMap<Entity, int>(eventCount, Allocator.TempJob);
        NativeMultiHashMap<Entity, int>.ParallelWriter constantIrrigationChangesPW = constantIrrigationChanges.AsParallelWriter();

        NativeList<Entity> keyList = new NativeList<Entity>(eventCount, Allocator.TempJob);
        NativeList<Entity>.ParallelWriter keyListPW = keyList.AsParallelWriter();

        ProcessIrrigationChangeEventsJob processEventsJob = new ProcessIrrigationChangeEventsJob()
        {
            constantIrrigationChangeEventHandle = this.GetComponentTypeHandle<ConstantIrrigationChangeEvent>(true),
            entityHandle = this.GetEntityTypeHandle(),
            constantIrrigationChangesPW = constantIrrigationChangesPW,
            keyListPW = keyListPW,
            commandBufferPW = commandBuffer
        };

        UpdateConstantIrrigationJob updateJob = new UpdateConstantIrrigationJob()
        {
            mapKeys = keyList.AsDeferredJobArray(),
            map = constantIrrigationChanges,
            constantIrrigationData = GetComponentDataFromEntity<ConstantIrrigation>(false),
            ecb = commandBuffer
        };

        JobHandle job1Handle = processEventsJob.ScheduleParallel(eventEntitiesQuery, 1, this.Dependency);
        JobHandle job2Handle = updateJob.Schedule(constantIrrigationChanges.Count(), 10, job1Handle);

I checked that the keyList is actually filled (temporarily added a job1Handle.Complete() before scheduling the second one) but the second job will throw errors for trying to access elements of the zero length mapKeys array. Am I using this deferred array stuff wrong, I pretty much sticked to the example in the documentation as far as I can tell (it wasn’t used in a system there, though).

Also, I found what’s causing the other issue I wrote about in my first post: It’s me trying to access the Count() of the MultiHashMap in line 25. Makes perfect sense to me now with job 1 already scheduled and me trying to access the map (even if just for the count). Guess I would also need to pass this count in some kind of deferred way to the second job since it first has to be calculated by job No.1. How can I solve this?

4

Yes you can’t use the count like that too, that count is also zero. You can use IJobParallelForDefer, which is in the experimental Jobs package, allows you to either pass a pointer to the count or the NativeList.

Your NativeMultiHashMap usage seems weird to me though. Why use that if you’re going to for loop over all keys

5

Alright that works! Here’s the changes I made to the code:

JobHandle job1Handle = processEventsJob.ScheduleParallel(eventEntitiesQuery, 1, this.Dependency);
        JobHandle job2Handle = updateJob.Schedule<UpdateConstantIrrigationJob, Entity>(keyList, 1, job1Handle);

And the job looks like this now:

[BurstCompatible]
    public struct UpdateConstantIrrigationJob : IJobParallelForDefer
    {
        [ReadOnly]
        public NativeList<Entity> mapKeys;

        [ReadOnly]
        public NativeMultiHashMap<Entity, int> map;

        [NativeDisableParallelForRestriction]
        public ComponentDataFromEntity<ConstantIrrigation> constantIrrigationData;

        public EntityCommandBuffer.ParallelWriter ecb;

        [BurstCompatible]
        public void Execute(int index)
        {
            for (int i = 0; i < map.Count(); ++i)
            {
                Entity entity = mapKeys[i];
                int sum = 0;

                if (map.TryGetFirstValue(entity, out int value, out NativeMultiHashMapIterator<Entity> iterator))
                {
                    sum = value;

                    while (map.TryGetNextValue(out value, ref iterator))
                    {
                        sum += value;
                    }

                    if (constantIrrigationData.HasComponent(entity))
                    {
                        ConstantIrrigation constIrrigation = constantIrrigationData[entity];
                        constIrrigation.value = math.max(constIrrigation.value + sum, 0);
                        constantIrrigationData[entity] = constIrrigation;

                        if(constIrrigation.value == 0)
                        {
                            ecb.RemoveComponent<ConstantIrrigation>(index, entity);
                        }
                    } else
                    {
                        ecb.AddComponent(index, entity, new ConstantIrrigation()
                        {
                            value = sum
                        });
                    }
                }
            }
        }
    }

Would you say this is the proper way of doing this? I would have imagined that this is kind of a very common case with one job working on something and another depending job to use that as input and you hardly find anything about IJobParallelForDefer online.

About your remark regarding my usage of MultiHashMap: Yeah, it’s somewhat weird but I wanted a way to go through the map in a parallel job and couldn’t find another option to do that. I think at some point of the DOTS development there were specialized job types to go through MultiHashMaps but it wasn’t available for me and to me it seems it might have been deprecated or something. I’m very open for suggestion to do this in a more direct way, if you have an idea.

Thanks a lot so far, buddy! You’re really helping me out here.

6

I guess if you have to go through the map in parallel that’s fine.

I would prefer using Entities.ForEach with DynamicBuffer if that’s possible, instead of the MultiHashMap

Also how does that Parallel Job work? You’re iterating through the whole map for each index, instead you need to only use the given index

7

Usually you wouldn’t want to call map.Count() in each iteration of the loop; it is not simply a fast lookup like in an array. In your case, perhaps you intended mapKeys.Length instead.

You can use foreach(var value in map.GetValuesForKey(entity)) to replace TryGetFirstValue and TryGetNextValue

8

You’re right, don’t know how I overlooked that. I guess when I was trying out different approaches this somehow was left in. I removed it now, thanks again.

9

I also integrated your suggestions, which made the code more straightforward. Thanks a lot!