Variations on producer consumer in C++

Image made by Stefaan Himpe in libreoffice draw, available under CC-BY-SA license.

Problem

Paraphrased from wikipedia:

The consumer producer problem is a classical example of a multi-process synchronization problem. The problem describes two processes, the producer and the consumer, who share a common buffer used as a queue. The producer's job is to generate a piece of data, put it into the buffer and start again. At the same time, the consumer is consuming the data (i.e., removing it from the buffer and doing something with it) one piece at a time.

As an example consider a printer server. The printer server receives requests to print long documents. We don't want the printer server to refuse new requests while the printer is still printing. Therefore we run it in a separate thread. After receiving the request, the printer server queues it in a printer queue and immediately returns to listening for new requests. The printer server is a producer. An application monitors the printer queue and prints the requested documents, one by one. The application is a consumer.

Challenges

• I still have too little practical experience in multithreaded programming. This article is written as part of my investigation into what is possible. If you spot mistakes or inaccuracies, please comment!
• How do we push and pop work items on/from a queue in a multi-threaded environment?
• If the producer is faster than the consumer, how do we prevent the queue from exploding in size, eventually filling up all available memory?
• How can a consumer wait for data to appear on the queue?
• How can we wait for a queue to become empty? How do we wait for a consumer to finish processing everything in its queue? What's the difference?

Preliminaries

Synchronization

In a multi-threaded environment it is imperative that we use synchronization mechanisms to avoid situations like the producer trying to add something on the queue while simultaneously the consumer tries to get something from the queue. Without synchronization mechanisms this can lead to crashes or data corruption.

Technology choices

To implement multi-threaded things in C++ in a portable way I have chosen the following technologies:

• Boost: a C++ library that turns "C++" into "C++, batteries included"
• CMake: a cross-platform C++ build system

CMakeLists.txt

Here's my CMakeLists.txt file (New BSD License). This file defines how CMake can find the boost libraries, and build the project.

project(ProducerConsumer)
cmake_minimum_required(VERSION 2.8)
set(Boost_MULTITHREADED TRUE)
find_package (Boost COMPONENTS system thread)
include_directories (${Boost_INCLUDE_DIRS})
link_directories (${Boost_LIBRARY_DIRS})
aux_source_directory(. SRC_LIST)
add_executable(${PROJECT_NAME} ${SRC_LIST})
target_link_libraries(${PROJECT_NAME} ${Boost_LIBRARIES} )

To use this CMakeLists.txt file, you'd create a directory(say at: /home/user/myproject/src or c:\development\myproject\src) containing the following files

• CMakeLists.txt (shown above)
• main.cpp (i.e. the rest of the code appearing later in this blog entry)

Then, to build the software, you'd create a build directory outside the src directory(say at: /home/user/myproject/build or c:\development\myproject\build) and from inside that directory you'd run on the command line:

If you have gnu make installed:

cmake ../src
make

If you don't have gnu make installed on your system, try cmake --help to see how you can use it to generate a project file for eclipse, kdevelop, visual studio, ninja, codeblocks, ...

Creating a producer and consumer for testing

Here's a simple producer and consumer, both derived from a WorkerThread class (both need to run independently in a separate thread). WorkerThread implements all the things that all producers and consumers have in common, but leaves the actual work to be implemented by derived classes by means of the purely virtual method Work().
In an attempt to reduce code duplication, I've templatized the WorkerThread with a QueueType that itself is templatized with QueueContents (e.g. int or some class PrintingParameters). In the rest of this blog entry, different QueueTypes will be presented.

Attempt 1: ThreadSafeQueue

Code

This is based on the code published here.

Test program

Output

About to construct queue
About to construct producer
About to construct consumer
About to start producer
About to start consumer
Waiting for producer to finish
[PRODUCER] Produced value 0
[CONSUMER] Consumed value 0
[PRODUCER] Produced value 1
[PRODUCER] Produced value 2
[PRODUCER] Produced value 3
[PRODUCER] Produced value 4
[PRODUCER] Produced value 5
[PRODUCER] Produced value 6
[PRODUCER] Produced value 7
[PRODUCER] Produced value 8
[PRODUCER] Produced value 9
Waiting for consumer to finish
[CONSUMER] Consumption of value 0 completely handled.
[CONSUMER] Consumed value 1
[CONSUMER] Consumption of value 1 completely handled.
[CONSUMER] Consumed value 2
[CONSUMER] Consumption of value 2 completely handled.
[CONSUMER] Consumed value 3
[CONSUMER] Consumption of value 3 completely handled.
[CONSUMER] Consumed value 4
[CONSUMER] Consumption of value 4 completely handled.
[CONSUMER] Consumed value 5
[CONSUMER] Consumption of value 5 completely handled.
[CONSUMER] Consumed value 6
[CONSUMER] Consumption of value 6 completely handled.
[CONSUMER] Consumed value 7
[CONSUMER] Consumption of value 7 completely handled.
[CONSUMER] Consumed value 8
[CONSUMER] Consumption of value 8 completely handled.
[CONSUMER] Consumed value 9
[CONSUMER] Consumption of value 9 completely handled.
Queue should be empty after all threads finished: 1

Discussion

• Each method accessing the queue first waits until it can acquire a mutex (obviously, each method has to try to acquire the same mutex). This is needed to ensure that other threads cannot modify the queue this method is operating on. This could lead to data corruption or crashes.
• At some point (e.g. in method TryPop) we check the condition "m_queue.size()==0". If we'd replace this condition with a call to "bool Empty()", the code would deadlock. The reason is that "Empty()" tries to acquire the same mutex that "TryPop(...)" has already acquired.
• To wait for data to appear on the queue, we wait for a condition variable. This is by far the most efficient way to wait for something to happen. In some code bases one finds constructs like "while (!DataAvailable) sleep(100);". Such approach consumes more resources and if one is not careful it is not thread-safe.
• When the producer is faster than the consumer (which I simulated in the test program by giving the consumer a 1 ms delay, and giving the producer no delay), it will keep on pushing values on the queue, and the queue size will keep increasing. This is clearly visible in the output of the program. At some point the available memory on the system would be exhausted causing a hang up or crash. This is a serious problem that needs an improved design: a BoundedThreadSafeQueue. A BoundedThreadSafeQueue is parameterized with a maximum size. Pushing data on the queue will block until the queue has room for more entries.

Attempt 2: BoundedThreadSafeQueue

Code

Test program

Output

About to construct queue
About to construct producer
About to construct consumer
About to start producer
About to start consumer
Waiting for producer to finish
Produced value 0
Produced value 1
Produced value 2
Consumed value 0
Produced value 3
Consumed value 1
Produced value 4
Consumed value 2
Produced value 5
Consumed value 3
Produced value 6
Consumed value 4
Produced value 7
Consumed value 5
Produced value 8
Consumed value 6
Produced value 9
Waiting for consumer to finish
Consumed value 7
Consumed value 8
Consumed value 9
Queue should be empty after all threads finished: 1

Discussion

• The test program instantiated a bounded queue with size 3, with a consumer that is slower than the producer. The output of the test program shows that the consumer first needs to consume items before pushing can proceed when the queue is full.
• But we have a different problem now. In real life, consumption of an item will take some time (printing a large document can take several minutes). The queue can now signal that it is empty, but an empty queue does not mean that all processing is finished. (The items are removed from the queue before their processing starts.) So how can we extend our design to be able to wait until the queue and associated processing is completely finished?

Attempt 3: BoundedThreadSafeQueueSignalWorkloadDone

The basic idea is for the consumer of the queue data to inform the queue that processing the data is finished. Ideally we'd have a kind of condition variable that we can set/reset manually to signal that the queue + associated processing is completely finished (idle) or not. Someone waiting for the queue and associated processing to finish completely can then wait for the condition variable to become true. Boost doesn't offer this out of the box, but based on the code published here, it is possible to create such a manually resetable condition variable ourselves.

The consumer class needs a small update: we have to call ProcessingFinished() on the queue after finishing processing the information.

Output

[MAIN] About to construct queue
[MAIN] About to construct producer
[MAIN] About to construct consumer
[MAIN] About to start producer
[MAIN] About to start consumer
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[PRODUCER] Produced value 0
[CONSUMER] Consumed value 0
[PRODUCER] Produced value 1
[PRODUCER] Produced value 2
[PRODUCER] Produced value 3
[PRODUCER] Produced value 4
[QUEUE] Wait until idle
[QUEUE] ProcessingFinished but queue not empty. ProcessingDone flag remains low.
[CONSUMER] Consumption of value 0 completely handled.
[CONSUMER] Consumed value 1
[PRODUCER] Produced value 5
[QUEUE] ProcessingFinished but queue not empty. ProcessingDone flag remains low.
[CONSUMER] Consumption of value 1 completely handled.
[CONSUMER] Consumed value 2
[PRODUCER] Produced value 6
[QUEUE] ProcessingFinished but queue not empty. ProcessingDone flag remains low.
[CONSUMER] Consumption of value 2 completely handled.
[CONSUMER] Consumed value 3
[PRODUCER] Produced value 7
[QUEUE] ProcessingFinished but queue not empty. ProcessingDone flag remains low.
[CONSUMER] Consumption of value 3 completely handled.
[CONSUMER] Consumed value 4
[PRODUCER] Produced value 8
[QUEUE] ProcessingFinished but queue not empty. ProcessingDone flag remains low.
[CONSUMER] Consumption of value 4 completely handled.
[CONSUMER] Consumed value 5
[PRODUCER] Produced value 9
[QUEUE] ProcessingFinished but queue not empty. ProcessingDone flag remains low.
[CONSUMER] Consumption of value 5 completely handled.
[CONSUMER] Consumed value 6
[QUEUE] ProcessingFinished but queue not empty
[CONSUMER] Consumption of value 6 completely handled.
[CONSUMER] Consumed value 7
[QUEUE] ProcessingFinished but queue not empty
[CONSUMER] Consumption of value 7 completely handled.
[CONSUMER] Consumed value 8
[QUEUE] ProcessingFinished but queue not empty
[CONSUMER] Consumption of value 8 completely handled.
[CONSUMER] Consumed value 9
[CONSUMER] Consumption of value 9 completely handled.
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[QUEUE] Wait until idle
[QUEUE] Consumer is completely idle
[MAIN] Queue should be empty after all threads finished: 1
[MAIN] Waiting for producer to finish
[MAIN] Waiting for consumer to finish

Discussion

• Writing a meaningful test program is less obvious here.
• From the output: when the program starts, it starts producer and consumer in a separate thread, and then starts waiting in the main thread until the queue + associated processing is completely finished.
  • Just after constructing the producer and consumer, the channel is still completely idle: no work has been pushed onto the queue yet.
  • Then the producer starts producing values, and the next wait until idle blocks until the processing is completely finished. After that, WaitUntilCompletelyIdle immediately returns true (until another producer inserts work into the queue).

Capture colored console output in a tkinter window

The problem

I want to run a console script that emits ansi color codes to generate a colorful output and capture that colorful output in a tkinter window. Ideally I want this to work for programs like cmake or emerge which semi-intelligently switch off color codes if they detect they are not running on a terminal. This is all tested under linux. Some if it might work on windows too, but I haven't investigated.

Approach

I will show some increasingly sophisticated approaches using python 3.2 and try to outline their limits or problems. In order not to overload the presentation, this section details some functions that are used by the different approaches below. First of all in a file ansicolortext.py I have the following MIT licensed code. It is a tkinter Text widget that tranforms ansi color code sequences into tkinter color tags. Next, in a file timeout.py I have the following code, created by Matt Harrison under MIT license and posted as an ActiveState recipe: Finally, here's a function to create the user interface (a tkinter window with a resizable AnsiColorText widget)

Approach 1

If you google for this type of stuff almost always someone will come up with something as follows: This simple approach works beautifully for the example given (ls --color) and is highly recommended if it works for you. For some programs, however, this doesn't work. Many programs try to detect if they are running on a terminal or if they are part of a piped series of commands, and in the latter case suppress color codes. Examples of such programs include CMake and grep with the --color option. The end result is that your output shows up in black and white.

