thread_pool_sample.cpp

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.

/**
 * @file thread_pool_sample.cpp
 * @brief Thread pool sample with logger integration and batch job submission
 * @example thread_pool_sample.cpp
 *
 * Shows how to create a thread pool with multiple workers, enqueue a large
 * batch of callback jobs, and use the logger system for output. This is the
 * canonical "hello world" for thread_system with logging.
 *
 * @see thread_pool, callback_job, log_module
 */

#include <iostream>
#include <memory>
#include <chrono>

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

using kcenon::thread::utils::formatter;
using namespace kcenon::thread;

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;

auto initialize_logger() -> std::optional<std::string>
{
	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();
}

auto create_default(const uint16_t& worker_counts)
	-> std::tuple<std::shared_ptr<thread_pool>, kcenon::common::VoidResult>
{
	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() };
}

auto store_job(std::shared_ptr<thread_pool> thread_pool) -> kcenon::common::VoidResult
{
	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();
}

auto main() -> int
{
	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;
}