时间戳

首页 > 解决方案 > 后台工作人员中异步等待的中介死锁 - 如何检测线程调用自身

c# - 后台工作人员中异步等待的中介死锁 - 如何检测线程调用自身

问题描述

我有一个中介，我最近需要在后台线程上一次同步一个消息调度，但它是锁定的，如下所示。

我将命令发布到队列并从 TaskCompletionSource 返回任务：

public Task<object> Send(object command, CancellationToken cancellationToken)
{
    var item = new CommandItem() { Command = request, Tcs = new TaskCompletionSource<object>(), Ct = cancellationToken };            
    this.queue.Writer.WriteAsync(item); // just write and immediatly return the tcs
    return item.Tcs.Task;
}

然后从后台工作人员中获取它，并创建处理程序：

var item = await this.queue.Reader.ReadAsync(cancellationToken);
// work out command  type snipped
var command = item.Command as LockMeGoodCommand;
var handler = new LockMeGoodCommandHandler();
var result = await handler.Handle(command, item.Ct);
item.Tcs.SetResult(result);

然后处理它，当命令处理程序被发送到命令处理程序内时，以下锁定（当使用后台线程时，但在线程内是可以的）：

public async Task<int> Handle(LockMeGoodCommand command, CancellationToken cancellationToken)
{
   Console.WriteLine(command.GetType().Name);

   // this would get the result but will lock forever when using background worker bus implementation
   var otherResult = await this.commandBus.Send(new BoringCommand(), cancellationToken);

   // perform some action based on the result - but we never get here
   Console.WriteLine("otherResult is " + otherResult);

   return 3;
}

** 问题和潜在修复 **

我相信我们可以通过检测后台线程是否从其线程内向其自身发布（通过然后调用 Send() 以发布新命令的命令处理程序）来避免死锁，如果是，则不应使用任何线程机制（发布到命令队列或 TaskCompletionSource），而应该直接直接处理任务。

我试图检测线程但它不工作，所以我在上面的处理程序中将手动标志 isSameThread 设置为 true var otherResult = await this.commandBus.Send(new BoringCommand(), cancellationToken, true)，我可以确认它工作正常并且避免了死锁。

此修复中有任何警告吗？如何检测后台线程是否正在请求发送命令（线程如何检测自己）以及如何完成以下代码（从DispatchOnBackgroundThread.Send()包含此自调用检测（这样我就可以取消 isSameThread 标志)?

这似乎更复杂，因为每个等待都会给出不同的线程 ID。

// in thread start we set the thread id of the background thread
this.workerThreadId = System.Threading.Thread.CurrentThread.ManagedThreadId;

public Task<object> Send(object command, CancellationToken cancellationToken, bool isSameThread = false)
{
    Console.WriteLine($"this.workerThreadId: {this.workerThreadId}, Thread.CurrentThread.ManagedThreadId: {Thread.CurrentThread.ManagedThreadId}");

    // below doesnt work gives different numbers so i use flag instead
    // this.workerThreadId == Thread.CurrentThread.ManagedThreadId
    if (isSameThread == true)
    {
        if (command is BoringCommand boringCommand)
        {
            var handler = new BoringCommandHandler();
            return handler.Handle(boringCommand, cancellationToken).ContinueWith(t => (object)t);

        }
        else if (command is LockMeGoodCommand lockMeGoodCommand)
        {
            var handler = new LockMeGoodCommandHandler(this);
            return handler.Handle(lockMeGoodCommand, cancellationToken).ContinueWith(t => (object)t);
        }
        else
            throw new Exception("unknown");
    }
    else
    {
        var item = new CommandItem() { Command = command, Tcs = new TaskCompletionSource<object>(), Ct = cancellationToken };
        this.queue.Writer.WriteAsync(item); // just write and immediatly return the cts
        return item.Tcs.Task;
    }
}

**代码演示问题**

using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

namespace TestDeadlock
{
    class BoringCommand { }
    class LockMeGoodCommand { }    

    class BoringCommandHandler
    {
        public Task<int> Handle(BoringCommand command, CancellationToken cancellationToken)
        {
            Console.WriteLine(command.GetType().Name);         
            return Task.FromResult(1);
        }
    }
    class LockMeGoodCommandHandler
    {
        private readonly DispatchOnBackgroundThread commandBus;

        public LockMeGoodCommandHandler(DispatchOnBackgroundThread commandBus) => this.commandBus = commandBus;

        public async Task<int> Handle(LockMeGoodCommand command, CancellationToken cancellationToken)
        {
            Console.WriteLine(command.GetType().Name);

            // this locks forever
            var otherResult = await this.commandBus.Send(new BoringCommand(), cancellationToken);
            Console.WriteLine("otherResult is " + otherResult);
            return 3;
        }
    }

