TaskQueue.hpp

#pragma once
 
#include <Core/CoreMacros.hpp>
#include <Core/RaCore.hpp>
#include <Core/Utils/Index.hpp>
#include <Core/Utils/Timer.hpp> // Ra::Core::TimePoint
#include <atomic>
#include <condition_variable>
#include <deque>
#include <iosfwd>
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <string>
#include <thread>
#include <utility>
#include <vector>
 
namespace Ra {
namespace Core {
class Task;
}
} // namespace Ra
 
namespace Ra {
namespace Core {
class RA_CORE_API TaskQueue
{
  public:
    using TaskId = Utils::Index;
 
    struct TimerData {
        Utils::TimePoint start;
        Utils::TimePoint end;
        uint threadId;
        std::string taskName;
    };
 
  public:
    explicit TaskQueue( uint numThreads );
 
    ~TaskQueue();
 
    //
    // Task management
    //
 
    TaskId registerTask( std::unique_ptr<Task> task );
 
    void removeTask( TaskId taskId );
 
    TaskId getTaskId( const std::string& taskName ) const;
 
    void addDependency( TaskId predecessor, TaskId successor );
 
    bool addDependency( const std::string& predecessors, TaskId successor );
    bool addDependency( TaskId predecessor, const std::string& successors );
 
    void addPendingDependency( const std::string& predecessors, TaskId successor );
    void addPendingDependency( TaskId predecessor, const std::string& successors );
 
    //
    // Task queue operations
    //
 
    void startTasks();
 
    void runTasksInThisThread();
 
    void waitForTasks();
 
    const std::vector<TimerData>& getTimerData();
 
    void flushTaskQueue();
 
    void printTaskGraph( std::ostream& output ) const;
 
  private:
    void runThread( uint id );
 
    void queueTask( TaskId task );
 
    void detectCycles();
 
    void resolveDependencies();
 
  private:
    using wlock = std::unique_lock<std::shared_mutex>;
    using rlock = std::shared_lock<std::shared_mutex>;
 
    std::vector<std::thread> m_workerThreads;
 
    //
    // mutex protected variables.
    //
 
    std::vector<std::unique_ptr<Task>> m_tasks;
    std::vector<std::vector<TaskId>> m_dependencies;
 
    std::vector<std::pair<TaskId, std::string>> m_pendingDepsPre;
    std::vector<std::pair<std::string, TaskId>> m_pendingDepsSucc;
 
    std::vector<TimerData> m_timerData;
 
    std::vector<uint> m_remainingDependencies;
    std::deque<TaskId> m_taskQueue;
    uint m_processingTasks;
 
    std::atomic_bool m_shuttingDown;
    std::condition_variable_any m_threadNotifier;
    std::condition_variable_any m_waitForTasksNotifier;
 
    mutable std::shared_mutex m_mutex;
};
 
} // namespace Core
} // namespace Ra