Java 8 CompletableFuture - fan out implementation

Question

I was wondering what is the best way to implement a fan out type of functionality with Java 8 Completable future. I recently rewrote a function that had a bunch of old Future instances and then calling get in a loop, blocking on each one, to a somewhat cleaner variant using CompletableFuture. However I am seeing about 2x drop in performance so I am assuming something is not quite right in the way I'm using the new API. The code looks something like this:

if (!clinet.login()) {
        throw new LoginException("There was a login error");
    }
    CompletableFuture<List<String>> smths = CompletableFuture
            .supplyAsync(client::getSmth);

    CompletableFuture<List<Data>> smths2 = smths.thenApply(client::getInformation)
            .thenApplyAsync((list) -> list.stream().map(obj -> mapper.map(obj, Data.class)).collect(toList()));

    List<CompletableFuture<Map<String, AnotherData>>> waitGroup = new ArrayList<>();
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getIvPercentileM12M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getIvPercentileM6M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getIvPercentile2M6M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getIvPercentile2M12M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getIvPercentile2M24M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getHvPercentileM6M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getHvPercentile2M6M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getHvPercentileM12M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getHvPercentile2M12M));
    waitGroup.add(notablesFuture.thenComposeAsync(clientb::getHvPercentile2M24M));

    CompletableFuture
            .allOf(waitGroup.toArray(new CompletableFuture[waitGroup.size()]));

    List<Data> data = smths2.join();
    Map<String, Set<AnotherData>> volPercent = waitGroup.stream()
            .map(CompletableFuture::join)
            .flatMap((e) -> e.entrySet().stream())
            .collect(groupingBy(Map.Entry::getKey,
                    mapping(Map.Entry::getValue,
                            toSet())));

    data.forEach((d) -> {
        Set<AnotherData> asdasd = volPercent.get(d.getSymbol());
        if (asdasd != null) {
            d.add(asdasd);
        }
    });
    return stocks;

client::getInformation is a blocking network call returning a List, all the clientb.* are doing is something like:

 return CompletableFuture.supplyAsync(() -> blockingNetworkCall(params, symbols)
            .entrySet().stream()
            .collect(Collectors.toMap(Map.Entry::getKey, value -> new Data(value.getValue(), TimePeriod.M1, TimePeriod.Y1))));

The original code looked something like this:

 List<String> symbols = client.block().get();
    Future<Map<String, Data>> smth = client.block2(symbols);
    Future<Map<String, Double>> ivM6MResultsFuture = clientB.getIvdataM6M(symbols);
    Future<Map<String, Double>> ivM12MResultsFuture = clientB.getIvdataM12M(symbols);
    Future<Map<String, Double>> iv2M6MResultsFuture = clientB.getIvdata2M6M(symbols);
    Future<Map<String, Double>> iv2M12MResultsFuture = clientB.getIvdata2M12M(symbols);
    Future<Map<String, Double>> iv2M24MResultsFuture = clientB.getIvdata2M24M(symbols);
    Future<Map<String, Double>> hvM6MResultsFuture = clientB.getHvdataM6M(symbols);
    Future<Map<String, Double>> hvM12MResultsFuture = clientB.getHvdataM12M(symbols);
    Future<Map<String, Double>> hv2M6MResultsFuture = clientB.getHvdata2M6M(symbols);
    Future<Map<String, Double>> hv2M12MResultsFuture = clientB.getHvdata2M12M(symbols);
    Future<Map<String, Double>> hv2M24MResultsFuture = clientB.getHvdata2M24M(symbols);
    Map<String, Data> doughResults = smth.get();
    Map<String, Double> ivM6MResults = ivM6MResultsFuture.get();
    Map<String, Double> ivM12MResults = ivM12MResultsFuture.get();
    Map<String, Double> iv2M6MResults = iv2M6MResultsFuture.get();
    Map<String, Double> iv2M12MResults = iv2M12MResultsFuture.get();
    Map<String, Double> iv2M24MResults = iv2M24MResultsFuture.get();
    Map<String, Double> hvM6MResults = hvM6MResultsFuture.get();
    Map<String, Double> hvM12MResults = hvM12MResultsFuture.get();
    Map<String, Double> hv2M6MResults = hv2M6MResultsFuture.get();
    Map<String, Double> hv2M12MResults = hv2M12MResultsFuture.get();
    Map<String, Double> hv2M24MResults = hv2M24MResultsFuture.get();

with a big for loop to map all the futures together and aggregate a result. Hopefully it's clear from the code what its doing, but essentially:

1. I make one network call that gets a list
2. Based on this list I call an internal service that generates some objects
3. Based on the first list I spawn a bunch of tasks to fetch various data
4. I populate the objects generated in 2 with the items in 3 - essentially 2 and 3 can run concurrently since they are not dependent.

Two main problems:

1. Do you see any problems with my CompletableFuture usage, and any room to improve the implementation based on the outlined criteria? Currently it's about 2x slower than regular blocking .get() old Futures, that is given as reference.

2. I am a little bit annoyed by the way joining is done, by having to call .allOf() with a void result, is there a better way to do that in the API that I am missing?

