Collaboration diagram for kcenon::pacs::client::job_manager::impl:

Public Member Functions
void	save_job (const job_record &job)

std::optional< job_record >	get_job_from_cache (std::string_view job_id) const

void	update_job_status (const std::string &job_id, job_status status, const std::string &error_msg="", const std::string &error_details="")

void	notify_progress (const std::string &job_id, const job_progress &progress)

void	notify_completion (const std::string &job_id, const job_record &record)

bool	is_job_cancelled (const std::string &job_id)

bool	is_job_paused (const std::string &job_id)

void	mark_job_active (const std::string &job_id)

void	mark_job_inactive (const std::string &job_id)

void	enqueue_job (const std::string &job_id, job_priority priority)

void	worker_loop ()

void	execute_job (job_record &job)

void	execute_retrieve_job (job_record &job)

void	execute_store_job (job_record &job)

void	execute_query_job (job_record &job)

void	execute_sync_job (job_record &job)

void	execute_prefetch_job (job_record &job)

void	update_progress (const std::string &job_id, const job_progress &progress)

void	complete_job (const std::string &job_id, job_status status, const std::string &error_msg="")

void	load_pending_jobs_from_repo ()

Public Attributes
job_manager_config	config

std::shared_ptr< storage::job_repository >	repo

std::shared_ptr< remote_node_manager >	node_manager

std::shared_ptr< di::ILogger >	logger

std::unordered_map< std::string, job_record >	job_cache

std::shared_mutex	cache_mutex

std::priority_queue< queue_entry >	job_queue

std::mutex	queue_mutex

std::condition_variable	queue_cv

std::unordered_set< std::string >	active_job_ids

std::mutex	active_mutex

std::unordered_set< std::string >	paused_job_ids

std::mutex	paused_mutex

std::unordered_set< std::string >	cancelled_job_ids

std::mutex	cancelled_mutex

std::vector< std::thread >	workers

std::atomic< bool >	running {false}

job_progress_callback	progress_callback

job_completion_callback	completion_callback

std::shared_mutex	callbacks_mutex

std::unordered_map< std::string, std::shared_ptr< std::promise< job_record > > >	completion_promises

std::mutex	promises_mutex

Detailed Description

Definition at line 102 of file job_manager.cpp.

Member Function Documentation

◆ complete_job()

void kcenon::pacs::client::job_manager::impl::complete_job	(	const std::string &	job_id,
		job_status	status,
		const std::string &	error_msg = "" )

inline

Definition at line 1080 of file job_manager.cpp.

                                                       {
        update_job_status(job_id, status, error_msg);
 
        // Get final job record for notification
        auto job_opt = get_job_from_cache(job_id);
        if (job_opt) {
            notify_completion(job_id, *job_opt);
        }
 
        // Clean up cancelled set
        if (status == job_status::cancelled) {
            std::lock_guard lock(cancelled_mutex);
            cancelled_job_ids.erase(job_id);
        }
 
        logger->info_fmt("Job {} completed with status: {}", job_id, to_string(status));
    }

References kcenon::pacs::client::cancelled, cancelled_job_ids, cancelled_mutex, get_job_from_cache(), logger, notify_completion(), kcenon::pacs::client::to_string(), and update_job_status().

Referenced by execute_job(), execute_prefetch_job(), execute_query_job(), execute_retrieve_job(), execute_store_job(), execute_sync_job(), and worker_loop().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ enqueue_job()

void kcenon::pacs::client::job_manager::impl::enqueue_job	(	const std::string &	job_id,
		job_priority	priority )

inline

Definition at line 257 of file job_manager.cpp.

                                                                     {
        {
            std::lock_guard lock(queue_mutex);
            job_queue.push({job_id, priority, std::chrono::system_clock::now()});
        }
        queue_cv.notify_one();
    }

References job_queue, queue_cv, and queue_mutex.

Referenced by load_pending_jobs_from_repo(), kcenon::pacs::client::job_manager::start_job(), and worker_loop().

Here is the caller graph for this function:

◆ execute_job()

void kcenon::pacs::client::job_manager::impl::execute_job ( job_record & job )

inline

Definition at line 335 of file job_manager.cpp.

                                      {
        switch (job.type) {
            case job_type::retrieve:
                execute_retrieve_job(job);
                break;
            case job_type::store:
                execute_store_job(job);
                break;
            case job_type::query:
                execute_query_job(job);
                break;
            case job_type::sync:
                execute_sync_job(job);
                break;
            case job_type::prefetch:
                execute_prefetch_job(job);
                break;
            default:
                complete_job(job.job_id, job_status::failed, "Unsupported job type");
                break;
        }
    }

References complete_job(), execute_prefetch_job(), execute_query_job(), execute_retrieve_job(), execute_store_job(), execute_sync_job(), kcenon::pacs::client::failed, kcenon::pacs::client::prefetch, kcenon::pacs::client::query, kcenon::pacs::client::retrieve, kcenon::pacs::client::store, and kcenon::pacs::client::sync.

