Building a load-balanced task scheduler - Part 3: Parent-child relationships

Continuing from where we left off last time, this post explains how parent-child relationships are handled inside the task scheduler, and how streaming tasks can be split automatically by the scheduler.

If you’re new to this series, make sure to read the other articles first:

• Part 1: Explains the basics behind a task scheduler, work-stealing, load-balancing, etc., and explains a very simple scheduler implementation.
• Part 2: Discusses Molecule’s task model in detail.

Today, we will have a look at how parent-child relationships are implemented in Molecule’s task scheduler. Among the implementation’s requirements were the following:

• A task should be able to have N child tasks.
• Adding relationships should not make the code more complicated.
• Waiting for a task to be completed must properly synchronize across its children as well, e.g. a task is only considered completed if all his children and the task himself have finished executing.

Keeping track of relationships

A naive implementation would simply add a list of child tasks to a Task (e.g. stored in a std::vector<>), and check for completion whenever scheduler::Wait() is called on a task. Such an implementation should immediately set off your alarm bells:

• By adding a std::vector<> we loose the ability to relocate/memcpy a Task.
• Dynamic memory allocations. We care about memory!

Instead of adding an explicit list of all its children to a Task, we turn the problem on its head and store a parent in each Task instead. Furthermore, as long as we don’t need to identify each and every child, we can simply store the number of unfinished children in our Task.

Basic idea

The basic idea behind implementing parent-child relationships is now simply the following:

• Each Task stores the number of still open tasks – a task with N child tasks has N+1 still open tasks (one for himself, N for the children). The N+1 bit is important!
• Each task done executing tells its parent (recursively) that it has finished.
• A task can only be executed if the number of open tasks equals 1. This ensures that parents/children are automatically synchronized because no parent task is ever being worked on unless all of its children have finished already.

While this sounds simple in theory, there are still some things we have to watch out for:

• As discussed, when pulling a task from the queue we must wait until all its children have finished executing. But we need to ensure that the thread that pulled the task from the queue does some work in the meantime, otherwise we either block the CPU (by busy looping), or never work on the task because it’s no longer contained in the queue.
• When waiting until a task has finished, we again should be doing work in the meantime.

Implementation

With the above in mind, we first add the needed members to our Task structure:

struct Task
{
  char unusedFreelistAlias[sizeof(void*)];
  AtomicInt m_openTasks;
  int32_t m_generation;
  TaskOffset m_parent;
  Kernel m_kernel;
  TaskData m_taskData;
};

As can be seen, m_openTasks is implemented as an atomic counter because we need to ensure that incrementing/decrementing the number of still open tasks is done in a thread-safe manner across multiple cores/threads. The implementation is quite simple, and will be discussed at the end of this post.

Separating all the different parts of the scheduler’s logic into small functions leads to surprisingly simple worker threads:

Thread::ReturnValue Work(const core::Thread& thread)
{
  while (thread.ShouldRun())
  {
    Task* task = WaitUntilTaskIsAvailable(thread);
    WorkOnTask(task);
  }

  return 0;
}

WaitUntilTaskIsAvailable() literally does what it says on the tin – it waits until a task has become available in the queue by means of a condition variable, as discussed in part 1. Because the function only returns if there’s something in the queue, we don’t need to check the returned task for being valid.

WorkOnTask() is simple as well:

void WorkOnTask(Task* task)
{
  while (!CanExecuteTask(task))
  {
    // the task cannot be executed at this time, work on another item
    HelpWithWork();
  }

  // execute the kernel and finish the task
  (task->m_kernel)(task->m_taskData);

  FinishTask(task);
}

As discussed above, a task can only be executed if m_openTasks == 1 (remember that a task having N children will always have N+1 open tasks). Because children notify their parent of completion, this automatically makes sure that a parent is only executed if all its children have finished (we will get to the implementation of FinishTask() in a minute).

Note that rather than calling WaitUntilTaskIsAvailable() whenever a task cannot be executed immediately, we call HelpWithWork(). This is crucial, because whereas the first waits on the condition variable until a task is added to the queue, the latter simply checks if there’s work currently available, and yields CPU resources if not. If we were to wait on the condition variable, we’d introduce a race condition were a worker thread would pull the task from the queue and wait until any other task becomes available – however, if nobody adds a task this would block indefinitely.

After executing the task, the implementation calls FinishTask():

void FinishTask(Task* task)
{
  // IMPORTANT: make sure to store the result in a local variable and use that for comparing against zero later.
  // otherwise, the code contains a data race because other child tasks could change task->m_openTasks in the meantime.
  const int32_t openTasks = (--task->m_openTasks);
  if (task->m_parent != Task::NO_PARENT)
  {
    // tell our parent that we're finished
    Task* parent = GetTask(task->m_parent);
    FinishTask(parent);
  }

  if (openTasks == 0)
  {
    // this task has finished completely, remove it
    ReturnTask(task);
  }
}

This simply decrements the number of open tasks (using an atomic variable), and informs the parent (if any) about the task’s completion. Whenever m_openTasks hits zero, we remove the task and return it to the allocator (the problem of how to make sure that we don’t access “old” Task members was discussed in part 2).

Like stated above, we should be doing work while waiting on a task to become completed:

void Wait(const TaskId& taskId)
{
  // wait until the task and all its children have completed
  while (!IsTaskFinished(taskId))
  {
    // help with working while the task isn't finished
    HelpWithWork();
  }
}

IsTaskFinished() was introduced in the last part of the series, so the only thing missing is HelpWithWork() which was briefly discussed above:

