8
\$\begingroup\$

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?
\$\endgroup\$
  • \$\begingroup\$ You're confusing atomic safety with application thread safety, essentially - ie Volatile.Read(ref workloadsRemaining) will get you the state of the variable, but your comparison (==) is NOT part of the guarantee; you can read an actual 0, but by the time you go to compare it, workloadsRemaining is now 1, so the condition should maybe be false... you still have timing issues, and aren't necessarily any better off. Same with WorkloadSet.add(...) - two entrances to an instance might have operations reordered/interleaved in fun ways. Shared mutable state is evil in threading. \$\endgroup\$ – Clockwork-Muse Jan 5 '14 at 4:32
  • \$\begingroup\$ @ WorkloadSet.add -> Only the master thread can add to the workload, so I didn't make that safe. I need to digest the other stuff you said though :) \$\endgroup\$ – Xenoprimate Jan 5 '14 at 4:37
  • \$\begingroup\$ Ah okay I get you - I see your point but the 'workloads remaining' variable can only ever go down (not up) unless someone calls .Add() - and only the master thread can do that. The master thread will not check for completion until it knows its added all the workloads. So I think that's safe? \$\endgroup\$ – Xenoprimate Jan 5 '14 at 4:39
  • 3
    \$\begingroup\$ If there's one thing I've heard continuously, it's that you can't anticipate how your code is going to be used - and this includes by you. Unless you make it actually safe by design (so that it can only be put together a specific way), it's going to be "misused"; for instance, what if a workload processor decides that a result requires more work-sets - simple, just call Add(...)! Given current processor trends, I too want a multithreaded engine; however, I'm thinking more along the lines of immutable/functional code so I don't have to worry about locks at all.... (ish...) \$\endgroup\$ – Clockwork-Muse Jan 5 '14 at 4:50
  • \$\begingroup\$ "and context switching as much as possible" - forgive me, I don't pay attention to threading as much as I should, but isn't switching a thread a context switch? \$\endgroup\$ – Dan Pantry Aug 21 '14 at 21:37
3
\$\begingroup\$

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 ?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.