StaTaskScheduler 和 STA 线程消息泵 [英] StaTaskScheduler and STA thread message pumping
问题描述
TL;DR:StaTaskScheduler 运行的任务内部出现死锁. 长版:
我正在使用 StaTaskScheduler 来自 Parallel Team 的 ParallelExtensionsExtras，用于托管第三方提供的一些遗留 STA COM 对象.StaTaskScheduler 实现细节的描述如下:
好消息是 TPL 的实现能够运行在MTA 或 STA 线程，并考虑到相关差异底层 API，如 WaitHandle.WaitAll(仅支持 MTA当该方法提供多个等待句柄时线程).
我认为这意味着 TPL 的阻塞部分将使用泵送消息的等待 API，例如 CoWaitForMultipleHandles，避免在STA线程上调用时出现死锁情况.
在我的情况下，我相信以下情况正在发生:进程内 STA COM 对象 A 调用进程外对象 B，然后期望从 B 通过作为传出调用的一部分进行回调.
以简化形式:
var result = await Task.Factory.StartNew(() =>{//进程内对象Avar a = new A();//进程外对象 Bvar b = new B();//A调用B，B在Method调用期间回调A返回 a.Method(b);}, CancellationToken.None, TaskCreationOptions.None, staTaskScheduler);问题是，a.Method(b) 永远不会返回.据我所知，发生这种情况是因为 BlockingCollection 内某处的阻塞等待不会抽取消息，所以我对引用语句的假设可能是错误的.
EDITED 相同的代码在测试 WinForms 应用程序的 UI 线程上执行时有效(即提供 TaskScheduler.FromCurrentSynchronizationContext() 而不是 staTaskScheduler 到 Task.Factory.StartNew).
解决这个问题的正确方法是什么?我是否应该实现一个自定义同步上下文，该上下文将使用 CoWaitForMultipleHandles 显式泵送消息，并将其安装在由 StaTaskScheduler 启动的每个 STA 线程上?
如果是这样，BlockingCollection 的底层实现是否会调用我的 SynchronizationContext.Wait 方法?我可以使用 SynchronizationContext.WaitHelper 实现SynchronizationContext.Wait?<小时>已编辑，其中一些代码显示托管 STA 线程在进行阻塞等待时不会泵送.代码是一个完整的控制台应用程序，可以复制/粘贴/运行:
使用系统；使用 System.Collections.Concurrent;使用 System.Runtime.InteropServices；使用 System.Threading;使用 System.Threading.Tasks;命名空间 ConsoleTestApp{课程计划{//启动并运行一个 STA 线程静态无效 RunStaThread(布尔泵){//使用 BlockingCollection.Take 测试阻塞等待var tasks = new BlockingCollection();var thread = new Thread(() =>{//创建一个简单的 Win32 窗口var hwndStatic = NativeMethods.CreateWindowEx(0, "Static", String.Empty, NativeMethods.WS_POPUP,0, 0, 0, 0, IntPtr.Zero, IntPtr.Zero, IntPtr.Zero, IntPtr.Zero);//使用自定义的 WndProc 对其进行子类化IntPtr prevWndProc = IntPtr.Zero;var newWndProc = new NativeMethods.WndProc((hwnd, msg, wParam, lParam) =>{如果(味精== NativeMethods.WM_TEST)Console.WriteLine("WM_TEST 已处理");返回 NativeMethods.CallWindowProc(prevWndProc, hwnd, msg, wParam, lParam);});prevWndProc = NativeMethods.SetWindowLong(hwndStatic, NativeMethods.GWL_WNDPROC, newWndProc);if (prevWndProc == IntPtr.Zero)抛出新的应用程序异常()；//向其发送测试 WM_TEST 消息NativeMethods.PostMessage(hwndStatic, NativeMethods.WM_TEST, IntPtr.Zero, IntPtr.Zero);//BlockingCollection 阻塞而不抽，NativeMethods.WM_TEST 永远不会到达尝试 { var task = tasks.Take();}catch (Exception e) { Console.WriteLine(e.Message);}如果(泵){//NativeMethods.WM_TEST 会到达，因为 Win32 MessageBox 泵Console.WriteLine("现在开始抽...");NativeMethods.MessageBox(IntPtr.Zero, "Pumping Messages, press OK to stop...", String.Empty, 0);}});thread.SetApartmentState(ApartmentState.STA);线程.开始()；线程睡眠(2000)；//这会导致 STA 线程结束tasks.CompleteAdding();thread.Join();}静态无效主(字符串 [] args){Console.WriteLine("测试不抽...");RunStaThread(false);Console.WriteLine("
用抽水测试...");RunStaThread(真);Console.WriteLine("按回车退出");Console.ReadLine();}}//互操作静态类 NativeMethods{[DllImport("user32")]public static extern IntPtr SetWindowLong(IntPtr hwnd, int nIndex, WndProc newProc);[DllImport("user32")]public static extern IntPtr CallWindowProc(IntPtr lpPrevWndFunc, IntPtr hwnd, int msg, int wParam, int lParam);[DllImport("user32.dll")]public static extern IntPtr CreateWindowEx(int dwExStyle, string lpClassName, string lpWindowName, int dwStyle, int x, int y, int nWidth, int nHeight, IntPtr hWndParent, IntPtr hMenu, IntPtr hInstance, IntPtr lpParam);[DllImport("user32.dll")]public static extern bool PostMessage(IntPtr hwnd, uint msg, IntPtr wParam, IntPtr lParam);[DllImport("user32.dll")]public static extern int MessageBox(IntPtr hwnd, string text, String caption, int options);公共委托 IntPtr WndProc(IntPtr hwnd, int msg, int wParam, int lParam);public const int GWL_WNDPROC = -4;public const int WS_POPUP = unchecked((int)0x80000000);公共常量 int WM_USER = 0x0400；public const int WM_TEST = WM_USER + 1;}} 这会产生输出:
<前>不抽水测试...集合参数为空，并且已被标记为关于添加的完成.用抽水测试...集合参数为空，并且已被标记为关于添加的完成.现在开始抽...已处理 WM_TEST按 Enter 退出我对您问题的理解:您正在使用 StaTaskScheduler 仅用于为您的旧 COM 对象组织经典的 COM STA 单元.您未在StaTaskScheduler 的STA 线程上运行WinForms 或WPF 核心消息循环.也就是说，您没有在该线程内使用 Application.Run、Application.DoEvents 或 Dispatcher.PushFrame 之类的任何内容.如果这是一个错误的假设，请纠正我.
就其本身而言，StaTaskScheduler 不会在它创建的 STA 线程上安装任何同步上下文.因此，您依靠 CLR 为您抽取消息.我只在 CLR 中的公寓和抽水 作者:Chris Brumme:
我一直说托管阻塞将在以下情况下执行一些泵送"在 STA 线程上调用.确切地知道什么不是很好吗会被抽吗?不幸的是，抽水是一种黑色艺术超出凡人的理解.在 Win2000 及更高版本上，我们只需委托给OLE32 的 CoWaitForMultipleHandles 服务.
这表明 CLR 使用 CoWaitForMultipleHandles 内部用于 STA 线程.此外，COWAIT_DISPATCH_WINDOW_MESSAGES 标志的 MSDN 文档提及这个:
... 在 STA 中只是发送的一小组特殊情况的消息.
我做了对此进行了一些研究，但无法抽身带有 CoWaitForMultipleHandles 的示例代码中的 WM_TEST，我们在对您的问题的评论中讨论了这一点.我的理解是，上述一小组特殊情况的消息 实际上仅限于某些 COM 编组器特定的消息，并且不包括任何常规的通用消息，例如你的WM_TEST.
所以，回答你的问题:
<块引用>... 我应该实现一个自定义的同步上下文，这将使用 CoWaitForMultipleHandles 显式泵送消息，并安装它在由 StaTaskScheduler 启动的每个 STA 线程上?
是的，我相信创建自定义同步上下文并覆盖 SynchronizationContext.Wait 确实是正确的解决方案.
但是，您应该避免使用 CoWaitForMultipleHandles，并使用 MsgWaitForMultipleObjectsEx 代替.如果 MsgWaitForMultipleObjectsEx 指示队列中有待处理的消息，您应该使用 PeekMessage(PM_REMOVE) 和 DispatchMessage 手动泵送它.然后您应该继续等待句柄，所有这些都在同一个 SynchronizationContext.Wait 调用中.
注意 MsgWaitForMultipleObjectsEx 和 MsgWaitForMultipleObjects.如果在队列中已经看到一条消息(例如，使用 PeekMessage(PM_NOREMOVE) 或 GetQueueStatus)，则后者不会返回并保持阻塞，但未被删除.这对抽水不利，因为您的 COM 对象可能正在使用诸如 PeekMessage 之类的东西来检查消息队列.这可能会导致 MsgWaitForMultipleObjects 在非预期时阻塞.
OTOH, MsgWaitForMultipleObjectsEx 带有 MWMO_INPUTAVAILABLE 标志没有这样的缺点，在这种情况下会返回.
不久前我创建了一个自定义版本的 StaTaskScheduler (此处作为 ThreadAffinityTaskScheduler 提供)以尝试解决 不同的问题:为后续的 await 延续维护一个具有线程亲和性的线程池.如果您跨多个 await 使用 STA COM 对象，则线程关联至关重要.原StaTaskScheduler 仅当其池限制为 1 个线程时才会表现出此行为.
所以我继续对您的 WM_TEST 案例进行了更多实验.最初，我安装了一个标准的实例 SynchronizationContext STA 线程上的类.WM_TEST 消息没有被抽取，这是预期的.
然后我覆盖了 SynchronizationContext.Wait 将其转发到 SynchronizationContext.WaitHelper.它确实被调用了，但仍然没有泵.
最后，我实现了一个全功能的消息泵循环，这是它的核心部分:
//核心循环var msg = new NativeMethods.MSG();而(真){//带有 MWMO_INPUTAVAILABLE 的 MsgWaitForMultipleObjectsEx 返回，//即使在消息队列中已经看到但没有删除的消息nativeResult = NativeMethods.MsgWaitForMultipleObjectsEx(计数，等待句柄，(uint) 剩余超时，QS_MASK,NativeMethods.MWMO_INPUTAVAILABLE);if (IsNativeWaitSuccessful(count, nativeResult, out managedResult) || WaitHandle.WaitTimeout == managedResult)返回托管结果；//有一条消息，泵送并发送它if (NativeMethods.PeekMessage(out msg, IntPtr.Zero, 0, 0, NativeMethods.PM_REMOVE)){NativeMethods.TranslateMessage(ref msg);NativeMethods.DispatchMessage(ref msg);}如果(hasTimedOut())返回 WaitHandle.WaitTimeout;}这确实有效，WM_TEST 得到了提升.以下是您的测试的改编版本:
public static async Task RunAsync(){使用 (var staThread = new Noseratio.ThreadAffinity.ThreadWithAffinityContext(staThread: true, pumpMessages: true)){Console.WriteLine(初始线程#"+ Thread.CurrentThread.ManagedThreadId);等待 staThread.Run(async () =>{Console.WriteLine(在 STA 线程#"+ Thread.CurrentThread.ManagedThreadId);//创建一个简单的 Win32 窗口IntPtr hwnd = CreateTestWindow();//发布一些 WM_TEST 消息Console.WriteLine(发布一些 WM_TEST 消息...");NativeMethods.PostMessage(hwnd, NativeMethods.WM_TEST, new IntPtr(1), IntPtr.Zero);NativeMethods.PostMessage(hwnd, NativeMethods.WM_TEST, new IntPtr(2), IntPtr.Zero);NativeMethods.PostMessage(hwnd, NativeMethods.WM_TEST, new IntPtr(3), IntPtr.Zero);Console.WriteLine(按回车继续...");等待 ReadLineAsync();Console.WriteLine(等待后，线程#"+ Thread.CurrentThread.ManagedThreadId);Console.WriteLine("队列中的待处理消息:" + (NativeMethods.GetQueueStatus(0x1FF) >> 16 != 0));Console.WriteLine(退出 STA 线程#"+ Thread.CurrentThread.ManagedThreadId);}, CancellationToken.None);}Console.WriteLine(当前线程#"+ Thread.CurrentThread.ManagedThreadId);}输出:
<前>初始线程 #9在 STA 线程 #10发布一些 WM_TEST 消息...按 Enter 继续...WM_TEST 已处理:1WM_TEST 已处理:2WM_TEST 已处理:3等待后，线程 #10队列中的待处理消息:False退出 STA 线程 #10当前线程 #12按任意键退出请注意，此实现支持线程关联(它在 await 之后停留在线程 #10)和消息泵.完整的源代码包含可重用的部分(ThreadAffinityTaskScheduler 和 ThreadWithAffinityContext)，并且可用 这里是独立的控制台应用程序.它尚未经过彻底测试，因此请自行承担使用风险.
TL;DR: A deadlock inside a task run by StaTaskScheduler. Long version:
I'm using StaTaskScheduler from ParallelExtensionsExtras by Parallel Team, to host some legacy STA COM objects supplied by a third party. The description of the StaTaskScheduler implementation details says the following:
The good news is that TPL’s implementation is able to run on either MTA or STA threads, and takes into account relevant differences around underlying APIs like WaitHandle.WaitAll (which only supports MTA threads when the method is provided multiple wait handles).
I thought that would mean the blocking parts of TPL would use a wait API which pumps messages, like CoWaitForMultipleHandles, to avoid deadlock situations when called on an STA thread.
In my situation, I believe the following is happening: in-proc STA COM object A makes a call to out-of-proc object B, then expects a callback from B via as a part of the outgoing call.
In a simplified form:
var result = await Task.Factory.StartNew(() =>
{
// in-proc object A
var a = new A();
// out-of-proc object B
var b = new B();
// A calls B and B calls back A during the Method call
return a.Method(b);
}, CancellationToken.None, TaskCreationOptions.None, staTaskScheduler);
The problem is, a.Method(b) never returns. As far as I can tell, this happens because a blocking wait somewhere inside BlockingCollection<Task> does not pump messages, so my assumption about the quoted statement is probably wrong.
EDITED The same code works when is executed on the UI thread of the test WinForms application (that is, providing TaskScheduler.FromCurrentSynchronizationContext() instead of staTaskScheduler to Task.Factory.StartNew).
What is the right way to solve this? Should I implemented a custom synchronization context, which would explicitly pump messages with CoWaitForMultipleHandles, and install it on each STA thread started by StaTaskScheduler?
If so, will the underlying implementation of BlockingCollection be calling my SynchronizationContext.Wait method? Can I use SynchronizationContext.WaitHelper to implement SynchronizationContext.Wait?
EDITED with some code showing that a managed STA thread doesn't pump when doing a blocking wait. The code is a complete console app ready for copy/paste/run:
using System;
using System.Collections.Concurrent;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
namespace ConsoleTestApp
{
class Program
{
// start and run an STA thread
static void RunStaThread(bool pump)
{
// test a blocking wait with BlockingCollection.Take
var tasks = new BlockingCollection<Task>();
var thread = new Thread(() =>
{
// Create a simple Win32 window
var hwndStatic = NativeMethods.CreateWindowEx(0, "Static", String.Empty, NativeMethods.WS_POPUP,
0, 0, 0, 0, IntPtr.Zero, IntPtr.Zero, IntPtr.Zero, IntPtr.Zero);
// subclass it with a custom WndProc
IntPtr prevWndProc = IntPtr.Zero;
var newWndProc = new NativeMethods.WndProc((hwnd, msg, wParam, lParam) =>
{
if (msg == NativeMethods.WM_TEST)
Console.WriteLine("WM_TEST processed");
return NativeMethods.CallWindowProc(prevWndProc, hwnd, msg, wParam, lParam);
});
prevWndProc = NativeMethods.SetWindowLong(hwndStatic, NativeMethods.GWL_WNDPROC, newWndProc);
if (prevWndProc == IntPtr.Zero)
throw new ApplicationException();
// post a test WM_TEST message to it
NativeMethods.PostMessage(hwndStatic, NativeMethods.WM_TEST, IntPtr.Zero, IntPtr.Zero);
// BlockingCollection blocks without pumping, NativeMethods.WM_TEST never arrives
try { var task = tasks.Take(); }
catch (Exception e) { Console.WriteLine(e.Message); }
if (pump)
{
// NativeMethods.WM_TEST will arrive, because Win32 MessageBox pumps
Console.WriteLine("Now start pumping...");
NativeMethods.MessageBox(IntPtr.Zero, "Pumping messages, press OK to stop...", String.Empty, 0);
}
});
thread.SetApartmentState(ApartmentState.STA);
thread.Start();
Thread.Sleep(2000);
// this causes the STA thread to end
tasks.CompleteAdding();
thread.Join();
}
static void Main(string[] args)
{
Console.WriteLine("Testing without pumping...");
RunStaThread(false);
Console.WriteLine("
Test with pumping...");
RunStaThread(true);
Console.WriteLine("Press Enter to exit");
Console.ReadLine();
}
}
// Interop
static class NativeMethods
{
[DllImport("user32")]
public static extern IntPtr SetWindowLong(IntPtr hwnd, int nIndex, WndProc newProc);
[DllImport("user32")]
public static extern IntPtr CallWindowProc(IntPtr lpPrevWndFunc, IntPtr hwnd, int msg, int wParam, int lParam);
[DllImport("user32.dll")]
public static extern IntPtr CreateWindowEx(int dwExStyle, string lpClassName, string lpWindowName, int dwStyle, int x, int y, int nWidth, int nHeight, IntPtr hWndParent, IntPtr hMenu, IntPtr hInstance, IntPtr lpParam);
[DllImport("user32.dll")]
public static extern bool PostMessage(IntPtr hwnd, uint msg, IntPtr wParam, IntPtr lParam);
[DllImport("user32.dll")]
public static extern int MessageBox(IntPtr hwnd, string text, String caption, int options);
public delegate IntPtr WndProc(IntPtr hwnd, int msg, int wParam, int lParam);
public const int GWL_WNDPROC = -4;
public const int WS_POPUP = unchecked((int)0x80000000);
public const int WM_USER = 0x0400;
public const int WM_TEST = WM_USER + 1;
}
}
This produces the output:
Testing without pumping... The collection argument is empty and has been marked as complete with regards to additions. Test with pumping... The collection argument is empty and has been marked as complete with regards to additions. Now start pumping... WM_TEST processed Press Enter to exit
My understanding of your problem: you are using StaTaskScheduler only to organize the classic COM STA apartment for your legacy COM objects. You're not running a WinForms or WPF core message loop on the STA thread of StaTaskScheduler. That is, you're not using anything like Application.Run, Application.DoEvents or Dispatcher.PushFrame inside that thread. Correct me if this is a wrong assumption.
By itself, StaTaskScheduler doesn't install any synchronization context on the STA threads it creates. Thus, you're relying upon the CLR to pump messages for you. I've only found an implicit confirmation that the CLR pumps on STA threads, in Apartments and Pumping in the CLR by Chris Brumme:
I keep saying that managed blocking will perform "some pumping" when called on an STA thread. Wouldn’t it be great to know exactly what will get pumped? Unfortunately, pumping is a black art which is beyond mortal comprehension. On Win2000 and up, we simply delegate to OLE32’s CoWaitForMultipleHandles service.
This indicates the CLR uses CoWaitForMultipleHandles internally for STA threads. Further, the MSDN docs for COWAIT_DISPATCH_WINDOW_MESSAGES flag mention this:
... in STA is only a small set of special-cased messages dispatched.
I did some research on that, but could not get to pump the WM_TEST from your sample code with CoWaitForMultipleHandles, we discussed that in the comments to your question. My understanding is, the aforementioned small set of special-cased messages is really limited to some COM marshaller-specific messages, and doesn't include any regular general-purpose messages like your WM_TEST.
So, to answer your question:
... Should I implemented a custom synchronization context, which would explicitly pump messages with CoWaitForMultipleHandles, and install it on each STA thread started by StaTaskScheduler?
Yes, I believe that creating a custom synchronization context and overriding SynchronizationContext.Wait is indeed the right solution.
However, you should avoid using CoWaitForMultipleHandles, and use MsgWaitForMultipleObjectsEx instead. If MsgWaitForMultipleObjectsEx indicates there's a pending message in the queue, you should manually pump it with PeekMessage(PM_REMOVE) and DispatchMessage. Then you should continue waiting for the handles, all inside the same SynchronizationContext.Wait call.
Note there's a subtle but important difference between MsgWaitForMultipleObjectsEx and MsgWaitForMultipleObjects. The latter doesn't return and keeps blocking, if there's a message already seen in the queue (e.g., with PeekMessage(PM_NOREMOVE) or GetQueueStatus), but not removed. That's not good for pumping, because your COM objects might be using something like PeekMessage to inspect the message queue. That might later cause MsgWaitForMultipleObjects to block when not expected.
OTOH, MsgWaitForMultipleObjectsEx with MWMO_INPUTAVAILABLE flag doesn't have such shortcoming, and would return in this case.
A while ago I created a custom version of StaTaskScheduler (available here as ThreadAffinityTaskScheduler) in attempt to solve a different problem: maintaining a pool of threads with thread affinity for subsequent await continuations. The thread affinity is vital if you use STA COM objects across multiple awaits. The original StaTaskScheduler exhibits this behavior only when its pool is limited to 1 thread.
So I went ahead and did some more experimenting with your WM_TEST case. Originally, I installed an instance of the standard SynchronizationContext class on the STA thread. The WM_TEST message didn't get pumped, which was expected.
Then I overridden SynchronizationContext.Wait to just forward it to SynchronizationContext.WaitHelper. It did get called, but still didn't pump.
Finally, I implemented a full-featured message pump loop, here's the core part of it:
// the core loop
var msg = new NativeMethods.MSG();
while (true)
{
// MsgWaitForMultipleObjectsEx with MWMO_INPUTAVAILABLE returns,
// even if there's a message already seen but not removed in the message queue
nativeResult = NativeMethods.MsgWaitForMultipleObjectsEx(
count, waitHandles,
(uint)remainingTimeout,
QS_MASK,
NativeMethods.MWMO_INPUTAVAILABLE);
if (IsNativeWaitSuccessful(count, nativeResult, out managedResult) || WaitHandle.WaitTimeout == managedResult)
return managedResult;
// there is a message, pump and dispatch it
if (NativeMethods.PeekMessage(out msg, IntPtr.Zero, 0, 0, NativeMethods.PM_REMOVE))
{
NativeMethods.TranslateMessage(ref msg);
NativeMethods.DispatchMessage(ref msg);
}
if (hasTimedOut())
return WaitHandle.WaitTimeout;
}
This does work, WM_TEST gets pumped. Below is an adapted version of your test:
public static async Task RunAsync()
{
using (var staThread = new Noseratio.ThreadAffinity.ThreadWithAffinityContext(staThread: true, pumpMessages: true))
{
Console.WriteLine("Initial thread #" + Thread.CurrentThread.ManagedThreadId);
await staThread.Run(async () =>
{
Console.WriteLine("On STA thread #" + Thread.CurrentThread.ManagedThreadId);
// create a simple Win32 window
IntPtr hwnd = CreateTestWindow();
// Post some WM_TEST messages
Console.WriteLine("Post some WM_TEST messages...");
NativeMethods.PostMessage(hwnd, NativeMethods.WM_TEST, new IntPtr(1), IntPtr.Zero);
NativeMethods.PostMessage(hwnd, NativeMethods.WM_TEST, new IntPtr(2), IntPtr.Zero);
NativeMethods.PostMessage(hwnd, NativeMethods.WM_TEST, new IntPtr(3), IntPtr.Zero);
Console.WriteLine("Press Enter to continue...");
await ReadLineAsync();
Console.WriteLine("After await, thread #" + Thread.CurrentThread.ManagedThreadId);
Console.WriteLine("Pending messages in the queue: " + (NativeMethods.GetQueueStatus(0x1FF) >> 16 != 0));
Console.WriteLine("Exiting STA thread #" + Thread.CurrentThread.ManagedThreadId);
}, CancellationToken.None);
}
Console.WriteLine("Current thread #" + Thread.CurrentThread.ManagedThreadId);
}
The output:
Initial thread #9 On STA thread #10 Post some WM_TEST messages... Press Enter to continue... WM_TEST processed: 1 WM_TEST processed: 2 WM_TEST processed: 3 After await, thread #10 Pending messages in the queue: False Exiting STA thread #10 Current thread #12 Press any key to exit
Note this implementation supports both the thread affinity (it stays on the thread #10 after await) and the message pumping. The full source code contains re-usable parts (ThreadAffinityTaskScheduler and ThreadWithAffinityContext) and is available here as self-contained console app. It hasn't been thoroughly tested, so use it at your own risk.
这篇关于StaTaskScheduler 和 STA 线程消息泵的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！
