3
\$\begingroup\$

I receive a message that contains a collection of items that need to be processed.
Processing each item involves sequential calls to various services.
The services return a CompletableFuture that will be completed when the result is available.
Data directed to certain services needs to be sent in a specific order.

For the purpose of this example operationX can be processed in any order and take any amount of time to return.
operationY needs to be executed in the natural order of the data.

To improve performance we want to kick off as many operationX's as we can, so we don't have to wait for each one.

The below solution works, but I don't like it for a few reasons.

  • Mainly because I had to implement my own concurrency controls
  • It requires explicit synchronization in places.
  • I feel like there must be a better, more concise, and common way of achieving the same thing.

The code:

import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.stream.Collectors;
import java.util.stream.IntStream;


public class CompletableFutureOrderedExecutionDemo {

    final Random RANDOM = new Random();

    public static void main(String... args) {
        CompletableFutureOrderedExecutionDemo demo = new CompletableFutureOrderedExecutionDemo();
        demo.processBatchAsynchronously(IntStream.rangeClosed(1, 10).boxed().collect(Collectors.toList()));
    }

    //Example of the equivalent sequential solution
    void processBatchSynchronously(List<Integer> items){
        for(Integer item: items){
            Integer xResult = operationX(item).join();
            Integer yResult = operationY(xResult).join();
        }
    }

    void processBatchAsynchronously(List<Integer> items) {
        OrderedCompletableFutureScheduler<Integer> ocfs = new OrderedCompletableFutureScheduler<>();
        List<CompletableFuture<Integer>> results = items.stream()
                .map(i ->
                        ocfs.enqeue(operationX(i))
                                .thenCompose(this::operationY)
                )
                .collect(Collectors.toList());
        results.forEach(CompletableFuture::join);
    }

    CompletableFuture<Integer> operationX(Integer item) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(RANDOM.nextInt(100));
            } catch (InterruptedException e) {
            }
            System.out.println("X " + item);
            return item;
        });
    }

    CompletableFuture<Integer> operationY(Integer item) {
        System.out.println("Y " + item);
        return CompletableFuture.supplyAsync(() -> item);
    }


    class OrderedCompletableFutureScheduler<T> {
        ConcurrentLinkedQueue<Tup> queue = new ConcurrentLinkedQueue<>();

        CompletableFuture<T> enqeue(CompletableFuture<T> original) {
            CompletableFuture<T> delegate = new CompletableFuture<>();
            original.exceptionally(t -> {
                delegate.completeExceptionally(t);
                return null;
            });
            queue.add(new Tup(original, delegate));
            original.thenRun(() -> {
                synchronized (this) {
                    while (queue.peek().original.isDone()) {
                        Tup head = queue.poll();
                        if (!head.original.isCompletedExceptionally()) {
                            head.delegate.complete(head.original.join());
                        }
                    }
                }
            });
            return delegate;
        }

        class Tup {
            CompletableFuture<T> original;
            CompletableFuture<T> delegate;

            Tup(CompletableFuture<T> original, CompletableFuture<T> delegate) {
                this.original = original;
                this.delegate = delegate;
            }
        }

    }

}

Edit: Additionally it would be good if the implementation catered for multiple types, IE if operation X returned a String and operation Y accepted a String as an argument but returned a Long.
It is unclear to me why the compose method restricts the return type to the same as the parent.

\$\endgroup\$
1
\$\begingroup\$

To better understand your case I imagine such situation:

You have a list of episodes that you wish to watch. You are downloading them concurrently but you will play them in their correct order.

  • operationX is a downloading task.
  • operationY is playing task.

So what constraints do we have?

  1. Episodes are being downloaded concurrently.
  2. Episode will be played after downloading it.
  3. Episode will be played after playing previous one.

First is easy. 2 and 3 are more complicated because you cannot simple schedule playing an episode as a continuation of its downloading and playing previous one. But let’s check what you can do:

void processBatchAsynchronously(List<Integer> items) {
    executeAsync(items).join();
}

