Decoupling work and threads

[ben-manes]

I stumbled across your project - neat!

You might find this stackoverflow answer helpful. The idea is to decouple the task from the thread of execution, which can be trivially done using a CompletableFuture. This becomes a poor-man's solution to your problem, where you can prune the mapping by either using a weak valued cache or a whenComplete callback. The asker then wrote this blog after some private emails and his own analysis.

Hope that helps!

[q3769]

Yes, that has been bothering me for a while... although in reality as-is, the uneveness of work distribution among the execeutors usually is not too much, especially when the runtime/resource needed for each job is roughly the same. Nevertheless, the API could use some decoupling like in this idea, instead of totally relying on the mercy of the hashing algorithm for the work distribution.

Hope I can start to work on this soon. Thanks for the idea and pointers!! Wish more folks were interested enough to contribute PRs but understand the use cases of this may not be that many, speically outside the field of async and messaging....

[q3769]

Hi @ben-manes , I know it's a lot to ask but if you could take a glance at this PR when you get a chance #3 ?

Specifically with a complete code overhaul per this idea, now the whole thing is only about 30 lines of real code here: https://github.com/q3769/conseq4j/blob/f3c2143c2d6dff50e09a394cce4af494a87c1bc4/src/main/java/conseq4j/Conseq.java

I am specially not sure whether to use Caffeine, Guava MapMaker, or just JDK concurrent map to host the weak-valued execution queues (CompletableFutures). Wondering if Caffeine (or any kind of cache) is an overkill for this simple case, seeing that the queues/CompletableFutures are ephemeral by nature and depending on the weak reference machanism to clear out of memeory....

Any help is much appreicated.

[ben-manes]

MapMaker is quite old and does not support the Java 8 atomic compute methods. They are inefficiently tacked onto Guava's Cache, but have also been buggy. That's moot since Guava recommends Caffeine for caching.

You can use normal ConcurrentHashMap and clean up using a whenComplete outside of the computation that does a conditional removal, e.g. map.remove(key, future). If it is the last in the chain then this will discard the entry, else no op. You might guard that write with a cheap read check to avoid lock contention (if assuming the removal will usually fail).

The stackoverflow example did not handle error cases. If prior future fails then the dependents using thenRunAsync won't be executed. The given task could be wrapped with error handling to suppress and log any exception.

You're also very much welcome to do something different or fancier. I only offered it for inspiration. I had a use-case where I needed this type of behavior for framework integration but did not want to invest much as unpopular (JCache). I had realized that small snippet did what I wanted trivially, which was neat and let me not waste more effort. You might have more imaginative ideas on where your library evolves, but I'm glad it helps your iterations towards that.

[q3769]

Hmm... I am afraid now I'm out of my element here with the error scenario. As a matter of fact, if we back out one step, with the way my code is currently done in this PR, I am not even sure how it could work in the happy-path scenairo, any more.

I mean, suppose

1. the void runAsync(Object sequenceKey, Runnable task) method starts on a main thread t-m-1 with task1.
2. It goes and gets a new intsance of CompletableFuture cf1 from the compute() method; and suppose task1 takes 10 seconds to complete on CompletableFuture thread t-cf-1.
3. Now, main thread t-m-1 does not wait on t-cf-1's 10 seconds and instead exits right away, which immediately renders cf1 elegible for GC/evict from the cache even though cf1 is still in the middle of running task1.
4. Two seconds later, another main thread t-m-2 gets another task2 with the same sequence key as task1.
5. t-m-2 looks in the cache and finds cf1 is already GC'ed/evicted/null'ed in step 3. That causes another new instance of CompletableFuture cf2 to be computed to run task2, meanwhile main thread t-m-2 exits immediately.
6. Cf2 runs task2 for 5 seconds. When task2 finishes at the 7th second (2 + 5), task1 will still need to run for another 3 seconds (10-7) to finshish.

That is not good, because under the same sequenceKey, task2 is supposed to be trailing task1 with the "cf1.thenRunAsync()", but now task2 finishes three sec ahead of task1, running on cf2. What am I missing....?

[ben-manes]

