<commit_before>#include <Core/Tasks/TaskQueue.hpp>
#include <Core/Tasks/Task.hpp>

#include <stack>
#include <iostream>
#include <algorithm>

namespace Ra
{
    namespace Core
    {

        TaskQueue::TaskQueue( uint numThreads )
            : m_processingTasks( 0 ), m_shuttingDown( false )
        {
            CORE_ASSERT( numThreads > 0, " You need at least one thread" );
            m_workerThreads.reserve( numThreads );
            for ( uint i = 0 ; i < numThreads; ++i )
            {
                m_workerThreads.emplace_back( std::thread( &TaskQueue::runThread, this, i ) );
            }
        }

        TaskQueue::~TaskQueue()
        {
            flushTaskQueue();
            m_shuttingDown = true;
            m_threadNotifier.notify_all();
            for ( auto& t :  m_workerThreads )
            {
                t.join();
            }
        }

        TaskQueue::TaskId TaskQueue::registerTask( Task* task )
        {
            m_tasks.emplace_back( std::unique_ptr<Task> ( task ) );
            m_dependencies.push_back( std::vector<TaskId>() );
            m_remainingDependencies.push_back( 0 );
            TimerData tdata;
            tdata.taskName = task->getName();
            m_timerData.push_back( tdata );

            CORE_ASSERT( m_tasks.size() == m_dependencies.size(), "Inconsistent task list" );
            CORE_ASSERT( m_tasks.size() == m_remainingDependencies.size(), "Inconsistent task list" );
            CORE_ASSERT( m_tasks.size() == m_timerData.size(), "Inconsistent task list" );
            return TaskId( m_tasks.size() - 1 );
        }

        void TaskQueue::addDependency( TaskQueue::TaskId predecessor, TaskQueue::TaskId successor )
        {
            CORE_ASSERT( ( predecessor != InvalidTaskId ) && ( predecessor < m_tasks.size() ), "Invalid predecessor task" );
            CORE_ASSERT( ( successor != InvalidTaskId )   && ( successor < m_tasks.size() ), "Invalid successor task" );
            CORE_ASSERT( predecessor != successor, "Cannot add self-dependency" );

            CORE_ASSERT( std::find( m_dependencies[predecessor].begin(), m_dependencies[predecessor].end(), successor ) == m_dependencies[predecessor].end(), "Cannot add a dependency twice" );

            m_dependencies[predecessor].push_back( successor );
            ++m_remainingDependencies[successor];
        }

        bool TaskQueue::addDependency(const std::string &predecessors, TaskQueue::TaskId successor)
        {
            bool added = false;
            for (uint i = 0; i < m_tasks.size(); ++i)
            {
                if (m_tasks[i]->getName() == predecessors )
                {
                    added = true;
                    addDependency( i, successor);
                }
            }
            return added;
        }

        bool TaskQueue::addDependency(TaskQueue::TaskId predecessor, const std::string &successors)
        {
            bool added = false;
            for (uint i = 0; i < m_tasks.size(); ++i)
            {
                if (m_tasks[i]->getName() == successors )
                {
                    added = true;
                    addDependency( predecessor, i );
                }
            }
            return added;
        }

        void TaskQueue::addPendingDependency(const std::string &predecessors, TaskQueue::TaskId successor)
        {
            m_pendingDepsSucc.push_back(std::make_pair(predecessors, successor));
        }

        void TaskQueue::addPendingDependency( TaskId predecessor, const std::string& successors)
        {
            m_pendingDepsPre.push_back(std::make_pair(predecessor, successors));
        }

        void TaskQueue::resolveDependencies()
        {
           for ( const auto& pre : m_pendingDepsPre )
           {
               ON_ASSERT(bool result =) addDependency( pre.first, pre.second );
               CORE_WARN_IF( !result, "Pending dependency unresolved : " << pre.second);
           }
           for ( const auto& pre : m_pendingDepsSucc )
           {
               ON_ASSERT(bool result =) addDependency( pre.first, pre.second );
               CORE_WARN_IF( result, "Pending dependency unresolved : " << pre.first );
           }
           m_pendingDepsPre.clear();
           m_pendingDepsSucc.clear();
        }

        void TaskQueue::queueTask( TaskQueue::TaskId task )
        {
            CORE_ASSERT( m_remainingDependencies[task] == 0, " Task" << m_tasks[task]->getName() <<"has unmet dependencies" );
            m_taskQueue.push_front( task );
        }