CompletableFuture<Void> executeAsync(List<Integer> items) {
    List<CompletableFuture<Integer>> xTasks = items
            .stream()
            .map(this::operationX)
            .collect(toList());

    CompletableFuture<Integer> yTask = completedFuture(null);
    for (CompletableFuture<Integer> xTask : xTasks) {
        final CompletableFuture<Integer> yTaskFinal = yTask;
        yTask = xTask.thenCompose(result -> yTaskFinal.thenCompose(ignored -> operationY(result)));
    }

    return yTask.thenRun(() -> {});
}

How to read this?

  • start all x operations and remember them as a list of tasks.
  • for each x-task schedule a continuation that schedules execution of y operation after previous y operation is finished.
  • return task that completes when last y operation completes (to skip result this operation).

do you like functional programming? You can think about as a some kind of reduction:

CompletableFuture<Void> executeAsync(List<Integer> items) {
    return items.stream()
            .map(this::operationX)
            .reduce(completedFuture(null), (y, x) -> x.thenCompose(r -> y.thenCompose(ignored -> operationY(r))))
            .thenRun(() -> {});

}
\$\endgroup\$
5
  • \$\begingroup\$ There are a couple of issues with this, the first one is that there is no requirement to wait for each opY to complete before scheduling the next, rather there is only a requirement to wait for the last opY request to be sent. The second issue is an oversight in the original implementation where I didn’t cater for different types. Using multiple types makes using compose difficult. \$\endgroup\$
    – Magnus
    Nov 23 '16 at 6:16
  • \$\begingroup\$ I am not sure if I understand it correctly. You said that you want to execute operationY in in the natural order. What does it mean? I understand that second operationX cannot be started before finishing first operation. \$\endgroup\$
    – Karol
    Nov 23 '16 at 9:52
  • \$\begingroup\$ So lets say operation Y is a sending a message over a socket and returning a response, the order the message is placed onto the socket matters, but the response can come back any time and its not really important, other messages should be placed onto the socket before waiting for the response to come back. Operation X can also be a remote call, but the order it is sent doesn't matter. \$\endgroup\$
    – Magnus
    Nov 23 '16 at 10:00
  • \$\begingroup\$ You are invoking an operation that sends message and receives response - returned future completes when response is received. How do you want to determine whether message was sent? \$\endgroup\$
    – Karol
    Nov 23 '16 at 11:09
  • \$\begingroup\$ Well this example doesnt really have implementation details of the task, but the order in which the method operationY is invoked determines the order it is sent, operation y would be blocking until sent, but waiting for the response is non-blocking. \$\endgroup\$
    – Magnus
    Nov 23 '16 at 11:16
1
\$\begingroup\$

I'm not sure about your exact requirements, but I'll try to express the idea.

If I understand correctly the operationX is blocking by its nature and basically you're looking for fire and forget type of behavior. The only thing that is required is the order in which operationY should be processed must be the same as the order of items arrived in process

I think you can solve your problem by refactoring your api into composable functions, which also allow it to be more declarative.

class CompletableFutureOrderedExecutionDemo  {
     String operationX(int i) {
        return "foo-" + i;
     }
     
     long operationY(String someFoo) {
        return someFoo.length;
     }

     Funcrion<Integer, Long> combiner() {
        Function<Integer, String> operX = this::operationX;
        Function<String, Long> operY = this::operationY;
        return operY.compose(operX);
     }

     void process(List<Integer> items) {
        items.forEach(i -> CompletableFuture.runAsync(() -> combiner().apply(i)))
     }   
}

ps. sorry for formatting (answer from phone)

\$\endgroup\$
2
  • \$\begingroup\$ Also on mobile, but as far as I can tell, this doesn't provide any guarantees over when operation y will run, other than after its related operation x. Rather the items that have a faster opX will run their opY sooner. \$\endgroup\$
    – Magnus
    Nov 23 '16 at 21:19
  • \$\begingroup\$ Basically this code ensures that operationY will be submitted for execution in the data arrived order, but does not guarantee that operY it will be completed in the data arrived order. You want achieve the later ? \$\endgroup\$
    – marknorkin
    Nov 24 '16 at 7:52

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.