c++ - 如何在 C++ 中创建高效的多线程任务调度程序？

Question

我想用 C++ 创建一个非常高效的任务调度系统。

基本思路是这样的:

class Task {
    public:
        virtual void run() = 0;
};

class Scheduler {
    public:
        void add(Task &task, double delayToRun);
};

在Scheduler之后，应该有一个固定大小的线程池，用来运行任务(我不想为每个任务创建一个线程)。 delayToRun 表示 task 不会立即执行，而是在 delayToRun 秒后执行(从添加到 的点开始测量调度程序)。

(delayToRun 当然是一个“至少”的值。如果系统已加载，或者如果我们向调度程序询问不可能，它将无法处理我们的请求.但它应该尽力而为)

这是我的问题。如何高效实现delayToRun功能？我正在尝试使用互斥锁和条件变量来解决这个问题。

我看到了两种方式:

带有管理线程

Scheduler 包含两个队列:allTask​​sQueue 和 tasksReadyToRunQueue。在 Scheduler::add 处将任务添加到 allTask​​sQueue。有一个管理线程，它等待的时间最短，因此它可以将任务从 allTask​​sQueue 放到 tasksReadyToRunQueue。工作线程等待 tasksReadyToRunQueue 中可用的任务。

如果 Scheduler::add 在 allTask​​sQueue 前面添加一个任务(一个任务，它的值为 delayToRun 所以它应该去在当前最快运行的任务之前)，则需要唤醒管理器任务，以便更新等待时间。

这种方法可以认为效率低下，因为它需要两个队列，并且需要两个 condvar.signals 才能使任务运行(一个用于 allTask​​sQueue->tasksReadyToRunQueue，和一个用于向工作线程发出实际运行任务的信号)

没有管理器线程

调度程序中有一个队列。在 Scheduler::add 处将任务添加到此队列中。工作线程检查队列。如果它是空的，它会在没有时间限制的情况下等待。如果不为空，则等待最快的任务。

1. 如果工作线程等待的条件变量只有一个:这种方法可以认为是低效的，因为如果队列前面增加了一个任务(前面的意思是，如果有N个工作线程，那么任务索引 所有工作线程需要被唤醒以更新他们正在等待的时间。

2. 如果每个线程有单独的条件变量，那么我们可以控制唤醒哪个线程，所以这种情况下我们不需要唤醒所有线程(我们只需要唤醒线程等待时间最长，所以我们需要管理这个值)。我目前正在考虑实现这一点，但确定确切的细节很复杂。有没有关于这种方法的建议/想法/文档？

---

这个问题有更好的解决方案吗？我正在尝试使用标准 C++ 功能，但如果它们提供更好的解决方案，我也愿意使用平台相关(我的主要平台是 linux)工具(如 pthreads)，甚至是 linux 特定工具(如 futexes)。

Answer 1

通过使用单个池线程等待“next to在一个条件变量上运行”任务(如果有的话)，其余的池线程无限期地等待第二个条件变量。

池线程将按照以下几行执行伪代码:

pthread_mutex_lock(&queue_lock);

while (running)
{
    if (head task is ready to run)
    {
        dequeue head task;
        if (task_thread == 1)
            pthread_cond_signal(&task_cv);
        else
            pthread_cond_signal(&queue_cv);

        pthread_mutex_unlock(&queue_lock);
        run dequeued task;
        pthread_mutex_lock(&queue_lock);
    }
    else if (!queue_empty && task_thread == 0)
    {
        task_thread = 1;
        pthread_cond_timedwait(&task_cv, &queue_lock, time head task is ready to run);
        task_thread = 0;
    }
    else
    {
        pthread_cond_wait(&queue_cv, &queue_lock);
    }
}

pthread_mutex_unlock(&queue_lock);

如果您更改下一个要运行的任务，则执行:

if (task_thread == 1)
    pthread_cond_signal(&task_cv);
else
    pthread_cond_signal(&queue_cv);

持有 queue_lock。

在这种方案下，所有的唤醒都直接在一个线程上，只有一个优先级任务队列，不需要管理线程。