Конкурентность, асинхронность и параллелизм в C++

std::thread worker([]() {
    std::cout << "Я выполняюсь в отдельном потоке!
";
});
worker.join(); // ждём завершения потока

std::vector<int> v = {5, 8, 1, 3, 7, 9, 2};
std::sort(std::execution::par, v.begin(), v.end());

task<int> async_computation() {
    int result = co_await long_running_operation();
    co_return result * 2;
}

// Создаём больше потоков, чем у нас ядер процессора
const auto num_cores = std::thread::hardware_concurrency();
std::cout << "Количество ядер: " << num_cores << '\n';
 
std::vector<std::thread> threads;
for (int i = 0; i < num_cores * 2; i++) {
    threads.emplace_back([]() {
        // Интенсивные вычисления
        for (int j = 0; j < 1000000; j++) {
            std::this_thread::sleep_for(std::chrono::microseconds(1));
        }
    });
}
 
for (auto& t : threads) {
    t.join();
}

// Потенциальный источник трудноуловимого бага
int shared_data = 0;
bool data_ready = false;
 
// Поток-писатель
void writer() {
    shared_data = 42;  // Строка A
    data_ready = true; // Строка B
}
 
// Поток-читатель
void reader() {
    while (!data_ready) {
        std::this_thread::yield(); // Даём другим потокам поработать
    }
    assert(shared_data == 42); // Может сработать! Почему?
}

// Корректная версия с использованием std::atomic
std::atomic<int> shared_data{0};
std::atomic<bool> data_ready{false};
 
void writer() {
    shared_data.store(42, std::memory_order_relaxed);
    data_ready.store(true, std::memory_order_release);
}
 
void reader() {
    while (!data_ready.load(std::memory_order_acquire)) {
        std::this_thread::yield();
    }
    assert(shared_data.load(std::memory_order_relaxed) == 42); // Всегда верно
}

template <typename T>
class ThreadSafeQueue {
private:
    mutable std::mutex mtx;  // mutable позволяет модифицировать в const методах
    std::queue<T> data_queue;
    std::condition_variable data_cond;
    
public:
    ThreadSafeQueue() = default;
    
    void push(T new_value) {
        std::lock_guard<std::mutex> lock(mtx);
        data_queue.push(std::move(new_value));
        data_cond.notify_one();  // Уведомляем один ждущий поток
    }
    
    bool try_pop(T& value) {
        std::lock_guard<std::mutex> lock(mtx);
        if(data_queue.empty()) return false;
        
        value = std::move(data_queue.front());
        data_queue.pop();
        return true;
    }
    
    std::shared_ptr<T> try_pop() {
        std::lock_guard<std::mutex> lock(mtx);
        if(data_queue.empty()) return nullptr;
        
        std::shared_ptr<T> res(
            std::make_shared<T>(std::move(data_queue.front())));
        data_queue.pop();
        return res;
    }
    
    void wait_and_pop(T& value) {
        std::unique_lock<std::mutex> lock(mtx);
        // Ждём, пока очередь не станет непустой
        data_cond.wait(lock, [this]{ return !data_queue.empty(); });
        
        value = std::move(data_queue.front());
        data_queue.pop();
    }
    
    std::shared_ptr<T> wait_and_pop() {
        std::unique_lock<std::mutex> lock(mtx);
        data_cond.wait(lock, [this]{ return !data_queue.empty(); });
        
        std::shared_ptr<T> res(
            std::make_shared<T>(std::move(data_queue.front())));
        data_queue.pop();
        return res;
    }
    
    bool empty() const {
        std::lock_guard<std::mutex> lock(mtx);
        return data_queue.empty();
    }
};

// Безблокировочный счетчик
class LockFreeCounter {
    std::atomic<uint64_t> counter{0};
public:
    void increment() {
        counter.fetch_add(1, std::memory_order_relaxed);
    }
    
    uint64_t getValue() const {
        return counter.load(std::memory_order_relaxed);
    }
};

std::promise<int> promise;
std::future<int> future = promise.get_future();
 
// Поток-вычислитель
std::thread calculator([&promise] {
    int result = compute_something_complex();
    promise.set_value(result);
});
 
// Основной поток продолжает работу...
// ...и когда нужен результат:
int answer = future.get();  // Блокируемся, если результат ещё не готов

auto future = std::async(std::launch::async, []() {
    return process_huge_amount_of_data();
});
 
// Делаем другую работу, пока данные обрабатываются...
auto result = future.get();

task<std::string> fetch_data(std::string url) {
    http_response response = co_await http_client.get(url);
    if (response.status_code != 200) {
        co_return "Error: " + std::to_string(response.status_code);
    }
    co_return response.body;
}

generator<int> fibonacci(int max = std::numeric_limits<int>::max()) {
    int a = 0, b = 1;
    co_yield a;  // Выдаём первое число
    
    if (max > 0) {
        co_yield b;  // Выдаём второе число
        
        for (int i = 2; i < max; ++i) {
            int next = a + b;
            if (next < 0) break;  // Проверка переполнения
            
            co_yield next;
            a = b;
            b = next;
        }
    }
}

for (int fib : fibonacci(10)) {
    std::cout << fib << " ";
}  // Выведет: 0 1 1 2 3 5 8 13 21 34

thread_local int counter = 0;  // У каждого потока свой счётчик
 
void increment_counter() {
    ++counter;  // Изменяем только локальную копию для текущего потока
    std::cout << "Поток #" << std::this_thread::get_id() 
              << ": counter = " << counter << '\n';
}

std::vector<std::thread> threads;
for (int i = 0; i < 5; ++i) {
    threads.emplace_back([]() {
        for (int j = 0; j < 3; ++j) {
            increment_counter();
        }
    });
}
 
for (auto& t : threads) {
    t.join();
}

std::vector<int> huge_data(1000000, 42);  // Большой вектор
 
std::thread worker([data = huge_data]() {
    // Здесь происходит копирование огромного вектора!
    process_data(data);
});

std::vector<int> huge_data(1000000, 42);
 
std::thread worker([data = std::move(huge_data)]() {
    // Данные перемещены вместо копирования
    process_data(data);
});
 
// Теперь huge_data пуст в главном потоке
assert(huge_data.empty());

std::packaged_task<int(const std::vector<int>&)> task(compute_sum);
std::future<int> result = task.get_future();
 
// Перемещаем задачу в поток
std::thread worker(std::move(task), data);
 
// Позже получаем результат
int sum = result.get();

class Actor {
private:
    ThreadSafeQueue<std::function<void()>> message_queue;
    std::thread worker_thread;
    std::atomic<bool> done{false};
    
    void process_messages() {
        while (!done) {
            std::function<void()> message;
            if (message_queue.try_pop(message)) {
                message();  // Выполняем сообщение
            } else {
                std::this_thread::yield();
            }
        }
    }
    
public:
    Actor() : worker_thread(&Actor::process_messages, this) {}
    
    ~Actor() {
        done = true;
        worker_thread.join();
    }
    
    // Отправка сообщения актору
    template<typename Func>
    void send(Func&& msg) {
        message_queue.push(std::forward<Func>(msg));
    }
};

std::barrier sync_point(3);  // 3 потока должны достичь барьера
 
void worker(int id) {
    // Фаза 1: подготовка
    std::cout << "Поток " << id << " готовит данные...
";
    prepare_data(id);
    
    // Ждём, пока все потоки закончат подготовку
    sync_point.arrive_and_wait();
    
    // Фаза 2: обработка данных со всех потоков
    std::cout << "Поток " << id << " обрабатывает все данные...
";
    process_all_data();
}

template<typename T>
class LockFreeStack {
private:
    struct Node {
        T data;
        Node* next;
    };
    
    std::atomic<Node*> head{nullptr};
    
public:
    void push(const T& value) {
        Node* new_node = new Node{value, nullptr};
        
        Node* old_head = head.load(std::memory_order_relaxed);
        do {
            new_node->next = old_head;
        } while (!head.compare_exchange_weak(old_head, new_node, 
                 std::memory_order_release, std::memory_order_relaxed));
    }
    
    bool pop(T& result) {
        Node* old_head = head.load(std::memory_order_relaxed);
        
        do {
            if (!old_head) return false;  // Стек пуст
        } while (!head.compare_exchange_weak(old_head, old_head->next,
                 std::memory_order_acquire, std::memory_order_relaxed));
                 
        result = old_head->data;
        delete old_head;  // Потенциальная проблема!
        return true;
    }
};

struct TaggedPointer {
    Node* ptr;
    uint64_t tag;  // Увеличивается при каждом изменении указателя
};
 
std::atomic<TaggedPointer> head;

struct Data { int x, y, z; };
Data my_data{1, 2, 3};
 
void thread_func() {
    std::atomic_ref<Data> atomic_data(my_data);
    
    Data expected{1, 2, 3};
    Data desired{4, 5, 6};
    
    if (atomic_data.compare_exchange_strong(expected, desired)) {
        // Успешно заменили {1,2,3} на {4,5,6}
    }
}

std::vector<int> data(10000);
// Заполнение вектора...
 
// Параллельный обход с fork-join семантикой
std::for_each(std::execution::par, data.begin(), data.end(), 
[](int& val) {
    // Каждый элемент обрабатывается в своём потоке
    val = complex_transformation(val);
});

class ThreadPool {
private:
    std::vector<std::thread> workers;
    ThreadSafeQueue<std::function<void()>> tasks;
    std::atomic<bool> stop{false};
    
public:
    ThreadPool(size_t threads = std::thread::hardware_concurrency()) {
        for(size_t i = 0; i < threads; ++i) {
            workers.emplace_back([this] {
                while(!stop) {
                    std::function<void()> task;
                    if(tasks.wait_and_pop(task)) {
                        task();
                    }
                }
            });
        }
    }
    
    ~ThreadPool() {
        stop = true;
        for(auto& worker: workers) {
            worker.join();
        }
    }
    
    template<typename F>
    void enqueue(F&& f) {
        tasks.push(std::forward<F>(f));
    }
};

// 1. Создание потока с функцией
void do_work(int x, double y) {
    // сложные вычисления
}
 
std::thread t1(do_work, 42, 3.14);
 
// 2. Создание потока с лямбдой
std::thread t2([]() {
    for (int i = 0; i < 1000; ++i) {
        calculate_something();
    }
});
 
// 3. Создание потока с методом класса
class Worker {
public:
    void process(int iterations) {
        // обработка
    }
};
 
Worker w;
std::thread t3(&Worker::process, &w, 1000);
 
// 4. Передача владения потоком
std::thread t4(heavy_computation);
std::thread t5 = std::move(t4);  // t4 больше не владеет потоком
 
// 5. Не забываем присоединять потоки!
t1.join();
t2.join();
t3.join();
t5.join();

class scoped_thread {
    std::thread t;
public:
    explicit scoped_thread(std::thread t_) : t(std::move(t_)) {
        if (!t.joinable()) 
            throw std::logic_error("No thread");
    }
    
    ~scoped_thread() {
        t.join();
    }
    
    scoped_thread(const scoped_thread&) = delete;
    scoped_thread& operator=(const scoped_thread&) = delete;
};
 
// Теперь код становится безопасным
{
    scoped_thread t{std::thread(do_work, 42, 3.14)};
    // можно не беспокоиться о вызове join()
}

// 1. Защита критического раздела с помощью mutex
std::mutex data_mutex;
std::vector<int> shared_data;
 
void add_to_data(int value) {
    std::lock_guard<std::mutex> lock(data_mutex);
    shared_data.push_back(value);
}
 
// 2. Использование atomic для простых типов
std::atomic<int> counter{0};
 
void increment() {
    ++counter;  // Атомарный инкремент
}
 
// 3. Двойная проверка с ленивой инициализацией (DCLP)
class Singleton {
private:
    static std::mutex singleton_mutex;
    static std::atomic<Singleton*> instance;
    
    Singleton() = default;
    
public:
    static Singleton* getInstance() {
        Singleton* p = instance.load(std::memory_order_acquire);
        if (!p) {  // Первая проверка без блокировки
            std::lock_guard<std::mutex> lock(singleton_mutex);
            p = instance.load(std::memory_order_relaxed);
            if (!p) {  // Вторая проверка под блокировкой
                p = new Singleton();
                instance.store(p, std::memory_order_release);
            }
        }
        return p;
    }
};
 
std::mutex Singleton::singleton_mutex;
std::atomic<Singleton*> Singleton::instance{nullptr};

class ThreadPool {
private:
    std::vector<std::thread> workers;
    std::queue<std::function<void()>> tasks;
    
    std::mutex queue_mutex;
    std::condition_variable condition;
    bool stop;
    
public:
    ThreadPool(size_t threads) : stop(false) {
        for(size_t i = 0; i < threads; ++i)
            workers.emplace_back(
                [this] {
                    for(;;) {
                        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();
                    }
                }
            );
    }
    
