Thread pool sample with logger integration and batch job submission. More...

#include <iostream>
#include <memory>
#include <chrono>
#include "logger/core/logger.h"
#include <kcenon/thread/utils/formatter.h>
#include <kcenon/thread/core/thread_pool.h>

Include dependency graph for thread_pool_sample.cpp:

Go to the source code of this file.

Functions
auto	initialize_logger () -> std::optional< std::string >

auto	create_default (const uint16_t &worker_counts) -> std::tuple< std::shared_ptr< thread_pool >, kcenon::common::VoidResult >

auto	store_job (std::shared_ptr< thread_pool > thread_pool) -> kcenon::common::VoidResult

auto	main () -> int

Variables
bool	use_backup_ = false

uint32_t	max_lines_ = 0

uint16_t	wait_interval_ = 100

uint32_t	test_line_count_ = 1000000

log_module::log_types	file_target_ = log_module::log_types::None

log_module::log_types	console_target_ = log_module::log_types::Information

log_module::log_types	callback_target_ = log_module::log_types::None

uint16_t	thread_counts_ = 10

Detailed Description

Thread pool sample with logger integration and batch job submission.

Definition in file thread_pool_sample.cpp.

Function Documentation

◆ create_default()

auto create_default ( const uint16_t & worker_counts ) -> std::tuple<std::shared_ptr<thread_pool>, kcenon::common::VoidResult>

Examples: thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 59 of file thread_pool_sample.cpp.

{
    std::shared_ptr<thread_pool> pool;
 
    try
    {
        pool = std::make_shared<thread_pool>();
    }
    catch (const std::bad_alloc& e)
    {
        return { nullptr, kcenon::common::error_info{
            kcenon::thread::error_code::allocation_failed,
            e.what(),
            "thread_pool_sample"} };
    }
 
    std::vector<std::unique_ptr<thread_worker>> workers;
    workers.reserve(worker_counts);
    for (uint16_t i = 0; i < worker_counts; ++i)
    {
        workers.push_back(std::make_unique<thread_worker>());
    }
 
    auto enqueue_result = pool->enqueue_batch(std::move(workers));
    if (enqueue_result.is_err())
    {
        return { nullptr, enqueue_result.error() };
    }
 
    return { pool, kcenon::common::ok() };
}

Referenced by main().

Here is the caller graph for this function:

◆ initialize_logger()

auto initialize_logger ( ) -> std::optional<std::string>

Definition at line 38 of file thread_pool_sample.cpp.

{
    log_module::set_title("thread_pool_sample");
    log_module::set_use_backup(use_backup_);
    log_module::set_max_lines(max_lines_);
    log_module::file_target(file_target_);
    log_module::console_target(console_target_);
    log_module::callback_target(callback_target_);
    // Note: This demonstrates the logger callback feature - std::cout is intentionally used here
    log_module::message_callback(
        [](const log_module::log_types& type, const std::string& datetime,
           const std::string& message)
        { std::cout << formatter::format("[{}][{}] {}\n", datetime, type, message); });
    if (wait_interval_ > 0)
    {
        log_module::set_wake_interval(std::chrono::milliseconds(wait_interval_));
    }
 
    return log_module::start();
}

References callback_target_, console_target_, file_target_, kcenon::thread::utils::formatter::format(), max_lines_, use_backup_, and wait_interval_.

Referenced by main().

Here is the call graph for this function:

Here is the caller graph for this function:

◆ main()

auto main ( ) -> int

Definition at line 118 of file thread_pool_sample.cpp.

{
    auto error_message = initialize_logger();
    if (error_message.has_value())
    {
        std::cerr << formatter::format("error starting logger: {}\n",
                                       error_message.value_or("unknown error"));
        return 0;
    }
 
    std::shared_ptr<thread_pool> thread_pool = nullptr;
    kcenon::common::VoidResult pool_init_result;
    std::tie(thread_pool, pool_init_result) = create_default(thread_counts_);
    if (pool_init_result.is_err())
    {
        log_module::write_error("error creating thread pool: {}",
                                pool_init_result.error().message);
 
        return 0;
    }
 
    log_module::write_information("created {}", thread_pool->to_string());
 
    auto store_result = store_job(thread_pool);
    if (store_result.is_err())
    {
        log_module::write_error("error storing job: {}", store_result.error().message);
 
        thread_pool.reset();
 
        return 0;
    }
 
    auto start_result = thread_pool->start();
    if (start_result.is_err())
    {
        log_module::write_error("error starting thread pool: {}",
                                start_result.error().message);
 
        thread_pool.reset();
 
        return 0;
    }
 
    log_module::write_information("started {}", thread_pool->to_string());
 
    auto stop_result = thread_pool->stop();
    if (stop_result.is_err())
    {
        log_module::write_error("error stopping thread pool: {}",
                                stop_result.error().message);
    }
 
    log_module::write_information("stopped {}", thread_pool->to_string());
 
    thread_pool.reset();
 
    log_module::stop();
 
    return 0;
}

References create_default(), kcenon::thread::utils::formatter::format(), initialize_logger(), kcenon::thread::thread_pool::start(), kcenon::thread::thread_pool::stop(), store_job(), thread_counts_, and kcenon::thread::thread_pool::to_string().

Here is the call graph for this function:

◆ store_job()

auto store_job ( std::shared_ptr< thread_pool > thread_pool ) -> kcenon::common::VoidResult

Examples: thread_pool_sample.cpp, typed_thread_pool_sample.cpp, and typed_thread_pool_sample_2.cpp.

Definition at line 92 of file thread_pool_sample.cpp.

{
    std::vector<std::unique_ptr<job>> jobs;
    jobs.reserve(test_line_count_);
 
    for (auto index = 0; index < test_line_count_; ++index)
    {
        jobs.push_back(std::make_unique<callback_job>(
            [index](void) -> kcenon::common::VoidResult
            {
                log_module::write_debug("Hello, World!: {}", index);
                return kcenon::common::ok();
            }));
    }
 
    auto result = thread_pool->enqueue_batch(std::move(jobs));
    if (result.is_err())
    {
        return result.error();
    }
 
    log_module::write_sequence("enqueued jobs: {}", test_line_count_);
 
    return kcenon::common::ok();
}