void HelpWithWork(void)
{
  Task* task = GetAvailableTask();
  if (task)
  {
    WorkOnTask(task);
  }
  else
  {
    // no item available now, yield CPU resources
    core::Thread::Yield();
  }
}

Regarding the scheduler implementation, that’s about it. It’s still simple and readable, with only a few dozen lines of actual source code as can be seen above. Parent-child relationships are handled automatically using the atomic m_openTasks counter, and synchronization is simple as well.

Automatically splitting streaming tasks

With the above in place, we can now turn our attention to automatically splitting streaming tasks into several subtasks without having to worry about correct synchronization because it’s done implicitly (as long as we setup our task data structures correctly).

Splitting is done based on simple heuristics, e.g. the scheduler knows about L1 and L2 cache sizes, and splits tasks accordingly. (In this case, heuristics is just a fancy term for choosing how many tasks we need to create based on e.g. L1 cache size – it’s not an exact science, and can be configured by the user).

Furthermore, splitting is done in linear fashion while we’re still dealing with a non-load-balanced scheduler – a divide-and-conquer approach would be more feasible when using work-stealing, and we’ll handle that as well.

For now, consider this simple linear splitting:

TaskId AddStreamingTask(const KernelData& kernelData, const InputStream& is0, const OutputStream& os0, uint32_t elementCount, Kernel kernel)
{
  const int N = DetermineNumberOfTasks(is0, os0, elementCount);

  // add a root task used for synchronization
  Task* root = ObtainTask();
  root->m_kernel = &EmptyKernel;
  root->m_openTasks = N + 1;

  QueueTask(root);

  TaskOffset rootOffset = GetTaskOffset(root);
  {
    // split the task into several subtasks, according to the size of the input/output streams
    const uint32_t perElementCount = elementCount / N;
    for (int i=0; i<N; ++i)
    {
      Task* task = ObtainTask();
      task->m_kernel = kernel;
      task->m_taskData.m_kernelData = kernelData.m_data;
      task->m_taskData.m_streamingData.m_elementCount = perElementCount;
      task->m_taskData.m_streamingData.m_inputStreams[0] = static_cast<char*>(is0.m_data) + i*is0.m_elementStride*perElementCount;
      task->m_taskData.m_streamingData.m_outputStreams[0] = static_cast<char*>(os0.m_data) + i*os0.m_elementStride*perElementCount;
      task->m_openTasks = 1;
      task->m_parent = rootOffset;

      QueueTask(task);
    }
    // queue leftover tasks here...
  }

  return TaskId(rootOffset, root->m_generation);
}

Note that we keep adding tasks to the system while some worker threads are probably racing away and want to start working on both the parent and child tasks. Again, this is implicitly handled without us having to worry about race conditions. On other architectures like the PS3, we could start DMA’ing the respective stream data to the SPUs, and start working on tasks while the PPU queues more tasks to the scheduler.

And that’s about it, quite simply really.

Performance

In order to do some performance comparison, I’ve taken an easily parallelizable task as an example: character skinning. The performance figures on my i7-CPU with 4 cores and 8 logical threads (4 cores á 2 logical threads) are the following (the measurements were conducted over a series of runs and averaged):

Serial: 683.98 ms
Parallel: 177.20 ms

That’s a 3.85x speed improvement which is not bad considered we haven’t even touched load-balancing yet. Other tasks have shown speed-ups of almost 5x – in some cases the logical threads help, in other cases they don’t. It all depends on memory workloads and ALUs used.

Atomic variables

Finishing today’s post, I briefly want to discuss Molecule’s implementation of atomic variables because they have been talked about a lot in this series already.

In Molecule, an AtomicInt is nothing more than a simple class offering operators, as well as a conversion/cast operator so it can be more easily used as if it was a regular integer. The interface is the following:

ME_ALIGNED_SYMBOL(class AtomicInt, 4)
{
public:
  inline explicit AtomicInt(int32_t value);
  inline AtomicInt& operator+=(int32_t value);
  inline AtomicInt& operator++(void);
  inline AtomicInt& operator--(void);
  inline AtomicInt& operator=(int32_t value);
  inline operator int32_t() const;

private:
  volatile int32_t m_value;
};

Because atomic operations can only be carried out correctly for 32-bit aligned memory addresses, the class is declared to be aligned to 4 bytes (using __declspec(align(4)) on MSVC).

Internally, AtomicInt uses the (platform-dependent) atomic functions implemented elsewhere:

namespace atomic
{
  /// Atomically increments the integer at the memory location being pointed to
  inline __declspec(naked) void __fastcall Increment(volatile int32_t*)
  {
    __asm
    {
      lock inc dword ptr [ecx]
      ret
    }
  }

  /// Atomically decrements the integer at the memory location being pointed to
  inline __declspec(naked) void __fastcall Decrement(volatile int32_t*)
  {
    __asm
    {
      lock dec dword ptr [ecx]
      ret
    }
  }

  /// Atomically adds a value to the integer at the memory location being pointed to
  inline __declspec(naked) void __fastcall Add(volatile int32_t*, int32_t)
  {
    __asm
    {
      lock add dword ptr [ecx], edx
      ret
    }
  }
}

There’s two things to note here:

• __declspec(naked): Declaring a naked function allows us to directly use inline assembly without having to write a function prolog/epilogue.
• __fastcall: The fastcall calling convention makes sure to pass the first two arguments to the function in registers (ecx and edx, respectively). This allows us to directly use them in the inline assembly.

And that concludes today’s post. Next time, we will take a look at how CPU caches can easily kill your multicore performance if you’re not careful.

About these ads