# A Simple Thread Pool in Scala

I wrote the following that spawns n threads, uses them to process a queue of jobs, then returns a result.

As well as any general suggestions, I'd like feedback on the following:

• How safe is SyncQueue? Previously, it went into a deadlock on inexpensive tasks, but I changed the notify to notifyAll, and I haven't had any problems since. I'd still like it looked at though.

• Is there a better way to delay execution of the jobs? I'm using an implicit to make the delay method available; but it would be nice to have it be completely implicit.

• Does the JVM prevent the stdout from interleaving? I remember back from c++ that outputting over different threads at once ending up creating a mess of interleaved text, but this doesn't. For the sole purpose of testing, is outputting text from several threads at once in any way "harmful"?

SyncQueue.scala - A theoretically thread-safe, mutable FIFO queue:

package threadPool

import scala.collection.mutable.Queue

class SyncQueue[A] {
private val q: Queue[A] = new Queue

def nQueued: Int = synchronized {
q.length
}

def available: Boolean = synchronized {
nQueued > 0
}

def pop: A = synchronized {
if (!available) { wait; pop }
else q.dequeue
}

def push(x: A) = synchronized {
q enqueue x; notifyAll
}

def toList: List[A] = synchronized {
q.toList
}

def clearQ = synchronized {
q.clear
}

}


JobQ.scala - A wrapper over 2 SyncQueues that helps with adding jobs/collecting results:

package threadPool

import scala.concurrent._

class JobQ[Result] {
type Job = () => Result
type PossibleResult = Either[Throwable,Result]

private val workQ = new SyncQueue[Job]
private val resQ = new SyncQueue[PossibleResult]

//So it knows when all jobs are finished
private var runningJobs = 0

//Waits until all started jobs are finished
def waitForJobsToFinish(checkDelayMS: Int) =
while(!allJobsFinished)

def allJobsFinished: Boolean = synchronized {
runningJobs == 0
}

def jobsAvailable: Boolean =
workQ.available

def resultsAvailable: Boolean =
resQ.available

def giveJob(job: Job) = synchronized {
workQ push job
runningJobs += 1
}

def giveJobs(jobs: Seq[Job]) =
jobs map (giveJob(_))

def getJob: Job =
workQ.pop

def giveResult(result: PossibleResult) = {
resQ push result
runningJobs -= 1
}

def getResults: List[PossibleResult] = {
val xs = resQ.toList
resQ.clearQ
xs
}

}


Worker.scala - The Runnable used by each thread. It forms an infinite loop of taking a job, processing it, and queuing the result:

package threadPool

class Worker[Result](jobQ: JobQ[Result]) extends Runnable {
def run = while (true) {

val job = jobQ.getJob //blocks until a job is made available

val result: Either[Throwable,Result] =
try {
Right( job() ) //Long computation
} catch { case e: Throwable =>
Left(e)
}

jobQ giveResult result
}

}


Timer.scala - Used to assist the timing in the test:

package threadPool

import java.util.Date

case class Timer(startTime: Long = new Date().getTime) {
private def curMs: Long = new Date().getTime

def restart: Timer = Timer(curMs)
def stop: Long = curMs - startTime
def lap: (Long, Timer) = { val curTime = curMs
(curTime - startTime,Timer(curTime))
}
}

object Timer {
def timeBlock(body: => Unit): Long = {
val t = Timer()
body
t.stop
}
}


package threadPool

import java.lang.Runtime._
import scala.util.Random._

object Implicits {
implicit class delayCall[A](body: => A) {
def delay: (() => A) =
() => body
}
}

//By default, it spawns 1 thread per available processor
def this() = this(Runtime.getRuntime.availableProcessors)

type Job = () => Result
type PossibleResult = Either[Throwable,Result]

val jobQ: JobQ[Result] = new JobQ
}

def start = threads map (_.start)

def giveJob(job: Job) =
jobQ giveJob job

def giveJobs(jobs: Seq[Job]) =
jobs map (giveJob(_))

def getResultsIfDone: Option[List[PossibleResult]] =
if(jobQ.jobsAvailable) None
else Some(jobQ.getResults)

def waitForResults: List[PossibleResult] = {
jobQ waitForJobsToFinish 500
jobQ.getResults
}

}


The Main - Just a sample case:

object ThreadPoolTest extends App {
import Implicits._

val nJobs = 10

//Returns the time taken to execute, to be summed and compared later
def expensiveLong(id: Int): Long = Timer.timeBlock {
val s = scala.util.Random.nextInt(20000)
println(s"Starting expensive task $id:${s / 60000.0} minutes")
println(s"\tEnding $id: Started${s / 60000.0} minutes ago")
}

val jobs: List[() => Long] = (1 to nJobs).toList map { id =>
expensiveLong(id).delay
}

pool.giveJobs(jobs)

pool.start

var rs: List[Either[Throwable,Long]] = Nil

//timeBlock will time all the executions to compare against
val realTime:Long = Timer.timeBlock {
rs = pool.waitForResults
}

//Print results
println(rs)

//For this test, I'm having it crash on an error, because any exceptions would invalidate the results (sum of times taken)
val checkedResults: List[Long] = rs map {
case Left(e)    => throw e
case Right(r)   => r
}

val sum = checkedResults.
foldLeft(0L)(_+_)

println(s"Done:\n\tTotal Time Needed:\t${sum / 60000.0} minutes\n\tTime Spent:\t\t\t${realTime / 60000.0} minutes")

println(((sum * 1.0) / realTime) + "x faster")

}


Since posting this, I've noticed a couple things:

• giveResult in JobQ isn't synchronized, which I believe is the cause of a "deadlock" problem I noticed when running inexpensive tasks (not actually a deadlock, but for the better part of today, that's what I was trying to diagnose).
• I saw a post that mentioned it's good practice not to lock on this, so in JobQ and SyncQueue, I switched to using a separate lock object (defined as class Lock)

I haven't changed the above code though.

• Have you encountered Futures (and actors) in Scala but rejected them for this? They would make a much better solution and largely (or entirely) remove the need to block. – itsbruce Jan 27 '15 at 9:47
• @itsbruce Akka isn't a mess that I want to deal with yet; given how hard the installation is. And futures, I've used, but they don't seem that useful. From what I can tell, the onSuccess etc. callbacks it uses force you to use side effects, and if the result you need isn't calculated by the time you need it, you have to wait for it anyways. – Carcigenicate Jan 27 '15 at 14:48
• Actors have side effects, Futures do not (they just, as you say, block when you ask for their value if it is not yet in). Futures do simplify the work you have done in that code and are composable. – itsbruce Jan 27 '15 at 15:05
• @itsbruce If you could explain something, I would appreciate it. Every tutorial I've ever seen on them puts a println in the success callback to demonstrate the calculation finishing, but doesn't actually show how to obtain the result; besides using Await. Do you always use Await, but use onSuccess to activate some side side-effect? And if you do always use Await, then aren't you blocking anyways? Unless the goal is to do enough after you start the Future that by the time you need the result, it's ready. – Carcigenicate Jan 27 '15 at 15:12
• @itsbruce And, the other point of me writing and posting this was to learn thread safety. – Carcigenicate Jan 27 '15 at 15:13