c++ - 安全地同步 COM 线程

Question

我过去在多线程方面做过很多工作，但我对 COM 还很陌生。无论如何，这是我的问题:

我创建了一个工作线程，它注册为一个 STA，并创建了一个 COM 对象。然后工作线程和主线程尝试相互通信。使用 CoMarshalInterThreadInterfaceInStream 和 CoGetInterfaceAndReleaseStream，我可以让线程调用其他线程中 COM 对象的方法。

这是工作线程的样子:

void workerThread()
{
  CoInitialize(NULL);
  MyLib::IFooPtr foo = ...; // create my COM object

  // Marshall it so the main thread can talk to it
  HRESULT hr = CoMarshalInterThreadInterfaceInStream(foo.GetIID(),
                                                     foo.GetInterfacePtr(),
                                                     &m_stream);
  if (FAILED(hr)) {
    // handle failure
  }

  // begin message loop, to keep this STA alive
  MSG msg;
  BOOL bRet;
  while( (bRet = GetMessage( &msg, NULL, 0, 0 )) != 0)
  { 
    if (bRet == -1)  break;

    DispatchMessage(&msg); 
  }
}

在主线程中:

// launch the thread
m_worker = boost::thread (&workerThread);

// get the interface proxy
MyLib::IFooPtr foo;
LPVOID vp (NULL);
HRESULT hr = CoGetInterfaceAndReleaseStream(m_stream, foo.GetIID(), &vp);
if (SUCCEEDED(hr)) foo.Attach(static_cast<MyLib::IFoo*>(vp));

这会创建对象(这需要一段时间来初始化)，并允许主线程与其对话，并且一切都与 COM 公寓的东西正确同步。据我通过阅读 msdn 可以看出，这似乎是正确的做事方式。现在主线程可以使用它的代理来调用我的 COM 对象上的方法，工作线程将通过消息队列接收这些调用，并正确地调度它们。

但是，如何同步这些线程？

显然，在这种情况下，我希望主线程等待调用 CoGetInterfaceAndReleaseStream，直到工作线程通过 CoMarshalInterThreadInterfaceInStream 创建该流之后。但我怎样才能安全地做到这一点？

来自 MSDN ，我应该使用像 MsgWaitForMultipleObjects 这样的东西，所以我可以等待 my_condition 或 new_message_arrived，然后我可以做类似的事情:

// verbatim from msdn
while (TRUE)
{
   // wait for the event and for messages
   DWORD dwReturn = ::MsgWaitForMultipleObjects(1,
                     &m_hDoneLoading, FALSE, INFINITE, QS_ALLINPUT);

   // this thread has been reawakened. Determine why
   // and handle appropriately.
   if (dwReturn == WAIT_OBJECT_0)
     // our event happened.
     break ;
   else if (dwReturn == WAIT_OBJECT_0 + 1)
   {
     // handle windows messages to maintain
     // client liveness
     MSG msg ;
     while(::PeekMessage(&msg, NULL, 0, 0, PM_REMOVE))
       ::DispatchMessage(&msg) ;
   }
}

但是如何将 boost::thread.join() 和 boost::condition.wait() 与 MsgWaitForMultipleObjects 混合使用？这有可能吗，还是我必须做其他事情来避免竞争条件？

Answer 1

您的主线程有一个消息队列(必须是，因为是 STA 主机)，为什么不简单地向它发送消息，PostThreadMessage ？发布用户消息 (WM_USER +X)，您的普通主线程消息泵可以处理此用户消息，作为 COM 对象已将接口(interface)编码到流中并且主线程可以安全调用 CoGetInterfaceAndReleaseStream。

我必须指出，在您当前的设计中，您的工作线程基本上只是运行一个额外的消息泵。从主线程对接口(interface)上的任何方法的任何调用都将阻塞，等待工作线程从其消息队列中获取消息、处理调用、响应，然后主线程将恢复。所有操作至少与将 COM 对象托管在主线程中一样慢，再加上在两个 STA 之间来回编码 COM 的开销。由于 COM STA 的工作方式，两个线程之间基本上没有任何并发​​性。你确定这是你想要的吗？

编辑

(省略了一堆细节，如线程数、超时处理、为每个工作人员分配流/IID/CLSID 等)

在.h中:

HANDLE m_startupDone;
volatile int m_threadStartCount;

工作线程:

void workerThread()
{
  CoInitialize(NULL);
  MyLib::IFooPtr foo = ...; // create my COM object

  // Marshall it so the main thread can talk to it
  HRESULT hr = CoMarshalInterThreadInterfaceInStream(foo.GetIID(),
                                                     foo.GetInterfacePtr(),
                                                     &m_stream);
  if (FAILED(hr)) {
    // handle failure
    // remember to decrement and signal *even on failure*
  }

  if (0 == InterlockedDecrement(&m_threadStartCount))
  {
     SetEvent (m_startupDone);
  } 

  // begin message loop, to keep this STA alive
  MSG msg;
  BOOL bRet;
  while( (bRet = GetMessage( &msg, NULL, 0, 0 )) != 0)
  { 
    if (bRet == -1)  break;

    DispatchMessage(&msg); 
  }
}

在主线程中:

m_startupDone = CreateEvent (NULL, FALSE, FALSE, NULL);
m_threadStartCount = <number of workerthreads>

// launch the thread(s)
m_worker = boost::thread (&workerThread);
m_worker2 = boost::thread (&workerThread);
...

// now wait for tall the threads to create the COM object(s)
if (WAIT_OBJECT0 != WaitForSingleObject(m_startupDone, ...))
{
   // handle failure like timeout
}
// By now all COM objects are guaranteed created and marshaled, unmarshall them all in main
// here must check if all threads actually succeeded (could be as simple as m_stream is not NULL)

// get the interface proxy
MyLib::IFooPtr foo;
LPVOID vp (NULL);
HRESULT hr = CoGetInterfaceAndReleaseStream(m_stream, foo.GetIID(), &vp);
if (SUCCEEDED(hr)) foo.Attach(static_cast<MyLib::IFoo*>(vp));