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 EngineComponent currentComponent;

int numLogicalCores = Environment.ProcessorCount;
int poolSize = numLogicalCores;

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

for (int i = 0; i < poolSize; i++) {
Name = "OphSlave-" + i,
IsBackground = true
};
}

}

/// <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();

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

int range = currentComponent.GetRange();
int blockSize = currentComponent.actualBlockSize;
int numWorkloadsLessOne = range / blockSize - 1;
completionWaiter.Reset();

// Pre-execute
currentComponent.PreExecute();

for (int wl = 0; wl < numWorkloadsLessOne; ) {
}

// Do the work
}

// Wait for every thread to be done
completionWaiter.SpinOnce();
}

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

SpinWait workWaiter = new SpinWait();

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

workWaiter.SpinOnce();
}

do {
}
}
}


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

[FieldOffset(0)]
internal long AsLong;
}


/// <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>
private long[] workloads = new long[0];
private int producerNextIndex = 0;
private int consumerNextIndex = 0;

/// <summary>
/// Returns true if all workloads have been completed, false if not.
/// </summary>
get {
}
}

/// <summary>
/// Creates a new Workload Set.
/// </summary>
/// <param name="capacity">The initial capacity to use.</param>
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;

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>
Interlocked.Increment(ref producerNextIndex);
}

/// <summary>
/// Lets the calling thread reserve a workload, if one is waiting, for its exclusive execution.
/// </summary>
/// <returns>True if the workload struct was set, false if not (i.e. false if there is no waiting work).</returns>
int consumerNextIndexSnapshot;

do {
int producerNextIndexSnapshot = producerNextIndex;

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

return true;
}

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


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?
• 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. Jan 5 '14 at 4:32
• @ 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 :) Jan 5 '14 at 4:37
• 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? Jan 5 '14 at 4:39
• 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...) Jan 5 '14 at 4:50
• "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?
– Dan
Aug 21 '14 at 21:37

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;

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

for (int i = 0; i < poolSize; i++) {
Name = "OphSlave-" + i,
IsBackground = true
};
}

}


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 ?