Approach 2, attempt 1

In order to fool programs like CMake or grep into producing color codes we need to pretend we are a terminal. The way to do this in linux is by using a pseudoterminal. According to wikipedia,

In some operating systems, including Unix, a pseudo terminal is a pseudo-device pair that provides a text terminal interface without an associated device, such as a virtual console, computer terminal or serial port. Instead, a process assumes the role of the underlying hardware for the pseudo terminal session.

Pseudoterminals can be accessed using Python's pty module. pty.openpty() opens a new pseudo-terminal pair and returns a pair of file descriptors (master, slave), for the master and the slave end, respectively. The terminal emulator process is associated with the master device and the shell is associated with the slave. The terminal emulator process (master) receives input from the keyboard and mouse using windowing events, and is thus able to transmit these characters to the shell (slave), giving the shell the appearance of the terminal emulator being an underlying hardware object. Any terminal operations performed by the shell in a terminal emulator session are received and handled by the terminal emulator process itself (such as terminal resizing or terminal resets). This approach looks clean and simple, but it has a problem: os.read blocks when no more data comes in, and so our script hangs before the UI can even come up. One proposal could be to use a timeout when reading data using os.read.

Approach 2, attempt 2

This approach now deals with the blocking: upon a timeout the communication is stopped. It suffers from a different problem though: the UI doesn't appear before all data has come in. In the case of a CMake script this can take minutes to hours to complete. Not exactly what we wanted... therefore we need a slightly smarter approach.

Approach 2, attempt 3

The idea is this: we will run all the communication in a thread of its own. The UI will run in the main thread. The communication thread will send the data it receives to the main thread by means of a Queue object. The main thread will read data from the Queue and display it in the text widget as it receives it.

The TextUpdater object T lives in the program's main thread. It reads data from the communication thread t by means of Queue q and sends it to the Text widget text, which itself is a child widget of the Tk root window root. The TextUpdater thread is asked to start working 100ms after root.mainloop() is started. Like that the UI has ample time to get itself ready for processing. The communication thread ThreadCommand t opens a pseudoterminal and uses it to communicate with the program being run. It stops itself and flags itself as not running anymore if no more data comes from the process (i.e. if the reading timed out). Reading data from the process uses a version of os.read with a timeout added to it (read2). The data read by read2 is slightly processed before it is sent to the communication thread: only complete lines of text are sent to avoid that we break up text in the middle of a color code (this would give trouble because the AnsiColorText widget isn't smart enough to deal with half-finished color codes).

If the timeout value used in read2 is too small, the timeout can occur too soon (especially with processes that take some time to generate output, like CMake) so this part is still a weak spot in this approach. Too long a timeout causes one to wait long after the process finishes to really close the communication thread. If you know of better ways to handle this, please comment! After text is written to the Text widget, we call root.update() to give Tk() the chance to handle pending events. This makes sure that the text widget is updated as input comes in, and it also makes sure that you can resize the window or scroll through the displayed data while data is still coming in.

An alternative without timeout decorator

As I found out today, there's an alternative without timeout decorater, by using the poll call of Python's select module. It still uses a timeout mechanism to decide if there's data to be read. This time the timeout mechanism is provided by the select module directly.

Screenshot

Here's a screenshot of CMake producing colored output in a Tk window. While CMake is running, the window can be resized and you can navigate through it with arrow/pageUp/pageDown/... keys.

YouNeedTests: a python3 based C++ unit testing tool