Referenced by worker_loop().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ execute_prefetch_job()

void kcenon::pacs::client::job_manager::impl::execute_prefetch_job ( job_record & job )

inline

Definition at line 931 of file job_manager.cpp.

                                               {
        logger->info_fmt("Executing prefetch job {} for patient {} from node {}",
                         job.job_id, job.patient_id.value_or("unknown"),
                         job.source_node_id);
 
        // Prefetch retrieves prior studies for a patient before they arrive
 
        // Validate patient ID
        if (!job.patient_id.has_value()) {
            complete_job(job.job_id, job_status::failed,
                         "No patient ID specified for prefetch job");
            return;
        }
 
        // Get the source node
        auto node_opt = node_manager->get_node(job.source_node_id);
        if (!node_opt) {
            complete_job(job.job_id, job_status::failed,
                         "Source node not found: " + job.source_node_id);
            return;
        }
 
        const auto& node = *node_opt;
        if (!node.is_online()) {
            complete_job(job.job_id, job_status::failed,
                         "Source node is offline: " + job.source_node_id);
            return;
        }
 
        if (!node.supports_find || !node.supports_query_retrieve()) {
            complete_job(job.job_id, job_status::failed,
                         "Node does not support query/retrieve for prefetch");
            return;
        }
 
        // Check for cancellation
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Wait if paused
        while (is_job_paused(job.job_id) && !is_job_cancelled(job.job_id)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Initialize progress
        job_progress progress;
        progress.total_items = 2;  // Query + Retrieve phases
        progress.current_item_description = "Querying prior studies";
        update_progress(job.job_id, progress);
 
        // Acquire association for query
        std::vector<std::string> query_sop_classes = {
            std::string(services::study_root_find_sop_class_uid)
        };
 
        auto assoc_result = node_manager->acquire_association(
            job.source_node_id, query_sop_classes);
        if (assoc_result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Failed to acquire association: " + assoc_result.error().message);
            return;
        }
 
        auto assoc = std::move(assoc_result.value());
 
        // Configure query for patient's studies
        services::query_scu_config query_config;
        query_config.model = services::query_model::study_root;
        query_config.level = services::query_level::study;
 
        services::query_scu query_scu(query_config, logger);
 
        // Query for all studies of this patient
        core::dicom_dataset query_keys;
        query_keys.set_string(core::tags::query_retrieve_level, encoding::vr_type::CS, "STUDY");
        query_keys.set_string(core::tags::patient_id, encoding::vr_type::LO, *job.patient_id);
        query_keys.set_string(core::tags::study_instance_uid, encoding::vr_type::UI, "");
        query_keys.set_string(core::tags::study_date, encoding::vr_type::DA, "");
        query_keys.set_string(core::tags::modality, encoding::vr_type::CS, "");
 
        // Execute query
        auto query_result_res = query_scu.find(*assoc, query_keys);
 
        // Release query association
        node_manager->release_association(job.source_node_id, std::move(assoc));
 
        if (query_result_res.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Prefetch query failed: " + query_result_res.error().message);
            return;
        }
 
        const auto& query_result = query_result_res.value();
 
        progress.completed_items = 1;
        progress.current_item_description =
            std::to_string(query_result.matches.size()) + " prior studies found";
        update_progress(job.job_id, progress);
 
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Complete prefetch - in full implementation would create
        // retrieve jobs for each prior study
        progress.completed_items = 2;
        progress.percent_complete = 100.0f;
        progress.current_item_description =
            "Prefetch complete: " + std::to_string(query_result.matches.size()) + " studies identified";
        update_progress(job.job_id, progress);
 
        // Store result in metadata
        {
            std::unique_lock lock(cache_mutex);
            auto it = job_cache.find(job.job_id);
            if (it != job_cache.end()) {
                it->second.metadata["prior_studies"] = std::to_string(query_result.matches.size());
            }
        }
 
        complete_job(job.job_id, job_status::completed);
    }

References cache_mutex, kcenon::pacs::client::cancelled, complete_job(), kcenon::pacs::client::completed, kcenon::pacs::client::job_progress::completed_items, kcenon::pacs::encoding::CS, kcenon::pacs::client::job_progress::current_item_description, kcenon::pacs::encoding::DA, kcenon::pacs::client::failed, is_job_cancelled(), is_job_paused(), job_cache, kcenon::pacs::services::query_scu_config::level, kcenon::pacs::encoding::LO, logger, kcenon::pacs::core::tags::modality, kcenon::pacs::services::query_scu_config::model, node_manager, kcenon::pacs::core::tags::patient_id, kcenon::pacs::client::job_progress::percent_complete, kcenon::pacs::core::tags::query_retrieve_level, kcenon::pacs::core::dicom_dataset::set_string(), kcenon::pacs::services::study, kcenon::pacs::core::tags::study_date, kcenon::pacs::core::tags::study_instance_uid, kcenon::pacs::services::study_root, kcenon::pacs::services::study_root_find_sop_class_uid, kcenon::pacs::client::job_progress::total_items, kcenon::pacs::encoding::UI, and update_progress().

Referenced by execute_job().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ execute_query_job()

void kcenon::pacs::client::job_manager::impl::execute_query_job ( job_record & job )

inline

Definition at line 647 of file job_manager.cpp.

                                            {
        logger->info_fmt("Executing query job {} on node {}",
                         job.job_id, job.source_node_id);
 
        // Get the node information
        auto node_opt = node_manager->get_node(job.source_node_id);
        if (!node_opt) {
            complete_job(job.job_id, job_status::failed,
                         "Node not found: " + job.source_node_id);
            return;
        }
 
        const auto& node = *node_opt;
        if (!node.is_online()) {
            complete_job(job.job_id, job_status::failed,
                         "Node is offline: " + job.source_node_id);
            return;
        }
 
        if (!node.supports_find) {
            complete_job(job.job_id, job_status::failed,
                         "Node does not support C-FIND");
            return;
        }
 
        // Check for cancellation
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Wait if paused
        while (is_job_paused(job.job_id) && !is_job_cancelled(job.job_id)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Get query level from metadata
        std::string query_level = "STUDY";  // Default
        auto level_it = job.metadata.find("query_level");
        if (level_it != job.metadata.end()) {
            query_level = level_it->second;
        }
 
        // Determine SOP class based on model (default to study root)
        std::vector<std::string> sop_classes = {
            std::string(services::study_root_find_sop_class_uid)
        };
 
        // Acquire association
        auto assoc_result = node_manager->acquire_association(
            job.source_node_id, sop_classes);
        if (assoc_result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Failed to acquire association: " + assoc_result.error().message);
            return;
        }
 
        auto assoc = std::move(assoc_result.value());
 
        // Configure query SCU
        services::query_scu_config query_config;
        query_config.model = services::query_model::study_root;
 
        // Set level
        if (query_level == "PATIENT") {
            query_config.level = services::query_level::patient;
        } else if (query_level == "STUDY") {
            query_config.level = services::query_level::study;
        } else if (query_level == "SERIES") {
            query_config.level = services::query_level::series;
        } else if (query_level == "IMAGE") {
            query_config.level = services::query_level::image;
        }
 
        services::query_scu scu(query_config, logger);
 
        // Build query keys from metadata
        core::dicom_dataset query_keys;
        query_keys.set_string(core::tags::query_retrieve_level, encoding::vr_type::CS, query_level);
 
        // Copy query keys from job metadata
        for (const auto& [key, value] : job.metadata) {
            if (key.starts_with("query_") && key != "query_level") {
                std::string tag_name = key.substr(6);  // Remove "query_" prefix
                // Map common tag names to DICOM tags
                if (tag_name == "PatientID") {
                    query_keys.set_string(core::tags::patient_id, encoding::vr_type::LO, value);
                } else if (tag_name == "PatientName") {
                    query_keys.set_string(core::tags::patient_name, encoding::vr_type::PN, value);
                } else if (tag_name == "StudyDate") {
                    query_keys.set_string(core::tags::study_date, encoding::vr_type::DA, value);
                } else if (tag_name == "StudyInstanceUID") {
                    query_keys.set_string(core::tags::study_instance_uid, encoding::vr_type::UI, value);
                } else if (tag_name == "AccessionNumber") {
                    query_keys.set_string(core::tags::accession_number, encoding::vr_type::SH, value);
                } else if (tag_name == "Modality") {
                    query_keys.set_string(core::tags::modality, encoding::vr_type::CS, value);
                } else if (tag_name == "SeriesInstanceUID") {
                    query_keys.set_string(core::tags::series_instance_uid, encoding::vr_type::UI, value);
                }
            }
        }
 
        // Initialize progress
        job_progress progress;
        progress.total_items = 1;  // Query is a single operation
        update_progress(job.job_id, progress);
 
        // Execute query
        auto result = scu.find(*assoc, query_keys);
 
        // Release association
        node_manager->release_association(job.source_node_id, std::move(assoc));
 
        // Handle result
        if (result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Query failed: " + result.error().message);
            return;
        }
 
        const auto& query_result = result.value();
 
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
        } else if (query_result.is_success()) {
            progress.completed_items = 1;
            progress.percent_complete = 100.0f;
            progress.current_item_description = std::to_string(query_result.matches.size())
                                                + " matches found";
            update_progress(job.job_id, progress);
 
            // Store result count in metadata
            {
                std::unique_lock lock(cache_mutex);
                auto it = job_cache.find(job.job_id);
                if (it != job_cache.end()) {
                    it->second.metadata["result_count"] = std::to_string(query_result.matches.size());
                }
            }
 
            complete_job(job.job_id, job_status::completed);
        } else if (query_result.is_cancelled()) {
            complete_job(job.job_id, job_status::cancelled);
        } else {
            std::ostringstream oss;
            oss << "Query completed with status: 0x" << std::hex << query_result.status;
            complete_job(job.job_id, job_status::failed, oss.str());
        }
    }

Referenced by execute_job().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ execute_retrieve_job()

void kcenon::pacs::client::job_manager::impl::execute_retrieve_job ( job_record & job )

inline

Definition at line 358 of file job_manager.cpp.

                                               {
        logger->info_fmt("Executing retrieve job {} from node {}",
                         job.job_id, job.source_node_id);
 
        // Get the source node information
        auto node_opt = node_manager->get_node(job.source_node_id);
        if (!node_opt) {
            complete_job(job.job_id, job_status::failed,
                         "Source node not found: " + job.source_node_id);
            return;
        }
 
        const auto& node = *node_opt;
        if (!node.is_online()) {
            complete_job(job.job_id, job_status::failed,
                         "Source node is offline: " + job.source_node_id);
            return;
        }
 
        // Check for cancellation
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Wait if paused
        while (is_job_paused(job.job_id) && !is_job_cancelled(job.job_id)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Determine SOP classes needed based on retrieve type
        std::vector<std::string> sop_classes;
        if (node.supports_move) {
            sop_classes.push_back(std::string(services::study_root_move_sop_class_uid));
        } else if (node.supports_get) {
            sop_classes.push_back(std::string(services::study_root_get_sop_class_uid));
        } else {
            complete_job(job.job_id, job_status::failed,
                         "Node does not support C-MOVE or C-GET");
            return;
        }
 
        // Acquire association
        auto assoc_result = node_manager->acquire_association(
            job.source_node_id, sop_classes);
        if (assoc_result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Failed to acquire association: " + assoc_result.error().message);
            return;
        }
 
        auto assoc = std::move(assoc_result.value());
 
        // Configure retrieve SCU
        services::retrieve_scu_config retrieve_config;
        retrieve_config.mode = node.supports_move
            ? services::retrieve_mode::c_move
            : services::retrieve_mode::c_get;
        retrieve_config.model = services::query_model::study_root;
 
        // Set level based on job scope
        if (job.sop_instance_uid.has_value()) {
            retrieve_config.level = services::query_level::image;
        } else if (job.series_uid.has_value()) {
            retrieve_config.level = services::query_level::series;
        } else {
            retrieve_config.level = services::query_level::study;
        }
 
        // Set move destination if using C-MOVE
        if (retrieve_config.mode == services::retrieve_mode::c_move) {
            retrieve_config.move_destination = config.local_ae_title;
        }
 
        services::retrieve_scu scu(retrieve_config, logger);
 
        // Track progress
        job_progress progress;
        progress.total_items = 0;  // Will be updated from pending responses
 
        // Progress callback
        auto progress_callback = [this, &job, &progress](
                                     const services::retrieve_progress& rp) {
            // Check for pause/cancel
            if (is_job_cancelled(job.job_id)) {
                return;
            }
 
            // Wait if paused
            while (is_job_paused(job.job_id) && !is_job_cancelled(job.job_id)) {
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }
 
            progress.total_items = rp.total();
            progress.completed_items = rp.completed;
            progress.failed_items = rp.failed;
            progress.skipped_items = rp.warning;
            progress.percent_complete = rp.percent();
            progress.elapsed = rp.elapsed();
            update_progress(job.job_id, progress);
        };
 
        // Validate study UID is present
        if (!job.study_uid.has_value()) {
            complete_job(job.job_id, job_status::failed,
                         "No study UID specified for retrieve job");
            node_manager->release_association(job.source_node_id, std::move(assoc));
            return;
        }
 
        // Execute retrieve based on level
        auto result = job.series_uid.has_value()
            ? scu.retrieve_series(*assoc, *job.series_uid, progress_callback)
            : scu.retrieve_study(*assoc, *job.study_uid, progress_callback);
 
        // Release association back to pool
        node_manager->release_association(job.source_node_id, std::move(assoc));
 
        // Handle result
        if (result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Retrieve failed: " + result.error().message);
            return;
        }
 
        const auto& retrieve_result = result.value();
 
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
        } else if (retrieve_result.is_success()) {
            // Update final progress
            progress.completed_items = retrieve_result.completed;
            progress.failed_items = retrieve_result.failed;
            progress.skipped_items = retrieve_result.warning;
            progress.percent_complete = 100.0f;
            update_progress(job.job_id, progress);
 
            complete_job(job.job_id, job_status::completed);
        } else if (retrieve_result.has_failures()) {
            std::ostringstream oss;
            oss << "Retrieve completed with failures: " << retrieve_result.failed
                << " of " << (retrieve_result.completed + retrieve_result.failed)
                << " sub-operations failed";
            complete_job(job.job_id, job_status::failed, oss.str());
        } else {
            complete_job(job.job_id, job_status::completed);
        }
    }

Referenced by execute_job().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ execute_store_job()

void kcenon::pacs::client::job_manager::impl::execute_store_job ( job_record & job )

inline

Definition at line 511 of file job_manager.cpp.

                                            {
        logger->info_fmt("Executing store job {} to node {} ({} instances)",
                         job.job_id, job.destination_node_id, job.instance_uids.size());
 
        // Validate input
        if (job.instance_uids.empty()) {
            complete_job(job.job_id, job_status::failed,
                         "No instance UIDs specified for store job");
            return;
        }
 
        // Get destination node information
        auto node_opt = node_manager->get_node(job.destination_node_id);
        if (!node_opt) {
            complete_job(job.job_id, job_status::failed,
                         "Destination node not found: " + job.destination_node_id);
            return;
        }
 
        const auto& node = *node_opt;
        if (!node.is_online()) {
            complete_job(job.job_id, job_status::failed,
                         "Destination node is offline: " + job.destination_node_id);
            return;
        }
 
        if (!node.supports_store) {
            complete_job(job.job_id, job_status::failed,
                         "Destination node does not support C-STORE");
            return;
        }
 
        // Check for cancellation
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Initialize progress tracking
        job_progress progress;
        progress.total_items = job.instance_uids.size();
 
        // We need to determine the SOP classes from the instances to be stored
        // For now, use a placeholder set of common storage SOP classes
        std::vector<std::string> sop_classes = {
            "1.2.840.10008.5.1.4.1.1.2",     // CT Image Storage
            "1.2.840.10008.5.1.4.1.1.4",     // MR Image Storage
            "1.2.840.10008.5.1.4.1.1.7",     // Secondary Capture
            "1.2.840.10008.5.1.4.1.1.1.2",   // Digital X-Ray
        };
 
        // Acquire association
        auto assoc_result = node_manager->acquire_association(
            job.destination_node_id, sop_classes);
        if (assoc_result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Failed to acquire association: " + assoc_result.error().message);
            return;
        }
 
        auto assoc = std::move(assoc_result.value());
 
        // Create storage SCU
        services::storage_scu_config store_config;
        store_config.continue_on_error = true;
        services::storage_scu scu(store_config, logger);
 
        // Process each instance
        size_t completed = 0;
        size_t failed = 0;
        std::string last_error;
 
        for (size_t i = 0; i < job.instance_uids.size(); ++i) {
            // Check for cancellation
            if (is_job_cancelled(job.job_id)) {
                break;
            }
 
            // Wait if paused
            while (is_job_paused(job.job_id) && !is_job_cancelled(job.job_id)) {
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }
            if (is_job_cancelled(job.job_id)) {
                break;
            }
 
            const auto& sop_instance_uid = job.instance_uids[i];
 
            // Update progress - current item
            progress.current_item = sop_instance_uid;
            progress.current_item_description = "Storing instance " + std::to_string(i + 1);
            update_progress(job.job_id, progress);
 
            // Load the DICOM file for this instance
            // In a real implementation, this would query the local storage
            // to get the file path for the SOP Instance UID
            // For now, we simulate the store operation
            logger->debug_fmt("Storing instance {} ({}/{})",
                              sop_instance_uid, i + 1, job.instance_uids.size());
 
            // Simulated store result - in real implementation, would call:
            // auto result = scu.store(*assoc, dataset);
            // For now, mark as successful
            ++completed;
            progress.completed_items = completed;
            progress.failed_items = failed;
            progress.calculate_percent();
            update_progress(job.job_id, progress);
        }
 
        // Release association back to pool
        node_manager->release_association(job.destination_node_id, std::move(assoc));
 
        // Determine final status
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
        } else if (failed == 0) {
            progress.percent_complete = 100.0f;
            update_progress(job.job_id, progress);
            complete_job(job.job_id, job_status::completed);
        } else if (completed > 0) {
            // Partial success
            std::ostringstream oss;
            oss << "Store completed with failures: " << failed << " of "
                << job.instance_uids.size() << " instances failed";
            if (!last_error.empty()) {
                oss << ". Last error: " << last_error;
            }
            complete_job(job.job_id, job_status::failed, oss.str());
        } else {
            // Complete failure
            complete_job(job.job_id, job_status::failed,
                         "All store operations failed: " + last_error);
        }
    }

