performance - 使用内部同步实现作业列表

Question

我正在开发一个简单的作业线程框架，它与 id Tech 5 Challenges 中描述的框架非常相似。 .在最基本的层面上，我有一组作业列表，我想在一堆 CPU 线程中调度这些列表(使用标准线程池进行实际调度。)但是，我想知道这个信号/等待的东西在wait list里面可以有效地实现。据我了解，如果尚未执行信号 token ，则等待 token 会阻止列表执行。这隐含地意味着信号之前的一切都必须在信号被引发之前完成。所以假设我们有一个这样的列表:

J1, J2, S, J3, W, J4

那么调度可以这样进行:

#1: J1, J2, J3
<wait for J1, J2, run other lists if possible>
#2: J4

然而，这并不像看起来那么容易，因为给定一组列表，我必须在 ready 之间移动其中的一些。和 waiting并且还有特殊的代码来收集信号之前的所有作业并在它们上面标记一些东西，这样当且仅当它们都完成时它们才能触发信号(例如，这意味着不再可能将作业添加到列表中)被执行，因为以下信号访问先前插入的作业。)

有没有任何“标准”的方式来有效地实现这一点？我还想知道如何最好地安排作业列表执行，现在，每个核心获取一个作业列表，并安排其中的所有作业，这提供了非常好的扩展(对于 0.7 毫秒的 32k 作业，我得到 101%，我猜部分原因是单线程版本有时会被调度到不同的内核上。)

Answer 1

这是一种相对简单的调度算法。一些问题起初看起来很棘手，但实际上并非如此(信号/等待和缓存位置)。我将解释这些技术，然后给出我编写的一些代码来说明这些概念，最后给出一些关于调优的注释。

使用的算法

有效地处理信号/等待起初看起来很棘手，但实际上非常容易。由于信号/等待对不能嵌套或重叠，因此在任何给定时间实际上只能有两个被满足和一个被等待。只需保持一个指向最近未满足信号的“CurrentSignal”指针即可进行簿记。

确保内核不会在列表之间跳来跳去太多，并且给定列表不会在太多内核之间共享也相对容易:每个内核不断从同一个列表中获取作业，直到它阻塞，然后切换到另一个列表。为了防止所有内核聚集在一个列表中，每个列表都保留了一个 WorkerCount，用于说明有多少内核正在使用它，并且这些列表被组织起来，以便内核首先选择具有较少 worker 的列表。

锁定可以通过只锁定调度程序或您在任何时候处理的列表来保持简单，不要同时锁定两者。

您对在列表已经开始执行后将作业添加到列表表示了一些担忧。事实证明，支持这一点几乎是微不足道的:当将作业添加到当前已完成的列表时，它所需要的只是从列表调用调度程序，以便调度程序可以调度新作业。

数据结构

以下是您需要的基本数据结构:

class Scheduler
{
  LinkedList<JobList>[] Ready; // Indexed by number of cores working on list
  LinkedList<JobList> Blocked;
  int ReadyCount;
  bool Exit;

  public:
    void AddList(JobList* joblist);
    void DoWork();

  internal:
    void UpdateQueues(JobList* joblist);

    void NotifyBlockedCores();
    void WaitForNotifyBlockedCores();
}

class JobList
{
  Scheduler Scheduler;
  LinkedList<JobList> CurrentQueue;

  LinkedList<Job> Jobs;            // All jobs in the job list
  LinkedList<SignalPoint> Signals; // All signal/wait pairs in the job list,
                                      plus a dummy

  Job* NextJob;                    // The next job to schedule, if any
  int NextJobIndex;                // The index of NextJob

  SignalPoint* CurrentSignal;      // First signal not fully satisfied

  int WorkerCount;                 // # of cores executing in this list

  public:
    void AddJob(Job* job);
    void AddSignal();
    void AddWait();

