kms

Concurrent implementation of the Daytime Protocol in Haskell

One example from Parallel and Concurrent Programming in Haskell is a concurrent network server. The server given in the book implements an informally specified doubling protocol, where each submitted line gets parsed as an Integer and returns the double of the input.

Back in May, Andrew Clarkson gave a talk at the PyMNtos meeting about Asynchronous IO in Python. As an example, he included an asynchronous version of a Daytime server.

Let’s take the Haskell doubling server, and make it a (TCP) Daytime server.

… but first, some imports.

> import Control.Monad (forever)

forever is useful for making loops out of void functions (IO ()).

> import Text.Printf (printf)

printf behaves similarly to how it does in other languages: it takes a format string and items to interpolate into the format string.

> import Data.Time (getCurrentTime, formatTime)
> import System.Locale (defaultTimeLocale)

These are used to get the current time, and format it with the default locale.

> import System.IO (
>     Handle

A handle is a place for IO to stream data.

>   , stdout

stdout is the default handle used for things like putStrLn.

>   , hPutStrLn

This is the handle equivalents of putStrLn. Rather than assuming the standard handle, the handle is passed in explicitly. One can define putStrLn (in point-free style) as putStrLn = hPutStrLn stdout.

>   , hClose

Handles need to be closed when you are done with them.

>   )

> import Network (
>     withSocketsDo

This is needed on Windows to initialize the networking subsystem. It is here for portability reasons.

>   , listenOn

This opens a socket on a specified port.

>   , PortID(PortNumber)

To specify the port, we’ll pass in a PortNumber.

>   , accept

accept takes a socket and returns a tuple of a Handle, a host, and the port.

>   )

> import Control.Concurrent (
>     forkFinally

forkFinally creates a new (light-weight) thread to run a specified process and when it completes, it “finally” runs another command. You’ll see when we get there.

>   , threadDelay

It turns out our single-threaded implementation is quite efficient, so we’ll add a slight delay to make clear how concurrency affects the server.

>   )

This program has four different main functions. The fourth one is the concurrent Daytime server, so we’ll use that implementation as our main main:

> main = main4

First, lets write a non-server version of what a Daytime server does. According to the specification:

“Once a connection is established the current date and time is sent out the connection as a ascii character string (and any data received is thrown away). The service closes the connection after sending the quote.”

A non-server of this would just be a simple version of date

> main1 :: IO ()
> main1 = do
>   ct <- getCurrentTime
>   let time = formatTime defaultTimeLocale "%F %T" ct
>   putStrLn time

This outputs the time, and halts.

In order to work with sockets, however, we’ll need to use the Handle equivalent program. We’ll also adding a slight delay to make the benefits of concurrency clear later.

To convert main1 to use handles explicitly, we’ll pass an output handle in, and use hPutStrLn instead of putStrLn:

> mainWith :: Handle -> IO ()
> mainWith outH = do
>   threadDelay (10^6) -- to simulate "real" work.
>   ct <- getCurrentTime
>   let time = formatTime defaultTimeLocale "%F %T" ct
>   hPutStrLn outH time

Ignoring the threadDelay, the equivalent program to main1 would be:

> main2 :: IO ()
> main2 = mainWith stdout

Now we can get on with implementing the server. The specification establishes port 13 for the Daytime protocol.¹

> port :: Int
> port = 13

Let’s start out with the server implementation from the book:

> main3 :: IO ()
> main3 = withSocketsDo $ do
>   sock <- listenOn (PortNumber (fromIntegral port))
>   printf "Listening on port %d\n" port
>   forever $ do
>     (handle, host, port) <- accept sock
>     printf "Accepted connection from %s: %s\n" host (show port)

Here’s where things diverge. We use our mainWith in place of talk. Since we can pass handles into mainWith, we can pass handles returned by accept into mainWith:

>     mainWith handle

Don’t forget to close the handle after the connection has been handled (as specified)!

>     hClose handle

This handles each connection in sequence. Since handling the connection takes over a second of work (due to the thread delay), it can only respond to one connection per second.

Let’s add some concurrency! This first part is the same:

> main4 :: IO ()
> main4 = withSocketsDo $ do
>   sock <- listenOn (PortNumber (fromIntegral port))
>   printf "Listening on port %d\n" port
>   forever $ do
>     (handle, host, port) <- accept sock
>     printf "Accepted connection from %s: %s\n" host (show port)

Here’s where forkFinally comes in. Instead of calling mainWith handle, we fork a new thread to call it. When the thread completes, we (finally) close the handle.

>     forkFinally (mainWith handle)
>                 (\_ -> hClose handle)

Since a new thread is created for each connection, we are no longer limited to one connection per second.

To run this file²:

runhaskell daytime.lhs

To test the concurrency³:

yes "nc localhost 13" | parallel -j 32

This streams commands to connect to localhost on port 13, and uses parallel to have 32 worker threads running those commands. With main3, you should see one response per second, whereas with main4, you should see 32 responses per second.