First, a bit of terminology. A distributed program consists of a set of processes that may communicate with one another by sending and receiving messages. A process is like a thread. Processes run concurrently with one another, and every process has a unique ProcessId. There are a couple of important differences between threads and processes, however:

We start by defining the type of messages that our processes will send and receive:

distrib-ping/ping.hs

data Message = Ping ProcessId
             | Pong ProcessId
  deriving (Typeable, Generic)          -- 

instance Binary Message                 -- 

The Ping message contains the ProcessId of the process that sent it so that the target of the message knows where to send the response. The Pong response also includes the ProcessId of the responder so that the master process can tell which process a particular response comes from.

Messages in a distributed program can be sent over the network, which Involves serializing the Haskell data into a stream of bytes before it is sent and deserializing the bytes back into Haskell data at the other end. The distributed-process framework uses the Binary class from the binary package to implement serialization and deserialization, and hence every message type must be an instance of Binary.

The serialization format is under your control. If you want, you can define your own Binary instance that uses a specialized serialization format. Normally, however, you’ll just want an automatically derived Binary instance. Fortunately, the binary package^[54] lets you derive Binary instances using GHC’s DeriveGeneric extension.^[55] To do this, we first derive the Generic class () and then declare an instance of Binary for Message (); GHC fills in the method definitions of this instance for us.

Message types must also be an instance of Typeable, because they can be sent to dynamically typed channels (more about this later). For Typeable, we can derive the instance directly ().

Typeable and Binary are normally packaged up together and referred to as Serializable using the following class provided by Control.Distributed.Process.Serializable:

class (Binary a, Typeable a) => Serializable a
instance (Binary a, Typeable a) => Serializable a

There’s nothing magic about Serializable. Just think of Serializable a as shorthand for (Binary a, Typeable a). You’ll see Serializable used a lot in the Control.Distributed.Process APIs.

Next, we’ll write the code for a “ping server” process. The ping server must wait for a Ping message and then respond with a Pong message.

pingServer :: Process ()
pingServer = do
  Ping from <- expect                              -- 
  say $ printf "ping received from %s" (show from) -- 
  mypid <- getSelfPid                              -- 
  send from (Pong mypid)                           -- 

First of all, notice that we are in the Process monad. As we mentioned earlier, virtually all of the Control.Distributed.Process API is in this monad, and only code running in the Process monad can communicate with other processes and spawn new processes. There has to be a way to get into Process in the first place; we’ll see how that happens shortly, but for now let’s assume we’re already in Process and we need to program the ping server.

At we receive the next message using expect:

expect :: Serializable a => Process a

The expect function receives a message sent directly to this process. Each process has a channel associated with it, and the channel can receive messages of any type. The expect call receives a message of a particular type, where the type is determined by the context. If the type cannot be determined, the compiler will complain that the type is ambiguous, and the usual fix is to add a type signature. In the example just shown, the type of messages to receive is determined by the pattern match on the result, which matches directly on the Ping constructor and thus forces expect to receive messages of the type Message.

The expect function is a little like Haskell’s read function, in that it returns a value whose type depends on the context. But whereas read fails if its argument cannot be parsed as the desired type, expect skips over messages in the queue that do not match and returns the first one that matches. Messages that don’t match the expected type are left in the channel for the time being.

If there are no messages of the right type, expect will block until one arrives. Therefore, it should be used with care: the other messages in the queue are ignored while expect is waiting for the right kind of message to arrive, which could lead to a deadlock. We’ll see later how to wait for several different types of message at the same time.

The say function, called at , causes a message to be logged, which is a useful way to debug your program. Usually, the message will be logged to stderr, but it might be sent somewhere else if the transport layer overrides the default logging process.

At we call getSelfPid to obtain the ProcessId of the current process. The ProcessId of the current process is needed because the Pong message will contain it:

getSelfPid  :: Process ProcessId

And at we send a response back to the originator of the Ping. The function send is used to send a message to a process, and it has the following type:

send :: (Serializable a) => ProcessId -> a -> Process ()

