kcenon::thread::dag_scheduler Class Reference

DAG-based job scheduler with dependency management. More...

#include <dag_scheduler.h>

Collaboration diagram for kcenon::thread::dag_scheduler:

Public Member Functions
	dag_scheduler (std::shared_ptr< thread_pool > pool, dag_config config={})
	Constructs a DAG scheduler with a thread pool.
	~dag_scheduler ()
	Destructor - cancels any pending jobs.
	dag_scheduler (const dag_scheduler &)=delete
auto	operator= (const dag_scheduler &) -> dag_scheduler &=delete
	dag_scheduler (dag_scheduler &&) noexcept
auto	operator= (dag_scheduler &&) noexcept -> dag_scheduler &
auto	add_job (std::unique_ptr< dag_job > j) -> job_id
	Adds a job to the DAG.
auto	add_job (dag_job_builder &&builder) -> job_id
	Adds a job using a builder.
auto	add_dependency (job_id dependent, job_id dependency) -> common::VoidResult
	Adds a dependency between jobs.
auto	remove_job (job_id id) -> common::VoidResult
	Removes a job from the DAG (only if not yet started)
auto	execute_all () -> std::future< common::VoidResult >
	Executes all jobs in dependency order.
auto	execute (job_id target) -> std::future< common::VoidResult >
	Executes a specific job and its dependencies.
auto	cancel_all () -> void
	Cancels all pending jobs.
auto	wait () -> common::VoidResult
	Waits for all jobs to complete.
auto	reset () -> common::VoidResult
	Resets the scheduler for reuse.
auto	get_job_info (job_id id) const -> std::optional< dag_job_info >
	Gets information about a specific job.
auto	get_all_jobs () const -> std::vector< dag_job_info >
	Gets information about all jobs.
auto	get_jobs_in_state (dag_job_state state) const -> std::vector< dag_job_info >
	Gets jobs in a specific state.
auto	get_ready_jobs () const -> std::vector< job_id >
	Gets IDs of ready jobs (dependencies satisfied)
auto	has_cycles () const -> bool
	Checks if the DAG has cycles.
auto	get_execution_order () const -> std::vector< job_id >
	Gets topological execution order.
template<typename T >
auto	get_result (job_id id) const -> const T &
	Gets the result from a completed job.
auto	to_dot () const -> std::string
	Exports the DAG as DOT format (Graphviz)
auto	to_json () const -> std::string
	Exports the DAG as JSON format.
auto	get_stats () const -> dag_stats
	Gets execution statistics.
auto	get_config () const -> const dag_config &
	Gets the configuration.

Private Member Functions
auto	on_job_completed (job_id id) -> void
	Called when a job completes successfully.
auto	on_job_failed (job_id id, const std::string &error) -> void
	Called when a job fails.
auto	schedule_ready_jobs () -> void
	Schedules ready jobs for execution.
auto	execute_job (job_id id) -> void
	Executes a single job.
auto	topological_sort () const -> std::vector< job_id >
	Performs topological sort.
auto	detect_cycle () const -> bool
	Detects cycles using DFS.
auto	are_dependencies_satisfied (job_id id) const -> bool
	Checks if a job's dependencies are all satisfied.
auto	skip_dependents (job_id failed_id) -> void
	Marks dependents as skipped due to dependency failure.
auto	cancel_dependents (job_id failed_id) -> void
	Cancels dependents due to dependency failure.
auto	is_execution_complete () const -> bool
	Checks if execution is complete.

Static Private Member Functions
static auto	get_state_color (dag_job_state state) -> std::string
	Gets the state color for DOT visualization.

Private Attributes
std::shared_ptr< thread_pool >	pool_
	Thread pool for job execution.
dag_config	config_
	Configuration.
std::unordered_map< job_id, std::unique_ptr< dag_job > >	jobs_
	Job storage (job_id -> dag_job)
std::unordered_map< job_id, std::vector< job_id > >	dependencies_
	Dependency graph (job -> jobs it depends on)
std::unordered_map< job_id, std::vector< job_id > >	dependents_
	Reverse dependency graph (job -> jobs that depend on it)
std::shared_mutex	mutex_
	Mutex for thread-safe access.
std::condition_variable_any	completion_cv_
	Condition variable for waiting on completion.
std::atomic< bool >	executing_ {false}
	Flag indicating execution is in progress.
std::atomic< bool >	cancelled_ {false}
	Flag indicating cancellation was requested.
std::atomic< std::size_t >	running_count_ {0}
	Number of jobs currently running.
std::atomic< std::size_t >	active_callbacks_ {0}
	Number of active callbacks (for safe destruction)
std::chrono::steady_clock::time_point	execution_start_time_
	Execution start time.
std::optional< common::error_info >	first_error_
	First error encountered during execution.
std::unordered_map< job_id, std::size_t >	retry_counts_
	Retry count per job.

Detailed Description

DAG-based job scheduler with dependency management.

The dag_scheduler manages jobs with dependencies, ensuring they execute in the correct order. Jobs are represented as a Directed Acyclic Graph (DAG) where edges represent dependencies.

Key Features

• Automatic dependency resolution
• Parallel execution of independent jobs
• Cycle detection
• Multiple failure handling policies
• Result passing between jobs
• DOT/JSON visualization export

Thread Safety

All public methods are thread-safe and can be called from any thread. Internal state is protected by a shared_mutex for optimal read performance.

Usage Example

auto pool = std::make_shared<thread_pool>("pool");
pool->start();
 
dag_scheduler scheduler(pool);
 
auto job_a = scheduler.add_job(
    dag_job_builder("fetch")
        .work([]{ return fetch_data(); })
        .build()
);
 
auto job_b = scheduler.add_job(
    dag_job_builder("process")
        .depends_on(job_a)
        .work([]{ return process_data(); })
        .build()
);
 
scheduler.execute_all().wait();

Definition at line 78 of file dag_scheduler.h.

Constructor & Destructor Documentation

dag_scheduler() [1/3]

kcenon::thread::dag_scheduler::dag_scheduler ( std::shared_ptr< thread_pool > pool, dag_config config = {} )

explicit

Constructs a DAG scheduler with a thread pool.

Parameters

pool	The thread pool to use for job execution
config	Optional configuration (default: dag_config{})