    template<class F>
    void enqueue(F&& f) {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            if(stop)
                throw std::runtime_error("enqueue on stopped ThreadPool");
                
            tasks.emplace(std::forward<F>(f));
        }
        condition.notify_one();
    }
    
    ~ThreadPool() {
        {
            std::unique_lock<std::mutex> lock(queue_mutex);
            stop = true;
        }
        condition.notify_all();
        for(std::thread &worker: workers)
            worker.join();
    }
};

ThreadPool pool(4);  // Создаём пул из 4 потоков
 
// Ставим задачи в очередь
auto result1 = pool.enqueue([](){ return heavy_computation1(); });
auto result2 = pool.enqueue([](){ return heavy_computation2(); });
 
// Получаем результаты
auto value1 = result1.get();
auto value2 = result2.get();

template<typename T>
class BoundedBuffer {
private:
    std::mutex mutex;
    std::condition_variable not_full;
    std::condition_variable not_empty;
    size_t capacity;
    std::queue<T> queue;
    bool closed;
    
public:
    explicit BoundedBuffer(size_t capacity) : capacity(capacity), closed(false) {}
    
    void close() {
        std::lock_guard<std::mutex> lock(mutex);
        closed = true;
        not_empty.notify_all();  // Разблокируем ждущих потребителей
    }
    
    bool push(T item) {
        std::unique_lock<std::mutex> lock(mutex);
        
        if (closed)
            return false;
            
        not_full.wait(lock, [this] { 
            return queue.size() < capacity || closed; 
        });
        
        if (closed)
            return false;
            
        queue.push(std::move(item));
        not_empty.notify_one();
        return true;
    }
    
    bool pop(T& item) {
        std::unique_lock<std::mutex> lock(mutex);
        
        not_empty.wait(lock, [this] { 
            return !queue.empty() || closed; 
        });
        
        if (queue.empty())
            return false;  // Буфер закрыт и пуст
            
        item = std::move(queue.front());
        queue.pop();
        not_full.notify_one();
        return true;
    }
};

BoundedBuffer<std::string> buffer(100);  // Буфер на 100 элементов
 
// Поток-производитель
std::thread producer([&buffer]() {
    for (int i = 0; i < 1000; ++i) {
        std::string data = generate_data();
        if (!buffer.push(data)) {
            break;  // Буфер закрыт
        }
    }
});
 
// Поток-потребитель
std::thread consumer([&buffer]() {
    std::string item;
    while (buffer.pop(item)) {
        process_data(item);
    }
});
 
// Ждём завершения работы
producer.join();
buffer.close();  // Сообщаем потребителю, что данных больше не будет
consumer.join();

// 1. Использование std::lock для одновременного захвата нескольких мьютексов
std::mutex m1, m2;
 
void safe_operation() {
    std::lock(m1, m2);  // Атомарно захватывает оба мьютекса без риска дедлока
    std::lock_guard<std::mutex> lock1(m1, std::adopt_lock);
    std::lock_guard<std::mutex> lock2(m2, std::adopt_lock);
    
    // Критический раздел
}
 
// 2. Использование std::scoped_lock (C++17) для еще более краткой записи
void modern_safe_operation() {
    std::scoped_lock lock(m1, m2);  // Автоматически захватывает все мьютексы
    
    // Критический раздел
}
 
// 3. Установка тайм-аута на блокировки
void timeout_operation() {
    std::unique_lock<std::mutex> lock(m1, std::defer_lock);
    if (lock.try_lock_for(std::chrono::milliseconds(100))) {
        // Успели захватить блокировку вовремя
    } else {
        // Обработка ситуации, когда блокировка не получена
    }
}

// Отладочная версия мьютекса с проверкой иерархии
class hierarchical_mutex {
    std::mutex mtx;
    const unsigned long hierarchy_value;
    unsigned long previous_hierarchy_value;
    static thread_local unsigned long this_thread_hierarchy_value;
    
public:
    explicit hierarchical_mutex(unsigned long value) : 
        hierarchy_value(value), 
        previous_hierarchy_value(0) {}
        
    void lock() {
        check_for_hierarchy_violation();
        mtx.lock();
        update_hierarchy_value();
    }
    
    void unlock() {
        this_thread_hierarchy_value = previous_hierarchy_value;
        mtx.unlock();
    }
    
