public class ParallelismTools
extends Object
ParallelismTools.submit(Callable) are deferred until they
are waited for (ParallelismTools.get(Future), ParallelismTools.join(List),
ParallelismTools.joinAll(List) or ParallelismTools.joinOrdered(List), or the returned
Future's own get()) or polled (its own isDone()), which executes them on
that thread, so a task submitted that
way is never executed if it is never waited for (or polled); all others are
executed immediately, including those given to
ParallelismTools.supplyAsync(Callable), so that the returned CompletableFuture is
already complete and work can be chained on it in any way (a deferred task
could not be run by waiting for work chained on it).
An exception (or error) thrown by a task is rethrown the same way by every
method here that waits for tasks, or runs them: an unchecked exception with a
suppressed exception added to it recording the stack of the rethrowing
thread; a checked exception, which cannot be rethrown as-is, wrapped in a
RuntimeException. When parallel processing is disabled, the unchecked
exception rethrown is always the original. When it is enabled, as
ForkJoinTask.join() does (and so a parallel stream), one thrown on another
thread by a task run as a ForkJoinTask
(ParallelismTools.forEach(Collection, Consumer),
ParallelismTools.invokeAll(Collection, Function), ParallelismTools.invokeAll(Callable[]) and
ParallelismTools.join(CountedCompleter)) is rethrown as a copy where possible: of the
same type, with the original as its cause. Otherwise, it is the original. The
exception is ParallelismTools.stream(Collection), whose terminal operations are the
caller's own.| Modifier and Type | Field and Description |
|---|---|
static String |
RW_PARALLEL
Name of the environment variable to disable parallel processing, set RW_PARALLEL=0 (or
RW_PARALLEL=false, case insensitive) to disable.
|
| Constructor and Description |
|---|
ParallelismTools() |
| Modifier and Type | Method and Description |
|---|---|
static <T> RunnableFuture<T> |
adapt(Callable<T> task)
Adapt a task-with-return value into a RunnableFuture object that implements
the Future interface to be executed by the current thread (as opposed to
submitting it to the thread pool).
|
static <T> Stream<List<T>> |
chunkedStream(@NotNull List<T> items)
The given items in chunks of consecutive items, in order, for a task to be run on each chunk in parallel: when
parallel processing is enabled, the chunks of
ParallelismTools.chunkRanges(int, int) (roughly four per
ParallelismTools.maxParallelism() thread, or one per item, if there are fewer), else a single chunk of all the items. |
static Stream<int[]> |
chunkRanges(int start,
int end)
The indices from start (inclusive) to end (exclusive) in chunks of consecutive indices, in order, for a task to
be run on each chunk in parallel: when parallel processing is enabled, roughly four chunks per
ParallelismTools.maxParallelism() thread (or one per index, if there are fewer), else a single chunk of all the indices. |
static Stream<int[]> |
chunkRanges(int start,
int end,
int minChunkSize)
As
ParallelismTools.chunkRanges(int, int), but with at least the given number of indices in each chunk (but the last),
as for ParallelismTools.chunkSize(int, int). |
static int |
chunkSize(int size)
The number of items in each chunk when the given number of items are split into chunks for a task to be run on
each in parallel, as
ParallelismTools.chunkRanges(int, int) does: when parallel processing is enabled, roughly four
chunks per ParallelismTools.maxParallelism() thread (or one item per chunk, if there are fewer items), else a single
chunk of all the items. |
static int |
chunkSize(int size,
int minChunkSize)
As
ParallelismTools.chunkSize(int), but with at least the given number of items in each chunk (but the last), so fewer
chunks when there are few items (a single one, if there are no more items than that): e.g. |
static <T> void |
forEach(Collection<T> items,
Consumer<T> task)
Run the specified task on all items, blocking until all have been completed.
|
static <T> T |
get(Future<T> future)
Block until the task behind the given Future is complete.
|
static boolean |
getParallel()
Global getter for current parallel processing state.
|
static <T> List<Future<T>> |
invokeAll(Callable<T>... tasks)
Given a list of tasks-with-return-value, block until all tasks
have been completed, using
ForkJoinTask.invokeAll(Collection): all
but the first are forked (to the current thread's own queue, if it is a
thread of a ForkJoinPool, so to that pool, otherwise to the common pool),
the first is executed by the current thread, which then waits for the
others, executing those it can take back from its queue (how many depends
on the JDK), and, on a thread of a ForkJoinPool, also helping the threads
that took the others, or letting the pool compensate for it blocking. |
static <T,R> List<R> |
invokeAll(Collection<T> items,
Function<T,R> task)
Run the specified task on all items, blocking until all have been completed.
|
static void |
invokeAll(Runnable... tasks)
Given a list of tasks-without-return-value, block until all tasks
have been completed.
|
static <T> void |
invokeAllRunnable(Collection<T> items,
Consumer<T> task)
Deprecated.
|
static <T> Deque<Future<T>> |
invokeFirstSubmitRest(Callable<T>... tasks)
Deprecated.
Submit each task with
ParallelismTools.submit(Callable) into a List, and wait for
them in order with ParallelismTools.joinOrdered(List), which also runs any not yet
started. |
static void |
join(@NotNull CountedCompleter<?> completer)
Block until the given CountedCompleter has completed, meanwhile running on this thread any of its descendants
(tasks whose chain of completers leads to it) that no thread has started yet, rather than only blocking.
|
static <T> void |
join(List<? extends Future<? extends T>> futures)
For a given List of Futures, block until all are complete.
|
static void |
joinAll(@NotNull List<? extends Future<?>> futures)
For a given List of Futures, block until all are complete, waiting for
each in order: meanwhile, tasks not yet claimed are stolen in order (the
oldest first, as the pool's workers also take them), so that the current
thread helps complete the tasks roughly in the List's order, unlike
ParallelismTools.join(List), which steals them in reverse order. |
static <T> T |
joinFirst(Deque<Future<T>> futures)
Deprecated.
Wait for the Futures in order with
ParallelismTools.joinOrdered(List), which
likewise steals subsequent unclaimed tasks while waiting for each. |
static <T> Iterator<T> |
joinOrdered(@NotNull List<? extends Future<? extends T>> futures)
For a given List of Futures, return an Iterator over their values, in
order, each of whose next() waits for the next Future as
ParallelismTools.joinAll(List) does: meanwhile stealing tasks not yet claimed in
order, then setting its element to null, and returning its value; or, if
its task threw, cancelling the Futures of those after it and rethrowing
(after which the Iterator must not be used). |
static <E> Iterator<E> |
joinUnordered(@NotNull List<E> items,
@NotNull Function<? super E,? extends CompletableFuture<?>> futureOf)
For a given List of items that each have a CompletableFuture (such as pairs of a CompletableFuture and what it is
for), return an Iterator over the items in the order their CompletableFutures are done, each of whose next()
waits for one of those not yet returned to be done: one already done, if any; else one whose task this thread
runs itself, the first in the List's order not yet claimed by another thread (so, when parallel processing is
disabled, as by
ParallelismTools.supplyAsync(Callable), each is already done and they are returned in the List's
order); else, as every one is running on another thread, whichever is done first. |
static <F extends CompletableFuture<?>> |
joinUnordered(@NotNull List<F> futures)
As
ParallelismTools.joinUnordered(List, Function), for a List of CompletableFutures themselves. |
static int |
maxParallelism()
The number of threads that run tasks in parallel: the common ForkJoinPool's threads
(by default, as many as there are processors minus one) plus the calling thread,
which also runs tasks while waiting for them; or just the calling thread if parallel
processing is disabled.
|
static boolean |
setParallel(boolean parallel)
Global setter to control parallel processing.
|
static <T> Stream<T> |
stream(@NotNull Collection<T> items)
A stream of the given items: a parallel one (run on the common ForkJoinPool, as for
ParallelismTools.forEach(Collection, Consumer)) when parallel processing is enabled, else a sequential one. |
static <T> Stream<T> |
stream(T[] items)
A stream of the given array's items, as
ParallelismTools.stream(Collection) is of a collection's: a parallel one (run on
the common ForkJoinPool) when parallel processing is enabled, else a sequential one, in the array's order. |
static <T> Future<T> |
submit(Callable<T> task)
Submit a task-with-return-value to the common ForkJoinPool, as a
FutureTask
(rather than a ForkJoinTask): any thread waiting for it with ParallelismTools.get(Future),
ParallelismTools.join(List) or ParallelismTools.joinAll(List) runs it itself if no other thread
has claimed it yet, wherever it is in the pool's queues (and so even if the pool has
no threads), and any exception it throws is kept as-is. |
static <T> void |
submit(@NotNull CountedCompleter<?> parent,
@NotNull List<T> items,
@NotNull Consumer<T> task)
Run the specified task on all items without waiting for it, in chunks that are children of the given
CountedCompleter: so that it only completes once the task has been run on all items, and a thread waiting for it
(or any of its completers) helps run any chunks not yet started.
|
static void |
submit(@NotNull CountedCompleter<?> parent,
@NotNull Runnable task)
Run the specified task without waiting for it, as a child of the given CountedCompleter: so that it only
completes once the task has run, and a thread waiting for it (or any of its completers) runs the task itself if
no other thread has started it.
|
static Future<?> |
submit(Runnable task)
Submit a task-without-return-value to the thread pool, as
ParallelismTools.submit(Callable) does
(so, when parallel processing is disabled, it is deferred until waited for). |
static <T> CompletableFuture<T> |
supplyAsync(Callable<T> task)
Submit a task-with-return-value to the common ForkJoinPool, as
ParallelismTools.submit(Callable) does, but returning a
CompletableFuture, so that further work can be chained on its completion (by default, run by the thread that
completes it), in any way. |
static CompletableFuture<Void> |
supplyAsync(Runnable task)
Submit a task-without-return-value to the common ForkJoinPool, as
ParallelismTools.supplyAsync(Callable) does. |
public static final String RW_PARALLEL
public static boolean setParallel(boolean parallel)
ParallelismTools.RW_PARALLEL environment variable.parallel - Enable parallel processing.public static boolean getParallel()
public static <T> Future<T> submit(Callable<T> task)
FutureTask
(rather than a ForkJoinTask): any thread waiting for it with ParallelismTools.get(Future),
ParallelismTools.join(List) or ParallelismTools.joinAll(List) runs it itself if no other thread
has claimed it yet, wherever it is in the pool's queues (and so even if the pool has
no threads), and any exception it throws is kept as-is.
When parallel processing is disabled, the task is not submitted, but deferred until
it is waited for, with ParallelismTools.get(Future), ParallelismTools.join(List),
ParallelismTools.joinAll(List), ParallelismTools.joinOrdered(List) or the
returned Future's own get(), or polled with its isDone(), which executes it on that
thread (see LazyFutureTask). So a task that is never waited for (or polled)
is never executed: for one whose result is not needed, use
ParallelismTools.supplyAsync(Callable), which executes it immediately instead. Nor can a
timed get() bound how long the task takes: with no other thread to run it, it runs
the task to completion on the waiting thread, whatever the timeout (as executing it
immediately would). So a timed get() can only serve as a watchdog when parallel
processing is enabled.T - Type returned by task.task - Task to be performed.public static Future<?> submit(Runnable task)
ParallelismTools.submit(Callable) does
(so, when parallel processing is disabled, it is deferred until waited for).task - Task to be performed.public static <T> CompletableFuture<T> supplyAsync(Callable<T> task)
ParallelismTools.submit(Callable) does, but returning a
CompletableFuture, so that further work can be chained on its completion (by default, run by the thread that
completes it), in any way. Any exception the task throws completes the CompletableFuture exceptionally. A thread
waiting for it, whether with ParallelismTools.get(Future), ParallelismTools.join(List), ParallelismTools.joinAll(List),
ParallelismTools.joinOrdered(List) or ParallelismTools.joinUnordered(List), or with the CompletableFuture's own get() or
join(), runs the task itself if no other thread has started it (and a timed get() then runs it to the end,
whatever the timeout).
When parallel processing is disabled, the task is executed immediately on the current thread, so that the
returned CompletableFuture is already complete: work chained on it then runs as soon as it is chained, on the
thread chaining it.T - Type returned by task.task - Task to be performed.public static CompletableFuture<Void> supplyAsync(Runnable task)
ParallelismTools.supplyAsync(Callable) does.task - Task to be performed.public static <T> T get(Future<T> future)
T - Type returned by task.future - Future representing previously submitted task.@Deprecated
@SafeVarargs
public static <T> Deque<Future<T>> invokeFirstSubmitRest(@NotNull
Callable<T>... tasks)
ParallelismTools.submit(Callable) into a List, and wait for
them in order with ParallelismTools.joinOrdered(List), which also runs any not yet
started.ParallelismTools.submit(Callable), until they are passed to
ParallelismTools.joinFirst(Deque) (or waited for by ParallelismTools.get(Future),
ParallelismTools.join(List), ParallelismTools.joinAll(List), ParallelismTools.joinOrdered(List) or the
Future's own get(), or polled by its own isDone()).T - Type returned by all tasks.tasks - List of tasks to be executed (if none, an empty Deque is returned).@Deprecated public static <T> T joinFirst(Deque<Future<T>> futures)
ParallelismTools.joinOrdered(List), which
likewise steals subsequent unclaimed tasks while waiting for each.T - Type returned by all tasks.futures - A Deque of Future objects corresponding to previously
submitted tasks.public static <T> void join(List<? extends Future<? extends T>> futures)
ParallelismTools.joinAll(List), which steals them in order instead.T - Type returned by all tasks.futures - A List of Future objects corresponding to previously
submitted tasks.public static void joinAll(@NotNull
@NotNull List<? extends Future<?>> futures)
ParallelismTools.join(List), which steals them in reverse order. Useful when the
tasks' results are consumed in the List's order as they complete (e.g. by
work chained on their CompletableFutures), so that fewer results are held
until the earlier ones have completed.
Once each Future has completed, its element is set to null, so that the
List no longer keeps it (or its result) from being garbage collected once
nothing else refers to it (e.g. once work chained on it has consumed its
result), rather than until the List itself is. So the List must be
modifiable, and must not be read by anything else while (or after) this
runs: work chained on a Future should be given the Future itself, rather
than look it up in the List later.
If a task threw an exception, it is rethrown (as described for this
class) once the tasks before it have completed,
after cancelling the Futures of those after it (or of that one and those
after it, if the wait for it was interrupted), leaving their elements in
the List: so that, when this returns or throws, every Future given to it
is done, and work chained on them can complete, without waiting for a
thread that may never run their tasks (such as when the pool has no
threads). A task not yet started is then never run; one that is running
runs to the end, but its result is discarded.futures - A modifiable List of Future objects corresponding to
previously submitted tasks, whose elements are accessed by
index (so should be a random-access list, such as an
ArrayList), and are each set to null once completed.public static <T> Iterator<T> joinOrdered(@NotNull
@NotNull List<? extends Future<? extends T>> futures)
ParallelismTools.joinAll(List) does: meanwhile stealing tasks not yet claimed in
order, then setting its element to null, and returning its value; or, if
its task threw, cancelling the Futures of those after it and rethrowing
(after which the Iterator must not be used). Useful when each value is
consumed by the waiting thread (e.g. copied elsewhere) before it waits
for the next, rather than by work chained on the Futures.
The given List is modified: each next() sets the element of the Future
it waited for to null, leaving those not yet reached (and that of a task
that threw) as they are, so that the List no longer keeps a Future (or
its value) once it has been waited for. So the List must be modifiable,
and may be read between calls to next() (e.g. to cancel the Futures not
yet waited for), when each Future already waited for is null. Futures may
also be added to its end between calls to next() (e.g. as tasks are
submitted while earlier ones are consumed): hasNext() and next() then
include them, and next() may steal their tasks.T - Type returned by all tasks.futures - A modifiable List of Future objects corresponding to
previously submitted tasks, whose elements are accessed by
index (so should be a random-access list, such as an
ArrayList).public static <F extends CompletableFuture<?>> Iterator<F> joinUnordered(@NotNull
@NotNull List<F> futures)
ParallelismTools.joinUnordered(List, Function), for a List of CompletableFutures themselves.F - Type of the CompletableFutures.futures - A modifiable List of CompletableFutures, e.g. from ParallelismTools.supplyAsync(Callable).public static <E> Iterator<E> joinUnordered(@NotNull
@NotNull List<E> items,
@NotNull
@NotNull Function<? super E,? extends CompletableFuture<?>> futureOf)
ParallelismTools.supplyAsync(Callable), each is already done and they are returned in the List's
order); else, as every one is running on another thread, whichever is done first. Useful when each result is
consumed by this thread as soon as it is ready, in whatever order the tasks complete. Its value, or the exception
its task threw, is not taken here: get it with ParallelismTools.get(Future), which rethrows such an exception as
described for this class.
As with ParallelismTools.joinOrdered(List), the given List is modified: each next() sets the element of the item it
returns to null, so that the List no longer keeps it once returned. So the List must be modifiable, and may be
read between calls to next() (e.g. to close the results of those not yet returned if consuming one fails), when
each item already returned is null. Items may also be added to its end between calls to next(): hasNext() and
next() then include them.
Only CompletableFutures can be waited for this way, as for any one of them (CompletableFuture.anyOf()): any other
Future (e.g. from ParallelismTools.submit(Callable)) could only be polled.E - Type of the items.items - A modifiable List of items, none of them null, whose elements are accessed by index (so should be a
random-access list, such as an ArrayList).futureOf - The CompletableFuture of an item.public static void join(@NotNull
@NotNull CountedCompleter<?> completer)
completer - The CountedCompleter to wait for.public static <T> Stream<T> stream(@NotNull
@NotNull Collection<T> items)
ParallelismTools.forEach(Collection, Consumer)) when parallel processing is enabled, else a sequential one.
Unlike the other methods here, exceptions thrown by its terminal operation are exactly as the JDK throws them:
without a suppressed exception recording the stack of the rethrowing thread (and, when parallel, one thrown on
another thread is rethrown as a copy where possible, as described for this class).T - Item type.items - The items to stream.items.parallelStream() if ParallelismTools.getParallel(), else items.stream().public static <T> Stream<T> stream(@NotNull
T[] items)
ParallelismTools.stream(Collection) is of a collection's: a parallel one (run on
the common ForkJoinPool) when parallel processing is enabled, else a sequential one, in the array's order.T - Item type.items - The items to stream.stream(Arrays.asList(items)), which wraps the array without copying it.public static int chunkSize(int size)
ParallelismTools.chunkRanges(int, int) does: when parallel processing is enabled, roughly four
chunks per ParallelismTools.maxParallelism() thread (or one item per chunk, if there are fewer items), else a single
chunk of all the items.size - The number of items.public static int chunkSize(int size,
int minChunkSize)
ParallelismTools.chunkSize(int), but with at least the given number of items in each chunk (but the last), so fewer
chunks when there are few items (a single one, if there are no more items than that): e.g. when each chunk has a
cost of its own (such as a buffer, or a task's overhead) that is only worth paying for enough items.size - The number of items.minChunkSize - The fewest items in each chunk (but the last).public static Stream<int[]> chunkRanges(int start,
int end)
ParallelismTools.maxParallelism() thread (or one per index, if there are fewer), else a single chunk of all the indices.
Every index is in exactly one chunk, and no chunk is empty.start - The first index.end - One past the last index.{from, to}: the indices from from
(inclusive) to to (exclusive); none if end is not after start. Call parallel() on it to run a task on
each chunk in parallel.public static Stream<int[]> chunkRanges(int start,
int end,
int minChunkSize)
ParallelismTools.chunkRanges(int, int), but with at least the given number of indices in each chunk (but the last),
as for ParallelismTools.chunkSize(int, int).start - The first index.end - One past the last index.minChunkSize - The fewest indices in each chunk (but the last).{from, to}: the indices from from
(inclusive) to to (exclusive); none if end is not after start.public static <T> Stream<List<T>> chunkedStream(@NotNull
@NotNull List<T> items)
ParallelismTools.chunkRanges(int, int) (roughly four per
ParallelismTools.maxParallelism() thread, or one per item, if there are fewer), else a single chunk of all the items.
Each chunk is a view of the given list (see List.subList(int, int)), so the list must not be
structurally modified while any chunk is in use.T - Item type.items - The items, which should be a random-access list (such as an ArrayList), as each chunk starts at
an index.public static <T> void forEach(Collection<T> items,
Consumer<T> task)
T - item typeitems - the items to call the task withtask - the task that should be executed for all itemspublic static <T> void submit(@NotNull
@NotNull CountedCompleter<?> parent,
@NotNull
@NotNull List<T> items,
@NotNull
@NotNull Consumer<T> task)
ParallelismTools.forEach(Collection, Consumer) does,
the items are split into roughly four chunks per ParallelismTools.maxParallelism() thread (or one per item, if there are
fewer), each of which is forked to the common ForkJoinPool. If parallel processing is disabled, the task is
instead run on all items immediately, on the calling thread, with ParallelismTools.forEach(Collection, Consumer).
If the task throws for any item, the parent (and, unless it is overridden not to, each of its completers)
completes exceptionally (see ParallelismTools.join(CountedCompleter)); if parallel processing is disabled, the exception
is instead rethrown immediately, as described for this class.T - Item type.parent - The CountedCompleter to run the chunks as children of. Its pending count is increased by the
number of chunks; it must still call tryComplete() itself, once it has submitted all its children.items - The items, whose elements are accessed by index (so should be a random-access list, such as an
ArrayList), and which must not be modified until the parent has completed.task - The task to run on each item.public static void submit(@NotNull
@NotNull CountedCompleter<?> parent,
@NotNull
@NotNull Runnable task)
ParallelismTools.join(CountedCompleter)); if parallel processing is disabled, the exception is instead
rethrown immediately, as described for this class.parent - The CountedCompleter to run the task as a child of. Its pending count is increased by one; it must
still call tryComplete() itself, once it has submitted all its children.task - The task to run.@Deprecated
public static <T> void invokeAllRunnable(Collection<T> items,
Consumer<T> task)
ParallelismTools.forEach(Collection, Consumer)public static void invokeAll(@NotNull
Runnable... tasks)
ParallelismTools.forEach(Collection, Consumer) with each task as an item.tasks - List of tasks-without-return-value.public static <T,R> List<R> invokeAll(Collection<T> items,
Function<T,R> task)
ParallelismTools.forEach(Collection, Consumer), including how exceptions are rethrown.T - item typeR - type returned by the taskitems - the items to call the task withtask - the task that should be executed for all items@SafeVarargs public static <T> List<Future<T>> invokeAll(Callable<T>... tasks)
ForkJoinTask.invokeAll(Collection): all
but the first are forked (to the current thread's own queue, if it is a
thread of a ForkJoinPool, so to that pool, otherwise to the common pool),
the first is executed by the current thread, which then waits for the
others, executing those it can take back from its queue (how many depends
on the JDK), and, on a thread of a ForkJoinPool, also helping the threads
that took the others, or letting the pool compensate for it blocking.
If any task throws an exception, one of them (not necessarily the first in
order) is rethrown as described for this class. In that case, the other
tasks may have been cancelled or not run. When parallel processing is
disabled, all tasks are executed in order on the current thread, and then
the first exception (in order) is rethrown.T - Type returned by all tasks.tasks - List of tasks-with-return-value.public static <T> RunnableFuture<T> adapt(Callable<T> task)
T - Type returned by task.task - Task with return value.public static int maxParallelism()