Orc has no communication primitives like pi-calculus channels^[1] or Erlang mailboxes^[2]. Instead, it makes use of sites to create channels of communication.

The most frequently used of these sites is Buffer. When called, it publishes a new asynchronous FIFO channel. That channel is a site with two methods: get and put. The call c.get() takes the first value from channel c and publishes it, or blocks waiting for a value if none is available. The call c.put(v) puts v as the last item of c and publishes a signal.

A channel may be closed to indicate that it will not be sent any more values. If the channel c is closed, c.put(v) always halts (without modifying the state of the channel), and c.get() halts once c becomes empty. The channel c may be closed by calling either c.close(), which returns a signal once c becomes empty, or c.closenb(), which returns a signal immediately.

In the section on Cor, we were introduced to lists: how to construct them, and how to match them against patterns. While it is certainly feasible to write a specific function with an appropriate pattern match every time we want to access a list, it is helpful to have a handful of common operations on lists and reuse them.

One of the most common uses for a list is to send each of its elements through a sequential combinator. Since the list itself is a single value, we want to walk through the list and publish each one of its elements in parallel as a value. The library function each does exactly that.

Suppose we want to send the message invite to each email address in the list inviteList:

each(inviteList) >address> Email(address, invite)

Orc also adopts many of the list idioms of functional programming. The Orc library contains definitions for most of the standard list functions, such as map and fold. Many of the list functions internally take advantage of concurrency to make use of any available parallelism; for example, the map function dispatches all of the mapped calls concurrently, and assembles the result list once they all return using a fork-join.

Sometimes a source of data is not explicitly represented by a list or other data structure. Instead, it is made available through a site, which returns the values one at a time, each time it is called. We call such a site a stream. It is analogous to an iterator in a language like Java. Functions can also be used as streams, though typically they will not be pure functions, and should only return one value. A call to a stream may halt, to indicate that the end of the data has been reached, and no more values will become available. It is often useful to detect the end of a stream using the otherwise combinator.

Streams are common enough in Orc programming that there is a library function to take all of the available publications from a stream; it is called repeat, and it is analogous to each for lists.

def repeat(f) = f() >x> (x | repeat(f))

The repeat function calls the site or function f with no arguments, publishes its return value, and recurses to query for more values. repeat should be used with sites or functions that block until a value is available. Notice that if any call to f halts, then repeat(f) consequently halts.

For example, it is very easy to treat a channel c as a stream, reading any values put on the channel as they become available:

repeat(c.get)

While lists are a very useful data structure, they are not indexed; it is not possible to get the nth element of a list in constant time. However, this capability is often needed in practice, so the Orc standard library provides a function IArray to create immutable arrays. Once initialized, an immutable array cannot be rewritten; it can only be read.

IArray takes two arguments: an array size, and a function to initialize the array. The function takes the index being initialized as an argument (indices start at 0), and publishes the value to be stored at that array position. Here are a few examples:

{- Create an array of 10 elements; element i is the ith power of 2 -}
IArray(10, lambda(i) = 2 ** i)

{- Create an array of 5 elements; each element is a newly created buffer -}
IArray(5, lambda(_) = Buffer())

The array is used like a function; the call A(i) returns the ith element of the array A. A call with an out-of-bounds index halts.

{- Create an array of 3 buffers -}
val A = IArray(10, lambda(_) = Buffer())

{- Send true on the 0th channel
   and listen for a value on the 0th channel. -}
A(0).put(true) | A(0).get()

Since arrays are accessed by index, there is a library function specifically designed to make programming with indices easier. The function upto(n) publishes all of the numbers from 0 to n-1 simultaneously; thus, it is very easy to access all of the elements of an array simultaneously. Suppose we have an array A of n email addresses and would like to send the message m to each one.

upto(n) >i> A(i) >address> Email(address, m)

Variables in Orc are immutable. There is no assignment operator, and there is no way to change the value of a bound variable. However, it is often useful to have mutable state when writing certain algorithms. The Orc library contains two sites that offer simple mutable storage: Ref and Cell.

The Ref site creates rewritable reference cells.

val r = Ref(0)
println(r.read()) >>
r.write(2) >> 
println(r.read()) >>
stop

These are very similar to ML's ref cells. r.write(v) stores the value v in the reference r, overwriting any previous value, and publishes a signal. r.read() publishes the current value stored in r.

However, unlike in ML, a reference cell can be left initially empty by calling Ref with no arguments. A read operation on an empty cell blocks until the cell is written.

{- Create a cell, and wait 1 second before initializing it. -}
{- The read operation blocks until the write occurs. -}
val r = Ref()
r.read() | Rtimer(1000) >> r.write(1) >> stop

The Orc library also offers write-once reference cells, using the Cell site. A write-once cell has no initial value. Read operations block until the cell has been written. A write operation succeeds only if the cell is empty; subsequent write operations simply halt.

{- Create a cell, try to write to it twice, and read it -}
{- The read will block until a write occurs
   and only one write will succeed. -}
val r = Cell()
  Rtimer(1000) >> r.write(2) >> println("Wrote 2") >> stop
| Rtimer(1000) >> r.write(3) >> println("Wrote 3") >> stop
| r.read()

A word of caution: References, cells, and other mutable objects may be accessed concurrently by many different parts of an Orc program, so race conditions may arise.

{- Create a cell, try to write to it twice, and read it -}
{- The read will block until a write occurs
   and only one write will succeed. -}
val r = Cell()
  Rtimer(1000) >> r := 2 >> println("Wrote 2") >> stop
| Rtimer(1000) >> r := 3 >> println("Wrote 3") >> stop
| r?

Orc does not have any explicit looping constructs. Most of the time, where a loop might be used in other languages, Orc programs use one of two strategies:

Matching a value against multiple patterns, as we have seen it so far, is a linear process, and requires a def whose clauses have patterns in their argument lists. Such a match is linear; each pattern is tried in order until one succeeds.

What if we want to match a value against multiple patterns in parallel, executing every clause that succeeds? Fortunately, this is very easy to do in Orc. Suppose we have an expression F which publishes pairs of integers, and we want to publish a signal for each 3 that occurs.

We could write:

F >(x,y)>
  ( if(x=3) >> signal
  | if(y=3) >> signal ) 

F >x>
  ( x >(3,_)> signal
  | x >(_,3)> signal ) 

This parallel matching technique is sometimes used as an alternative to pattern matching using function clauses, but only when the patterns are mutually exclusive.

For example,

def helper([]) = 0
def helper([_]) = 1
def helper(_:_:_) = 2
helper([4,6])

[4,6] >x>
  x >[]> 0
| x >[_]> 1
| x >_:_:_> 2

def helper([]) = 0
def helper([_]) = 1
def helper(_) = 2
helper([5])

[5] >x>
  x >[]> 0
| x >[_]> 1
| x >_> 2

One of the most common concurrent idioms is a fork-join: run two processes concurrently, and wait for a result from each one. This is very easy to express in Orc. Whenever we write a val declaration, the process computing that value runs in parallel with the rest of the program. So if we write two val declarations, and then form a tuple of their results, this performs a fork-join.

val x = F
val y = G
(x,y)

Fork-joins are a fundamental part of all Orc programs, since they are created by all nested expression translations. In fact, the fork-join we wrote above could be expressed even more simply as just:

(F,G)

def initAll([]) = signal
def initAll(m:ms) = ( m.init() , initAll(ms) ) >> signal

def auction([]) = 0
def auction(b:bs) = max(b.ask(), auction(bs))

Note that all bidders are called simultaneously. Also note that if some bidder fails to return a bid, then the auction will never complete. Later we will see a different solution that addresses the issue of non-termination.

M()  >x>  F | N()  >y>  G

( M() , N() )  >(x,y)>  ( F | G )

