异步在后台工作器中等待中介程序死锁-如何检测调用自身的线程 [英] Mediator deadlock on async await within background worker - how to detect thread calling itself
问题描述
我有一个调解器，最近需要在后台线程上一次将消息分发时进行同步，但它已锁定，如下所示。
I have a mediator which I have recently needed to synchronize one at a time message dispatch on a background thread but it is locking, demonstrated below.
我发布了一个命令到队列并从TaskCompletionSource返回任务：
I post a command to a queue and return a task from a 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;
}
然后从后台工作程序中拾取它，并创建处理程序：
It then is picked up from the background worker, and handler created:
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);
然后处理它，当命令处理程序发送到命令处理程序中时，下面的内容被锁定（当使用后台线程，但在线程内时可以）：
It is then handled, with the below locking up when the command handler is send to within the command handler (when using a background thread, but within thread it is OK):
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;
}
**问题和可能的解决方法**
** Question and potential fix **
我相信我们可以通过检测后台线程是否正在从其线程内部向自身发布（通过命令处理程序，然后调用Send（）来发布新命令）来避免死锁，如果是这样，它就不应使用任何线程机制（发布到命令队列或TaskCompletionSource），而应直接直接处理任务。
我已经尝试过来检测线程，但是它不起作用，所以我在 var otherResult =上面的处理程序中将手动标志isSameThread设置为true，等待this.commandBus.Send（new BoringCommand（），cancelleToken，true）和我可以确认它是否有效并且避免了僵局。
I have tried to detect the thread but it is not working, so i set the manual flag isSameThread to true within my handler above var otherResult = await this.commandBus.Send(new BoringCommand(), cancellationToken, true) and I can confirm it works and the deadlock is avoided.
此修复程序有任何警告吗？如何检测后台线程是否正在请求发送命令（线程如何检测自身）以及如何完成下面的代码（来自 DispatchOnBackgroundThread.Send（）包括此自调用检测（因此我可以取消isSameThread标志）？
Any caveats in this fix? How would one detect if the background thread is requesting to send a command (how can a thread detect itself) and how would one finish off the below code (from DispatchOnBackgroundThread.Send() to include this self-calling detection (so I can do away with the isSameThread flag)?
由于每次等待都将给出不同的线程，因此似乎涉及更多
It would seem this is more involved as each await will give a different thread 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;
}
}
**代码演示问题**
** Code demonstrating issue **
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();
}
}
}
**简单的非线程无需锁定即可工作的中介器实现**
** Simple non-threaded mediator implementation that works without locking **
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");
}
}
推荐答案
感谢对于斯蒂芬（Stephen）的回答和彼得（Peter）的评论，当您说谢谢的时候，确实确实让人眼花clear乱。
Thanks for Stephen for the answer and Peter for the comments, it indeed blindingly clear when stated thank you,
只有一个代码循环（不是特定线程；请参阅下文），由
负责处理队列。在处理每个命令时，
会等待该命令的处理程序。
There is one code loop (not a specific thread; see below) that is responsible for processing the queue. As it processes each command, it awaits that command's handler.
有一个命令处理程序正在等待另一个命令进行处理。
但是，这将无法工作，因为将不会再处理
进一步的命令；在此
完成之前，代码循环不会使下一个命令出队。
There is a command handler that awaits another command to be handled. However, this cannot work because no further commands will be processed; the code loop will not dequeue the next command until this one completes.
考虑到上述情况，我发现了一个
With the above in mind I have found a way to handle without any threading hacks (detecting stack/re-entrance depth etc) or schedulers.
在下面的示例中，我将注入到处理程序中而不是循环调用类，但不执行任何排队的另一种命令处理程序分派器，而是直接在线程内处理。
In the example below I "inject" into the handler not the looping calling class, but a different type of command handler dispatcher which does not do any queuing, it instead processes directly within the thread.
下面的内容是在线程循环内调用的，那么就没有相互依赖关系了。
The below is called from within the thread loop, then there is no inter-dependency:
public class DispatchInCallingThread: ICommandBus
{
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("cough furball");
}
public void Start(CancellationToken cancellationToken) { }
public Task StopAsync() { return Task.CompletedTask; }
}
在后台线程中，这是注入到实例化的命令处理程序中：
And within the background thread, this is the injection into the instantiated command handler:
else if (item.Command is LockMeGoodCommand lockMeGoodCommand)
{
var handler = new LockMeGoodCommandHandler(this.dispatchInCallingThread);
var result = await handler.Handle(lockMeGoodCommand, item.Ct);
item.Tcs.SetResult(result);
}
现在代码将永久运行（将需要为取消令牌实施正确的关闭逻辑
Now the code runs forever (will need to implement proper shutdown logic for cancellation token source being set):
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 ICommandBus commandBus;
public LockMeGoodCommandHandler(ICommandBus commandBus) => this.commandBus = commandBus;
public async Task<int> Handle(LockMeGoodCommand command, CancellationToken cancellationToken)
{
Console.WriteLine(command.GetType().Name);
var otherResult = await this.commandBus.Send(new BoringCommand(), cancellationToken);
var otherResult2 = await this.commandBus.Send(new BoringCommand(), cancellationToken);
return 3;
}
}
public interface ICommandBus
{
Task<object> Send(object request, CancellationToken cancellationToken);
void Start(CancellationToken cancellationToken);
Task StopAsync();
}
public class DispatchOnBackgroundThread : ICommandBus
{
private readonly Channel<CommandItem> queue = Channel.CreateUnbounded<CommandItem>();
private Task worker = null;
private readonly DispatchInCallingThread dispatchInCallingThread = new DispatchInCallingThread();
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, cancellationToken); // just write and immediatly return the cts
return item.Tcs.Task;
}
public void Start(CancellationToken cancellationToken)
{
var scheduler = new ConcurrentExclusiveSchedulerPair();
this.worker = Task.Factory.StartNew(async () =>
{
CommandItem item = null;
try
{
while (cancellationToken.IsCancellationRequested == false)
{
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 = handler.Handle(boringCommand, item.Ct);
item.Tcs.SetResult(result);
}
else if (item.Command is LockMeGoodCommand lockMeGoodCommand)
{
var handler = new LockMeGoodCommandHandler(this.dispatchInCallingThread);
var result = await handler.Handle(lockMeGoodCommand, item.Ct);
item.Tcs.SetResult(result);
}
else
throw new Exception("unknown");
}
}
catch (TaskCanceledException)
{
if (item != null)
item.Tcs.SetCanceled();
}
Console.WriteLine("exit background thread");
})
.Unwrap();
}
public async Task StopAsync()
{
this.queue.Writer.Complete();
await this.worker;
}
}
public class DispatchInCallingThread: ICommandBus
{
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");
}
public void Start(CancellationToken cancellationToken) { }
public Task StopAsync() { return Task.CompletedTask; }
}
class Program
{
static async Task Main(string[] args)
{
await TestDispatchOnBackgroundThread();
}
static async Task TestDispatchOnBackgroundThread()
{
var cts = new CancellationTokenSource();
Console.CancelKeyPress += delegate {
Console.WriteLine("setting cts.Cancel()");
cts.Cancel();
};
var threadStrategy = new DispatchOnBackgroundThread();
threadStrategy.Start(cts.Token);
while (cts.IsCancellationRequested == false)
{
Console.WriteLine("***************** sending new batch ****************");
var result1 = await threadStrategy.Send(new BoringCommand(), cts.Token);
var result3 = await threadStrategy.Send(new LockMeGoodCommand(), cts.Token);
Thread.Sleep(1000);
}
await threadStrategy.StopAsync();
}
}
}
有关详细信息，请参见 https://stackoverflow.com/a/61791817/915839 可以通过依赖项注入实现工作线程内的线程内分派
For further info, the actual implementation with dependency injection is here https://stackoverflow.com/a/61791817/915839 which was able to dynamically switch to in-thread dispatch within the worker thread
这篇关于异步在后台工作器中等待中介程序死锁-如何检测调用自身的线程的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