(Image retrieved from http://blog.hinshelwood.com/. According to google image search this image is labeled for reuse. Please let me know if you disagree.)

Unit testing in C++

Perhaps you recognize the following story. You're finished writing C++ code, and now you want to add some unit tests, because it makes the metrics look good. You're using a well-established C++ unit testing framework, like CppUnit. Adding a new test requires manually adding files to the solution, adding seemingly duplicated information in .cpp and .h files (i.e. implementations and declarations) and invoking obscure macros. Or perhaps you copy an existing test file and start editing it to test new stuff. Or worse: you extend an existing test with some extra lines to avoid creating all the boilerplate code over and over again.

However, being a programmer and not a file copier, I would really appreciate if we could use our computer a bit more efficiently to specify tests.

YouNeedTests: Python3's doctest meets fitnesse for C++ code

doctest? fitnesse?

The brilliant thing about python's doctest module is how it extracts tests from comments embedded in the source code. This leads to executable documentation which by its very nature never needs to become outdated. Doctest is typically used to write unit tests: tests written by programmers for programmers, trying to test little bits of code in isolation.

The brilliant thing about fitnesse is how it specifies different test scenarios in table form. Fitnesse is typically used to define acceptance tests: tests written by business and QA people and intended for business and QA people. Acceptance tests usually involve more objects than unit tests.

requirements

I wanted a tool that could be used with C++, and that transforms comments embedded in the C++ code into unit tests. I also wanted a way to specify different test scenarios in table form. Because I prefer python over C++ in matters of text processing, I decided this was an ideal opportunity to wet my feet in several "hot" and "established" technologies:

• git as source code management system. The brilliant thing about git is its distributed model, its insane flexibility, speed and compactness, and its superb support for branching and merging.
• python3, the somewhat controversial, non-backward compatible successor to the very popular python2 language. I still have to find out what's so brilliant about python3 - but so far python3 hasn't been a negative experience.
• yaml to format the comments from which the code will be generated. The brilliant thing about yaml is how it combines expressiveness of XML with a much more human friendly syntax. Human friendly enough that I didn't create a domain specific language to specify tests in.
• googletest as backend unit testing framework. The brilliant thing about the googletest framework is how it requires fairly little boilerplate code to write simple tests. The second brilliant thing about googletest is that it supports death-tests, i.e. you can check if a piece of code crashes as expected.
• CMake for the build system. The brilliant thing about CMake is how cross-platform, easy to use and versatile it is as a build system. It also comes with built-in provision for testing (and loads of stuff I haven't discovered yet).
• mako to generate boiler plate code and build scripts. The brilliant thing about mako templates is their ease of use and the fast speeds with which they are rendered.

Example

Here's a simple comment that will 6 independent tests from a table of scenarios. Not all tests need to have a table, one can also embed raw code if desired. A comment like the above consists of several parts:

• a testsuite name
• a testsuite type: for now 3 types are supported:
  • COLUMNBASED: CODE section contains a list of tables. Each line in each table is an independent test.
  • ROWBASED: CODE section contains a list of tables. Each table becomes a test. Each line in a table is one step in the test.
  • RAW: CODE section contains normal C++ code.
• a LINK section: this specifies which .cpp files should be linked in with the test
• an INCLUDE section: this specifies which .h files to include in with the test
• a STUBS section: this can contain arbitrary C++ code to resolve linker errors. Only stub those parts of the code which are not testing. There's no point in testing stubs :)
• a PRE section: code in the PRE section is executed before anything from the table is execute
• a POST section: code in the POST section is executed after executing statements from the table, but before executing assertions from the table
• a CODE section. In case of a COLUMNBASED or ROWBASED test suite, the CODE section contains a list of tables. Each table has a table header that specifies code templates with placeholders $1, $2, .... The placeholders will be filled in with values from the table cells in the same column. Each table has a NULL column (the ~ symbol) which separates the statements on a line from the assertions.
  • The line containing the string 'twovalues' in table 'TABLE1', e.g. will result in the following lines of code being generated: where the double comma (",,") causes the code template to be unrolled. (Of course the system will also generate all the rest of the required boilerplate code and buildscripts, finally leading to this .cpp file (buildscript not shown here): Despite being used with a toy example, the table format already results in quite some space and boiler-plate reduction :)

All the generated tests can be run at once by issuing "make test" in the output folder:

More detailed test results are available by individually running tests:

Proof of concept code!

Got curious? Hop over to Gitorious! LGPL'd proof of concept code - for now I only tried to use it on debian sid amd64.