1
\$\begingroup\$

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)
            Thread sleep checkDelayMS

    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
    }
}

ThreadPool.scala - Spawns the threads, and manages the queues:

package threadPool

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

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

class ThreadPool[Result](nThreads: Int) {
    //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
    val threads = 1 to nThreads map { _=>
        new Thread( new Worker(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 nThreads = 4
    val nJobs = 10

    val pool = new ThreadPool[Long](nThreads)

    //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")
            Thread.sleep(s)
            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")

    println((realTime / nThreads / 1000.0) + " seconds per thread")

}

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.

\$\endgroup\$
  • \$\begingroup\$ 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. \$\endgroup\$ – itsbruce Jan 27 '15 at 9:47
  • \$\begingroup\$ @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. \$\endgroup\$ – Carcigenicate Jan 27 '15 at 14:48
  • \$\begingroup\$ 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. \$\endgroup\$ – itsbruce Jan 27 '15 at 15:05
  • \$\begingroup\$ @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. \$\endgroup\$ – Carcigenicate Jan 27 '15 at 15:12
  • \$\begingroup\$ @itsbruce And, the other point of me writing and posting this was to learn thread safety. \$\endgroup\$ – Carcigenicate Jan 27 '15 at 15:13

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.