Parallel.h

#pragma once
 
#include "Context.h"
 
#include "Services/TaskManager.h"
#include "Platform/Instrumentation.h"
 
#include "Foundation/ComponentModel/Component.h"
#include "Foundation/Platform/Threads.h"
 
namespace Cogs
{
  namespace Core
  {
    namespace Serial
    {
      template<typename Pool, typename Func>
      void processComponents(Pool & pool, Func func)
      {
        size_t i = 0;
        for (auto & component : pool) {
          func(component, i++);
        }
      }
    }
 
    namespace Parallel
    {
      struct TaskScope
      {
        TaskScope(TaskManager * taskManager, TaskId group) : taskManager(taskManager), group(group) {}
        TaskScope(TaskScope && other) = default;
        ~TaskScope() { if (taskManager && group.isValid()) taskManager->destroy(group); }
        void Wait() { taskManager->wait(group); }
      private:
        TaskManager * taskManager = nullptr;
        TaskId group = NoTask;
      };
 
      template<typename Func>
      TaskScope forEach(Context * context, size_t count, Func f, const char * scopeTag = "")
      {
        TaskQueueId queue = TaskManager::GlobalQueue;
        const size_t hwc = context->taskManager->getQueueConcurrency(queue);
        if (count <= 1 || hwc < 1) {
          if (count) {
            DynamicCpuInstrumentationScope(SCOPE_PARALLEL, "forEach", scopeTag);
            for (size_t j = 0; j < count; ++j) {
              f(j);
            }
          }
          return TaskScope{ context->taskManager.get(), NoTask };
        }
 
        const size_t perTask = (count / hwc) + 1;
 
        auto group = context->taskManager->createGroup(queue);
 
        for (size_t i = 0; i < hwc; ++i) {
          const size_t begin = i * perTask;
          const size_t end = std::min(begin + perTask, count);
 
          if (begin >= count) continue;
 
          context->taskManager->enqueueChild(group, [=]()
          {
            DynamicCpuInstrumentationScope(SCOPE_PARALLEL, "forEach", scopeTag);
 
            for (size_t j = begin; j < end; ++j) {
              f(j);
            }
          });
        }
 
        return TaskScope{ context->taskManager.get(), group };
      }
 
      template<typename Func>
      TaskScope forEachRanged(Context* context, size_t count, Func f, size_t minSize = 0, const char* scopeTag = "")
      {
        TaskQueueId queue = TaskManager::GlobalQueue;
        const size_t hwc = context->taskManager->getQueueConcurrency(queue);
        if (count < minSize || hwc < 1) {
          if (count) {
            DynamicCpuInstrumentationScope(SCOPE_PARALLEL, "forEachRanged", scopeTag);
            f(0, count);
          }
          return TaskScope{ context->taskManager.get(), NoTask };
        }
        const size_t perTask = std::max(minSize, (count / hwc) + 1);
 
        TaskId group = context->taskManager->createGroup(queue);
 
        for (size_t i = 0; i < hwc; ++i) {
          const size_t begin = i * perTask;
          const size_t end = std::min(begin + perTask, count);
 
          if (begin >= count) continue;
 
          context->taskManager->enqueueChild(group, [=]()
          {
            DynamicCpuInstrumentationScope(SCOPE_PARALLEL, "forEachRanged", scopeTag);
            f(begin, end);
          });
        }
 
        return TaskScope{ context->taskManager.get(), group };
      }
 
      template<typename Pool, typename Func>
      TaskId processComponents(Context * context, Pool & pool, const char * scopeName, Func func, TaskId taskGroup = NoTask)
      {
        const size_t numComponents = pool.size();
        TaskQueueId queue = taskGroup.isValid() ? taskGroup.queueId : TaskManager::GlobalQueue;
        const size_t hwc = context->taskManager->getQueueConcurrency(queue);
        if (numComponents <= 1 || hwc < 1) {
          if (numComponents) {
            DynamicCpuInstrumentationScope(SCOPE_PARALLEL, "processComponents", scopeName);
            for (size_t j = 0; j < numComponents; j++) {
              func(pool[static_cast<SizeType>(j)], j);
            }
          }
          return NoTask;
        }
 
        if (!taskGroup.isValid()) {
          taskGroup = context->taskManager->createGroup(queue);
        }
        const size_t perTask = (numComponents / hwc) + 1;
        for (size_t i = 0; i < hwc; ++i) {
          const size_t begin = i * perTask;
          const size_t end = std::min(begin + perTask, numComponents);
 
          if (begin >= numComponents) continue;
 
          context->taskManager->enqueueChild(taskGroup, [&pool, scopeName, func, begin, end]()
          {
            DynamicCpuInstrumentationScope(SCOPE_PARALLEL, "processComponents", scopeName);
            for (size_t j = begin; j < end; ++j) {
              func(pool[static_cast<SizeType>(j)], j);
            }
          });
        }
        return taskGroup;
      }
 
      template<typename Pool, typename Func>
      TaskId processComponents(Context * context, Pool & pool, Func func, TaskId taskGroup = NoTask)
      {
        return processComponents(context, pool, "Parallel::processComponents", func, taskGroup);
      }
    }
  }
}