kcenon::pacs::workflow::task_scheduler Class Reference

#include <task_scheduler.h>

Collaboration diagram for kcenon::pacs::workflow::task_scheduler:

Public Member Functions
	task_scheduler (storage::index_database &database, const task_scheduler_config &config={})
	Construct task scheduler.
	task_scheduler (storage::index_database &database, storage::file_storage &file_storage, std::shared_ptr< kcenon::thread::thread_pool > thread_pool, const task_scheduler_config &config={})
	Construct task scheduler with storage and thread pool.
	task_scheduler (storage::index_database &database, storage::file_storage &file_storage, std::shared_ptr< kcenon::common::interfaces::IExecutor > executor, const task_scheduler_config &config={})
	Construct task scheduler with IExecutor (recommended)
	~task_scheduler ()
	Destructor - ensures graceful shutdown.
	task_scheduler (const task_scheduler &)=delete
	Non-copyable.
task_scheduler &	operator= (const task_scheduler &)=delete
	task_scheduler (task_scheduler &&)=delete
	Non-movable.
task_scheduler &	operator= (task_scheduler &&)=delete
void	start ()
	Start the scheduler service.
void	stop (bool wait_for_completion=true)
	Stop the scheduler service.
auto	is_running () const noexcept -> bool
	Check if the scheduler is running.
auto	schedule_cleanup (const cleanup_config &config) -> task_id
	Schedule cleanup task.
auto	schedule_archive (const archive_config &config) -> task_id
	Schedule archive task.
auto	schedule_verification (const verification_config &config) -> task_id
	Schedule verification task.
auto	schedule (const std::string &name, const std::string &description, std::chrono::seconds interval, task_callback_with_result callback) -> task_id
	Schedule a custom task with interval.
auto	schedule (const std::string &name, const std::string &description, const cron_schedule &cron_expr, task_callback_with_result callback) -> task_id
	Schedule a custom task with cron schedule.
auto	schedule_once (const std::string &name, const std::string &description, std::chrono::system_clock::time_point execute_at, task_callback_with_result callback) -> task_id
	Schedule a one-time task.
auto	schedule (scheduled_task task) -> task_id
	Schedule a task with full definition.
auto	list_tasks () const -> std::vector< scheduled_task >
	List all scheduled tasks.
auto	list_tasks (task_type type) const -> std::vector< scheduled_task >
	List tasks by type.
auto	list_tasks (task_state state) const -> std::vector< scheduled_task >
	List tasks by state.
auto	get_task (const task_id &id) const -> std::optional< scheduled_task >
	Get a specific task by ID.
auto	cancel_task (const task_id &id) -> bool
	Cancel a scheduled task.
auto	pause_task (const task_id &id) -> bool
	Pause a scheduled task.
auto	resume_task (const task_id &id) -> bool
	Resume a paused task.
auto	trigger_task (const task_id &id) -> bool
	Trigger immediate execution of a task.
auto	update_schedule (const task_id &id, const kcenon::pacs::workflow::schedule &new_schedule) -> bool
	Update task schedule.
auto	get_execution_history (const task_id &id, std::size_t limit=100) const -> std::vector< task_execution_record >
	Get execution history for a task.
auto	get_recent_executions (std::size_t limit=100) const -> std::vector< task_execution_record >
	Get all recent executions.
void	clear_history (const task_id &id, std::size_t keep_last=0)
	Clear execution history for a task.
auto	get_stats () const -> scheduler_stats
	Get scheduler statistics.
auto	pending_count () const noexcept -> std::size_t
	Get number of pending tasks.
auto	running_count () const noexcept -> std::size_t
	Get number of running tasks.
auto	save_tasks () const -> bool
	Save all tasks to persistence storage.
auto	load_tasks () -> std::size_t
	Load tasks from persistence storage.
void	set_task_complete_callback (task_scheduler_config::task_complete_callback callback)
	Set the task complete callback.
void	set_error_callback (task_scheduler_config::task_error_callback callback)
	Set the error callback.

Private Member Functions
void	run_loop ()
	Background thread main loop.
void	execute_cycle ()
	Execute a single scheduler cycle.
auto	execute_task (scheduled_task &task) -> task_execution_record
	Execute a single task.
auto	calculate_next_run (const kcenon::pacs::workflow::schedule &sched, std::chrono::system_clock::time_point from=std::chrono::system_clock::now()) const -> std::optional< std::chrono::system_clock::time_point >
	Calculate next run time for a schedule.
auto	calculate_next_cron_run (const cron_schedule &cron, std::chrono::system_clock::time_point from) const -> std::optional< std::chrono::system_clock::time_point >
	Calculate next run time for cron schedule.
auto	generate_task_id () const -> task_id
	Generate unique task ID.
auto	generate_execution_id () const -> std::string
	Generate unique execution ID.
auto	create_cleanup_callback (const cleanup_config &config) -> task_callback_with_result
	Create cleanup task callback.
auto	create_archive_callback (const archive_config &config) -> task_callback_with_result
	Create archive task callback.
auto	create_verification_callback (const verification_config &config) -> task_callback_with_result
	Create verification task callback.
void	record_execution (const task_id &task_id, const task_execution_record &record)
	Record task execution.
void	update_stats (const task_execution_record &record)
	Update statistics after execution.
auto	serialize_tasks () const -> std::string
	Serialize tasks to JSON.
auto	deserialize_tasks (const std::string &json) -> std::size_t
	Deserialize tasks from JSON.

Private Attributes
storage::index_database &	database_
	Reference to PACS index database.
storage::file_storage *	file_storage_ {nullptr}
	Optional reference to file storage.
std::shared_ptr< kcenon::thread::thread_pool >	thread_pool_
	Thread pool for parallel task execution (legacy)
std::shared_ptr< kcenon::common::interfaces::IExecutor >	executor_
	IExecutor for task execution (recommended, Issue #487)
task_scheduler_config	config_
	Service configuration.
std::thread	scheduler_thread_
	Background scheduler thread.
std::mutex	mutex_
	Mutex for thread synchronization.
std::condition_variable	cv_
	Condition variable for sleep/wake.
std::atomic< bool >	stop_requested_ {false}
	Flag to signal shutdown.
std::atomic< bool >	running_ {false}
	Flag indicating scheduler is running.
std::map< task_id, scheduled_task >	tasks_
	All scheduled tasks (id -> task)
std::mutex	tasks_mutex_
	Mutex for tasks map.
std::map< task_id, std::vector< task_execution_record > >	execution_history_
	Execution history (task_id -> records)
std::mutex	history_mutex_
	Mutex for execution history.
std::atomic< std::size_t >	running_count_ {0}
	Running task count.
scheduler_stats	stats_
	Scheduler statistics.
std::mutex	stats_mutex_
	Mutex for statistics.
std::chrono::steady_clock::time_point	start_time_
	Start time for uptime calculation.
std::atomic< uint64_t >	next_task_id_ {1}
	Next task ID counter.
std::atomic< uint64_t >	next_execution_id_ {1}
	Next execution ID counter.

Detailed Description

Definition at line 162 of file task_scheduler.h.

Constructor & Destructor Documentation

task_scheduler() [1/5]

kcenon::pacs::workflow::task_scheduler::task_scheduler ( storage::index_database & database, const task_scheduler_config & config = {} )

explicit

Construct task scheduler.

Parameters

database	Reference to the PACS index database
config	Service configuration

Definition at line 68 of file task_scheduler.cpp.

    : database_(database)
    , config_(config) {
    // Schedule built-in tasks from config
    if (config_.cleanup) {
        schedule_cleanup(*config_.cleanup);
    }
    if (config_.archive) {
        schedule_archive(*config_.archive);
    }
    if (config_.verification) {
        schedule_verification(*config_.verification);
    }
 
    if (config_.auto_start && config_.enabled) {
        start();
    }
}

References kcenon::pacs::workflow::task_scheduler_config::archive, kcenon::pacs::workflow::task_scheduler_config::auto_start, kcenon::pacs::workflow::task_scheduler_config::cleanup, config_, kcenon::pacs::workflow::task_scheduler_config::enabled, if(), and kcenon::pacs::workflow::task_scheduler_config::verification.

Here is the call graph for this function:

task_scheduler() [2/5]

kcenon::pacs::workflow::task_scheduler::task_scheduler	(	storage::index_database &	database,
		storage::file_storage &	file_storage,
		std::shared_ptr< kcenon::thread::thread_pool >	thread_pool,
		const task_scheduler_config &	config = {} )

Construct task scheduler with storage and thread pool.

Parameters

database	Reference to the PACS index database
file_storage	Reference to file storage for cleanup/archive
thread_pool	Thread pool for parallel task execution
config	Service configuration

Definition at line 89 of file task_scheduler.cpp.

    : database_(database)
    , file_storage_(&file_storage)
    , thread_pool_(std::move(thread_pool))
    , config_(config) {
 
    // Load persisted tasks if configured
    if (config_.restore_on_startup && !config_.persistence_path.empty()) {
        load_tasks();
    }
 
    // Schedule built-in tasks from config
    if (config_.cleanup) {
        schedule_cleanup(*config_.cleanup);
    }
    if (config_.archive) {
        schedule_archive(*config_.archive);
    }
    if (config_.verification) {
        schedule_verification(*config_.verification);
    }
 
    if (config_.auto_start && config_.enabled) {
        start();
    }
}

References kcenon::pacs::workflow::task_scheduler_config::archive, kcenon::pacs::workflow::task_scheduler_config::auto_start, kcenon::pacs::workflow::task_scheduler_config::cleanup, config_, kcenon::pacs::workflow::task_scheduler_config::enabled, if(), kcenon::pacs::workflow::task_scheduler_config::persistence_path, kcenon::pacs::workflow::task_scheduler_config::restore_on_startup, and kcenon::pacs::workflow::task_scheduler_config::verification.

Here is the call graph for this function:

task_scheduler() [3/5]

kcenon::pacs::workflow::task_scheduler::task_scheduler	(	storage::index_database &	database,
		storage::file_storage &	file_storage,
		std::shared_ptr< kcenon::common::interfaces::IExecutor >	executor,
		const task_scheduler_config &	config = {} )

Construct task scheduler with IExecutor (recommended)

This constructor accepts the standardized IExecutor interface from common_system, enabling better testability and decoupling from specific thread pool implementations.

Parameters

database	Reference to the PACS index database
file_storage	Reference to file storage for cleanup/archive
executor	IExecutor for task execution (from common_system)
config	Service configuration

See also

Issue #487 - IExecutor integration

Definition at line 120 of file task_scheduler.cpp.

    : database_(database)
    , file_storage_(&file_storage)
    , executor_(std::move(executor))
    , config_(config) {
 
    // Load persisted tasks if configured
    if (config_.restore_on_startup && !config_.persistence_path.empty()) {
        load_tasks();
    }
 
    // Schedule built-in tasks from config
    if (config_.cleanup) {
        schedule_cleanup(*config_.cleanup);
    }
    if (config_.archive) {
        schedule_archive(*config_.archive);
    }
    if (config_.verification) {
        schedule_verification(*config_.verification);
    }
 
    if (config_.auto_start && config_.enabled) {
        start();
    }
}

References kcenon::pacs::workflow::task_scheduler_config::archive, kcenon::pacs::workflow::task_scheduler_config::auto_start, kcenon::pacs::workflow::task_scheduler_config::cleanup, config_, kcenon::pacs::workflow::task_scheduler_config::enabled, if(), kcenon::pacs::workflow::task_scheduler_config::persistence_path, kcenon::pacs::workflow::task_scheduler_config::restore_on_startup, and kcenon::pacs::workflow::task_scheduler_config::verification.

Here is the call graph for this function:

~task_scheduler()

kcenon::pacs::workflow::task_scheduler::~task_scheduler

(

)

Destructor - ensures graceful shutdown.

Definition at line 151 of file task_scheduler.cpp.

                                {
    stop(true);
 
    // Persist tasks before destruction
    if (!config_.persistence_path.empty()) {
        save_tasks();
    }
}

References config_, kcenon::pacs::workflow::task_scheduler_config::persistence_path, save_tasks(), and stop().

Here is the call graph for this function:

task_scheduler() [4/5]

kcenon::pacs::workflow::task_scheduler::task_scheduler ( const task_scheduler & )

delete

Non-copyable.

task_scheduler() [5/5]

kcenon::pacs::workflow::task_scheduler::task_scheduler ( task_scheduler && )

delete

Non-movable.

Member Function Documentation

calculate_next_cron_run()

auto kcenon::pacs::workflow::task_scheduler::calculate_next_cron_run ( const cron_schedule & cron, std::chrono::system_clock::time_point from ) const -> std::optional<std::chrono::system_clock::time_point>

nodiscardprivate

Calculate next run time for cron schedule.

Parameters

cron	Cron schedule
from	Starting time point

Returns

Next run time

Definition at line 1022 of file task_scheduler.cpp.

                                                    {
 
    // Simple cron calculation - advance minute by minute until match
    // This is a simplified implementation; production would use more
    // efficient algorithms
 
    auto time = from + std::chrono::minutes{1};  // Start from next minute
    auto end_time = from + std::chrono::hours{24 * 365};  // Max 1 year ahead
 
    // Helper to parse cron field
    auto matches_field = [](const std::string& field, int value, [[maybe_unused]] int max) -> bool {
        if (field == "*") return true;
 
        // Check for step values (*/n)
        if (field.starts_with("*/")) {
            int step = std::stoi(field.substr(2));
            return (value % step) == 0;
        }
 
        // Check for range (n-m)
        auto dash_pos = field.find('-');
        if (dash_pos != std::string::npos) {
            int start = std::stoi(field.substr(0, dash_pos));
            int end = std::stoi(field.substr(dash_pos + 1));
            return value >= start && value <= end;
        }
 
        // Check for list (n,m,o)
        if (field.find(',') != std::string::npos) {
            std::istringstream iss(field);
            std::string item;
            while (std::getline(iss, item, ',')) {
                if (std::stoi(item) == value) {
                    return true;
                }
            }
            return false;
        }
 
        // Single value
        return std::stoi(field) == value;
    };
 
    while (time < end_time) {
        auto time_t = std::chrono::system_clock::to_time_t(time);
        std::tm tm = *std::localtime(&time_t);
 
        bool matches =
            matches_field(cron.minute, tm.tm_min, 59) &&
            matches_field(cron.hour, tm.tm_hour, 23) &&
            matches_field(cron.day_of_month, tm.tm_mday, 31) &&
            matches_field(cron.month, tm.tm_mon + 1, 12) &&
            matches_field(cron.day_of_week, tm.tm_wday, 6);
 
        if (matches) {
            // Round to start of minute
            auto duration = time.time_since_epoch();
            auto minutes = std::chrono::duration_cast<std::chrono::minutes>(duration);
            return std::chrono::system_clock::time_point(minutes);
        }
 
        time += std::chrono::minutes{1};
    }
 
    return std::nullopt;
}

calculate_next_run()

auto kcenon::pacs::workflow::task_scheduler::calculate_next_run ( const kcenon::pacs::workflow::schedule & sched, std::chrono::system_clock::time_point from = std::chrono::system_clock::now() ) const -> std::optional<std::chrono::system_clock::time_point>

nodiscardprivate

Calculate next run time for a schedule.

Parameters

sched	Schedule definition
from	Starting time point

Returns

Next run time

Definition at line 995 of file task_scheduler.cpp.

                                                    {
 
    return std::visit([this, from](const auto& s) ->
        std::optional<std::chrono::system_clock::time_point> {
        using T = std::decay_t<decltype(s)>;
 
        if constexpr (std::is_same_v<T, interval_schedule>) {
            if (s.start_at && *s.start_at > from) {
                return s.start_at;
            }
            return from + s.interval;
        } else if constexpr (std::is_same_v<T, cron_schedule>) {
            return calculate_next_cron_run(s, from);
        } else if constexpr (std::is_same_v<T, one_time_schedule>) {
            if (s.execute_at > from) {
                return s.execute_at;
            }
            return std::nullopt;  // Already passed
        }
 
        return std::nullopt;
    }, sched);
}

Referenced by execute_cycle().

Here is the caller graph for this function:

cancel_task()

auto kcenon::pacs::workflow::task_scheduler::cancel_task

(

const task_id &

id

)

-> bool

Cancel a scheduled task.

Removes the task from the schedule. If the task is currently running, it will complete but not be rescheduled.

Parameters

id	Task ID to cancel

Returns

true if task was found and cancelled

Definition at line 428 of file task_scheduler.cpp.

                                                          {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    auto it = tasks_.find(id);
    if (it == tasks_.end()) {
        return false;
    }
 
    it->second.state = task_state::cancelled;
    it->second.enabled = false;
    it->second.updated_at = std::chrono::system_clock::now();
 
    integration::logger_adapter::info(
        "Cancelled scheduled task task_id={} name={}",
        id, it->second.name);
 
    return true;
}

References kcenon::pacs::workflow::cancelled, and kcenon::pacs::integration::logger_adapter::info().

Here is the call graph for this function:

clear_history()

void kcenon::pacs::workflow::task_scheduler::clear_history	(	const task_id &	id,
		std::size_t	keep_last = 0 )

Clear execution history for a task.

Parameters

id	Task ID
keep_last	Number of records to keep

Definition at line 585 of file task_scheduler.cpp.

                                                                         {
    std::lock_guard<std::mutex> lock(history_mutex_);
 
    auto it = execution_history_.find(id);
    if (it == execution_history_.end()) {
        return;
    }
 
    auto& records = it->second;
    if (records.size() <= keep_last) {
        return;
    }
 
    // Keep only the last N records
    records.erase(
        records.begin(),
        records.begin() + static_cast<std::ptrdiff_t>(records.size() - keep_last));
}

References execution_history_, and history_mutex_.

create_archive_callback()

auto kcenon::pacs::workflow::task_scheduler::create_archive_callback ( const archive_config & config ) -> task_callback_with_result

nodiscardprivate

Create archive task callback.

Parameters

config

Archive configuration

Returns

Task callback

Definition at line 1224 of file task_scheduler.cpp.

                                 {
    return [this, config]() -> std::optional<std::string> {
        integration::logger_adapter::info(
            "Running archive task archive_after_days={} destination={}",
            config.archive_after.count(),
            config.destination);
 
        try {
            // Calculate cutoff date
            auto now = std::chrono::system_clock::now();
            auto cutoff = now - config.archive_after;
 
            // Convert to YYYYMMDD format
            auto cutoff_time_t = std::chrono::system_clock::to_time_t(cutoff);
            std::tm tm = *std::localtime(&cutoff_time_t);
            std::ostringstream date_oss;
            date_oss << std::put_time(&tm, "%Y%m%d");
            std::string cutoff_date = date_oss.str();
 
            // Query for studies older than cutoff
            storage::study_query query;
            query.study_date_to = cutoff_date;
            query.limit = config.max_archives_per_cycle;
 
            auto studies_result = database_.search_studies(query);
            if (!studies_result.is_ok()) {
                integration::logger_adapter::error(
                    "Failed to query studies error={}",
                    studies_result.error().message);
                return "Archive failed: " + studies_result.error().message;
            }
 
            std::size_t archived_count = 0;
            std::size_t failed_count = 0;
 
            // Create destination directory if needed
            std::filesystem::path dest_path(config.destination);
            if (!std::filesystem::exists(dest_path)) {
                std::filesystem::create_directories(dest_path);
            }
 
            for (const auto& study : studies_result.value()) {
                // Get all files for this study
                auto file_paths_result = database_.get_study_files(study.study_uid);
                if (!file_paths_result.is_ok() || file_paths_result.value().empty()) {
                    continue;
                }
                const auto& file_paths = file_paths_result.value();
 
                // Create study archive directory
                std::filesystem::path study_dest = dest_path / study.study_uid;
                if (!std::filesystem::exists(study_dest)) {
                    std::filesystem::create_directories(study_dest);
                }
 
                bool archive_success = true;
 
                for (const auto& src_file : file_paths) {
                    std::filesystem::path src_path(src_file);
                    if (!std::filesystem::exists(src_path)) {
                        integration::logger_adapter::warn(
                            "Source file not found file_path={}", src_file);
                        continue;
                    }
 
                    // Determine destination filename
                    std::filesystem::path dest_file = study_dest / src_path.filename();
 
                    try {
                        // Copy file to archive location
                        std::filesystem::copy_file(
                            src_path, dest_file,
                            std::filesystem::copy_options::overwrite_existing);
 
                        // Verify copy if configured
                        if (config.verify_after_archive) {
                            auto src_size = std::filesystem::file_size(src_path);
                            auto dest_size = std::filesystem::file_size(dest_file);
                            if (src_size != dest_size) {
                                integration::logger_adapter::error(
                                    "Archive verification failed: size mismatch "
                                    "src={} dest={}", src_file, dest_file.string());
                                archive_success = false;
                                break;
                            }
                        }
                    } catch (const std::filesystem::filesystem_error& e) {
                        integration::logger_adapter::error(
                            "Failed to archive file src={} dest={} error={}",
                            src_file, dest_file.string(), e.what());
                        archive_success = false;
                        break;
                    }
                }
 
                if (archive_success) {
                    ++archived_count;
 
                    // Delete originals if configured
                    if (config.delete_after_archive && file_storage_ != nullptr) {
                        for (const auto& file_path : file_paths) {
                            std::filesystem::path p(file_path);
                            std::string sop_uid = p.stem().string();
                            (void)file_storage_->remove(sop_uid);
                        }
                        (void)database_.delete_study(study.study_uid);
                    }
 
                    integration::logger_adapter::debug(
                        "Archived study study_uid={} files={}",
                        study.study_uid, file_paths.size());
                } else {
                    ++failed_count;
                }
            }
 
            integration::logger_adapter::info(
                "Archive task completed archived={} failed={}",
                archived_count, failed_count);
 
            if (failed_count > 0) {
                return "Archive completed with " + std::to_string(failed_count) + " failures";
            }
 
            return std::nullopt;  // Success
        } catch (const std::exception& e) {
            return std::string("Archive failed: ") + e.what();
        }
    };
}

References kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::integration::logger_adapter::info(), and kcenon::pacs::integration::logger_adapter::warn().

Here is the call graph for this function:

create_cleanup_callback()

auto kcenon::pacs::workflow::task_scheduler::create_cleanup_callback ( const cleanup_config & config ) -> task_callback_with_result

nodiscardprivate

Create cleanup task callback.

Parameters

config

Cleanup configuration

Returns

Task callback

Definition at line 1108 of file task_scheduler.cpp.

                                 {
    return [this, config]() -> std::optional<std::string> {
        integration::logger_adapter::info(
            "Running cleanup task retention_days={}",
            config.default_retention.count());
 
        try {
            // Calculate cutoff date
            auto now = std::chrono::system_clock::now();
            auto cutoff = now - config.default_retention;
 
            // Convert to YYYYMMDD format for database query
            auto cutoff_time_t = std::chrono::system_clock::to_time_t(cutoff);
            std::tm tm = *std::localtime(&cutoff_time_t);
            std::ostringstream date_oss;
            date_oss << std::put_time(&tm, "%Y%m%d");
            std::string cutoff_date = date_oss.str();
 
            // Query for studies older than cutoff
            storage::study_query query;
            query.study_date_to = cutoff_date;
            query.limit = config.max_deletions_per_cycle;
 
            auto studies_result = database_.search_studies(query);
            if (!studies_result.is_ok()) {
                integration::logger_adapter::error(
                    "Failed to query studies error={}",
                    studies_result.error().message);
                return "Cleanup failed: " + studies_result.error().message;
            }
 
            std::size_t deleted_count = 0;
            std::size_t skipped_count = 0;
 
            for (const auto& study : studies_result.value()) {
                // Check modality-specific retention
                auto modality_retention = config.retention_for(study.modalities_in_study);
                auto modality_cutoff = now - modality_retention;
                auto modality_cutoff_t = std::chrono::system_clock::to_time_t(modality_cutoff);
                std::tm mod_tm = *std::localtime(&modality_cutoff_t);
                std::ostringstream mod_oss;
                mod_oss << std::put_time(&mod_tm, "%Y%m%d");
                std::string modality_cutoff_date = mod_oss.str();
 
                // Skip if study is newer than modality-specific retention
                if (study.study_date > modality_cutoff_date) {
                    ++skipped_count;
                    continue;
                }
 
                // Check exclusion patterns
                bool excluded = false;
                for (const auto& pattern : config.exclude_patterns) {
                    if (study.study_description.find(pattern) != std::string::npos) {
                        excluded = true;
                        break;
                    }
                }
                if (excluded) {
                    ++skipped_count;
                    continue;
                }
 
                if (config.dry_run) {
                    integration::logger_adapter::info(
                        "Dry-run: would delete study study_uid={} study_date={} modality={}",
                        study.study_uid, study.study_date, study.modalities_in_study);
                    ++deleted_count;
                    continue;
                }
 
                // Delete files from storage if not database_only
                if (!config.database_only && file_storage_ != nullptr) {
                    auto file_paths_result = database_.get_study_files(study.study_uid);
                    if (file_paths_result.is_ok()) {
                        for (const auto& file_path : file_paths_result.value()) {
                            // Extract SOP UID from file path for removal
                            std::filesystem::path p(file_path);
                            std::string sop_uid = p.stem().string();
                            auto remove_result = file_storage_->remove(sop_uid);
                            if (remove_result.is_err()) {
                                integration::logger_adapter::warn(
                                    "Failed to remove file file_path={} error={}",
                                    file_path, remove_result.error().message);
                            }
                        }
                    }
                }
 
                // Delete from database
                auto delete_result = database_.delete_study(study.study_uid);
                if (delete_result.is_err()) {
                    integration::logger_adapter::error(
                        "Failed to delete study study_uid={} error={}",
                        study.study_uid, delete_result.error().message);
                    continue;
                }
 
                ++deleted_count;
                integration::logger_adapter::debug(
                    "Deleted study study_uid={} study_date={}",
                    study.study_uid, study.study_date);
            }
 
            integration::logger_adapter::info(
                "Cleanup task completed deleted={} skipped={} dry_run={}",
                deleted_count, skipped_count, config.dry_run);
 
            return std::nullopt;  // Success
        } catch (const std::exception& e) {
            return std::string("Cleanup failed: ") + e.what();
        }
    };
}

References kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::integration::logger_adapter::info(), and kcenon::pacs::integration::logger_adapter::warn().

Here is the call graph for this function:

create_verification_callback()

auto kcenon::pacs::workflow::task_scheduler::create_verification_callback ( const verification_config & config ) -> task_callback_with_result

nodiscardprivate

Create verification task callback.

Parameters

config

Verification configuration

Returns

Task callback

Definition at line 1356 of file task_scheduler.cpp.

                                 {
    return [this, config]() -> std::optional<std::string> {
        integration::logger_adapter::info(
            "Running verification task check_checksums={} check_db={}",
            config.check_checksums,
            config.check_db_consistency);
 
        try {
            std::size_t verified = 0;
            std::size_t errors = 0;
            std::size_t missing_files = 0;
 
            // Database integrity check
            if (config.check_db_consistency) {
                auto db_result = database_.verify_integrity();
                if (db_result.is_err()) {
                    integration::logger_adapter::error(
                        "Database integrity check failed error={}",
                        db_result.error().message);
                    ++errors;
                } else {
                    integration::logger_adapter::debug("Database integrity check passed");
                }
            }
 
            // Get studies for verification
            storage::study_query query;
            query.limit = config.max_verifications_per_cycle;
            auto studies_result = database_.search_studies(query);
            if (!studies_result.is_ok()) {
                integration::logger_adapter::error(
                    "Failed to query studies error={}",
                    studies_result.error().message);
                return "Verification failed: " + studies_result.error().message;
            }
 
            for (const auto& study : studies_result.value()) {
                // Get all files for this study
                auto file_paths_result = database_.get_study_files(study.study_uid);
                if (!file_paths_result.is_ok()) {
                    integration::logger_adapter::error(
                        "Failed to get study files study_uid={} error={}",
                        study.study_uid, file_paths_result.error().message);
                    ++errors;
                    continue;
                }
 
                for (const auto& file_path : file_paths_result.value()) {
                    std::filesystem::path path(file_path);
 
                    // Check file existence
                    if (!std::filesystem::exists(path)) {
                        ++missing_files;
                        integration::logger_adapter::warn(
                            "Missing file detected file_path={} study_uid={}",
                            file_path, study.study_uid);
 
                        if (config.repair_on_failure) {
                            // Extract SOP UID and remove orphaned database record
                            std::string sop_uid = path.stem().string();
                            (void)database_.delete_instance(sop_uid);
                            integration::logger_adapter::info(
                                "Removed orphaned database record sop_uid={}", sop_uid);
                        }
                        continue;
                    }
 
                    // Checksum verification
                    if (config.check_checksums) {
                        // Get file size as basic verification
                        // Full checksum verification would require storing checksums in DB
                        try {
                            auto file_size = std::filesystem::file_size(path);
                            if (file_size == 0) {
                                ++errors;
                                integration::logger_adapter::warn(
                                    "Empty file detected file_path={}", file_path);
                            }
                        } catch (const std::filesystem::filesystem_error& e) {
                            ++errors;
                            integration::logger_adapter::error(
                                "Cannot read file file_path={} error={}",
                                file_path, e.what());
                        }
                    }
 
                    ++verified;
                }
            }
 
            // File storage integrity check
            if (file_storage_ != nullptr) {
                auto storage_result = file_storage_->verify_integrity();
                if (storage_result.is_err()) {
                    integration::logger_adapter::warn(
                        "Storage integrity check reported issues error={}",
                        storage_result.error().message);
                }
            }
 
            integration::logger_adapter::info(
                "Verification task completed verified={} errors={} missing_files={}",
                verified, errors, missing_files);
 
            if (errors > 0 || missing_files > 0) {
                std::ostringstream oss;
                oss << "Verification found " << errors << " errors and "
                    << missing_files << " missing files";
                return oss.str();
            }
 
            return std::nullopt;  // Success
        } catch (const std::exception& e) {
            return std::string("Verification failed: ") + e.what();
        }
    };
}

References kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::integration::logger_adapter::info(), and kcenon::pacs::integration::logger_adapter::warn().

Here is the call graph for this function:

deserialize_tasks()

auto kcenon::pacs::workflow::task_scheduler::deserialize_tasks ( const std::string & json ) -> std::size_t

private

Deserialize tasks from JSON.

Parameters

json	JSON string

Returns

Number of tasks loaded

Definition at line 1554 of file task_scheduler.cpp.

                                                                         {
    // Note: Full deserialization is not supported without a proper JSON library.
    //
    // Limitations:
    // 1. Callbacks cannot be serialized/deserialized (function pointers)
    // 2. Schedule details are not currently serialized
    // 3. Tags and other complex fields are not serialized
    //
    // For production use, consider:
    // 1. Using nlohmann/json or similar library for proper JSON parsing
    // 2. Storing task configurations separately and re-creating tasks on startup
    // 3. Using a database-backed persistence layer
    //
    // Current implementation returns 0 (no tasks loaded) to indicate
    // that manual task registration is required after restart.
 
    if (json.empty()) {
        return 0;
    }
 
    integration::logger_adapter::warn(
        "deserialize_tasks called but full deserialization is not implemented. "
        "Tasks must be re-registered programmatically after scheduler restart.");
 
    return 0;
}

References kcenon::pacs::integration::logger_adapter::warn().

Here is the call graph for this function:

execute_cycle()

void kcenon::pacs::workflow::task_scheduler::execute_cycle ( )

private

Execute a single scheduler cycle.

Checks for due tasks and executes them.

Definition at line 755 of file task_scheduler.cpp.

                                   {
    auto now = std::chrono::system_clock::now();
    std::vector<task_id> due_tasks;
 
    // Find due tasks
    {
        std::lock_guard<std::mutex> lock(tasks_mutex_);
 
        for (auto& [id, task] : tasks_) {
            if (!task.enabled || task.state != task_state::pending) {
                continue;
            }
 
            if (task.next_run_at && *task.next_run_at <= now) {
                due_tasks.push_back(id);
            }
        }
    }
 
    if (due_tasks.empty()) {
        return;
    }
 
    // Sort by priority (higher first)
    {
        std::lock_guard<std::mutex> lock(tasks_mutex_);
        std::ranges::sort(due_tasks, [this](const auto& a, const auto& b) {
            return tasks_[a].priority > tasks_[b].priority;
        });
    }
 
    // Execute tasks (respecting max concurrent limit)
    std::size_t executed = 0;
    std::size_t succeeded = 0;
    std::size_t failed = 0;
 
    for (const auto& id : due_tasks) {
        if (running_count_.load() >= config_.max_concurrent_tasks) {
            break;  // Max concurrent reached
        }
 
        std::lock_guard<std::mutex> lock(tasks_mutex_);
        auto it = tasks_.find(id);
        if (it == tasks_.end()) {
            continue;
        }
 
        auto& task = it->second;
 
        // Mark as running
        task.state = task_state::running;
        ++running_count_;
 
        // Execute task
        auto record = execute_task(task);
 
        // Update task state
        task.state = (record.state == task_state::completed)
            ? task_state::pending
            : record.state;
        task.last_run_at = record.started_at;
        task.last_execution = record;
        ++task.execution_count;
 
        if (record.state == task_state::completed) {
            ++task.success_count;
            ++succeeded;
        } else {
            ++task.failure_count;
            ++failed;
        }
 
        // Calculate next run time
        if (task.enabled && task.state == task_state::pending) {
            task.next_run_at = calculate_next_run(
                task.task_schedule,
                std::chrono::system_clock::now());
 
            // For one-time tasks, disable after execution
            if (std::holds_alternative<one_time_schedule>(task.task_schedule)) {
                task.enabled = false;
            }
        }
 
        --running_count_;
        ++executed;
 
        // Record execution history
        record_execution(id, record);
        update_stats(record);
 
        // Invoke callbacks
        if (config_.on_task_complete) {
            config_.on_task_complete(id, record);
        }
        if (record.state == task_state::failed && config_.on_task_error) {
            config_.on_task_error(id, record.error_message.value_or("Unknown error"));
        }
    }
 
    // Log cycle summary
    if (executed > 0) {
        integration::logger_adapter::info(
            "Scheduler cycle completed executed={} succeeded={} failed={}",
            executed, succeeded, failed);
 
        // Record metrics
        integration::monitoring_adapter::increment_counter(
            "scheduler_tasks_executed", static_cast<int64_t>(executed));
        integration::monitoring_adapter::increment_counter(
            "scheduler_tasks_succeeded", static_cast<int64_t>(succeeded));
        integration::monitoring_adapter::increment_counter(
            "scheduler_tasks_failed", static_cast<int64_t>(failed));
 
        // Update stats
        {
            std::lock_guard<std::mutex> lock(stats_mutex_);
            stats_.last_cycle_at = std::chrono::system_clock::now();
        }
 
        // Invoke cycle callback
        if (config_.on_cycle_complete) {
            config_.on_cycle_complete(executed, succeeded, failed);
        }
    }
}

References calculate_next_run(), kcenon::pacs::workflow::completed, config_, execute_task(), kcenon::pacs::workflow::failed, kcenon::pacs::integration::monitoring_adapter::increment_counter(), kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::workflow::scheduler_stats::last_cycle_at, kcenon::pacs::workflow::task_scheduler_config::max_concurrent_tasks, kcenon::pacs::workflow::task_scheduler_config::on_cycle_complete, kcenon::pacs::workflow::task_scheduler_config::on_task_complete, kcenon::pacs::workflow::task_scheduler_config::on_task_error, kcenon::pacs::workflow::pending, record_execution(), kcenon::pacs::workflow::running, running_count_, stats_, stats_mutex_, tasks_, tasks_mutex_, and update_stats().

Referenced by run_loop().

Here is the call graph for this function:

Here is the caller graph for this function:

execute_task()

auto kcenon::pacs::workflow::task_scheduler::execute_task ( scheduled_task & task ) -> task_execution_record

private

Execute a single task.

Parameters

task	Task to execute

Returns

Execution record

Definition at line 882 of file task_scheduler.cpp.

                             {
    task_execution_record record;
    record.execution_id = generate_execution_id();
    record.task_id = task.id;
    record.started_at = std::chrono::system_clock::now();
    record.state = task_state::running;
 
    integration::logger_adapter::debug(
        "Executing task task_id={} name={}",
        task.id, task.name);
 
    auto start_time = std::chrono::steady_clock::now();
 
    // Retry logic: attempt up to (max_retries + 1) times
    std::size_t attempt = 0;
    const std::size_t max_attempts = task.max_retries + 1;
 
    while (attempt < max_attempts) {
        attempt++;
 
        if (attempt > 1) {
            integration::logger_adapter::info(
                "Retrying task task_id={} attempt={}/{}",
                task.id, attempt, max_attempts);
 
            // Wait before retry
            std::this_thread::sleep_for(task.retry_delay);
        }
 
        try {
            if (!task.callback) {
                record.state = task_state::failed;
                record.error_message = "No callback defined";
                break;  // No point retrying without callback
            }
 
            // Execute with timeout if configured
            if (task.timeout.count() > 0) {
                // Run callback in async task with timeout
                auto future = std::async(std::launch::async, task.callback);
 
                auto status = future.wait_for(task.timeout);
                if (status == std::future_status::timeout) {
                    record.state = task_state::failed;
                    record.error_message = "Task execution timed out after " +
                        std::to_string(task.timeout.count()) + " seconds";
                    integration::logger_adapter::warn(
                        "Task timed out task_id={} timeout_seconds={}",
                        task.id, task.timeout.count());
                    // Continue to retry if attempts remain
                    continue;
                }
 
                // Get result (may throw)
                auto result = future.get();
                if (result.has_value()) {
                    record.state = task_state::failed;
                    record.error_message = *result;
                } else {
                    record.state = task_state::completed;
                    break;  // Success, no need to retry
                }
            } else {
                // No timeout, execute directly
                auto result = task.callback();
 
                if (result.has_value()) {
                    record.state = task_state::failed;
                    record.error_message = *result;
                } else {
                    record.state = task_state::completed;
                    break;  // Success, no need to retry
                }
            }
        } catch (const std::exception& e) {
            record.state = task_state::failed;
            record.error_message = e.what();
            integration::logger_adapter::error(
                "Task execution failed task_id={} attempt={} error={}",
                task.id, attempt, e.what());
        }
 
        // If we've exhausted all retries and still failed
        if (attempt >= max_attempts && record.state == task_state::failed) {
            integration::logger_adapter::error(
                "Task failed after {} attempts task_id={}",
                max_attempts, task.id);
        }
    }
 
    record.ended_at = std::chrono::system_clock::now();
 
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(
        std::chrono::steady_clock::now() - start_time);
 
    integration::logger_adapter::info(
        "Task execution completed task_id={} name={} state={} duration_ms={} attempts={}",
        task.id, task.name, to_string(record.state), duration.count(), attempt);
 
    // Record metrics
    integration::monitoring_adapter::record_histogram(
        "scheduler_task_duration_ms",
        static_cast<double>(duration.count()));
 
    if (attempt > 1) {
        integration::monitoring_adapter::increment_counter(
            "scheduler_task_retries_total");
    }
 
    return record;
}

References kcenon::pacs::workflow::completed, kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::workflow::failed, kcenon::pacs::integration::monitoring_adapter::increment_counter(), kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::integration::monitoring_adapter::record_histogram(), kcenon::pacs::workflow::running, and kcenon::pacs::integration::logger_adapter::warn().

Referenced by execute_cycle().

Here is the call graph for this function:

Here is the caller graph for this function:

generate_execution_id()

auto kcenon::pacs::workflow::task_scheduler::generate_execution_id ( ) const -> std::string

nodiscardprivate

Generate unique execution ID.

Returns

New unique execution ID

Definition at line 1097 of file task_scheduler.cpp.

                                                              {
    auto id = next_execution_id_.fetch_add(1);
 
    // Add timestamp for uniqueness
    auto now = std::chrono::system_clock::now();
    auto millis = std::chrono::duration_cast<std::chrono::milliseconds>(
        now.time_since_epoch()).count();
 
    return "exec_" + std::to_string(millis) + "_" + std::to_string(id);
}

References next_execution_id_.

generate_task_id()

auto kcenon::pacs::workflow::task_scheduler::generate_task_id ( ) const -> task_id

nodiscardprivate

Generate unique task ID.

Returns

New unique task ID

Definition at line 1092 of file task_scheduler.cpp.

                                                       {
    auto id = next_task_id_.fetch_add(1);
    return "task_" + std::to_string(id);
}

References next_task_id_.

get_execution_history()

auto kcenon::pacs::workflow::task_scheduler::get_execution_history ( const task_id & id, std::size_t limit = 100 ) const -> std::vector<task_execution_record>

nodiscard

Get execution history for a task.

Parameters

id	Task ID
limit	Maximum number of records to return

Returns

Vector of execution records (most recent first)

Definition at line 541 of file task_scheduler.cpp.

                                                               {
    std::lock_guard<std::mutex> lock(history_mutex_);
 
    auto it = execution_history_.find(id);
    if (it == execution_history_.end()) {
        return {};
    }
 
    const auto& records = it->second;
    if (records.size() <= limit) {
        return records;
    }
 
    // Return most recent records
    return std::vector<task_execution_record>(
        records.end() - static_cast<std::ptrdiff_t>(limit),
        records.end());
}

get_recent_executions()

auto kcenon::pacs::workflow::task_scheduler::get_recent_executions ( std::size_t limit = 100 ) const -> std::vector<task_execution_record>

nodiscard

Get all recent executions.

Parameters

limit

Maximum number of records to return

Returns

Vector of execution records (most recent first)

Definition at line 562 of file task_scheduler.cpp.

                                        {
    std::lock_guard<std::mutex> lock(history_mutex_);
 
    std::vector<task_execution_record> all_records;
 
    for (const auto& [id, records] : execution_history_) {
        all_records.insert(all_records.end(), records.begin(), records.end());
    }
 
    // Sort by start time (most recent first)
    std::ranges::sort(all_records, [](const auto& a, const auto& b) {
        return a.started_at > b.started_at;
    });
 
    if (all_records.size() <= limit) {
        return all_records;
    }
 
    all_records.resize(limit);
    return all_records;
}

get_stats()

auto kcenon::pacs::workflow::task_scheduler::get_stats ( ) const -> scheduler_stats

nodiscard

Get scheduler statistics.

Returns

Current scheduler statistics

Definition at line 608 of file task_scheduler.cpp.

                                                        {
    std::lock_guard<std::mutex> lock(stats_mutex_);
 
    scheduler_stats result = stats_;
 
    // Calculate uptime
    if (running_.load()) {
        auto now = std::chrono::steady_clock::now();
        result.uptime = std::chrono::duration_cast<std::chrono::seconds>(
            now - start_time_);
    }
 
    result.running_tasks = running_count_.load();
 
    {
        std::lock_guard<std::mutex> tasks_lock(tasks_mutex_);
        result.scheduled_tasks = tasks_.size();
    }
 
    return result;
}

References running_, running_count_, kcenon::pacs::workflow::scheduler_stats::running_tasks, kcenon::pacs::workflow::scheduler_stats::scheduled_tasks, start_time_, stats_, stats_mutex_, tasks_, tasks_mutex_, and kcenon::pacs::workflow::scheduler_stats::uptime.

get_task()

auto kcenon::pacs::workflow::task_scheduler::get_task ( const task_id & id ) const -> std::optional<scheduled_task>

nodiscard

Get a specific task by ID.

Parameters

id

Task ID

Returns

Task if found, nullopt otherwise

Definition at line 416 of file task_scheduler.cpp.

                                   {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    auto it = tasks_.find(id);
    if (it == tasks_.end()) {
        return std::nullopt;
    }
 
    return it->second;
}

is_running()

auto kcenon::pacs::workflow::task_scheduler::is_running ( ) const -> bool

nodiscardnoexcept

Check if the scheduler is running.

Returns

true if the scheduler is actively running

Definition at line 201 of file task_scheduler.cpp.

                                                       {
    return running_.load();
}

References running_.

list_tasks() [1/3]

auto kcenon::pacs::workflow::task_scheduler::list_tasks ( ) const -> std::vector<scheduled_task>

nodiscard

List all scheduled tasks.

Returns

Vector of all scheduled tasks

Definition at line 373 of file task_scheduler.cpp.

                                                                   {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    std::vector<scheduled_task> result;
    result.reserve(tasks_.size());
 
    for (const auto& [id, task] : tasks_) {
        result.push_back(task);
    }
 
    return result;
}

References tasks_, and tasks_mutex_.

list_tasks() [2/3]

auto kcenon::pacs::workflow::task_scheduler::list_tasks ( task_state state ) const -> std::vector<scheduled_task>

nodiscard

List tasks by state.

Parameters

state

Task state to filter by

Returns

Vector of matching tasks

Definition at line 401 of file task_scheduler.cpp.

                                 {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    std::vector<scheduled_task> result;
 
    for (const auto& [id, task] : tasks_) {
        if (task.state == state) {
            result.push_back(task);
        }
    }
 
    return result;
}

list_tasks() [3/3]

auto kcenon::pacs::workflow::task_scheduler::list_tasks ( task_type type ) const -> std::vector<scheduled_task>

nodiscard

List tasks by type.

Parameters

type	Task type to filter by

Returns

Vector of matching tasks

Definition at line 386 of file task_scheduler.cpp.

                                 {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    std::vector<scheduled_task> result;
 
    for (const auto& [id, task] : tasks_) {
        if (task.type == type) {
            result.push_back(task);
        }
    }
 
    return result;
}

load_tasks()

auto kcenon::pacs::workflow::task_scheduler::load_tasks

(

)

-> std::size_t

Load tasks from persistence storage.

Returns

Number of tasks loaded

Definition at line 681 of file task_scheduler.cpp.

                                             {
    if (config_.persistence_path.empty()) {
        return 0;
    }
 
    try {
        std::ifstream file(config_.persistence_path);
        if (!file) {
            integration::logger_adapter::debug(
                "No persistence file found path={}",
                config_.persistence_path);
            return 0;
        }
 
        std::string json((std::istreambuf_iterator<char>(file)),
                        std::istreambuf_iterator<char>());
        file.close();
 
        std::size_t count = deserialize_tasks(json);
 
        integration::logger_adapter::info(
            "Loaded tasks from persistence path={} count={}",
            config_.persistence_path, count);
 
        return count;
    } catch (const std::exception& e) {
        integration::logger_adapter::error(
            "Failed to load tasks error={}", e.what());
        return 0;
    }
}

References kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), and kcenon::pacs::integration::logger_adapter::info().

