c# - 异步 TPL 死锁与第三方 lib aka wild goose chase

Question

在为此度过了非常令人沮丧且毫无成效的一天之后，我在这里发帖寻求帮助。

我正在使用以未知方式启动网络连接的第三方库(但我知道它是非托管库的托管包装器)。它通过调用事件让您了解连接状态 StatusChanged(status) .

显然调用网络的成本很高，我的 Service 可能不需要它, 我注入(inject)一个 AsyncLazy<Connection>然后在必要时调用它。 ParallelForEachAsync 访问该服务这是我为处理 Tasks 所做的扩展同时，基于 this post .

如果顺序访问，一切都很好。任何并发性，即使是 2 个并行任务也会在 90% 的时间内导致死锁。我知道这肯定与第三方库如何与我的代码交互有关，因为 a) 我无法使用相同的结构重现效果但不调用它和 b) 事件 StatusChanged(Connecting)接收正常，此时我假设网络操作已启动并且我从未收到 StatusChanged(Connected) 的回调.

这是代码结构的尽可能忠实的重现，不幸的是它没有重现死锁。

关于如何解决这个问题有什么想法吗？

class Program
{
    static void Main(string[] args)
    {
        AsyncContext.Run(() => MainAsync(args));
    }

    static async Task MainAsync(string[] args)
    {
        var lazy = new AsyncLazy<Connection>(() => ConnectionFactory.Create());
        var service = new Service(lazy);

        await Enumerable.Range(0, 100)
            .ParallelForEachAsync(10, async i =>
            {
                await service.DoWork();
                Console.WriteLine("did some work");
            }, CancellationToken.None);
    }
}

class ConnectionFactory
{
    public static Task<Connection> Create()
    {
        var tcs = new TaskCompletionSource<Connection>();
        var session = new Session();

        session.Connected += (sender, args) =>
        {
            Console.WriteLine("connected");
            tcs.SetResult(new Connection());
        };

        session.Connect();

        return tcs.Task;
    }
}

class Connection
{
    public async Task DoSomethinElse()
    {
        await Task.Delay(1000);
    }
}

class Session
{
    public event EventHandler Connected;

    public void Connect()
    {
        Console.WriteLine("Simulate network operation with unknown scheduling");
        Task.Delay(100).Wait();

        Connected(this, EventArgs.Empty);
    }
}

class Service
{
    private static Random r = new Random();
    private readonly AsyncLazy<Connection> lazy;

    public Service(AsyncLazy<Connection> lazy)
    {
        this.lazy = lazy;
    }

    public async Task DoWork()
    {
        Console.WriteLine("Trying to do some work, will connect");
        await Task.Delay(r.Next(0, 100));
        var connection = await lazy;
        await connection.DoSomethinElse();
    }
}

public static class AsyncExtensions
{
    public static async Task<AsyncParallelLoopResult> ParallelForEachAsync<T>(
            this IEnumerable<T> source,
            int degreeOfParallelism,
            Func<T, Task> body,
            CancellationToken cancellationToken)
    {
        var partitions = Partitioner.Create(source).GetPartitions(degreeOfParallelism);

        bool wasBroken = false;
        var tasks =
            from partition in partitions
            select Task.Run(async () =>
            {
                using (partition)
                {
                    while (partition.MoveNext())
                    {
                        if (cancellationToken.IsCancellationRequested)
                        {
                            Volatile.Write(ref wasBroken, true);
                            break;
                        }

                        await body(partition.Current);
                    }
                }
            });

        await Task.WhenAll(tasks)
            .ConfigureAwait(false);

        return new AsyncParallelLoopResult(Volatile.Read(ref wasBroken));
    }
}

public class AsyncParallelLoopResult
{
    public bool IsCompleted { get; private set; }

    internal AsyncParallelLoopResult(bool isCompleted)
    {
        IsCompleted = isCompleted;
    }
}

编辑

我想我明白为什么会这样，但不确定如何解决。当上下文正在等待 DoWork 时, DoWork正在等待惰性连接。

这个丑陋的 hack 似乎解决了它:

Connection WaitForConnection()
    {
        connectionLazy.Start();
        var awaiter = connectionLazy.GetAwaiter();
        while (!awaiter.IsCompleted)
            Thread.Sleep(50);
        return awaiter.GetResult();
    }

还有更优雅的解决方案吗？

Answer 1

我怀疑第 3 方库需要某种 STA 抽取。这在旧式异步代码中很常见。

我有一个 AsyncContextThread 类型，您可以尝试将 true 传递给构造函数以启用手动 STA 泵送。 AsyncContextThread 与 AsyncContext 类似，只是它在新线程(本例中为 STA 线程)中运行上下文。

static void Main(string[] args)
{
  using (var thread = new AsyncContextThread(true))
  {
    thread.Factory.Run(() => MainAsync(args)).Wait();
  }
}

或

static void Main(string[] args)
{
  AsyncContext.Run(() => async
  {
    using (var thread = new AsyncContextThread(true))
    {
      await thread.Factory.Run(() => MainAsync(args));
    }
  }
}

请注意，AsyncContextThread 不会在所有 STA 场景中工作。我在做(一些相当扭曲的)需要真正的 UI 线程(WPF 或 WinForms 线程)的 COM 互操作时遇到了问题；出于某种原因，STA 泵送对于那些 COM 对象来说是不够的。