Workflow organization: synchronization channel

Imagine the architecture of a typical application:

There is a workflow engine that performs some functionality, let's say file copying (archiving, searching for prime numbers). In general, something long.
This stream should periodically report information about the current file being copied, as well as be able to handle errors, for example, an error of insufficient disk space.

The graphical interface of such an application should allow you to start the file copying process, be able to pause the copying, and, in case of an error, display the corresponding dialog with a question for the user.
')
It would seem, how can you make a mistake in such a simple situation?

Multithreading problems

When an additional stream appears in the program, immediately there is a problem of interaction between the threads. Even if the stream does nothing and does not communicate with anyone, there is always the problem of stopping the stream correctly.

Even when working with high-level wrapper classes on threads, it’s easy to do something wrong if you don’t fully understand the correctness of working with threads. Therefore, in this article we will talk about working with threads at the WinAPI level.

And so, back to our example.

The engine workflow must somehow inform the GUI thread of its state (the current file being copied), notify about pausing, and also initiate an error message.

The two main methods of notification - asynchronous and synchronous

Asynchronous method - a workflow notifies its status with asynchronous messages (PostMessage).
After sending such a message, the flow, as a rule, does not wait for a response and continues its work.
And if it is impossible to continue, the thread waits for a control command from the GUI.

Synchronous method - a workflow notifies you of its status by synchronous calls (SendMessage), with waiting for the completion of processing such calls.
This method is convenient because the workflow, at the time of processing messages, is in a state that is known in advance. No need for excessive synchronization.

Also, the third option is not a little important - polling the engine state by timer. Thus, it is best to implement frequently changing states, such as progress of implementation. But this method does not apply to the topic of this article.

The asynchronous method has its advantages, but it will be about synchronous messages, the main benefit of which is simplicity.

Pitfalls: SendMessage + stop flow

When I see a workflow, I immediately wonder how it interacts with the GUI and how it is stopped.

Be careful if the workflow directly or indirectly calls the blocking function SendMessage for the thread's GUI. On the example of WinAPI, it could be something completely innocuous, for example, some kind of call to SetText, which internally calls SendMessage WM_SETTEXT. In this case, you need to be especially careful when trying to stop a thread in handlers for pressing buttons and when closing an application (if the GUI stream is the main thread of the application). This is not entirely obvious, I will try to explain further.

The correct way to terminate a stream is to wait until it completes, using one of the WaitFor functions, passing the HANDLE parameter to the stream. Moreover, it is necessary to wait for the thread to completely stop - no timeouts followed by a call to TerminateThread. For example:

// INFINITE ,  -       ,     WaitForSingleObject(hThread, INFINITE); 

If this is not done, unpredictable consequences are possible (program hangs and crashes).
Especially if the thread is inside a dynamically connected library, which should also be unloaded here.

And so, once again about the SendMessage problem: if we wait for the completion of the stream in the window message handler, then we will block this very processing of window messages. And the workflow, in turn, will send the message and wait until it is processed. Thus, we are guaranteed to get a deadlock on each thread.
One solution in the case of synchronous messages is not just to wait for the completion of the flow, but to scroll through the window messages until the flow is complete (the crutch of course, but also has the right to exist)

The second architectural problem is that if the workflow directly calls the GUI code, then you need to take care of synchronization. Synchronization of threads is spread over the entire program.

The solution to these problems

The workflow must be isolated inside the engine interface.

All notifications from the engine should come synchronously and in the context of the client flow, on the principle of COM Single-Threaded Apartments.

Calls must occur synchronously and transparently: the worker thread calls a function that does not return until the GUI thread processes this call.
But at the same time, the workflow must be able to terminate, even at the time of calling such a function.

As a result, the GUI engine interface will be single-threaded, which will greatly simplify the work with such an engine.

Implementation variant and example in C ++

To implement this behavior, you can create a reusable object that will provide context switching for threads when calling GUI code.

I called such an object - the synchronization channel.

And so, we do a kind of synchronization channel, with the help of which the workflow of the engine will call callback functions implemented by the GUI.