Definition at line 18 of file dag_scheduler.cpp.

    : pool_(std::move(pool))
    , config_(std::move(config))
{
}

~dag_scheduler()

kcenon::thread::dag_scheduler::~dag_scheduler

(

)

Destructor - cancels any pending jobs.

Definition at line 24 of file dag_scheduler.cpp.

{
    cancel_all();
 
    // Wait for in-flight callbacks to finish accessing members
    // (e.g., completion_cv_.notify_all() after mutex unlock)
    while (active_callbacks_.load(std::memory_order_acquire) > 0)
    {
        std::this_thread::yield();
    }
}

References active_callbacks_, and cancel_all().

Here is the call graph for this function:

dag_scheduler() [2/3]

kcenon::thread::dag_scheduler::dag_scheduler ( const dag_scheduler & )

delete

dag_scheduler() [3/3]

kcenon::thread::dag_scheduler::dag_scheduler ( dag_scheduler && other )

noexcept

Definition at line 36 of file dag_scheduler.cpp.

    : pool_(std::move(other.pool_))
    , config_(std::move(other.config_))
    , jobs_(std::move(other.jobs_))
    , dependencies_(std::move(other.dependencies_))
    , dependents_(std::move(other.dependents_))
    , executing_(other.executing_.load())
    , cancelled_(other.cancelled_.load())
    , running_count_(other.running_count_.load())
    , execution_start_time_(other.execution_start_time_)
    , first_error_(std::move(other.first_error_))
    , retry_counts_(std::move(other.retry_counts_))
{
}

Member Function Documentation

add_dependency()

auto kcenon::thread::dag_scheduler::add_dependency ( job_id dependent, job_id dependency ) -> common::VoidResult

nodiscard

Adds a dependency between jobs.

Parameters

dependent	The job that depends on another
dependency	The job being depended on

Returns

Success or error if cycle detected or jobs not found

Thread Safety: Thread-safe (acquires exclusive lock)

Definition at line 109 of file dag_scheduler.cpp.

{
    std::unique_lock lock(mutex_);
 
    // Check if jobs exist
    if (jobs_.find(dependent) == jobs_.end())
    {
        return make_error_result(error_code::job_invalid,
                                 "Dependent job not found: " + std::to_string(dependent));
    }
    if (jobs_.find(dependency) == jobs_.end())
    {
        return make_error_result(error_code::job_invalid,
                                 "Dependency job not found: " + std::to_string(dependency));
    }
 
    // Add dependency
    dependencies_[dependent].push_back(dependency);
    dependents_[dependency].push_back(dependent);
 
    // Check for cycles
    if (config_.detect_cycles && detect_cycle())
    {
        // Rollback
        dependencies_[dependent].pop_back();
        dependents_[dependency].pop_back();
        return make_error_result(error_code::invalid_argument,
                                 "Adding dependency would create a cycle");
    }
 
    // Update job's dependencies
    jobs_[dependent]->add_dependency(dependency);
 
    return common::ok();
}

References kcenon::thread::invalid_argument, kcenon::thread::job_invalid, and kcenon::thread::make_error_result().

Here is the call graph for this function:

add_job() [1/2]

auto kcenon::thread::dag_scheduler::add_job ( dag_job_builder && builder ) -> job_id

nodiscard

Adds a job using a builder.

Parameters

builder

The job builder

Returns

The assigned job ID

Thread Safety: Thread-safe (acquires exclusive lock)

Definition at line 104 of file dag_scheduler.cpp.

{
    return add_job(builder.build());
}

add_job() [2/2]

auto kcenon::thread::dag_scheduler::add_job ( std::unique_ptr< dag_job > j ) -> job_id

nodiscard

Adds a job to the DAG.

Parameters

j	The job to add

Returns

The assigned job ID

Thread Safety: Thread-safe (acquires exclusive lock)

Definition at line 74 of file dag_scheduler.cpp.

{
    std::unique_lock lock(mutex_);
 
    auto id = j->get_dag_id();
 
    // Add dependencies from the job
    for (const auto& dep : j->get_dependencies())
    {
        dependencies_[id].push_back(dep);
        dependents_[dep].push_back(id);
    }
 
    // Check for cycles if configured
    if (config_.detect_cycles && detect_cycle())
    {
        // Rollback
        dependencies_.erase(id);
        for (const auto& dep : j->get_dependencies())
        {
            auto& deps = dependents_[dep];
            deps.erase(std::remove(deps.begin(), deps.end(), id), deps.end());
        }
        return INVALID_JOB_ID;
    }
 
    jobs_[id] = std::move(j);
    return id;
}

References kcenon::thread::INVALID_JOB_ID.

are_dependencies_satisfied()

auto kcenon::thread::dag_scheduler::are_dependencies_satisfied ( job_id id ) const -> bool

nodiscardprivate

Checks if a job's dependencies are all satisfied.

Parameters

id	The job ID to check

Returns

true if all dependencies are completed

Definition at line 1026 of file dag_scheduler.cpp.

{
    auto it = dependencies_.find(id);
    if (it == dependencies_.end() || it->second.empty())
    {
        return true;  // No dependencies
    }
 
    for (const auto& dep : it->second)
    {
        auto job_it = jobs_.find(dep);
        if (job_it == jobs_.end())
        {
            return false;  // Dependency not found
        }
        if (job_it->second->get_state() != dag_job_state::completed)
        {
            return false;  // Dependency not completed
        }
    }
 
    return true;
}

References kcenon::thread::completed.

Referenced by get_ready_jobs().

Here is the caller graph for this function:

cancel_all()

auto kcenon::thread::dag_scheduler::cancel_all

(

)

-> void

Cancels all pending jobs.

Jobs that are already running will complete, but no new jobs will start.

Thread Safety: Thread-safe

Definition at line 322 of file dag_scheduler.cpp.

{
    std::unique_lock lock(mutex_);
    cancelled_.store(true);
 
    for (auto& [id, job] : jobs_)
    {
        auto state = job->get_state();
        if (state == dag_job_state::pending || state == dag_job_state::ready)
        {
            job->set_state(dag_job_state::cancelled);
            if (config_.state_callback)
            {
                config_.state_callback(id, state, dag_job_state::cancelled);
            }
        }
    }
 
    completion_cv_.notify_all();
}

