pars::ev::runner Struct Reference

#include <runner.h>

Public Member Functions
	runner (hf_registry &hfs)
std::size_t	next_job_id ()
void	start_thread (std::size_t spec_hash, std::packaged_task< void(std::stop_token, job)> task, job j)
auto	count_threads ()
void	stop_all_threads ()
bool	can_exec (int s_id, std::size_t spec_hash)
void	exec (job j)
template<template< typename > typename kind_of, event_c event_t> requires kind_c<kind_of>
void	exec (kind_of< event_t > ke)
void	add_pipe (const net::pipe &p)
void	remove_pipe (const net::pipe &p)
	stop all running and remove all pending jobs for pipe p
void	associate_job_to_pipe (const int j_id, const int p_id)
void	stop_thread (const int j_id)

Detailed Description

Definition at line 45 of file runner.h.

Constructor & Destructor Documentation

runner()

pars::ev::runner::runner ( hf_registry & hfs )

inline

Definition at line 48 of file runner.h.

    : hf_registry_m{hfs}
  {
  }

Member Function Documentation

add_pipe()

void pars::ev::runner::add_pipe ( const net::pipe & p )

inline

Definition at line 150 of file runner.h.

  {
    auto guard = std::lock_guard{mtx_m};
 
    auto p_id = p.id();
 
    pipe_jobs_m.insert({p_id, {}});
  }

References pars::net::pipe::id().

Here is the call graph for this function:

associate_job_to_pipe()

void pars::ev::runner::associate_job_to_pipe ( const int j_id, const int p_id )

inline

Definition at line 175 of file runner.h.

  {
    if (j_id <= 0)
      throw std::runtime_error(std::format("Job #{}: invalid Job!", p_id));
 
    if (p_id <= 0)
      return;
 
    auto guard = std::lock_guard{mtx_m};
 
    // add j_id to the jobs related to p, unless p was removed with a previous
    // call to remove_pipe(p) (ie: pipe_jobs_m does not contains p.id())
    if (pipe_jobs_m.contains(p_id))
    {
      pipe_jobs_m.at(p_id).push_back(j_id);
 
      pars::debug(SL, lf::event,
                  "Job #{} pushed and associated to Pipe {:X} [# size: {}]",
                  j_id, p_id, pipe_jobs_m.at(p_id).size());
    }
  }

References pars::debug(), pars::event, and SL.

Here is the call graph for this function:

can_exec()

bool pars::ev::runner::can_exec ( int s_id, std::size_t spec_hash )

inline

Definition at line 95 of file runner.h.

  {
    auto lock = hf_registry_m.lock();
 
    return hf_registry_m.has_handler_for(s_id, spec_hash);
  }

count_threads()

auto pars::ev::runner::count_threads ( )

inline

Definition at line 72 of file runner.h.

  {
    auto guard = std::lock_guard{mtx_m};
 
    return futures_m.size();
  }

exec() [1/2]

void pars::ev::runner::exec ( job j )

inline

Definition at line 102 of file runner.h.

  {
    auto spec_hash = j.spec_hash();
 
    auto s_id = j.socket_id();
 
    auto lock = hf_registry_m.lock();
 
    if (!hf_registry_m.has_handler_for(s_id, spec_hash))
    {
      pars::err(SL, lf::event,
                "Unable to find handler for Spec 0x{:X} on Socket {}, skip "
                "message ...",
                spec_hash, s_id);
 
      return;
    }
 
    auto& hf = hf_registry_m.handler_for(s_id, spec_hash);
 
    lock.unlock();
 
    pars::debug(SL, lf::event, "Job #{}: Running Handler [{}]", j.id(),
                demangle(hf_registry_m.type_for(spec_hash)->name()));
 
    try
    {
      {
        auto guard = std::lock_guard{mtx_m};
 
        process_futures();
      }
 
      hf(std::move(j));
    }
    catch (...)
    {
      process_exception(s_id, spec_hash);
    }
  }

References pars::debug(), pars::err(), pars::event, pars::ev::job::id(), SL, pars::ev::job::socket_id(), and pars::ev::job::spec_hash().

Referenced by exec().

Here is the call graph for this function:

Here is the caller graph for this function:

exec() [2/2]

template<template< typename > typename kind_of, event_c event_t>
requires kind_c<kind_of>

void pars::ev::runner::exec ( kind_of< event_t > ke )

inline

Definition at line 145 of file runner.h.

  {
    exec(make_job(next_job_id(), std::move(ke)));
  }

References exec(), and next_job_id().

Here is the call graph for this function:

next_job_id()

std::size_t pars::ev::runner::next_job_id ( )

inline

Definition at line 53 of file runner.h.

{ return next_job_id_m++; }

Referenced by exec().

Here is the caller graph for this function:

remove_pipe()

void pars::ev::runner::remove_pipe ( const net::pipe & p )

inline

stop all running and remove all pending jobs for pipe p

Definition at line 160 of file runner.h.

  {
    auto guard = std::lock_guard{mtx_m};
 
    auto p_id = p.id();
 
    if (!pipe_jobs_m.at(p_id).empty())
    {
      // stop all running jobs for pipe p
      stop_runnings(pipe_jobs_m.at(p_id));
    }
 
    pipe_jobs_m.erase(p_id);
  }

References pars::net::pipe::id().

Here is the call graph for this function:

start_thread()

void pars::ev::runner::start_thread ( std::size_t spec_hash, std::packaged_task< void(std::stop_token, job)> task, job j )

inline

Definition at line 55 of file runner.h.

  {
    auto j_id = j.id();
 
    auto s_id = j.socket_id();
 
    auto guard = std::lock_guard{mtx_m};
 
    futures_m.emplace_back(j_id, s_id, spec_hash, task.get_future());
 
    auto th_res = threads_m.try_emplace(j_id, std::move(task), std::move(j));
 
    if (!th_res.second)
      futures_m.pop_back();
  }

References pars::ev::job::id(), and pars::ev::job::socket_id().

Here is the call graph for this function:

stop_all_threads()

void pars::ev::runner::stop_all_threads ( )

inline

Definition at line 79 of file runner.h.

  {
    auto guard = std::lock_guard{mtx_m};
 
    for (const auto& t : threads_m)
    {
      if (stop_possible(t.first))
        request_stop(t.first);
    }
 
    while (!futures_m.empty())
      process_futures();
 
    pipe_jobs_m.clear();
  }

stop_thread()

void pars::ev::runner::stop_thread ( const int j_id )

inline

Definition at line 197 of file runner.h.

  {
    auto guard = std::lock_guard(mtx_m);
 
    if (stop_possible(j_id))
      request_stop(j_id);
  }

---

The documentation for this struct was generated from the following file:

• /home/runner/work/pars-cxx/pars-cxx/lib/include/pars/ev/runner.h