multi_process_monitoring_integration.cpp File Reference

Multi-process monitoring integration with multiple thread pools. More...

#include <kcenon/thread/core/thread_pool.h>
#include <kcenon/common/interfaces/monitoring_interface.h>
#include <kcenon/thread/interfaces/thread_context.h>
#include <kcenon/thread/core/callback_job.h>
#include <kcenon/thread/core/thread_worker.h>
#include <kcenon/thread/utils/formatter.h>
#include <chrono>
#include <thread>
#include <iostream>
#include <vector>
#include <random>

Include dependency graph for multi_process_monitoring_integration.cpp:

Go to the source code of this file.

Classes
class	sample_monitoring

Functions
int	main ()

Detailed Description

Multi-process monitoring integration with multiple thread pools.

Definition in file multi_process_monitoring_integration.cpp.

Function Documentation

main()

int main

(

)

Definition at line 90 of file multi_process_monitoring_integration.cpp.

           {
    std::cout << "=== Multi-Process Monitoring Integration Example ===\n\n";
 
    // Create monitoring instance
    auto monitoring = std::make_shared<sample_monitoring>();
 
    // Create thread context with monitoring
    thread_context context(nullptr, monitoring);
 
    // Create multiple thread pools with unique names
    auto primary_pool = std::make_shared<thread_pool>("primary_pool", context);
    auto secondary_pool = std::make_shared<thread_pool>("secondary_pool", context);
 
    // Display pool instance IDs
    std::cout << formatter::format("Primary pool instance ID: {}\n", primary_pool->get_pool_instance_id());
    std::cout << formatter::format("Secondary pool instance ID: {}\n\n", secondary_pool->get_pool_instance_id());
 
    // Add workers then start pools
    {
        std::vector<std::unique_ptr<thread_worker>> workers;
        for (int i = 0; i < 3; ++i) workers.push_back(std::make_unique<thread_worker>());
        auto r = primary_pool->enqueue_batch(std::move(workers));
        if (r.is_err()) {
            std::cerr << "Failed to add workers to primary_pool: " << r.error().message << "\n";
            return 1;
        }
    }
    {
        std::vector<std::unique_ptr<thread_worker>> workers;
        for (int i = 0; i < 2; ++i) workers.push_back(std::make_unique<thread_worker>());
        auto r = secondary_pool->enqueue_batch(std::move(workers));
        if (r.is_err()) {
            std::cerr << "Failed to add workers to secondary_pool: " << r.error().message << "\n";
            return 1;
        }
    }
 
    auto start_primary = primary_pool->start();
    if (start_primary.is_err()) {
        std::cerr << "Failed to start primary_pool: " << start_primary.error().message << "\n";
        return 1;
    }
    auto start_secondary = secondary_pool->start();
    if (start_secondary.is_err()) {
        std::cerr << "Failed to start secondary_pool: " << start_secondary.error().message << "\n";
        return 1;
    }
 
    // Report initial metrics
    primary_pool->report_metrics();
    secondary_pool->report_metrics();
 
    std::cout << "\n--- Submitting jobs ---\n";
 
    // Submit jobs to primary pool
    for (int i = 0; i < 10; ++i) {
        auto job = std::make_unique<callback_job>(
            [i]() {
                std::this_thread::sleep_for(std::chrono::milliseconds(50 + i * 10));
                std::cout << formatter::format("Primary job {} completed\n", i);
                return kcenon::common::ok();
            },
            formatter::format("primary_job_{}", i)
        );
        auto r = primary_pool->enqueue(std::move(job));
        if (r.is_err()) {
            std::cerr << "enqueue to primary_pool failed: " << r.error().message << "\n";
        }
    }
 
    // Submit jobs to secondary pool
    for (int i = 0; i < 5; ++i) {
        auto job = std::make_unique<callback_job>(
            [i]() {
                std::this_thread::sleep_for(std::chrono::milliseconds(100));
                std::cout << formatter::format("Secondary job {} completed\n", i);
                return kcenon::common::ok();
            },
            formatter::format("secondary_job_{}", i)
        );
        auto r = secondary_pool->enqueue(std::move(job));
        if (r.is_err()) {
            std::cerr << "enqueue to secondary_pool failed: " << r.error().message << "\n";
        }
    }
 
    // Periodically report metrics while jobs are processing
    for (int i = 0; i < 3; ++i) {
        std::this_thread::sleep_for(std::chrono::milliseconds(200));
        std::cout << "\n--- Metrics Update ---\n";
        primary_pool->report_metrics();
        secondary_pool->report_metrics();
    }
 
    // Stop pools
    std::cout << "\n--- Stopping pools ---\n";
    auto stop_primary = primary_pool->stop();
    if (stop_primary.is_err()) {
        std::cerr << "Error stopping primary_pool: " << stop_primary.error().message << "\n";
    }
    auto stop_secondary = secondary_pool->stop();
    if (stop_secondary.is_err()) {
        std::cerr << "Error stopping secondary_pool: " << stop_secondary.error().message << "\n";
    }
 
    // Final metrics
    std::cout << "\n--- Final Metrics ---\n";
    primary_pool->report_metrics();
    secondary_pool->report_metrics();
 
    std::cout << "\n=== Example completed ===\n";
 
    return 0;
}

References kcenon::thread::utils::formatter::format().

Here is the call graph for this function: