ThreadDispatcher implementation (application freezes)

[AZV-EU]

Hello!

I'd like to quickly ask if my implementation of ThreadDispatcher is correct, because when I run DispatchThread the app it throws an error (workerBody is null)...

Could you please help? I hope it doesn't take too long to figure out what's wrong...

internal struct ThreadDispatcher : IThreadDispatcher
    {
        struct Worker
        {
            public Thread Thread;
            public AutoResetEvent Signal;
        }

        readonly Worker[] workers;
        public int ThreadCount { get; set; }

        readonly BufferPool[] bufferPools;

        volatile Action<int>? workerBody;
        int remainingWorkerCounter;
        readonly AutoResetEvent finished;
        public void DispatchWorkers(Action<int> workerBody)
        {
            Debug.Assert(this.workerBody == null);
            this.workerBody = workerBody;
            SignalThreads(workers.Length);
            DispatchThread(0);
            finished.WaitOne();
            this.workerBody = null;
        }
        void DispatchThread(int workerIndex)
        {
            Debug.Assert(workerBody != null);
            workerBody!.Invoke(workerIndex);

            if (Interlocked.Decrement(ref remainingWorkerCounter) == -1)
            {
                finished.Set();
            }
        }

        public BufferPool GetThreadMemoryPool(int workerIndex)
            => bufferPools[workerIndex];
        void SignalThreads(int maximumWorkerCount)
        {
            int maximumWorkersToSignal = maximumWorkerCount - 1;
            var workersToSignal = maximumWorkersToSignal < workers.Length ? maximumWorkersToSignal : workers.Length;
            remainingWorkerCounter = workersToSignal;
            for (int i = 0; i < workersToSignal; ++i)
                workers[i].Signal.Set();
        }
        void WorkerLoop(object untypedSignal)
        {
            var (signal, workerIndex) = ((AutoResetEvent, int))untypedSignal;
            while (true)
            {
                signal.WaitOne();
                if (disposed)
                    return;
                DispatchThread(workerIndex);
            }
        }

        volatile bool disposed;
        public void Dispose()
        {
            if (!disposed)
            {
                disposed = true;
                SignalThreads(ThreadCount);
                for (int i = 0; i < bufferPools.Length; ++i)
                    bufferPools[i].Clear();
                foreach (var worker in workers)
                {
                    worker.Thread.Join();
                    worker.Signal.Dispose();
                }
            }
        }
        public ThreadDispatcher(int threadCount, BufferPool bufferPool, int threadPoolBlockAllocationSize = 16384)
        {
            ThreadCount = threadCount;
            workers = new Worker[threadCount - 1];
            finished = new AutoResetEvent(false);
            bufferPools = new BufferPool[threadCount];
            workerBody = null;
            remainingWorkerCounter = 0;
            disposed = false;

            for (int i = 0; i < workers.Length; ++i)
            {
                workers[i] = new Worker { Thread = new Thread(WorkerLoop!), Signal = new AutoResetEvent(false) };
                workers[i].Thread.IsBackground = true;
                workers[i].Thread.Start((workers[i].Signal, i + 1));
            }
            for (int i = 0; i < bufferPools.Length; ++i)
            {
                bufferPools[i] = new BufferPool(threadPoolBlockAllocationSize);
            }
        }
    }

[RossNordby]

At a glance, nothing jumps out as wrong. That looks pretty close to the default BepuUtilities.ThreadDispatcher implementation, reordered, plus some nullability annotations.

Is it possible that DispatchWorkers is being called at the same time from multiple threads? That implementation is not reentrant, so that would definitely cause problems.

[AZV-EU]

How can I check that?

I'm pretty sure i'm calling it from the main thread, as I specifically disabled multithreading for the app...
What else could be the cause?

[RossNordby]

How can I check that?

Look at all instances where DispatchWorkers is called and ensure that the paths to those calls couldn't accidentally occur from multiple threads simultaneously. Race conditions are not a trivial thing to debug.

What else could be the cause?

Anything which sets the workerBody to null while a worker is still running. The only times that workerBody is set to null within the dispatcher itself are upon construction, and after job completion. This is why I suggested simultaneous dispatches as a cause; the state of the dispatcher would be corrupted and some workers would end up thinking there is work to be done, but the workerBody could have already been set to null by the ending of a different call.

As a sanity check, try using the original BepuUtilities.ThreadDispatcher unmodified. There could be some fiddly difference I'm not noticing.

[AZV-EU]

As a sanity check, trying using the original BepuUtilities.ThreadDispatcher unmodified. There could be some fiddly difference I'm not noticing.

Sadly, I can't access this class when using nuget import. Should I try manually importing from the github page?

[RossNordby]

To get the latest stuff, you'll need to clone the repo and add it as a project reference.

[AZV-EU]

Thank you very much!

Manually cloning the repo, building and adding the references did the trick.