We know which ProcessId to send the Pong to because it was contained in the original Ping message.

Now we need to be able to create processes running pingServer. Although in this example we will be creating the process on the local node, in general we might be creating the process on another node. Functions that will be executed remotely in this way need to be declared explicitly.^[56] The following declaration invokes a bit of Template Haskell magic that creates the necessary infrastructure to allow pingServer to be executed remotely:^[57]

remotable ['pingServer]

Next, we will write the code for the master process. As you might expect, this is an operation of type Process ():

master :: Process ()
master = do
  node <- getSelfNode                               -- 

  say $ printf "spawning on %s" (show node)
  pid <- spawn node $(mkStaticClosure 'pingServer)  -- 

  mypid <- getSelfPid                               -- 
  say $ printf "sending ping to %s" (show pid)
  send pid (Ping mypid)                             -- 

  Pong _ <- expect                                  -- 
  say "pong."

  terminate                                         -- 

spawn :: NodeId -> Closure (Process ()) -> Process ProcessId

All that remains to complete the program is to define our main function, and here it is:

main :: IO ()
main = distribMain (\_ -> master) Main.__remoteTable

The main function calls distribMain from DistribUtils, which is a small module of utilities provided with the sample code to make these examples a bit less cluttered. The distribMain function is a wrapper around the lower-level startup facilities from the distributed-process-simplelocalnet package. It starts up the distributed-process framework with the distributed-process-simplelocalnet backend on a single node.

The first argument to distribMain is the Process computation to run as the master process on the node. It has type [NodeId] -> Process (), where the list of NodeIds are the other nodes in our distributed network. Because this example is running on a single node, we ignore the [NodeId] and just invoke the master function as our master process.

The second argument to distribMain is the metadata used to execute remote calls; in this case we pass Main.__remoteTable, which is generated by the Template Haskell call to remotable we showed earlier.

When you run the program, you should see output like this:^[58]

$ ./ping
pid://localhost:44444:0:3: spawning on nid://localhost:44444:0
pid://localhost:44444:0:3: sending ping to pid://localhost:44444:0:4
pid://localhost:44444:0:4: ping received from pid://localhost:44444:0:3
pid://localhost:44444:0:3: pong.

Each of these messages corresponds to one of the calls to say in the example program, and they are tagged with the date, time, and ProcessId of the process that called say.

In this section, we built the simplest distributed program possible: it spawns a single child process and performs a simple ping/pong message exchange. Here are the key things to take away:

There is a certain amount of boilerplate associated with distributed programming: deriving Binary instances, declaring remotable functions with remotable, starting up the framework with distribMain, and so on. Remember that the distributed-process framework is currently implemented as a library entirely in Haskell. There is no support for distributed programming built into the language or GHC itself, and this accounts for some of the boilerplate. As the framework matures, distributed programming will likely become a smoother experience.

First, I’ll illustrate starting multiple nodes on the same machine and then progress on to multiple machines.

A distributed program consists of a single master node and one or more slave nodes. The master is the node that begins with a process running; the slave nodes just wait until processes are spawned on them.

Let’s start by creating two slave nodes:

$ ./ping-multi slave 44445 &
[3] 58837
$ ./ping-multi slave 44446 &
[4] 58847

The ping-multi program takes two command-line arguments; these are interpreted by the distrbMain function and tell it how to initialize the framework. The first argument is either master or slave and indicates which kind of node to create. The second argument is the TCP port number that this node should use to communicate on, with the default being 44444.^[59] Always use different port numbers when creating multiple nodes on the same machine.

I used & to create these as background processes in the shell. If you’re on Windows, just open a few Command Prompt windows and run the program in each one.

Having started the slaves, we now start the master node:

$ ./ping-multi
pid://localhost:44444:0:3: spawning on nid://localhost:44445:0
pid://localhost:44444:0:3: spawning on nid://localhost:44446:0
pid://localhost:44444:0:3: pinging pid://localhost:44445:0:4
pid://localhost:44444:0:3: pinging pid://localhost:44446:0:4
pid://localhost:44446:0:4: ping received from pid://localhost:44444:0:3
pid://localhost:44445:0:4: ping received from pid://localhost:44444:0:3
pid://localhost:44444:0:3: All pongs successfully received

The first thing to note is that the master node automatically found the two slave nodes. The distributed-process-simplelocalnet package includes a peer discovery mechanism that is designed to automatically locate and connect to other instances running on the same machine or other machines on the local network.

It is also possible to restart the master without restarting the slaves—try invoking ping-multi again, and you should see the same result. The new master node discovers and reconnects to the existing slaves.

If we have multiple machines connected on the same network, we can run a distributed Haskell program on them. The first step is to distribute the binary to all the machines; every machine must be running the same binary. A mismatch in the binary on different machines can cause strange failures, such as errors when decoding messages.

Next, we start the slaves as before, but this time we start slaves on the remote machines and pass an extra argument:

$ ./ping-multi slave 192.168.1.100 44444
$ ./ping-multi slave 192.168.1.101 44444

(The above commands are executed on the appropriate machines.) The second argument is new and gives the IP address that identifies the slave. This is the address that the other nodes will use to contact it, so it must be an address that resolves to the correct machine. It doesn’t have to be an IP address, but using IP addresses is simpler and eliminates a potential source of failure (the DNS).

When the slaves are running, we can start the master:

$ ./ping-multi master 44444
pid://localhost:44444:0:3: spawning on nid://192.168.1.100:44444:0
pid://localhost:44444:0:3: spawning on nid://192.168.1.101:44444:0
pid://localhost:44444:0:3: pinging pid://192.168.1.100:44444:0:5
pid://localhost:44444:0:3: pinging pid://192.168.1.101:44444:0:5
pid://192.168.1.100:44444:0:5: ping received from pid://localhost:44444:0:3
pid://192.168.1.101:44444:0:5: ping received from pid://localhost:44444:0:3
pid://localhost:44444:0:3: All pongs successfully received

The program successfully identified the remote nodes, spawned a processes on each one, and exchanged ping-pong messages with the process on each node.

In the previous section, we waited for a response from each child process in turn, whereas the old waitForPongs version processed the messages in the order they arrived. In this case it isn’t a problem, but suppose some of these messages required a response. Then we might have introduced some extra latency: if a process toward the end of the list replies early, it won’t get a response until the master process has dealt with the messages from the other processes earlier in the list, some of which might take a while to reply.

So we need a way to wait for messages from multiple channels simultaneously. The distributed-process framework has an elegant way to do this. Channels can be merged together to make a single channel that receives messages from any of the original channels. There are two ways to do this:

mergePortsBiased :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)
mergePortsRR     :: Serializable a => [ReceivePort a] -> Process (ReceivePort a)

The difference is in the order in which messages arrive on the merged channel. In mergePortsBiased, each receive searches the ports in left-to-right order for a message, returning the first message it finds. The alternative is mergePortsRR (the RR stands for "round robin") which also searches left to right, but rotates the list by one element after each receive, with the leftmost port moving to the end of the list.

One important thing to note is that merging channels does not affect the original channel; we can still receive messages from either source, and indeed there is no problem with merging multiple overlapping sets of channels.^[61]

Here is the ping example with channels, where instead of waiting for the responses one by one, we merge the channels together and wait for all the responses simultaneously.

distrib-ping/ping-tc-merge.hs

master :: [NodeId] -> Process ()
master peers = do

  ps <- forM peers $ \nid -> do
          say $ printf "spawning on %s" (show nid)
          spawn nid $(mkStaticClosure 'pingServer)

  ports <- forM ps $ \pid -> do
    say $ printf "pinging %s" (show pid)
    (sendport,recvport) <- newChan
    send pid (Ping sendport)
    return recvport

  oneport <- mergePortsBiased ports     -- 
  waitForPongs oneport ps               -- 

  say "All pongs successfully received"
  terminate

waitForPongs :: ReceivePort ProcessId -> [ProcessId] -> Process ()
waitForPongs _ [] = return ()
waitForPongs port ps = do
  pid <- receiveChan port
  waitForPongs port (filter (/= pid) ps)

In a distributed system, parts of the running program may fail at any time due to circumstances beyond our control. Such a failure typically results in one or more of the processes in our network becoming disconnected without warning; there is no exception and no opportunity to clean up whatever it was doing. Perhaps the hardware it was running on failed, or the network on which we were communicating with it stopped working.

A far-reaching approach for such failures can be seen in Erlang, a programming language with distributed programming at its heart. The only mechanism for communication is message passing, so every concurrent Erlang program is fundamentally distributable. The Erlang designers promote a particular philosophy for dealing with failure, often known by its catchphrase: "Let it crash." The basic principle is that since in a distributed system we must already be prepared for a process to simply disappear, we might as well deal with all kinds of failure in this way because doing so makes failure handling much simpler. And since failure handling is difficult to test, making it simpler is highly desirable.

Concretely, instead of trying to enumerate local failure conditions and handle them in some way, we can just let them propagate to the top of the process and let the process die. The distributed program must be prepared for this eventuality already (since this is a distributed system), so the system will recover in some way: perhaps by restarting the failed process in some known-good state and logging the failure somewhere.

Thus the granularity at which we have to consider failure is the process, and we can design our applications such that individual processes can fail without catastrophic consequences. A process will probably have some internal state that is lost when it dies, but the parent should know how to construct the initial state to restart the process or to propagate the failure to a higher layer that can.

We will need a few changes to the data structures compared with the multithreaded server. When one client sends a message to another client connected to a different server, we need to know where to send the message. So each server will need to keep a list of all the clients connected to any server in the network, along with the server to which the client is connected. The information about a client now has two possibilities: either it is a local client (connected to this server), or a remote client (connected to a different server).

distrib-chat/chat.hs

type ClientName = String

data Client
  = ClientLocal   LocalClient
  | ClientRemote  RemoteClient

data RemoteClient = RemoteClient
       { remoteName :: ClientName
       , clientHome :: ProcessId
       }

data LocalClient = LocalClient
       { localName      :: ClientName
       , clientHandle   :: Handle
       , clientKicked   :: TVar (Maybe String)
       , clientSendChan :: TChan Message
       }

clientName :: Client -> ClientName
clientName (ClientLocal  c) = localName c
clientName (ClientRemote c) = remoteName c

newLocalClient :: ClientName -> Handle -> STM LocalClient
newLocalClient name handle = do
  c <- newTChan
  k <- newTVar Nothing
  return LocalClient { localName      = name
                     , clientHandle   = handle
                     , clientSendChan = c
                     , clientKicked   = k
                     }

LocalClient is what we previously called Client, and RemoteClient is a client connected to another server. The Client type is now a disjunction of these two, with constructors ClientLocal and ClientRemote.

The Message type is as before, except that we need to derive Typeable and Binary, because Messages will be sent over the network:

data Message = Notice String
             | Tell ClientName String
             | Broadcast ClientName String
             | Command String
  deriving (Typeable, Generic)

instance Binary Message

Servers need to communicate with one another, and the kinds of messages they need to send are richer than Message. For example, servers need to tell one another when a new client connects, or one client kicks another. So we have a new type for messages sent between servers, which we call PMessage:

data PMessage
  = MsgServers            [ProcessId]
  | MsgSend               ClientName Message
  | MsgBroadcast          Message
  | MsgKick               ClientName ClientName
  | MsgNewClient          ClientName ProcessId
  | MsgClientDisconnected ClientName ProcessId
  deriving (Typeable, Generic)

instance Binary PMessage

Most of these are self-explanatory, except for one: MsgServers is a special message sent to each server node when it starts up, telling it the ProcessIds of all the server nodes in the network.

The Server type previously contained only the mapping from ClientName to Client, but now it needs some more information:

data Server = Server
  { clients   :: TVar (Map ClientName Client)
  , proxychan :: TChan (Process ())
  , servers   :: TVar [ProcessId]
  , spid      :: ProcessId
  }

newServer :: [ProcessId] -> Process Server
newServer pids = do
  pid <- getSelfPid
  liftIO $ do
    s <- newTVarIO pids
    c <- newTVarIO Map.empty
    o <- newTChanIO
    return Server { clients = c, servers = s, proxychan = o, spid = pid }

clients is the client mapping, as before; servers is the list of other server ProcessIds, and spid is the ProcessId of this server (for convenience).

The proxychan field pertains to an added bit of complexity in our distributed architecture. Remember that we are leaving as much of the existing server infrastructure intact as possible; that means the existing server threads are ordinary forkIO threads. A forkIO thread cannot perform operations in the Process monad, yet we certainly need to be able to do that somehow because certain actions by a client must trigger communication with other servers in the network. So the trick we use is a proxy, which is a process that reads actions from a TChan and performs them in the Process monad. To have a Process action performed from an IO thread, we simply queue it on the proxy TChan. Each server has a single proxy channel, created when the server starts up and stored in the proxychan field of Server.

Next, we need a few small utilities. First, a way to send a Message to a LocalClient:

sendLocal :: LocalClient -> Message -> STM ()
sendLocal LocalClient{..} msg = writeTChan clientSendChan msg

The following function, sendRemote, sends a PMessage to a remote server. To do this, it needs to use the proxychan (which it gets from the Server) and it needs the pid of the destination process:

sendRemote :: Server -> ProcessId -> PMessage -> STM ()
sendRemote Server{..} pid pmsg = writeTChan proxychan (send pid pmsg)

Now that we can send both local and remote messages, we can define sendMessage, which sends a Message to any client:

sendMessage :: Server -> Client -> Message -> STM ()
sendMessage server (ClientLocal client) msg =
    sendLocal client msg
sendMessage server (ClientRemote client) msg =
    sendRemote server (clientHome client) (MsgSend (remoteName client) msg)

A variant sends a message to a named client or returns False if the client is not connected:

sendToName :: Server -> ClientName -> Message -> STM Bool
sendToName server@Server{..} name msg = do
    clientmap <- readTVar clients
    case Map.lookup name clientmap of
        Nothing     -> return False
        Just client -> sendMessage server client msg >> return True

Next, we consider broadcasting messages. First, we need a way to send a PMessage to all the connected servers:

sendRemoteAll :: Server -> PMessage -> STM ()
sendRemoteAll server@Server{..} pmsg = do
    pids <- readTVar servers
    mapM_ (\pid -> sendRemote server pid pmsg) pids

We also need a broadcastLocal function that sends a message to the local clients only:

broadcastLocal :: Server -> Message -> STM ()
broadcastLocal server@Server{..} msg = do
    clientmap <- readTVar clients
    mapM_ sendIfLocal (Map.elems clientmap)
  where
    sendIfLocal (ClientLocal c)  = sendLocal c msg
    sendIfLocal (ClientRemote _) = return ()

This function works by calling an auxiliary function sendIfLocal on each of the clients, which calls sendLocal if the client is local and does nothing if the client is remote.

Putting sendRemoteAll and broadcastLocal together, we can broadcast a Message to everyone:

broadcast :: Server -> Message -> STM ()
broadcast server@Server{..} msg = do
    sendRemoteAll server (MsgBroadcast msg)
    broadcastLocal server msg

The rest of the local server code is almost identical to that in “A Chat Server”, so we don’t reproduce it here. The only important differences are that we need to inform other servers whenever a client connects or disconnects by calling sendRemoteAll with a MsgNewClient or MsgClientDisconnected respectively.

The interesting part is how we handle distribution. Previously, the main function was responsible for setting up the network socket and accepting new connections. This is now delegated to a function socketListener, which is otherwise identical to the previous main:

socketListener :: Server -> Int -> IO ()
socketListener server port = 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)
      forkFinally (talk server handle)
                  (\_ -> hClose handle)

We need a function to implement the proxy, described above in “Sending Messages”. All it does is repeatedly read Process () values from the proxychan and execute them:

proxy :: Server -> Process ()
proxy Server{..} = forever $ join $ liftIO $ atomically $ readTChan proxychan

Now, the chatServer function is the main Process () action that implements a chat server:

chatServer :: Int -> Process ()
chatServer port = do
  server <- newServer []
  liftIO $ forkIO (socketListener server port)           -- 
  spawnLocal (proxy server)                              -- 
  forever $ do m <- expect; handleRemoteMessage server m -- 

handleRemoteMessage :: Server -> PMessage -> Process ()
handleRemoteMessage server@Server{..} m = liftIO $ atomically $
  case m of
    MsgServers pids  -> writeTVar servers (filter (/= spid) pids) -- 
    MsgSend name msg -> void $ sendToName server name msg         -- 
    MsgBroadcast msg -> broadcastLocal server msg                 -- 
    MsgKick who by   -> kick server who by                        -- 

    MsgNewClient name pid -> do                                   -- 
        ok <- checkAddClient server (ClientRemote (RemoteClient name pid))
        when (not ok) $
          sendRemote server pid (MsgKick name "SYSTEM")

    MsgClientDisconnected name pid -> do                          -- 
         clientmap <- readTVar clients
         case Map.lookup name clientmap of
            Nothing -> return ()
            Just (ClientRemote (RemoteClient _ pid')) | pid == pid' ->
              deleteClient server name
            Just _ ->
              return ()

Now that the server code is in place, we just need to write the code to start up the whole distributed network. The main function invokes master on the master node:

port :: Int
port = 44444

master :: [NodeId] -> Process ()
master peers = do

  let run nid port = do
         say $ printf "spawning on %s" (show nid)
         spawn nid ($(mkClosure 'chatServer) port)

  pids <- zipWithM run peers [port+1..]
  mypid <- getSelfPid
  let all_pids = mypid : pids
  mapM_ (\pid <- send pid (MsgServers)) all_pids

  chatServer port

main = distribMain master Main.__remoteTable

The master function is fairly straightforward. It spawns chatServer on each of the slaves, using increasing port numbers, and then sends a MsgServers message to each server process containing a list of all the server ProcessIds.^[62]

We can start up a few nodes on a single machine like so:

$ ./chat slave 55551 & ./chat slave 55552 & ./chat master 55553
pid://localhost:55553:0:3: spawning on nid://localhost:55552:0
pid://localhost:55553:0:3: spawning on nid://localhost:55551:0
Listening on port 44444
Listening on port 44445
Listening on port 44446

(Remember the port numbers given on the command line are the ports used by the distributed-process framework; the ports that the chat server listens to are hardcoded to 44444, 44445, …)

Then connect to one of the nodes:

$ nc localhost 44445
What is your name?
Fred
*** Fred has connected

And connect to a different node:

$ nc localhost 44446
What is your name?
Bob
*** Bob has connected
hi
<Bob>: hi

We should now see the new activity on the first connection:

*** Bob has connected
<Bob>: hi

Our distributed server works only with a fixed set of nodes, which makes it quite limited. In practice, we want to be able to add and remove nodes from the network at will. Nodes will disconnect due to network and hardware outages, and we would like to be able to add new nodes without restarting the entire network.

My sketch implementation can be found in distrib-chat/chat-noslave.hs, but you might want to try implementing this for yourself. Some hints on how to go about it follow.

We need to abandon the master/slave architecture; every node will be equal. Instead of using our DistribUtils module, we can use the following sequence to initialize the simplelocalnet backend and start up a node:

distrib-chat/chat-noslave.hs

main = do
 [port, chat_port] <- getArgs
 backend <- initializeBackend "localhost" port
                              (Main.__remoteTable initRemoteTable)
 node <- newLocalNode backend
 Node.runProcess node (master backend chat_port)

Now the function master has type Backend -> String -> Process () and runs on every node. The outline of the rest of the implementation is as follows: