Low-lock Multi-threading Implementation

Question

I'm designing in my spare time a game engine (for fun, not so much for profit, haha). I wanted to design the 'core pipeline' as efficiently as possible. Having a quad-core CPU, I decided to take advantage of parallel processing.

I wanted to implement a lock-free algorithm (or at least, low-locking) to make the pipeline as quick as possible (and to avoid expensive things like lock contention, kernel-mode locking, and context switching as much as possible).

Without further preamble, here is my implementation (slightly shortened):

EnginePipeline.cs

public static partial class EnginePipeline {
    private static EngineComponent[] componentPipeline;
    private static volatile bool isRunning = false;
    private static volatile bool exitFlag = false;

    private static Thread[] threadPool;
    private static EngineComponent currentComponent;
    private static WorkloadSet currentWorkloads = new WorkloadSet(100);

    private static void InitThreadPool() {
        int numLogicalCores = Environment.ProcessorCount;
        int poolSize = numLogicalCores;
        if (PipelineConfig.MaxThreads > 0 && numLogicalCores > PipelineConfig.MaxThreads) poolSize = PipelineConfig.MaxThreads;

        // One less than the number of cores because the master thread will be used too
        poolSize -= 1;
        if (poolSize < 0) poolSize = 0;

        threadPool = new Thread[poolSize];
        for (int i = 0; i < poolSize; i++) {
            threadPool[i] = new Thread(ThreadWaitForWork) {
                Name = "OphSlave-" + i,
                IsBackground = true
            };
            threadPool[i].Start();
        }

        Thread.CurrentThread.Priority = ThreadPriority.AboveNormal;
        Thread.CurrentThread.Name = "OphMaster";
    }

    /// <summary>
    /// Run the pipeline. This method will block indefinitely until something calls
    /// <see cref="Shutdown"/> or <see cref="TerminateWithError"/>.
    /// </summary>
    public static void Run() { 
        isRunning = true;

        SpinWait completionWaiter = new SpinWait();
        PipelineWorkload workload = new PipelineWorkload();

        while (!exitFlag) {
            for (int i = 0; i < componentPipeline.Length; ++i) {
                // Set current component
                Volatile.Write(ref currentComponent, componentPipeline[i]);

                // Calculate the workloads
                int range = currentComponent.GetRange();
                int blockSize = currentComponent.actualBlockSize;
                int numWorkloadsLessOne = range / blockSize - 1;
                currentWorkloads.Reset(numWorkloadsLessOne + 1);
                completionWaiter.Reset();

                // Pre-execute
                currentComponent.PreExecute();

                // Add the work
                for (int wl = 0; wl < numWorkloadsLessOne; ) {
                    workload.BlockStartInc = wl * blockSize;
                    workload.BlockEndEx = ++wl * blockSize;
                    currentWorkloads.Add(workload);
                }
                workload.BlockStartInc = numWorkloadsLessOne * blockSize;
                workload.BlockEndEx = (range + 1);
                currentWorkloads.Add(workload);

                // Do the work
                while (currentWorkloads.Reserve(ref workload)) {
                    currentComponent.Execute(workload.BlockStartInc, workload.BlockEndEx);
                    currentWorkloads.Complete();
                }

                // Wait for every thread to be done
                while (!currentWorkloads.AllWorkloadsCompleted) {
                    completionWaiter.SpinOnce();
                }

                // Post-execute
                currentComponent.PostExecute();
            }
        }
    }

    private static void ThreadWaitForWork() { 
        SpinWait workWaiter = new SpinWait();
        PipelineWorkload workload = new PipelineWorkload();


        while (!exitFlag) {
            workWaiter.Reset();

            while (!currentWorkloads.Reserve(ref workload)) {
                workWaiter.SpinOnce();
            }

            EngineComponent currentComponentSnapshot = Volatile.Read(ref currentComponent);

            do {
                currentComponentSnapshot.Execute(workload.BlockStartInc, workload.BlockEndEx);
                currentWorkloads.Complete();
            } while (currentWorkloads.Reserve(ref workload));
        }
    }
}

PipelineWorkload.cs

[StructLayout(LayoutKind.Explicit)]
public struct PipelineWorkload {
    [FieldOffset(0)]
    public int BlockStartInc;
    [FieldOffset(4)]
    public int BlockEndEx;

    [FieldOffset(0)]
    internal long AsLong;
}

WorkloadSet.cs

/// <summary>
/// Represents a set of <see cref="PipelineWorkload"/>s that the master thread has created, and that are to be
/// consumed and executed by the master and its slaves.
/// </summary>
/// <remarks>
/// <para>The master adds all the work at the beginning of each tick by calling <see cref="Add"/> with each workload.
/// All the time, the slave threads will be calling <see cref="Reserve"/> to check for added work.</para>
/// 
/// <para>Once the master has added all work, it will keep calling <see cref="Reserve"/> to also chip in on remaining
/// work, until <see cref="Reserve"/> returns false; after which it will keep checking <see cref="AllWorkloadsCompleted"/>.
/// Each worker thread can decrement the number of workloads remaining by calling <see cref="Complete"/>.
/// Once all workloads are complete, the next tick begins, and the master thread will call <see cref="Reset"/>.</para>
/// </remarks>
public sealed class WorkloadSet {
    private long[] workloads = new long[0];
    private int producerNextIndex = 0;
    private int consumerNextIndex = 0;
    private int workloadsRemaining = 0;

    /// <summary>
    /// Returns true if all workloads have been completed, false if not.
    /// </summary>
    public bool AllWorkloadsCompleted {
        get {
            return Volatile.Read(ref workloadsRemaining) == 0;
        }
    }

    /// <summary>
    /// Creates a new Workload Set.
    /// </summary>
    /// <param name="capacity">The initial capacity to use.</param>
    public WorkloadSet(int capacity) {
        Reset(capacity);
    }

    /// <summary>
    /// Resets the state of this set, and optionally re-sizes the internal buffer according to the requisite capacity.
    /// The buffer will only be re-sized if necessary.
    /// </summary>
    /// <param name="capacity">The requisite capacity on this tick.</param>
    public void Reset(int capacity) {
        Assure.IsGreaterThan(capacity, 0, "WorkloadSet capacity must be at least 1.");

        producerNextIndex = 0;
        consumerNextIndex = 0;
        Volatile.Write(ref workloadsRemaining, 0);

        if (workloads.Length < capacity) workloads = new long[capacity * 2]; /* Double the capacity to eliminate the chance
                                                                                * of the next call to Reset causing another
                                                                                * resize if it is (workloads.Length + n) where
                                                                                * n isn't double the size again.
                                                                                */
    }

    /// <summary>
    /// Add a new element to the set.
    /// </summary>
    /// <param name="workload">The workload to add.</param>
    public void Add(PipelineWorkload workload) {
        Volatile.Write(ref workloads[producerNextIndex], workload.AsLong);
        Interlocked.Increment(ref workloadsRemaining);
        Interlocked.Increment(ref producerNextIndex);
    }

    /// <summary>
    /// Lets the calling thread reserve a workload, if one is waiting, for its exclusive execution.
    /// </summary>
    /// <param name="workload">The workload struct to be set.</param>
    /// <returns>True if the workload struct was set, false if not (i.e. false if there is no waiting work).</returns>
    public bool Reserve(ref PipelineWorkload workload) {
        int consumerNextIndexSnapshot;

        do {
            consumerNextIndexSnapshot = Volatile.Read(ref consumerNextIndex);
            int producerNextIndexSnapshot = producerNextIndex;

            if (consumerNextIndexSnapshot == producerNextIndexSnapshot) return false;
        }
        while (Interlocked.CompareExchange(ref consumerNextIndex, consumerNextIndexSnapshot + 1, consumerNextIndexSnapshot) != consumerNextIndexSnapshot);

        workload.AsLong = workloads[consumerNextIndexSnapshot];
        return true;
    }

    /// <summary>
    /// Reports that a <see cref="PipelineWorkload"/> that was previously reserved by this thread has completed execution.
    /// </summary>
    public void Complete() {
        Interlocked.Decrement(ref workloadsRemaining);
    }
}

---

I'm interested in two things:

1. Efficiency: Is this implementation as fast as it could be (not considering things like choosing a different language etc.)
2. Correctness: Are there any parts of this implementation that could actually result in memory corruption or dead/live lock?

Answer 1

you have a little bit of rewriting that can be done on this chunk of code

private static void InitThreadPool() {
    int numLogicalCores = Environment.ProcessorCount;
    int poolSize = numLogicalCores;
    if (PipelineConfig.MaxThreads > 0 && numLogicalCores > PipelineConfig.MaxThreads) poolSize = PipelineConfig.MaxThreads;

    // One less than the number of cores because the master thread will be used too
    poolSize -= 1;
    if (poolSize < 0) poolSize = 0;

    threadPool = new Thread[poolSize];
    for (int i = 0; i < poolSize; i++) {
        threadPool[i] = new Thread(ThreadWaitForWork) {
            Name = "OphSlave-" + i,
            IsBackground = true
        };
        threadPool[i].Start();
    }

    Thread.CurrentThread.Priority = ThreadPriority.AboveNormal;
    Thread.CurrentThread.Name = "OphMaster";
}

I want to change a Variable name, and I am sure that there are more than should be changed as well. but I have only singled out this method so far.

so this

private static Thread[] threadPool;

should be this

private static Thread[] threads;

and then instead of counting the logical cores/processors and changing the poolSize variable so many times, just use a ternary statement to assign the value you want to the poolSize variable like this

// One less than the number of cores because the master thread will be used too
int poolSize = (PipelineConfig.MaxThreads > 0 && Environment.ProcessorCount > PipelineConfig.MaxThreads) ? PipelinConfig.MaxThreads - 1 : Environment.ProcessorCount - 1;
if (poolSize < 0) poolSize = 0;

And then following the logic of that ternary statement MaxThreads is greater than 0 meaning 1 or more and ProcessorCount has to be greater than MaxThreads, so you could get rid of this:

if (poolSize < 0) poolSize = 0;

but, I think there will be issues. What if PipelineConfig.MaxThreads is the same as Environment.ProcessorCount ?