#scala #actor
#скала #актер
Вопрос:
Проблема: Мне нужно написать приложение для обработки нескольких сотен файлов, на обработку каждого из которых потребуется несколько сотен мегабайт и несколько секунд. Я написал это, используя Future[Report] объекты, созданные с помощью Executors.newFixedThreadPool() , но получил ошибки нехватки памяти, потому что List[Future[Report]] объект, возвращенный ExecutorService.invokeAll() , удерживал промежуточную память, используемую каждым процессом. Я решил проблему, вернув Report объекты из локальных методов в процессорах после вычисления Report значений (всего несколько сотен строк на Report ) вместо того, чтобы выполнять вычисления в call методе (из интерфейса Callable ).
Я хотел бы попробовать решить это с помощью Scala Actors вместо этого. Я создал класс, который принимает последовательность заданий (параметризованные типы для заданий, результатов и функции обработки) и обрабатывает каждое из них в одном из настраиваемого числа Worker экземпляров (подкласс Actor ). Далее следует код.
Проблемы:
-
Я не уверен, что моя обработка правильная.
-
Мне не нравится использовать
CountDownLatchдля задержки возврата результата от диспетчера. -
Я бы предпочел написать более «функциональную» версию диспетчера, которая не изменяет
jobsQueueсписок илиworkersхэш-карту, возможно, заимствуяloopструктуру с хвостовой рекурсией из Clojure (я использовал@tailrec def loopметод в другом коде Scala).
Я с нетерпением жду публикации «Актеров в Скала» Филиппа Халлера и Фрэнка Соммерса.
Вот код:
package multi_worker
import scala.actors.Actor
import java.util.concurrent.CountDownLatch
object MultiWorker {
private val megabyte = 1024 * 1024
private val runtime = Runtime.getRuntime
}
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.
*/
class Worker(id: Int)
extends Actor{
def act() {
// Ask the dispatcher for an initial job
dispatcher ! SendJob(id)
loop {
react {
case Process(job) =>
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 =>
dispatcher ! Stopped(id)
exit()
}
}
}
}
val latch = new CountDownLatch(1)
var res = List.empty[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() {
var jobQueue = jobs
var workers = (0 until actorCount).map(id => (id, new Worker(id))).toMap
workers.values.foreach(_.start())
loop {
react {
case ReportResult(id, result) =>
res = result :: res
if (jobQueue.isEmpty amp;amp; workers.isEmpty) {
latch.countDown()
exit()
}
case SendJob(id) =>
if (!jobQueue.isEmpty) {
workers(id) ! Process(jobQueue.head)
jobQueue = jobQueue.tail
}
case Stopped(id) =>
workers = workers - id
}
}
}
}
dispatcher.start()
/**
* Get the results of the processing -- wait for the dispatcher to finish
* before returning.
*/
def results: List[B] = {
latch.await()
res
}
}
Ответ №1:
После беглого ознакомления я бы предложил следующее обновление:
val resultsChannel = new Channel[List[B]] // used instead of countdown latch to get the results
val dispatcher = new Actor {
def act = loop(Nil, (0 to actorCount).map(id =>
(id, new Worker(id).start.asInstanceOf[Worker])).toMap,
Nil)
@tailrec
def loop(jobQueue: List[A], // queue, workers and results are immutable lists, passed recursively through the loop
workers: Map[Int, Worker],
res: List[B]):Unit = react {
case ReportResult(id, result) =>
val results = result :: res
if (results.size == jobs.size) { // when the processing is finished, sends results to the output channel
resultsChannel ! results
}
loop(jobQueue, workers, results)
case SendJob(id) =>
if (!jobQueue.isEmpty) {
workers(id) ! Process(jobQueue.head)
loop(jobQueue.tail, workers, res)
}
case Stopped(id) =>
loop(jobQueue, workers - id, res)
}
}
dispatcher.start()
def results: List[B] = {
resultsChannel.receive {
case results => results // synchronously wait for the data in the channel
}
}
Комментарии:
1. Прекрасный код. Я бы немного разбил это — методы извлечения и тому подобное — для удобства чтения, но концепция действительно приятная.
Ответ №2:
Вот окончательная версия, которую я придумал (благодаря Василю Ременюку). 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
}
}
}