References kcenon::thread::cancelled, kcenon::thread::pending, and kcenon::thread::ready.

Referenced by ~dag_scheduler().

Here is the caller graph for this function:

cancel_dependents()

auto kcenon::thread::dag_scheduler::cancel_dependents ( job_id failed_id ) -> void

private

Cancels dependents due to dependency failure.

Parameters

failed_id

The failed job ID

Definition at line 1097 of file dag_scheduler.cpp.

{
    auto it = dependents_.find(failed_id);
    if (it == dependents_.end())
    {
        return;
    }
 
    std::queue<job_id> to_cancel;
    for (const auto& dep : it->second)
    {
        to_cancel.push(dep);
    }
 
    while (!to_cancel.empty())
    {
        auto current = to_cancel.front();
        to_cancel.pop();
 
        auto job_it = jobs_.find(current);
        if (job_it == jobs_.end())
        {
            continue;
        }
 
        auto old_state = job_it->second->get_state();
        if (old_state == dag_job_state::pending || old_state == dag_job_state::ready)
        {
            job_it->second->set_state(dag_job_state::cancelled);
            if (config_.state_callback)
            {
                config_.state_callback(current, old_state, dag_job_state::cancelled);
            }
 
            // Also cancel dependents of cancelled job
            auto dep_it = dependents_.find(current);
            if (dep_it != dependents_.end())
            {
                for (const auto& dep : dep_it->second)
                {
                    to_cancel.push(dep);
                }
            }
        }
    }
}

References kcenon::thread::cancelled, kcenon::thread::pending, and kcenon::thread::ready.

detect_cycle()

auto kcenon::thread::dag_scheduler::detect_cycle ( ) const -> bool

nodiscardprivate

Detects cycles using DFS.

Returns

true if cycle detected

Definition at line 982 of file dag_scheduler.cpp.

{
    enum class Color { White, Gray, Black };
 
    std::unordered_map<job_id, Color> color;
    for (const auto& [id, _] : jobs_)
    {
        color[id] = Color::White;
    }
 
    std::function<bool(job_id)> dfs = [&](job_id id) -> bool {
        color[id] = Color::Gray;
 
        auto it = dependencies_.find(id);
        if (it != dependencies_.end())
        {
            for (const auto& dep : it->second)
            {
                if (color[dep] == Color::Gray)
                {
                    return true;  // Back edge = cycle
                }
                if (color[dep] == Color::White && dfs(dep))
                {
                    return true;
                }
            }
        }
 
        color[id] = Color::Black;
        return false;
    };
 
    for (const auto& [id, _] : jobs_)
    {
        if (color[id] == Color::White && dfs(id))
        {
            return true;
        }
    }
 
    return false;
}

References dependencies_, and jobs_.

Referenced by has_cycles().

Here is the caller graph for this function:

execute()

auto kcenon::thread::dag_scheduler::execute ( job_id target ) -> std::future<common::VoidResult>

nodiscard

Executes a specific job and its dependencies.

Parameters

target

The target job to execute

Returns

Future that completes when the target and dependencies are done

Thread Safety: Thread-safe

Definition at line 230 of file dag_scheduler.cpp.

{
    auto promise = std::make_shared<std::promise<common::VoidResult>>();
    auto future = promise->get_future();
 
    std::thread([this, target, promise]() {
        {
            std::unique_lock lock(mutex_);
 
            auto it = jobs_.find(target);
            if (it == jobs_.end())
            {
                promise->set_value(make_error_result(error_code::job_invalid,
                                                     "Target job not found: " + std::to_string(target)));
                return;
            }
 
            if (executing_.exchange(true))
            {
                promise->set_value(make_error_result(error_code::thread_already_running,
                                                     "Execution already in progress"));
                return;
            }
 
            cancelled_.store(false);
            first_error_.reset();
            execution_start_time_ = std::chrono::steady_clock::now();
 
            // Get all dependencies of target (transitively)
            std::unordered_set<job_id> needed_jobs;
            std::stack<job_id> to_visit;
            to_visit.push(target);
 
            while (!to_visit.empty())
            {
                auto current = to_visit.top();
                to_visit.pop();
 
                if (needed_jobs.count(current) > 0)
                {
                    continue;
                }
                needed_jobs.insert(current);
 
                auto deps_it = dependencies_.find(current);
                if (deps_it != dependencies_.end())
                {
                    for (const auto& dep : deps_it->second)
                    {
                        to_visit.push(dep);
                    }
                }
            }
 
            // Mark needed jobs as ready if dependencies satisfied
            for (const auto& id : needed_jobs)
            {
                auto job_it = jobs_.find(id);
                if (job_it != jobs_.end() && job_it->second->get_state() == dag_job_state::pending)
                {
                    bool deps_in_needed = true;
                    auto deps_it = dependencies_.find(id);
                    if (deps_it != dependencies_.end())
                    {
                        for (const auto& dep : deps_it->second)
                        {
                            if (needed_jobs.count(dep) == 0)
                            {
                                deps_in_needed = false;
                                break;
                            }
                        }
                    }
 
                    if (deps_in_needed && are_dependencies_satisfied(id))
                    {
                        job_it->second->set_state(dag_job_state::ready);
                    }
                }
            }
        }
 
        schedule_ready_jobs();
 
        auto result = wait();
        executing_.store(false);
        promise->set_value(result);
    }).detach();
 
    return future;
}

References kcenon::thread::job_invalid, kcenon::thread::make_error_result(), kcenon::thread::pending, kcenon::thread::ready, and kcenon::thread::thread_already_running.

Here is the call graph for this function:

execute_all()

auto kcenon::thread::dag_scheduler::execute_all ( ) -> std::future<common::VoidResult>

nodiscard

Executes all jobs in dependency order.

Returns

Future that completes when all jobs are done

Thread Safety: Thread-safe

Definition at line 185 of file dag_scheduler.cpp.