  internal:
    void Ready { get; }
    void GetNextReadyJob(Job& job, int& jobIndex);
    void MarkJobCompleted(Job job, int jobIndex);
}
class SignalPoint
{
  int SignalJobIndex = int.MaxValue;
  int WaitJobIndex = int.MaxValue;
  int IncompleteCount = 0;
}

请注意，给定作业列表的信号点最方便地与实际作业列表分开存储。

调度器实现

调度程序跟踪作业列表，将它们分配给内核，并执行作业列表中的作业。

AddList 向调度程序添加作业。根据它是否有任何工作要做(即是否有任何作业已添加到其中)，它必须放在就绪或阻塞队列中，因此只需调用 UpdateQueues。

void Scheduler.AddList(JobList* joblist)
{
  joblist.Scheduler = this;
  UpdateQueues(joblist);
}

UpdateQueues 集中了队列更新逻辑。请注意选择新队列的算法，以及当工作可用时通知空闲核心:

void Scheduler.UpdateQueues(JobList* joblist)
{
  lock(this)
  {
    // Remove from prior queue, if any
    if(joblist.CurrentQueue!=null)
    {
      if(joblist.CurrentQueue!=Blocked) ReadyCount--;
      joblist.CurrentQueue.Remove(joblist);
    }

    // Select new queue
    joblist.CurrentQueue = joblist.Ready ? Ready[joblist.WorkerCount] : Blocked;

    // Add to new queue
    joblist.CurrentQueue.Add(joblist);
    if(joblist.CurrentQueue!=Blocked)
      if(++ReadyCount==1) NotifyBlockedCores();
  }
}

DoWork 是一个普通的调度程序工作，除了:1. 它选择 worker 最少的 JobList，2. 它从给定的作业列表处理作业，直到它不能再工作为止，以及 3. 它存储 jobIndex 以及作业，以便joblist 可以轻松更新完成状态(实现细节)。

void Scheduler.DoWork()
{
  while(!Exit)
  {
    // Get a job list to work on
    JobList *list = null;
    lock(this)
    {
      for(int i=0; i<Ready.Length; i++)
        if(!Ready[i].Empty)
        {
          list = Ready[i].First;
          break;
        }
      if(list==null)  // No work to do
      {
        WaitForNotifyBlockedCores();
        continue;
      }
      list.WorkerCount++;
      UpdateQueues(list);
    }

    // Execute jobs in the list as long as possible
    while(true)
    {
      int jobIndex;
      Job job;
      if(!GetNextReadyJob(&job, &jobIndex)) break;

      job.Execute();

      list.MarkJobCompleted(job, jobIndex);
    }

    // Release the job list
    lock(this)
    {
      list.WorkerCount--;
      UpdateQueues(list);
    }
  }
}

作业列表实现

JobList 跟踪信号/等待如何与作业穿插，并跟踪哪些信号/等待对在它们的信号点之前已经完成了所有事情。

构造函数创建一个虚拟信号点来添加作业。每当添加新的“信号”时，该信号点就会成为真正的信号点(并添加新的虚拟对象)。

JobList.JobList()
{
  // Always have a dummy signal point at the end
  Signals.Add(CurrentSignal = new SignalPoint());
}

AddJob 将作业添加到列表中。它在 SignalPoint 中被标记为不完整。作业实际执行时，同一个SignalPoint的IncompleteCount递减。还必须告诉调度程序事情可能已经改变，因为新作业可以立即执行。请注意，在释放“this”上的锁后调用调度程序以避免死锁。

void JobList.AddJob(Job job)
{
  lock(this)
  {
    Jobs.Add(job);
    Signals.Last.IncompleteCount++;
    if(NextJob == null)
      NextJob = job;
  }
  if(Scheduler!=null)
    Scheduler.UpdateQueues(this);
}

AddSignal 和 AddWait 将信号和等待添加到作业列表中。注意AddSignal实际上是新建了一个SignalPoint，AddWait只是在之前创建的SignalPoint中填写了等待点信息。

