Создание своего пула потоков на C++

std::atomic<bool> stop{false};  // Флаг остановки пула
std::atomic<size_t> active_tasks{0};  // Счётчик активных задач

std::queue<Task> tasks;
std::mutex queue_mutex;
 
void enqueue(Task task) {
    std::lock_guard<std::mutex> lock(queue_mutex);
    tasks.push(std::move(task));
}

std::condition_variable condition;
 
void worker_thread() {
    while (true) {
        std::unique_lock<std::mutex> lock(queue_mutex);
        condition.wait(lock, [this] { 
            return stop || !tasks.empty(); 
        });
        
        if (stop && tasks.empty())
            return;
            
        auto task = std::move(tasks.front());
        tasks.pop();
        lock.unlock();
        
        task();  // Выполняем задачу без блокировки мьютекса
    }
}

// Производитель (клиентский код)
pool.enqueue([](){ return heavy_computation(); });
 
// Потребитель (рабочий поток)
while (true) {
    Task task = queue.pop(); // Блокируется, если очередь пуста
    task();
}

class ThreadPoolMonitor {
private:
    std::mutex stats_mutex;
    size_t completed_tasks{0};
    size_t failed_tasks{0};
public:
    void task_completed() {
        std::lock_guard<std::mutex> lock(stats_mutex);
        ++completed_tasks;
    }
    
    void task_failed() {
        std::lock_guard<std::mutex> lock(stats_mutex);
        ++failed_tasks;
    }
    
    std::pair<size_t, size_t> get_stats() {
        std::lock_guard<std::mutex> lock(stats_mutex);
        return {completed_tasks, failed_tasks};
    }
};

class ActiveObject {
private:
    ThreadPool pool;
public:
    template<class F, class... Args>
    auto execute(F&& f, Args&&... args) {
        using RetType = std::invoke_result_t<F, Args...>;
        return pool.enqueue(std::forward<F>(f), std::forward<Args>(args)...);
    }
};

class WorkStealingQueue {
private:
    std::deque<Task> tasks;
    std::mutex mutex;
    
public:
    void push(Task task) {
        std::lock_guard<std::mutex> lock(mutex);
        tasks.push_back(std::move(task));
    }
    
    std::optional<Task> pop() { // Для владельца очереди
        std::lock_guard<std::mutex> lock(mutex);
        if (tasks.empty()) return std::nullopt;
        
        Task task = std::move(tasks.back());
        tasks.pop_back();
        return task;
    }
    
    std::optional<Task> steal() { // Для "воров"
        std::lock_guard<std::mutex> lock(mutex);
        if (tasks.empty()) return std::nullopt;
        
        Task task = std::move(tasks.front());
        tasks.pop_front();
        return task;
    }
};

#include <vector>
#include <queue>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <future>
#include <memory>
 
class ThreadPool {
public:
    // Конструктор создаёт указанное количество рабочих потоков
    ThreadPool(size_t threads);
    
    // Деструктор останавливает все потоки и освобождает ресурсы
    ~ThreadPool();
    
    // Метод для добавления задачи в пул
    template<class F, class... Args>
    auto enqueue(F&& f, Args&&... args) 
        -> std::future<typename std::result_of<F(Args...)>::type>;
    
private:
    // Вектор рабочих потоков
    std::vector<std::thread> workers;
    
    // Очередь задач
    std::queue<std::function<void()>> tasks;
    
    // Мьютекс для защиты доступа к очереди
    std::mutex queue_mutex;
    
    // Условная переменная для оповещения потоков о новых задачах
    std::condition_variable condition;
    
    // Флаг остановки пула
    bool stop;
};

ThreadPool::ThreadPool(size_t threads) : stop(false) {
    for(size_t i = 0; i < threads; ++i) {
        workers.emplace_back([this] {
            while(true) {
                std::function<void()> task;
                
                {
                    // Блокируем доступ к очереди
                    std::unique_lock<std::mutex> lock(this->queue_mutex);
                    
                    // Ждём, пока не появится задача или не будет вызван деструктор
                    this->condition.wait(lock, [this] { 
                        return this->stop || !this->tasks.empty(); 
                    });
                    
                    // Если пул остановлен и нет задач, завершаем поток
                    if(this->stop && this->tasks.empty())
                        return;
                    
                    // Берём задачу из очереди
                    task = std::move(this->tasks.front());
                    this->tasks.pop();
                }
                
                // Выполняем задачу (вне зоны действия блокировки!)
                task();
            }
        });
    }
}

ThreadPool::~ThreadPool() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    
    // Оповещаем все ждущие потоки
    condition.notify_all();
    
    // Ждём завершения всех потоков
    for(std::thread &worker: workers)
        worker.join();
}

template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type> {
    
    // Определяем тип результата задачи
    using return_type = typename std::result_of<F(Args...)>::type;
    
    // Создаём packaged_task, который обернёт нашу функцию
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    );
    
    // Получаем future для возврата результата
    std::future<return_type> result = task->get_future();
    
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        
        // Проверяем, не остановлен ли пул
        if(stop)
            throw std::runtime_error("enqueue на остановленный ThreadPool");
        
        // Добавляем задачу в очередь
        tasks.emplace([task]() { (*task)(); });
    }
    
    // Оповещаем один из ждущих потоков о новой задаче
    condition.notify_one();
    
    return result;
}

int main() {
    // Создаём пул с 4 рабочими потоками
    ThreadPool pool(4);
    
    // Вектор для хранения результатов
    std::vector<std::future<int>> results;
    
    // Добавляем 8 задач в пул
    for(int i = 0; i < 8; ++i) {
        results.emplace_back(
            pool.enqueue([i] {
                // Имитируем сложные вычисления
                std::this_thread::sleep_for(std::chrono::seconds(1));
                return i * i;
            })
        );
    }
    
    // Выводим результаты
    for(auto& result : results)
        std::cout << result.get() << ' ';
    std::cout << std::endl;
    
    return 0;
}

void worker_loop() {
    while (true) {
        std::function<void()> task;
        
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            condition.wait(lock, [this] { return stop || !tasks.empty(); });
            
            if (stop && tasks.empty())
                return;
                
            task = std::move(tasks.front());
            tasks.pop();
        }
        
        try {
            task();
        }
        catch (const std::exception& e) {
            std::cerr << "Exception in thread pool: " << e.what() << std::endl;
            // Можно добавить логгер или колбек для обработки
        }
        catch (...) {
            std::cerr << "Unknown exception in thread pool" << std::endl;
        }
    }
}

class PriorityThreadPool {
private:
    using TaskPair = std::pair<int, std::function<void()>>;
    
    // Компаратор для сравнения приоритетов (меньшее значение = выше приоритет)
    struct TaskCompare {
        bool operator()(const TaskPair& lhs, const TaskPair& rhs) {
            return lhs.first > rhs.first;
        }
    };
    
    std::priority_queue<TaskPair, std::vector<TaskPair>, TaskCompare> tasks;
    // Остальные члены класса как в базовом пуле
    
public:
    // Конструктор и деструктор как в базовом пуле
    
    template<class F, class... Args>
    auto enqueue(int priority, F&& f, Args&&... args) -> std::future<typename std::result_of<F(Args...)>::type> {
        using return_type = typename std::result_of<F(Args...)>::type;
        
        auto task = std::make_shared<std::packaged_task<return_type()>>(
            std::bind(std::forward<F>(f), std::forward<Args>(args)...)
        );
        
        std::future<return_type> result = task->get_future();
        
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            if (stop)
                throw std::runtime_error("enqueue на остановленный PriorityThreadPool");
                
            tasks.emplace(priority, [task]() { (*task)(); });
        }
        
        condition.notify_one();
        return result;
    }
};

class WorkStealingThreadPool {
private:
    std::vector<std::unique_ptr<WorkStealingQueue>> local_queues;
    std::vector<std::thread> workers;
    std::atomic<bool> stop{false};
    
    // Простая реализация work-stealing очереди
    class WorkStealingQueue {
    private:
        std::deque<std::function<void()>> queue;
        std::mutex mutex;
        
    public:
        void push(std::function<void()> task) {
            std::lock_guard<std::mutex> lock(mutex);
            queue.push_back(std::move(task));
        }
        
        bool pop(std::function<void()>& task) {
            std::lock_guard<std::mutex> lock(mutex);
            if (queue.empty())
                return false;
                
            task = std::move(queue.back());
            queue.pop_back();
            return true;
        }
        
        bool steal(std::function<void()>& task) {
            std::lock_guard<std::mutex> lock(mutex);
            if (queue.empty())
                return false;
                
            task = std::move(queue.front());
            queue.pop_front();
            return true;
        }
    };
    
public:
    // Реализация методов для работы с локальными очередями
};

// Добавляем в класс ThreadPool
std::atomic<size_t> active_tasks{0};
std::atomic<size_t> completed_tasks{0};
std::atomic<size_t> failed_tasks{0};
 
// В методе worker_loop, перед выполнением задачи
++active_tasks;
 
try {
    task();
    ++completed_tasks;
}
catch (...) {
    ++failed_tasks;
    // Обработка исключения
}
 
// После выполнения задачи
--active_tasks;

void ThreadPool::shutdown() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for (std::thread &worker : workers)
        worker.join();
}
 
void ThreadPool::shutdown_now() {
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
        tasks.clear(); // Очищаем очередь задач
    }
    condition.notify_all();
    for (std::thread &worker : workers)
        worker.join();
}

void ThreadPool::drain_and_shutdown() {
    // Блокируем добавление новых задач
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop_accepting = true; // Новый флаг, запрещающий enqueue
    }
    
    // Ждём, пока все задачи не будут выполнены
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        drain_condition.wait(lock, [this] { 
            return tasks.empty() && active_tasks == 0; 
        });
    }
    
    // Завершаем потоки
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for (std::thread &worker : workers)
        worker.join();
}

bool ThreadPool::shutdown_with_timeout(std::chrono::milliseconds timeout) {
    auto start_time = std::chrono::steady_clock::now();
    
    // Блокируем добавление новых задач
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop_accepting = true;
    }
    
    // Ждём с таймаутом
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        bool all_done = drain_condition.wait_for(lock, timeout, [this] { 
            return tasks.empty() && active_tasks == 0; 
        });
        
        if (!all_done) {
            // Таймаут истёк, но задачи остались
            stop = true;
            condition.notify_all();
            
            for (std::thread &worker : workers)
                if (worker.joinable())
                    worker.join();
                    
            return false; // Не все задачи завершены
        }
    }
    
    // Нормальное завершение
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        stop = true;
    }
    condition.notify_all();
    for (std::thread &worker : workers)
        worker.join();
        
    return true; // Все задачи завершены
}

class AdaptiveThreadPool {
private:
std::atomic<size_t> target_threads{0}; // Целевое количество потоков
std::atomic<size_t> active_threads{0}; // Активные потоки
std::atomic<size_t> pending_tasks{0};  // Ожидающие задачи
std::mutex resize_mutex;               // Защита операций изменения размера
 
void monitor_and_adjust() {
    while (!stop) {
        std::this_thread::sleep_for(std::chrono::seconds(1));
        
        size_t waiting = pending_tasks.load();
        size_t current = active_threads.load();
        
        // Политика масштабирования: если очередь растёт, добавляем потоки
        if (waiting > current && current < max_threads) {
            add_worker();
        }
        // Если потоков слишком много и они простаивают, убираем часть
        else if (waiting == 0 && current > min_threads && idle_threads > 2) {
            remove_worker();
        }
    }
}

void add_worker() {
std::lock_guard<std::mutex> lock(resize_mutex);
workers.emplace_back([this] {
    ++active_threads;
    worker_loop();
    --active_threads;
});
++target_threads;
}

void remove_worker() {
std::lock_guard<std::mutex> lock(resize_mutex);
if (workers.empty()) return;
 
{
    std::lock_guard<std::mutex> queue_lock(queue_mutex);
    tasks.push(Task::create_termination_token());
}
condition.notify_one();
--target_threads;
}

void worker_loop() {
while (true) {
    Task task = get_next_task();
    
    if (task.is_termination_token()) {
        // Этот поток должен завершиться
        std::lock_guard<std::mutex> lock(resize_mutex);
        if (active_threads <= min_threads) {
            // Не завершаемся, если достигли минимума
            continue;
        }
        return; // Завершаем поток
    }
    
    execute_task(task);
}
}

// Замена обычной очереди на приоритетную
std::priority_queue<
    std::pair<int, std::function<void()>>,     // Элемент: приоритет + задача
    std::vector<std::pair<int, std::function<void()>>>,  // Контейнер
    std::function<bool(
        const std::pair<int, std::function<void()>>&,
        const std::pair<int, std::function<void()>>&
    )>                                          // Компаратор
> tasks;

using Task = std::function<void()>;
using PriorityTask = std::pair<int, Task>;
 
struct PriorityCompare {
    bool operator()(const PriorityTask& a, const PriorityTask& b) {
        return a.first > b.first;  // Меньший приоритет => ниже в очереди
    }
};
 
std::priority_queue<PriorityTask, std::vector<PriorityTask>, PriorityCompare> tasks;

template<class F, class... Args>
auto enqueue(int priority, F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type> {
    using return_type = typename std::result_of<F(Args...)>::type;
    
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    );
    
    std::future<return_type> result = task->get_future();
    
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(stop)
            throw std::runtime_error("enqueue на остановленный ThreadPool");
        
        // Добавляем задачу с указанным приоритетом
        tasks.emplace(priority, [task]() { (*task)(); });
    }
    
    condition.notify_one();
    return result;
}

// Массив очередей по приоритетам
std::array<std::queue<Task>, PRIORITY_LEVELS> priority_queues;
 
Task get_next_task() {
    std::unique_lock<std::mutex> lock(queue_mutex);
    
    // Ищем задачу, начиная с наивысшого приоритета
    for (int p = 0; p < PRIORITY_LEVELS; ++p) {
        if (!priority_queues[p].empty()) {
            Task task = std::move(priority_queues[p].front());
            priority_queues[p].pop();
            return task;
        }
    }
    
    // Если очереди пусты, ждём новых задач
    condition.wait(lock, [this] {
        for (int p = 0; p < PRIORITY_LEVELS; ++p) {
            if (!priority_queues[p].empty()) return true;
        }
        return stop;
    });
    
    // После пробуждения проверяем состояние пула
    if (stop && all_queues_empty())
        return Task();  // Возвращаем пустую задачу для завершения потока
    
    // Повторяем поиск задачи
    for (int p = 0; p < PRIORITY_LEVELS; ++p) {
        if (!priority_queues[p].empty()) {
            Task task = std::move(priority_queues[p].front());
            priority_queues[p].pop();
            return task;
        }
    }
    
    return Task();  // Никогда не должны дойти до этой строки
}

struct AgingTask {
    Task task;
    int base_priority;
    std::chrono::steady_clock::time_point submit_time;
    
    int current_priority() const {
        auto age = std::chrono::steady_clock::now() - submit_time;
        auto seconds = std::chrono::duration_cast<std::chrono::seconds>(age).count();
        return std::max(0, base_priority - static_cast<int>(seconds / AGING_FACTOR));
    }
};

class CancellationToken {
private:
    std::atomic<bool> cancelled{false};
 
public:
    void cancel() { cancelled = true; }
    bool is_cancelled() const { return cancelled; }
};
 
using CancellationTokenPtr = std::shared_ptr<CancellationToken>;

template<class F, class... Args>
auto enqueue_cancellable(CancellationTokenPtr token, F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type> {
    using return_type = typename std::result_of<F(Args...)>::type;
    
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        [token, f=std::forward<F>(f), args=std::make_tuple(std::forward<Args>(args)...)]() -> return_type {
            // Проверяем, не отменена ли задача
            if (token && token->is_cancelled()) {
                throw std::runtime_error("Task was cancelled");
            }
            
            return std::apply(f, args);
        }
    );
    
    // Остальной код как в обычном enqueue
}

pool.enqueue_cancellable(token, [token](int x) {
    for (int i = 0; i < 1000000; ++i) {
        // Периодически проверяем отмену
        if (i % 1000 == 0 && token->is_cancelled()) {
            std::cout << "Задача отменена на итерации " << i << std::endl;
            return 0; // или бросаем исключение
        }
        
        // Выполняем работу...
    }
    return 42;
});

template<class F, class... Args>
auto enqueue(F&& f, Args&&... args) 
    -> std::future<typename std::result_of<F(Args...)>::type> {
    using return_type = typename std::result_of<F(Args...)>::type;
    
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        [f=std::forward<F>(f), args=std::make_tuple(std::forward<Args>(args)...)]() {
            try {
                return std::apply(f, args);
            } catch (...) {
                // Передаём исключение через future
                std::throw_with_nested(std::runtime_error("Exception in thread pool task"));
            }
        }
    );
    
    // Остальной код как обычно
}

try {
    int result = future.get();
    // Используем результат
} catch (const std::exception& e) {
    std::cerr << "Ошибка при выполнении асинхронной задачи: " << e.what() << std::endl;
    
    // Можно развернуть вложенные исключения
    try {
        std::rethrow_if_nested(e);
    } catch (const std::exception& nested) {
        std::cerr << "Вложенное исключение: " << nested.what() << std::endl;
    }
}

template<class F, class... Args>
auto ThreadPool::enqueue(F&& f, Args&&... args) 
-> std::future<typename std::invoke_result_t<F(Args...)>> {
    using return_type = typename std::invoke_result_t<F(Args...)>;
    
    // Создаём packaged_task, связывающий функцию с promise
    auto task = std::make_shared<std::packaged_task<return_type()>>(
        std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    );
    
    // Получаем future, связанный с promise внутри packaged_task
    std::future<return_type> result = task->get_future();
    
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");
            
        // Оборачиваем task в функцию без параметров и результата
        tasks.emplace([task](){ (*task)(); });
    }
    
    condition.notify_one();
    return result;
}

template<class F, class... Args>
auto ThreadPool::enqueue_extended(F&& f, Args&&... args) 
-> std::future<typename std::invoke_result_t<F(Args...)>> {
    using return_type = typename std::invoke_result_t<F(Args...)>;
    
    // Создаём promise вручную для большего контроля
    auto promise_ptr = std::make_shared<std::promise<return_type>>();
    std::future<return_type> future = promise_ptr->get_future();
    
    auto task_function = [promise_ptr, f=std::forward<F>(f), 
                         args=std::make_tuple(std::forward<Args>(args)...)]() {
        try {
            if (is_cancelled()) {
                promise_ptr->set_exception(
                    std::make_exception_ptr(std::runtime_error("Task cancelled"))
                );
                return;
            }
            
            // Выполняем функцию и передаём результат в promise
            if constexpr (std::is_void_v<return_type>) {
                std::apply(f, args);
                promise_ptr->set_value();
            } else {
                promise_ptr->set_value(std::apply(f, args));
            }
        } catch (...) {
            // Передаём исключение через promise
            promise_ptr->set_exception(std::current_exception());
        }
    };
    
    // Оставшаяся часть идентична базовой реализации
    {
        std::unique_lock<std::mutex> lock(queue_mutex);
        if(stop)
            throw std::runtime_error("enqueue on stopped ThreadPool");
            
        tasks.emplace(std::move(task_function));
    }
    
    condition.notify_one();
    return future;
}

// Ждёт завершения всех futures из диапазона
template<typename Iterator>
void wait_all(Iterator begin, Iterator end) {
    for (auto it = begin; it != end; ++it) {
        it->wait();
    }
}
 
// Ждёт завершения хотя бы одной future, возвращает индекс
template<typename Iterator>
size_t wait_any(Iterator begin, Iterator end) {
    while (true) {
        for (auto it = begin; it != end; ++it) {
            auto status = it->wait_for(std::chrono::milliseconds(0));
            if (status == std::future_status::ready) {
                return std::distance(begin, it);
            }
        }
        std::this_thread::yield();  // Уступаем процессорное время
    }
}

void benchmark_cpu_bound(ThreadPoolInterface& pool, size_t tasks_count) {
    auto start = std::chrono::high_resolution_clock::now();
    
    std::vector<std::future<uint64_t>> results;
    for (size_t i = 0; i < tasks_count; ++i) {
        results.push_back(pool.enqueue([i] {
            // Имитация тяжелых вычислений
            uint64_t sum = 0;
            for (uint64_t j = 0; j < 10000000; ++j) {
                sum += j % (i + 1);
            }
            return sum;
        }));
    }
    
    // Ждём завершения всех задач
    for (auto& f : results) {
        f.get();
    }
    
    auto end = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> duration = end - start;
    
    std::cout << "Время выполнения: " << duration.count() << " секунд\n";
}

Наш basic_pool:      3.42
Наш ws_pool:         2.18
Boost.Thread:        3.57
Intel TBB:           2.47
std::async:          5.83

struct TimeConstrainedTask {
    std::function<void()> task;
    std::chrono::steady_clock::time_point deadline;
    uint8_t priority;
    
    // Сравнение с учётом и дедлайна, и базового приоритета
    bool operator<(const TimeConstrainedTask& other) const {
        // Если дедлайны сильно различаются, они важнее
        auto time_diff = deadline - other.deadline;
        if (std::abs(time_diff.count()) > critical_threshold) {
            return deadline < other.deadline;
        }
        // Иначе используем базовый приоритет
        return priority > other.priority;
    }
};

void set_real_time_priority(std::thread& thread, int priority) {
    sched_param sch_params;
    sch_params.sched_priority = priority;
    if (pthread_setschedparam(thread.native_handle(), SCHED_FIFO, &sch_params)) {
        std::cerr << "Failed to set thread priority: " << strerror(errno) << std::endl;
    }
}

template<typename T, size_t PoolSize = 1024>
class ObjectPool {
private:
    std::array<T, PoolSize> objects;
    std::bitset<PoolSize> used_flags;
    std::mutex pool_mutex;
    
public:
    T* acquire() {
        std::lock_guard<std::mutex> lock(pool_mutex);
        for (size_t i = 0; i < PoolSize; ++i) {
            if (!used_flags[i]) {
                used_flags[i] = true;
                return &objects[i];
            }
        }
        return nullptr; // Пул исчерпан
    }
    
    void release(T* obj) {
        std::lock_guard<std::mutex> lock(pool_mutex);
        ptrdiff_t index = obj - &objects[0];
        if (index >= 0 && index < PoolSize) {
            used_flags[index] = false;
        }
    }
};

class ThreadPoolService : public boost::asio::io_service::service {
private:
ThreadPool& thread_pool;
boost::asio::io_service::id service_id;
 
public:
ThreadPoolService(boost::asio::io_service& ios, ThreadPool& pool)
  : boost::asio::io_service::service(ios), thread_pool(pool) {}
 
static boost::asio::io_service::id id;
 
// Реализуем метод для отправки задач в наш пул
template<typename Handler>
void post(Handler handler) {
    thread_pool.enqueue([handler = std::move(handler)]() mutable {
        handler();
    });
}
 
// Необходимые методы для интерфейса service
void shutdown_service() override {}
};
 
// Нужен для идентификации сервиса в io_service
boost::asio::io_service::id ThreadPoolService::id;

boost::asio::io_service ios;
ThreadPool pool(8);  // Пул с 8 потоками
auto& service = boost::asio::use_service<ThreadPoolService>(ios);
// Теперь ios.post() будет использовать наш пул

struct ThreadPoolAwaiter {
ThreadPool& pool;
std::function<void()> task;
 
bool await_ready() const noexcept { return false; }
 
void await_suspend(std::coroutine_handle<> handle) {
    pool.enqueue([handle, this]() {
        task();
        handle.resume();
    });
}
 
void await_resume() const noexcept {}
};
 
// Использование
ThreadPool pool(4);
 