Here is the call graph for this function:

operator=() [1/2]

task_scheduler & kcenon::pacs::workflow::task_scheduler::operator= ( const task_scheduler & )

delete

operator=() [2/2]

task_scheduler & kcenon::pacs::workflow::task_scheduler::operator= ( task_scheduler && )

delete

pause_task()

auto kcenon::pacs::workflow::task_scheduler::pause_task

(

const task_id &

id

)

-> bool

Pause a scheduled task.

Temporarily suspends task execution. The task remains scheduled but will not execute until resumed.

Parameters

id	Task ID to pause

Returns

true if task was found and paused

Definition at line 447 of file task_scheduler.cpp.

                                                         {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    auto it = tasks_.find(id);
    if (it == tasks_.end()) {
        return false;
    }
 
    if (it->second.state == task_state::running) {
        return false;  // Cannot pause a running task
    }
 
    it->second.state = task_state::paused;
    it->second.updated_at = std::chrono::system_clock::now();
 
    integration::logger_adapter::info(
        "Paused scheduled task task_id={} name={}",
        id, it->second.name);
 
    return true;
}

References kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::workflow::paused, and kcenon::pacs::workflow::running.

Here is the call graph for this function:

pending_count()

auto kcenon::pacs::workflow::task_scheduler::pending_count ( ) const -> std::size_t

