<commit_before>#include "Task/TaskManager.h"

#include <cassert>

#include "Task/TaskQueue.h"
#include "Platform/Os.h"

namespace Flourish
{
    struct RootAndDependencies
    {
        RootAndDependencies(TaskManager* taskManager, TaskId root, std::vector<TaskId> dependenices)
        {
            _taskManager = taskManager;
            _root = root;
            _dependencies = dependenices;
        }
        
        TaskManager* _taskManager;
        TaskId _root;
        std::vector<TaskId> _dependencies;
    };
    
    void WaitForRootThenAddDependencies(void* data)
    {
        auto rootAndDependencies = (RootAndDependencies*)data;
        rootAndDependencies->_taskManager->Wait(rootAndDependencies->_root);
        for(auto& dependancy : rootAndDependencies->_dependencies)
        {
            rootAndDependencies->_taskManager->FinishAdd(dependancy);
        }
        delete rootAndDependencies;
    }
    
	TaskManager::TaskManager(int32_t numThreads)
		: _nextTaskId(1)
        , _allTasks(nullptr)
        , _numThreads(numThreads)
        , _numQueuedTasks(0)
		, _workerThreads(nullptr)
        , _taskQueues(nullptr)
        , _waitForWorkMutex()
		, _workMaybeAvailable()
        , _exiting(false)
	{
        CreateTasks();
		CreateAndStartWorkerThreads();
	}

	TaskManager::~TaskManager()
	{
        _exiting = true;
        _workMaybeAvailable.notify_all();
        for (uint32_t threadIdx = 0; threadIdx < _numThreads; threadIdx++)
        {
            _workerThreads[threadIdx].join();
        }
		delete[] _workerThreads;
	}

	TaskId TaskManager::BeginAdd(WorkItem workItem)
	{
        TaskId taskId = _nextTaskId++;
        auto task = GetTaskFromId(taskId);
		task->_id = taskId;
		task->_workItem = workItem;
        task->_added = false;
        task->_complete = false;
		return task->_id;
	}
    
    TaskId TaskManager::AddDependentTasks(TaskId root, std::vector<TaskId> dependencies)
    {
        WorkItem waitForRootThenAddDependencies(&WaitForRootThenAddDependencies, new RootAndDependencies(this, root, dependencies));
        auto wrapperTaskId = BeginAdd(waitForRootThenAddDependencies);
        FinishAdd(wrapperTaskId);
        return wrapperTaskId;
    }
    
    void TaskManager::AddChild(TaskId parent, TaskId child)
    {
        auto childTask = GetTaskFromId(child);
        if(childTask == nullptr)
        {
            return;
        }
        childTask->_parentId = parent;
    }
    
    void TaskManager::FinishAdd(TaskId id)
    {
        auto task = GetTaskFromId(id);
        if(task != nullptr)
        {
            // The first task queue is the one for this non-worker thread
            assert(!task->_added);
            task->_added = true;
            _taskQueues[0]->Push(task);
            _numQueuedTasks++;
            _workMaybeAvailable.notify_one();
        }
    }
    
    TaskId TaskManager::AddTaskWithNoChildrenOrDependencies(Flourish::WorkItem workItem)
    {
        auto taskId = BeginAdd(workItem);
        FinishAdd(taskId);
        return taskId;
    }

	void TaskManager::Wait(TaskId id)
	{
        auto task = GetTaskFromId(id);
        if(task == nullptr)
        {
            return;
        }
		while (!task->_complete)
		{
            WaitForTaskAndExecute(0);
		}
	}
    
    void TaskManager::CreateTasks()
    {
        _allTasks = new Task[MaxConcurrentTasks];
    }

	void TaskManager::CreateAndStartWorkerThreads()
	{
        if((int32_t)_numThreads == TaskManager::AutomaticallyDetectNumThreads)
        {
            _numThreads = GetIdealNumThreads();
        }
        _workerThreads = new std::thread[_numThreads];
        _taskQueues = new TaskQueue*[_numThreads + 1]; // The current, non-worker thread also gets a queue
        for (uint32_t queueIdx = 0; queueIdx < _numThreads + 1; queueIdx++)
        {
            _taskQueues[queueIdx] = new TaskQueue();
        }
		for (uint32_t threadIdx = 0; threadIdx < _numThreads; threadIdx++)
		{
			_workerThreads[threadIdx] = std::thread(&TaskManager::WorkerThreadFunc, this, threadIdx + 1);
        }
	}

	void TaskManager::WorkerThreadFunc(int32_t threadIdx)
	{
		while(!_exiting)
		{
            WaitForTaskAndExecute(threadIdx);
		}
	}
    
    void TaskManager::WaitForTaskAndExecute(int32_t threadIdx)
    {
        auto task = GetTaskToExecute(threadIdx);
        if(task == nullptr)
        {
            if(_numQueuedTasks > 0)
            {
                // Task was added whilst we were looking for one, back to the top!
                return;
            }
            // Wait for a new task to be added
            std::unique_lock<std::mutex> lock(_waitForWorkMutex);
            _workMaybeAvailable.wait(lock);
            lock.unlock();
            return;
        }
        _numQueuedTasks--;
        task->_workItem();
        FinishTask(task);
    }
    
    int32_t TaskManager::GetIdealNumThreads()
    {
        // TODO: Actually work this out from something sensible
        return 5;
    }
    