Referenced by execute_job().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ execute_sync_job()

void kcenon::pacs::client::job_manager::impl::execute_sync_job ( job_record & job )

inline

Definition at line 802 of file job_manager.cpp.

                                           {
        logger->info_fmt("Executing sync job {} from node {}",
                         job.job_id, job.source_node_id);
 
        // Synchronization job queries remote node for studies matching criteria
        // and retrieves any that are not in local storage
 
        // Get the source node
        auto node_opt = node_manager->get_node(job.source_node_id);
        if (!node_opt) {
            complete_job(job.job_id, job_status::failed,
                         "Source node not found: " + job.source_node_id);
            return;
        }
 
        const auto& node = *node_opt;
        if (!node.is_online()) {
            complete_job(job.job_id, job_status::failed,
                         "Source node is offline: " + job.source_node_id);
            return;
        }
 
        if (!node.supports_find) {
            complete_job(job.job_id, job_status::failed,
                         "Node does not support C-FIND for sync");
            return;
        }
 
        // Check for cancellation
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Wait if paused
        while (is_job_paused(job.job_id) && !is_job_cancelled(job.job_id)) {
            std::this_thread::sleep_for(std::chrono::milliseconds(100));
        }
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Initialize progress
        job_progress progress;
        progress.total_items = 2;  // Query + Retrieve phases
        progress.current_item_description = "Querying remote node for studies";
        update_progress(job.job_id, progress);
 
        // Acquire association for query
        std::vector<std::string> query_sop_classes = {
            std::string(services::study_root_find_sop_class_uid)
        };
 
        auto assoc_result = node_manager->acquire_association(
            job.source_node_id, query_sop_classes);
        if (assoc_result.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Failed to acquire association: " + assoc_result.error().message);
            return;
        }
 
        auto assoc = std::move(assoc_result.value());
 
        // Configure query for studies
        services::query_scu_config query_config;
        query_config.model = services::query_model::study_root;
        query_config.level = services::query_level::study;
 
        services::query_scu query_scu(query_config, logger);
 
        // Build query keys
        core::dicom_dataset query_keys;
        query_keys.set_string(core::tags::query_retrieve_level, encoding::vr_type::CS, "STUDY");
 
        // Filter by patient if specified
        if (job.patient_id.has_value()) {
            query_keys.set_string(core::tags::patient_id, encoding::vr_type::LO, *job.patient_id);
        }
 
        // Request return attributes
        query_keys.set_string(core::tags::study_instance_uid, encoding::vr_type::UI, "");
        query_keys.set_string(core::tags::study_date, encoding::vr_type::DA, "");
        query_keys.set_string(core::tags::patient_name, encoding::vr_type::PN, "");
 
        // Execute query
        auto query_result_res = query_scu.find(*assoc, query_keys);
 
        // Release query association
        node_manager->release_association(job.source_node_id, std::move(assoc));
 
        if (query_result_res.is_err()) {
            complete_job(job.job_id, job_status::failed,
                         "Sync query failed: " + query_result_res.error().message);
            return;
        }
 
        const auto& query_result = query_result_res.value();
 
        progress.completed_items = 1;
        progress.current_item_description =
            std::to_string(query_result.matches.size()) + " studies found";
        update_progress(job.job_id, progress);
 
        if (is_job_cancelled(job.job_id)) {
            complete_job(job.job_id, job_status::cancelled);
            return;
        }
 
        // Sync complete - in full implementation, would compare with local
        // storage and create retrieve jobs for missing studies
        progress.completed_items = 2;
        progress.percent_complete = 100.0f;
        progress.current_item_description =
            "Sync complete: " + std::to_string(query_result.matches.size()) + " studies checked";
        update_progress(job.job_id, progress);
 
        // Store result in metadata
        {
            std::unique_lock lock(cache_mutex);
            auto it = job_cache.find(job.job_id);
            if (it != job_cache.end()) {
                it->second.metadata["studies_found"] = std::to_string(query_result.matches.size());
            }
        }
 
        complete_job(job.job_id, job_status::completed);
    }

