Introduction
Like many of my projects, this one started out due to a question in microsoft.public.vc.mfc. Someone had written a program that was supposed to transform the output of a program and convert the stderr and stdout streams of a child process to different representations, and write them out to the stdout handler of the filter process.
Unfortunately, it was a really bad example of a program that would work for this purpose. There is a problem with anonymous pipes; they can't be asynchronous. So, the problem was to grab data from stderr and stdout, but blocking on one pipe means that output that came via the other pipe would not be seen. So, the solution was to use ::PeekNamedPipe to see if there was any data to read. If there was, then a ReadFile was issued to read it. The problem with this was that it was in an infinite loop, and it was polling each time. There was no way it could block waiting for input, so as a result, it would run continuously, ultimately using close to 100% of the CPU time, accomplishing nothing with most of it.
So, the problem was: how to create a program that would block, but would be able to receive data from stderr or stdout as the data became available.
The answer was obvious: threads.
The problem with threads is that far too many people suffer from klostiphobia, the morbid fear of threads, so never think to use threads when they are appropriate, or necessary. In this case, they were necessary.
So, I decided to write a program to demonstrate how threads could be used for this purpose.
Most of my articles on threading use GUI apps, so this shows how threading can be used in a console app.
The Test Program
I wrote a little test program, called tester, which simply generates a random number and, based on that random number, writes a line to either stdout (using an integer value and the imaginative contents "output") or stderr (using an integer value and the equally imaginative contents "error").
int _tmain(int argc, _TCHAR* argv[])
{
for(int i = 0; i < 100; i++)
{
int r = rand();
if(r & 0x100)
{
_ftprintf(stderr, _T("%4d: error\n"), i);
fflush(stderr);
}
else
{
_ftprintf(stdout, _T("%4d: output\n"), i);
fflush(stdout);
}
int w = rand() % 500;
Sleep(200 + w);
}
return 0;
}
The output from running this program by itself is shown below:
Generating Decorated Console Output
The goal of the poster was to enclose these in some HTML commands that would cause, with appropriate HTML definitions around them, to display the data in some suitable form.
I generalized this to support writing directly to a console window. So, when run under my piper program, the lines to stdout and stderr display differently, as shown below:
Generating HTML Output
However, if I add the -html option to the command line, I get HTML commands wrapped around them, such as:
To make this display work, I had to add (by hand) to this HTML file the following declarations in the <head>...</head> region:
<style type="text/css">
p.stdout {margin-top:0; margin-bottom:0; color:green}
p.stderr {margin-top:0; margin-bottom:0; font-weight:bold; color:red}
</style>
The Program
The structure of the program was to create two threads: one thread to handle stdout and one thread to handle stdin. The main thread would gather the data from the two worker threads and display it. When both threads terminated, the main thread would then exit.
The question was what mechanisms to implement to do this. I naturally fell back on my favorite inter-thread queuing mechanism, the I/O Completion Port. I/O Completion Ports are cool, and I describe several techniques in my accompanying article on them. This is just another generalization of that mechanism, with some interesting twists and features to illustrate their generality.
I used VS.NET 2003 to generate this program; to generate it, I asked for a console application, with ATL support, so I could use the CString data type. Note that this is not really possible in VS6. In VS.NET, several important classes were moved out of MFC and moved into the ATL domain, making them usable in console apps and other non-MFC contexts.
_tmain(): Command Line Handling
The command line handling is fairly straightforward, but to give some elegance to the solution, I created a CommandLine class that held all the parameters.
class CommandLine {
public:
CommandLine() { HTML = FALSE; IsUnicode = FALSE; program = NULL; }
BOOL HTML;
BOOL IsUnicode;
LPTSTR program;
};
I wanted to be able to support Unicode pipes, so I added an option to treat the incoming data as Unicode data. The HTML member tells whether to use console decoration or generate the HTML I illustrated above.
int _tmain(int argc, _TCHAR* argv[])
{
if(argc == 1)
{
CString module;
LPTSTR p = module.GetBuffer(MAX_PATH);
::GetModuleFileName(NULL, p, MAX_PATH);
module.ReleaseBuffer();
int n = module.ReverseFind(_T('\\'));
if(n < 0)
n = 0;
module = module.Mid(n + 1);
n = module.ReverseFind(_T('.'));
if(n < 0)
n = module.GetLength();
module = module.Left(n);
_ftprintf(stderr, _T("Usage:\n%s [-u] [-html] command\n"), module);
return Result::INVALID_ARGUMENT;
}
The above code may look a little odd, but I learned years ago in writing console apps that users will rename them. So, if I hardwire the name of the executable into the program, the "usage" message will show the wrong program name. This code merely extracts the file name and displays it properly as part of the usage message.
To deal with the return values, since this is a console app, I wanted to have unique codes for many of the error returns, but I didn't want to have to worry about assigning values to them, so I created a class to represent them.
CommandLine cmd;
for(int i = 1; i < argc; i++)
{
CString arg = argv[i];
if(arg[0] == _T('-'))
{
if(arg == _T("-u"))
{
cmd.IsUnicode = TRUE;
continue;
}
if(arg == _T("-html"))
{
cmd.HTML = TRUE;
continue;
}
_ftprintf(stderr, _T("Unrecognized option \"%s\"\n"), arg);
return Result::INVALID_ARGUMENT;
}
if(cmd.program != NULL)
{
_ftprintf(stderr,
_T("Two command directives given:\n [1] \"%s\"\n [2]\"%s\"\n"),
cmd,
arg);
return Result::TWO_COMMANDS;
}
cmd.program = argv[i];
}
if(cmd.program == NULL)
{
_ftprintf(stderr, _T("need program to run\n"));
return Result::NO_PROGRAM;
}
Class Result
The result types are defined by the class:
class Result {
public:
typedef enum { SUCCESS = 0,
NO_PROGRAM,
INVALID_ARGUMENT,
TWO_COMMANDS,
IOCP_FAILED,
IOCP_ERROR,
THREAD_FAILURE,
STDOUT_CREATION_FAILED,
STDERR_CREATION_FAILED,
CREATEPROCESS_FAILED
} Type;
};
Class SmartHandle
One of the problems in many resource allocation schemes, such as opening handles, is that you end up having to make sure everything is closed properly when you terminate a subroutine. There are some smart classes available, but I wanted to illustrate how easy it is to write one. So, I wrote the class SmartHandle, which closes the handle when the variable goes out of scope.
class SmartHandle {
public:
SmartHandle() { handle = NULL; }
SmartHandle(HANDLE h) { handle = h; }
virtual ~SmartHandle() { if(handle != NULL) ::CloseHandle(handle); }
public:
operator HANDLE() { return handle; }
operator LPHANDLE() { return & handle; }
bool operator==(HANDLE h) { return handle == h; }
SmartHandle & operator=(HANDLE h) { handle = h; return *this; }
public:
void Close() { if(handle != NULL) ::CloseHandle(handle); handle = NULL; }
protected:
HANDLE handle;
};
There is one limitation on this class: you must not explicitly call ::CloseHandle. The SmartHandle::Close method must be used, or exceptions can be thrown.
_tmain(): Create I/O Completion Port
The following code creates the I/O Completion Port:
SmartHandle iocp = ::CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
if(iocp == NULL)
{
DWORD err = ::GetLastError();
_ftprintf(stderr, _T("CreateIoCompletionPort failed, error %s\n"),
ErrorString(err));
return Result::IOCP_FAILED;
}
This creates an I/O Completion Port that is not associated with any file handle.
ErrorString
The ErrorString function is very simple, and is a stripped-down version of my more general ErrorString function suitable for this simple program.
CString ErrorString(UINT err)
{
LPTSTR msg;
FormatMessage(FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM,
NULL,
err,
0,
(LPTSTR) &msg,
0,
NULL);
CString result = msg;
LocalFree(msg);
return result;
}
_tmain(): Create Pipes
SECURITY_ATTRIBUTES sa = {sizeof(SECURITY_ATTRIBUTES), NULL, TRUE};
SmartHandle stdout_read;
SmartHandle stdout_write;
SmartHandle stderr_read;
SmartHandle stderr_write;
static const UINT PIPE_BUFFER_SIZE = 32;
if(!::CreatePipe((LPHANDLE)stdout_read, (LPHANDLE)stdout_write, &sa, PIPE_BUFFER_SIZE))
{
DWORD err = ::GetLastError();
_tprintf(_T("stdout pipe failure: %s\n"), ErrorString(err));
return Result::STDOUT_CREATION_FAILED;
}
if(!::CreatePipe((LPHANDLE)stderr_read, (LPHANDLE)stderr_write, &sa, PIPE_BUFFER_SIZE))
{
DWORD err = ::GetLastError();
_tprintf(_T("stderr pipe failure: %s\n"), ErrorString(err));
return Result::STDERR_CREATION_FAILED;
}
This creates four handles representing the read-side and write-side of the stdout and stderr pipes.
_tmain(): Create Process
STARTUPINFO startup = {sizeof(STARTUPINFO)};
startup.dwFlags = STARTF_USESTDHANDLES | STARTF_USESHOWWINDOW;
startup.wShowWindow = SW_HIDE;
startup.hStdOutput = stdout_write;
startup.hStdError = stderr_write;
PROCESS_INFORMATION procinfo;
if(!::CreateProcess(NULL, cmd.program, NULL, NULL,
TRUE, CREATE_NEW_CONSOLE, NULL, NULL, &startup, &procinfo))
{
DWORD err = ::GetLastError();
_tprintf(_T("CreateProcess failed for \"%s\": %s"),
cmd.program, ErrorString(err));
return Result::CREATEPROCESS_FAILED;
}
::CloseHandle(procinfo.hProcess); ::CloseHandle(procinfo.hThread);
stdout_write.Close(); stderr_write.Close();
After successful completion, there will be no further need of the process and thread handles returned by CreateProcess, so they are closed. Since these handles are not in a SmartHandle structure, they are closed explicitly with ::CloseHandle. The handles this process has to the write side of the stdout and stderr handles are closed.
_tmain(): Thread Creation
To create the thread, I will use _beginthreadex. But there's a problem I have to solve first. By default, a console app assumes that it is not going to be multithreaded, so it is configured with the "single threaded" C runtime library. To create threads, I have to reconfigure the build to use the "multithreaded" C runtime library. Select the project in the Solutiontab, right click on it, and ask for Properties. The screen below will be displayed:
Select the configuration you want (in this case, the Debug configuration), select the C/C++ property, select Code Generation, go to the Runtime Library option, and drop it down. Select the appropriate Multi-threaded library. For the Release configuration, this would be the Multi-threaded runtime library; for the Debug configuration, it would be the Multi-threaded Debug library. Alternatively, you may choose to use the DLL versions of the C runtime library.
unsigned id;
SmartHandle stdoutThread = (HANDLE)_beginthreadex(NULL, 0, reader,
new ThreadParms(stdout_read, SourceFlags::StdOut,
iocp, cmd.IsUnicode), 0, &id);
if(stdoutThread == NULL)
{
DWORD err = ::GetLastError();
_ftprintf(stderr, _T("Thread creation for stdout failed, error %s\n"),
ErrorString(err));
return Result::THREAD_FAILURE;
}
stdoutThread.Close();
SmartHandle stderrThread = (HANDLE)_beginthreadex(NULL, 0, reader,
new ThreadParms(stderr_read, SourceFlags::StdErr,
iocp, cmd.IsUnicode), 0, &id);
if(stderrThread == NULL)
{
DWORD err = ::GetLastError();
_ftprintf(stderr, _T("Thread creation for stderr failed, error %s\n"),
ErrorString(err));
return Result::THREAD_FAILURE;
}
stderrThread.Close();
To pass parameters to the thread, I use the ThreadParms class. It packages up the HANDLE of the stream to read, a flag that we will use to distinguish which stream is notifying us about an event, the handle of the I/O Completion Port, and the Boolean flag to indicate if the child pipe is Unicode or not. If one of these fails, it will return, and note that all currently-open handles are all SmartHandle objects, so the handles will be implicitly closed by the destructor SmartHandle::~SmartHandle, so we don't need to keep track of the handles on our own.
Class ThreadParms
class ThreadParms {
public:
ThreadParms(HANDLE h, SourceFlags::FlagType f, HANDLE io, BOOL uni) {
stream = h;
flags = f;
iocp = io;
IsUnicode = uni;
}
public:
HANDLE stream;
SourceFlags::FlagType flags;
HANDLE iocp;
BOOL IsUnicode;
};
The I/O Completion Port Protocol
I am going to use the I/O Completion Port for inter-thread communication from the worker threads to the main thread. To do this, I will use ::PostQueuedCompletionStatus which allows me to pass three parameters: a DWORD, a ULONG_PTR, and a pointer, which is nominally an LPOVERLAPPED pointer, but in fact can be any pointer of our choosing for ::PostQueuedCompletionStatus.
I could have chosen a variety of techniques to handle this, and I chose the following encoding. There is no reason to use this design in preference to other designs, such as packaging everything up in an object pointed to by the pointer parameter. Any combination not shown below would be an error.
DWORD NumberOfBytesTransferred |
ULONG_PTR CompletionKey |
LPOVERLAPPED Overlapped |
Meaning |
SourceFlags::StdOut |
0 |
(LPOVERLAPPED)(CString *) |
stdout line to display |
SourceFlags::StdErr |
0 |
(LPOVERALLPED)(CString *) |
stderr line to display |
| 0 |
SourceFlags::StdOut |
NULL |
stdout has terminated |
| 0 |
SourceFlags::StdErr |
NULL |
stderr has terminated |
Class SourceFlags
The flags are specified in the SourceFlags class:
class SourceFlags {
public:
typedef enum { None = 0, StdOut=1, StdErr=2 } FlagType;
};
_tmain(): Receive Thread Messages
SourceFlags::FlagType broken = SourceFlags::None;
Result::Type result = Result::SUCCESS;
while(broken != (SourceFlags::StdOut | SourceFlags::StdErr))
{
OVERLAPPED * ovl;
DWORD bytesRead;
ULONG_PTR key;
BOOL ok = ::GetQueuedCompletionStatus(iocp, &bytesRead,
&key, &ovl, INFINITE);
if(!ok)
{
DWORD err = ::GetLastError();
result = Result::IOCP_ERROR;
_ftprintf(stderr, _T("GetQueuedCompletionStatus failed, error %s\n"),
ErrorString(err));
break;
}
broken = (SourceFlags::FlagType)(broken | (int)key);
if(key != 0)
continue;
CString * s = (CString *)ovl;
WriteToOutput(*s, (SourceFlags::FlagType)bytesRead, cmd);
delete s;
}
The trick here is that as each thread finishes, it sends a termination notification. When both threads have sent their termination notification, the main thread will exit the loop, and then the program will terminate.
_tmain(): Cleanup
stdout_read.Close();
stderr_read.Close();
return result;
}
WriteToOutput
This method takes a pointer to the string, the flags that indicate the source, and a pointer to the command line options structure (which is used to determine the format of the output, for example). Its responsibility is to "wrap" the string it is given in whatever context is required to make it "display" correctly. If the output is directed to a console, I do this by setting the text attributes of the console buffer to display in the correct colors; if it is going to HTML (using the -html flag in the command line), I put the right kind of environment around it. There isn't anything really deep going on here.
void WriteToOutput(const CString & s, SourceFlags::FlagType flag, CommandLine & cmd)
{
if(cmd.HTML)
{
CString classname;
switch(flag)
{
case SourceFlags::StdOut:
classname = _T("stdout");
break;
case SourceFlags::StdErr:
classname = _T("stderr");
break;
}
_ftprintf(stdout, _T("<p.%s>%s</p>\n"), classname, ToHTML(s));
}
else
{
HANDLE console = ::GetStdHandle(STD_OUTPUT_HANDLE);
CONSOLE_SCREEN_BUFFER_INFO info;
::GetConsoleScreenBufferInfo(console, &info);
switch(flag)
{
case SourceFlags::StdOut:
::SetConsoleTextAttribute(console, FOREGROUND_INTENSITY | FOREGROUND_GREEN);
break;
case SourceFlags::StdErr:
::SetConsoleTextAttribute(console, FOREGROUND_INTENSITY | FOREGROUND_RED);
break;
}
_fputts(s, stdout);
_fputts(_T("\r\n"), stdout);
::SetConsoleTextAttribute(console, info.wAttributes);
}
}
ToHTML: HTML Conversion
Because the output might contain characters such as '<', '>', or '&', which have meaning in HTML as metacharacters for formatting, it is necessary to convert such characters so they are not going to cause a conflict with the HTML rendering engine. The translations are:
| Character |
Translation |
| & |
& |
| < |
< |
| > |
> |
CString ToHTML(const CString & s)
{
CString t = s;
t.Replace(_T("&"), _T("&")); t.Replace(_T("<"), _T("<"));
t.Replace(_T(">"), _T(">"));
return t;
}
Reader: Top-level Thread Function
This is the top-level thread function. As such, it is coded very simply: while there is something in the pipe, send a notification to the main thread for each line in the data. Then, flush any pending partial line, and finally, send a notification that the thread has terminated.
UINT __stdcall reader(LPVOID p)
{
ThreadParms * parms = (ThreadParms *)p;
PipeReader pipe(parms->stream, parms->IsUnicode);
CString Prefix;
while(TRUE)
{
if(!pipe.Read())
{
break;
}
FormatAndOutput(pipe.GetString(), Prefix, parms);
}
if(!Prefix.IsEmpty())
{
CString text(_T("\r\n"));
FormatAndOutput(text, Prefix, parms);
}
::PostQueuedCompletionStatus(parms->iocp, 0, parms->flags, NULL);
return 0;
}
FormatAndOutput: Line Splitter
This function simply splits up the packet that comes back (which can contain several lines), and sends each complete line to the main thread for subsequent formatting and display. If there is any partial line left over, that is placed in the Prefix variable and will be concatenated to the front of the next line that comes in on the next iteration. Because this means that if the last few characters of the pipe stream will be held in the Prefix because they do not end with a line terminator sequence, it is necessary to "flush" these by calling it one last time with a newline sequence and the remaining characters of the partial line, but only if there is a remaining partial line. That is the code shown in the reader thread.
void FormatAndOutput(CString text, CString & prefix, ThreadParms * parms)
{
text = prefix + text;
while(TRUE)
{
int n = text.Find(_T("\r\n"));
if(n < 0)
{
prefix = text;
return;
}
CString * s = new CString(text.Left(n));
::PostQueuedCompletionStatus(parms->iocp, (DWORD)parms->flags, 0, (LPOVERLAPPED)s);
text = text.Mid(n+2);
}
}
The important feature here is that the string which is passed across the thread boundary is allocated from the heap, and disposed of by the recipient.
Class PipeReader
This is a class I adapted from another application I did. What it does is read from the pipe and handle all the details of what happens if an odd number of bytes is read from a pipe which is expected to be sending Unicode data. Otherwise, there's nothing really deep going on.
class PipeReader {
protected:
static const UINT MAX_BUFFER = 1024;
public:
PipeReader(HANDLE str, BOOL uni) { Init(); stream = str; IsUnicode = uni; }
CString GetString() {
if(IsUnicode)
return CString((LPCWSTR)buffer);
else
return CString((LPCSTR)buffer); }
BOOL Read() {
if(Offset == 1)
buffer[0] = reread;
if(!ReadFile(stream, &buffer[Offset], MAX_BUFFER -
(IsUnicode ? sizeof(WCHAR) : sizeof(char)), &bytesRead, NULL))
return FALSE;
if(IsUnicode)
{
if((Offset + bytesRead) & 1)
{
Offset = 1; reread = buffer[Offset + bytesRead - 1]; buffer[Offset + bytesRead - 1] = 0; bytesRead--; }
else
{
Offset = 0; }
buffer[Offset + bytesRead] = 0;
buffer[Offset + bytesRead + 1] = 0; }
else
{
buffer[bytesRead] = '\0';
}
return TRUE;
}
protected:
void Init() { stream = NULL; Offset = 0; IsUnicode = FALSE; }
BOOL IsUnicode;
HANDLE stream;
protected:
BYTE buffer[MAX_BUFFER];
DWORD Offset;
BYTE reread;
DWORD bytesRead;
};
Summary: Use Threads
Whenever you get into a situation where you end up polling, because you can't block the thread that is polling, consider instead using secondary threads to perform the computations. If there is nothing to do, this application consumes zero CPU time.
As far as programming style, note that there is not a single global variable in any of this code. None are needed, none are used. There is no explicit synchronization; all synchronization is implicit in the use of ::GetQueuedCompletionStatus and the corresponding ::PostQueuedCompletionStatus calls. The best synchronization is no synchronization. This code is a combination of the "positive handoff" model (wherein responsibility for an object is handed off from one thread to another) and the "central manager" model (the output stream, whether console or HTML stream, is managed by a single thread).
History
10 Jun 2008 - Small bugs fixed in ErrorString and ToHTML
Submit an article or tip
Post your Blog
News 
Suggestion 
Question 
Bug 
Answer 
Joke 
Rant 
Admin 