    bool try_lock() {
        check_for_hierarchy_violation();
        if (!mtx.try_lock()) return false;
        update_hierarchy_value();
        return true;
    }
    
private:
    void check_for_hierarchy_violation() {
        if (this_thread_hierarchy_value <= hierarchy_value) {
            throw std::logic_error("hierarchy violation");
        }
    }
    
    void update_hierarchy_value() {
        previous_hierarchy_value = this_thread_hierarchy_value;
        this_thread_hierarchy_value = hierarchy_value;
    }
};
 
thread_local unsigned long hierarchical_mutex::this_thread_hierarchy_value(ULONG_MAX);

std::vector<double> data(10000000);
// Заполняем вектор...
 
// Последовательная сортировка
auto start = std::chrono::high_resolution_clock::now();
std::sort(data.begin(), data.end());
auto seq_time = std::chrono::high_resolution_clock::now() - start;
 
// Сбрасываем данные и выполняем параллельную сортировку
// ...
start = std::chrono::high_resolution_clock::now();
std::sort(std::execution::par, data.begin(), data.end());
auto par_time = std::chrono::high_resolution_clock::now() - start;
 
std::cout << "Ускорение: " << seq_time.count() / par_time.count() << "x\n";

// Применяем функцию к каждому элементу параллельно
std::transform(
    std::execution::par,
    genome_data.begin(), genome_data.end(),
    genome_data.begin(),
    [](const Nucleotide& n) {
        return process_nucleotide(n);
    }
);

class shared_mutex_guard {
private:
std::shared_mutex& mutex;
bool exclusive;
 
public:
shared_mutex_guard(std::shared_mutex& m, bool exclusive = false)
    : mutex(m), exclusive(exclusive) {
    if (exclusive) {
        mutex.lock();
    } else {
        mutex.lock_shared();
    }
}
 
~shared_mutex_guard() {
    if (exclusive) {
        mutex.unlock();
    } else {
        mutex.unlock_shared();
    }
}
 
// Запрещаем копирование
shared_mutex_guard(const shared_mutex_guard&) = delete;
shared_mutex_guard& operator=(const shared_mutex_guard&) = delete;
};

std::shared_mutex rwlock;
std::vector<Data> shared_data;
 
void reader_thread() {
    // Захватываем блокировку для чтения (могут быть и другие читатели)
    shared_mutex_guard guard(rwlock, false);
    for (const auto& item : shared_data) {
        process_data(item);
    }
}
 
void writer_thread() {
    // Эксклюзивная блокировка - никто не может читать или писать
    shared_mutex_guard guard(rwlock, true);
    shared_data.push_back(generate_new_data());
}

template<typename Mutex>
bool with_timeout_retry(Mutex& mtx, std::function<void()> action, 
                      int max_retries = 3) {
for (int attempt = 0; attempt < max_retries; ++attempt) {
    if (mtx.try_lock_for(std::chrono::milliseconds(100))) {
        std::lock_guard<Mutex> guard(mtx, std::adopt_lock);
        action();
        return true;
    }
    // Логирование или другое действие при неудачной попытке
    std::this_thread::sleep_for(
        std::chrono::milliseconds(50 * (attempt + 1))
    );
}
return false;
}

void robust_operation(std::mutex& mtx, int timeout_ms) {
std::unique_lock<std::mutex> lock(mtx, std::defer_lock);
if (!lock.try_lock_for(std::chrono::milliseconds(timeout_ms))) {
    // Логируем потенциальный дедлок
    report_potential_deadlock("Operation timed out, potential deadlock");
    
    // В крайнем случае можно аварийно завершить программу
    // std::abort();
    
    // Либо пробуем альтернативный путь без блокировки
    fallback_operation_without_lock();
}
else {
    // Нормальное выполнение с захваченным мьютексом
    protected_operation();
}
}

class DeadlockDetector {
private:
std::unordered_map<std::thread::id, std::set<std::mutex*>> thread_waits_for;
std::mutex detector_mutex;
 
public:
void thread_will_lock(std::mutex* m) {
    std::lock_guard<std::mutex> lock(detector_mutex);
    thread_waits_for[std::this_thread::get_id()].insert(m);
    check_for_cycles();
}
 
void thread_did_lock(std::mutex* m) {
    std::lock_guard<std::mutex> lock(detector_mutex);
    thread_waits_for[std::this_thread::get_id()].erase(m);
}
 
private:
void check_for_cycles() {
    // Реализация поиска циклов в графе зависимостей
    // ...
}
};