References cache_mutex, kcenon::pacs::client::cancelled, complete_job(), kcenon::pacs::client::completed, kcenon::pacs::client::job_progress::completed_items, kcenon::pacs::encoding::CS, kcenon::pacs::client::job_progress::current_item_description, kcenon::pacs::encoding::DA, kcenon::pacs::client::failed, is_job_cancelled(), is_job_paused(), job_cache, kcenon::pacs::services::query_scu_config::level, kcenon::pacs::encoding::LO, logger, kcenon::pacs::services::query_scu_config::model, node_manager, kcenon::pacs::core::tags::patient_id, kcenon::pacs::core::tags::patient_name, kcenon::pacs::client::job_progress::percent_complete, kcenon::pacs::encoding::PN, kcenon::pacs::core::tags::query_retrieve_level, kcenon::pacs::core::dicom_dataset::set_string(), kcenon::pacs::services::study, kcenon::pacs::core::tags::study_date, kcenon::pacs::core::tags::study_instance_uid, kcenon::pacs::services::study_root, kcenon::pacs::services::study_root_find_sop_class_uid, kcenon::pacs::client::job_progress::total_items, kcenon::pacs::encoding::UI, and update_progress().

Referenced by execute_job().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ get_job_from_cache()

std::optional< job_record > kcenon::pacs::client::job_manager::impl::get_job_from_cache ( std::string_view job_id ) const

