C++ 中的工作窃取算法

2025 年 5 月 15 日 | 阅读 8 分钟

引言

在并行计算系统性能中，当存在多个处理器时，任务分配变得至关重要。工作窃取算法是一种适用于此环境的有效负载平衡方法。工作窃取方法允许已完成任务的线程“窃取”具有未完成任务的线程，从而保持系统平衡并减少空闲处理器。它现已广泛应用于支持并行性的框架中，例如 Intel 的 Threading Building Blocks (TBB) 和 Java 中的 ForkJoinPool。

在本文中，我们将首先概述工作窃取算法的概念，然后提供一个实现它的 C++ 代码。此外，还将讨论工作窃取算法的适当用法和限制。

问题陈述

在多线程模型中，一些线程会比其他线程更早完成它们的部分，因此一些资源将保持未使用状态。这就产生了如何在线程之间有效地平衡负载的问题，以便在其他线程有工作时，没有线程是空闲的。传统的任务分配方法通常使用集中式队列。这种方法也存在可伸缩性问题，因为可能会出现队列瓶颈。工作窃取算法的目标是创建一个分布式任务队列系统，空闲线程可以从其他线程的队列中“窃取”任务，以保持平衡并提高整体效率。

工作窃取算法概述

工作窃取算法通过允许每个线程拥有自己的双端队列（deque）来存储任务来操作。线程通常遵循以下规则：

每个线程主要在其自己的 deque 上操作，它以 LIFO（后进先出）顺序推送和弹出任务。
如果一个线程完成其任务且其 deque 为空，它会尝试从另一个线程的 deque 中“窃取”任务。
任务从 deque 的底部窃取，允许受害者线程将其最近的本地任务保留在顶部。
这种方法最大限度地减少了争用，因为大多数访问都是针对线程自己的 deque。

这种分散式方法允许更好的可伸缩性并避免单一的争用点，因为每个线程管理自己的任务队列，并且只有在任务用完时才与其他线程交互。

C++ 中的实现

以下是使用每个工作线程的 deque 实现工作窃取算法的一个简单 C++ 示例。这是一个用于说明目的的最小示例，缺少生产环境中所需的一些优化和错误处理。

程序 1

所需库

我们将使用 C++ 标准库的线程支持和容器进行实现。您可以使用 C++11 或更新的编译器编译此代码。

#include <iostream>
#include <deque>
#include <thread>
#include <mutex>
#include <vector>
#include <memory> // For std::unique_ptr
#include <atomic>

// Simple implementation of std::make_unique for C++11
template <typename T, typename... Args>
std::unique_ptr<T> make_unique(Args&&... args) {
    return std::unique_ptr<T>(new T(std::forward<Args>(args)...));
}

// A class representing a task that can be run by a thread
struct Task {
    int task_id;
    Task(int id) : task_id(id) {}
    void execute() const {
        std::cout << "Executing task " << task_id << " on thread " << std::this_thread::get_id() << std::endl;
    }
};

// Class representing the worker thread with a deque and a work-stealing mechanism
class Worker {
public:
    Worker() : stop_flag(false) {}

    // Adds a task to the worker's deque
    void push_task(Task task) {
        std::lock_guard<std::mutex> lock(mutex);
        deque.push_front(task);
    }

    // Attempts to steal a task from another worker
    std::unique_ptr<Task> steal_task() {
        std::lock_guard<std::mutex> lock(mutex);
        if (!deque.empty()) {
            Task task = deque.back();
            deque.pop_back();
            return make_unique<Task>(task);
        }
        return nullptr;
    }

    // Runs the worker, executing tasks or stealing if the deque is empty
    void run() {
        while (!stop_flag) {
            auto task = pop_task();
            if (!task) {
                // Attempt to steal from another worker
                for (Worker* other : *other_workers) {
                    if (other != this) {
                        task = other->steal_task();
                        if (task) break;
                    }
                }
            }

            if (task) {
                task->execute();
            } else {
                // No work available, so sleep briefly to reduce contention
                std::this_thread::yield();
            }
        }
    }

