Line data    Source code

       1             : /*!
       2             :  * \file esys/repo/manifest/runtasks.h
       3             :  * \brief Execute a list of tasks
       4             :  *
       5             :  * \cond
       6             :  * __legal_b__
       7             :  *
       8             :  * Copyright (c) 2020 Michel Gillet
       9             :  * Distributed under the MIT License.
      10             :  * (See accompanying file LICENSE.txt or
      11             :  * copy at https://opensource.org/licenses/MIT)
      12             :  *
      13             :  * __legal_e__
      14             :  * \endcond
      15             :  *
      16             :  */
      17             : 
      18             : #pragma once
      19             : 
      20             : #include "esys/repo/esysrepo_defs.h"
      21             : #include "esys/repo/queuemt.h"
      22             : #include "esys/repo/manifest/taskbase.h"
      23             : #include "esys/repo/manifest/workerthread.h"
      24             : #include "esys/repo/progress/consoleobserver.h"
      25             : #include "esys/repo/progress/progresssession.h"
      26             : 
      27             : #include <esys/log/user.h>
      28             : 
      29             : #include <string>
      30             : #include <memory>
      31             : #include <vector>
      32             : #include <thread>
      33             : #include <atomic>
      34             : #include <condition_variable>
      35             : 
      36             : namespace esys
      37             : {
      38             : 
      39             : namespace repo
      40             : {
      41             : 
      42             : namespace manifest
      43             : {
      44             : 
      45             : /*! \class RunTasks esys/repo/manifest/runtasks.h "esys/repo/manifest/runtasks.h"
      46             :  * \brief Execute a list of tasks
      47             :  */
      48             : class ESYSREPO_API RunTasks : public log::User
      49             : {
      50             : public:
      51             :     //! Default constructor
      52             :     RunTasks();
      53             : 
      54             :     //! Destructor
      55             :     ~RunTasks() override;
      56             : 
      57             :     //! Add a task to do
      58             :     /*!
      59             :      * \param[in] task the task
      60             :      */
      61             :     void add(std::shared_ptr<TaskBase> task);
      62             : 
      63             :     //! Add a task with a fixed progress/queue id (e.g. manifest repo index)
      64             :     /*!
      65             :      * \param[in] task the task
      66             :      * \param[in] id stable id used for progress events (``RepoAssignedEvent`` etc.)
      67             :      */
      68             :     void add(std::shared_ptr<TaskBase> task, std::size_t id);
      69             : 
      70             :     //! Set the job count
      71             :     /*!
      72             :      * \param[in] job_count the job count
      73             :      */
      74             :     void set_job_count(std::size_t job_count);
      75             : 
      76             :     //! Get the job count
      77             :     /*!
      78             :      * \return the job count
      79             :      */
      80             :     std::size_t get_job_count() const;
      81             : 
      82             :     //! Set the progress session used while running tasks (non-owning)
      83             :     /*!
      84             :      * \param[in] progress_session the session, or nullptr
      85             :      */
      86             :     void set_progress_session(progress::ProgressSession *progress_session);
      87             : 
      88             :     //! Get the progress session
      89             :     /*!
      90             :      * \return the session, or nullptr
      91             :      */
      92             :     progress::ProgressSession *get_progress_session() const;
      93             : 
      94             :     //! Enable or disable the default console \\r status observer
      95             :     /*!
      96             :      * When no external ProgressSession is set, RunTasks owns a session and
      97             :      * registers a ConsoleObserver if enabled (default true). Quiet / non-TTY
      98             :      * callers may disable this to suppress the status line.
      99             :      *
     100             :      * \param[in] console_progress true to show the compact status line
     101             :      */
     102             :     void set_console_progress(bool console_progress);
     103             : 
     104             :     //! Whether the default console status observer is enabled
     105             :     /*!
     106             :      * \return true if enabled
     107             :      */
     108             :     bool get_console_progress() const;
     109             : 
     110             :     //! Mark dispatch as paused (visibility + future controls; ADR-0018)
     111             :     void set_paused(bool paused);
     112             : 
     113             :     //! Whether dispatch is paused
     114             :     bool get_paused() const;
     115             : 
     116             :     //! Mark dispatch as under backpressure (visibility; ADR-0018)
     117             :     void set_backpressure(bool backpressure);
     118             : 
     119             :     //! Whether dispatch is under backpressure
     120             :     bool get_backpressure() const;
     121             : 
     122             :     //! Execute all tasks
     123             :     /*!
     124             :      * \return 0 if successful, < 0 otherwise
     125             :      */
     126             :     Result run();
     127             : 
     128             :     //! Get the worker thread count
     129             :     /*!
     130             :      * \return the worker thread count
     131             :      */
     132             :     std::size_t get_worker_thread_count();
     133             : 
     134             :     //! worker thread
     135             :     /*!
     136             :      * \param[in] work_thread std::shared_ptr<WorkerThread> work_thread
     137             :      */
     138             :     void worker_thread(std::shared_ptr<WorkerThread> work_thread);
     139             : 
     140             :     //! Redraw thread: prints ConsoleObserver status (not a progress producer)
     141             :     void cout_thread();
     142             : 
     143             :     //! wait done
     144             :     void wait_done();
     145             : 
     146             : protected:
     147             :     //!< \cond DOXY_IMPL
     148             : 
     149             :     //! Remove a worker thread from the active list
     150             :     /*!
     151             :      * \param[in] worker_thread the worker thread to remove
     152             :      */
     153             :     void remove(std::shared_ptr<WorkerThread> worker_thread);
     154             : 
     155             :     //! Publish a SchedulerStateEvent snapshot (no-op without a progress session)
     156             :     void push_scheduler_state();
     157             : 
     158             :     class RemoveWorkerThreadGuard
     159             :     {
     160             :     public:
     161          76 :         RemoveWorkerThreadGuard(RunTasks *run_tasks, std::shared_ptr<WorkerThread> worker_thread)
     162          76 :             : m_run_tasks(run_tasks)
     163         152 :             , m_work_thread(worker_thread)
     164             :         {
     165             :         }
     166             : 
     167          76 :         ~RemoveWorkerThreadGuard()
     168             :         {
     169         152 :             m_run_tasks->remove(m_work_thread);
     170          76 :         }
     171             : 
     172             :     private:
     173             :         //!< \cond DOXY_IMPL
     174             :         RunTasks *m_run_tasks = nullptr;
     175             :         std::shared_ptr<WorkerThread> m_work_thread;
     176             :         //!< \endcond
     177             :     };
     178             : 
     179             : private:
     180             :     //!< \cond DOXY_IMPL
     181             :     std::atomic_bool m_done{false};
     182             :     std::atomic_bool m_any_failed{false};
     183             :     std::condition_variable m_done_cond;
     184             :     std::mutex m_done_mutex;
     185             :     QueueMT<std::shared_ptr<TaskBase>> m_tasks;
     186             :     QueueMT<std::shared_ptr<TaskBase>> m_tasks_done;
     187             :     QueueMT<std::shared_ptr<TaskBase>> m_tasks_running;
     188             :     std::vector<std::thread> m_threads;
     189             :     QueueMT<std::shared_ptr<WorkerThread>> m_worker_threads;
     190             :     std::thread m_cout_thread;
     191             :     std::size_t m_job_count = 1;
     192             :     std::size_t m_cout_interval_ms = 500;
     193             :     std::size_t m_task_id = 0;
     194             :     std::size_t m_worker_thread_count = 0;
     195             :     progress::ProgressSession *m_progress_session = nullptr;
     196             :     progress::ConsoleObserver *m_console_observer = nullptr;
     197             :     bool m_console_progress = true;
     198             :     std::atomic_bool m_paused{false};
     199             :     std::atomic_bool m_backpressure{false};
     200             :     //!< \endcond
     201             : };
     202             : 
     203             : } // namespace manifest
     204             : 
     205             : } // namespace repo
     206             : 
     207             : } // namespace esys