5
\$\begingroup\$

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.

\$\endgroup\$
3
\$\begingroup\$
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());
}
\$\endgroup\$
3
  • \$\begingroup\$ That was a pretty good explanation. To clarify this CompletableFuture .allOf(waitGroup.toArray(new CompletableFuture[waitGroup.size()])); was done to make sure everything in the waitgroup is completed before we proceed further. IE block until the previous pipeline completes. It's quite ugly so if you can suggest a more idiomatic way of doing it that would be great. \$\endgroup\$
    – Feras
    Jun 15 '15 at 17:40
  • \$\begingroup\$ @Feras I believe you also need to call on the CompletableFuture returned by allOf(...) for it to work, e.g. a join(), at least that's what I gather from the tutorials online and my own little experimentation. I'm also not sure if you came across the same article, but this is suggesting something similar to what you have (search for its sequence() implementation), so I suppose you are not that far off. :) \$\endgroup\$
    – h.j.k.
    Jun 16 '15 at 1:18
  • \$\begingroup\$ Turns out I was wrong in my understanding of allOf(...), wells looks like I learned something new too. \$\endgroup\$
    – h.j.k.
    Jun 16 '15 at 2:34

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.