    // Sets a flag to stop the worker
    void stop() { stop_flag = true; }

    void set_other_workers(const std::vector<Worker*>& workers) {
        other_workers = &workers;
    }

private:
    // Pops a task from the deque
    std::unique_ptr<Task> pop_task() {
        std::lock_guard<std::mutex> lock(mutex);
        if (!deque.empty()) {
            Task task = deque.front();
            deque.pop_front();
            return make_unique<Task>(task);
        }
        return nullptr;
    }

    std::deque<Task> deque;
    std::mutex mutex;
    std::atomic<bool> stop_flag;
    const std::vector<Worker*>* other_workers;
};

int main() {
    const int num_threads = 4;
    std::vector<std::thread> threads;
    std::vector<Worker> workers(num_threads);

    // Set the other_workers vector for each worker for stealing
    std::vector<Worker*> worker_ptrs;
    for (auto& worker : workers) worker_ptrs.push_back(&worker);
    for (auto& worker : workers) worker.set_other_workers(worker_ptrs);

    // Launch threads
    for (int i = 0; i < num_threads; ++i) {
        threads.emplace_back([&workers, i] { workers[i].run(); });
    }

    // Distribute tasks to workers
    for (int i = 0; i < 10; ++i) {
        workers[i % num_threads].push_task(Task(i));
    }

    // Stop workers after a short delay
    std::this_thread::sleep_for(std::chrono::seconds(1));
    for (auto& worker : workers) {
        worker.stop();
    }

    // Join threads
    for (auto& thread : threads) {
        if (thread.joinable()) {
            thread.join();
        }
    }

    return 0;
}   

输出

Executing task 6 on thread 22639988467392
Executing task 2 on thread 22639988467392
Executing task 0 on thread 22639988467392
Executing task 4 on thread 22639988467392
Executing task 8 on thread 22639988467392
Executing task 1 on thread 22639988467392
Executing task 5 on thread 22639988467392
Executing task 9 on thread 22639988467392
Executing task 3 on thread 22639988467392
Executing task 7 on thread 22639986366144

代码解释

Worker 类：每个 worker 都有一个用于其任务的 deque、一个用于管理并发访问的互斥锁以及一个添加任务的方法 (push_task)。如果 worker 的 deque 为空，它会尝试从其他 worker 窃取任务。
任务执行：每个 worker 在循环中运行，从其 deque 中弹出任务并执行它们。如果未找到任务，它会尝试从其他 worker 窃取任务。stop_flag 用于控制 worker 何时停止操作。
任务窃取：当 worker 的 deque 中没有任务时，它会迭代其他 worker，尝试从其 deque 的底部窃取任务。如果没有可用的任务，它会短暂地放弃控制以减少 CPU 负载。
Main 函数：主函数初始化 worker 并启动一组线程，每个线程运行一个 worker。它还在 worker 之间分配一些初始任务。短暂延迟后，所有 worker 都停止，并连接线程。

程序 2

#include <iostream>
#include <deque>
#include <vector>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <unordered_map>
#include <functional>
#include <atomic>
#include <future>

// Represents a task with dependencies and priority
class Task {
public:
    using TaskFunc = std::function<void()>;

    Task(int id, TaskFunc func, int priority = 0)
        : task_id(id), func(std::move(func)), priority(priority), dependencies(0) {}

    void execute() const {
        func();
    }

    void add_dependency() {
        dependencies.fetch_add(1, std::memory_order_relaxed);
    }

    void resolve_dependency() {
        dependencies.fetch_sub(1, std::memory_order_relaxed);
    }

    bool is_ready() const {
        return dependencies.load(std::memory_order_relaxed) == 0;
    }

    int get_priority() const {
        return priority;
    }