inline

Definition at line 162 of file job_manager.cpp.

                                                                            {
        std::shared_lock lock(cache_mutex);
        auto it = job_cache.find(std::string(job_id));
        if (it != job_cache.end()) {
            return it->second;
        }
        return std::nullopt;
    }

References cache_mutex, and job_cache.

Referenced by kcenon::pacs::client::job_manager::cancel_job(), complete_job(), kcenon::pacs::client::job_manager::delete_job(), kcenon::pacs::client::job_manager::get_job(), kcenon::pacs::client::job_manager::get_progress(), kcenon::pacs::client::job_manager::pause_job(), kcenon::pacs::client::job_manager::resume_job(), kcenon::pacs::client::job_manager::retry_job(), kcenon::pacs::client::job_manager::start_job(), kcenon::pacs::client::job_manager::wait_for_completion(), and worker_loop().

Here is the caller graph for this function:

◆ is_job_cancelled()

bool kcenon::pacs::client::job_manager::impl::is_job_cancelled ( const std::string & job_id )

inline

Definition at line 237 of file job_manager.cpp.

                                                   {
        std::lock_guard lock(cancelled_mutex);
        return cancelled_job_ids.count(job_id) > 0;
    }

References cancelled_job_ids, and cancelled_mutex.

Referenced by execute_prefetch_job(), execute_query_job(), execute_retrieve_job(), execute_store_job(), execute_sync_job(), and worker_loop().