{
    auto promise = std::make_shared<std::promise<common::VoidResult>>();
    auto future = promise->get_future();
 
    // Start execution in a new thread
    std::thread([this, promise]() {
        {
            std::unique_lock lock(mutex_);
            if (executing_.exchange(true))
            {
                promise->set_value(make_error_result(error_code::thread_already_running,
                                                     "Execution already in progress"));
                return;
            }
 
            cancelled_.store(false);
            first_error_.reset();
            execution_start_time_ = std::chrono::steady_clock::now();
 
            // Mark all jobs with satisfied dependencies as ready
            for (auto& [id, job] : jobs_)
            {
                if (job->get_state() == dag_job_state::pending)
                {
                    if (are_dependencies_satisfied(id))
                    {
                        job->set_state(dag_job_state::ready);
                    }
                }
            }
        }
 
        // Schedule ready jobs
        schedule_ready_jobs();
 
        // Wait for completion
        auto result = wait();
        executing_.store(false);
        promise->set_value(result);
    }).detach();
 
    return future;
}

References kcenon::thread::make_error_result(), kcenon::thread::pending, kcenon::thread::ready, and kcenon::thread::thread_already_running.

Here is the call graph for this function:

execute_job()

auto kcenon::thread::dag_scheduler::execute_job ( job_id id ) -> void

private

Executes a single job.

Parameters

id	The job ID to execute

Definition at line 868 of file dag_scheduler.cpp.

{
    std::unique_lock lock(mutex_);
 
    auto it = jobs_.find(id);
    if (it == jobs_.end())
    {
        return;
    }
 
    if (!it->second->try_transition_state(dag_job_state::ready, dag_job_state::running))
    {
        return;  // Already running or in terminal state
    }
 
    it->second->record_start_time();
    ++running_count_;
 
    if (config_.state_callback)
    {
        config_.state_callback(id, dag_job_state::ready, dag_job_state::running);
    }
 
    lock.unlock();
 
    // Create a callback job to execute the dag_job
    auto wrapper_job = std::make_unique<callback_job>(
        [this, id]() -> common::VoidResult {
            std::shared_lock slock(mutex_);
            auto job_it = jobs_.find(id);
            if (job_it == jobs_.end())
            {
                return make_error_result(error_code::job_invalid, "Job not found");
            }
 
            auto& dag_job_ptr = job_it->second;
            slock.unlock();
 
            auto result = dag_job_ptr->do_work();
 
            if (result.is_ok())
            {
                on_job_completed(id);
            }
            else
            {
                on_job_failed(id, result.error().message);
            }
 
            return result;
        },
        "dag_job_" + std::to_string(id)
    );
 
    pool_->enqueue(std::move(wrapper_job));
}

References kcenon::thread::result< T >::is_ok(), kcenon::thread::job_invalid, kcenon::thread::make_error_result(), kcenon::thread::ready, and kcenon::thread::running.

Here is the call graph for this function:

get_all_jobs()

auto kcenon::thread::dag_scheduler::get_all_jobs ( ) const -> std::vector<dag_job_info>

nodiscard

Gets information about all jobs.

Returns

Vector of job info for all jobs

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 405 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    std::vector<dag_job_info> result;
    result.reserve(jobs_.size());
 
    for (const auto& [id, job] : jobs_)
    {
        auto info = job->get_info();
        auto dep_it = dependents_.find(id);
        if (dep_it != dependents_.end())
        {
            info.dependents = dep_it->second;
        }
        result.push_back(std::move(info));
    }
 
    return result;
}

References dependents_, kcenon::thread::info, jobs_, and mutex_.

get_config()

auto kcenon::thread::dag_scheduler::get_config ( ) const -> const dag_config&

inlinenodiscard

Gets the configuration.

Returns

Current configuration

Definition at line 306 of file dag_scheduler.h.

{ return config_; }

References config_.

get_execution_order()

auto kcenon::thread::dag_scheduler::get_execution_order ( ) const -> std::vector<job_id>

nodiscard

Gets topological execution order.

Returns

Vector of job IDs in execution order

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 476 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
    return topological_sort();
}

References mutex_, and topological_sort().

Here is the call graph for this function:

get_job_info()

auto kcenon::thread::dag_scheduler::get_job_info ( job_id id ) const -> std::optional<dag_job_info>

nodiscard

Gets information about a specific job.

Parameters

id	The job ID

Returns

Job info, or std::nullopt if not found

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 383 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    auto it = jobs_.find(id);
    if (it == jobs_.end())
    {
        return std::nullopt;
    }
 
    auto info = it->second->get_info();
 
    // Add dependents from our tracking
    auto dep_it = dependents_.find(id);
    if (dep_it != dependents_.end())
    {
        info.dependents = dep_it->second;
    }
 
    return info;
}

References kcenon::thread::info.

get_jobs_in_state()

auto kcenon::thread::dag_scheduler::get_jobs_in_state ( dag_job_state state ) const -> std::vector<dag_job_info>

nodiscard

Gets jobs in a specific state.

Parameters

state

The state to filter by

Returns

Vector of job info for matching jobs

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 426 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    std::vector<dag_job_info> result;
 
    for (const auto& [id, job] : jobs_)
    {
        if (job->get_state() == state)
        {
            auto info = job->get_info();
            auto dep_it = dependents_.find(id);
            if (dep_it != dependents_.end())
            {
                info.dependents = dep_it->second;
            }
            result.push_back(std::move(info));
        }
    }
 
    return result;
}

References kcenon::thread::info.

get_ready_jobs()

auto kcenon::thread::dag_scheduler::get_ready_jobs ( ) const -> std::vector<job_id>

nodiscard

Gets IDs of ready jobs (dependencies satisfied)

Returns

Vector of job IDs that can be executed

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 449 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    std::vector<job_id> result;
 
    for (const auto& [id, job] : jobs_)
    {
        if (job->get_state() == dag_job_state::ready)
        {
            result.push_back(id);
        }
        else if (job->get_state() == dag_job_state::pending && are_dependencies_satisfied(id))
        {
            result.push_back(id);
        }
    }
 
    return result;
}

References are_dependencies_satisfied(), jobs_, mutex_, kcenon::thread::pending, and kcenon::thread::ready.

Here is the call graph for this function:

get_result()

template<typename T >

auto kcenon::thread::dag_scheduler::get_result ( job_id id ) const -> const T&

inlinenodiscard

Gets the result from a completed job.

Template Parameters

T	The expected result type

Parameters

id	The job ID

Returns

The result value

Exceptions

