TaskFlow Part 1

A couple of weeks ago, I developed a thin wrapper around C# Tasks and called it TaskExtensions. The main purpose of it was to provide 2 interfaces to write Task structs.

• ITask
• ITaskParallelFor

The idea is to provide a fluent way of writing threads similar to Unity’s Job System.

ITask represents writing a single task to be executed on a single thread. This is similar to writing a regular Task in C# like so.

Task.Factory.StartNew(() => { /* Perform a single unit of work */ };

ITaskParallelFor represents a unit of work that is partioned into multiple threads. For example, let’s say you have an array that has 1000 elements. You want to split this between 10 threads. You can split it like so:

This would be pretty simple as you can write a for loop like so:

var someArray = new int[1000];
for (var i = 0; i < 10; i++) {
    Task.Factory.StartNew(() => {
        var offset = i * 100; // Represent our start offset, each thread starts at a multiple of 100
        for (var x = 0; x < 100; x++) {
            someArray[x + offset] += 1;
        }
    });
}

Now the problem is, what happens when you don’t have an event number of elements to process each slice for. For example, if you have 1001 elements, your last slice of work will have 101 elements to process. You could launch 11 threads where the last thread will only process the last unit of work, but the cost to launch a thread to process the last element is not ideal.

ITaskParallelFor will calculate the partitions such that your last thread will take the remaining slice regardless of what you define as the unit of work.

The need for TaskExtensions was to provide an idiomatic way to launch threads to process a large amount of work without having to manually calculate each slice and catch for odd cases.

How to write an ITask and ITaskParallelFor  

Some examples below

ITask  

// ITask
public struct SingleTask : ITaskFor {

    public int[] Data;

    public void Execute() {
        for (var i = 0; i < Data.Length; i++) {
            /* Do some work with Data */
        }
    }
}

async void Update() {
    // SOME_DATA is an array that is stored somewhere, you're providing a 
    // mutable reference to the task.
    await new SingleTask { Data = SOME_DATA }.Schedule();
}

ITaskParallelFor  

public struct MultiTask : ITaskFor {

    public int[] Data;

    public void Execute(int index) {
        ref var element = ref Data[index];
        /* Do some work on element */
    }
}

async void Update() {
    // SOME_DATA is an array that is stored somewhere, you're providing a 
    // mutable reference to the task.
    // Assume SOME_DATA contains 1000 elements , and we want to schedule, 10 threads
    await new MultiTask { Data = SOME_DATA }.Schedule(total: SOME_DATA.Length, workPerTask: 100);
}

Now this is only part 1 and the TaskExtensions wrapper changed quite a bit. The next part is pretty much a rewrite of the TaskExtensions (now renamed to TaskFlow), which introduces concepts such as

• TaskHandle
• TaskGraph with Dependency Tracking
  • Topological sorting

This upcoming version makes it more like Unity’s Job System and is mainly meant for preemptive scheduling (async - await will still exist).