    Task* TaskManager::GetTaskToExecute(uint32_t currentThreadQueueIdx)
    {
        auto ourQueue = _taskQueues[currentThreadQueueIdx];<commit_msg>Actually detect the number of hardware threads<commit_after>#include "Task/TaskManager.h"

#include <cassert>

#include "Task/TaskQueue.h"
#include "Platform/Os.h"

namespace Flourish
{
    struct RootAndDependencies
    {
        RootAndDependencies(TaskManager* taskManager, TaskId root, std::vector<TaskId> dependenices)
        {
            _taskManager = taskManager;
            _root = root;
            _dependencies = dependenices;
        }
        
        TaskManager* _taskManager;
        TaskId _root;
        std::vector<TaskId> _dependencies;
    };
    
    void WaitForRootThenAddDependencies(void* data)
    {
        auto rootAndDependencies = (RootAndDependencies*)data;
        rootAndDependencies->_taskManager->Wait(rootAndDependencies->_root);
        for(auto& dependancy : rootAndDependencies->_dependencies)
        {
            rootAndDependencies->_taskManager->FinishAdd(dependancy);
        }
        delete rootAndDependencies;
    }
    
	TaskManager::TaskManager(int32_t numThreads)
		: _nextTaskId(1)
        , _allTasks(nullptr)
        , _numThreads(numThreads)
        , _numQueuedTasks(0)
		, _workerThreads(nullptr)
        , _taskQueues(nullptr)
        , _waitForWorkMutex()
		, _workMaybeAvailable()
        , _exiting(false)
	{
        CreateTasks();
		CreateAndStartWorkerThreads();
	}

	TaskManager::~TaskManager()
	{
        _exiting = true;
        _workMaybeAvailable.notify_all();
        for (uint32_t threadIdx = 0; threadIdx < _numThreads; threadIdx++)
        {
            _workerThreads[threadIdx].join();
        }
		delete[] _workerThreads;
	}

	TaskId TaskManager::BeginAdd(WorkItem workItem)
	{
        TaskId taskId = _nextTaskId++;
        auto task = GetTaskFromId(taskId);
		task->_id = taskId;
		task->_workItem = workItem;
        task->_added = false;
        task->_complete = false;
		return task->_id;
	}
    
    TaskId TaskManager::AddDependentTasks(TaskId root, std::vector<TaskId> dependencies)
    {
        WorkItem waitForRootThenAddDependencies(&WaitForRootThenAddDependencies, new RootAndDependencies(this, root, dependencies));
        auto wrapperTaskId = BeginAdd(waitForRootThenAddDependencies);
        FinishAdd(wrapperTaskId);
        return wrapperTaskId;
    }
    
    void TaskManager::AddChild(TaskId parent, TaskId child)
    {
        auto childTask = GetTaskFromId(child);
        if(childTask == nullptr)
        {
            return;
        }
        childTask->_parentId = parent;
    }
    
    void TaskManager::FinishAdd(TaskId id)
    {
        auto task = GetTaskFromId(id);
        if(task != nullptr)
        {
            assert(!task->_added); // If this is hit a task was added twice
            task->_added = true;
            _taskQueues[0]->Push(task);
            _numQueuedTasks++;
            _workMaybeAvailable.notify_one();
        }
    }
    
    TaskId TaskManager::AddTaskWithNoChildrenOrDependencies(Flourish::WorkItem workItem)
    {
        auto taskId = BeginAdd(workItem);
        FinishAdd(taskId);
        return taskId;
    }

	void TaskManager::Wait(TaskId id)
	{
        auto task = GetTaskFromId(id);
        if(task == nullptr)
        {
            return;
        }
		while (!task->_complete)
		{
            WaitForTaskAndExecute(0);
		}
	}
    
    void TaskManager::CreateTasks()
    {
        _allTasks = new Task[MaxConcurrentTasks];
    }

	void TaskManager::CreateAndStartWorkerThreads()
	{
        if((int32_t)_numThreads == TaskManager::AutomaticallyDetectNumThreads)
        {
            _numThreads = GetIdealNumThreads();
        }
        _workerThreads = new std::thread[_numThreads];
        _taskQueues = new TaskQueue*[_numThreads + 1]; // The current, non-worker thread also gets a queue
        for (uint32_t queueIdx = 0; queueIdx < _numThreads + 1; queueIdx++)
        {
            _taskQueues[queueIdx] = new TaskQueue();
        }
		for (uint32_t threadIdx = 0; threadIdx < _numThreads; threadIdx++)
		{
			_workerThreads[threadIdx] = std::thread(&TaskManager::WorkerThreadFunc, this, threadIdx + 1);
        }
	}

	void TaskManager::WorkerThreadFunc(int32_t threadIdx)
	{
		while(!_exiting)
		{
            WaitForTaskAndExecute(threadIdx);
		}
	}
    
    void TaskManager::WaitForTaskAndExecute(int32_t threadIdx)
    {
        auto task = GetTaskToExecute(threadIdx);
        if(task == nullptr)
        {
            if(_numQueuedTasks > 0)
            {
                // Task was added whilst we were looking for one, back to the top!
                return;
            }
            // Wait for a new task to be added
            std::unique_lock<std::mutex> lock(_waitForWorkMutex);
            _workMaybeAvailable.wait(lock);
            lock.unlock();
            return;
        }
        _numQueuedTasks--;
        task->_workItem();
        FinishTask(task);
    }
    
    int32_t TaskManager::GetIdealNumThreads()
    {
        auto numConcurrentThreads = std::thread::hardware_concurrency();
        assert(numConcurrentThreads != 0); // If you hit this, it can't be detected
        return numConcurrentThreads - 1;
    }
    
    Task* TaskManager::GetTaskToExecute(uint32_t currentThreadQueueIdx)
    {
        auto ourQueue = _taskQueues[currentThreadQueueIdx];
{'fraction_non_alphanumeric': 0.0982469659025236, 'fraction_numerical': 0.005875553843190137, 'mean_word_length': 2.351517107811491, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3382483', 'n_tokens_mistral': 3166, 'n_tokens_neox': 3038, 'n_words': 563}