Purely functional parallelism in Scala by Wiem Zine El Abidine

Example:

Assuming we have many Tasks (extends 'Runnable') and we create a thread pool with 2 workers.

val service: ExecutorService = Executors.newFixedThreadPool(2)
service.submit(t1)
service.submit(t2)
service.submit(t3)

Note: we can assume that our tasks will be executed in logical threads that will run concurrently with main execution thread of the program. You can see what I mean in this picture: the 2 painters are the threads (workers) they will execute the tasks, t1 and t3 are executed by the same thread but separately and t2 is executed by the other one.

There are many types of executor service and every type has its use case. (FixedThreadPool, CachedThreadPool, ScheduledThreadPool, SingleThreadPool…)

//TODO: we need to generalize our design to be useful for every type of executor Service.

2. 'Runnable' doesn’t return a meaningful value because 'run' returns 'Unit'. We cannot test the result of our parallel computations, we cannot control 'Runnable' because it’s hard to know about its internal behavior, so let’s think about a better approach to get the result of the computation after running the task.

We’re going to have an 'ExecutorService' and we will 'submit' our tasks to it, submit can take 'Runnable' or 'Callable[A]'

We can define our tasks as a 'Callable[A]' specifying the type of our computation result, when we 'submit' our tasks to the 'ExecutorService' they will be evaluated in a separate logical thread asynchronously because each of them returns a java 'Future[A]'

trait Callable[A] {
  def call: A
}
class ExecutorService {
  def submit[A](a: Callable[A]): Future[A]
}

2. Data type and functions

The data structure should be a container of a value that will be executed in separate thread and it requires an 'ExecutorService' to run the result and return it in the future, let’s call it 'Par':

type Par[A] = ExecutorService => Future[A]

In order to interpret 'Par' we need to define a method that requires an 'ExecutorService' and returns 'Future[A]'

def run[A](es: ExecutorService)(par: Par[A]): Future[A] = par(es)

Let’s start with a simple implementation that wraps a constant value, a function that creates a 'Par' of a single value that is always done and can’t be cancelled:

def unit[A](a: A): Par[A] = Par(_ => UnitFuture(a))
private case class UnitFuture[A](get: A) extends Future[A] {
  def cancel(x: Boolean): Boolean = false
  def get(v: Long, unit: java.util.concurrent.TimeUnit): A = get
  def isCancelled(): Boolean = false
  def isDone(): Boolean = true
}

Note, we’re using 'Future' which doesn’t have a purely functional interface.

We need to submit a computation that will run it in a logical thread:

def fork(a: => Par[A]): Par[A] = es =>
    es.submit(new Callable[A] {
      def call: A = run(es)(a).get
    })

Here we see call by name: the 'a' will be evaluated only when it will be called (later in 'def call'.)

Let’s implement 'map2' to combine two parallel computations

def map2[A, B, C](a: Par[A], b: Par[B])(f: (A, B) => C): Par[C] =
  (es: ExecutorService) => {
    val af = run(es)(a)
    val bf = run(es)(b)
    UnitFuture(f(af.get, bf.get))
  }

But this doesn’t evaluate the call of 'a' and 'b' in separate logical thread.

So we can fork 'a' and 'b' : 'map2(fork(a), fork(b))'

Let’s take a break and talk about other thing

In the red book, there is a section about The algebra of an API, it’s about how in FP it is important to have proof properties and laws about code. In general when you implement tests for the code, you try to break the expected behavior of the computation to reveal if there are bugs. In FP it’s quite similar but we have to do that during the implementation of the Algebra respecting some rules and strong FP properties. That’s why it’s easy to reason about a pure functional code.

There are two laws for the implementation of 'Par': the law of mapping, and the law of forking.

Let’s talk about that shortly, first of all we need the 'map' function here:

def map[A, B](a: Par[A])(f: A => B): Par[B] =
  map2(a, unit(()))((a, _) => f(a))

• The law of mapping: identity and composition

def equals[A, B](e: ExecutorService)(p1: Par[A], p2: Par[B]): Boolean =
  run(e)(p1).get == run(e)(p2).get

def lazyUnit[A](a: => A): Par[A] = fork(unit(a))

equals(es)(fork(lazyUnit(1)), lazyUnit(1))

sealed trait Future[A] {
  private[parallelism] def apply(k: A => Unit): Unit
}

def unit[A](a: A): Par[A] = (_: ExecutorService) => new Future[A] {
  def apply(cb: A => Unit): Unit = cb(a)
}

def fork[A](a: => Par[A]): Par[A] =
  es =>
    new Future[A] {
      def apply(cb: A => Unit): Unit =
        es.submit(new Callable[Unit] { def call = a(es)(cb) })
  }

def run[A](es: ExecutorService)(p: Par[A]): A = ???

def unsafeRun[A](es: ExecutorService)(p: Par[A]): A = {
  val ref = new AtomicReference[A]
  val latch = new CountDownLatch(1)
  p(es) { a =>
    ref.set(a)
    latch.countDown()
  }
  ref.get
}