[concurrency-interest] CountedCompleters

Wolfgang Baltes wolfgang.baltes at laposte.net
Mon Apr 16 12:01:50 EDT 2012

I apologize for a mistake in the last paragraph of my memo: using 
quietlyJoinUnforked() in the SDK7 sample code for MergeSort does have a 
non-negligible performance impact (not no impact as stated). There is 
better performance in case of recursions which produce many tasks that 
are not explicitly forked, and it reduces the number of extra threads 
significantly, allowing larger problems to be solved with smaller memory 


On 2012-04-16 16:48, Wolfgang Baltes wrote:
> Thanks, Doug, for a this addition to the FJ framework. I think that
> CountedCompleter will address the needs of an entire class of
> applications in an efficient and simple to use manner.
> I used the code and noticed that method doJoin() has become more
> effective in avoiding blocking threads, and as a result fewer extra
> threads are created. I found the performance, compared to
> RecursiveAction, to be equal or insignificantly different. This
> reduces the problem described in item 3 below.
> However, at the same time, CountedCompleter does not fully satisfy the
> needs for a class of problems I work on. To this end, here are a few
> enhancements I would like to suggest:
> 1: Symmetrically to onCompletion(), provide
> onExceptionalCompletion(Throwable). This allows filtering exception
> propagation. There are cases where the propagation of the exception is
> desired, and others where a corrective action is taken instead, such
> as a retry.
> 2: As a further enhancement to 1: enable any Throwable, including
> checked exceptions. This allows the use of a CountedCompleter as a
> CompletionHandler for asynchronous IO operations or as a wrapper for
> MethodHandles (which throw Throwable) without adding extra logic to
> capture and convert an IO exception. I read the documentation which
> explains why this is currently limited to unchecked exceptions. While
> I can agree with this in general, I feel the argument is weak for
> CountedCompleter if it is there to support asynchronous tasks/events.
> (May I add that using this type of framework is not for the
> faint-hearted anyway!?)
> 3: Provide a method to join a task that is not forked and/or not
> completable, while minimizing worker thread blocking. For example,
> CountedCompleter allows creating chains of dependent tasks. Unless the
> ultimate task (the last in the chain) is forked/exists on the task
> stack AND can complete because all dependencies are resolved, joining
> it will block the worker thread. I noticed (and my testing is limited
> to a few test cases and therefore not representative) the blocking and
> the creation of other worker threads, ultimately running out of memory
> or reaching the thread count limit. If this task is not forked, then
> join()/quietlyJoin() will block the worker thread. The following code
> is my (inexpert) attempt to provide a remedy. It is based on the
> assumption that a task that depends on others for completion is not
> forked until all dependencies are resolved. For example, a
> CountedCompleter implementing CompletionHandler would fork itself
> ("implicit fork") when the IO operation is done. This works very well
> in my test cases, but at this time I would not claim it to be
> universally applicable or error free. It is shown here more to
> demonstrate the attempt rather than as a reference implementation.
> With access to private data structures, this can be done more
> elegantly and more reliably.
>         static final int RETRIES = 16;
>         static final long WAIT_TIMEOUT = 1_000;     // Timeout in
> microseconds.
>         public final void quietlyJoinUnforked() {
>             this.doJoinUnforked(false);
>         }
>         public final void quietlyJoinUnforkedInterruptibly()
>         throws InterruptedException {
>             if (this.doJoinUnforked(true)) {
>                 throw new InterruptedException();
>             }
>         }
>         public final boolean doJoinUnforked(final boolean
> interruptibly) {
>             int retries = RETRIES;
>             boolean wasInterrupted = false;
>             while (!this.isDone()) {
>                 ForkJoinTask<?> t;
>                 if ((t = pollTask()) != null) {
>                     t.quietlyInvoke();
>                     if (t == this) {
>                         break;
>                     }
>                 }
>                 else {
>                     if (retries-- > 0) {
>                         Thread.yield();
>                         continue;
>                     }
>                     wasInterrupted = Thread.interrupted();
>                     try {
>                         // get(...) is used as a timed join(). It is
> assumed that
>                         // other code will perform get() on this task
> to retrieve
>                         // the task's result or exception.
>                         this.get(WAIT_TIMEOUT, TimeUnit.MICROSECONDS);
>                         break;
>                     }
>                     catch (final InterruptedException consumed) {
>                         if (!interruptibly) {
>                             wasInterrupted = true;
>                             continue;
>                         }
>                         return true;
>                     }
>                     catch (final ExecutionException ignored) {
>                         // See comment on get() above.
>                         break;
>                     }
>                     catch (final TimeoutException ignored) {
>                         continue;
>                     }
>                 }
>                 retries = RETRIES;
>             }
>             if (wasInterrupted && !interruptibly) {
>                 Thread.currentThread().interrupt();
>                 return false;
>             }
>             return wasInterrupted;
>         }
> As already mentioned this works quite well in a number of cases. For
> example, adding this method to the example MergeSort code and calling
> quietlyJoinUnforked(), results in the same overall performance,
> reduces the number of extra blocked worker threads to 1 if any
> (instead of up to 8 for the unmodified code; on a PC with 4
> hyper-threading cores/8 threads), and allows for some extra
> (recreational?) freedom in joining the right and left sub-tasks in any
> order. It works in cases where no sub-task is forked explicitly. I
> observed that worker thread blocking only occurs towards the end of a
> large recursion, suggesting that worker threads only block - as
> intended - when there is no other work available (sometimes while
> implicit forking has not yet happened).
> Wolfgang.
> On 2012-04-09 16:16, Doug Lea wrote:
>> After sitting on multiple variations for months, I committed
>> CountedCompleter, a completion-based flavor of ForkJoinTask.
>> As mentioned a few times over the past year, the main motivation
>> is to better support tasks that perform IO or other base
>> actions that may (or may not) take a lot of time to execute.
>> As is the case with JDK7 async IO and other completion-based
>> frameworks, the most common path to efficiency is for such tasks
>> to arrange continuation actions that occur upon their completion.
>> The main twist for CountedCompleters is that continuations
>> might be dependent on multiple actions, not just one. (Or in
>> other words, the continuations must be preceded by a specialized,
>> "bottom-up" form of join.)
>> The CountedCompleter abstract class provides a minimal basis
>> for these kinds of tasks. While some of the mechanics are
>> reminiscent of other FJ-like frameworks such as Intel TBB,
>> CountedCompleters are designed to fit smoothly with other
>> kinds of ForkJoinTasks (like RecursiveActions), and so still
>> allow people to use the more pleasant Future-style conventions
>> rather than count-based bottom-up joining unless they need them.
>> At the same time, the CountedCompleter class exposes enough
>> mechanics to allow all sorts of tweaks that people can use
>> to improve performance.
>> In particular, in addition to usually being the best way to deal
>> with IO etc bound tasks, CountedCompleters sometimes fare better
>> than RecursiveActions in programs that entail lots of garbage
>> collection because GC can have similar impact on task variability.
>> Even though targeted for JDK8, versions of CountedCompleter
>> appear in the jsr166y and main repositories, not jsr166e. This is
>> because they require a non-public hook into modified ForkJoinTask
>> exception handling mechanics in order to properly propagate
>> exceptional completions. For sources, docs, and jar files, see
>> the usual links at
>> http://gee.cs.oswego.edu/dl/concurrency-interest/index.html
>> The API docs include more details and some examples:
>> http://gee.cs.oswego.edu/dl/jsr166/dist/docs/java/util/concurrent/CountedCompleter.html
>> I also added a few (with more to come) test/demo programs that
>> illustrate
>> other usages. See CCBoxedLongSort and CCJacobi in
>> http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/test/loops/
>> Please try these out. As always, comments and suggestions
>> (hopefully based on usage experience) would be welcome.
>> -Doug
>> _______________________________________________
>> Concurrency-interest mailing list
>> Concurrency-interest at cs.oswego.edu
>> http://cs.oswego.edu/mailman/listinfo/concurrency-interest
> _______________________________________________
> Concurrency-interest mailing list
> Concurrency-interest at cs.oswego.edu
> http://cs.oswego.edu/mailman/listinfo/concurrency-interest

More information about the Concurrency-interest mailing list