These are some notes for the 3rd course of Scala Specialization on Cousera.
- 1. Week 1 Basics
- 2. Week 2 Task-Parallelism
- 3. Week 3 Data-Parallelism
- 4. Week 4 Data Structures
We assume:
- Our parallel programming model applied for multicore or multiprocessor systems with shared memory.
- Operating system and the JVM as the underlying runtim environments.
Operating System : software that manages hardware and software resources, and shedules program executions.
Process : an instance of a program that is executing in the OS.
The OS multiplexes many different processes and a limited number of CPUs, so that they get time slices of execution. This mechanism is called multitasking.
Two different processes cannot access each other's memorty directly, i.e., they are isolated.
Thread : Each process can contain multiple independent concurrency units called threads.
Threads can be started from within the same program, and they share the same memory address space.
Each thread has a program counter and a program stack.
Each JVM process starts with a main thread.
To start additional threads:
- Define a Thread subclass.
- Instantiate a new Thread object.
- Call
startmethod on the Thread object.
The Thread subclass defines the code that the thread will excute. The same custom Thread subclass can be used to start multiple threads.
Run in sbt scala console with paste mode.
scala> :paste
// Entering paste mode (ctrl-D to finish)
class HelloThread extends Thread {
override def run() {
println("Hello world!")
}
}
val t = new HelloThread
t.start()
t.join()Hello world!
defined class HelloThread
t: HelloThread = Thread[Thread-8,5,]
Try 2 threads at the same time.
class HelloThread extends Thread {
override def run() {
println("Hello ")
println("world!")
}
}
def main() {
val t = new HelloThread
val s = new HelloThread
t.start()
s.start()
t.join()
s.join()
}scala> main()
Hello
Hello
world!
world!
scala> main()
Hello
world!
Hello
world!
scala> main()
Hello
world!
Hello
world!
The previous demo showed that separate statements in two threads can overlap.
In some cases, we want to ensure that a sequence of statements in a specific thread executes at once.
An operation is atomic if it appears as if it occurred instantaneously from the point of view of other threads.
A demo:
private var uidCount = 0L
def getUniqueId(): Long = {
uidCount = uidCount + 1
uidCount
}
def startThread() = {
val t = new Thread {
override def run() {
val uids = for (i <- 0 until 10) yield getUniqueId()
println(uids)
}
}
t.start()
t
}
startThread
startThreadgetUniqueId: ()Long
startThread: ()Thread
res0: Thread = Thread[Thread-53,5,run-main-group-2]
scala> Vector(1, 2, 3, 4, 5, 6, 7, 8, 10, 12)
Vector(1, 7, 9, 11, 13, 14, 15, 16, 17, 18)
The synchronized block is used to achieve atomicity. Code block after a synchronized call on an object x is never executed by two threads at the same time.
//The synchronized method must be invoded on an instance of some object.
private val x = new AnyRef {}
private var uidCount = 0L
def getUniqueId(): Long = x.synchronized {
uidCount = uidCount + 1
uidCount
}
def startThread() = {
val t = new Thread {
override def run() {
val uids = for (i <- 0 until 10) yield getUniqueId()
println(uids)
}
}
t.start()
t
}
startThread
startThreadVector(2, 3, 4, 5, 6, 7, 8, 9, 10, 11)
Vector(1, 12, 13, 14, 15, 16, 17, 18, 19, 20)
getUniqueId: ()Long
startThread: ()Thread
res0: Thread = Thread[Thread-3,5,]
Different threads use the synchronized block to agree on unique values. The synchronized block is an example of a synchronization primitive.
Invocations of the synchronized block can nest.
class Account(private var amount: Int = 0) {
def transfer(target: Account, n: Int) =
this.synchronized {
target.synchronized {
this.amount -= n
target.amount += n
}
}
}
def startThread(a: Account, b: Account, n: Int) = {
val t = new Thread {
override def run(): Unit = {
for (i <- 0 until n) {
a.transfer(b, 1)
}
}
}
t.start()
t
}
val a1 = new Account(500000)
val a2 = new Account(700000)
val t = startThread(a1, a2, 150000)
val s = startThread(a2, a1, 150000)
t.join()
s.join()Deadlock is a scenario in which two or more threads compete for resources(such as monitor ownership), and wait for each to finish without releasing the already acquired resources.
One approach to resolve deadlocks is to always acquire resources in the same order. This assume an ordering relationship on the resources.
private val x = new AnyRef {}
private var uidCount = 0L
def getUniqueId(): Long = {
uidCount = uidCount + 1
uidCount
}
class Account(private var amount: Int = 0) {
val uid = getUniqueId()
private def lockAndTransfer(target: Account, n: Int) =
this.synchronized {
target.synchronized {
this.amount -= n
target.amount += n
}
}
def transfer(target: Account, n: Int) =
if (this.uid < target.uid) this.lockAndTransfer(target, n)
else target.lockAndTransfer(this, n)
}
def startThread(a: Account, b: Account, n: Int) = {
val t = new Thread {
override def run(): Unit = {
for (i <- 0 until n) {
a.transfer(b, 1)
}
}
}
t.start()
t
}
val a1 = new Account(500000)
val a2 = new Account(700000)
val t = startThread(a1, a2, 150000)
val s = startThread(a2, a1, 150000)
t.join()
s.join()Memory model is a set of rules that describes how threads interact when accessing shared memory.
Java Memory Model - the memory model for the JVM
Following is two rules of JVM(a subset of the whole rules) which should be remembered in this course:
- Two threads writing to separate locations in memory do not need synchronization.
- A thread X that calls
joinmethod on another thread Y is guaranteed to observe all the writes by thread Y afterjoinreturns.
The parallelism constructs in the remainder of the course are implemented in terms of:
- threads
- synchronization primitives such as
synchronized
Given expressions e1 and e2 , compute them in parallel and return the pair of results
parallel(e1, e2)Parallelism could be done by a recursive algorithm.
def parallel[A, B](taskA: =>A, taskB: =>B): (A, B) = {...}- returns the same value as given
- benefit:
parallel(a, b)can be faster than(a, b) - it takes its arguments as
by name, indicated with=> Aand=> B
For parallelism, need to pass unevaluated computations(call by name).
Memory is bottleneck. Multi-processors share the memory space of RAM. The computation time can not be less that the time it takes to fetch the data from memory to processors.
val(v1, v2) = parallel(e1, e2)The minimum time that this parallel expression time is the maximum of the running times of e1 and e2.
val(v1, v2) = parallel(e1, e2)alternatively using task construct:
val t1 = task(e1)
val t2 = task(e2)
val v1 = t1.join
val v2 = t2.joint = task e starts computation e "in the background"
tis atask, which performs computation ofe- current computation proceeds in parallel with
t - to obtain the result of
e, uset.join t.joinblocks and waits until the results is computed- subsequent
t.joincalls quickly return the same result
Here is a minimal interface for tasks:
def task(c: => A): Task[A]
trait Task[A] {
def join: A
}task and join establish maps between computations and tasks
In terms of the value computed the equation task(e).join==e holds. We can omit writing .join if we also define an implicit conversion:
implicit def getJoin[T](x: Task[T]): T = x.joinPerformance : a key motivation for paralellism
How to estimate it?
- empirical measurement
- asymptotic analysis
Asymptotic analysis is important to understand how algorithms scale when:
- inputs get larger
- we have more hardware parallelism available
We examine worst-cast(as opposed to average) bounds.
We would like to speak about the asymptotic complexity of parallel code
- but this depends on available parallel resources
- we introduce two measures for a program
Work W(e) : number of steps e would take if there was no parallelism
- this is simply the sequential execution time
- treat all
parallel(e1, e2)as(e1, e2)
Depth
- we take maximum of running times for arguments of parallel
Key rules are:
$W(parallel(e1, e2)) = W(e1) + W(e2) + c2$ $D(parallel(e1, e2)) = max(D(e1), D(e2)) + c1$
If we divide work in equal parts, for depth it counts only once!
For parts of code where we do not use parallel explicityly, we must add up costs. For function call or operation
$W(f(e1, ..., en)) = W(e1) + ... + W(en) + W(f)(v1, ..., vn)$ $D(f(e1, ..., en)) = D(e1) + ... + D(en) + D(f)(v1, ..., vn)$
Note: we assume(reasonably) that constants are such that
Suppose we know W(e) and D(e) and our platfrom has P parallel threads.
It is reasonalbe to use this estimate for running time:
D(e) + W(e) / P
Suppose that we have two parts of a sequential computaion:
- part1 takes fraction
fof the computation time, e.g., 40%. - part2 takes the remaining
1 - ffraction of time, e.g., 60%, and we can speed it up.
If we make part2 P times faster the speedup is
1 / (f + (1 - f) / P)
For P = 100 and f = 0.4 we obtain 2.46. Even if we speed the second part infinitely, we can obtain at most 1 / 0.4 = 2.5 speed up.
Testing : ensures that parts of the program are behaving according to the intended behavior.
Benchmarking : computes performance metrics for parts of the program.
Typically, testing yields a binary output - a program or its part is either correct or not. Benchmarking usually yields a continous value, which denotes the extent to which the program is correct.
- Performance benefits are the main reason why we are writing parallel programs in the first place.
- Benchmarking parallel programs is even more important than benchmarking sequential programs.
Performance(specifically, running time) is subject to many factors:
- processor speed
- number of proce ssors
- memory access latency and throughput(affectc contention)
- cache behavior(e.g. false sharing, associativity effects)
- running behavior(e.g. garbage collection, JIT complication, thread scheduling)
See What Every Programmer Should Know About Memory, by Ulrich Drepper.
Measuring performance is difficult - usually, the a performance metric is a random variable.
- multiple repetions
- statistical treatment: mean and variance
- eliminating outliers
- ensuring steady state(warm-up)
- preventing anomalies(GC, JIT complication, aggressive optimizations)
See Statistically Rigorous Java Performance Evaluation, by Georges, Buytaert and Eechhout.
ScalaMeter is a benchmarking and performance regression testing framework of JVM.
- performance regression testing: comparing performance of the current program run against known previous runs
- benchmarking: measuring performance of the current(part of the) program
Add ScalaMeter as a dependency:
libraryDependencies += "com.storm-enroute" %% "scalameter-core" % "0.6"
Import the contents of the ScalaMeter package, and measure:
import org.ScalaMeter._
val time = measure {
(0 until 1000000).toArray
}
println(s"Array initialization time: $time ms")When a JVM programs starts, it undergoes a period of warmup, after which it achieves its maximum performance.
- first, the program is interpreted
- then, parts of the program are compiled into machine code
- later, the JVM may choose to apply additional dynamic optimizations
- eventually, the program reaches steady state
Usually, we want to measure steady state program performance.
ScalaMeter Warmer objects run the benchmarked code until detecting steady state.
import org.scalameter._
val time = withWarmer(new Warmer.Default) measure {
(0 until 1000000).toArray
}Measure.Default: plain running timeIgnoringGC: running time without GC pausesOutlierElimination: remove statistical outliersMemoryFootprint: memory footprint of an objectGarbageCollectionCycles: total number of GC pauses- newer ScalaMeter version can also measure method invocation counts and boxing counts
import org.scalameter._
val time = withMeasurer(new Measurer.MemoryFootprint) measure {
(0 until 1000000).toArray
}Sort in parallel
- recursivley sort the tow halves of the array in parallel
- sequentially merge the two array halves by copying into a temporary array
- copy the demporary array backi into the original array
def parMergeSort(xs: Array[Int], maxDepth: Int): Unit = {
// At each level of the merge sort, we will alternate between the source array xs and the intermediate array ys
val ys = new Array[Int](xs.length)
def sort(from: Int, until: Int, depth: Int): Unit = {
if (depth == maxDepth) {
// sequential sort
quickSort(xs, from, until - from)
} else {
val mid = (from + until) / 2
parallel(sort(mid, until, depth + 1), sort(from, mid, depth + 1))
val flip = (maxDepth - depth) % 2 == 0
val src = if (flip) ys else xs
val dst = if (flip) xs else ys
merge(src, dst, from, mid, until)
}
}
def merge(src: Array[Int], dst: Array[Int], from: Int, mid: Int, until: Int): Unit ={
// Given an array src consisting of two sorted intervals, merge those interval into the dst array.
// The merge implementation is sequential, so it will note be gone through here.
// Exerise: How would you implement merge function in parallel?
}
sort(0, xs.length, 0)
}def copy(src: Array[Int], target: Array[Int], from: Int, until: Int, depth: Int): Unit = {
if (depth == maxDepth) {
Array.copy(src, from, targt, from, until - from)
} else {
val mid = (from + until) / 2
val right = parallel(
copy(src, target, mid, until, depth + 1)
copy(src, target, from, mid, depth + 1)
)
}
}
if (maxDepth % 2 == 0) copy(ys, xs, 0, xs.length, 0)Parallel processing of collections is important. It is one of the main applications of parallelsim today.
We examine conditions when this can be done:
- properties of collections: ablility to split, combine
- properties of operations: associativity, independence
Opertations on collections are key to functional programming
map: apply function to each element
List(1,3,8).map(x => x*x) == List(1,9,64)
fold: combine elements with a given collection
List(1,3,8).fold(100)((s,x) => s + x) == 112
scan: combine folds of all list prefixes
List(1,3,8).scan(100)((s,x) => s + x) == List(100, 101, 104, 112)
We use List to speicfy the results of operations. Lists are not good for parallel implementations because we cannot efficiently
- split them in half (need to search for the middle)
- combine them (concatenation needs linear time)
We use for now these alternatives:
arrays: imperative (recall array sum)trees: can be implemented functionally
Map applies a given function to each list element
List(a1, a2, ..., a3).map(f) == List(f(a1), f(a2), ..., f(an))
Properties to keep in mind:
list.map(x => x) == listlist.map(f.compose(g)) = list.map(g).map(f)
Recall (f.compose(g))(x) == f(g(x))
// sequential definition
def mapSeq[A, B](lst: List[A], f: A => B): List[B] = lst match {
case Nil => Nil
case h :: t => f(h) :: mapSwq(t, f)
}We would like a version that parallelizes
- computations of
f(h)for different elementsh - finding the elements themselves (list is not a good choice)
def mapASegSeq[A,B](inp: Array[A], left: Int, right: Int, f: A => B, out: Array[B]): Unit = {
// Writes to out(i) for left <= i <= right-1
var i = left
while (i < right) {
out(i) = f(inp(i))
i = i+1
}
}def mapASegPar[A,B](inp: Array[A], left: Int, right: Int, f: A => B, out: Array[B]): Unit = {
// Writes to out(i) for left <= i <= right-1
if (right - left < threshold)
mapASegSeq(inp, left, right, f, out)
else {
val mid = left + (right - left) / 2
parallel(
mapASegPar(inp, left, mid, f, out),
mapASegPar(inp, mid, right, f, out)
)
}
}- writes need to be disjoint (otherwise: non-deterministic behavior)
- threshold needs to be large enough (otherwise we lose efficiency)
Consider trees where
- leaves store array segments
- non-leaf node stores number of elements
We assume that our tress are balanced: we can explore branches in parallel
sealed abstract class Tree[A] { val size: Int }
case class Leaf[A](a: Array[A]) extends Tree[A] {
override val size = a.size
}
case class Node[A](l: Tree[A], r: Tree[A]) extends Tree[A] {
override val size = l.size + r.size
}
def mapTreePar[A:Manifest,B:Manifest](t: Tree[A], f: A => B): Tree[B] = t match {
case Leaf(a) => {
val len = a.length; val b = new Array[B](len)
var i = 0
while (i < len) { b(i) = f(a(i)); i = i + 1 }
Leaf(b)
}
case Node(l, r) => {
val (lb,rb) = parallel(mapTreePar(l,f), mapTreePar(r,f))
Node(lb,rb)
}
}Arrays
- (+) random access to elements, on shared memory can share array
- (+) good memory locality
- (-) imperative: must ensure parallel tasks write to disjoint parts
- (-) expensive to concatenate
Immutable trees:
- (+) purely functional, produce new trees, keep old ones
- (+) no need to worry about disjointness of writes by parallel tasks
- (+) efficient to combine two trees
- (-) high memory allocation overhead
- (-) bad locality
Fold combines elements with a given operation
List(1,3,8).fold(100)((s,x) => s + x) == 112
Fold takes among others a binary operation, but variants differ:
- whether they take an initial element or assume non-empty list
- in which order they combine operations of collection
List(1,3,8).foldLeft(100)((s,x) => s - x) == ((100 - 1) - 3) - 8 == 88
List(1,3,8).foldRight(100)((s,x) => s - x) == 1 - (3 - (8 - 100)) == -94
List(1,3,8).reduceLeft((s,x) => s - x) == (1 - 3) - 8 == -10
List(1,3,8).reduceRight(100)((s,x) => s - x) == 1 - (3 - 8) == 6
To enable parallel operations, we look at associative operations
- addiction, string concatenation (but not minus)
Operation f: (A,A) => A is associative iff for every
If we write
Consequence: consider two expressions with same list of operands connected with
Each expression built from values connected with
- leaves are the values
- nodes are
$\otimes$
Result of evaluating the expression is given by a reduce of this tree.
Sequential Version:
// sequential definition
// reduce can be thought of as relacing the constructor Node with given f
def reduce[A](t: Tree[A], f: (A,A) => A): A = t match {
case Leaf(v) => v
case Node(l, r) => f(reduce[A](l, f), reduce[A](r, f)) // Node -> f
}Parallel Version:
// parallel definition
def reduce[A](t: Tree[A], f: (A,A) => A): A = t match {
case Leaf(v) => v
case Node(l, r) => {
val (lV, rV) = parallel(reduce[A](l, f), reduce[A](r, f))
f(lV, rV)
}
}If f denotes
reduce(Node(Leaf(x), Node(Leaf(y), Leaf(z))), f) ==
reduce(Node(Node(Leaf(x), Leaf(y)), Leaf(z)), f)Observe: can use a list to describe the ordering of elements of a tree
def toList[A](t: Tree[A]): List[A] = t match {
case Leaf(v) => List(v)
case Node(l, r) => toList[A](l) ++ toList[A](r)
}Suppose we also have tree map:
def map[A,B](t: Tree[A], f: A => B): Tree[B] = t match {
case Leaf(v) => Leaf(f(v))
case Node(l, r) => Node(map[A,B](l, f), map[A,B](r, f))
}Express toList using map and reduce:
toList(t) == reduce(map(t, List(_)), _ ++ _)
Express associativety results consistency up to element order in Scala:
Consequence of associativity(Scala): if f: (A,A) => A is associative, t1: Tree[A] and t2: Tree[B] satisfies toList(t1) == toList(t2), then: reduce(t1, f) == reduce(t2, f).
Often we work with collections where we only know the ordering and not the tree structure, e.g., arrays. We can convert it into a balanced tree then do tree reduction.
Because of associativety, we can choose any tree that preserves the order of elements of the original collection.
Tree reduction replaces Node constructor with f, so we can just use f directly instead of building tree nodes.
When the segment is small, it is faster to process it sequentially.
// Parallel array reduce
// Implementation that computes fold of a given array segment with a given associative operation f
def reduceSeg[A](inp: Array[A], left: Int, right: Int, f: (A,A) => A): A = {
if (right - left < threshold) {
var res = inp(left); var i = left + 1
while (i < right) { res = f(res, inp(i)); i = i + 1 }
res
} else {
mid = (left + right) / 2
val (a1, a2) = parallel(
reduceSeg(inp, left, mid, f),
reduceSeg(inp, mid, rigt, f)
)
f(a1, a2)
}
}
def reduce[A](inp: Array[A], f: (A,A) => A): A =
reduceSeg(inp, 0, inp.length, f)Operation f: (A,A) => A is associative iff for every
Operation f: (A,A) => A is commutative iff for every
For correctness of reduce, we need (just) associativity.
reduce(map(a, power(abs(_), p)), _ + _).
map can be used together with reduce to avoid intermediate collections.
Addition is commutative but not associative
scala> val e = 1e-200
e: Double = 1.0E-200
scala> val x = 1e200
x: Double = 1.0E200
scala> val mx = -x
mx: Double = -1.0E200
scala> (x + mx) + e
res0: Double = 1.0E-200
scala> x + (mx + e)
res1: Double = 0.0
scala> (x + mx) + e = x + (mx + e)
res2: Boolean = falseMultiplication is commutative but not associative
scala> (e * x) * x
res3: Double = 1.0E200
scala> e * (x * x)
res4: Double = Infinity
scala> (e * x) * x == e * (x * x)
res5: Boolean = falseSuppose f1: (A1,A1) => A1 and f2, (A2,A2) => A2 are associative.
Then f: ((A1,A2),(A1,A2)) => (A1,A2) defined by
f((x1,x2),(y1,y2)) = (f1(x1,y1),f2(x2,y2))
is also associative.
It's similar to construct associative operations on for n-tuples.
Given a collction of integers, compute the average.
// Two reduction solution.
val sum = reduce(collection, _ + _)
val length = reduce(map(collection, (x: Int) => 1), _ + _)
sum / length
// single reduction solution.
// f((sum1, length1), (sum2, length2)) = (sum1 + sum2, length1 + length2)
val (sum, length) = reduce(map(collection, (x: Int) => (x, 1), f)
sum / lengthIf f satisfies f(f(x,y),z) = f(f(y,z),x)(symmetry) and f(x,y) = f(y,x)(commutativity) for every x, y, z, we have f(f(x,y),z) = f(x, f((y,z)))(associativity).
scanLeft: list of the folds of all list prefixes
List(1,3,8).scanLeft(100)((s,x) => s + x) == List(100, 101, 104, 112)
scanRight is different from scanLeft
List(1,3,8).scanRight(100)((s,x) => s + x) == List(112, 104, 101, 100)
where
// Sequential scanLeft.
// Assume out.length >= inp.length + 1.
def scanLeft[A](inp: Array[A], a0: A, f: (A,A) => A, out: Array[A]): Unit = {
out(0) = a0
var a = a0
var i = 0
while (i < inp.length) {
a = f(a,inp(i))
i = i + 1
out(i) = a
}
}Assume input is given in array inp and that you have reduceSeg1 and mapSeg functions on array segments:
def reduceSeg1[A](inp: Array[A], left: Int, right: Int, a0: A, f: (A,A) => A): A
def mapSeg[A,B](inp: Array[A], left: Int, right: Int, fi: (Int, A) => B, out: Array[B]): Unit
def scanLeft[A](inp: Array[A], a0: A, f: (A,A) => A, out: Array[A]): Unit = {
val fi = { (i: Int, v: A) => reduceSeg1(inp, 0, i, a0, f) }
mapSe�g(inp, 0, inp.length, fi, out)
val last = inp.length - 1
out(last + 1) = f(out(last), inp(last))
}Reuse computations by saving the intermediate results in a tree. We assume that input collection is also (another) tree.
// Trees storing the input collection only have values in leaves
sealed abstract class Tree[A]
case class Leaf[A](a: A) extends Tree[A]
case class Node[A](l: Tree[A], r: Tree[A]) extends Tree[A]
// Trees storing intermediate values have (res) value in nodes
sealed abstract class TreeRes[A] { val res: A }
case class LeafRes[A](override val res: A) extends TreeRes[A]
case class NodeRes[A[(l: TreeRes[A], override val res: A, r: TreeRes[A])
def reduceRes[A](t: Tree[A], f: (A,A) => A): TreeRes[A] = t match {
case Leaf(v): LeafRes(v)
case Node(l, r): {
val (tL, tR) = (reduceRes(l, f), reduceRes(r, f))
val res = f(tL.res, tR.res)
NodeRes(lt, res, rt)
}
}
// Parallel version for reduceRes
def upsweep[A](t: Tree[A], f: (A,A) => A): TreeRes[A] = t match {
case Leaf(v) => LeafRes(v)
case Node(l, r) => {
val (tL, tR) = parallel(upsweep(l, f), upsweep(r, f))
val res = f(tL.res, tR.res)
NodeRes(lt, res, rt)
}
}// a0 is reduce of all elements left of the tree
def downsweep[A](t: TreeRes[A], a0: A, f: (A, A) => A): Tree[A] = t match {
case LeafRes(a) => Leaf(f(a0, a))
case NodeRes(l, _, r) => {
val (tL, tR) = parallel(
downsweep(l, a0, f),
downsweep(r, f(a0, l.res), f)
)
Node(tL, tR)
}
}def scanLeft[A](t: Tree[A], a0: A, f: (A,A) => A): Tree[A] = {
val tRes = upsweep(t, f)
val scan1 = downsweep(tRes, a0, f)
prepend(a0, scan1)
}
def prepend[A](x: A, t: Tree[A]): Tree[A] = t match {
case Leaf(v) => Node(Leaf(x), Leaf(v))
case Node(l, r) => Node(prepend(x, l), r)
}The only difference compared to previouse TreeRes: each Leaf now keeps track of the array segment range (from, to) from which res is computed.
// Intermediate tree for array reduce
// We do not keep track of the array elements in the Leaf itself.
// We instead pass around a reference to the input array.
sealed abstract class TreeResA[A] { val res: A }
case class Leaf[A](from: Int, to: Int, override val res: A) extends TreeResA[A]
case class Node[A](l: TreeResA[A], override val res: A, r: TreeResA[A]) extends TreeResA[A]
def upsweep[A](inp: Array[A], from: Int, to: Int, f: (A,A) => A): TreeResA[A] = {
if (to - from < threshold)
Leaf(from, to, reduceSeg1(inp, from + 1, to, inp(from), f))
else {
val mid = from + (to - from) / 2
val (tL, tR) = parallel(
upsweep(inp, from, mid, f),
upsweep(inp, mid, to, f)
)
Node(tL, f(tL.res, rR.res), tR)
}
}
def downsweep[A](inp: Array[A], a0: A, f: (A,A) => A, t: TreeResA[A], out: Array[A]): Tree[A] = t macth {
case Leaf(from, to, res) => scanLeftSeg(inp, from, to, a0, f, out)
case Node(l, _, r) => {
val (_,_) = parallel(
downsweep(inp, a0, f, l, out),
downsweep(inp, f(a0, l.res), f, r, out)
)
}
}
def scanLeft[A](inp: Array[A], a0: A, f: (A,A) => A, out: Array[A]) = {
val t = upsweep(inp, 0, inp.length, f)
downsweep(inp, a0, f, t, out) // fills out [1..inp.length]
out(0) = a0
}The simplest form of data-parallel programming is the parallel for loop.
def initializeArray(xs: Array[Int])(v: Int): Unit {
for (i <- (0 until xs.length).par) {
xs(i) = v
}
}Although simple, parallel for loop allows writing interesting programs.
Render a set of complex numbers in the plane for which sequence
We approximate the definition of the Mandelbrot set - as long as the absolute value of maxIterations
private def computePixel(xc: Double, yc: Double, maxIterations: Int): Int = {
var i = 0
var x, y = 0.0
while (x * x + y * y < 4 && i < maxIterations) {
val xt = x * x - y * y + xc
val yt = 2 * x * y + yc
x = xt;
y = yt;
i += 1
}
color(i)
}
// render the set using data-parallel programming
def parRender(): Unit = {
for (idx <- (0 until image.length).par) {
val (xc, yc) = coordinatesFor(idx)
image(idx) = computePixel(xc, yc, maxIterations)
}
}Different data-parallel programs have different workloads.
workload is a function that maps each input element to the amount of work required to process it.
Uniform workload is defined by a constant function:
Irregular workload is defined by an arbitrary function:
The workload depends on the problem instance.
Goal of the data-parallel scheduler: efficiently balance the workload across processors without any knowledge about the
In Scala, most collection operations can become data-parallel. The .par call converts a sequential collection to a parallel collection.
(1 until 1000).par
.filter(n => n % 3 == 0)
.count(n => n.toString == n.toString.reverse)// implement the method using the foldLeft method
def sum(xs: Array[Int]): Int = xs.par.foldLeft(0)(_ + _)This implementation does not execute in parallel.
Notice the foldLeft signature, we will find that to compute the final result, the intermediate functions must be invoked sequentially, one after another.
def foldLeft[B](z: B)(f: (B, A) => B): BOperations foldRight, reduceLeft, reduceRight, scanLeftand scanRight similarly must process the elements sequentially.
fold signature:
def fold(z: A)(f: (A, A) => A): AThe accumulation will have the same type as the elements in the collection instead of having another type parameter for the accumulation type in the above foldLeft signature.
The fold operation can process the elements in a reduction tree, so it can execute in parallel.
def sum = {
xs.par.fold(0)(_ + _)
}
def max = {
xs.par.fold(Int.MinValue)(math.max)
// xs.par.fold(Int.MinValue)((x, y) => if (x > y) x else y)
}Given a list of "paper", "rock" and "scissors" strings, find out who won:
Array("paper", "rock", "paper", "scissors").par.fold("")(play)
def play(a: String, b: String): String = List(a, b).sorted match {
case List("paper", "scissors") => "scissors"
case List("paper", "rock") => "paper"
case List("rock", "scissors") => "rock"
case List(a, b) if a == b => a
case List("", b) => b
}// The play operator is commutative, but not associative.
play(play("paper", "rock"), play("paper", "scissors")) == "scissors"
play("paper", play("rock", play("paper", "scissors"))) == "paper"In order for the fold operation to work correctly, the following relations must hold:
f(a, f(b, c)) == f(f(a, b), c)
f(z, a) == f(a, z) == a
We say that the neutral element z and the binary operator f must form a monoid.
Commutativity does not matter for fold - the following relation is not necessary:
f(a, b) == f(b, a)
The fold operation can only produce values of the same type as the collection that it is called on.
def aggregate[B](z: B)(f: (B, A) => B, g: (B, B) => B): B
// Count the number of vowels in a character array.
Array('E', 'P', 'F', 'L').par.aggregate(0)(
(count, c) => if (isVowel(c)) count + 1 else count,
_ + _
)Transformer combinations, such as map, filter, flatMap and groupBy, do not return a single value, but instead return new collections as results. They can also execute in parallel.
Traversable[T]: collection of elements with typeT, with operations implemented usingforeachIterable[T]: collection of elements with typeT, with operations implemented usingiteratorSeq[T]: an ordered sequence of elements with typeTSet[T]: a set of elements with typeT(no duplicates)Map[K, T]: a mpa of keys with typeKassociated with values of typeV(no duplicate keys)
Traits ParIterable[T], ParSeq[T], ParSet[T] and ParMap[K, V] are the parallel counterparts of different sequential traits.
For code that is agnostic about parallelism, there exists a separate hierarchy of generic collection traits GenIterable[T], GenSeq[T], GenSet[T] and GenMap[K, V].
Generic collection traits allow us to write code that is unaware of parallelism.
Example - find the largest palindrome in the sequence:
def largestPalindrome(xs: GenSeq[Int]): Int = {
xs.aggregate(Int.MinValue)(
(largest, n) =>
if (n > largest && n.toString == n.toString.reverse) n else largest,
math.max
)
}
val array = (0 until 1000000).toArray
largestPalindrome(array)In Scala, a sequential collection can be converted into a parallel one by calling par.
ParArray[T]: parallel array of objects, counterpart ofArrayandArrayBufferParRange: parallel range of integers, counterpart ofRangeParVector[T]: parallel vector, counter part ofVectorimmutable.ParHashSet[T]: counterpart ofimmutable.HashSetimmutable.ParHashMap[K, V]: counterpart ofimmutable.HashMapmutable.ParHashSet[T]: counterpart ofmutable.HashSetmutable.ParHashMap[K, V]: counterpart ofmutable.HashMapParTrieMap[K, V]: thread-safe parallel map with atomic snapshots, counterpart ofTrieMap- for other collections,
parcreates the closest parallel collection - e.g. aListis converted to aParVector
def intersection(a: GenSet[Int], b: GenSet[Int]): Set[Int] = {
val result = mutable.Set[Int]()
for (x <- a) if (b contains x) result += x
result
}
// correct for sequential computation
intersection((0 until 1000).toSet, (0 until 1000 by 4).toSet)
// incorrect because internal result is mutable.Set, parallel computation will lead problems
intersection((0 until 1000).par.toSet, (0 until 1000 by 4).par.toSet)Avoid mutations to the same memory locations without proper synchronization.
// correct implementation, use a concurrent collection which can be mutated by multiple threads
import java.util.concurrent
def intersection(a: GenSet[Int], b: GenSet[Int]) = {
val result = new ConcurrentSkipListSet[Int]()
for (x <- a) if (b contains x) result += x
result
}
intersection((0 until 1000).toSet, (0 until 1000 by 4).toSet)
intersection((0 until 1000).par.toSet, (0 until 1000 by 4).par.toSet)// correct implementation, avoid side-effects by using the correct combinators
import java.util.concurrent
def intersection(a: GenSet[Int], b: GenSet[Int]): GenSet[Int] = {
if (a.size < b.size) a.filter(b(_))
else b.filter(a(_))
}
intersection((0 until 1000).toSet, (0 until 1000 by 4).toSet)
intersection((0 until 1000).par.toSet, (0 until 1000 by 4).par.toSet)Never modify a parallel collection on which a data-parallel operation is in progress.
- Never write to a collection that is concurrently traversed.
- Never read from a collection that is concurrently modified.
In either case, program non-deterministically prints different results, or crashes.
TrieMap is an exception to above rules.
The snapshot method can be used to efficiently grab the current state:
val graph = concurrent.TrieMap[Int, Int]() ++= (0 until 100000).map(i => (i, i + 1))
graph(graph.size - 1) = 0
val previous = graph.snapshot()
for ((k, v) <- graph.par) graph(k) = previous(v)
val violation = graph.find({ case (i, v} => v != (i + 2) % graph.size })
println(s"violation: $violation")- iterators
- splitters
- builders
- combiners
Each iterable collection can create its own iterator object.
// simplified iterator trait
trait Iterator[A] {
def next(): A
def hasNext: Boolean
}
def iterator: Iterator[A] // on every collectionnextcan be called only ifhasNextreturnstrue- after
hasNextreturnsfalse, it will always returnfalse
// implement foldLeft on an iterator
trait Iterator[T] {
def hasNext: Boolean
def next(): T
def foldLeft[S](z: S)(f: (S, T) => S): S = {
var result = z
while (hasNext) result = f(result, next())
result
}
}Splitter is a counterpart of iterator used for parallel programming.
// simplified splitter trait
trait Splitter[A] extends Iterator[A] {
def split: Seq[Splitter[A]]
def remaining: Int
}
def splitter: Splitter[A] // on every parallel collection- after calling
split, the original splitter is left in an undefined state - the resulting splitters traverse disjoint subsets of the original splitter
- remaining is an estimate on the number of remaining elements
-
splitis an efficient method -$O(log n)$ or better
// implement fold on an splitter
trait Splitter[T] {
def split: Seq[Splitter[T]]
def remaining: Int
def fold(z: T)(f: (T, T) => T): T = {
if (remaining < threshold) foldLeft(z)(f)
else {
val children: Seq[Task[T]] = for (child <- split) yield task { child.fold(z)(f) }
children.map(_.join()).foldLeft(z)(f)
}
}
}Builders are abstractions used for creating new collections.
// simplified Builder trait
trait Builder[A, Repr] {
def +=(elem: A): Builder[A, Repr]
def result: Repr
}
def newBuilder: Builder[A, Repr] // on every collection- calling
resultreturns a collection of typeRepr, containing the elements that were previously added with += - calling
resultleaves theBuilderin an undefined state
trait Traversable[T] {
def foreach(f: T => Unit): Unit
def newBuilder: Builder[T, Traversable[T]]
def filter(p: T => Boolean): Traversable[T] = {
val b = newBuilder
for (x <- this) if (p(x)) b += x
b.result
}
}A combiner is a parallel version of a builder.
trait Combiner[A, Repr] extends Builder[A, Repr] {
def combine(that: Combiner[A, Repr]): Combiner[A, Repr]
}
def newCombiner: Combiner[T, Repr] // on every parallel collection- calling
combinereturns a new combiner that contains elements of input combiners - calling
combineleaves both original Combiners in an undefined state -
combineis an efficient method:$O(log n)$ or better
// Exercise: implement a parallel filter method using splitter and newCombiner