    int get_id() const {
        return task_id;
    }

private:
    int task_id;
    TaskFunc func;
    int priority;
    std::atomic<int> dependencies;
};

// Represents a worker thread with a deque for task management
class Worker {
public:
    Worker() : stop_flag(false) {}

    void push_task(const std::shared_ptr<Task>& task) {
        std::lock_guard<std::mutex> lock(mutex);
        task_queue.push_front(task);
    }

    std::shared_ptr<Task> pop_task() {
        std::lock_guard<std::mutex> lock(mutex);
        if (!task_queue.empty()) {
            auto task = task_queue.front();
            task_queue.pop_front();
            return task;
        }
        return nullptr; // No task available
    }

    std::shared_ptr<Task> steal_task() {
        std::lock_guard<std::mutex> lock(mutex);
        if (!task_queue.empty()) {
            auto task = task_queue.back();
            task_queue.pop_back();
            return task;
        }
        return nullptr; // No task to steal
    }

    void set_other_workers(const std::vector<Worker*>& workers) {
        other_workers = &workers;
    }

    void run() {
        while (!stop_flag) {
            auto task = pop_task();

            if (!task) {
                for (Worker* other : *other_workers) {
                    if (other != this) {
                        task = other->steal_task();
                        if (task) break;
                    }
                }
            }

            if (task && task->is_ready()) {
                task->execute();
            } else {
                std::this_thread::yield();
            }
        }
    }

    void stop() {
        stop_flag = true;
    }

private:
    std::deque<std::shared_ptr<Task>> task_queue;
    std::mutex mutex;
    const std::vector<Worker*>* other_workers = nullptr;
    std::atomic<bool> stop_flag;
};

// Task Scheduler using Work-Stealing
class TaskScheduler {
public:
    TaskScheduler(size_t num_threads)
        : workers(num_threads), threads(num_threads) {
        initialize_workers();
    }

    ~TaskScheduler() {
        stop();
    }

    void submit_task(const std::shared_ptr<Task>& task) {
        if (!task->is_ready()) {
            throw std::runtime_error("Task has unresolved dependencies!");
        }
        assign_task_to_worker(task);
    }

    void resolve_dependency(int task_id) {
        auto it = task_map.find(task_id);
        if (it != task_map.end()) {
            auto& task = it->second;
            task->resolve_dependency();

            if (task->is_ready()) {
                assign_task_to_worker(task);
            }
        }
    }

    void run() {
        for (size_t i = 0; i < workers.size(); ++i) {
            threads[i] = std::thread([this, i] { workers[i].run(); });
        }
    }

    void stop() {
        for (auto& worker : workers) {
            worker.stop();
        }

        for (auto& thread : threads) {
            if (thread.joinable()) {
                thread.join();
            }
        }
    }

private:
    std::vector<Worker> workers;
    std::vector<std::thread> threads;
    std::unordered_map<int, std::shared_ptr<Task>> task_map;

    void initialize_workers() {
        std::vector<Worker*> worker_ptrs;
        for (auto& worker : workers) {
            worker_ptrs.push_back(&worker);
        }

        for (auto& worker : workers) {
            worker.set_other_workers(worker_ptrs);
        }
    }

    void assign_task_to_worker(const std::shared_ptr<Task>& task) {
        static size_t current_worker = 0;
        workers[current_worker % workers.size()].push_task(task);
        current_worker++;
    }
};

