15

I have a set of suppliers that all suppply the same result but with different (and varying) speed.

I want an elegant way to start off the suppliers at the same time and as soon as one of them has produced a value, return it (discarding the other results).

I've tried using parallel streams and the Stream.findAny() for this but it always seems to block until all results have been produced.

Here's a unit test demonstrating my problem:

import org.junit.Test;

import java.util.Collections;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Supplier;
import java.util.stream.Stream;

import static org.junit.Assert.*;

public class RaceTest {

    @Test
    public void testRace() {
        // Set up suppliers
        Set<Supplier<String>> suppliers = Collections.newSetFromMap(new ConcurrentHashMap<>());
        suppliers.add(() -> "fast"); // This supplier returns immediately
        suppliers.add(() -> {
            try {
                Thread.sleep(10_000);
                return "slow";
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }); // This supplier takes 10 seconds to produce a value

        Stream<Supplier<String>> stream = suppliers.parallelStream();
        assertTrue(stream.isParallel()); // Stream can work in parallel
        long start = System.currentTimeMillis();
        Optional<String> winner = stream
                .map(Supplier::get)
                .findAny();
        long duration = System.currentTimeMillis() - start;
        assertTrue(winner.isPresent()); // Some value was produced
        assertEquals("fast", winner.get()); // The value is "fast"
        assertTrue(duration < 9_000); // The whole process took less than 9 seconds
    }
}

The result of the test is that the last assertion fails as the whole test takes about 10 seconds to complete.

What am I doing wrong here?

3
  • 2
    I just tested your code and it doesn't fail (JDK 1.8.0_51 here). But findAny is nondeterministic so it's just luck.
    – Tunaki
    Oct 6, 2015 at 11:17
  • 1
    If you add the fast suppliers 4 times (set size = 5), it works as expected, I suspect that when the stream is too small, it reverts to a sequential execution.
    – assylias
    Oct 6, 2015 at 11:27
  • 2
    There is no need for the source to be a Concurrent… collection here. Any collection will do, as long as you don’t modify it during the operation. In fact, due to the way it works internally, an ordinary collection performs better at this task, even when it comes to parallel execution. But as already said by others, the Stream API is not the right tool here.
    – Holger
    Oct 6, 2015 at 12:16

3 Answers 3

12

In this case, you are better off using Callable instead of Supplier (same functional signature) and use the good old concurrency API that exists since Java 5:

Set<Callable<String>> suppliers=new HashSet<>();
suppliers.add(() -> "fast"); // This supplier returns immediately
suppliers.add(() -> {
        Thread.sleep(10_000);
        return "slow";
    }
);

ExecutorService es=Executors.newCachedThreadPool();
try {

    String result = es.invokeAny(suppliers);
    System.out.println(result);

} catch (InterruptedException|ExecutionException ex) {
    Logger.getLogger(MyClass.class.getName()).log(Level.SEVERE, null, ex);
}
es.shutdown();

Note, how the entire “run all and return the fastest” becomes a single method invocation…

It also has the bonus of canceling/ interrupting all pending operations, as soon as one result is available, so the slow operation won’t actually wait the full ten seconds here (well, in most cases, as the timing is not deterministic).

3
  • The Javadoc of invokeAny says "Executes the given tasks, returning the result of one that has completed successfully". Does this implies that the one returned is always the first task that has completed (and not another one)?
    – Tunaki
    Oct 6, 2015 at 12:22
  • 1
    @Tunaki: the interface can’t specify such a guaranty as it depends on the logic of the actual executor implementation, whether this is the case. E.g., in case of a thread pool executor, you must ensure that it has either no limit on the thread number or at least as much free threads as tasks are available for this to work as intended. The executor returned by newCachedThreadPool() has no limit. But while returning the fastest is the obvious intention of invokeAny, if you want to put it as formal as possible, there is never such a guaranty when it comes to concurrency…
    – Holger
    Oct 6, 2015 at 12:30
  • This worked perfectly, and also handles failed tasks just like I intended (throw exception only if no task is successful) Oct 6, 2015 at 15:17
4

The code you are currently using is nondeterministic. Quoting the Javadoc of findAny():

The behavior of this operation is explicitly nondeterministic; it is free to select any element in the stream.

You could use a CompletionService and submit all the tasks to it. Then, CompletionService.take() will return the Future of the first completed task.

long start = System.currentTimeMillis();
ExecutorService executor = Executors.newFixedThreadPool(suppliers.size());
CompletionService<String> completionService = new ExecutorCompletionService<>(executor);
suppliers.forEach(s -> completionService.submit(() -> s.get()));
String winner = completionService.take().get();
long duration = System.currentTimeMillis() - start;
assertEquals("fast", winner); // The value is "fast"
assertTrue(duration < 9_000); // The whole process took less than 9 seconds
1
  • 1
    This is IMHO much more deterministic and idiomatic than the currently accepted answer. Or to state it that way: This is exactly what CompletionService is for!
    – Marco13
    Oct 6, 2015 at 21:43
2

Stream API is not suitable for such things as it does not guarantee when tasks are finished. The better solution would be to use CompletableFuture:

long start = System.currentTimeMillis();
String winner = CompletableFuture
        .anyOf(suppliers.stream().map(CompletableFuture::supplyAsync)
                .toArray(CompletableFuture[]::new)).join().toString();
long duration = System.currentTimeMillis() - start;
assertEquals("fast", winner); // The value is "fast"
assertTrue(duration < 9_000); // The whole process took less than 9 seconds

Note that it still may not launch all the suppliers in parallel if common FJP has not enough parallelism level. To fix this you can create your own pool which has the required parallelism level:

long start = System.currentTimeMillis();
ForkJoinPool fjp = new ForkJoinPool(suppliers.size());
String winner = CompletableFuture
        .anyOf(suppliers.stream().map(s -> CompletableFuture.supplyAsync(s, fjp))
                .toArray(CompletableFuture[]::new)).join().toString();
long duration = System.currentTimeMillis() - start;
assertEquals("fast", winner); // The value is "fast"
assertTrue(duration < 9_000); // The whole process took less than 9 seconds
fjp.shutdownNow();
0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.