Argthtjtr

I have the following workflow:
There are n records that need to be retrieved over the network and subsequently n expensive computations that need to be done on each. Put in code, this would look like:

List<Integer> ids = {1,2,....n};

ids.forEach(id -> {

    Record r = RetrieveRecord(id); // Blocking IO

    ProcessRecord(r); // CPU Intensive

})

I would like to convert the blocking part into async so that the time is minimized with a single thread- essentially, by ensuring that record i+1 is being retrieved when record i is being processed. So that the execution would look like:

Retrieve(1).start()

Retrieve(1).onEnd(() -> { start Retrieve(2), Process(1) })

Retrieve(2).onEnd(() -> { start Retrieve(3), Process(2) })

....

Now I can come up with the naive way to implement this with a List<> and CompletableFuture, but this would require me to handle the first record differently.

Is there a more elegant way of solving this with something like reactive streams?
A solution that would maybe let me easily configure how many records Process() can trail behind Retreive()?

So you have N tasks and want to run them in parallel but no more than K tasks simultaneously. Most natural way is to have a task generator and a permission counter with K permissions initially. Task generator creates K tasks and waits for more permissions. Each permission is owned by some task and is returned when the task ends. Standard permission counter in Java is class java.util.concurrent.Semaphore:

List<Integer> ids = {1,2,....n};

Semaphore sem = new Semaphore(K);

ids.forEach(id -> {

    sem.aquire();

    CompletableFuture<Data> fut = Retrieve(id);

    fut.thenRun(sem::release);

    fut.thenAcceptAsync(this::ProcessRecord, someExecutor);

})

Since the task generator occupies only one thread, there is little sense to make it asynchronous. If, however, you don't want to use a dedicated thread for task generator and want to implement asynchronous solution, then the main question is what class can play the role of asynchronous permission counter. You have 3 options:

• use implicit asynchronous permission counter which is a part of reactive streams, found in RxJava, project Reactor etc.


• use explicit asynchronous semaphore org.df4j.core.boundconnector.permitstream.Semafor included in my asynchronous library df4j
  


• make it yourself

Solution using df4j, with explicit asynchronous semaphore:

import org.df4j.core.boundconnector.permitstream.Semafor;

import org.df4j.core.tasknode.Action;

import org.df4j.core.tasknode.messagestream.Actor;



import java.util.Arrays;

import java.util.Iterator;

import java.util.List;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.CountDownLatch;

import java.util.concurrent.ForkJoinPool;



public class AsyncSemaDemo extends Actor {

    List<Integer> ids = Arrays.asList(1, 2, 3, 4, 5);

    Semafor sema = new Semafor(this, 2);

    Iterator<Integer> iter = ids.iterator();

    int tick = 100; // millis

    CountDownLatch done = new CountDownLatch(ids.size());

    long start = System.currentTimeMillis();



    private void printClock(String s) {

        long ticks = (System.currentTimeMillis() - start)/tick;

        System.out.println(Long.toString(ticks) + " " + s);

    }



    CompletableFuture<Integer> Retrieve(Integer e) {

        return CompletableFuture.supplyAsync(() -> {

            printClock("Req " + e + " started");

            try {

                Thread.sleep(tick); // Network

            } catch (InterruptedException ex) {

            }

            printClock(" Req " + e + " done");

            return e;

        }, executor);

    }



    void ProcessRecord(Integer s) {

        printClock(" Proc " + s + " started");

        try {

            Thread.sleep(tick*2); // Compute

        } catch (InterruptedException ex) {

        }

        printClock("  Proc " + s + " done");

    }



    @Action

    public void act() {

        if (iter.hasNext()) {

            CompletableFuture<Integer> fut = Retrieve(iter.next());

            fut.thenRun(sema::release);

            fut.thenAcceptAsync(this::ProcessRecord,  executor)

            .thenRun(done::countDown);

        } else {

            super.stop();

        }

    }



    public static void main(String args) throws InterruptedException {

        AsyncSemaDemo asyncSemaDemo = new AsyncSemaDemo();

        asyncSemaDemo.start(ForkJoinPool.commonPool());

        asyncSemaDemo.done.await();

    }

}

its log should be:

0 Req 1 started

0 Req 2 started

1  Req 1 done

1  Proc 1 started

1 Req 3 started

1  Req 2 done

1  Proc 2 started

1 Req 4 started

2  Req 3 done

2  Proc 3 started

2 Req 5 started

2  Req 4 done

2  Proc 4 started

3   Proc 1 done

3  Req 5 done

3  Proc 5 started

3   Proc 2 done

4   Proc 3 done

4   Proc 4 done

5   Proc 5 done

Note how this solution is close to my previous answer with standard java.util.concurrent.Semaphore.

Here's what I finally came up with that seems to get the job done:

Flowable.just(1,2,3,4,5,6) // Completes in 1 + 6 * 3 = 19 secs

    .concatMapEager(v->

            Flowable.just(v)

            .subscribeOn(Schedulers.io())

            .map(  e->{

                System.out.println(getElapsed("Req " + e + " started");

                Thread.sleep(1000); // Network: 1 sec

                System.out.println(getElapsed("Req " + e + " done");

                return e;

            }, requestsOnWire, 1) // requestsOnWire = K = 2

           .blockingSubscribe(new DisposableSubscriber<Integer>() {

        @Override

        protected void onStart() {

            request(1);

        }

        @Override

        public void onNext(Integer s) {

            request(1);

            System.out.println("Proc " + s + " started");

            try {

                Thread.sleep(3000); // Compute: 3 secs

                System.out.println("Proc " + s + " done");

            } catch (InterruptedException e) {

                e.printStackTrace();

            }

        }

        @Override

        public void onError(Throwable t) {



        }

        @Override

        public void onComplete() {



        }

    });

Below is the execution order. Note that at any given point of time, there are 1 record being processed, at most 2 requests on wire and at most 2 unprocessed records in memory (Process trails behind by K=2) records:

 0 secs: Req 1 started

       : Req 2 started

 1 secs: Req 2 done

       : Req 1 done

       : Proc 1 started

       : Req 3 started

       : Req 4 started

 2 secs: Req 3 done

       : Req 4 done

 4 secs: Proc 1 done

       : Proc 2 started

       : Req 5 started

 5 secs: Req 5 done

 7 secs: Proc 2 done

       : Proc 3 started

       : Req 6 started

 8 secs: Req 6 done

10 secs: Proc 3 done

       : Proc 4 started

13 secs: Proc 4 done

       : Proc 5 started

16 secs: Proc 5 done

       : Proc 6 started

19 secs: Proc 6 done

Hope there are no anti-patterns/pitfalls here.