Terminal Concurrency: The Printing Process

One simple way to see some of the problems that concurrency can cause is to try printing to the terminal concurrently. Try loading up GHCi, and typing import Control.Concurrent followed by:

mapM_ (forkIO . putStr . replicate 100) ['a', 'b']

This code spawns off two threads: one that prints 100 ‘a’s, and one that prints 100 ‘b’s. Your results will likely differ from mine, and the results will likely differ each time, but here’s a sample (with line breaks added for readability):

aabababababaaaaaaaaaaaaabababababababababababababababababab
ababababababababababababababababababbbbbbbbbbbbbbbbbbbbbbbbbbbbb
bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa

Technically, this is non-determinism: the observable behaviour of the program is different each time, even though there is no obvious randomness in our program. The theoretical implications can wait for another day — this is also a practical issue. If we have several concurrently executing processes wanting to write to stdout, how can we stop this messy interleaving from occurring? We want to get all 100 ‘a’s together, and all 100 ‘b’s together.

In CHP, the answer is the printer process. The printer process continually reads Strings from its input channel, and prints them:

printer :: Chanin String -> CHP ()
printer input = forever $ readChannel input >>= liftIO . putStr

We need to connect this printer process to the processes that want to do the printing. A naive attempt might be:

unsafePrinting :: CHP ()
unsafePrinting = do
  c <- oneToOneChannel
  printer (reader c)
    <|*|> writeChannel (writer c) (replicate 100 'a')
    <|*|> writeChannel (writer c) (replicate 100 'b')

This is unsafe, however. The reason is implied by the name oneToOneChannel:

CHP’s standard channels are designed for use by one writer and one reader at a time. Using the output twice in parallel as shown above will sometimes give a run-time error (sometimes you may get away with it, which can make locating the bug hard). The compiler will not stop you writing this code, as encoding the safe-usage restriction in the type system would be very cumbersome. Instead, you must use a shared channel.

We want to share the writing end, but do not need to share the reading end; thus we need an anyToOneChannel, allowing anyone (of multiple writers) to send to one recipient. With this channel, we must use a claim function to grab the shared writing end for ourselves — a form of mutual exclusion. Hence we can get 100 ‘a’s together and 100 ‘b’s together (in either order) by using:

safePrinting :: CHP ()
safePrinting = do
  c <- anyToOneChannel
  printer (reader c)
    <|*|> claim (writer c) (flip writeChannel $ replicate 100 'a')
    <|*|> claim (writer c) (flip writeChannel $ replicate 100 'b')

The claim function takes a shared channel end (e.g. Shared Chanout String), and a function that takes the claimed end (which has the same type as a normal end: Chanout String) and performs some block of code. The claim function claims the channel-end for the duration of the call; releasing is automatic at the end of the block. If you try to use a shared channel-end without claiming it, you’ll get a type error. You can perform multiple communications on the channel during the claim block — we could choose to write “a” to the channel 100 times rather than 100 ‘a’s at once, and the resulting output would still have the ‘a’s uninterrupted.

You would use this printing process in your real application by running it in parallel with the whole system (probably at the very top level), and passing out the shared channel-end to any process that may want to print a message. Each process will then have to claim the channel to send messages, and it is really this claiming process that serialises the messages. Only the printing process actually prints the messages on stdout, which has the effect of stopping any concurrent use of stdout (and hence stopping this garbledness). You could make the printing process print to stderr or to a logfile, if you want to use this method to keep track of errors/warnings/debug messages in your concurrent application.