// Example usage
int main() {
    TaskScheduler scheduler(4);

    auto task1 = std::make_shared<Task>(1, []() {
        std::cout << "Task 1 executed on thread " << std::this_thread::get_id() << std::endl;
    });

    auto task2 = std::make_shared<Task>(2, []() {
        std::cout << "Task 2 executed on thread " << std::this_thread::get_id() << std::endl;
    });

    auto task3 = std::make_shared<Task>(3, []() {
        std::cout << "Task 3 executed on thread " << std::this_thread::get_id() << std::endl;
    });

    auto task4 = std::make_shared<Task>(4, []() {
        std::cout << "Task 4 executed on thread " << std::this_thread::get_id() << std::endl;
    });

    // Adding dependencies
    task2->add_dependency();
    task3->add_dependency();

    scheduler.submit_task(task1);
    scheduler.submit_task(task4);

    scheduler.run();

    // Resolving dependencies
    std::this_thread::sleep_for(std::chrono::seconds(1));
    scheduler.resolve_dependency(2);
    scheduler.resolve_dependency(3);

    scheduler.stop();

    return 0;
}   

输出

Task 1 executed on thread 140491566859968
Task 4 executed on thread 140491556370112

说明

任务类
- 它包含依赖关系，这些依赖关系包括依赖关系。
- 原子计数器用于跟踪依赖关系，并确保此类依赖关系的安全和线程安全的递减。
Worker 类
- 它们维护一个私有的本地任务双端队列，并根据需要从中获取任务。
- 任务在本地执行，或者在没有任务的情况下，甚至可以从其他工作线程中窃取。
TaskScheduler 类
- 监督分散在 worker 任务池中的 worker 线程网络。
- 监控提交的任务并实时解决这些任务。
主函数
- 展示了调度程序如何通过带有依赖关系和优先级的任务图来使用。
- 任务在清除其依赖关系之前不会执行。

优点和权衡

工作窃取算法可以很好地补充线程，因为它允许工作负载平衡，从而增加 CPU 利用率和空闲时间。

锁争用：即使算法允许大多数 deque 访问是本地的，但由于需要利用任务窃取概念，算法仍然允许大多数访问失效。
任务开销：任务切换和窃取过多会导致一种任务开销，这反过来使得工作窃取对于持续时间非常短或没有计算要执行的任务来说并不实用。

结论

总之，工作窃取是一种处理多线程应用程序中动态工作负载的绝佳方法。它允许将部分工作分配到分布式队列中，并让空闲线程帮助工作量过大的线程。它使解决方案更具资源效率和高度可伸缩性。尽管该算法有其缺点，但它已经变得流行，并且更适用于需要多样化和可调整任务的领域，因此它成为并行计算领域中广泛使用的算法。

下一主题C++ 中矩阵 I 中字母数字模式的匹配

C++ 中的工作窃取算法

引言

问题陈述

工作窃取算法概述

C++ 中的实现

程序 1

代码解释

程序 2

说明

优点和权衡

结论

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

C++ 教程

C++ 控制语句

C++ 函数

C++ 数组

C++ 类和对象

C++ 构造函数

C++ 继承

C++ 多态

C++ 抽象

C++ 命名空间

C++ 模板

C++ 字符串

C++ 指针

信号处理

C++ 异常

C++ 文件与流

C++ STL 教程

面试题

选择题

C++ 程序

C++ STL Stack

C++ STL Bitset

C++ STL Deque

C++ STL List

C++ STL Map

C++ STL Math

C++ STL priority_queue

C++ STL Queue

C++ STL Multiset

C++ STL Multimap

C++ STL Set

C++ STD Strings

C++ STL Vector

C++ 操纵符

C++ STL Algorithms

C++ Algorithm

C++ Iterators

C++ 杂项

C++ 中的工作窃取算法

引言

问题陈述

工作窃取算法概述

C++ 中的实现

程序 1

代码解释

程序 2

说明

优点和权衡

结论

相关帖子

C++ 中在七段显示器中使用最少段查找元素

C++ std::mem_fun_ref

C++ 中修改后的 ECMAScript 正则表达式语法

C++ 中的卢恩算法

C++ 中检查两个给定字符串的分割子字符串的连接是否构成回文

C++ 代码查找哈希冲突的索引

C++ 中的 std::get_money

C++ 中的摇晃排序

C++ 与 Erlang 的区别

C++ 中 DWORD 和无符号整型的区别