See the code below for how I would design this with respect to error handling and cancellations. The dispatch method has no library dependencies. To optionally return a result to the caller we use an intermediate data holder so that the future is always successful and unwrap when handing it back. That adds some complexity that might not be needed, but does force us to consider that callers can mutate the future via cancelation or early completion. Since we don't want that to mess with our pipeline, returning back a different future to the caller protects us but requires propagation if canceled. Cancelation may seem tricky since we want to not run the task if it hasn't started yet and don't want to run the next dependent prematurely, but it is merely a flag to no op when our turn. Since CompletableFuture does not honor interrupting an in-flight task we don't need to deal with that complexity either.

I hope that makes sense (and has no bugs!). The core logic is still quite small (60 LOC), maximimizes parallelization, and doesn't require any additional libraries.

Java test

import java.time.Duration;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executor;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;

import org.awaitility.Awaitility;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import com.google.common.base.Stopwatch;
import com.google.common.util.concurrent.Uninterruptibles;

public final class DispatchQueueTest {
  ConcurrentMap<Object, CompletableFuture<Result>> dispatchQueues;
  Executor executor = ForkJoinPool.commonPool();

  @BeforeMethod
  public void before() {
    dispatchQueues = new ConcurrentHashMap<>();
  }

  @Test
  public void dispatch_sucess() {
    var stopwatch = Stopwatch.createStarted();
    var a1 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a1%n", stopwatch);
    });
    var a2 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a2%n", stopwatch);
    });
    var b1 = dispatch("b", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(1));
      System.out.printf("%s: Ran b1%n", stopwatch);
    });
    var b2 = dispatch("b", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran b2%n", stopwatch);
    });

    CompletableFuture.allOf(a1, a2, b1, b2).join();
    Awaitility.await().until(dispatchQueues::isEmpty);
  }

  @Test
  public void dispatch_failure() {
    var stopwatch = Stopwatch.createStarted();
    var a1 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a1%n", stopwatch);
      throw new IllegalStateException("a1");
    });
    var a2 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a2%n", stopwatch);
    });
    var b1 = dispatch("b", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(1));
      System.out.printf("%s: Ran b1%n", stopwatch);
    });
    var b2 = dispatch("b", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
    }).orTimeout(2, TimeUnit.SECONDS).whenComplete((r, e) -> {
      System.out.printf("%s: Ran b2 - %s%n", stopwatch, (e == null) ? r : e);
    });

    CompletableFuture.allOf(a1, a2, b1, b2).exceptionally(t -> null).join();
    Awaitility.await().until(dispatchQueues::isEmpty);
  }

  @Test
  public void dispatch_cancel() {
    var stopwatch = Stopwatch.createStarted();
    var a1 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a1%n", stopwatch);
    });
    var a2 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a2%n", stopwatch);
    });
    a2.exceptionally(t -> {
      System.out.printf("%s: Canceled a2 - %s%n", stopwatch, t);
      return null;
    });
    var a3 = dispatch("a", () -> {
      Uninterruptibles.sleepUninterruptibly(Duration.ofSeconds(3));
      System.out.printf("%s: Ran a3%n", stopwatch);
    });
    CompletableFuture.runAsync(() -> a2.cancel(false),
        CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));

    CompletableFuture.allOf(a1, a2, a3).exceptionally(t -> null).join();
    Awaitility.await().until(dispatchQueues::isEmpty);
  }

  CompletableFuture<Void> dispatch(Object key, Runnable task) {
    return dispatch(key, () -> {
      task.run();
      return null;
    });
  }

  <V> CompletableFuture<V> dispatch(Object key, Supplier<V> supplier) {
    var mayRun = new AtomicBoolean(true);
    var future = dispatchQueues.compute(key, (k, queue) -> {
      return (queue == null)
          ? CompletableFuture.supplyAsync(wrap(supplier, mayRun), executor)
          : queue.thenApplyAsync(ignored -> wrap(supplier, mayRun).get(), executor);
    });

    var f = new CompletableFuture<V>();
    future.whenComplete((result, ignored) -> {
      if (result.error() == null) {
        @SuppressWarnings("unchecked")
        V castedValue = (V) result.value();
        f.complete(castedValue);
      } else {
        f.completeExceptionally(result.error);
      }

      if (dispatchQueues.get(key) == future) {
        dispatchQueues.remove(key, future);
      }
    });
    f.exceptionally(error -> {
      if (f.isCancelled()) {
        mayRun.set(false);
      }
      return null;
    });
    return f;
  }

  static Supplier<Result> wrap(Supplier<?> supplier, AtomicBoolean mayRun) {
    return () -> {
      try {
        return mayRun.get()
            ? new Result(supplier.get(), null)
            : new Result(null, new CancellationException());
      } catch (Throwable t) {
        System.out.println("Failed: " + t); // log & suppress
        return new Result(null, t);
      }
    };
  }

  record Result(Object value, Throwable error) {
    static Result success(Object value) {
      return new Result(value, null);
    }
    static Result error(Throwable error) {
      return new Result(null, error);
    }
  }
}

