简单的自写 JobScheduler 导致死锁？答案

【问题标题】：Simple selfwritten JobScheduler causes deadlock?简单的自写 JobScheduler 导致死锁？
【发布时间】：2022-11-11 21:18:57
【问题描述】：

我为学习目的编写了自己的 JobScheduler。这个想法非常简单，它启动 n 个线程，从并发队列中提取作业/任务，处理它们，一旦完成，它将通知一个事件，以便主线程可以等待它完成（如果他愿意的话）。

线程循环看起来像这样......


    internal long ItemCount;  // The amount of jobs to process
    internal ManualResetEventSlim Event { get; set; }  // Event to notify worker threads for new items 
    internal ConcurrentQueue<JobMeta> Jobs { get; set; }  // Jobs

    private void Loop(CancellationToken token) {
        
        Loop:

        // Break if cancellation is requested
        if (token.IsCancellationRequested) return;
        
        // Make threads wait, the event tells them when new jobs arrived
        Event.Wait(token);
        if (Jobs.TryDequeue(out var jobMeta)) {  // Concurrent, dequeue one at a time

            // Make other threads wait once no more items are in the queue
            if(Interlocked.Decrement(ref ItemCount) == 0) Event.Reset();

            jobMeta.Job.Execute();      // Execute job
            jobMeta.JobHandle.Set(); // ManualResetEvent.Set to notify the main e.g.
        }
        goto Loop;
    }

    // Notify threads about new arrived jobs 
    public void NotifyThreads() {
        Interlocked.Exchange(ref ItemCount, Jobs.Count);  // Set ItemCount
        Event.Set();  // Notify
    }

    // Enqueues new job
    public JobHandle Schedule(IJob job) {

        var handle = new ManualResetEvent(false);
        var jobMeta = new JobMeta{ JobHandle = handle, Job = job};
        Jobs.Enqueue(jobMeta);
        return handle;
    }

但是，如果我执行以下操作，有时这会导致死锁：

var jobHandle = threadScheduler.Schedule(myJob);  // JobHandle is a ManualResetEvent
threadScheduler.NotifyThreads();

for(var index = 0; index < 10000; index++){
   
   var otherJobHandle = threadScheduler.Schedule(otherJob);
   threadScheduler.NotifyThreads();
   otherJobHandle.Wait();
}

jobHandle.Wait();  // Deadlock sometimes...

为什么这会导致死锁？逻辑问题在哪里？一个普通的 JobScheduler 会是什么样子（因为我一般找不到关于这个主题的任何好的信息）？

很高兴有任何帮助！

【问题讨论】：

使用BlockingCollection<T> 类可以大大简化您尝试做的事情，而不是使用低级别的ManualResetEventSlims 和ConcurrentQueue<T>s。
@TheodorZoulias 谢谢！但是我也遇到了阻塞集合的死锁......所以我想那部分不是问题：/
很难说问题出在哪里，因为您没有提供 MVCE ... fx Schedule 返回 JobHandle (public JobHandle Schedule(IJob job) 但您返回 ManualResetEvent ... 谁知道还有哪些其他代码丢失/更改这里
您是否尝试过用单个 BlockingCollection<JobMeta> 替换所有三个状态字段（long ItemCount、ManualResetEventSlim Event 和 ConcurrentQueue<JobMeta> Jobs），但您仍然遇到死锁？
@TheodorZoulias 正是......而且它变得越来越奇怪。只有当我将它作为发行版运行时...调试我当前的代码和带有blockingcollection的修改版本才能完美地工作。

标签： c# multithreading asynchronous deadlock

【解决方案1】：

我认为您的最后一个 sn-p 不会导致死锁，因为 jobHandle 首先入队，它的句柄应该已经设置。

我可以在您的实施中找到 2 个问题。首先是这一行：

Interlocked.Exchange(ref ItemCount, Jobs.Count);

考虑一下代码是否按以下顺序执行：

                         //queue.count = 1
1. Jobs.TryDequeue       //queue.count = 0
2. Interlocked.Exchange  //ItemCount = 0
3. Interlocked.Decrement //ItemCount = -1

ItemCount 减少了两次。所以你应该做的是在工作排队时增加ItemCount。

Interlocked.Increment(ref ItemCount);
Jobs.Enqueue(jobMeta);

第二个问题在循环中，想想这段代码执行顺序：

1. Jobs.TryDequeue
2. Interlocked.Decrement
  3. Interlocked.Increment
  3. Jobs.Enqueue
  3. Event.Set
4. Event.Reset

最后等待句柄的状态未设置，它可能会导致死锁，因为您的尝试是应该在排队作业后设置句柄。在这种情况下，您需要以相反的顺序运行代码：

1. Jobs.TryDequeue
2. Event.Reset // anyway reset the handle
3. if(Interlocked.Decrement > 0)
      Event.Set

使用这种方法，无论何时在Interlocked.Decrement 之前将作业入队，等待句柄的状态都是正确的。

上述方法可能不够高效，因为我们需要在每个循环中切换等待句柄，我建议您为队列添加一个 while-loop：

int count = 0;
while(Jobs.TryDequeue(out var jobMeta)) {
    ++count;
    ...
}

if(count > 0) {
    Event.Reset();
    if(Interlocked.Add(ref ItemCount, -count) > 0)
        Event.Set();
}

【讨论】：