A C++ Wrapper for WaitForMultipleObjects API

• Download demo - 15.1 KB
• Download source - 4.2 KB

Introduction

The WaitForMultipleObjects Win32 API is a powerful construct that can be used to write very efficient and easy to maintain multithreaded applications. The API allows you to implement complex software consisting of independent units of execution logic using a single worker thread that can detect and process multiple wait handles’ transition to signaled state instead of spawning individual worker threads for each of them. However, marrying this API to C++ objects requires some plumbing code to translate the detection of the signalled state into the associated C++ object method call. In this article, I’ll show how a base class can abstract and hide this logic and provide the client classes with a natural C++ facade to easily associate a wait handle with a specific C++ object’s method.

Background 

Though WaitForMultipleObjects provides an effective mechanism for writing efficient multithreaded programs, the related code that involves the API call and processing of its return value to take the appropriate action is often repeated across multiple modules. Besides involving repetitive code, such an approach is not very flexible as adding or removing a wait handle to existing list could require rearranging of the existing code. Also, the code that deals with the calling and processing of the API’s return value are mixed together with the program’s functional logic reducing code readability and consequently its maintainability. A typical implementation usually looks like:

class MyDaemon {
public:
    MyDaemon()
    {}
    void Start() {
        // initialize resources and spawn the worker thread
    }
    void Stop() {
        // set shutdown event and wait on worker thread
    }
private:
    void HandleEventResource() { // functional logic
    }
    void HandleSemaphoreResource() { // functional logic
    }
    // worker thread 
    unsigned int ThreadProc() {
        bool fMore = true;
        HANDLE aHandles[4] = {0}
        aHandle[0] = m_hShutdown;
        aHandle[1] = m_hEvent;
        aHandle[2] = m_hSemaphore;
        do {
            DWORD dwRet = ::WaitForMultipleObjects(
                    _countof(aHandle), aHandle, FALSE, INFINITE, TRUE);
            switch (dwRet) {
            case WAIT_OBJECT_0:	// m_hShutdown signalled
                fMore = false;
                break;
            case WAIT_OBJECT_0+1: // m_hEvent signalled
                HandleEventResource();
                break;
            case WAIT_OBJECT_0+2: // m_hSemaphore released
                HandleSemaphoreResource();
                break;
            default:
                fMore = false;	// error
                break;
            }
        } while (fMore);
        return 0;
    }
    static unsigned int __stdcall _ThreadProc(void* p) {
        _ASSERTE(p != NULL);
        return reinterpret_cast<WFMOHandler*>(p)->ThreadProc();
    }
private:
    HANDLE m_hShutdown; // thread shutdown event
    HANDLE m_hEvent;
    HANDLE m_hSemaphore;
};

Here the WaitForMultipleObjects API call and the handling of its return value is mixed with the functional logic of the MyDaemon class which constitutes the main application logic. If one were required to monitor more wait handles, the worker thread body has to be modified to accommodate the additional handles and process their respective return values. And as more and more handles are added, the switch-case body statement gets larger and larger thereby making it harder to maintain and increasing the chances of an inadvertent error due to a typo creeping in.

A typical solution that is often employed is to use a table (std::map) that stores the handles to be waited upon and their corresponding handler class method pointers. For example:

// in the class declaration
typedef void (MyDaemon ::*PFNHANDLER)();
typedef std::map<HANDLE, PFNHANDLER> HANDLERTABLE;
HANDLERTABLE m_handlers;

// in the thread body
std::vector<HANDLE> handles;
handles.resize(m_handlers.size());
int i=0;
for (auto it=m_handlers.begin(); it!=m_handlers.end(); it++)
	handles[i++] = it->first;
DWORD dwRet = ::WaitForMultipleObjectsEx(
                  handles.size(), &handles[0], FALSE, INFINITE, TRUE);
