multithreading - 改进线程调度策略

Question

我需要帮助来增强我正在研究的线程调度策略。

背景

为了设置上下文，我有几个(20-30,000)个“任务”需要执行。每个任务都可以独立执行。实际上，任务的执行时间范围在 40 毫秒到 5 分钟之间变化。此外，重新运行时每个单独的任务都需要相同的时间。

我需要选项来控制这些任务的执行方式，所以我想出了一个调度引擎，可以根据各种策略来调度这些任务。最基本的策略是 FCFS，即我的任务一个接一个地按顺序执行。第二个是批处理策略，调度器有一个桶大小“b”，它控制有多少线程可以并行运行。调度程序将为它获得的第一个“b”任务启动非阻塞线程，然后等待那些启动的任务完成，然后继续下一个“b”任务，并行启动它们，然后等待完成。一次处理的每个“b”组任务称为一个批处理，因此称为批处理调度。

现在，有了批处理调度，事件在批处理开始时开始增加，当线程开始创建时，然后在中间达到峰值，此时大多数线程都在运行，然后随着我们阻塞和等待而减弱用于线程重新加入。当批处理大小“b”= 1 时，批处理调度变为 FCFS 调度。

改进批处理调度的一种方法是我将其称为并行调度——调度程序将确保，如果存在足够数量的任务，“b”个线程将在任何时间点保持运行。线程数最初会增加到“b”，然后将计数保持在“b”个运行线程，直到最后一组任务完成执行。为了随时保持“b”个线程的执行，我们需要在旧线程执行完的那一刻启动一个新线程。与批量调度(平均案例场景)相比，这种方法可以减少完成处理所有任务所花费的时间。

我需要帮助的地方

我必须实现并行调度的逻辑如下。如果有人可以帮助我，我将不胜感激:

• 我们能否避免使用启动任务列表？我正在用那个因为我需要确定什么时候Commit() 退出，所有任务都有完成执行，所以我只是循环通过所有 startedTasks 和 block 直到它们完成。一电流问题是列表会很长。

--或者--

• 是否有更好的并行调度方法？

(也欢迎任何其他建议/策略 - 这里的主要目标是在批大小“b”的限制内缩短整体执行时间)

ParallelScheduler 伪代码

// assume all variable access/updates are thread safe
Semaphore S: with an initial capacity of "b"
Queue<Task> tasks
List<Task> startedTasks
bool allTasksCompleted = false;

// The following method is called by a callee
// that wishes to start tasks, it can be called any number of times
// passing various task items
METHOD void ScheduleTask( Task t )

    if the PollerThread not started yet then start it 
    // starting PollerThead will call PollerThread_Action


    // set up the task so that when it is completed, it releases 1
    // on semaphore S
    // assume OnCompleted is executed when the task t completes
    // execution after a call to t.Start()
    t.OnCompleted() ==> S.Release(1)

    tasks.Enqueue ( t )


// This method is called when the callee
// wishes to notify that no more tasks are present that needs
// a ScheduleTask call.
METHOD void Commit()
    // assume that the following assignment is thread safe
    stopPolling = true;

    // assume that the following check is done efficiently
    wait until allTasksCompleted is set to true


// this is the method the poller thread once started will execute
METHOD void PollerThread_Action
    while ( !stopPolling )
        if ( tasks.Count > 0 )
            Task nextTask = tasks.Deque()
            // wait on the semaphore to relase one unit
            if ( S.WaitOne() )              
                // start the task in a new thread
                nextTask.Start()
                startedTasks.Add( nextTask )
    // we have been asked to start polling
    // this means no more tasks are going to be added
    // to the queue
    // finish off the remaining tasks
    while ( tasks.Count > 0 )
        Task nextTask = tasks.Dequeue()
        if ( S.WaitOne() )
            nextTask.Start()
            startedTasks.Add ( nextTask )

    // at this point, there are no more tasks in the queue
    // each task would have already been started at some
    // point
    for every Task t in startedTasks
        t.WaitUntilComplete() // this will block if a task is running, else exit immediately

    // now all tasks are complete
    allTasksCompleted = true

Answer 1

搜索“work stealing scheduler”——它是最高效的通用调度程序之一。周围还有一些开源和商业实现。

想法是拥有固定数量的工作线程，从队列中获取任务。但是为了避免所有线程共享的单个队列的拥塞(多 CPU 系统的性能问题非常糟糕)——每个线程都有自己的队列。当一个线程创建新任务时 - 它会将它们放入自己的队列中。完成任务后，线程从自己的队列中获取下一个任务。但是，如果线程的队列为空，它就会从其他线程的队列中“窃取”工作。