nodiscardnoexcept

Get number of pending tasks.

Returns

Number of tasks waiting to execute

Definition at line 630 of file task_scheduler.cpp.

                                                               {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    std::size_t count = 0;
    for (const auto& [id, task] : tasks_) {
        if (task.enabled && task.state == task_state::pending) {
            ++count;
        }
    }
    return count;
}

References kcenon::pacs::workflow::pending, tasks_, and tasks_mutex_.

record_execution()

void kcenon::pacs::workflow::task_scheduler::record_execution ( const task_id & task_id, const task_execution_record & record )

private

Record task execution.

Parameters

task_id	Task ID
record	Execution record

Definition at line 1475 of file task_scheduler.cpp.

                                                                          {
    std::lock_guard<std::mutex> lock(history_mutex_);
 
    auto& history = execution_history_[task_id];
    history.push_back(record);
 
    // Limit history size (keep last 1000 records per task)
    constexpr std::size_t max_history = 1000;
    if (history.size() > max_history) {
        history.erase(
            history.begin(),
            history.begin() + static_cast<std::ptrdiff_t>(history.size() - max_history));
    }
}

References execution_history_, and history_mutex_.

Referenced by execute_cycle().

Here is the caller graph for this function:

resume_task()

auto kcenon::pacs::workflow::task_scheduler::resume_task