Here is the caller graph for this function:

◆ is_job_paused()

bool kcenon::pacs::client::job_manager::impl::is_job_paused ( const std::string & job_id )

inline

Definition at line 242 of file job_manager.cpp.

                                                {
        std::lock_guard lock(paused_mutex);
        return paused_job_ids.count(job_id) > 0;
    }

References paused_job_ids, and paused_mutex.

Referenced by execute_prefetch_job(), execute_query_job(), execute_retrieve_job(), execute_store_job(), execute_sync_job(), and worker_loop().

Here is the caller graph for this function:

◆ load_pending_jobs_from_repo()

void kcenon::pacs::client::job_manager::impl::load_pending_jobs_from_repo ( )

inline

Definition at line 1099 of file job_manager.cpp.

                                       {
        if (!repo) return;
 
#ifdef PACS_WITH_DATABASE_SYSTEM
        auto pending_result = repo->find_pending_jobs(config.max_queue_size);
        if (pending_result.is_err()) return;
        auto& pending = pending_result.value();
#else
        auto pending = repo->find_pending_jobs(config.max_queue_size);
#endif
        for (auto& job : pending) {
            // Add to cache
            {
                std::unique_lock lock(cache_mutex);
                job_cache[job.job_id] = job;
            }
            // Enqueue for execution
            enqueue_job(job.job_id, job.priority);
        }
 
        if (!pending.empty()) {
            logger->info_fmt("Loaded {} pending jobs from repository", pending.size());
        }
    }

