Shared channels
Table of contents
Cake or disappointment?
All the examples we have seen so far feature inter-thread communication via channels, whose endpoints are held by exactly one thread. This scheme makes sure that there are no unexpected messages on buffers: senders write what is supposed to be written, and receivers read exactly values of the expected type. Compounded with the use of the forkWith primitive to create new threads and new channels, programs are expected not to deadlock.
But someone must take the last cake in the store. Here’s the scenario taken from Kokke, Morris, Wadler: Towards Races in Linear Logic. Log. Methods Comput. Sci. 16(4) (2020).
Ami and Boé are working from home one morning when they each get a craving for a slice of cake. Being denizens of the web, they quickly find the nearest store which does home deliveries. Unfortunately for them, they both order their cake at the same store, which has only one slice left. After that, all it can deliver is disappointment.
The crucial difference between this example and those seen so far is that Ami and Boé both order their cake at the same store. If the interaction with the store is via a channel (what else could it be?), then both cake lovers would have to share one channel endpoint.
Not any channel endpoint can be shared. If two threads share a channel of type !Int ; !Char and they do not synchronise, chances are that the Char message may arrive at the destination before the Int message. The only channels that can be shared are stateless, channels whose contents do not change over time, and in particular that cannot be explicitly closed. Intuitively, one such channel can be understood as !Int ; !Int ; ....
For channels that forever send values of type U we use a type of the form *!U. For a channel that forever selects label l or m, we write *+{l,m}. The table below summarises the unrestricted types.
| Type | Description |
|---|---|
*!U | Forever send U |
*?U | Forever receive U |
*+{l1,...,ln} | Forever send one of the l1,…,ln labels |
*&{l1,...,ln} | Forever receive one of the l1,…,ln labels |
As with linear types, unrestricted types come in dual pairs:
S | Dual S |
|---|---|
*!U | *?U |
*+{l1,...,ln} | *&{l1,...,ln} |
All shared types are of kind *C, meaning that a) they can be shared (*) and b) that they can be used to create channels (C).
We start with the type of the cake store, as seen from the side of the store: it either provides Cake or Disappointment.
type CakeStore = *+{Cake, Disappointment}
The cake store first selects Cake, then Disappointment, and then terminates. Because we are selecting on unrestricted channels, we use select_ and not select which should be kept for linear channels.
cakeStore : CakeStore -> ()
cakeStore c = c |> select_ Cake
|> select_ Disappointment ; ()
Notice how the function “abandons” channel c. The expression c |> select_ Cake |> select_ Disappointment is of type CakeStore. Because of its unrestricted nature, it can be discarded, in this case at the left-hand-side of the semicolon operator.
Cake lovers branch on the incoming label and print a message accordingly. Because we are pattern-matching on unrestricted channels we use a pattern (*&Cake), rather than (&Cake), which should be kept for linear channels.
cakeLover : String -> Dual CakeStore -> ()
cakeLover name (*&Cake) = putStrLn (name ++ " got cake!")
cakeLover name (*&Disappointment) = putStrLn (name ++ " got disappointment")
To run a system with a store and two clients, we create a CakeStore channel and distribute its two endpoints to the appropriate actors.
_ =
let (s, c) = channel @CakeStore in
fork (\_ -> cakeLover "Ami" c);
fork (\_ -> cakeStore s);
cakeLover "Boé" c
Let us analyse the possible outputs of the above program. One is
Ami got disappointment
Boé got cake!
Another is:
Boé got disappointment
Ami got cake!
But there are many others. The program terminates when thread cakeLover "Boé" c terminates, so there will always be a Boé message. But whether you’ll read an Ami message or not, depends on the interleaving of the various operations in the three threads.
There are different solutions to this problem. A simple one uses the fork-join pattern discussed in the next section. It requires an extra, separate, channel to communicate task completion. Another refactors the program, so that the same channel that conveys cakes or disappointments also signals task completion.
A summary of the basic elements of unrestricted interaction
The table below summarises what we have seen on unrestricted (shared) channel type operations. Being stateless, unrestricted channels cannot be closed, so, unlike linear channels, there is no Close/Wait pair. Send and receive type operators are likewise not available on unrestricted channels.
S | Dual S | Operation |
|---|---|---|
*!U | *?U | Value exchange |
*+{l1,...,ln} | *&{l1,...,ln} | Choice |
| Output | Input | |
| Positive | Negative | |
| Chaining available | Pattern matching available | |
| Nonblocking operation | Blocking operation |
Each positive type has a corresponding chaining operator:
| Positive type | Operator | Chaining operator |
|---|---|---|
*!U | send_ : forall #m (a : m T) -> a -> *!a -m-> *!a | c |> send_ v |> ... |
*+{l1,...,ln} | select_ l : *+{l1,...,ln} | c |> select_ l |> ... |
Dually, each negative type has a corresponding pattern:
| Negative type | Operator | Pattern |
|---|---|---|
*?U | receive_ : forall (a : 1T) -> *?a -> a | *?x ; p |
*&{l1,...,ln} | case exp of *&l p -> ... | *&l p |
Fork-join
Let us leave the cake store for a while. Not all the programs we have seen terminate as expected. What do you expect to read on the console after running this code?
main =
fork (\_ -> putChar 'A') ;
fork (\_ -> putChar 'B') ;
fork (\_ -> putChar 'C') ;
()
The precise answer is any sequence of distinct letters taken from the set {A, B, C}, of sizes 0 to 3. Yes, no output is a possible answer. It happens when main terminates before any of the three putChar threads managed to write their character.
If reading the three letters on the console is intended, then main must wait for the three threads to terminate. Since inter-thread communication is by message passing alone, we need a channel, a channel with three writers and one reader, that is a shared channel.
The fork-join pattern is a solution to this problem. It comes with three ingredients: a ForkJoin channel type
type ForkJoin
a join operation for child channels to communicate task completion:
join : ForkJoin -> ()
and a await operation for the parent thread to wait for a given number of children:
await : Int -> Dual ForkJoin -> ()
Using fork-join is easy. The parent threads creates a ForkJoin channel, distributes one end to each of its children (while forking them) and awaits the completion:
_ =
let (w, r) = channel @ForkJoin in
fork (\_ -> putChar 'A'; join w) ;
fork (\_ -> putChar 'B'; join w) ;
fork (\_ -> putChar 'C'; join w) ;
await 3 r
We are now guaranteed to read three distinct characters on the console, even if we cannot anticipate their order.
It is instructive to study the implementation of the fork-join primitives. A ForkJoin channel is a channel on which threads can select_ Over an unbounded number of times.
type ForkJoin = *+{Over}
To signal completion of a child thread to the parent waiting on the join channel, we select_ Over on a ForkJoin channel. The operation returns the continuation channel, of type ForkJoin. The semicolon operator discards the (unrestricted) channel and returns ().
join : ForkJoin -> ()
join c = select_ Over c ; ()
To wait until n child threads have signalled completion through the join channel, we use the await primitive which receives n Over labels, by making use of the times operation in the Prelude.
await : Int -> Dual ForkJoin -> ()
await n c = times @() n (\_ -> case c of *&Over -> ())
Session initiation
The shared protocol of the “Cake or Disappointment?” section was extremely simple: all the channel could convey was a Cake or a Disappointment option. What if we aim at more sophisticated protocols? (imagine an online store with your shopping cart).
Rather than an online store, we discuss a simple, yet useful abstraction, that of a reference cell. Reference cells should answer on shared channels. Clients seek read or write operations. These operations are of different natures. We could try a solution similar to the cake store, featuring two operations, Read and Write. The problem is that the former should be followed by a receive (?) operation and the latter by a send (!) operation, something that shared channels do not allow.
The uniform, stateless, nature of shared channels points into another direction: we split each interaction into two phases. First, session initiation runs on the shared channel and establishes a fresh linear channel between a client and the cell; then the cell operation itself runs on that linear channel. Session initiation is driven by the client: when it needs a session, the client creates a new linear channel, keeps one endpoint for itself and sends the other endpoint to the cell on the shared channel.
Cell operations are Read or Write, followed by the appropriate read or write operation (?a or !a). In any case, the channel must be closed. This is the type of the linear channel, as seen from the side of the client.
type CellOp : *T -> 1C
type CellOp a = +{Read: ?a, Write: !a} ; Close
The type of a reference cell of values of type a is a shared channel on which clients send the cell’s endpoint of a cell operation channel, that is, a Dual (CellOp a).
type CellRef : *T -> *C
type CellRef a = *!(Dual (CellOp a))
Notice that both CellRef and CellOp are type operators that expect unrestricted values (*). That is because the values are to be read and written an arbitrary number of times. Notice further that they are of base kind C; that allows creating channels of both types.
Session initiation requires the collaboration of the two parties: the reference cell and its clients. We start with the latter. All a client needs to do is to request a cell operation channel on the cell reference. Function request in the Prelude does exactly that: it creates a channel, sends one endpoint on the given shared channel and returns the other endpoint:
request : forall (a : 1C) -> *!a -> Dual a
Requesting does not block: the client can start using its endpoint right away, even before the cell picks up the other endpoint. The shared channel itself is not returned by the function: it can be reused as often as needed.
Given a CellRef a, function request returns a CellOp a. To write a value x to the cell, one first request a CellOp on which one selects Write, sends x and closes the channel:
write : forall (a : *T) -> a -> CellRef a -> ()
write x s = s |> request |> select Write |> sendAndClose x
The read operation is similar:
read : forall (a : *T) -> CellRef a -> a
read s = s |> request |> select Read |> receiveAndClose
We now address the server side of session initiation. The server accepts a request on the dual of the CellRef channel, of type *?(Dual (CellOp a)). Function accept receives the endpoint sent by some client and returns it; it blocks until some client issues a request:
accept : forall (a : 1C) -> *?a -> a
The cell server accepts a connection from some client, and dispatches on the client operation: Write or Read. Using receiveAndWait and sendAndWait we can easily consume a Dual (CellOp a) channel.
cell : forall (a : *T) -> a -> Dual (CellRef a) -> ()
cell n c =
case accept c of
&Write s -> cell (receiveAndWait s) c
&Read s -> sendAndWait n s ; cell n c
To test the cell and its clients we could try forking a few write and read threads, but we must make sure they all complete their tasks before the main thread ends. For that we use the fork-join pattern again. The below program forks three read threads and two write clients. Expect to read any sequence of numbers 0, 5 and 6, possibly duplicated or triplicated.
_ =
let c = forkWith (cell 0)
(j, a) = channel @ForkJoin in
fork (\_ -> c |> read |> print ; join j) ;
fork (\_ -> c |> read |> print ; join j) ;
fork (\_ -> c |> write 5 ; join j) ;
fork (\_ -> c |> write 6 ; join j) ;
fork (\_ -> c |> read |> print ; join j) ;
await 5 a
Session initiation requires request c on one channel endpoint and accept c on the other. The below table summarises the types and operators involved.
| Client | Server | |
|---|---|---|
request c | accept c | Operator |
*!U | *?U | Parameter type |
Dual U | U | Return type |
Both functions are part of the Prelude, but are not difficult to write. Function request creates a channel, sends one endpoint on the given channel and returns the other endpoint. Function accept is just receive_ on a shared channel that conveys channel endpoints.
request : forall (a : 1C) -> *!a -> Dual a
request @a c =
let (x, y) = channel @a in
send_ x c;
y
accept : forall (a : 1C) -> *?a -> a
accept @a c = receive_ @a c
Why should the client, rather than the server, create the new channel? Because the client is the party that knows when a session is needed. Each session is created on demand, exactly when some client asks for it, and the server picks it up when it is ready to serve. A server never creates sessions in advance that nobody may ever use.
Running servers
The Prelude contains a handy abbreviation for writing servers that serve clients sequentially, one at a time. The function we have in mind is:
runServer : forall (a : *T) (b : 1C) -> (a -> Dual b -> a) -> a -> *?(Dual b) -> Void @*T
where a represents the state of the server, b the linear channel on which to serve one particular client (as seen from the side of the client). The function accepts a function to handle a particular client, the initial value of the state, and the server’s end of the shared channel. Since it never returns, its result type is Void @*T.
Using runServer, receiveAndWait and sendAndWait we can consume a CellOp channel.
cell : forall (a : *T) -> a -> Dual (CellRef a) -> Void @*T
cell @a =
runServer serveOne
where
serveOne : forall a -> a -> Dual (CellOp a) -> a
serveOne _ (&Write c) = receiveAndWait c
serveOne x (&Read c) = sendAndWait x c ; x
Note: This is an example where the type checker needs a hand. Omitting the type annotation currently yields the below error. We decided to fix it by binding the signature’s type variables with @a patterns on the left-hand side, as the error message suggests.
Type mismatch:
|
9 | runServer serveOne
| ^^^^^^^^^^^^^^^^^^
Couldn't match expected type `forall (a : *T) -> a -> Dual (CellRef a) -> Void @*T`, taken from:
MemoryCell.fst:7:8–7:60
|
7 | cell : forall (a : *T) -> a -> Dual (CellRef a) -> Void @*T
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
with actual type `_ -> *?Dual _ -> Void @*T`, taken from:
StandardLib/Prelude.fst:456:63–456:90
|
456 | runServer : forall (a : *T) (b : 1C) -> (a -> Dual b -> a) -> a -> *?(Dual b) -> Void @*T
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^
Type inference could not determine a type argument here (shown as `_`).
Consider annotating the application with an explicit type argument (e.g. `f @a`),
binding the signature's type variables with `@a` patterns on the left-hand side.
Multiple producer, multiple consumer
Many programming languages offer channels of the multiple producer, single consumer (mpsc) sort. But in session types there is no notion of producer or consumer. This means that we may have more than one thread reading on the same channel (in fact zero or more).
Let us think of a remote predicate evaluator. To make things more precise, think of the gz, greater than zero, predicate. As before, we need a shared channel, on which sessions may be initiated. Let us call the shared channel PredEvalService and the linear channel PredEvalSession.
type PredEvalSession : 1C
type PredEvalSession = !Int ; ?Bool ; Close
type PredEvalService : *C
type PredEvalService = *!(Dual PredEvalSession)
As before these are two types of base kind C (meaning that channels can be created from the types). Session is linear, 1, and service is shared, *.
Clients interact on PredEvalService. They request a session on which they provide an integer and wait for the result, before closing the channel:
client : Int -> PredEvalService -> Bool
client n s =
request s |> send n |> receiveAndClose
The gz function performs session initiation via a call to accept. It then receives an integer and returns the corresponding boolean, before waiting for the channel to be closed.
gz : Dual PredEvalService -> ()
gz s =
let (x, c) = receive (accept s) in sendAndWait (x > 0) c
We may now place several copies of gz and client running in parallel. Again, we make use of the fork-join pattern to make sure all threads are given a chance to terminate.
_ =
let (c, s) = channel @PredEvalService
(j, a) = channel @ForkJoin in
fork (\_ -> s |> gz ; join j) ;
fork (\_ -> c |> client 5 |> print ; join j) ;
fork (\_ -> c |> client (-1) |> print ; join j) ;
fork (\_ -> s |> gz ; join j) ;
fork (\_ -> s |> gz ; join j) ;
fork (\_ -> c |> client 0 |> print ; join j) ;
await 6 a
The console should read True, False, False, in any order. The code above is brittle. Because each gz server serves one client only, one must have the same number of clients and servers. Any deviation from this pattern would leave a thread blocked indefinitely: an extra server waits forever in accept for a client that never comes, and an extra client waits forever for an answer on a session that no server picks up.