Scala, read file, process lines and write output to a new file using concurrent (akka), asynchronous APIs (nio2)

The trick here is to avoid potentially reading all the data into memory at once. If you iterate and send lines to workers, you run this risk because sending to an actor is async so you might read all the data into memory and it will sit in the mailboxes of the actors probably leading to an OOM exception. A better high level approach would be to use a single master actor and a pool of child workers underneath it for the processing. The trick here is to use a lazy stream over the file (like the Iterator returned from scala.io.Source.fromX) in the master and then use a work-pulling pattern in the workers to prevent their mailboxes with filling up with data. Then, when the iterator no longer has any lines, the master stops itself and that will stop the workers (and if necessary, you can also use this point to shutdown the actor system if that's what you really want to do).

Here is a very rough outline. Please note that I did not test this yet:

import akka.actor._
import akka.routing.RoundRobinLike
import akka.routing.RoundRobinRouter
import scala.io.Source
import akka.routing.Broadcast

object FileReadMaster{
  case class ProcessFile(filePath:String)
  case class ProcessLines(lines:List[String], last:Boolean = false)
  case class LinesProcessed(lines:List[String], last:Boolean = false)

  case object WorkAvailable
  case object GimmeeWork
}

class FileReadMaster extends Actor{
  import FileReadMaster._

  val workChunkSize = 10
  val workersCount = 10

  def receive = waitingToProcess

  def waitingToProcess:Receive = {
    case ProcessFile(path) =>
      val workers = (for(i <- 1 to workersCount) yield context.actorOf(Props[FileReadWorker])).toList
      val workersPool = context.actorOf(Props.empty.withRouter(RoundRobinRouter(routees = workers)))
      val it = Source.fromFile(path).getLines
      workersPool ! Broadcast(WorkAvailable)
      context.become(processing(it, workersPool, workers.size))

      //Setup deathwatch on all
      workers foreach (context watch _)
  }

  def processing(it:Iterator[String], workers:ActorRef, workersRunning:Int):Receive = {
    case ProcessFile(path) => 
      sender ! Status.Failure(new Exception("already processing!!!"))


    case GimmeeWork if it.hasNext =>
      val lines = List.fill(workChunkSize){
        if (it.hasNext) Some(it.next)
        else None
      }.flatten

      sender ! ProcessLines(lines, it.hasNext)

      //If no more lines, broadcast poison pill
      if (!it.hasNext) workers ! Broadcast(PoisonPill)

    case GimmeeWork =>
      //get here if no more work left

    case LinesProcessed(lines, last) =>
      //Do something with the lines

    //Termination for last worker
    case Terminated(ref)  if workersRunning == 1 =>
      //Done with all work, do what you gotta do when done here

    //Terminared for non-last worker
    case Terminated(ref) =>
      context.become(processing(it, workers, workersRunning - 1))

  }
}

class FileReadWorker extends Actor{
  import FileReadMaster._

  def receive = {
    case ProcessLines(lines, last) => 
      sender ! LinesProcessed(lines.map(_.reverse), last)
      sender ! GimmeeWork

    case WorkAvailable =>
      sender ! GimmeeWork
  }
}

The idea is that the master iterates over a file's contents and sends chunks of work to a pool of child workers. When the file processing starts, the master tells all the children that work is available. Each child then continues to request work until there is no more work left. When the master detects that the file is done being read, it broadcasts a poison pill to the children which will let them finish any outstanding work and then stop. When all of the children stop, the master can finish whatever cleanup is needed.

Again, this is very rough based on what I think you are asking about. If I'm off in any area, let me know and I can amend the answer.