(

const task_id &

id

)

-> bool

Resume a paused task.

Parameters

id	Task ID to resume

Returns

true if task was found and resumed

Definition at line 469 of file task_scheduler.cpp.

                                                          {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    auto it = tasks_.find(id);
    if (it == tasks_.end()) {
        return false;
    }
 
    if (it->second.state != task_state::paused) {
        return false;  // Not paused
    }
 
    it->second.state = task_state::pending;
    it->second.next_run_at = calculate_next_run(it->second.task_schedule);
    it->second.updated_at = std::chrono::system_clock::now();
 
    integration::logger_adapter::info(
        "Resumed scheduled task task_id={} name={}",
        id, it->second.name);
 
    return true;
}

References kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::workflow::paused, and kcenon::pacs::workflow::pending.

Here is the call graph for this function:

run_loop()

void kcenon::pacs::workflow::task_scheduler::run_loop ( )

private

Background thread main loop.

Definition at line 733 of file task_scheduler.cpp.

                              {
    integration::logger_adapter::debug("Task scheduler thread started");
 
    while (!stop_requested_.load()) {
        std::unique_lock<std::mutex> lock(mutex_);
 
        // Wait for check interval or until woken up
        cv_.wait_for(lock, config_.check_interval, [this]() {
            return stop_requested_.load();
        });
 
        if (stop_requested_.load()) {
            break;
        }
 
        lock.unlock();
        execute_cycle();
    }
 
    integration::logger_adapter::debug("Task scheduler thread stopped");
}

