Line data    Source code

       1             : /*!
       2             :  * \file esys/repo/manifest/runtasks_manifest.cpp
       3             :  * \brief
       4             :  *
       5             :  * \cond
       6             :  * __legal_b__
       7             :  *
       8             :  * Copyright (c) 2020-2023 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             : #include "esys/repo/esysrepo_prec.h"
      19             : #include "esys/repo/manifest/runtasks.h"
      20             : #include "esys/repo/progress/repoassignedevent.h"
      21             : #include "esys/repo/progress/repodoneevent.h"
      22             : #include "esys/repo/progress/repofailedevent.h"
      23             : #include "esys/repo/progress/schedulerstateevent.h"
      24             : 
      25             : #include <esys/log/consolelockguard.h>
      26             : 
      27             : #include <esys/trace/call.h>
      28             : #include <esys/trace/macros.h>
      29             : 
      30             : #include <chrono>
      31             : #include <sstream>
      32             : #include <iostream>
      33             : 
      34             : namespace esys::repo::manifest
      35             : {
      36             : 
      37          55 : RunTasks::RunTasks()
      38         275 :     : log::User()
      39             : {
      40          55 : }
      41             : 
      42          55 : RunTasks::~RunTasks()
      43             : {
      44          55 :     m_tasks.clear();
      45         275 : }
      46             : 
      47           2 : void RunTasks::add(std::shared_ptr<TaskBase> task)
      48             : {
      49           4 :     add(task, m_task_id);
      50           2 : }
      51             : 
      52         122 : void RunTasks::add(std::shared_ptr<TaskBase> task, std::size_t id)
      53             : {
      54         122 :     task->set_id(id);
      55         122 :     if (id >= m_task_id) m_task_id = id + 1;
      56         122 :     m_tasks.push_back(task);
      57         122 : }
      58             : 
      59          55 : void RunTasks::set_job_count(std::size_t job_count)
      60             : {
      61          55 :     m_job_count = job_count;
      62          55 : }
      63             : 
      64         186 : std::size_t RunTasks::get_job_count() const
      65             : {
      66         186 :     return m_job_count;
      67             : }
      68             : 
      69           1 : void RunTasks::set_progress_session(progress::ProgressSession *progress_session)
      70             : {
      71           1 :     m_progress_session = progress_session;
      72           1 : }
      73             : 
      74           0 : progress::ProgressSession *RunTasks::get_progress_session() const
      75             : {
      76           0 :     return m_progress_session;
      77             : }
      78             : 
      79           0 : void RunTasks::set_console_progress(bool console_progress)
      80             : {
      81           0 :     m_console_progress = console_progress;
      82           0 : }
      83             : 
      84           0 : bool RunTasks::get_console_progress() const
      85             : {
      86           0 :     return m_console_progress;
      87             : }
      88             : 
      89           0 : void RunTasks::set_paused(bool paused)
      90             : {
      91           0 :     m_paused.store(paused);
      92           0 :     push_scheduler_state();
      93           0 : }
      94             : 
      95           0 : bool RunTasks::get_paused() const
      96             : {
      97           0 :     return m_paused.load();
      98             : }
      99             : 
     100           0 : void RunTasks::set_backpressure(bool backpressure)
     101             : {
     102           0 :     m_backpressure.store(backpressure);
     103           0 :     push_scheduler_state();
     104           0 : }
     105             : 
     106           0 : bool RunTasks::get_backpressure() const
     107             : {
     108           0 :     return m_backpressure.load();
     109             : }
     110             : 
     111         355 : void RunTasks::push_scheduler_state()
     112             : {
     113         355 :     if (m_progress_session == nullptr) return;
     114             : 
     115         710 :     m_progress_session->push(progress::SchedulerStateEvent(
     116         355 :         m_tasks.size(), m_tasks_running.size(), m_worker_threads.size(), m_paused.load(), m_backpressure.load()));
     117             : }
     118             : 
     119         111 : std::size_t RunTasks::get_worker_thread_count()
     120             : {
     121         111 :     m_worker_thread_count = get_job_count();
     122             : 
     123         111 :     if (m_tasks.size() < m_worker_thread_count) m_worker_thread_count = m_tasks.size();
     124             : 
     125         111 :     return m_worker_thread_count;
     126             : }
     127             : 
     128          35 : Result RunTasks::run()
     129             : {
     130          35 :     Result result;
     131          35 :     ETRC_CALL_RET_NP(result);
     132             : 
     133          35 :     if (m_tasks.size() == 0) return ETRC_RET(ESYSREPO_RESULT(ResultCode::RUNTASKS_NO_TASK));
     134          35 :     if (static_cast<int>(get_job_count()) < 0) return ETRC_RET(ESYSREPO_RESULT(ResultCode::JOB_COUNT_NEGATIVE));
     135             : 
     136          35 :     progress::ProgressSession owned_session;
     137          35 :     progress::ConsoleObserver console_observer;
     138          35 :     progress::ProgressSession *const external_session = m_progress_session;
     139          35 :     const bool own_session = (external_session == nullptr);
     140             : 
     141          35 :     if (own_session) m_progress_session = &owned_session;
     142             : 
     143          35 :     const bool console_active = m_console_progress && own_session;
     144          35 :     if (console_active)
     145             :     {
     146          34 :         console_observer.set_enabled(true);
     147          34 :         m_progress_session->add_observer(&console_observer);
     148          34 :         m_console_observer = &console_observer;
     149             :     }
     150             :     else
     151             :     {
     152           1 :         m_console_observer = nullptr;
     153             :     }
     154             : 
     155          35 :     m_done = false;
     156          35 :     m_any_failed = false;
     157             : 
     158         111 :     for (int idx = 0; idx < get_worker_thread_count(); ++idx)
     159             :     {
     160          76 :         std::shared_ptr<WorkerThread> worker = std::make_shared<WorkerThread>(idx);
     161             : 
     162         152 :         m_threads.push_back(std::thread(&RunTasks::worker_thread, this, worker));
     163             : 
     164          76 :         m_worker_threads.push_back(worker);
     165          76 :     }
     166             : 
     167          35 :     push_scheduler_state();
     168             : 
     169          35 :     if (console_active) m_cout_thread = std::thread(&RunTasks::cout_thread, this);
     170             : 
     171          35 :     wait_done();
     172             : 
     173         111 :     for (auto &thread : m_threads)
     174             :     {
     175          76 :         if (thread.joinable()) thread.join();
     176             :     }
     177             : 
     178          35 :     if (m_cout_thread.joinable()) m_cout_thread.join();
     179             : 
     180          35 :     if (console_active) m_progress_session->remove_observer(&console_observer);
     181          35 :     m_console_observer = nullptr;
     182          35 :     if (own_session) m_progress_session = external_session;
     183             : 
     184          35 :     m_threads.clear();
     185          35 :     m_worker_threads.clear();
     186             : 
     187          36 :     if (m_any_failed) return ETRC_RET(ESYSREPO_RESULT(ResultCode::GIT_GENERIC_ERROR));
     188             : 
     189          68 :     return ETRC_RET(ESYSREPO_RESULT(ResultCode::OK));
     190          35 : }
     191             : 
     192          76 : void RunTasks::worker_thread(std::shared_ptr<WorkerThread> thread)
     193             : {
     194         228 :     RemoveWorkerThreadGuard remove_guard(this, thread);
     195          76 :     std::shared_ptr<TaskBase> task;
     196             : 
     197          76 :     std::ostringstream oss;
     198          76 :     oss << "Thread " << thread->get_id();
     199         152 :     ETRC_SET_THREAD_NAME(oss.str().c_str());
     200             : 
     201          76 :     int result = m_tasks.pop_front(task);
     202         198 :     while (result == 0)
     203             :     {
     204         122 :         task->set_worker_id(thread->get_id());
     205         122 :         task->set_progress_session(m_progress_session);
     206             : 
     207         122 :         if (m_progress_session != nullptr)
     208         244 :             m_progress_session->push(progress::RepoAssignedEvent(task->get_id(), thread->get_id()));
     209             : 
     210         122 :         m_tasks_running.push_back(task);
     211         122 :         push_scheduler_state();
     212             : 
     213         122 :         try
     214             :         {
     215         122 :             result = task->run();
     216         122 :             if (result < 0) task->set_result(result);
     217             :         }
     218           0 :         catch (const std::exception &e)
     219             :         {
     220           0 :             task->set_result(-1);
     221           0 :             task->add_errors(e.what());
     222           0 :         }
     223           0 :         catch (...)
     224             :         {
     225           0 :             task->set_result(-1);
     226           0 :             task->add_errors("Unknown exception.");
     227           0 :         }
     228         122 :         if ((result < 0) || (task->get_result() < 0))
     229             :         {
     230           1 :             m_any_failed = true;
     231           2 :             auto l_result = ESYSREPO_RESULT(ResultCode::GIT_GENERIC_ERROR);
     232           1 :             std::ostringstream err_oss;
     233           1 :             l_result.print(err_oss);
     234           1 :             error(err_oss.str());
     235           2 :             for (const auto &err : task->get_errors()) error(err);
     236             : 
     237           1 :             if (m_progress_session != nullptr)
     238             :             {
     239           1 :                 std::string fail_msg = "task failed";
     240           1 :                 if (!task->get_errors().empty()) fail_msg = task->get_errors().front();
     241           3 :                 m_progress_session->push(progress::RepoFailedEvent(task->get_id(), fail_msg));
     242           1 :             }
     243           1 :         }
     244         121 :         else if (m_progress_session != nullptr)
     245             :         {
     246         242 :             m_progress_session->push(progress::RepoDoneEvent(task->get_id()));
     247             :         }
     248             : 
     249         122 :         task->set_progress_session(nullptr);
     250         122 :         m_tasks_done.push_back(task);
     251             : 
     252         122 :         std::size_t id = task->get_id();
     253         302 :         auto remove_fct = [id](const std::shared_ptr<TaskBase> &task) { return (task->get_id() == id); };
     254         122 :         m_tasks_running.remove_if(remove_fct);
     255             : 
     256         122 :         push_scheduler_state();
     257             : 
     258         122 :         result = m_tasks.pop_front(task);
     259             :     }
     260          76 : }
     261             : 
     262          34 : void RunTasks::cout_thread()
     263             : {
     264         233 :     while (!m_done)
     265             :     {
     266         199 :         std::this_thread::sleep_for(std::chrono::milliseconds(m_cout_interval_ms));
     267             : 
     268         199 :         if (m_console_observer == nullptr) continue;
     269             : 
     270         199 :         const std::string line = m_console_observer->format_status_line();
     271         199 :         if (line.empty()) continue;
     272         165 :         {
     273         165 :             log::ConsoleLockGuard<log::User> lock(this);
     274             : 
     275             :             // Pad so a shorter line fully clears a previous longer \\r status.
     276         165 :             std::cout << '\r' << line << "                    " << std::flush;
     277         165 :         }
     278         199 :     }
     279             : 
     280             :     // Finish the status line so following summary logs are not overwritten.
     281          34 :     {
     282          34 :         log::ConsoleLockGuard<log::User> lock(this);
     283          34 :         std::cout << '\n' << std::flush;
     284          34 :     }
     285          34 : }
     286             : 
     287          35 : void RunTasks::wait_done()
     288             : {
     289          35 :     std::unique_lock<std::mutex> lock{m_done_mutex};
     290         145 :     m_done_cond.wait(lock, [this] { return (m_done == true); });
     291          35 : }
     292             : 
     293          76 : void RunTasks::remove(std::shared_ptr<WorkerThread> rem_worker_thread)
     294             : {
     295         380 :     auto remove_fct = [rem_worker_thread](const std::shared_ptr<WorkerThread> &worker_thread) {
     296         152 :         return (rem_worker_thread->get_id() == worker_thread->get_id());
     297          76 :     };
     298             : 
     299          76 :     m_worker_threads.remove_if(remove_fct);
     300          76 :     m_worker_thread_count = m_worker_threads.size();
     301          76 :     if (m_worker_thread_count == 0) m_done = true;
     302          76 :     m_done_cond.notify_one();
     303          76 :     push_scheduler_state();
     304          76 : }
     305             : 
     306             : } // namespace esys::repo::manifest