class JitterInjector {
private:
std::mt19937 rng;
std::uniform_int_distribution<> dist;
float probability;
 
public:
JitterInjector(float probability = 0.1) 
    : rng(std::random_device{}()), 
      dist(1, 100),
      probability(probability * 100) {}
      
void maybe_delay() {
    if (dist(rng) <= probability) {
        std::this_thread::sleep_for(
            std::chrono::milliseconds(dist(rng))
        );
    }
}
};
 
// Использование
void thread_function() {
    static JitterInjector jitter;
    
    for (int i = 0; i < 1000; ++i) {
        // Иногда вносим случайную задержку
        jitter.maybe_delay();
        
        // Обычные операции
        process_data();
    }
}

template<typename Func>
double measure_time(Func f, int repeat = 5) {
auto total = std::chrono::duration<double>(0);
    
for (int i = 0; i < repeat; ++i) {
    auto start = std::chrono::high_resolution_clock::now();
    f();
    auto end = std::chrono::high_resolution_clock::now();
    total += end - start;
}
    
return total.count() / repeat;
}
 
// Сравниваем разные реализации
auto sequential_time = measure_time([]{ sequential_algorithm(data); });
auto parallel_time = measure_time([]{ parallel_algorithm(data); });
auto parallel_opt_time = measure_time([]{ optimized_parallel_algorithm(data); });
 
std::cout << "Ускорение (базовое параллельное): " 
          << sequential_time / parallel_time << "x\n";
std::cout << "Ускорение (оптимизированное): " 
          << sequential_time / parallel_opt_time << "x\n";

std::atomic<int> x{0};
 
// Самый слабый порядок - минимум гарантий, максимум производительности
x.store(1, std::memory_order_relaxed);
 
// Гарантирует, что все предыдущие операции с памятью завершены
x.store(1, std::memory_order_release);
 
// Гарантирует, что все последующие операции с памятью начнутся после этой
int val = x.load(std::memory_order_acquire);
 
// Полная синхронизация — самые сильные гарантии, но наименьшая производительность
x.store(1, std::memory_order_seq_cst);  // По умолчанию

std::atomic<bool> data_ready{false};
int shared_data[10];
 
// Поток-писатель
void writer() {
    // Заполняем данные
    for (int i = 0; i < 10; ++i) {
        shared_data[i] = i*i;
    }
    
    // Устанавливаем флаг с порядком release - все предыдущие операции
    // станут видимы потоку, который сделает load с порядком acquire
    data_ready.store(true, std::memory_order_release);
}
 
// Поток-читатель
void reader() {
    // Ждём, используя acquire-семантику
    while (!data_ready.load(std::memory_order_acquire)) {
        std::this_thread::yield();
    }
    
    // Теперь можно безопасно читать данные
    for (int i = 0; i < 10; ++i) {
        assert(shared_data[i] == i*i);
    }
}

void process_image_better(Image& img, int num_threads) {
    // Разбиваем изображение на блоки, а не на строки
    const int block_size = 64;  // Размер блока, соответствующий кэш-линии
    const int width = img.width();
    const int height = img.height();
    
    std::vector<std::thread> threads;
    for (int t = 0; t < num_threads; ++t) {
        threads.emplace_back([&, t]() {
            // Каждый поток обрабатывает свой набор блоков
            for (int by = t; by < height/block_size; by += num_threads) {
                for (int bx = 0; bx < width/block_size; ++bx) {
                    // Обрабатываем один блок с хорошей локальностью
                    for (int y = by * block_size; 
                         y < (by+1) * block_size && y < height; ++y) {
                        for (int x = bx * block_size; 
                             x < (bx+1) * block_size && x < width; ++x) {
                            process_pixel(img, x, y);
                        }
                    }
                }
            }
        });
    }
    
    for (auto& t : threads) t.join();
}

void pin_thread_to_core(int core_id) {
#ifdef _GNU_SOURCE
    // Linux-специфичный код
    cpu_set_t cpuset;
    CPU_ZERO(&cpuset);
    CPU_SET(core_id, &cpuset);
    pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
#elif defined(_WIN32)
    // Windows-специфичный код
    SetThreadAffinityMask(GetCurrentThread(), (1ULL << core_id));
#endif
}
 