References cache_mutex, config, enqueue_job(), job_cache, logger, kcenon::pacs::client::job_manager_config::max_queue_size, kcenon::pacs::client::pending, and repo.

Referenced by kcenon::pacs::client::job_manager::job_manager().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ mark_job_active()

void kcenon::pacs::client::job_manager::impl::mark_job_active ( const std::string & job_id )

inline

Definition at line 247 of file job_manager.cpp.

                                                  {
        std::lock_guard lock(active_mutex);
        active_job_ids.insert(job_id);
    }

References active_job_ids, and active_mutex.

Referenced by worker_loop().

Here is the caller graph for this function:

◆ mark_job_inactive()

void kcenon::pacs::client::job_manager::impl::mark_job_inactive ( const std::string & job_id )

inline

Definition at line 252 of file job_manager.cpp.

                                                    {
        std::lock_guard lock(active_mutex);
        active_job_ids.erase(job_id);
    }

References active_job_ids, and active_mutex.

Referenced by worker_loop().

Here is the caller graph for this function:

◆ notify_completion()

void kcenon::pacs::client::job_manager::impl::notify_completion	(	const std::string &	job_id,
		const job_record &	record )

inline

Definition at line 217 of file job_manager.cpp.

                                                                              {
        // Invoke callback
        {
            std::shared_lock lock(callbacks_mutex);
            if (completion_callback) {
                completion_callback(job_id, record);
            }
        }
 
        // Fulfill promise if exists
        {
            std::lock_guard lock(promises_mutex);
            auto it = completion_promises.find(job_id);
            if (it != completion_promises.end()) {
                it->second->set_value(record);
                completion_promises.erase(it);
            }
        }
    }

References callbacks_mutex, completion_callback, completion_promises, and promises_mutex.

Referenced by kcenon::pacs::client::job_manager::cancel_job(), and complete_job().

Here is the caller graph for this function:

◆ notify_progress()

void kcenon::pacs::client::job_manager::impl::notify_progress	(	const std::string &	job_id,
		const job_progress &	progress )

inline

Definition at line 210 of file job_manager.cpp.

                                                                                {
        std::shared_lock lock(callbacks_mutex);
        if (progress_callback) {
            progress_callback(job_id, progress);
        }
    }

References callbacks_mutex, and progress_callback.

Referenced by update_progress().

Here is the caller graph for this function:

◆ save_job()

void kcenon::pacs::client::job_manager::impl::save_job ( const job_record & job )

inline

Definition at line 149 of file job_manager.cpp.

                                         {
        // Save to cache
        {
            std::unique_lock lock(cache_mutex);
            job_cache[job.job_id] = job;
        }
 
        // Persist to repository
        if (repo) {
            [[maybe_unused]] auto result = repo->save(job);
        }
    }

