[{TableOfContents}]

!! Associative Fold

We consider implementing a fold operator in parallel on a list of items. We
first consider the case when the fold operator is associative.

We define afold(b,xs) where b is a binary associative function over
two arguments, and xs is a non-empty list. The value returned by afold
is as follows:

[{orc runnable='false'

afold(_,[x])  = x
afold(b,x:xs) = b(x,afold(xs))
}]

The parallel implementation builds a sequence of lists. The initial
list is the given one. Each subsequent list is about half the size of
the previous list, obtained by folding each disjoint pair of adjecent
items. In the following, afold'(xs) builds the fold from xs. Execution
of one instance of afold'() is in parallel, though different instances
are executed in sequence.

[{orc runnable='false'

def afold(b, [x]) = x
def afold(b, xs) =
  def afold'([]) = []
  def afold'([x]) = [x]
  def afold'(x:y:xs) = b(x,y):afold'(xs)
  afold(b, afold'(xs))
}]

!! Associative, Commutative Fold

We devise a better strategy when the fold operator is both associative
and commutative. We define cfold(b,xs) where b is a binary associative and commutative
function over two arguments, and xs is a non-empty list. The value
returned by cfold is as follows:

[{orc runnable='false'

cfold(_,[x])  = x
cfold(b,x:xs) = b(x,cfold(xs))
}]

Our implementation differs from afold() in that all list items are put
in a buffer in arbitrary order, two items are folded as
soon as possible, and their result is put in the buffer.

[{orc runnable='false'

def cfold(b, xs) =
  val c = Buffer()
  
  --------- Transfer all items of the argument list to Buffer c ---
  -- The order of items in c is arbitrary.
  def xfer([])    = stop
  def xfer(x:xs)  = c.put(x) >> stop | xfer(xs)
  ------------------------ End of xfer ----------------------------

  ---------- combine(n) computes fold of n items from Buffer c ----
  def combine(1) =  c.get()
  def combine(m) =  c.get() >x> c.get() >y> ( c.put(b(x,y)) >> stop | combine(m-1))
  --------------------- End of combine(n) --------------------------

  {- Goal of cfold -}
  xfer(xs) | combine(length(xs))
}]

!! Map with Associative Fold

We generalize the functions defined earlier; first we map list items
using a supplied function and then compute their fold. And, we
distinguish between associative fold, and commmutative-associative fold.

Function mapafold(b,L,f), where b is an associative operator, L is a
non-empty list of items and f is a mapping function over items L, has
the following meaning:

[{orc runnable='false'

mapafold(b,L,f) = afold(b,f(L))
}]

However, we can compute the fold and map operations in parallel,
unlike the sequencing implied by the given definition.

The strategy is to build a sequence of buffers containing the values
F(L) in order.  Recursively, we fold the left half and the right of
the sequence in parallel, and then combine them. Observe that in order
to fold a single value f(x), it takes around log n sequential steps,
where n is the length of L. 

[{orc runnable='false'

def mapafold(_,[x],f) = f(x)
def mapafold(b,L,f) =

 val n = length(L)
 val c = fillArray(Array(n), lambda (_) = Buffer())

 ----- fill(i,L), for non-empty list L, puts f(L_j) in channel i+j---

 def fill(i,[])   = stop
 def fill(i,x:xs) = c(i)?.put(f(x)) >> stop | fill(i+1,xs)

 ------------------------End of fill(i,L)------------------------------

 ------ aggr(i,k) folds k items from the channels starting at i -------

 {- i >= 0 and k >= 1, k denotes the number of values to be folded,
    the goal is to compute:
      c(i).get(), if k = 1
      b(c(i).get(), c(i+1).get(), ... c(i+k-1).get()), if k > 1
 -}
        def aggr(i,1) = c(i)?.get()

 def aggr(i,k) =
 val r = k/2
 (aggr(i,r), aggr(i+r,k-r)) >(x,y)> b(x,y)

 -------------------------- End of aggr(i,k) --------------------------

 {- Goal of mapafold(b,L,f) -}
 fill(0,L) | aggr(0, n)
}]

!! Map with Commutative and Associative Fold
We use the same technique as in the previous commutative fold over a
list of items. We store f(x), for each x in the list in Buffer c.

[{orc runnable='false'

def mapcfold(b,L,f) =
  val c = Buffer()

  --------- Transfer map of the argument list to Buffer c ---
  -- The order in c is arbitrary
  def xfer([])    = stop
  def xfer(x:xs)  = c.put(f(x)) >> stop | xfer(xs)
  ------------------------ End of xfer ----------------------------

  ---------- combine(n) computes fold of n items from Buffer c ----
  def combine(1) =  c.get()
  def combine(n) =  c.get() >x> c.get() >y> ( c.put(b(x,y)) >> stop | combine(n-1))
  --------------------- End of combine(n) --------------------------

  {- Goal of mapcfold(b,L,f) -}
  xfer(L) | combine(length(L))
}]