Ladies(?) and Gentlemen,

I recently (yesterday) attempted to create an Akka Actor Dispatcher using the Scala embedded version of the ForkJoin library, just wanted to verify that I'm doing the Right Thing™

Here's a link to the code: https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/dispatch/FJDispatcher.scala

The design:

Since I'm just forking and never doing any joining, I've done "setAsyncMode(true)" on the ForkJoinPool.
Each Actors mailbox is a Runnable, and to avoid extra allocations I'm weaving in FJMailbox that wraps the mailbox as a ForkJoinTask so I can reuse it.

The mailbox is only available in the pool in a binary fashion (it's added to the pool when a message is added, if messages are added and it's already in the pool it just adds the message to the mailbox,  and it's removed from the pool when the processing is completed).

If a mailbox is added to the pool, and it's done so by a ForkJoinWorkerThread, I use ForkJoinTask.fork(), and if not, I simply add it to the pool. (I always reinitialize it before doing so), here's the code for that: https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/dispatch/FJDispatcher.scala#L73

At the end of processing a mailbox, it will, if it's done by a ForkJoinWorkerThread, it will call helpQuiesce() to make sure that things are getting processed. (It wasn't working until I found out I needed to call helpQuiesce)

Am I on the right track here?

Cheers,
√


--
Viktor Klang

Akka Tech Lead

Does anyone know if there is a version of JSR166Y for Java 5?

On Jul 4, 2011 6:00 PM, "Doug Lea" <dl <at> cs.oswego.edu> wrote:
>
> On 07/02/11 18:48, √iktor Ҡlang wrote:
>>
>> I recently (yesterday) attempted to create an Akka Actor Dispatcher using the
>> Scala embedded version of the ForkJoin library,
>
>
> First off, you should probably target/use/test with the
> current versions, either in jdk7 (snapshots but apparently
> to be released soon) or the jsr166y equivalents. Hopefully
> the Scala releases will adjust to do this soon rather than
> carrying around outdated ones.

Indeed, unfortunately I have a 0 external dependency restriction for this, so hopefully it'll be updated in Scala library for 2.10

>
>
>>
>> Here's a link to the code:
>> https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/dispatch/FJDispatcher.scala
>>
>> The design:
>>
>> Since I'm just forking and never doing any joining, I've done
>> "setAsyncMode(true)" on the ForkJoinPool.
>
>
> This is probably the right approach for messaging, since it
> processes tasks in per-thread-FIFO order. Beware that there
> is much less builtin support for this mode than normal
> recursive-task-based mode.

Yeah, I can imagine. I'm not fully clear on priorities of other workers, they'll only steal work when quiscing?

>
>> Each Actors mailbox is a Runnable, and to avoid extra allocations I'm weaving in
>> FJMailbox that wraps the mailbox as a ForkJoinTask so I can reuse it.
>>
>> If a mailbox is added to the pool, and it's done so by a ForkJoinWorkerThread, I
>> use ForkJoinTask.fork(), and if not, I simply add it to the pool. (I always
>> reinitialize it before doing so),
>
>
> Make sure you don't reinitialize if the task was ever cancelled
> or completed with an exception, since in those cases other threads
> may not see completion before the reset.

Ouch, that sounds nasty, what's the recommended work-around?

>
>
>
>> here's the code for that:
>> https://github.com/jboner/akka/blob/master/akka-actor/src/main/scala/akka/dispatch/FJDispatcher.scala#L73
>>
>
> In current ForkJoinPool, the submit/execute methods already automate
> the internal-fork vs external-submission decision so you can/should skip
> this.

Yeah, I was unsure on the behavior of the Scala Library embedded version.

Also, there can logically be several distinct dispatchers, so I need to make sure that it's not only a FJWorkerThread, but that the current thread belongs to the pool of the mailbox FJTask, is that done by auto?

>
>
>> At the end of processing a mailbox, it will, if it's done by a
>> ForkJoinWorkerThread, it will call helpQuiesce() to make sure that things are
>> getting processed.
>
>
> Yes, helpQuiesce is possible here. There are also the protected
> methods pollNextLocalTask and related support for locally
> processing tasks.

Alright, but what about stealing? Because it's important that Stealing is done so that other workers can help out

If tasks are never joined, then the pool
> mechanics don't otherwise help ensure timely completion.

Thanks for helping me sort this out,

Cheers,

√

>
> -Doug
>
>
> _______________________________________________
> Concurrency-interest mailing list
> Concurrency-interest <at> cs.oswego.edu
> http://cs.oswego.edu/mailman/listinfo/concurrency-interest

I wanted to announce that Java Concurrent Animated now contains some new animations.