Using I/O Completion Ports for Cross-Thread Messaging
|
Home |
 |
Back To Tips Page |
|
Using I/O Completion Ports for Cross-Thread Messaging |
|
You should never use SendMessage to send a message across thread boundaries. I discuss one alternative to this strategy in my essay on worker threads. However, this approach has its own problems.
For example, there is a serious defect in this when the data rate can be high, or bursty. You can get thousands of messages queued up in the input queue of your main GUI thread, and the result is that the mouse and keyboard become non-responsive. If you click the mouse, the mouse message is simply queued up behind hundreds or thousands of messages already in the queue. This gives the illusion that the app has crashed, because it is non-responsive.
In addition, the presence of all these messages interferes with the reception of WM_PAINT and WM_TIMER messages, which are only delivered to the message pump if there are no other pending messages.
I recently had to deal with a realtime input stream that could generate a lot of messages. This killed the user response.
My solution is one that I have not seen before: I used an I/O Completion Port to handle the message.
Consider a simple approach to sending a string across a thread boundary (this is from my essay):
In the thread which initiates the message:
CString * s = new CString;
s->Format(_T("The answer is %d"), answer);
wnd->PostMessage(UWM_MSG, (WPARAM)s);
In the receiving window:
ON_REGISTERED_MESSAGE(UWM_MSG, OnMyMessage)
LRESULT CMyView::OnMyMessage(WPARAM wParam, LPARAM)
{
CString * s = (CString *)wParam;
c_Log.AddString(*s);
delete s;
return 0;
}
This is clean, simple, and wrong1. Well, at least in the case of high data rates it is wrong, because it will glut up the main GUI thread.
1Albert Einstein remarked that "for every complex question there is a simple and wrong solution".
What I did was declare a wrapper for my message:
class MsgWrapper : public _OVERLAPPED {
public: // creation
MsgWrapper(CWnd * w, UINT m, WPARAM wP, LPARAM lP);
/* non-virtual */ ~MsgWrapper() { ...your code here... }
static BOOL Init();
static void Close();
public: // methods
BOOL Forward();
static MsgWrapper * GetQueuedMessage();
static BOOL Post(CWnd * wnd, UINT msg, WPARAM wParam = 0, LPARAM lParam = 0);
protected:
static HANDLE messagePort;
CWnd * wnd;
UINT msg;
WPARAM wParam;
LPARAM lParam;
};
The rationale behind this is that I need to pass a value through to represent the enqueued object. I must also pass an OVERLAPPED pointer. This combines the two.
In the implementation file, I do
/* static */ HANDLE MsgWrapper::messagePort = NULL;
/****************************************************************************
* MsgWrapper::MsgWrapper
* Inputs:
* CWnd * w: Window to which message is to be posted
* UINT m: Message ID
* WPARAM wP: message parameter
* LPARAM lP: message parameter
* Effect:
* Constructor
****************************************************************************/
MsgWrapper::MsgWrapper(CWnd * w, UINT m, WPARAM wP = 0, LPARAM lP = 0)
{
hEvent = NULL;
wnd = w;
msg = m;
wParam = wP;
lParam = lP;
}
/****************************************************************************
* MsgWrapper::Init
* Result: BOOL
* TRUE if successful
* FALSE if error
* Effect:
* Initializes the static members of the class
* Notes:
* An attempt to initialize more than once returns FALSE with ::GetLastError
* returning ERROR_INVALID_HANDLE
****************************************************************************/
/* static */ BOOL MsgWrapper::Init()
{
ASSERT(messagePort == NULL);
if(messagePort != NULL)
{/* already exists */
::SetLastError(ERROR_INVALID_HANDLE);
return FALSE;
} /* already exists */
messagePort = CreateIoCompletionPort(
INVALID_HANDLE_VALUE, // no file
NULL, // create it
0, // no key
1); // max threads
ASSERT(messagePort != NULL);
return messagePort != NULL;
}
/****************************************************************************
* MsgWrapper::Close
* Result: void
*
* Effect:
* Closes the handle
****************************************************************************/
/* static */ void MsgWrapper::Close()
{
if(messagePort != NULL)
::CloseHandle(messagePort);
messagePort = NULL;
}
/****************************************************************************
* MsgWrapper::GetQueuedMessage
* Result: MsgWrapper *
* NULL if the queue is empty
* Effect:
* Puts up wait cursor while doc/view is being created
****************************************************************************/
/* static */ MsgWrapper * MsgWrapper::GetQueuedMessage()
{
ASSERT(messagePort != NULL);
if(messagePort == NULL)
return NULL; // nothing to handle, no port exists
DWORD bytesRead; // ignored
ULONG_PTR key; // ignored
LPOVERLAPPED ovl; // MsgWrapper
if(!GetQueuedCompletionStatus(messagePort,
&bytesRead, // ignored
&key, // ignored
&ovl, // value to return
0)) // polled: 0 ms wait
{ /* no more */
return NULL;
} /* no more */
return (MsgWrapper *)ovl;
}
/****************************************************************************
* MsgWrapper::Post
* Inputs:
* CWnd * wnd: Window to which message is to be sent
* UINT m: Message to send
* WPARAM wParam: message parameter
* LPARAM lParam: message parameter
* Result: BOOL
* TRUE if successful
* FALSE if error
* Effect:
* Puts the message in the queue
****************************************************************************/
/* static */ BOOL MsgWrapper::Post(CWnd * wnd, UINT m, WPARAM wParam, LPARAM lParam)
{
ASSERT(messagePort != NULL);
if(messagePort == NULL)
return FALSE;
ASSERT(wnd != NULL); // must not be NULL! Must be a real window!
MsgWrapper * msg = new MsgWrapper(wnd, m, wParam, lParam);
return PostQueuedCompletionStatus(messagePort,
0, // bytes transferred
0,
msg);
}
/****************************************************************************
* MsgWrapper::Forward
* Result: BOOL
* TRUE if successful
* FALSE if error
* Effect:
* Forwards the message to the target and deletes it
****************************************************************************/
BOOL MsgWrapper::Forward()
{
BOOL result = wnd->PostMessage(msg, wParam, lParam));
delete this;
return result;
}
Note that it is extremely important that when you derive from a C struct, you should not add a virtual destructor, or any other virtual method. This is because the vtable, the virtual function pointer table, will be inserted in the structure as the first element. This will make it impossible to use the structure in a meaningful way.
Consider the following trivial program, illustrated the effects of virtual methods on a class. The offsetof macro is defined in stddef.h and returns the offset of a field within a structure.
#include "stdafx.h"
class NoVirtualMethods : public OVERLAPPED {
public:
~NoVirtualMethods() { }
public:
int whatever;
};
class VirtualMethods : public OVERLAPPED {
public:
virtual ~VirtualMethods() { }
public:
int whatever;
};
int _tmain(int argc, _TCHAR* argv[])
{
_tprintf(_T("NoVirtualMehods.whatever @ %d\n"),offsetof(class NoVirtualMethods, whatever));
_tprintf(_T("VirtualMethods.whatever @ %d\n"), offsetof(class VirtualMethods, whatever));
_tprintf(_T("NoVirtualMethods.Offset @ %d\n"), offsetof(class NoVirtualMethods, Offset));
_tprintf(_T("VirtualMethods.Offset @ %d\n"), offsetof(class VirtualMethods, Offset));
return 0;
}
which produces the following output:
Note that the offsets, such as the offset of a field like Offset or whatever are now incorrect. In particular, if the address of a VirtualMethods structure were passed to a method that expects an OVERLAPPED structure, the API call would not find the Offset value where it expected, in location 8, so incorrect results would be obtained (in fact, you would likely corrupt your file).
Now the way I use this is to create an I/O Completion port
BOOL CMyApp::InitInstance()
{
//... stuff here
MsgWrapper::Init();
//... more stuff here
}
int CMyApp::ExitInstance()
{
//... maybe stuff here
MsgWrapper::Close();
return CWinApp::ExitInstance();
}
Then I modify the OnIdle handler as shown below:
BOOL CMyApp::OnIdle(LONG lCount)
{
#define MAX_MESSAGE_QUANTUM 10
for(int i = 0; i < MAX_MESSAGE_QUANTUM; i++)
{ /* scan messages */
MsgWrapper * msg = MsgWrapper::GetQueuedMessage();
if(msg == NULL)
break;
msg->Forward();
} /* scan messages */
return CWinApp::OnIdle(lCount);
}
Note that nothing changes in how you write your handlers for your message; they are still normal ON_MESSAGE or ON_REGISTERED_MESSAGE type handlers (see my essay on Message Management for more details).
The trick here is the messages are not queued up in the message queue of the main GUI thread. By putting this code in the OnIdle handler, I only retrieve messages from the message queue if there is nothing else to do (so they come in lower than WM_PAINT and WM_TIMER). And then, I only retrieve some number (in this case, 10) at a time. Any mouse click or keyboard event that arrives has at most ten messages ahead of it. And as soon as there is something in the queue, OnIdle is no longer called, so it won't be called until the message queue is completely empty.
The above works because it use the OnIdle handler of the CWinApp-derived class. This only works for MDI and SDI apps. For dialog-based applications, this doesn't work because the message pump OnIdle invocation only works for the CWinThread class. For dialog-based apps, this dispatch happens in CWnd::RunModalLoop. So the above code works similarly in dialog-based apps, but the code is put in different places.
For one thing, it involves using the largely-undocumented WM_KICKIDLE message of folklore. And contrary to some parts of the legend, the handler should not be a void-returning method. Some folklore suggests that the message should be handled as
#include <afxpriv.h>
...
ON_MESSAGE_VOID(WM_KICKIDLE, OnKickIdle)
...
void CMyDialog::OnKickIdle()
{
... do your processing here
}
Unfortunately, this is erroneous advice. An examination of the code in CWnd::RunModalLoop (in wincore.cpp) shows that the critical code is
if ((dwFlags & MLF_NOKICKIDLE) ||
!SendMessage(WM_KICKIDLE, MSGF_DIALOGBOX, lIdleCount++))
{
// stop idle processing next time
bIdle = FALSE;
}
The point of this code is that idle messages will be sent until the OnKickIdle handler returns FALSE. But in our case, we need to keep processing idle messages until we run out of queued messages to process, so we have to return TRUE as long as there are messages. In addition, there are claims that no parameters are sent, but as can be clearly seen by the above code, there is a flag set indicating that the idle message is from a dialog box loop, and that there is an idle count comparable to the parameter of CWinApp::OnIdle. So don't be misled by parts of the folklore. Therefore, the code must be
#include <afxpriv.h>
ON_MESSAGE(WM_KICKIDLE, OnKickIdle)
LRESULT CMyDialog::OnKickIdle(WPARAM wParam, LPARAM lParam)
{
BOOL moretodo = FALSE; // for example, assume we only need to be called once
... and just maybe
moretodo = TRUE;
return (LRESULT)moretodo; // keep going if TRUE, stop if FALSE
}
Now, in my code, I don't care about the flags or the idle count, so the code resembles strongly the code of the OnIdle handler illustrated above:
LRESULT CMyDialog::OnKickIdle(WPARAM, LPARAM)
{
if(messagePort != NULL)
{ /* check for deferred messages */
DWORD bytesRead; // ignored
ULONG_PTR key; // ignored
LPOVERLAPPED ovl; // deferred message structure
#define MAX_MESSAGE_QUANTUM 10
for(int i = 0; i < MAX_MESSAGE_QUANTUM; i++)
{ /* scan messages */
MsgWrapper * msg = MsgWrapper::GetQueuedMessage();
if(msg == NULL)
return FALSE;
msg->wnd->PostMessage(msg->msg, msg->wParam, msg->lParam);
delete msg;
} /* scan messages */
return TRUE; // keep processing more
} /* check for deferred messages */
return FALSE; // stop idle processing unless we dispatched
} // CLocaleExplorerDlg::OnKickIdle
The I/O Completion Port must be created in the OnInitDialog handler for the main window dialog,
BOOL CMyDialog::OnInitDialog()
{
CDialog::OnInitDialog();
...
msgWrapper.Init();
Put the Close handler in the OnDestroy handler:
void CMyDialog::OnDestroy()
{
msgWrapper.Close();
}
This is a nice technique to avoid glutting up your main GUI queue when you have high-bandwidth thread-to-GUI communication. It does delay the processing of the messages (until the main GUI thread has time), and it will change the relative order of messages if some are sent via the traditional PostMessage route. But it had a significant impact on the perceived performance of a real-time system I did recently.
The above code has been extracted and massaged from an application that was considerably more complex. I have not had time to create a sample application. If you discover any problems with this code, let me know. When I have time to write a sample application, I will update this page.
1-Apr-06 Added description of how to add this to a dialog-based app.
Added caveat about no virtual methods.
20-Aug-07 Fixed numerous small bugs. Thanks to 'nobody' (yes, that's his
newsgroup handle) for pointing them out.
![[Dividing Line Image]](div.gif){kind=small}
The views expressed in these essays are those of the author, and in no way represent, nor are they endorsed by, Microsoft.