multithreading - NCompBus和WCF间歇性地无法完成TaskCompletionSource

Question

我对TaskCompletionSource有一个不寻常的问题，这让我感到困惑。我调用TaskSetResult后，有一个TaskCompletionSource等待任务完成。我在代码中的三个位置调用它:从WCF线程立即返回一个值到APM WCF BeginXXX EndXXX；从另一个WCF线程立即返回到APM；最后来自NServiceBus处理程序线程。

我从MS-PL提供的无处不在的ToAPM开始。 http://blogs.msdn.com/b/pfxteam/archive/2011/06/27/using-tasks-to-implement-the-apm-pattern.aspx

我注意到两个基于WCF的线程100％地起作用。在100小时的艰苦测试中，再加上广泛的单元测试中，我从未遇到过将完成的任务返回给AsyncCallback的任何失败。

在MS提供的ToAPM代码中，该代码在完成的任务上使用ContinueWith来在已启用计划的任务中调用AsyncCallback。

我尚 Unresolved 问题是NServiceBus线程在TaskCompletionSource对象上调用TrySetResult。我发现停电的时间，在不确定的时间段内，调用只是失败了。我在代码中为调用以及在ContinueWith代码中设置了断点。我总是在TrySetResult上获得断点，但有时仅在ContinueWith代码内的代码上得到断点。

以下信息有望对该问题有所启发。

我使用带有超时的CancellationTokenSource，并设置结果以在TaskCompletionSource obj上调用TrySetResult。当上述调用无法使任务完成时，将触发超时代码。此超时代码从未奏效。它成功100％的时间。

有趣的是，在从NServiceBus线程调用TrySetResult的同一代码中，当它起作用时，它就像在TaskCompletionSource obj上对TrySetResult一样容易地调用取消对象的Cancel。

当一个失败时，它们都会失败。

经过一段漫长的时间后，它又恢复了工作。

这是生产和QA环境中的WCF服务器，每个服务器显示相同的结果。

最奇怪的是，对于一个WCF连接，NServiceBus线程成功，而另一个同时失败。然后有时都工作，然后都失败了。同样，所有这些都在同一时间。

我尝试了多种方法来解决该问题，但均无济于事:

我将对TrySetResult的调用包装在TaskCompletionSource + ContinueWith中-失败

我将调用包装在Task.Factory.StartNew中-

失败

我直接调用它-失败

我真的不知道还能尝试什么。

我进行了检查，以确保TaskCompletionSource obj未完成，并且在停机期间未完成。
我进行了检查，以确保CancellationTokenSource对象没有被取消，或者在中断期间没有被取消的取消对象，但没有取消。

我检查了调试器中的对象，它们看起来不错。

他们只是有时不工作。

NserviceBus线程中是否存在有时导致调用无法正常工作的不一致？
我可以尝试一些线程编码(marshal)处理吗？

我到处搜索，却没有提到这一问题。它是独特的吗？

我完全困惑，需要一些想法。

Answer 1

从NServiceBus线程执行中删除该调用。使用诸如QueueUserWorkItem之类的线程或旋转自己的线程来隔离对TrySetResult的调用。由于执行恢复使用线程，因此您可能需要一些其他线程来处理吞吐量。醚旋转多个专用线程或使用线程池。我测试了在专用线程中调用TrySetResult，它们可以工作。

这是演示单个专用线程的代码:

    public static void Spin()
    {
        ClientThread = new Thread(new ThreadStart(() =>
        {
            while (true)
            {
                try
                {
                    if (!HasSomething.WaitOne(1000, false))
                        continue;

                    while (true)
                    {
                        WaitingAsyncData entry = null;
                        lock (qlocker)
                        {
                            if (!Trigger.Any())
                                break;

                            entry = Trigger.Dequeue();
                        }

                        if (entry == null)
                            break;

                        entry.TrySetResult("string");
                    }
                }
                catch
                {
                }
            }
        }));
        ClientThread.IsBackground = true;
        ClientThread.Start();
    }

这是ThreadPool示例代码:

    ThreadPool.QueueUserWorkItem(delegate
    {
        entry.TrySetResult("string");
    });

使用ThreadPool而不是静态线程可提供更大的灵活性和可伸缩性。