References kcenon::pacs::workflow::task_scheduler_config::check_interval, config_, cv_, kcenon::pacs::integration::logger_adapter::debug(), execute_cycle(), mutex_, and stop_requested_.

Referenced by start().

Here is the call graph for this function:

Here is the caller graph for this function:

running_count()

auto kcenon::pacs::workflow::task_scheduler::running_count ( ) const -> std::size_t

nodiscardnoexcept

Get number of running tasks.

Returns

Number of tasks currently executing

Definition at line 642 of file task_scheduler.cpp.

                                                               {
    return running_count_.load();
}

References running_count_.

save_tasks()

auto kcenon::pacs::workflow::task_scheduler::save_tasks

(

)

const -> bool

Save all tasks to persistence storage.

Returns

true if saved successfully

Definition at line 650 of file task_scheduler.cpp.

                                              {
    if (config_.persistence_path.empty()) {
        return false;
    }
 
    try {
        std::string json = serialize_tasks();
 
        std::ofstream file(config_.persistence_path);
        if (!file) {
            integration::logger_adapter::error(
                "Failed to open persistence file path={}",
                config_.persistence_path);
            return false;
        }
 
        file << json;
        file.close();
 
        integration::logger_adapter::debug(
            "Saved tasks to persistence path={}",
            config_.persistence_path);
 
        return true;
    } catch (const std::exception& e) {
        integration::logger_adapter::error(
            "Failed to save tasks error={}", e.what());
        return false;
    }
}

References config_, kcenon::pacs::integration::logger_adapter::debug(), kcenon::pacs::integration::logger_adapter::error(), kcenon::pacs::workflow::task_scheduler_config::persistence_path, and serialize_tasks().

Referenced by ~task_scheduler().

Here is the call graph for this function:

Here is the caller graph for this function:

schedule() [1/3]

auto kcenon::pacs::workflow::task_scheduler::schedule	(	const std::string &	name,
		const std::string &	description,
		const cron_schedule &	cron_expr,
		task_callback_with_result	callback ) -> task_id

Schedule a custom task with cron schedule.

Parameters

name	Task name
description	Task description
cron_expr	Cron schedule expression
callback	Task callback function

Returns

Task ID

Definition at line 293 of file task_scheduler.cpp.

                                                   {
 
    scheduled_task task;
    task.id = generate_task_id();
    task.name = name;
    task.description = description;
    task.type = task_type::custom;
    task.task_schedule = cron_expr;
    task.enabled = true;
    task.callback = std::move(callback);
    task.created_at = std::chrono::system_clock::now();
    task.updated_at = task.created_at;
 
    return schedule(std::move(task));
}

References kcenon::pacs::workflow::scheduled_task::callback, kcenon::pacs::workflow::scheduled_task::created_at, kcenon::pacs::workflow::custom, kcenon::pacs::workflow::scheduled_task::description, kcenon::pacs::workflow::scheduled_task::enabled, kcenon::pacs::workflow::scheduled_task::id, kcenon::pacs::workflow::scheduled_task::name, name, kcenon::pacs::workflow::scheduled_task::task_schedule, kcenon::pacs::workflow::scheduled_task::type, and kcenon::pacs::workflow::scheduled_task::updated_at.

schedule() [2/3]

auto kcenon::pacs::workflow::task_scheduler::schedule	(	const std::string &	name,
		const std::string &	description,
		std::chrono::seconds	interval,
		task_callback_with_result	callback ) -> task_id

Schedule a custom task with interval.

Parameters

name	Task name
description	Task description
interval	Interval between executions
callback	Task callback function

Returns

Task ID

Definition at line 273 of file task_scheduler.cpp.

                                                   {
 
    scheduled_task task;
    task.id = generate_task_id();
    task.name = name;
    task.description = description;
    task.type = task_type::custom;
    task.task_schedule = interval_schedule{interval, std::nullopt};
    task.enabled = true;
    task.callback = std::move(callback);
    task.created_at = std::chrono::system_clock::now();
    task.updated_at = task.created_at;
 
    return schedule(std::move(task));
}

References kcenon::pacs::workflow::scheduled_task::callback, kcenon::pacs::workflow::scheduled_task::created_at, kcenon::pacs::workflow::custom, kcenon::pacs::workflow::scheduled_task::description, kcenon::pacs::workflow::scheduled_task::enabled, kcenon::pacs::workflow::scheduled_task::id, kcenon::pacs::workflow::scheduled_task::name, name, kcenon::pacs::workflow::scheduled_task::task_schedule, kcenon::pacs::workflow::scheduled_task::type, and kcenon::pacs::workflow::scheduled_task::updated_at.

schedule() [3/3]

auto kcenon::pacs::workflow::task_scheduler::schedule

(

scheduled_task

task

)

-> task_id

Schedule a task with full definition.

Parameters

task	Complete task definition

Returns

Task ID

Definition at line 334 of file task_scheduler.cpp.

                                                            {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    // Calculate next run time if not set
    if (!task.next_run_at) {
        task.next_run_at = calculate_next_run(task.task_schedule);
    }
 
    auto id = task.id;
 
    // Check if task already exists
    auto it = tasks_.find(id);
    if (it != tasks_.end()) {
        // Update existing task
        it->second = std::move(task);
        integration::logger_adapter::info(
            "Updated scheduled task task_id={} name={}",
            id, it->second.name);
    } else {
        // Add new task
        tasks_.emplace(id, std::move(task));
        integration::logger_adapter::info(
            "Added scheduled task task_id={} name={}",
            id, tasks_[id].name);
    }
 
    // Update stats
    {
        std::lock_guard<std::mutex> stats_lock(stats_mutex_);
        stats_.scheduled_tasks = tasks_.size();
    }
 
    return id;
}

References kcenon::pacs::integration::logger_adapter::info(), and name.

Here is the call graph for this function:

schedule_archive()

auto kcenon::pacs::workflow::task_scheduler::schedule_archive

(

const archive_config &

config

)

-> task_id

Schedule archive task.

Creates an archive task with the specified configuration. If an archive task already exists, updates its configuration.