[ben-manes]

Now, main thread t-m-1 does not wait on t-cf-1's 10 seconds and instead exits right away, which immediately renders cf1 elegible for GC/evict from the cache even though cf1 is still in the middle of running task1.

You forgot that if cf1 is running task1 then the thread has a reference to the future, so it is not yet eligible for garbage collection. The thread has to complete the future (as it might have dependents). When the thread finishes then there are no more references, so it becomes eligible for collection and the cache may sometime later evict the entry.

[q3769]

@ben-manes Again really appreciate the cool code sample. I will dig further when I get to work on this more. One thing I noticed and am unsure about the is the blocking get() call when the queue is not null:

return (queue == null)
          ? CompletableFuture.supplyAsync(wrap(supplier, mayRun), executor)
          : queue.thenApplyAsync(ignored -> wrap(supplier, mayRun).get(), executor);

Doesn't that block the calling thread? Frankly, this is a bit out of my depth so I am not sure...

Meanwhile, I committed, in the PR, a more long-winded setup that has more threading and sync blocks than I wished. Basic idea is using a pub-sub listener to check and clean up and remove the thread from the running concurrent map as needed. The listener is run on a separate thread pool from the pool that services the payload processing.

[ben-manes]

wrap returns a Supplier<Result> so that wasn't a blocking get. Likely the code would be cleaner to have made the call to supplyAsync instead use a lambda supplier so that wrap returned the result directly.

I'll take a peek at your latest work 🙂

[q3769]

Ah, I see! Thanks a bunch also for the code; will study.... A couple of thoughts at the moment: 1. Do we really want to provide the API client a "Future" at all, or just a "void" which would reduce the complexity by a lot? i.e. "void dispatch(Object sequenceKey, Runnable task)" vs. "<T> Future<T> dispatch(Object sequenceKey, Callalbe<T> task)", or both (and more as in the jdk ExecutorService)? So far my real world use cases have been mostly "void". How about what you've seen, in terms of the need? 2. If we provide the "Future" to the client, might we as well make it Guava's "ListenableFuture", such as "<T> ListenableFuture<T> dispatch(Object sequenceKey, Callalbe<T> task)" ?

…

On Fri, Jan 21, 2022 at 10:56 PM Ben Manes ***@***.***> wrote: Now, main thread t-m-1 does not wait on t-cf-1's 10 seconds and instead exits right away, which immediately renders cf1 elegible for GC/evict from the cache even though cf1 is still in the middle of running task1. You forgot that if cf1 is running task1 then the thread has a reference to the future, so it is not yet eligible for garbage collection. The thread has to complete the future (as it might have dependents). When the thread finishes then there are no more references, so it becomes eligible for collection and the cache may sometime later evict the entry. — Reply to this email directly, view it on GitHub <#2 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/ADSJI4DOLB42BT66T4FURL3UXI2JVANCNFSM5MGSSWNQ> . You are receiving this because you commented.Message ID: ***@***.***>

[ben-manes]

I have no opinion on whether to return a future or not, it is your api and design choice. 🙂

It certainly does add to a lot of implementation complexity. I figured that if I was to write the code and tests then I might as well be more complete in case that was helpful. I could imagine that some users would want the results of the computations, so its unclear.

Another possible API would be to return an Executor based on the key, which would itself be responsible for sequencing the submitted work. Then the executor could be supplied elsewhere, like CompletableFuture.runAsync(task, executor), leaving you not responsible for managing the return value. That reminded me of Guava's MoreExecutors.newSequentialExecutor(executor), which appears to implement thesequentiall pipeline in lower-level Java 5 style code. A map of that for keying to the correct instance is probably easier to do directly by the user than managing another library dependency. Of course Guava is large, whereas your library could be a focused solution.

I prefer CompletableFuture over ListenableFuture, but I think they are equivalent enough to be a style choice. It is fairly easy to adapt between all of the callback future types, e.g. future-converter helps to handle that boilerplate.