void JobList.AddSignal()
{
  lock(this)
  {
    Signals.Last.SignalJobIndex = Jobs.Count;  // Reify dummy signal point
    Signals.Add(new SignalPoint());            // Create new dummy signal point
  }
}


void JobList.AddWait()
{
  lock(this)
  {
    Signals.Last.Previous.WaitJobIndex = Jobs.Count;
  }
}

Ready 属性确定列表是否已准备好分配给它的其他核心。如果下一个作业在开始之前等待信号，则可能有两个或三个内核在列表上工作而列表未“准备好”。

bool JobList.Ready
{
  get
  {
    lock(this)
    {
      return NextJob!=null &&
        (CurrentSignal==Signals.Last ||
         NextJobIndex < CurrentSignal.WaitJobIndex);
    }
  }
}

GetNextReadyJob 非常简单:如果我们准备好了，只需返回列表中的下一个作业。

void JobList.GetNextReadyJob(Job& job, int& jobIndex)
{
  lock(this)
  {
    if(!Ready) return false;
    jobIndex = list.NextJobIndex++;
    job = list.NextJob; list.NextJob = job.Next;
    return true;

  }
}

MarkJobCompleted 可能是最有趣的。由于信号和等待的结构，当前作业要么在 CurrentSignal 之前，要么在 CurrentSignal 和 CurrentSignal 之间。 )。我们需要减少未完成作业的数量。如果此计数为零，我们可能还需要继续处理下一个信号。当然，我们从不最终通过虚拟 SignalPoint。

请注意，此代码没有调用 Scheduler.UpdateQueue，因为我们知道调度程序将在一秒钟内调用 GetNextReadyJob，如果它返回 false，它无论如何都会调用 UpdateQueue。

void JobList.MarkJobCompleted(Job job, int jobIndex)
{
  lock(this)
  {
    if(jobIndex >= CurrentSignal.SignalJobIndex)
      CurrentSignal.Next.IncompleteCount--;
    else
    {
      CurrentSignal.IncompleteCount--;
      if(CurrentSignal.IncompleteCount==0)
        if(CurrentSignal.WaitJobIndex < int.MaxValue)
          CurrentSignal = CurrentSignal.Next;
    }
  }
}

基于列表长度、作业长度估计等进行调整

上面的代码并没有关注作业列表的长度，所以如果有一百个小作业列表和一个巨大的作业列表，那么每个核心可能会分别采取一个单独的小作业列表，然后全部聚集在巨大的作业列表上。一，导致效率低下。这可以通过使 Ready[] 成为优先级队列的数组来解决 (joblist.Jobs.Count - joblist.NextJobIndex) ，但为了提高效率，优先级仅在正常的 UpdateQueue 情况下才实际更新。

通过创建考虑信号/等待组合的数量和间隔来确定优先级的启发式方法，这可以变得更加复杂。最好使用作业持续时间和资源使用的分布来调整这种启发式方法。

如果单个作业的持续时间是已知的，或者如果对它们有好的估计，那么启发式可以使用估计的剩余持续时间而不仅仅是列表长度。

最后说明

这是您提出的问题的相当标准的解决方案。您可以使用我提供的算法，它们会起作用，包括锁定，但由于以下几个原因，您将无法编译我上面编写的代码:

它是 C++ 和 C# 语法的疯狂组合。我最初开始用 C# 编写，然后将一堆语法更改为 C++ 样式，因为我认为这更有可能用于此类项目。但是我留下了很多 C#-isms。幸运的是没有 LINQ ;-)。

LinkedList 详细信息有一些挥手。我假设列表可以执行 First、Last、Add 和 Remove，并且列表中的项目可以执行 Previous 和 Next。但是我没有将实际的 API 用于我所知道的任何真正的链表类。

我没有编译或测试它。我保证某处有一两个错误。

底线:您应该将上面的代码视为伪代码，即使它看起来像真正的 McCoy。

享受!