task<void> example_coroutine() {
    // ... какой-то код
    
    co_await ThreadPoolAwaiter{
        pool, 
        []() { /* тяжёлая задача */ }
    };
    
    // Продолжение после выполнения задачи
}

class ThreadPoolExecutor : public folly::Executor {
private:
ThreadPool& pool;
 
public:
ThreadPoolExecutor(ThreadPool& p) : pool(p) {}
 
void add(folly::Func func) override {
    pool.enqueue(std::move(func));
}
};

void integrate_with_libuv(uv_loop_t* loop, ThreadPool& pool) {
auto work = [&pool](uv_work_t* req) {
    // Получаем задачу из req->data и выполняем
    auto task = static_cast<std::function<void()>*>(req->data);
    pool.enqueue(*task);
};
 
auto after_work = [](uv_work_t* req, int status) {
    // Обработка завершения
    delete static_cast<std::function<void()>*>(req->data);
    delete req;
};
 
// Использование
auto req = new uv_work_t();
req->data = new std::function<void()>([]() { /* задача */ });
uv_queue_work(loop, req, work, after_work);
}

void transfer_between_accounts(Account& from, Account& to, double amount) {
    // Мьютексы захватываются в определённом порядке, предотвращая deadlock
    std::scoped_lock lock(from.mutex, to.mutex);
    from.balance -= amount;
    to.balance += amount;
}

bool ThreadPool::try_enqueue(Task task) {
    std::unique_lock<std::mutex> lock(queue_mutex);
    if (tasks.size() >= max_queue_size)
        return false;  // Отказ при переполнении очереди
    
    tasks.push(std::move(task));
    lock.unlock();
    condition.notify_one();
    return true;
}

struct alignas(64) Counter {
    std::atomic<size_t> value{0};
    // padding заполнит структуру до кратности линии кэша
    char padding[64 - sizeof(std::atomic<size_t>)];
};

// Потенциальная взаимоблокировка:
void thread_a() {
    std::lock_guard<std::mutex> lock_x(mutex_x);
    std::this_thread::sleep_for(std::chrono::milliseconds(10)); // Имитация работы
    std::lock_guard<std::mutex> lock_y(mutex_y);
    // Никогда не дойдём сюда, если thread_b уже захватил mutex_y
}
 
void thread_b() {
    std::lock_guard<std::mutex> lock_y(mutex_y);
    std::this_thread::sleep_for(std::chrono::milliseconds(10));
    std::lock_guard<std::mutex> lock_x(mutex_x);
    // Взаимная блокировка!
}

// Безопасный способ:
void transfer(Account& from, Account& to) {
    // Сравниваем адреса объектов для определения порядка блокировки
    if (&from < &to) {
        std::lock_guard<std::mutex> lock1(from.mutex);
        std::lock_guard<std::mutex> lock2(to.mutex);
        perform_transfer(from, to);
    } else {
        std::lock_guard<std::mutex> lock1(to.mutex);
        std::lock_guard<std::mutex> lock2(from.mutex);
        perform_transfer(from, to);
    }
}

void safer_transfer(Account& from, Account& to) {
    std::scoped_lock locks(from.mutex, to.mutex); // Атомарный захват обоих мьютексов
    perform_transfer(from, to);
}

bool attempt_transfer(Account& from, Account& to) {
    std::unique_lock<std::mutex> lock1(from.mutex, std::defer_lock);
    if (!lock1.try_lock_for(std::chrono::milliseconds(100)))
        return false;
        
    std::unique_lock<std::mutex> lock2(to.mutex, std::defer_lock);
    if (!lock2.try_lock_for(std::chrono::milliseconds(100))) {
        return false;
    }
    
    perform_transfer(from, to);
    return true;
}

void optimized_process() {
    Data local_data;
    
    { // Ограниченная область видимости для блокировки
        std::lock_guard<std::mutex> lock(data_mutex);
        local_data = shared_data; // Копируем под защитой мьютекса
    }
    
    // Обрабатываем копию без удержания мьютекса
    process_locally(local_data);
    
    {
        std::lock_guard<std::mutex> lock(data_mutex);
        update_shared_data(local_data); // Снова блокируем только для обновления
    }
}