Parallel reduce: Hopac, Asyncs, Tasks and Scala's Futures

Tuomas Hietanen posted a parallel reduce function that uses TPL Tasks. I found it interesting to compare performance of this function with analogues implemented using F# Asyncs, Hopac Jobs and Scala Futures.

The author uses noop long-running reduce function to show that it's really run in parallel. In this blog post we are benchmarking another aspect of the implementations: how much extra cost is introduced by a particular parallezation mechanism (library) itself.

We translate the original code almost as is to Tasks and Hopac:

	open Hopac
	open Hopac.Job.Infixes
	open System.Threading.Tasks
	open System
	let reduceParallelTasks<'a> f (ie: 'a array) =
	let rec reduceRec f (ie: 'a array) =
	match ie.Length with
	| 1 -> ie.[0]
	| 2 -> f ie.[0] ie.[1]
	| len ->
	let h = len / 2
	let o1 = Task.Run(fun _ -> reduceRec f ie.[0..h - 1])
	let o2 = Task.Run(fun _ -> reduceRec f ie.[h..])
	f o1.Result o2.Result
	match ie.Length with
	| 0 -> failwith "Sequence contains no elements"
	| _ -> reduceRec f ie
	let reduceParallelAsync<'a> f (ie: 'a array) =
	let rec reduceRec f (ie: 'a array) =
	async {
	match ie.Length with
	| 1 -> return ie.[0]
	| 2 -> return f ie.[0] ie.[1]
	| len ->
	let h = len / 2
	let! o1a = Async.StartChild <| reduceRec f ie.[0..h - 1]
	let! o2 = reduceRec f ie.[h..]
	let! o1 = o1a
	return f o1 o2
	}
	match ie.Length with
	| 0 -> failwith "Sequence contains no elements"
	| _ -> Async.RunSynchronously <| reduceRec f ie
	let reduceParallelHopac<'a> f (a: 'a array) =
	let rec reduceRec (f, ie: 'a array) =
	match ie.Length with
	| 1 -> Job.result ie.[0]
	| 2 -> Job.result (f ie.[0] ie.[1])
	| len ->
	let h = len / 2
	reduceRec (f, ie.[0..h - 1]) <*> Job.delayWith reduceRec (f, ie.[h..]) |>> fun (x, y) -> f x y
	match a.Length with
	| 0 -> failwith "Sequence contains no elements"
	| _ -> run <| reduceRec (f, a)
	let cleanup() =
	for _ in 1..5 do
	GC.Collect ()
	Threading.Thread.Sleep 50
	let a = [| 1L..5000000L |]
	Array.reduce (+) a
	cleanup()
	reduceParallelTasks (+) a
	cleanup()
	reduceParallelHopac (+) a
	cleanup()
	reduceParallelAsync (+) a

view raw parallel_reduce.fs hosted with ❤ by GitHub

And Scala's Futures:

	import concurrent.duration.Duration
	import concurrent.{Await, Future}
	import concurrent.ExecutionContext.Implicits.global
	object ParallelReduce {
	def reduce [A](f: (A, A) => A, a: Array[A]) : A = {
	def reduceRec(a: Array[A]) : Future[A] = {
	a.length match {
	case 1 => Future(a(0))
	case 2 => Future {
	f(a(0), a(1))
	}
	case len => {
	val h = len / 2
	val o1a = reduceRec(a.slice(0, h))
	for {
	o2 <- reduceRec(a.slice(h, a.length))
	o1 <- o1a
	} yield f(o1, o2)
	}
	}
	}
	a match {
	case Array() => throw new Exception("Sequence contains no elements")
	case _ => Await.result(reduceRec(a), Duration.Inf)
	}
	}
	def run() = {
	val a = Array.range(1, 5000000)
	Util.time { reduce((x: Int, y: Int) => x + y, a) }
	}
	}

view raw parallel_reduce.scala hosted with ❤ by GitHub

The results (Core i5, 4 cores):

• Sequential List.reduce: Real: 00:00:00.014, CPU: 00:00:00.015, GC gen0: 0, gen1: 0, gen2: 0 
• Tasks: Real: 00:00:01.790, CPU: 00:00:05.678, GC gen0: 36, gen1: 10, gen2: 1 
• Hopac: Real: 00:00:00.514, CPU: 00:00:01.482, GC gen0: 27, gen1: 2, gen2: 1 
• Asyncs: Real: 00:00:37.872, CPU: 00:01:48.405, GC gen0: 90, gen1: 29, gen2: 4
• Scala Futures: 4.8 seconds

(Hopac - 3.4 times faster, Asyncs - 21.1 times slower, Scala - 1.8 times slower)

Hopac is ~3.5 times faster than TPL. What's wrong with Asyncs? I don't know. Maybe they are not intended for highly concurrent scenarios. Or my code may not be the most efficient. Any ideas, guys?

Let's test the leaders on larger arrays:

(Hopac is 3.37 times faster, Scala is 1.5 times slower)

(Hopac is 5.25 times faster, Scala is 1.05 times slower)