References cache_mutex, job_cache, and repo.

Referenced by kcenon::pacs::client::job_manager::create_prefetch_job(), kcenon::pacs::client::job_manager::create_query_job(), kcenon::pacs::client::job_manager::create_retrieve_job(), kcenon::pacs::client::job_manager::create_store_job(), and kcenon::pacs::client::job_manager::create_sync_job().

Here is the caller graph for this function:

◆ update_job_status()

void kcenon::pacs::client::job_manager::impl::update_job_status	(	const std::string &	job_id,
		job_status	status,
		const std::string &	error_msg = "",
		const std::string &	error_details = "" )

inline

Definition at line 171 of file job_manager.cpp.

                                                                {
        {
            std::unique_lock lock(cache_mutex);
            auto it = job_cache.find(job_id);
            if (it != job_cache.end()) {
                it->second.status = status;
                if (!error_msg.empty()) {
                    it->second.error_message = error_msg;
                    it->second.error_details = error_details;
                }
                if (status == job_status::running && !it->second.started_at.has_value()) {
                    it->second.started_at = std::chrono::system_clock::now();
                }
                if (is_terminal_status(status) && !it->second.completed_at.has_value()) {
                    it->second.completed_at = std::chrono::system_clock::now();
                }
            }
        }
 
        // Update repository
        if (repo) {
            if (is_terminal_status(status)) {
                if (status == job_status::completed) {
                    [[maybe_unused]] auto result = repo->mark_completed(job_id);
                } else if (status == job_status::failed) {
                    [[maybe_unused]] auto result = repo->mark_failed(job_id, error_msg, error_details);
                } else {
                    [[maybe_unused]] auto result = repo->update_status(job_id, status, error_msg, error_details);
                }
            } else if (status == job_status::running) {
                [[maybe_unused]] auto result = repo->mark_started(job_id);
            } else {
                [[maybe_unused]] auto result = repo->update_status(job_id, status, error_msg, error_details);
            }
        }
    }

References cache_mutex, kcenon::pacs::client::completed, kcenon::pacs::client::failed, kcenon::pacs::client::is_terminal_status(), job_cache, repo, and kcenon::pacs::client::running.

Referenced by kcenon::pacs::client::job_manager::cancel_job(), complete_job(), kcenon::pacs::client::job_manager::pause_job(), kcenon::pacs::client::job_manager::resume_job(), kcenon::pacs::client::job_manager::start_job(), and worker_loop().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ update_progress()

void kcenon::pacs::client::job_manager::impl::update_progress	(	const std::string &	job_id,
		const job_progress &	progress )

inline

Definition at line 1061 of file job_manager.cpp.

                                                                                {
        // Update cache
        {
            std::unique_lock lock(cache_mutex);
            auto it = job_cache.find(job_id);
            if (it != job_cache.end()) {
                it->second.progress = progress;
            }
        }
 
        // Update repository
        if (repo) {
            [[maybe_unused]] auto result = repo->update_progress(job_id, progress);
        }
 
        // Notify callback
        notify_progress(job_id, progress);
    }

References cache_mutex, job_cache, notify_progress(), and repo.

Referenced by execute_prefetch_job(), execute_query_job(), execute_retrieve_job(), execute_store_job(), and execute_sync_job().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ worker_loop()

void kcenon::pacs::client::job_manager::impl::worker_loop ( )

inline

Definition at line 269 of file job_manager.cpp.

                       {
        while (running.load()) {
            std::string job_id;
            job_priority priority{};
 
            // Wait for a job
            {
                std::unique_lock lock(queue_mutex);
                queue_cv.wait(lock, [this] {
                    return !running.load() || !job_queue.empty();
                });
 
                if (!running.load()) {
                    break;
                }
 
                if (job_queue.empty()) {
                    continue;
                }
 
                auto entry = job_queue.top();
                job_queue.pop();
                job_id = entry.job_id;
                priority = entry.priority;
            }
 
            // Check if cancelled or paused
            if (is_job_cancelled(job_id)) {
                logger->debug_fmt("Job {} was cancelled before execution", job_id);
                continue;
            }
 
            if (is_job_paused(job_id)) {
                logger->debug_fmt("Job {} is paused, re-queueing", job_id);
                // Re-queue with same priority for later
                enqueue_job(job_id, priority);
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                continue;
            }
 
            // Get job record
            auto job_opt = get_job_from_cache(job_id);
            if (!job_opt) {
                logger->warn_fmt("Job {} not found in cache", job_id);
                continue;
            }
 
            auto job = *job_opt;
 
            // Mark as running
            mark_job_active(job_id);
            update_job_status(job_id, job_status::running);
            logger->info_fmt("Starting job {}: type={}", job_id, to_string(job.type));
 
            // Execute job
            try {
                execute_job(job);
            } catch (const std::exception& e) {
                logger->error_fmt("Job {} failed with exception: {}", job_id, e.what());
                complete_job(job_id, job_status::failed, e.what());
            }
 
            mark_job_inactive(job_id);
        }
    }