As a sidenote, I realize I'm doing a bit more work in the Java 8 variant with a bunch of streams and mapping happening, but the time difference is from 22sec in the old to 45secs in the new, and total items is about 200, so the majority is actually spent in networking and waiting and not the stream operations.

Answer 1

CompletableFuture<List<String>> smths = CompletableFuture
        .supplyAsync(client::getSmth);
CompletableFuture<List<Data>> smths2 = smths.thenApply(client::getInformation)
        .thenApplyAsync((list) -> list.stream().map(obj -> mapper.map(obj, Data.class))
        .collect(toList()));

Since smths is only used in the immediate line, you can consider combining both together:

CompletableFuture<List<Data>> smths2 = CompletableFuture
        .supplyAsync(client::getSmth).thenApply(client::getInformation)
        .thenApplyAsync(list -> list.stream().map(v -> mapper.map(v, Data.class))
        .collect(toList()));

On the same note, further down you are processing your smths2 by the following:

List<Data> data = smths2.join();
// Map something...
data.forEach((d) -> {
    Set<AnotherData> asdasd = volPercent.get(d.getSymbol());
    if (asdasd != null) {
        d.add(asdasd);
    }
});

That can probably be done in a more functional way:

// Map something...
List<Data> data = smths2.join();
data.forEach(v -> Optional.ofNullable(volPercent.get(v.getSymbol()))
    .ifPresent(/* add this to v? */));

Your code actually wouldn't work as a List<Data> can't add() a Set<AnotherData>, possibly only an addAll() if Data is the parent class of AnotherData. Please review this part.

CompletableFuture
        .allOf(waitGroup.toArray(new CompletableFuture[waitGroup.size()]));

The return value from the above doesn't seem to be used at all, copy-and-paste error? Looks like this is probably OK.

I'll like to review the chunk that is waitGroup.add(...), but since it isn't clear what notablesFuture is, and unless one assumes it's the equivalent of doing a List<String> symbols = client.block().get() based on your original code... oh.

Now, it looks like notablesFuture is actually smths, and that changes the answer significantly...

Take #2

First, you probably can use better method names than just getIvPercentileM12M, getHvPercentile2M24M etc. Second, instead of manually creating the List of CompletableFutures, you can probably Stream it too using a small helper method to use method references in-place:

private static <T, U> Function<T, U> ref(Function<T, U> ref) {
    return ref;
}

private static Map<String, Set<Double>> getMap(Client client,
        CompletableFuture<List<String>> notablesFuture) {
    return Stream.of(
            ref(client::getIvPercentileM12M),
            ref(client::getIvPercentileM6M),
            ref(client::getIvPercentile2M6M),
            ref(client::getIvPercentile2M12M),
            ref(client::getIvPercentile2M24M),
            ref(client::getHvPercentileM6M),
            ref(client::getHvPercentile2M6M),
            ref(client::getHvPercentileM12M),
            ref(client::getHvPercentile2M12M),
            ref(client::getHvPercentile2M24M))
        .map(notablesFuture::thenComposeAsync).map(CompletableFuture::join)
        .flatMap(e -> e.entrySet().stream())
        .collect(Collectors.groupingBy(Map.Entry::getKey,
                    Collectors.mapping(Map.Entry::getValue, Collectors.toSet())));
}

Your code in your current method then becomes:

CompletableFuture<List<String>> notablesFuture = CompletableFuture
        .supplyAsync(client::getSmth);
Map<String, Set<Double>> volPercent = getMap(clientb, notablesFuture);
List<Data> data = notablesFuture.thenApply(
            client::getInformation).thenApplyAsync(
            list -> list.stream().map(v -> mapper.map(v, Data.class))
                    .collect(Collectors.toList())).join();
// use volPercent and data

Alternatively, you can consider making use of the CompletableFuture semantics completely (pun unintended):

// modify getMap's return value as such:
private static CompletableFuture<Map<String, Set<Double>>> getMap(Client client,
        CompletableFuture<List<String>> notablesFuture) {
    return CompletableFuture.completedFuture(Stream.of(
            ref(client::getIvPercentileM12M),
            /* remaining method references... */)
        .map(notablesFuture::thenComposeAsync).map(CompletableFuture::join)
        /* remaining flatMap() and collect() steps... */ );
}

And the code in your current method can be then be modified into:

CompletableFuture<List<String>> notablesFuture = CompletableFuture
        .supplyAsync(client::getSmth);
List<Double> result = getMap(clientb, notablesFuture).thenCombineAsync(
            notablesFuture.thenApply(client::getInformation).thenApplyAsync(
                list -> list.stream().map(v -> mapper.map(v, Data.class))
                        .collect(Collectors.toList())), lookup()).join();

The lookup() method processes your Map and List results together in the solution you require, and since as mentioned above about not being able to add() a Set<A> to a List<B>, this will have to be open-ended... With that said, here's a sample implementation for reference:

private static BiFunction<Map<String, Set<Double>>, List<Data>, List<Double>> lookup() {
    return (map, list) -> list.stream().map(v -> map.get(v.getSymbol()))
            .filter(Objects::nonNull).flatMap(Set::stream)
            .collect(Collectors.toList());
}