Вот последняя версия, которую я придумал (спасибо Василу Ременюку).Операторы println, помеченные комментарием // DEBUG, должны показать прогрессию, а метод main является модульным тестом:
import scala.actors.Actor
import scala.actors.Channel
import scala.actors.Scheduler
import scala.annotation.tailrec
object MultiWorker {
private val megabyte = 1024 * 1024
private val runtime = Runtime.getRuntime
def main(args: Array[String]) {
val jobs = (0 until 5).map((value: Int) => value).toList
val multiWorker = new MultiWorker[Int, Int](jobs, 2, { value =>
Thread.sleep(100)
println(value)
value
})
println("multiWorker.results: " + multiWorker.results)
Scheduler.shutdown
}
}
class MultiWorker[A, B](jobs: List[A],
actorCount: Int,
process: (A) => B) {
import MultiWorker._
sealed abstract class Message
// Dispatcher -> Worker: Run this job and report results
case class Process(job: A) extends Message
// Worker -> Dispatcher: Result of processing
case class ReportResult(id: Int, result: B) extends Message
// Worker -> Dispatcher: I need work -- send me a job
case class SendJob(id: Int) extends Message
// Worker -> Dispatcher: I have stopped as requested
case class Stopped(id: Int) extends Message
// Dispatcher -> Worker: Stop working -- all jobs done
case class StopWorking() extends Message
/**
* A simple logger that can be sent text messages that will be written to the
* console. Used so that messages from the actors do not step on each other.
*/
object Logger
extends Actor {
def act() {
loop {
react {
case text: String => println(text)
case StopWorking => exit()
}
}
}
}
Logger.start()
/**
* A worker actor that will process jobs and return results to the
* dispatcher.
*/
case class Worker(id: Int)
extends Actor{
def act() {
// Ask the dispatcher for an initial job
dispatcher ! SendJob(id)
loop {
react {
case Process(job) =>
println("Worker(" + id + "): " + Process(job)) // DEBUG
val startTime = System.nanoTime
dispatcher ! ReportResult(id, process(job))
val endTime = System.nanoTime
val totalMemory = (runtime.totalMemory / megabyte)
val usedMemory = totalMemory - (runtime.freeMemory / megabyte)
val message = "Finished job " + job + " in " +
((endTime - startTime) / 1000000000.0) +
" seconds using " + usedMemory +
"MB out of total " + totalMemory + "MB"
Logger ! message
dispatcher ! SendJob(id)
case StopWorking() =>
println("Worker(" + id + "): " + StopWorking()) // DEBUG
dispatcher ! Stopped(id)
exit()
}
}
}
}
val resultsChannel = new Channel[List[B]]
/**
* The job dispatcher that sends jobs to the worker until the job queue
* (jobs: TraversableOnce[A]) is empty. It then tells the workers to
* stop working and returns the List[B] results to the caller.
*/
val dispatcher = new Actor {
def act() {
@tailrec
def loop(jobs: List[A],
workers: Map[Int, Worker],
acc: List[B]) {
println("dispatcher: loop: jobs: " + jobs + ", workers: " + workers + ", acc: " + acc) // DEBUG
if (!workers.isEmpty) { // Stop recursion when there are no more workers
react {
case ReportResult(id, result) =>
println("dispatcher: " + ReportResult(id, result)) // DEBUG
loop(jobs, workers, result :: acc)
case SendJob(id) =>
println("dispatcher: " + SendJob(id)) // DEBUG
if (!jobs.isEmpty) {
println("dispatcher: " + "Sending: " + Process(jobs.head) + " to " + id) // DEBUG
workers(id) ! Process(jobs.head)
loop(jobs.tail, workers, acc)
} else {
println("dispatcher: " + "Sending: " + StopWorking() + " to " + id) // DEBUG
workers(id) ! StopWorking()
loop(Nil, workers, acc)
}
case Stopped(id) =>
println("dispatcher: " + Stopped(id)) // DEBUG
loop(jobs, workers - id, acc)
}
} else {
println("dispatcher: " + "jobs: " + jobs + ", workers: " + workers + ", acc: " + acc) // DEBUG
resultsChannel ! acc
}
}
loop(jobs, (0 until actorCount).map(id => (id, new Worker(id).start.asInstanceOf[Worker])).toMap, Nil)
exit()
}
}.start()
/**
* Get the results of the processing -- wait for the dispatcher to finish
* before returning.
*/
def results: List[B] = {
resultsChannel.receive {
case results => results
}
}
}