Support basic Java Executor interface

[paulius-p]

Currently you can provide either ExecutorService, ScheduledExecutorService or implement custom Failsafe Scheduler.
It would be good if it would also accept java.util.concurrent.Executor interface since it is a common interface returned by other libs. My current use case is in gRPC, when splitting context for multi-threading, and gRPC methods #fixedContextExecutor return base Executor.

[whiskeysierra]

Executor is missing the notion of Future which I guess is problematic.

[jhalterman]

@whiskeysierra is right on - Failsafe wouldn't be able to return execution results from an Executor or determine if a result is a failure. You could use an Executor to run things that don't return a result though. The question then is what to do if Failsafe is asked to get something that returns a result... perhaps just not use the Executor in that case?

[paulius-p]

Yes, I understand the semantic predicament, but you could have a wrapper similarly like you have for the ExecutorService and use the CompletableFuture.supplyAsync(Supplier, Executor) to return the value?

[Tembrel]

Yes, I understand the semantic predicament, but you could have a wrapper similarly like you have for the ExecutorService and use the CompletableFuture.supplyAsync(Supplier, Executor) to return the value?

Seems reasonable to me. At the very least, couldn't the use of ExecutorService.submit in lines like this be replaced by something like:

promise.delegate = es instanceof ExecutorService ?
    ((ExecutorService) es).submit(completingCallable) :
    supplyAsync(callableToSupplier(completingCallable), es);

when es is an Executor but not an ExecutableService? I realize that I'm glossing over an important detail with the following simple implementation of callableToSupplier, but the general idea seems feasible:

static <T> Supplier<T> callableToSupplier(Callable<? extends T> callable) {
    return () -> {
	try {
	    return callable.call();
        } catch (RuntimeException ex) {
            throw ex;
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
    };
}

[whiskeysierra]

(@Tembrel The instanceof in a catch clause can usually be written shorter and less complicated using a second catch clause.)

[Tembrel]

@whiskeysierra - oops, thanks, updated comment.

This was just a sketch meant to provoke discussion of whether it is worth offering an overload of FailsafeExecutor.with that takes an Executor. I wouldn't implement it this way IRL but wanted to suggest that it would be possible. (For one thing, I wouldn't do the instanceof ExecutorService check so late in the game; the result of that test could be determined at construction time.)

The question remains: Is this worth doing?

[jhalterman]

@paulius-p I'm curious, when creating a fixedContextExecutor, what Executor do you pass in?

@Tembrel thanks for sparking some ideas with this.

If we were to support configuring an Executor, I think it would need to be separate from and probably in addition to a Scheduler/ExecutorService. The Executor would basically just wrap Runnables/Suppliers and would amount to doing something like this:

public CheckedRunnable wrapWithExecutor(Runnable r, Executor executor) {
  return () -> {
    executor.execute(() -> {
      try {
        r.run();
      } catch (RuntimeException ex) {
        throw ex;
      } catch (Exception ex) {
        throw new RuntimeException(ex);
      }
    });
  };
}

Failsafe.with(policy).run(wrapWithExecutor(this::connect, executor));

@paulius-p Is that basically how you think this should work - simple wrapping? Maybe something like:

Failsafe.with(policy).with(executor).run(this::connect);

...but I'm not sure if this use case is too niche and the workaround too straightforward. Input welcome...

[paulius-p]

@jhalterman Yes, that's similar to what I had in mind.

For the fixedContextExecutor, in my case, I would pass in some thread pool from Executors.newFixedThreadPool(int). But in the current approach, since I cannot pass the wrapped fixedContextExecutor, I'm calling the action from within the gRPC context directly in the action supplier block:

Failsafe.with(retryPolicy)
  .with(executorService)
  .getStageAsync(() -> forkedContext.call(this::someFutureAction));

So, from verbosity perspective it's not that bad, but supporting Executor directly would reduce it slightly.

[Fadelis]

Hey, looking at it again, it's not that hard to get the Future from Executor. Java already has a way for that and it's actually used in ExecutorService impl - using FutureTask.
I've taken a quick crack at it, and it seems pretty straightforward and shouldn't break any existing features, as all ExecutorService implement Executor - Fadelis@898f961

Maybe you could have a look at this again? Thanks!

[jhalterman]

@Fadelis I pushed a PR that is similar to your suggestion: #281. Let me know what you think.

[Fadelis]

@Fadelis I pushed a PR that is similar to your suggestion: #281. Let me know what you think.

@jhalterman Yes, this should work as well 👍 Thanks for looking into this!

[jhalterman]

Closed by #281.

[jhalterman]

This is fixed in 2.4.2.

[jhalterman]

@paulius-p There may be some additional improvements I'd like to make with this feature. Initially I thought that the Executor would only be used for async executions (runAsync, getAsync, etc), but it occurred to me that you may have wanted it to be called for sync executions as well (run, get). Did you intend for this to be used with sync executions also?

[paulius-p]

@paulius-p There may be some additional improvements I'd like to make with this feature. Initially I thought that the Executor would only be used for async executions (runAsync, getAsync, etc), but it occurred to me that you may have wanted it to be called for sync executions as well (run, get). Did you intend for this to be used with sync executions also?

@jhalterman My use case only dealt with the async executions, but I think it would be good if it supported sync executions as well 👍

[jhalterman]

Closed by c1f514f

[jhalterman]

3.0 has been released, which includes this improvement.

[paulius-p]

Closed by c1f514f

Sadly this commit broke this feature. Now the execution is not managed by the provided executor, but just wrapped and passed to the DEFAULT scheduler, where before the passed executor was the underlying scheduler. Not sure about the details why this is happening, but now any delegating actions (ie setting grpc context or spring security context) on the executor are no longer available within the supplied getAsync function.
Impl before this commit was much more lightweight and worked well.

[jhalterman]

@paulius-p Yea, Failsafe 2.4.x allowed an Executor to take the place of an ExecutorService, which was not intended. This allowed getAsync calls to potentially not actually be async, and to instead block whatever thread made the Failsafe call.

The intent with the impl for this feature was that an Executor could be used by itself for sync calls, or in conjunction with a Scheduler/ExecutorService for async calls. So if you want to perform an async call it will actually be async, and if you provide an Executor, it will be called inside the Scheduler/ExecutorService's thread.

Of course, I want to get things working well for you. Can you describe more detail about what Failsafe settings you're using, and where the contextual information is being set? I'm guessing the security context is just being set on a different thread than the execution runs on, so it's not available there. That said, the Executor should be run on the same thread as the Runnable/Supplier. Here's an example of what I mean (in Failsafe 3.1.0):

ThreadLocal<String> threadLocal = new ThreadLocal<>();
Executor executor = runnable -> {
  threadLocal.set("contextual info");
  runnable.run();
};
Failsafe.none().with(executor).runAsync(() -> {
  System.out.println(threadLocal.get());
}).get();

This prints:

contextual info

I'm not sure if this is how you were using things, but maybe you could describe in more detail.

[paulius-p]

@jhalterman Thanks for the reply, though you should be enjoying holidays at this time, this can definitely wait 🎄

It seems it's only affected when using the .getStageAsync method:

    ThreadLocal<String> threadLocal = new ThreadLocal<>();
    Executor executor = runnable -> {
      threadLocal.set("contextual info");
      runnable.run();
    };
    System.out.println(Failsafe.none().with(executor).getStageAsync(() -> {
      System.out.println(threadLocal.get());
      return CompletableFuture.completedFuture("result");
    }).join());

[jhalterman]

@paulius-p It turns out I never added support for Executor with getStageAsync, so I just pushed that now in 8fe9126. It should be supported with all the FailsafeExecutor methods now.

This test case shows what to expect now. Since Executor returns null, Failsafe always treats the supplied CompletionStage as a null result. You can try it out in master and let me know what you think.

[paulius-p]

@jhalterman yes, this partially works now, since it still returns null. Though I'm still not sure what was the reasoning behind returning null value and having all this explicit and complicated handling of Executor, since the functionality from ExecutorService that is used within failsafe is basically a wrapped Executor call (see the initial proposal Fadelis@898f961). Maybe I'm missing something, but seems like currently implemented behaviour supports only partial functionality and is more complex than it could be.

[jhalterman]

@paulius-p So there are two options for approaching this problem:

1. One way is to have the Executor call the user-provided Supplier/Runnable directly, then the result of the Executor is recorded. Of course this result would always be null or an exception. This is what the current Failsafe implementation does.
2. Another way is to have the Executor call a Runnable that wraps not just the user-provided Supplier/Runnable, but also the Failsafe code that records its result. This way the result from the Executor doesn't matter, it's just there for performing side-effects if needed, almost like an event listener. Also, if the Executor throws an exception, it won't be recorded as part of the Failsafe handling since recording a result happens before the Executor completes. This is what your commit does.

I'm guessing you'd prefer the second option? When you say things partially works now, is that because using an Executor returns null currently? Maybe you could share your thoughts on the options.