Timeout Decorator

[maarek]

This pull request is for Enhancement #67

Adds a decorator on some Supplier, Function, Runnable, Consumer, and Callable in and wraps the function in a single threaded executed Future with a timeout specified in the TimeoutConfig.

Upon Concurrent TimeoutException, ExecutableException, InterruptedException, wraps checked exception in a RuntimeException resilience4j.timeout.TimeoutException with cause being the checked exception.

Example Usage:

 // Default Config
Timeout timeout = Timeout.ofDefault();
CheckedRunnable timedRun = Timeout.decorateCheckedRunnable(timeout, this::doSomething);

// Custom Config
TimeoutConfig timeoutConig = TimeoutConfig.builder()
	.timeoutDuration(Duration.ofMillis(10000))
        .cancelOnException(FALSE)
	.build()
Timeout timeout = Timeout.of(timeoutConfig);
CheckedRunnable timedRun = Timeout.decorateCheckedRunnable(timeout, this::doSomething);

// Or Default Config with specified Timeout
Timeout timeout = Timeout.of(Duration.ofMillis(10000));
CheckedRunnable timedRun = Timeout.decorateCheckedRunnable(timeout, this::doSomething);

Note: It will wrap a functions Throwable in a Concurrent ExecutionException which will throw as a TimeoutException(ExecutionException(Throwable)). Not sure if this is the best approach.

[RobWin]

Thx for your PR. We have a look at it.
If you like Resilience4j, we would love to get your GitHub Star.

[coveralls]

Coverage increased (+0.02%) to 92.692% when pulling ca088b2 on maarek:timeout-decorator into 2f71963 on resilience4j:master.

[storozhukBM]

Hi @maarek,
Thank you for your contribution.
I want to set your expectations just to make sure that we're on the same page here.
You picked up very tough task. It requires careful solution and implementation review, so it will take a while. Also if your solution will be accepted Timeout will become the part of our core API, for you it means extra effort for Documentation, Performance and Stress tests.
But please don't rush to document everything, because your API may be changed during code review.

Here is first points to discuss:
Personally I consider your approach with separate threads is very questionable, because:

• Thread.interrupt approach isn't always working (some links from Google: 1, 2, 3), so it at least requires explicit documentation of what type of tasks can be decorated.
• Additional thread for each task can be real "surprise" (in a bad sense) for users that are very careful with their resources allocation.

Ideally you should decorate only already provided instances of java.util.concurrent.Future interface, this will give your users choice to pass some specific implementations that can handle Future.cancel method properly (close JDBC statement or IO Socket). There is plenty of such Futures implementations that our library users can choose of.

[maarek]

@storozhukBM Thanks for your comments. I was particularly interested in getting the ball rolling on this enhancement. I have no problem with changes to the implementation or API involved and I hope that we can come up with a solution that fits well into the framework.

[RobWin]

We could get some inspiration from Guava's TimeLimiter
The ExecutorService could be configurable. But then Timeout would be almost a Thread-Pool based BulkHead implementation.

[storozhukBM]

OK guys,

Here is my proposal:

1. For now we can leave in Timeout only decorators for java.util.concurrent.Future and Future supplier. This decorators still will be convinient to use with any type that implements Future interface and there are whole bunch of them almost in any async library or framework:
   1.1. CompletableFuture (from JDK)
   1.2. AsyncResult (from Spring)
   1.3. Promise (from Netty)
   1.4. HttpRequestFutureTask (from Apache commons)
   1.5. ListenableFuture (from Guava)
   1.6. Here can be your library too, just call 555-55-555 😄
   All this specific implementations have appropriate Future.cancel method provided.

2. In scope of other enhancement we can create our own implementation of ExecutorService which will integrate all our features: CircuitBreaker, RateLimiter, Retry, Bulkhead, Timeout...
   And also will be easy monitor-able and convinient to use.
   And I want also implement "Thread injection" algorithm described in this paper and also here.
   This will give us ability to maximize ExecutorService throughput with minimal resources possible (seems to be very cool feature).

Hope it all makes sense for you guys. I want to hear your thoughts on it.

[sdanzo]

@storozhukBM @maarek concerning Java stopping a thread, I've found that the following code snippet is very reliable to stopping Java threads (if the design above is conducive to using thread executors):

 executorService.shutdownNow();
 executorService.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
 executorService = null;

I also use a controlled thread shutdown approach with a synchronized Boolean, "isRunning" to aid in gracefully shutting down threads who are also monitoring that semaphore.

[gibffe]

Maybe we should take a step back for a second and think about "timing out" work without the use of threads, or at least using a small, fixed set that does not grow with the amount of work to cancel timeout.

Does the timeout has to be exact or could there be some slack - could the "good enough" approximation of requested timeout give us some headroom in possible implementations ?

I am thinking along the lines of a global entity that can timeout work, and signal it accordingly - interrupt threads, inject error emission, cancel disposables etc.

Just an idea.

[drmaas]

Maybe already discussed elsewhere, but why not add timeouts directly to CircuitBreaker? All you really need to do is time some piece of code, and if it exceeds the timeout threshold, execute onError. Otherwise execute onSuccess.

I say this because most of the use cases we have involve treating a timeout similarly to an exception. If too many timeouts occur we'd want the circuit breaker to open to prevent future executions.

The downside is that you'd have to rely on the code segments not running infinitely, and you'd be recording errors potentially a large amount of time after the timeout threshold was breached (say the threshold was 500 ms and the code took 10 seconds). Also, the work would not be cancellable.

[RobWin]

@drmaas The idea was to create a lightweight, extendable Circuit Breaker where you can add additional features by combining different decorators (higher order functions).
My initial concept was that if you send a message to a remote system, you should set a network timeout on the protocol level (e.g. HTTP, JDBC). Network timeout exceptions can already be handled by the CircuitBreaker.
And if you use an async library like Vert.x, Ratpack, Netty or RxJava, you should use the capabilities of the framework to timeout calls, since it's framework-specific. Most of the Future/Promise/Observable implementations allow to set a timeout easily.

RxJava: You can combine RxJava's timeout operator with our custom CircuitBreaker RxJava operator as follows:

 Observable<String> observable = observable.timeout(1, TimeUnit.SECOND)
    .lift(CircuitBreakerOperator.of(circuitBreaker))
    .subscribe()

Java8: You could decorate a Callable as follows:

static <T> Try.CheckedSupplier<T>  decorateCallable(CircuitBreaker circuitBreaker, ExecutorService executor, Callable<T> callable, Duration timeout){
        return () -> {
            Future<T> future = executor.submit(callable);
            return CircuitBreaker.decorateSupplier(circuitBreaker, () -> future.get(timeout.toMillis(), TimeUnit.MILLISECONDS)).get();
        };
    }

Vertx: You can create a timer which fails a Vert.x Future.

vertx.setTimer(1000, id -> {
  if (!future.completed()) {
    future.fail(“timed out”);
  }
});

Ratpack plans to add a timeout operator for it's Promise: ratpack/ratpack#495

We could implement a decorator which works for a set of interfaces like Java's Future (as @storozhukBM suggested).

We should not add timeout handling into the CircuitBreaker directly, because we don't execute calls in a different thread. Timeouts are also not working correctly in Hystrix, when you use ExecutionIsolationStrategy.SEMAPHORE. At least when I understand the issues correctly: Netflix/Hystrix#835 and Netflix/Hystrix#1345

[storozhukBM]

@RobWin 👍 💯 👍

[storozhukBM]

OK @maarek,
If you want to work on this issue, let's stick with Futures version.
For this feature can satisfy every paradigm.
For ExecutorService implementation I've created separated issue (see #146 ).
I'm ready to help with any your questions.

[maarek]

@storozhukBM

Sorry, I was toying with some ideas over the weekend on the ExecutorService impl.

When you say decorate a Future, are you wanting to return a Future or return the standard supplier type (Callable, Runnable, etc) to be used in the CircuitBreaker.

static <T, F extends Future<T>> Callable<T> decorateFuture(Timeout timeout, F future)

[storozhukBM]

We need to have only two decorators:

static <T, F extends Future<T>> F decorateFuture(Timeout timeout, F future)
static <T, F extends Future<T>> Supplier<F> decorateFutureSupplier(
    Timeout timeout,
    Supplier<F> futureSupplier
)

[RobWin]

Can we even decorate a Future? Does it actually make sense? We can decorate a Future with a CircuitBreaker, but not with a preconfigured timeout value, or?
We can only execute Future.get with a preconfigured value and return the result or a callable which returns the result. @maarek API signature is correct.

[maarek]

@RobWin This is what I struggled with. A Future is not technically decoratable. What I can do is Proxy future.get() and/or future.get(oldTimeout, oldUnit) to future.get(newTimeout, newUnit) if that is the route that we'd like to go. I am working on that now.

How does the CircuitBreaker decorate a Future now though? Through a future Supplier?

[RobWin]

No, not yet. But I guess input would be a Supplier<Future<R>> and result would be Future<R>.
Somehow similar to this implementation: https://github.com/resilience4j/resilience4j/blob/master/resilience4j-vertx/src/main/java/io/github/resilience4j/circuitbreaker/VertxCircuitBreaker.java

[RobWin]

I think I know how we can implement @storozhukBM signature now.
Smart idea @storozhukBM :)

[maarek]

Just pushed a preliminary update to the branch providing the two decorators through proxy.

Codacy and Travis hate me for some reason.

[RobWin]

I think we can implement this without a proxy by executing the Future and returning a new (Completable)Future which either contains the result of the decorated Future or the Exception/TimeoutException.

[maarek]

Does it matter then that you are converting the input Future from some F extends Future to CompletableFuture?

[RobWin]

CompletableFuture implements Future. We should not change the return type.

[storozhukBM]

NOTE: I renamed io.github.resilience4j.timeout.TimeoutException to TimeExceededException in order to avoid name collisions an and confusion with java.util.concurrent.TimeoutException.

My original idea was simpler :

package io.github.resilience4j.timeout;

import static java.util.Objects.requireNonNull;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author bstorozhuk
 */
public class TimeoutFuture<T> implements Future<T> {

    private final Future<T> target;
    private final long maxTimeoutInNanos;
    private final AtomicLong waitingStartTimeInNanos = new AtomicLong();

    public TimeoutFuture(TimeoutConfig config, Future<T> target) {
        maxTimeoutInNanos = config.getTimeoutDuration().toNanos();
        this.target = target;
    }

    @Override public boolean cancel(boolean mayInterruptIfRunning) {
        return target.cancel(mayInterruptIfRunning);
    }

    @Override public boolean isCancelled() {
        return target.isCancelled();
    }

    @Override public boolean isDone() {
        return target.isDone();
    }

    @Override public T get() throws InterruptedException, ExecutionException {
        try {
            long timeLeftInNanos = timeLeftInNanos();
            T result = target.get(timeLeftInNanos, NANOSECONDS);
            return result;
        } catch (TimeoutException e) {
            target.cancel(true);
            throw new TimeExceededException(e);
        }
    }

    @Override public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        requireNonNull(unit, "Unit can't be null");
        long timeLeftInNanos = timeLeftInNanos();
        long specifiedTimeoutInNanos = unit.toNanos(timeout);
        if (specifiedTimeoutInNanos > timeLeftInNanos) {
            return this.get();
        }
        T result = target.get(specifiedTimeoutInNanos, NANOSECONDS);
        return result;
    }

    private long timeLeftInNanos() {
        long startTimeInNanos = waitingStartTimeInNanos.get();
        if (startTimeInNanos == 0) {
            waitingStartTimeInNanos.set(System.nanoTime());
        }
        Long currentWaitingTimeInNanos = System.nanoTime() - startTimeInNanos;
        return maxTimeoutInNanos - currentWaitingTimeInNanos;
    }
}

Then you can use it like that:

    static <T> Future<T> decorateFuture(Timeout timeout, Future<T> future) {
        return new TimeoutFuture<>(timeout.getTimeoutConfig(), future);
    }

    static <T> Supplier<Future<T>> decorateFutureSupplier(Timeout timeout, Supplier<Future<T>> futureSupplier) {
        return () -> {
            Future<T> target = futureSupplier.get();
            return new TimeoutFuture<>(timeout.getTimeoutConfig(), target);
        };
    }

[maarek]

I went the proxy route figuring that we wouldn't want to return a different Future type than the input type. Even if they are all implement Future. The TimeoutFuture would never be usable in instances like CompletableFuture composition or Promise chaining.

Note: I'm not sure if those are valid cases but just something that came to mind.

[storozhukBM]

From the other point of view it is not good to generate proxy for each decorated object.
I agree that simply return Future was a bad idea. So maybe, @RobWin is right in his thoughts about always returning CompletableFuture it at least can be adapted to almost any kind of modern Promises from other libraries.

[storozhukBM]

@maarek please note that I changed TimeoutFuture code example to get rid of thread locals

[maarek]

Before I commit again: Should we return a CompletableFuture, a custom TimeoutFuture or TimeoutCompletableFuture? :)

[storozhukBM]

TimeoutCompletableFuture which will override get methods of the CompletableFuture

[RobWin]

Java 9’s CompletableFuture introduces new methods like orTimeout and completeOnTimeOut that provide built-in support for dealing with timeouts. It internally uses a ScheduledThreadExecutor and completes the CompletableFuture with a TimeoutException after the specified timeout has elapsed. It also returns another CompletableFuture, meaning you can further chain your computation pipeline and deal with the TimeoutException. See blog post: http://www.esynergy-solutions.co.uk/blog/asynchronous-timeouts-completable-future-java-8-and-9 or http://www.nurkiewicz.com/2014/12/asynchronous-timeouts-with.html
I don't think that we should reimplement this timeout mechanism.

I provide my idea for TimeLimiter in some minutes.

[RobWin]

What we actually want to achieve is that we want to call a potentially long running method asynchronously, but timeout when it takes to long and optionally cancel the long running method.
Furthermore, we want to capture the TimeoutException to trigger a CircuitBreaker.

What about this solution.

static <T> Callable<T> decorateFuture(TimeLimiter timeLimiter,
                                      Supplier<Future<T>> supplier){
    return () -> {
        Future<T> future = supplier.get();
        try {
            return future.get(timeLimiter.getTimeLimiterConfig().getTimeoutDuration(), TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            if(timeLimiter.getTimeLimiterConfig().cancelRunningFuture()){
                future.cancel(true);
            }
            throw e;
        }
    };
}

It allows to limit the duration of a Future and decorate the call of the future with a CircuitBreaker.

TimeLimiter timeLimiter = TimeLimiter.ofDefaults("id");
Callable<String> decoratedFuture = TimeLimiter.decorateFuture(timeLimiter,
    () -> CompletableFuture.supplyAsync(helloWorldService::returnHelloWorld));

Callable<String> callable = CircuitBreaker
    .decorateCallable(circuitBreaker, decoratedFuture);

Of course it's not a reactive, event-driven solution, but that's not possible with Java8 Futures.
That way we don't have to create a Future Proxy or return a different Future type. The Callable return type just allows the user to do exactly what we want.

[RobWin]

@maarek @storozhukBM What do you think about my suggestion?

[maarek]

@RobWin So the input being: Future or Supplier and maybe output for each being: Callable or Runnable?

It fits nicely with the input for CircuitBreaker without more work. I am fairly pragmatic so if it works as is with the rest of Resilience4j then I am all for it.

[RobWin]

Does it add any value for your use case?

[RobWin]

@maarek Would you like to add this and write some tests?
I would like to work on issues #148 and #51

[coveralls]

Coverage increased (+0.09%) to 92.764% when pulling a245f39 on maarek:timeout-decorator into 2f71963 on resilience4j:master.

[RobWin]

after a configurable duration, or?

[coveralls]

Coverage decreased (-0.9%) to 91.779% when pulling e798154 on maarek:timeout-decorator into 2f71963 on resilience4j:master.