The channel will have a Execute function, with the parameter boost :: function, where you can pass the functor created by boost :: bind. Thus, using this channel, you can call the callback function with any signature, for example:

 class IEngineEvents { public: virtual void OnProgress(int progress) = 0; ... }; //-  ... IEngineEvents* listener; //  ,  GUI syncChannel.Execute(boost::bind(&IEngineEvents::OnProgress, listener, 30)); 

The Execute function, as mentioned earlier, is synchronous — it does not complete until the callback function is completed. Except for the exception described below.

The channel must also have a Close function, the action of which is the following: all calls to the Execute function are completed, new calls to the Execute function fail. The workflow is released and thus the problem of stopping the workflow is solved — you can use the WaitFor function without having to scroll through window messages.

To switch the thread context, the example uses the standard Win32 message queue and the PostThreadMessage function.

The principle of operation is as follows: the worker thread sends a message using PostThreadMessage, and then waits until this message is processed, for this the separate event object is used in the example.
At this time, the GUI thread must process its window messages, one of which must be a message from the workflow, which it must process and notify the workflow about the completion of processing using the event object.

This implementation assumes the ProcessMessage function, which must be called from the window message processing loop or window procedure. It is possible to implement without such a function, for example, a channel can create an invisible window for itself, and process all messages inside. In addition, implementation is possible without the use of window messages in principle.

I would also like to say that the example is only of an introductory nature, and is not a ready-made solution.

 // SyncChannel.h class CSyncChannel { public: typedef boost::function<void()> CCallback; public: CSyncChannel(void); ~CSyncChannel(void); public: bool Create(DWORD clientThreadId); void Close(); bool Execute(CCallback callback); bool ProcessMessage(MSG msg); private: DWORD m_clientThreadId; CCallback m_callback; HANDLE m_deliveredEvent; volatile bool m_closeFlag; }; 

 // SyncChannel.cpp UINT WM_SYNC_CHANNEL_COMMAND = WM_APP + 500; CSyncChannel::CSyncChannel(void) : m_closeFlag(true) {} CSyncChannel::~CSyncChannel(void) {} bool CSyncChannel::Create(DWORD clientThreadId) { if (!m_closeFlag) { return false; } m_clientThreadId = clientThreadId; m_deliveredEvent = CreateEvent(NULL, TRUE, FALSE, NULL); if (!m_deliveredEvent) { return false; } m_closeFlag = false; return true; } void CSyncChannel::Close() { m_closeFlag = true; if (m_deliveredEvent) { CloseHandle(m_deliveredEvent); m_deliveredEvent = NULL; } } bool CSyncChannel::Execute(CCallback callback) { //        . //   ,        . //   Pause(),     // -    , //        "pause pending" if (m_closeFlag) { return false; } if (GetCurrentThreadId() == m_clientThreadId) { //    -   , //      ,    . //      ,        , //       -     . callback(); } else { //  Execute     , //        , //        . //      , //       . //      ,   , //      , //     . m_callback = callback; //     ,    //      ,     ResetEvent(m_deliveredEvent); //     ,   //    . //       ,  , //       CSyncChannel::ProcessMessage() if (!PostThreadMessage(m_clientThreadId, WM_SYNC_CHANNEL_COMMAND, NULL, NULL)) { return false; } // ,      CSyncChannel::ProcessMessage(), //     m_deliveredEvent, //       m_closeFlag //    m_closeFlag    //   WaitForMultipleObjects,      , //       . DWORD waitResult = WAIT_TIMEOUT; while (waitResult == WAIT_TIMEOUT && !m_closeFlag) { waitResult = WaitForSingleObject(m_deliveredEvent, 100); } if (waitResult != WAIT_OBJECT_0) { //      ,       return false; } } //          return true; } bool CSyncChannel::ProcessMessage(MSG msg) { //        if (msg.message != WM_SYNC_CHANNEL_COMMAND) { //         , //     return false; } if (!m_closeFlag) { //      , //         m_callback(); //   ,   . //       SetEvent(m_deliveredEvent); } return true; } 

An example of using the CSyncChannel class is given in the following article - Workflow organization: engine state management .