std::runtime_error	if job not found or not completed
std::bad_any_cast	if type doesn't match

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 255 of file dag_scheduler.h.

        {
            std::shared_lock lock(mutex_);
            auto it = jobs_.find(id);
            if (it == jobs_.end())
            {
                throw std::runtime_error("Job not found: " + std::to_string(id));
            }
            if (it->second->get_state() != dag_job_state::completed)
            {
                throw std::runtime_error("Job not completed: " + std::to_string(id));
            }
            return it->second->get_result<T>();
        }

References kcenon::thread::completed, jobs_, and mutex_.

get_state_color()

auto kcenon::thread::dag_scheduler::get_state_color ( dag_job_state state ) -> std::string

staticnodiscardprivate

Gets the state color for DOT visualization.

Parameters

state

The job state

Returns

Color string for DOT format

Definition at line 1144 of file dag_scheduler.cpp.

{
    switch (state)
    {
        case dag_job_state::pending:   return "#FFFFFF";  // White
        case dag_job_state::ready:     return "#87CEEB";  // Light blue
        case dag_job_state::running:   return "#FFFF00";  // Yellow
        case dag_job_state::completed: return "#90EE90";  // Light green
        case dag_job_state::failed:    return "#FF6B6B";  // Light red
        case dag_job_state::cancelled: return "#D3D3D3";  // Light gray
        case dag_job_state::skipped:   return "#FFA500";  // Orange
        default:                       return "#FFFFFF";
    }
}

References kcenon::thread::cancelled, kcenon::thread::completed, kcenon::thread::failed, kcenon::thread::pending, kcenon::thread::ready, kcenon::thread::running, and kcenon::thread::skipped.

Referenced by to_dot().

Here is the caller graph for this function:

get_stats()

auto kcenon::thread::dag_scheduler::get_stats ( ) const -> dag_stats

nodiscard

Gets execution statistics.

Returns

Current statistics

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 600 of file dag_scheduler.cpp.

{
    // Note: This is called from to_json which already holds the lock
    // For standalone calls, we need to be careful about deadlock
 
    dag_stats stats;
    stats.total_jobs = jobs_.size();
 
    for (const auto& [id, job] : jobs_)
    {
        switch (job->get_state())
        {
            case dag_job_state::completed: ++stats.completed_jobs; break;
            case dag_job_state::failed:    ++stats.failed_jobs;    break;
            case dag_job_state::pending:
            case dag_job_state::ready:     ++stats.pending_jobs;   break;
            case dag_job_state::running:   ++stats.running_jobs;   break;
            case dag_job_state::skipped:   ++stats.skipped_jobs;   break;
            case dag_job_state::cancelled: ++stats.cancelled_jobs; break;
        }
    }
 
    if (executing_.load())
    {
        stats.total_execution_time = std::chrono::duration_cast<std::chrono::milliseconds>(
            std::chrono::steady_clock::now() - execution_start_time_);
    }
 
    return stats;
}

References kcenon::thread::cancelled, kcenon::thread::dag_stats::cancelled_jobs, kcenon::thread::completed, kcenon::thread::dag_stats::completed_jobs, executing_, execution_start_time_, kcenon::thread::failed, kcenon::thread::dag_stats::failed_jobs, jobs_, kcenon::thread::pending, kcenon::thread::dag_stats::pending_jobs, kcenon::thread::ready, kcenon::thread::running, kcenon::thread::dag_stats::running_jobs, kcenon::thread::skipped, kcenon::thread::dag_stats::skipped_jobs, kcenon::thread::dag_stats::total_execution_time, and kcenon::thread::dag_stats::total_jobs.

Referenced by to_json().

Here is the caller graph for this function:

has_cycles()

auto kcenon::thread::dag_scheduler::has_cycles ( ) const -> bool

nodiscard

Checks if the DAG has cycles.

Returns

true if cycles are detected

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 470 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
    return detect_cycle();
}

References detect_cycle(), and mutex_.

Here is the call graph for this function:

is_execution_complete()

auto kcenon::thread::dag_scheduler::is_execution_complete ( ) const -> bool

nodiscardprivate

Checks if execution is complete.

Returns

true if all jobs are in terminal state

Definition at line 1159 of file dag_scheduler.cpp.

{
    for (const auto& [id, job] : jobs_)
    {
        auto state = job->get_state();
        if (state == dag_job_state::pending ||
            state == dag_job_state::ready ||
            state == dag_job_state::running)
        {
            return false;
        }
    }
    return true;
}

References jobs_, kcenon::thread::pending, kcenon::thread::ready, and kcenon::thread::running.

on_job_completed()

auto kcenon::thread::dag_scheduler::on_job_completed ( job_id id ) -> void

private

Called when a job completes successfully.

Parameters

id	The completed job ID

Definition at line 635 of file dag_scheduler.cpp.

{
    ++active_callbacks_;
    std::unique_lock lock(mutex_);
 
    auto it = jobs_.find(id);
    if (it == jobs_.end())
    {
        return;
    }
 
    auto old_state = it->second->get_state();
    it->second->set_state(dag_job_state::completed);
    it->second->record_end_time();
    --running_count_;
 
    if (config_.state_callback)
    {
        config_.state_callback(id, old_state, dag_job_state::completed);
    }
 
    if (config_.completion_callback)
    {
        config_.completion_callback(id);
    }
 
    // Check and update dependents
    auto dep_it = dependents_.find(id);
    if (dep_it != dependents_.end())
    {
        for (const auto& dependent : dep_it->second)
        {
            auto job_it = jobs_.find(dependent);
            if (job_it != jobs_.end() && job_it->second->get_state() == dag_job_state::pending)
            {
                if (are_dependencies_satisfied(dependent))
                {
                    job_it->second->set_state(dag_job_state::ready);
                    if (config_.state_callback)
                    {
                        config_.state_callback(dependent, dag_job_state::pending, dag_job_state::ready);
                    }
                }
            }
        }
    }
 
    lock.unlock();
 
    // Schedule more jobs
    schedule_ready_jobs();
 
    // Notify waiters
    completion_cv_.notify_all();
    --active_callbacks_;
}

References kcenon::thread::completed, kcenon::thread::pending, and kcenon::thread::ready.

on_job_failed()

auto kcenon::thread::dag_scheduler::on_job_failed ( job_id id, const std::string & error ) -> void

private

Called when a job fails.

Parameters

id	The failed job ID
error	The error message

Definition at line 692 of file dag_scheduler.cpp.