        void TaskQueue::detectCycles()
        {
#if defined (CORE_DEBUG)
            // Do a depth-first search of the nodes.
            std::vector<bool> visited( m_tasks.size(), false );
            std::stack<TaskId> pending;

            for (TaskId id = 0; id < m_tasks.size(); ++id)
            {
                if ( m_dependencies[id].size() == 0 )<commit_msg>Fix task queue assert, added better message<commit_after>#include <Core/Tasks/TaskQueue.hpp>
#include <Core/Tasks/Task.hpp>

#include <stack>
#include <iostream>
#include <algorithm>

namespace Ra
{
    namespace Core
    {

        TaskQueue::TaskQueue( uint numThreads )
            : m_processingTasks( 0 ), m_shuttingDown( false )
        {
            CORE_ASSERT( numThreads > 0, " You need at least one thread" );
            m_workerThreads.reserve( numThreads );
            for ( uint i = 0 ; i < numThreads; ++i )
            {
                m_workerThreads.emplace_back( std::thread( &TaskQueue::runThread, this, i ) );
            }
        }

        TaskQueue::~TaskQueue()
        {
            flushTaskQueue();
            m_shuttingDown = true;
            m_threadNotifier.notify_all();
            for ( auto& t :  m_workerThreads )
            {
                t.join();
            }
        }

        TaskQueue::TaskId TaskQueue::registerTask( Task* task )
        {
            m_tasks.emplace_back( std::unique_ptr<Task> ( task ) );
            m_dependencies.push_back( std::vector<TaskId>() );
            m_remainingDependencies.push_back( 0 );
            TimerData tdata;
            tdata.taskName = task->getName();
            m_timerData.push_back( tdata );

            CORE_ASSERT( m_tasks.size() == m_dependencies.size(), "Inconsistent task list" );
            CORE_ASSERT( m_tasks.size() == m_remainingDependencies.size(), "Inconsistent task list" );
            CORE_ASSERT( m_tasks.size() == m_timerData.size(), "Inconsistent task list" );
            return TaskId( m_tasks.size() - 1 );
        }

        void TaskQueue::addDependency( TaskQueue::TaskId predecessor, TaskQueue::TaskId successor )
        {
            CORE_ASSERT( ( predecessor != InvalidTaskId ) && ( predecessor < m_tasks.size() ), "Invalid predecessor task" );
            CORE_ASSERT( ( successor != InvalidTaskId )   && ( successor < m_tasks.size() ), "Invalid successor task" );
            CORE_ASSERT( predecessor != successor, "Cannot add self-dependency" );

            CORE_ASSERT( std::find( m_dependencies[predecessor].begin(), m_dependencies[predecessor].end(), successor ) == m_dependencies[predecessor].end(), "Cannot add a dependency twice" );

            m_dependencies[predecessor].push_back( successor );
            ++m_remainingDependencies[successor];
        }

        bool TaskQueue::addDependency(const std::string &predecessors, TaskQueue::TaskId successor)
        {
            bool added = false;
            for (uint i = 0; i < m_tasks.size(); ++i)
            {
                if (m_tasks[i]->getName() == predecessors )
                {
                    added = true;
                    addDependency( i, successor);
                }
            }
            return added;
        }

        bool TaskQueue::addDependency(TaskQueue::TaskId predecessor, const std::string &successors)
        {
            bool added = false;
            for (uint i = 0; i < m_tasks.size(); ++i)
            {
                if (m_tasks[i]->getName() == successors )
                {
                    added = true;
                    addDependency( predecessor, i );
                }
            }
            return added;
        }

        void TaskQueue::addPendingDependency(const std::string &predecessors, TaskQueue::TaskId successor)
        {
            m_pendingDepsSucc.push_back(std::make_pair(predecessors, successor));
        }

        void TaskQueue::addPendingDependency( TaskId predecessor, const std::string& successors)
        {
            m_pendingDepsPre.push_back(std::make_pair(predecessor, successors));
        }

        void TaskQueue::resolveDependencies()
        {
           for ( const auto& pre : m_pendingDepsPre )
           {
               ON_ASSERT(bool result =) addDependency( pre.first, pre.second );
               CORE_WARN_IF( !result, "Pending dependency unresolved : " << m_tasks[pre.first]->getName() <<" -> (" << pre.second<<")");
           }
           for ( const auto& pre : m_pendingDepsSucc )
           {
               ON_ASSERT(bool result =) addDependency( pre.first, pre.second );
               CORE_WARN_IF( !result, "Pending dependency unresolved : (" << pre.first  <<") -> "<< m_tasks[pre.second]->getName());
           }
           m_pendingDepsPre.clear();
           m_pendingDepsSucc.clear();
        }

        void TaskQueue::queueTask( TaskQueue::TaskId task )
        {
            CORE_ASSERT( m_remainingDependencies[task] == 0, " Task" << m_tasks[task]->getName() <<"has unmet dependencies" );
            m_taskQueue.push_front( task );
        }

        void TaskQueue::detectCycles()
        {
#if defined (CORE_DEBUG)
            // Do a depth-first search of the nodes.
            std::vector<bool> visited( m_tasks.size(), false );
            std::stack<TaskId> pending;

            for (TaskId id = 0; id < m_tasks.size(); ++id)
            {
                if ( m_dependencies[id].size() == 0 )
{'fraction_non_alphanumeric': 0.11810786381440193, 'fraction_numerical': 0.002008637139700713, 'mean_word_length': 1.916813122437024, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 35, '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': '4598666', 'n_tokens_mistral': 2861, 'n_tokens_neox': 2688, 'n_words': 576}