if (dwRet >= WAIT_OBJECT_0 && dwRet < (WAIT_OBJECT_0+handles.size())) {
	(*this.*m_handlers[handles[dwRet-WAIT_OBJECT_0]](); // invoke the handler
}

Though this will help avoid the long switch-case body and hard coding of the return value processing, it also poses some constraints. All the handler methods have to have the same type signature and handlers have to be methods of the same class. You can mitigate this by using pure C++ function pointers instead, but then to translate that into a C++ object’s method call, thunking code has to be provided, wherever such translation is needed.

In medium-to-large scale software, this pattern of code can be seen repeated across multiple modules, wherever the API is employed.

It’s this pattern (essentially the worker thread body) that I refer to as the WaitForMultipleObjects plumbing logic which is factored out as an independent class that can be reused with minimum effort. Needless to say, abstracting this code as an independent class brings all the classical benefits of modularization -- makes the application code cleaner, thereby increasing code readability and maintainability, and any enhancements to the base class code is immediately reflected to all its client classes.

Using the Code

The WaitForMultipleObjects API call and related logic are encapsulated in the class WFMOHandler which is contained in its entirety in wfmohandler.h. There is no corresponding CPP file. This class has an internal worker thread that blocks on WaitForMultipleObjects waiting for any of the registered wait handles to be signalled. And when one is signalled, the relevant handler is invoked.

To use, declare your class deriving it from WFMOFHandler and register the wait handle whose state you want to track as is show in the example below. Registration is done through the AddWaitHandle method. This method takes two arguments -- the wait handle and its handler function which is to be called when the handle is signaled. The handler function is to be provided as a function object. You can provide an explicitly created function object or you can compose one on the fly using the STL std::bind() facility. Both these two usages are shown below:

#include "wfmohandler.h"

// A simple class that implements an asynchronous 'recv' UDP socket.
// Socket binds to loopback address!
class AsyncSocket {
    USHORT m_port;
    WSAEVENT m_event;
    SOCKET m_socket;
public:
    AsyncSocket(USHORT port)
        : m_port(port)
        , m_event(::WSACreateEvent())
        , m_socket(::WSASocket(AF_INET, SOCK_DGRAM, IPPROTO_UDP, NULL, 0, 0))
    {
        // bind the socket
        struct sockaddr_in sin = {0};
        sin.sin_family = AF_INET;
        sin.sin_port = ::htons(port);
        sin.sin_addr.s_addr = ::inet_addr("127.0.0.1");
        if (m_event != NULL && m_socket != INVALID_SOCKET 
            && ::bind(m_socket, reinterpret_cast<const sockaddr*>(&sin), sizeof(sin)) == 0) {
            // put it in 'async' mode
            if (::WSAEventSelect(m_socket, m_event, FD_READ) == 0)
                return;
        }
        std::cerr << "Error initializing AsyncSocket, error code: " << WSAGetLastError() << std::endl;
        // something went wrong, release resources and raise an exception
        if (m_event != NULL) ::WSACloseEvent(m_event);
        if (m_socket != INVALID_SOCKET) ::closesocket(m_socket);
        throw std::exception("socket creation error");
    }
    ~AsyncSocket()
    {
        ::closesocket(m_socket);
        ::WSACloseEvent(m_event);
    }
    // for direct access to the embedded event handle
    operator HANDLE() { return m_event; }
    // Read all incoming packets in the socket's recv buffer. When all the packets 
    // in the buffer have been read, resets the associated Win32 event preparing it
    // for subsequent signalling when a new packet is copied into the buffer.
    void ReadIncomingPacket()
    {
        std::vector<char> buf(64*1024);
        struct sockaddr_in from = {0};
        int fromlen = sizeof(from);
        int cbRecd = ::recvfrom(m_socket, &buf[0], buf.size(), 
            0, reinterpret_cast<sockaddr*>(&from), &fromlen);
        if (cbRecd > 0) {
            std::cerr << cbRecd << " bytes received on port " << m_port << std::endl;
        } else {
            int rc = ::WSAGetLastError();
            if (rc == WSAEWOULDBLOCK) {
                // no more data, reset the event so that WaitForMult..will block on it
                ::WSAResetEvent(m_event);
            } else {
                // something else went wrong
                std::cerr << "Error receiving data from port " << m_port 
                      << ", error code: " << ::WSAGetLastError() << std::endl;
            }
        }
    }
};

class MyDaemon : public WFMOHandler {
    AsyncSocket m_socket1;
    AsyncSocket m_socket2;
    AsyncSocket m_socket3;
    class AnotherEventHandler { // an explicit functor
        AsyncSocket& m_socket;
    public:
        AnotherEventHandler(AsyncSocket& sock) : m_socket(sock) {}
        void operator()() {
            m_socket.ReadIncomingPacket(); // handle incoming socket data
        }
    };
    AnotherEventHandler m_aeh;
public:
    MyDaemon() 
        : WFMOHandler()
        , m_socket1(5000)
        , m_socket2(6000)
        , m_socket3(7000)
        , m_aeh(m_socket3)
    {
        // setup two handlers on the two AsyncSockets that we created
        WFMOHandler::AddWaitHandle(m_socket1, 
            std::bind(&AsyncSocket::ReadIncomingPacket, &m_socket1));
        WFMOHandler::AddWaitHandle(m_socket2, 
            std::bind(&AsyncSocket::ReadIncomingPacket, &m_socket2));
        WFMOHandler::AddWaitHandle(m_socket3, m_aeh);
    }
    virtual ~MyDaemon()
    {
        Stop();
        // just being graceful, WFMOHandler dtor will cleanup anyways 
        WFMOHandler::RemoveWaitHandle(m_socket2);
        WFMOHandler::RemoveWaitHandle(m_socket1);
    }
};

In the example above, the Win32 events associated with m_socket1 and m_socket2 are handled directly by AsyncSocket method ReadIncomingPacket. This method is converted into a function object using std::bind(). However, m_socket3 is handled by the explicitly defined function object, AnotherEventHandler.

RemoveWaitHandle is a method corresponding to AddWaitHandle, one that allows you to remove an already registered wait handle. These two methods provide a truly dynamic WaitForMultipleObjects wait-loop to which handles can be registered and deregistered as the situation demands it. Of course, given the limitation on the number of handles that WaitForMultipleObjects can track, there’s an upper limit to this (this is discussed in more details in Points of Interest section below).

WFMOHandler provides two additional methods -- Start() and Stop() -- which starts the worker thread and shuts it down respectively. Stop() is to be called at the end of the program or whenever the WFMOHandler object is to be deinitialized. If Stop() is not called explicitly, it will be called from WFMOHandler destructor. The code below is an excerpt from main() that shows how Start() is used to get the daemon going.

It needs to be highlighted that AddWaitHandle and RemoveWaitHandle only update the internal data structures that are used to hold the wait handle and its associated function object handler. The actual task of updating the wait array is performed in the worker thread (Add/Remove routines signal the worker thread). This has important ramifications for RemoveWaitHandle. One might be tempted to release the wait handle from the client code right after a RemoveWaitHandle call returns. This will invariably result in an exception as the worker thread would be blocking on this handle (along with others) while you attempt to close it. So to provide for graceful release of handle resources, WFMOHandler provides a callback through the virtual function OnWaitHandleRemoved, which will be called when the worker thread has actually removed the handle from its wait array. Clients can override this method to close the handle and release any associated resources. Note that this method is called from the context of the WFMOHandler internal worker thread. So if releasing the handle also involves updating client class data members, care must be taken to synchronize this with the other threads in the system.

try {
    MyDaemon md;
    md.Start();
    std::cout << "Daemon started, press Ctrl+C to stop." << std::endl;
    ::WaitForSingleObject(__hStopEvent, INFINITE);
} catch (std::exception e) {
    std::cerr << "std::exception: " << e.what() << std::endl;
} catch (...) {
    std::cerr << "Unknown exception" << std::endl;
}

Points of Interest

Behind the scenes, WFMOHandler uses combination of function and class templates to implement its seemingly simple interface.

Firstly, AddWaitHandle is a function template that is instantiated with the type passed to it, that is the type of the function object. Secondly, this function object type is used in the AddWaitHandle implementation to generate a concrete data type from a class template - WaitHandler, an internal class of WFMOHandler. Since WFMOHandler needs to treat all WaitHandler class template instantiations uniformly, WaitHandler class template is derived from WaitHandlerBase. WaitHandlerBase stores the wait handle and has a pure virtual function -- invoke(). The wait handle member is used to generate the wait handle array to be passed to the API and invoke() is called when the corresponding handle is signalled. Since invoke() is overridden in WaitHandler, which in turn calls the function object argument to AddWaitHandle, control is transferred to the client code.

Using function objects instead of function/method pointers of fixed type signature provides immense flexibility. There is virtually no restriction on the type of method or its arguments that can be bound as a wait handle handler. These methods can come from any class, not just the child of WFMOHandler. This is demonstrated in the example above -- the socket event handlers are set to AsyncSocket::ReadIncomingPacket.

Also, multiple instances of the same type can be easily managed as individual std::bind() call generates a new function object. Again this point is demonstrated by the sample code where m_socket1 and m_socket2 are handled by AsyncSocket::ReadIncomingPackets. Finally, using std::bind allows you to bind additional arguments to the handler method calls.

Caveats

One caveat is that the order of the synchronization handles in the array that is supplied to WaitForMultipleObjects is arbitrary in this implementation. This is because internally WFMOHandler uses a std::map to store the handler method pointers indexed by the wait handle which is then enumerated to build the wait array. So the order really depends on the handle's numeric value. However, if control over this is required, the code can be adapted to accommodate this. Either by using a different STL collection so the order in which the AddWaitHandle is called becomes the natural order of the wait handles or by extending the AddWaitHandle to include an additional parameter that indicates the handle priority. This handle priority can then be used to control the order of the handles in the wait array that is passed to the API.

Another caveat is that compiling this code requires VC++2010 (or newer) that includes std::bind() which supports variable number of arguments. Using an older compiler is still possible, but you will be restricted to binding handler methods that can only take a single argument using either std::bind1st or std::bind2nd. However, if you can link to boost libraries, it provides a std::bind() replacement that supports composition of function objects from methods that take more than 1 argument and works with older compilers such as VC++2008. I have not tried the code with the boost version of std::bind(), but it ought to work.

Lastly, though WaitForMultipleObjects can take an array of 64 handles, in the WFMOHandler implementation, you are restricted to 62 handles. This is because the first two are used internally by two event handles -- one for signalling thread shutdown and the other for rebuilding the wait handle array. These two can be collapsed into one event handle by employing a different mechanism to signal these two tasks, but for my needs support for 62 wait handles were more than sufficient.

I hope you find this useful. I worked on a project where 4 different modules employed WaitForMultipleObjects and I have successfully migrated three of them to use this base class. And other programmers tasked with maintaining the code couldn’t be happier as this model brought a lot of clarity to the overall code. Even if they did not understand how the std::bind() magically results in callbacks into the client object’s method, they could go about their jobs as application functionality code was neatly separated from the threading/signalling mechanics.

In another article I'll show how this can be extended to support Win32 timer routines which can eliminate the need for additional worker threads with embedded Sleep() calls or continuous polling to perform periodic tasks.

History

10 Dec 2013 - Initial release