Previous sections illustrate how Orc can use the fork-join idiom to process a fixed set of expressions or a list of values. Suppose that instead we wish to process all the publications of an expression F, and once this processing is complete, execute some expression G. For example, F publishes the contents of a text file, one line at a time, and we wish to print each line to the console using the site println, then publish a signal after all lines have been printed.

Sequential composition alone is not sufficient, because we have no way to detect when all of the lines have been processed. A recursive fork-join solution would require that the lines be stored in a traversable data structure like a list, rather than streamed as publications from F. A better solution uses the ; combinator to detect when processing is complete:

F >x> println(x) >> stop ; signal

Since ; only evaluates its right side if the left side does not publish, we suppress the publications on the left side using stop. Here, we assume that we can detect when F halts. If, for example, F is publishing the lines of the file as it receives them over a socket, and the sending party never closes the socket, then F never halts and no signal is published.

The otherwise combinator is also useful for trying alternatives in sequence. Consider an expression of the form F0 ; F1 ; F2 ; .... If F_i does not publish and halts, then F_i+1 is executed. We can think of the F_i's as a series of alternatives that are explored until a publication occurs.

Suppose that we would like to poll a list of buffers for available data. The list of buffers is ordered by priority. The first buffer in the list has the highest priority, so it is polled first. If it has no data, then the next buffer is polled, and so on.

Here is a function which polls a prioritized list of buffers in this way. It publishes the first item that it finds, removing it from the originating buffer. If all buffers are empty, the function halts. We use the getnb ("get non-blocking") method of the buffer, which retrieves the first available item if there is one, and halts otherwise.

def priorityPoll([]) = stop
def priorityPoll(b:bs) = b.getnb() ; priorityPoll(bs)

``Parallel or'' is a classic idiom of parallel programming. The ``parallel or'' operation executes two expressions F and G in parallel, each of which may publish a single boolean, and returns the disjunction of their publications as soon as possible. If one of the expressions publishes true, then the disjunction is true, so it is not necessary to wait for the other expression to publish a value. This holds even if one of the expressions is silent.

The ``parallel or'' of expressions F and G may be expressed in Orc as follows:

let(
    val a = F
    val b = G

      (a || b)
    | if(a) >> true 
    | if(b) >> true 
)

The expression (a || b) waits for both a and b to become available and then publishes their disjunction. However if either a or b is true we can publish true immediately regardless of whether the other variable is available. Therefore we run if(a) >> true and if(b) >> true in parallel to wait for either variable to become true and immediately publish the result true. Since more than one of these expressions may publish true, the surrounding let(...) is necessary to select and publish only the first result.

Timeout, the ability to execute an expression for at most a specified amount of time, is an essential ingredient of fault-tolerant and distributed programming. Orc accomplishes timeout using pruning and the Rtimer site. The following program runs F for at most one second, publishing its result if available and the value 0 otherwise.

let( F | Rtimer(1000) >> 0 )

def auction([]) = 0
def auction(b:bs) = 
  val bid = b.ask() | Rtimer(8000) >> 0
  max(bid, auction(bs))

val (r, b) = (F, true) | (Rtimer(t), false)
if b then G else H

val s = Some(F) | Rtimer(t) >> None()
  s >Some(r)> G
| s >None()>  H

We can use a timer to give a window of priority to one computation over another. In this example, we run expressions F and G concurrently. For one second, F has priority; F's result is published immediately, but G's result is held until the time interval has elapsed. If neither F nor G publishes a result within one second, then the first result from either is published.

val x = F
val y = G
let( y | Rtimer(1000) >> x )

A timer can be used to execute an expression repeatedly at regular intervals, for example to poll a service. Recall the definition of metronome from the previous chapter:

def metronome(t) = signal | Rtimer(t) >> metronome()

The following example publishes "tick" once per second and "tock" once per second after an initial half-second delay. The publications alternate: "tick tock tick tock ...". Note that this program is not defined recursively; the recursion is entirely contained within metronome.

  metronome(1000) >> "tick"
| Rtimer(500) >> metronome(1000) >> "tock"

The Orc combinators restrict the passing of values among their component expressions. However, some programs will require greater flexibility. For example, F <x< G provides F with the first publication of G, but what if F needs the first n publications of G? In cases like this we use channels or other stateful sites to redirect or store publications. We call this technique routing because it involves routing values from one execution to another.

val c = Buffer()
repeat(c.get) << 
    F >x> c.put(x) >> stop 
  | Rtimer(1000) >> c.closenb()

val c = Buffer()
repeat(c.get) << 
  F >x> 
    (if x >= 0 
        then c.put(x) >> stop
        else c.closenb())

val c = Buffer()
val done = Semaphore(0)
repeat(c.get) <<
    F >x> c.put(x) >> stop
  | done.acquire() >> c.closenb()

val c = Buffer()
val done = Semaphore(0)
def allow(0) = done.release() >> stop
def allow(n) = c.get() >x> (x | allow(n-1))
allow(5) << 
    F >x> c.put(x) >> stop 
  | done.acquire() >> c.closenb()

val r = Ref()
(F <x< r.read()) | (G >x> r.write(x))

val c = Buffer()
repeat(c.get) | (F >x> c.put(x) >> stop ; c.close() >> G)  

We can write a function interruptible that implements the interrupt idiom to execute any function in an interruptible way. interruptible(g) calls the function g, which is assumed to take no arguments, and silences its publications. It immediately publishes another function, which we can call at any time to terminate the execution of g. For simplicity, we assume that g itself publishes no values.

Here is a naive implementation that doesn't quite work:

def interruptible(f) =
  val done = Semaphore(0)
  done.release
    << f() >> stop 
     | done.acquire() >> c.closenb()

{- wrong! -}
val stopper = interruptible(g)
...

The function interruptible is correct, but the way it is used causes a strange error. The function g executes, but is always immediately terminated! This happens because the val declaration which binds stopper also kills all of the remaining computation in interruptible(g), including the execution of g itself.

The solution is to bind the variable differently:

def interruptible(f) =
  val done = Semaphore(0)
  done.release
    << f() >> stop 
     | done.acquire() >> c.closenb()

interruptible(g) >stopper>
...

This idiom, wherein a function publishes some value that can be used to monitor or control its execution, arises occasionally in Orc programming. When using this idiom, always remember to avoid terminating that execution accidentally. Since Orc is a structured concurrent language, every process is contained with some other process; kill the containing process, and the contained processes die too.

def lifter() =
  val c = Buffer()
  def loop() = c.get() >f> ( f() >> stop | loop() )
  c.put | loop()

def delayedPrint() = Rtimer(1500) >> println("Delayed 1.5 seconds")

lifter() >lift>
(
println("Running...") 
  << Rtimer(1000) | delayedPrint() | lift(delayedPrint)
)

We consider various concurrent implementations of the classic "list fold" function from functional programming:

def fold(_, [x])  = x
def fold(f, x:xs) = f(x, fold(xs))

This is a seedless fold (sometimes called fold1) which requires that the list be nonempty and uses its first element as a seed. This implementation is short-circuiting --- it may finish early if the reduction operator f does not use its second argument --- but it is not concurrent; no two calls to f can proceed in parallel. However, if f is associative, we can overcome this restriction and implement fold concurrently. If f is also commutative, we can further increase concurrency.

def afold(_, [x]) = x
def afold(f, xs) =
  def step([]) = []
  def step([x]) = [x]
  def step(x:y:xs) = f(x,y):step(xs)
  afold(f, step(xs))

def cfold(f, xs) =
  val c = Buffer()
  
  def xfer([])    = 0
  def xfer(x:xs)  = c.put(x) >> stop | xfer(xs)+1

  def combine(0) = stop
  def combine(1) =  c.get()
  def combine(m) =  c.get() >x> c.get() >y> 
                    ( c.put(f(x,y)) >> stop | combine(m-1))

  xfer(xs) >n> combine(n)