{
    ++active_callbacks_;
    std::unique_lock lock(mutex_);
 
    auto it = jobs_.find(id);
    if (it == jobs_.end())
    {
        return;
    }
 
    it->second->set_error_message(error);
    it->second->record_end_time();
 
    // Handle based on failure policy
    switch (config_.failure_policy)
    {
        case dag_failure_policy::retry:
        {
            auto& retry_count = retry_counts_[id];
            if (retry_count < config_.max_retries)
            {
                ++retry_count;
                // Reset state to ready for retry
                it->second->set_state(dag_job_state::ready);
                --running_count_;
 
                if (config_.state_callback)
                {
                    config_.state_callback(id, dag_job_state::running, dag_job_state::ready);
                }
 
                lock.unlock();
 
                // Wait before retry
                std::this_thread::sleep_for(config_.retry_delay);
 
                schedule_ready_jobs();
                return;
            }
            // Fall through to fail_fast if max retries exceeded
            [[fallthrough]];
        }
 
        case dag_failure_policy::fallback:
        {
            if (config_.failure_policy == dag_failure_policy::fallback && it->second->has_fallback())
            {
                // Try fallback
                auto fallback = it->second->get_fallback();
                auto result = fallback();
                if (result.is_ok())
                {
                    // Fallback succeeded
                    it->second->set_state(dag_job_state::completed);
                    --running_count_;
                    if (config_.state_callback)
                    {
                        config_.state_callback(id, dag_job_state::running, dag_job_state::completed);
                    }
                    lock.unlock();
                    schedule_ready_jobs();
                    completion_cv_.notify_all();
                    --active_callbacks_;
                    return;
                }
            }
            [[fallthrough]];
        }
 
        case dag_failure_policy::fail_fast:
        {
            auto old_state = it->second->get_state();
            it->second->set_state(dag_job_state::failed);
            --running_count_;
 
            if (!first_error_)
            {
                first_error_ = to_error_info(error_code::job_execution_failed, error);
            }
 
            if (config_.state_callback)
            {
                config_.state_callback(id, old_state, dag_job_state::failed);
            }
 
            if (config_.error_callback)
            {
                config_.error_callback(id, error);
            }
 
            // Cancel all dependents
            cancel_dependents(id);
            break;
        }
 
        case dag_failure_policy::continue_others:
        {
            auto old_state = it->second->get_state();
            it->second->set_state(dag_job_state::failed);
            --running_count_;
 
            if (!first_error_)
            {
                first_error_ = to_error_info(error_code::job_execution_failed, error);
            }
 
            if (config_.state_callback)
            {
                config_.state_callback(id, old_state, dag_job_state::failed);
            }
 
            if (config_.error_callback)
            {
                config_.error_callback(id, error);
            }
 
            // Skip dependents but continue others
            skip_dependents(id);
            break;
        }
    }
 
    lock.unlock();
    schedule_ready_jobs();
    completion_cv_.notify_all();
    --active_callbacks_;
}

References kcenon::thread::completed, kcenon::thread::continue_others, kcenon::thread::fail_fast, kcenon::thread::failed, kcenon::thread::fallback, kcenon::thread::result< T >::is_ok(), kcenon::thread::job_execution_failed, kcenon::thread::ready, kcenon::thread::retry, kcenon::thread::running, and kcenon::thread::to_error_info().

Here is the call graph for this function:

operator=() [1/2]

auto kcenon::thread::dag_scheduler::operator= ( const dag_scheduler & ) -> dag_scheduler &=delete

delete

operator=() [2/2]

auto kcenon::thread::dag_scheduler::operator= ( dag_scheduler && other ) -> dag_scheduler&

noexcept

Definition at line 51 of file dag_scheduler.cpp.

{
    if (this != &other)
    {
        pool_ = std::move(other.pool_);
        config_ = std::move(other.config_);
        jobs_ = std::move(other.jobs_);
        dependencies_ = std::move(other.dependencies_);
        dependents_ = std::move(other.dependents_);
        executing_.store(other.executing_.load());
        cancelled_.store(other.cancelled_.load());
        running_count_.store(other.running_count_.load());
        execution_start_time_ = other.execution_start_time_;
        first_error_ = std::move(other.first_error_);
        retry_counts_ = std::move(other.retry_counts_);
    }
    return *this;
}

remove_job()

auto kcenon::thread::dag_scheduler::remove_job ( job_id id ) -> common::VoidResult

nodiscard

Removes a job from the DAG (only if not yet started)

Parameters

id	The job ID to remove

Returns

Success or error if job is running or not found

Thread Safety: Thread-safe (acquires exclusive lock)

Definition at line 145 of file dag_scheduler.cpp.

{
    std::unique_lock lock(mutex_);
 
    auto it = jobs_.find(id);
    if (it == jobs_.end())
    {
        return make_error_result(error_code::job_invalid,
                                 "Job not found: " + std::to_string(id));
    }
 
    auto state = it->second->get_state();
    if (state == dag_job_state::running)
    {
        return make_error_result(error_code::job_invalid,
                                 "Cannot remove running job: " + std::to_string(id));
    }
 
    // Remove from dependency graphs
    dependencies_.erase(id);
    dependents_.erase(id);
 
    // Remove from other jobs' dependencies
    for (auto& [job_id, deps] : dependencies_)
    {
        deps.erase(std::remove(deps.begin(), deps.end(), id), deps.end());
    }
    for (auto& [job_id, deps] : dependents_)
    {
        deps.erase(std::remove(deps.begin(), deps.end(), id), deps.end());
    }
 
    jobs_.erase(it);
    return common::ok();
}

References kcenon::thread::job_invalid, kcenon::thread::make_error_result(), and kcenon::thread::running.

Here is the call graph for this function:

reset()

auto kcenon::thread::dag_scheduler::reset ( ) -> common::VoidResult

nodiscard

Resets the scheduler for reuse.

Clears all jobs and resets state. Cannot be called while jobs are running.

Thread Safety: Thread-safe (acquires exclusive lock)

Definition at line 359 of file dag_scheduler.cpp.

{
    std::unique_lock lock(mutex_);
 
    if (executing_.load())
    {
        return make_error_result(error_code::thread_already_running,
                                 "Cannot reset while execution is in progress");
    }
 
    jobs_.clear();
    dependencies_.clear();
    dependents_.clear();
    retry_counts_.clear();
    first_error_.reset();
    cancelled_.store(false);
 
    return common::ok();
}

References kcenon::thread::make_error_result(), and kcenon::thread::thread_already_running.

Here is the call graph for this function:

schedule_ready_jobs()

auto kcenon::thread::dag_scheduler::schedule_ready_jobs ( ) -> void

private

Schedules ready jobs for execution.

Definition at line 821 of file dag_scheduler.cpp.

{
    if (cancelled_.load())
    {
        return;
    }
 
    std::vector<job_id> ready_jobs;
 
    {
        std::shared_lock lock(mutex_);
        for (const auto& [id, job] : jobs_)
        {
            if (job->get_state() == dag_job_state::ready)
            {
                ready_jobs.push_back(id);
            }
        }
    }
 
    for (const auto& id : ready_jobs)
    {
        if (cancelled_.load())
        {
            break;
        }
 
        if (config_.execute_in_parallel)
        {
            execute_job(id);
        }
        else
        {
            // Execute sequentially
            execute_job(id);
            // Wait for completion before next job
            std::shared_lock lock(mutex_);
            completion_cv_.wait(lock, [this, id]() {
                auto it = jobs_.find(id);
                if (it == jobs_.end()) return true;
                auto state = it->second->get_state();
                return state != dag_job_state::running;
            });
        }
    }
}

References kcenon::thread::ready, and kcenon::thread::running.

skip_dependents()

auto kcenon::thread::dag_scheduler::skip_dependents ( job_id failed_id ) -> void

private

Marks dependents as skipped due to dependency failure.

Parameters

failed_id

The failed job ID

Definition at line 1050 of file dag_scheduler.cpp.

{
    auto it = dependents_.find(failed_id);
    if (it == dependents_.end())
    {
        return;
    }
 
    std::queue<job_id> to_skip;
    for (const auto& dep : it->second)
    {
        to_skip.push(dep);
    }
 
    while (!to_skip.empty())
    {
        auto current = to_skip.front();
        to_skip.pop();
 
        auto job_it = jobs_.find(current);
        if (job_it == jobs_.end())
        {
            continue;
        }
 
        auto old_state = job_it->second->get_state();
        if (old_state == dag_job_state::pending || old_state == dag_job_state::ready)
        {
            job_it->second->set_state(dag_job_state::skipped);
            if (config_.state_callback)
            {
                config_.state_callback(current, old_state, dag_job_state::skipped);
            }
 
            // Also skip dependents of skipped job
            auto dep_it = dependents_.find(current);
            if (dep_it != dependents_.end())
            {
                for (const auto& dep : dep_it->second)
                {
                    to_skip.push(dep);
                }
            }
        }
    }
}

References kcenon::thread::pending, kcenon::thread::ready, and kcenon::thread::skipped.

to_dot()

auto kcenon::thread::dag_scheduler::to_dot ( ) const -> std::string

nodiscard

Exports the DAG as DOT format (Graphviz)

Returns

DOT format string

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 486 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    std::ostringstream ss;
    ss << "digraph DAG {\n";
    ss << "    rankdir=TB;\n";
    ss << "    node [shape=box, style=filled];\n\n";
 
    // Nodes
    for (const auto& [id, job] : jobs_)
    {
        auto state = job->get_state();
        ss << "    " << id << " [label=\"" << job->get_name()
           << "\\n(" << dag_job_state_to_string(state) << ")\""
           << ", fillcolor=\"" << get_state_color(state) << "\"];\n";
    }
 
    ss << "\n";
 
    // Edges
    for (const auto& [dependent, deps] : dependencies_)
    {
        for (const auto& dependency : deps)
        {
            ss << "    " << dependency << " -> " << dependent << ";\n";
        }
    }
 
    ss << "}\n";
 
    return ss.str();
}

References kcenon::thread::dag_job_state_to_string(), dependencies_, kcenon::thread::job::get_name(), get_state_color(), jobs_, and mutex_.

Here is the call graph for this function:

to_json()

auto kcenon::thread::dag_scheduler::to_json ( ) const -> std::string

nodiscard

Exports the DAG as JSON format.

Returns

JSON format string

Thread Safety: Thread-safe (acquires shared lock)

Definition at line 520 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    std::ostringstream ss;
    ss << "{\n";
    ss << "  \"jobs\": [\n";
 
    bool first_job = true;
    for (const auto& [id, job] : jobs_)
    {
        if (!first_job) ss << ",\n";
        first_job = false;
 
        auto info = job->get_info();
        ss << "    {\n";
        ss << "      \"id\": " << id << ",\n";
        ss << "      \"name\": \"" << info.name << "\",\n";
        ss << "      \"state\": \"" << dag_job_state_to_string(info.state) << "\",\n";
        ss << "      \"dependencies\": [";
 
        bool first_dep = true;
        for (const auto& dep : info.dependencies)
        {
            if (!first_dep) ss << ", ";
            first_dep = false;
            ss << dep;
        }
        ss << "],\n";
 
        ss << "      \"dependents\": [";
        auto dep_it = dependents_.find(id);
        if (dep_it != dependents_.end())
        {
            first_dep = true;
            for (const auto& dep : dep_it->second)
            {
                if (!first_dep) ss << ", ";
                first_dep = false;
                ss << dep;
            }
        }
        ss << "],\n";
 
        auto wait_ms = info.get_wait_time().count();
        auto exec_ms = info.get_execution_time().count();
        ss << "      \"wait_time_ms\": " << wait_ms << ",\n";
        ss << "      \"execution_time_ms\": " << exec_ms;
 
        if (info.error_message)
        {
            ss << ",\n      \"error\": \"" << *info.error_message << "\"";
        }
 
        ss << "\n    }";
    }
 
    ss << "\n  ],\n";
 
    // Add statistics
    auto stats = get_stats();
    ss << "  \"stats\": {\n";
    ss << "    \"total_jobs\": " << stats.total_jobs << ",\n";
    ss << "    \"completed_jobs\": " << stats.completed_jobs << ",\n";
    ss << "    \"failed_jobs\": " << stats.failed_jobs << ",\n";
    ss << "    \"pending_jobs\": " << stats.pending_jobs << ",\n";
    ss << "    \"running_jobs\": " << stats.running_jobs << ",\n";
    ss << "    \"skipped_jobs\": " << stats.skipped_jobs << ",\n";
    ss << "    \"cancelled_jobs\": " << stats.cancelled_jobs << "\n";
    ss << "  }\n";
 
    ss << "}\n";
 
    return ss.str();
}

References kcenon::thread::dag_job_state_to_string(), dependents_, get_stats(), kcenon::thread::info, jobs_, and mutex_.

Here is the call graph for this function:

topological_sort()

auto kcenon::thread::dag_scheduler::topological_sort ( ) const -> std::vector<job_id>

nodiscardprivate

Performs topological sort.

Returns

Sorted job IDs, or empty if cycle detected

Definition at line 925 of file dag_scheduler.cpp.

{
    std::vector<job_id> result;
    std::unordered_map<job_id, int> in_degree;
    std::queue<job_id> ready_queue;
 
    // Initialize in-degrees
    for (const auto& [id, _] : jobs_)
    {
        in_degree[id] = 0;
    }
 
    for (const auto& [id, deps] : dependencies_)
    {
        in_degree[id] = static_cast<int>(deps.size());
    }
 
    // Find initial ready jobs (in-degree 0)
    for (const auto& [id, degree] : in_degree)
    {
        if (degree == 0)
        {
            ready_queue.push(id);
        }
    }
 
    // Process queue
    while (!ready_queue.empty())
    {
        auto current = ready_queue.front();
        ready_queue.pop();
        result.push_back(current);
 
        // Reduce in-degree of dependents
        auto it = dependents_.find(current);
        if (it != dependents_.end())
        {
            for (const auto& dependent : it->second)
            {
                --in_degree[dependent];
                if (in_degree[dependent] == 0)
                {
                    ready_queue.push(dependent);
                }
            }
        }
    }
 
    // If result doesn't contain all jobs, there's a cycle
    if (result.size() != jobs_.size())
    {
        return {};  // Cycle detected
    }
 
    return result;
}

References dependencies_, dependents_, and jobs_.

Referenced by get_execution_order().

Here is the caller graph for this function:

wait()

auto kcenon::thread::dag_scheduler::wait ( ) -> common::VoidResult

nodiscard

Waits for all jobs to complete.

Returns

Success or the first error encountered

Thread Safety: Thread-safe

Definition at line 343 of file dag_scheduler.cpp.

{
    std::shared_lock lock(mutex_);
 
    completion_cv_.wait(lock, [this]() {
        return is_execution_complete() || cancelled_.load();
    });
 
    if (first_error_)
    {
        return common::VoidResult(*first_error_);
    }
 
    return common::ok();
}

Member Data Documentation

active_callbacks_

std::atomic<std::size_t> kcenon::thread::dag_scheduler::active_callbacks_ {0}

private

Number of active callbacks (for safe destruction)

Tracks callbacks that may access members after releasing the mutex. The destructor waits for this to reach 0 before destroying members.

Definition at line 365 of file dag_scheduler.h.

{0};

Referenced by ~dag_scheduler().

cancelled_

std::atomic<bool> kcenon::thread::dag_scheduler::cancelled_ {false}

private

Flag indicating cancellation was requested.

Definition at line 352 of file dag_scheduler.h.

{false};

completion_cv_

std::condition_variable_any kcenon::thread::dag_scheduler::completion_cv_

private

Condition variable for waiting on completion.

Definition at line 342 of file dag_scheduler.h.

config_

dag_config kcenon::thread::dag_scheduler::config_

private

Configuration.

Definition at line 317 of file dag_scheduler.h.

Referenced by get_config().

dependencies_

std::unordered_map<job_id, std::vector<job_id> > kcenon::thread::dag_scheduler::dependencies_

private

Dependency graph (job -> jobs it depends on)

Definition at line 327 of file dag_scheduler.h.

Referenced by detect_cycle(), to_dot(), and topological_sort().

dependents_

std::unordered_map<job_id, std::vector<job_id> > kcenon::thread::dag_scheduler::dependents_

private

Reverse dependency graph (job -> jobs that depend on it)

Definition at line 332 of file dag_scheduler.h.

Referenced by get_all_jobs(), to_json(), and topological_sort().

executing_

std::atomic<bool> kcenon::thread::dag_scheduler::executing_ {false}

private

Flag indicating execution is in progress.

Definition at line 347 of file dag_scheduler.h.

{false};

Referenced by get_stats().

execution_start_time_

std::chrono::steady_clock::time_point kcenon::thread::dag_scheduler::execution_start_time_

private

Execution start time.

Definition at line 370 of file dag_scheduler.h.

Referenced by get_stats().

first_error_

std::optional<common::error_info> kcenon::thread::dag_scheduler::first_error_

private

First error encountered during execution.

Definition at line 375 of file dag_scheduler.h.

jobs_

std::unordered_map<job_id, std::unique_ptr<dag_job> > kcenon::thread::dag_scheduler::jobs_

private

Job storage (job_id -> dag_job)

Definition at line 322 of file dag_scheduler.h.

Referenced by detect_cycle(), get_all_jobs(), get_ready_jobs(), get_result(), get_stats(), is_execution_complete(), to_dot(), to_json(), and topological_sort().

mutex_

std::shared_mutex kcenon::thread::dag_scheduler::mutex_

mutableprivate

Mutex for thread-safe access.

Definition at line 337 of file dag_scheduler.h.

Referenced by get_all_jobs(), get_execution_order(), get_ready_jobs(), get_result(), has_cycles(), to_dot(), and to_json().

pool_

std::shared_ptr<thread_pool> kcenon::thread::dag_scheduler::pool_

private

Thread pool for job execution.

Definition at line 312 of file dag_scheduler.h.

retry_counts_

std::unordered_map<job_id, std::size_t> kcenon::thread::dag_scheduler::retry_counts_

private

Retry count per job.

Definition at line 380 of file dag_scheduler.h.

running_count_

std::atomic<std::size_t> kcenon::thread::dag_scheduler::running_count_ {0}

private

Number of jobs currently running.

Definition at line 357 of file dag_scheduler.h.

{0};

---

The documentation for this class was generated from the following files:

• include/kcenon/thread/dag/dag_scheduler.h
• src/dag/dag_scheduler.cpp