Dependency between operations in scala actors

Question

I am trying to parallelise a code using scala actors. That is my first real code with actors, but I have some experience with Java Mulithreading and MPI in C. However I am completely lost.

The workflow I want to realise is a circular pipeline and can be described as the following:

Each worker actor has a reference to another one, thus forming a circle
There is a coordinator actor which can trigger a computation by sending a StartWork() message
When a worker receives a StartWork() message, it process some stuff locally and sends DoWork(...) message to its neighbour in the circle.
The neighbours do some other stuff and sends in turn a DoWork(...) message to its own neighbour.
This continues until the initial worker receives a DoWork() message.
The coordinator can send a GetResult() message to the initial worker and wait for a reply.

The point is that the coordinator should only receive a result when data is ready. How can a worker wait that the job returned to it before answering the GetResult() message ?

To speed up computation, any worker can receive a StartWork() at any time.

Here is my first try pseudo-implementation of the worker:

class Worker( neighbor: Worker, numWorkers: Int ) {
   var ready = Foo()
   def act() {
     case StartWork() => { 
       val someData = doStuff()
       neighbor ! DoWork( someData, numWorkers-1 ) 
       }
     case DoWork( resultData, remaining ) => if( remaining == 0 ) {
         ready = resultData
       } else {
         val someOtherData = doOtherStuff( resultData )
         neighbor ! DoWork( someOtherData, remaining-1 )
      }
    case GetResult() => reply( ready )
  }
}

On the coordinator side:

worker ! StartWork()
val result = worker !? GetResult() // should wait

oxbow_lakes oxbow_lakes · Accepted Answer · 2010-03-17T17:56:29

Firstly, you clearly need to have some identifier of what constitutes a single piece of work, so that the GetResult can get the correct result. I guess the obvious solution is to have your actors keep a Map of the results and a Map of any waiting getters:

class Worker( neighbor: Worker, numWorkers: Int ) {
   var res: Map[Long, Result] = Map.empty
   var gets: Map[Long, OutputChannel[Any]] = Map.empty   
   def act() {
     ...
     case DoWork( id, resultData, remaining ) if remaining == 0 =>
       res += (id -> resultData)
       gets.get(id).foreach(_ ! res(id)) //reply to getters when result is ready
       gets -= id //clear out getter map now?
     case GetResult(id) if res.isDefinedAt(d) => //result is ready
       reply (res(id))
     case GetResult(id) => //no result ready 
       gets += (id -> sender)
   }
}

Note: the use of if in the matching condition can make message processing a bit clearer

Dependency between operations in scala actors

2 Answers