    public class DispatchOnBackgroundThread
    {
        private readonly Channel<CommandItem> queue = Channel.CreateUnbounded<CommandItem>();
        private Task worker = null;

        class CommandItem
        {
            public object Command { get; set; }
            public CancellationToken Ct { get; set; }
            public TaskCompletionSource<object> Tcs { get; set; }
        }

        public Task<object> Send(object command, CancellationToken cancellationToken)
        {
            var item = new CommandItem()
            { Command = command, Tcs = new TaskCompletionSource<object>(), Ct = cancellationToken };            
            this.queue.Writer.WriteAsync(item); // just write and immediatly return the tcs
            return item.Tcs.Task;
        }

        public void Start(CancellationToken cancellationToken)
        {
            this.worker = Task.Factory.StartNew(async () =>
            {
                try
                {                    
                    while (cancellationToken.IsCancellationRequested == false)
                    {
                        var item = await this.queue.Reader.ReadAsync(cancellationToken);

                        // simplified DI container magic to static invocation
                        if (item.Command is BoringCommand boringCommand)
                        {
                            var handler = new BoringCommandHandler();
                            var result = await handler.Handle(boringCommand, item.Ct);
                            item.Tcs.SetResult(result);
                        }
                        if (item.Command is LockMeGoodCommand lockMeGoodCommand)
                        {
                            var handler = new LockMeGoodCommandHandler(this);
                            var result = await handler.Handle(lockMeGoodCommand, item.Ct);
                            item.Tcs.SetResult(result);
                        }
                    }
                }
                catch (TaskCanceledException) { }
            },
            TaskCreationOptions.LongRunning)
            .Unwrap();
        }

        public async Task StopAsync()
        {
            this.queue.Writer.Complete();
            await this.worker;
        }
    }

    class Program
    {
        static async Task Main(string[] args)
        {
            var cts = new CancellationTokenSource();
            var threadStrategy = new DispatchOnBackgroundThread();
            threadStrategy.Start(cts.Token);

            var result1 = await threadStrategy.Send(new BoringCommand(), cts.Token);
            var result2 = await threadStrategy.Send(new LockMeGoodCommand(), cts.Token);

            cts.Cancel();
            await threadStrategy.StopAsync();
        }
    }
}

** 简单的非线程中介实现，无需锁定即可工作 **

public class DispatchInCallingThread
{
    public async Task<object> Send(object request, CancellationToken cancellationToken)
    {
        // simplified DI container magic to static invocation
        if (request is BoringCommand boringCommand)
        {
            var handler = new BoringCommandHandler();
            return await handler.Handle(boringCommand, cancellationToken);
        }
        else if (request is LockMeGoodCommand lockMeGoodCommand)
        {
            var handler = new LockMeGoodCommandHandler(this);
            return await handler.Handle(lockMeGoodCommand, cancellationToken);
        }
        else
            throw new Exception("unknown");
    }
}

标签： c#multithreadingconcurrencytask-parallel-librarymediator

解决方案

死锁的原因很简单：

  • 有一个代码循环（不是特定线程；见下文）负责处理队列。当它处理每个命令时，它await就是该命令的处理程序。
  • 有一个命令处理程序，它是要处理await的另一个命令。但是，这不起作用，因为不会处理更多命令；在此命令完成之前，代码循环不会使下一个命令出队。

换句话说，如果一个命令一次只能执行一个命令，那么一个命令执行另一个命令在逻辑上是不可能的。

有几种可能的方法可以解决这个问题。我不推荐“重入”的方式；重入是许多微妙的逻辑错误的原因。我推荐的方法是以下之一：

  1. 更改Send语义，使其成为“队列”语义。这意味着不可能得到命令结果；结果必须通过一些中介作为消息发送。
  2. 让代码循环而不是await命令处理程序，允许它循环返回并获取下一个命令。这意味着它不再“一次同步一个”。
  3. 将“一次同步一个”重新定义为“一次一个，但如果它正在await运行，则不算作一个”。在这种情况下，您可能可以使用类似的东西ConcurrentExclusiveSchedulerPair或Nito.AsyncEx.AsyncContext一次运行方法块。

旁注：LongRunning没有做你认为它正在做的事情。StartNewis not async-aware，因此该LongRunning标志仅适用于第一个代码await；之后，该 lambda 中的代码将在任意线程池线程（未LongRunning设置）上运行。替换StartNew为Task.Run将使代码更清晰。

推荐阅读