Parameters

config

Archive configuration

Returns

Task ID for the archive task

Definition at line 230 of file task_scheduler.cpp.

                                                                             {
    scheduled_task task;
    task.id = "archive_task";
    task.name = "Study Archival";
    task.description = "Archives studies to secondary storage";
    task.type = task_type::archive;
    task.task_schedule = config.archive_schedule;
    task.enabled = true;
    task.priority = 5;
    task.tags = {"maintenance", "archive"};
    task.callback = create_archive_callback(config);
    task.created_at = std::chrono::system_clock::now();
    task.updated_at = task.created_at;
 
    return schedule(std::move(task));
}

References kcenon::pacs::workflow::archive, kcenon::pacs::workflow::scheduled_task::callback, kcenon::pacs::workflow::scheduled_task::created_at, kcenon::pacs::workflow::scheduled_task::description, kcenon::pacs::workflow::scheduled_task::enabled, kcenon::pacs::workflow::scheduled_task::id, kcenon::pacs::workflow::scheduled_task::name, kcenon::pacs::workflow::scheduled_task::priority, kcenon::pacs::workflow::scheduled_task::tags, kcenon::pacs::workflow::scheduled_task::task_schedule, kcenon::pacs::workflow::scheduled_task::type, and kcenon::pacs::workflow::scheduled_task::updated_at.

schedule_cleanup()

auto kcenon::pacs::workflow::task_scheduler::schedule_cleanup

(

const cleanup_config &

config

)

-> task_id

Schedule cleanup task.

Creates a cleanup task with the specified configuration. If a cleanup task already exists, updates its configuration.

Parameters

config

Cleanup configuration

Returns

Task ID for the cleanup task

Definition at line 209 of file task_scheduler.cpp.

                                                                             {
    scheduled_task task;
    task.id = "cleanup_task";
    task.name = "Storage Cleanup";
    task.description = "Removes old studies based on retention policy";
    task.type = task_type::cleanup;
    task.task_schedule = config.cleanup_schedule;
    task.enabled = true;
    task.priority = 10;
    task.tags = {"maintenance", "storage"};
    task.callback = create_cleanup_callback(config);
    task.created_at = std::chrono::system_clock::now();
    task.updated_at = task.created_at;
 
    return schedule(std::move(task));
}

References kcenon::pacs::workflow::scheduled_task::callback, kcenon::pacs::workflow::cleanup, kcenon::pacs::workflow::scheduled_task::created_at, kcenon::pacs::workflow::scheduled_task::description, kcenon::pacs::workflow::scheduled_task::enabled, kcenon::pacs::workflow::scheduled_task::id, kcenon::pacs::workflow::scheduled_task::name, kcenon::pacs::workflow::scheduled_task::priority, kcenon::pacs::workflow::scheduled_task::tags, kcenon::pacs::workflow::scheduled_task::task_schedule, kcenon::pacs::workflow::scheduled_task::type, and kcenon::pacs::workflow::scheduled_task::updated_at.

schedule_once()

auto kcenon::pacs::workflow::task_scheduler::schedule_once	(	const std::string &	name,
		const std::string &	description,
		std::chrono::system_clock::time_point	execute_at,
		task_callback_with_result	callback ) -> task_id

Schedule a one-time task.

Parameters

name	Task name
description	Task description
execute_at	Execution time
callback	Task callback function

Returns

Task ID

Definition at line 313 of file task_scheduler.cpp.

                                                   {
 
    scheduled_task task;
    task.id = generate_task_id();
    task.name = name;
    task.description = description;
    task.type = task_type::custom;
    task.task_schedule = one_time_schedule{execute_at};
    task.enabled = true;
    task.callback = std::move(callback);
    task.created_at = std::chrono::system_clock::now();
    task.updated_at = task.created_at;
    task.next_run_at = execute_at;
 
    return schedule(std::move(task));
}

References kcenon::pacs::workflow::scheduled_task::callback, kcenon::pacs::workflow::scheduled_task::created_at, kcenon::pacs::workflow::custom, kcenon::pacs::workflow::scheduled_task::description, kcenon::pacs::workflow::scheduled_task::enabled, kcenon::pacs::workflow::scheduled_task::id, kcenon::pacs::workflow::scheduled_task::name, name, kcenon::pacs::workflow::scheduled_task::next_run_at, kcenon::pacs::workflow::scheduled_task::task_schedule, kcenon::pacs::workflow::scheduled_task::type, and kcenon::pacs::workflow::scheduled_task::updated_at.

schedule_verification()

auto kcenon::pacs::workflow::task_scheduler::schedule_verification

(

const verification_config &

config

)

-> task_id

Schedule verification task.

Creates a verification task with the specified configuration. If a verification task already exists, updates its configuration.

Parameters

config

Verification configuration

Returns

Task ID for the verification task

Definition at line 251 of file task_scheduler.cpp.

               {
    scheduled_task task;
    task.id = "verification_task";
    task.name = "Data Verification";
    task.description = "Verifies data integrity and consistency";
    task.type = task_type::verification;
    task.task_schedule = config.verification_schedule;
    task.enabled = true;
    task.priority = 8;
    task.tags = {"maintenance", "integrity"};
    task.callback = create_verification_callback(config);
    task.created_at = std::chrono::system_clock::now();
    task.updated_at = task.created_at;
 
    return schedule(std::move(task));
}

References kcenon::pacs::workflow::scheduled_task::callback, kcenon::pacs::workflow::scheduled_task::created_at, kcenon::pacs::workflow::scheduled_task::description, kcenon::pacs::workflow::scheduled_task::enabled, kcenon::pacs::workflow::scheduled_task::id, kcenon::pacs::workflow::scheduled_task::name, kcenon::pacs::workflow::scheduled_task::priority, kcenon::pacs::workflow::scheduled_task::tags, kcenon::pacs::workflow::scheduled_task::task_schedule, kcenon::pacs::workflow::scheduled_task::type, kcenon::pacs::workflow::scheduled_task::updated_at, and kcenon::pacs::workflow::verification.

serialize_tasks()

auto kcenon::pacs::workflow::task_scheduler::serialize_tasks ( ) const -> std::string

nodiscardprivate

Serialize tasks to JSON.

Returns

JSON string representation

Definition at line 1526 of file task_scheduler.cpp.

                                                        {
    // Simple JSON serialization
    // In production, use a proper JSON library
 
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    std::ostringstream oss;
    oss << "{\n  \"tasks\": [\n";
 
    bool first = true;
    for (const auto& [id, task] : tasks_) {
        if (!first) oss << ",\n";
        first = false;
 
        oss << "    {\n";
        oss << "      \"id\": \"" << task.id << "\",\n";
        oss << "      \"name\": \"" << task.name << "\",\n";
        oss << "      \"type\": \"" << to_string(task.type) << "\",\n";
        oss << "      \"enabled\": " << (task.enabled ? "true" : "false") << ",\n";
        oss << "      \"priority\": " << task.priority << "\n";
        oss << "    }";
    }
 
    oss << "\n  ]\n}";
 
    return oss.str();
}

References tasks_, tasks_mutex_, and kcenon::pacs::workflow::to_string().

Referenced by save_tasks().

Here is the call graph for this function:

Here is the caller graph for this function:

set_error_callback()

void kcenon::pacs::workflow::task_scheduler::set_error_callback

(

task_scheduler_config::task_error_callback

callback

)

Set the error callback.

Parameters

callback

Function called when a task fails

Definition at line 723 of file task_scheduler.cpp.

                                                       {
    std::lock_guard<std::mutex> lock(mutex_);
    config_.on_task_error = std::move(callback);
}

References config_, mutex_, and kcenon::pacs::workflow::task_scheduler_config::on_task_error.

set_task_complete_callback()

void kcenon::pacs::workflow::task_scheduler::set_task_complete_callback

(

task_scheduler_config::task_complete_callback

callback

)

Set the task complete callback.

Parameters

callback

Function called after each task execution

Definition at line 717 of file task_scheduler.cpp.

                                                          {
    std::lock_guard<std::mutex> lock(mutex_);
    config_.on_task_complete = std::move(callback);
}

References config_, mutex_, and kcenon::pacs::workflow::task_scheduler_config::on_task_complete.

start()

void kcenon::pacs::workflow::task_scheduler::start

(

)

Start the scheduler service.

Starts the background scheduler thread and begins monitoring for tasks to execute.

Definition at line 164 of file task_scheduler.cpp.

                           {
    if (running_.exchange(true)) {
        return;  // Already running
    }
 
    stop_requested_.store(false);
    start_time_ = std::chrono::steady_clock::now();
 
    scheduler_thread_ = std::thread([this]() {
        run_loop();
    });
 
    integration::logger_adapter::info(
        "Task scheduler started check_interval_sec={} max_concurrent={}",
        config_.check_interval.count(),
        config_.max_concurrent_tasks);
}

References kcenon::pacs::workflow::task_scheduler_config::check_interval, config_, kcenon::pacs::integration::logger_adapter::info(), kcenon::pacs::workflow::task_scheduler_config::max_concurrent_tasks, run_loop(), running_, scheduler_thread_, start_time_, and stop_requested_.

Here is the call graph for this function:

stop()

void kcenon::pacs::workflow::task_scheduler::stop

(

bool

wait_for_completion = true

)

Stop the scheduler service.

Gracefully stops the service, waiting for any in-progress tasks to complete.

Parameters

wait_for_completion

If true, waits for running tasks

Definition at line 182 of file task_scheduler.cpp.

                                                  {
    if (!running_.exchange(false)) {
        return;  // Already stopped
    }
 
    stop_requested_.store(true);
 
    // Wake up the scheduler thread
    cv_.notify_all();
 
    if (wait_for_completion && scheduler_thread_.joinable()) {
        scheduler_thread_.join();
    } else if (scheduler_thread_.joinable()) {
        scheduler_thread_.detach();
    }
 
    integration::logger_adapter::info("Task scheduler stopped");
}

References cv_, kcenon::pacs::integration::logger_adapter::info(), running_, scheduler_thread_, and stop_requested_.

Referenced by ~task_scheduler().

Here is the call graph for this function:

Here is the caller graph for this function:

trigger_task()

auto kcenon::pacs::workflow::task_scheduler::trigger_task

(

const task_id &

id

)

-> bool

Trigger immediate execution of a task.

Executes the task immediately, regardless of its schedule.

Parameters

id	Task ID to execute

Returns

true if task was found and triggered

Definition at line 492 of file task_scheduler.cpp.

                                                           {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    auto it = tasks_.find(id);
    if (it == tasks_.end()) {
        return false;
    }
 
    if (!it->second.enabled || it->second.state == task_state::running) {
        return false;
    }
 
    // Set next run to now to trigger immediate execution
    it->second.next_run_at = std::chrono::system_clock::now();
 
    // Wake up scheduler
    cv_.notify_one();
 
    integration::logger_adapter::info(
        "Triggered immediate execution task_id={} name={}",
        id, it->second.name);
 
    return true;
}

References kcenon::pacs::integration::logger_adapter::info(), and kcenon::pacs::workflow::running.

Here is the call graph for this function:

update_schedule()

auto kcenon::pacs::workflow::task_scheduler::update_schedule	(	const task_id &	id,
		const kcenon::pacs::workflow::schedule &	new_schedule ) -> bool

Update task schedule.

Parameters

id	Task ID
new_schedule	New schedule

Returns

true if task was found and updated

Definition at line 517 of file task_scheduler.cpp.

            {
    std::lock_guard<std::mutex> lock(tasks_mutex_);
 
    auto it = tasks_.find(id);
    if (it == tasks_.end()) {
        return false;
    }
 
    it->second.task_schedule = new_schedule;
    it->second.next_run_at = calculate_next_run(new_schedule);
    it->second.updated_at = std::chrono::system_clock::now();
 
    integration::logger_adapter::info(
        "Updated schedule for task task_id={} name={}",
        id, it->second.name);
 
    return true;
}

References kcenon::pacs::integration::logger_adapter::info().

Here is the call graph for this function:

update_stats()

void kcenon::pacs::workflow::task_scheduler::update_stats ( const task_execution_record & record )

private

Update statistics after execution.

Parameters

record

Execution record

Definition at line 1491 of file task_scheduler.cpp.

                                                                     {
    std::lock_guard<std::mutex> lock(stats_mutex_);
 
    ++stats_.total_executions;
 
    if (record.state == task_state::completed) {
        ++stats_.successful_executions;
    } else if (record.state == task_state::failed) {
        ++stats_.failed_executions;
    } else if (record.state == task_state::cancelled) {
        ++stats_.cancelled_executions;
    }
 
    // Update average and max execution time
    if (auto dur = record.duration()) {
        auto ms = dur->count();
 
        // Update max
        if (ms > stats_.max_execution_time.count()) {
            stats_.max_execution_time = std::chrono::milliseconds(ms);
        }
 
        // Update running average
        if (stats_.total_executions == 1) {
            stats_.avg_execution_time = std::chrono::milliseconds(ms);
        } else {
            auto current_avg = stats_.avg_execution_time.count();
            auto new_avg = current_avg +
                           (ms - current_avg) /
                           static_cast<int64_t>(stats_.total_executions);
            stats_.avg_execution_time = std::chrono::milliseconds(new_avg);
        }
    }
}

References kcenon::pacs::workflow::scheduler_stats::avg_execution_time, kcenon::pacs::workflow::cancelled, kcenon::pacs::workflow::scheduler_stats::cancelled_executions, kcenon::pacs::workflow::completed, kcenon::pacs::workflow::failed, kcenon::pacs::workflow::scheduler_stats::failed_executions, kcenon::pacs::workflow::scheduler_stats::max_execution_time, stats_, stats_mutex_, kcenon::pacs::workflow::scheduler_stats::successful_executions, and kcenon::pacs::workflow::scheduler_stats::total_executions.

Referenced by execute_cycle().

Here is the caller graph for this function:

Member Data Documentation

config_

task_scheduler_config kcenon::pacs::workflow::task_scheduler::config_

private

Service configuration.

Definition at line 677 of file task_scheduler.h.

Referenced by execute_cycle(), run_loop(), save_tasks(), set_error_callback(), set_task_complete_callback(), start(), task_scheduler(), task_scheduler(), task_scheduler(), and ~task_scheduler().

cv_

std::condition_variable kcenon::pacs::workflow::task_scheduler::cv_

private

Condition variable for sleep/wake.

Definition at line 686 of file task_scheduler.h.

Referenced by run_loop(), and stop().

database_

storage::index_database& kcenon::pacs::workflow::task_scheduler::database_

private

Reference to PACS index database.

Definition at line 665 of file task_scheduler.h.

execution_history_

std::map<task_id, std::vector<task_execution_record> > kcenon::pacs::workflow::task_scheduler::execution_history_

private

Execution history (task_id -> records)

Definition at line 701 of file task_scheduler.h.

Referenced by clear_history(), and record_execution().

executor_

std::shared_ptr<kcenon::common::interfaces::IExecutor> kcenon::pacs::workflow::task_scheduler::executor_

private

IExecutor for task execution (recommended, Issue #487)

Definition at line 674 of file task_scheduler.h.

file_storage_

storage::file_storage* kcenon::pacs::workflow::task_scheduler::file_storage_ {nullptr}

private

Optional reference to file storage.

Definition at line 668 of file task_scheduler.h.

{nullptr};

history_mutex_

std::mutex kcenon::pacs::workflow::task_scheduler::history_mutex_

mutableprivate

Mutex for execution history.

Definition at line 704 of file task_scheduler.h.

Referenced by clear_history(), and record_execution().

mutex_

std::mutex kcenon::pacs::workflow::task_scheduler::mutex_

mutableprivate

Mutex for thread synchronization.

Definition at line 683 of file task_scheduler.h.

Referenced by run_loop(), set_error_callback(), and set_task_complete_callback().

next_execution_id_

std::atomic<uint64_t> kcenon::pacs::workflow::task_scheduler::next_execution_id_ {1}

mutableprivate

Next execution ID counter.

Definition at line 722 of file task_scheduler.h.

{1};

Referenced by generate_execution_id().

next_task_id_

std::atomic<uint64_t> kcenon::pacs::workflow::task_scheduler::next_task_id_ {1}

mutableprivate

Next task ID counter.

Definition at line 719 of file task_scheduler.h.

{1};

Referenced by generate_task_id().

running_

std::atomic<bool> kcenon::pacs::workflow::task_scheduler::running_ {false}

private

Flag indicating scheduler is running.

Definition at line 692 of file task_scheduler.h.

{false};

Referenced by get_stats(), is_running(), start(), and stop().

running_count_

std::atomic<std::size_t> kcenon::pacs::workflow::task_scheduler::running_count_ {0}

private

Running task count.

Definition at line 707 of file task_scheduler.h.

{0};

Referenced by execute_cycle(), get_stats(), and running_count().

scheduler_thread_

std::thread kcenon::pacs::workflow::task_scheduler::scheduler_thread_

private

Background scheduler thread.

Definition at line 680 of file task_scheduler.h.

Referenced by start(), and stop().

start_time_

std::chrono::steady_clock::time_point kcenon::pacs::workflow::task_scheduler::start_time_

private

Start time for uptime calculation.

Definition at line 716 of file task_scheduler.h.

Referenced by get_stats(), and start().

stats_

scheduler_stats kcenon::pacs::workflow::task_scheduler::stats_

private

Scheduler statistics.

Definition at line 710 of file task_scheduler.h.

Referenced by execute_cycle(), get_stats(), and update_stats().

stats_mutex_

std::mutex kcenon::pacs::workflow::task_scheduler::stats_mutex_

mutableprivate

Mutex for statistics.

Definition at line 713 of file task_scheduler.h.

Referenced by execute_cycle(), get_stats(), and update_stats().

stop_requested_

std::atomic<bool> kcenon::pacs::workflow::task_scheduler::stop_requested_ {false}

private

Flag to signal shutdown.

Definition at line 689 of file task_scheduler.h.

{false};

Referenced by run_loop(), start(), and stop().

tasks_

std::map<task_id, scheduled_task> kcenon::pacs::workflow::task_scheduler::tasks_

private

All scheduled tasks (id -> task)

Definition at line 695 of file task_scheduler.h.

Referenced by execute_cycle(), get_stats(), list_tasks(), pending_count(), and serialize_tasks().

tasks_mutex_

std::mutex kcenon::pacs::workflow::task_scheduler::tasks_mutex_

mutableprivate

Mutex for tasks map.

Definition at line 698 of file task_scheduler.h.

Referenced by execute_cycle(), get_stats(), list_tasks(), pending_count(), and serialize_tasks().

thread_pool_

std::shared_ptr<kcenon::thread::thread_pool> kcenon::pacs::workflow::task_scheduler::thread_pool_

private

Thread pool for parallel task execution (legacy)

Definition at line 671 of file task_scheduler.h.

---

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

• include/kcenon/pacs/workflow/task_scheduler.h
• src/workflow/task_scheduler.cpp