// Использование:
std::thread worker([core_id]() {
    pin_thread_to_core(core_id);
    
    // Работа потока...
});

#include <tbb/flow_graph.h>
 
void process_dependency_graph(const Graph& dependency_graph) {
    tbb::flow::graph g;
    
    // Создаём узлы для каждой задачи
    std::vector<tbb::flow::function_node<void, void>> nodes;
    for (int i = 0; i < dependency_graph.num_tasks(); ++i) {
        nodes.emplace_back(g, tbb::flow::unlimited, 
            [i](const tbb::flow::continue_msg&) {
                execute_task(i);
            });
    }
    
    // Устанавливаем связи между узлами согласно графу зависимостей
    for (const auto& edge : dependency_graph.edges()) {
        tbb::flow::make_edge(nodes[edge.from], nodes[edge.to]);
    }
    
    // Запускаем выполнение
    for (int i = 0; i < dependency_graph.num_tasks(); ++i) {
        if (dependency_graph.has_no_inputs(i)) {
            nodes[i].try_put(tbb::flow::continue_msg());
        }
    }
    
    g.wait_for_all();
}

#include <immintrin.h>
 
// Векторизация сложения массивов с помощью AVX
void simd_vector_add(float* a, float* b, float* result, size_t size) {
    // Обрабатываем по 8 элементов за раз с AVX
    for (size_t i = 0; i < size; i += 8) {
        __m256 va = _mm256_loadu_ps(&a[i]);
        __m256 vb = _mm256_loadu_ps(&b[i]);
        __m256 vr = _mm256_add_ps(va, vb);
        _mm256_storeu_ps(&result[i], vr);
    }
}

std::vector<float> a(10000), b(10000), result(10000);
// Заполняем векторы...
 
// Используем политику par_unseq для параллельной векторизации
std::transform(
    std::execution::par_unseq,
    a.begin(), a.end(),
    b.begin(),
    result.begin(),
    [](float x, float y) { return x + y; }
);

// Выравнивание данных для эффективной векторизации
alignas(32) float aligned_data[1024];  // Выравнивание по 32 байта для AVX
 
// Подсказка для компилятора (Intel)
#pragma ivdep
for (int i = 0; i < size; ++i) {
    result[i] = complex_calculation(input[i]);
}

#include <CL/sycl.hpp>
 
void matrix_multiply_gpu(const float* a, const float* b, float* c, 
                          int m, int n, int k) {
    cl::sycl::queue queue;
    
    cl::sycl::buffer<float, 2> buf_a({m, k}, a);
    cl::sycl::buffer<float, 2> buf_b({k, n}, b);
    cl::sycl::buffer<float, 2> buf_c({m, n}, c);
    
    queue.submit([&](cl::sycl::handler& h) {
        auto a_acc = buf_a.get_access<cl::sycl::access::mode::read>(h);
        auto b_acc = buf_b.get_access<cl::sycl::access::mode::read>(h);
        auto c_acc = buf_c.get_access<cl::sycl::access::mode::write>(h);
        
        h.parallel_for<class matrix_mult>(
            cl::sycl::range<2>{m, n},
            [=](cl::sycl::id<2> idx) {
                int row = idx[0];
                int col = idx[1];
                float sum = 0.0f;
                
                for (int i = 0; i < k; ++i) {
                    sum += a_acc[row][i] * b_acc[i][col];
                }
                
                c_acc[row][col] = sum;
            }
        );
    });
    queue.wait();
}

class TestBarrier {
    std::atomic<int> count{0};
    std::atomic<int> generation{0};
    int thread_count;
public:
    TestBarrier(int threads) : thread_count(threads) {}
    
    void sync_point(int id) {
        int gen = generation.load();
        if (count.fetch_add(1) == thread_count - 1) {
            count.store(0);
            generation.fetch_add(1);
        } else {
            while (generation.load() == gen) {
                std::this_thread::yield();
            }
        }
    }
};

void thread_func(int id, TestBarrier& barrier, SharedData& data) {
    // Этап 1: все потоки читают исходные данные
    auto local_copy = data.get_value();
    barrier.sync_point(id);
    
    // Этап 2: все одновременно модифицируют данные
    data.update(id, local_copy);
    barrier.sync_point(id);
    
    // Этап 3: проверка результатов
    verify_data(data);
}