Backpressure support

[mattrjacobs]

RxJava supports backpressure. I don't believe that Hystrix currently participates in backpressure, as it is creating an Observable from scratch by wrapping logic around either a scheduled HystrixCommand.run() or a HystrixObservableCommand.construct().

I need to write some code examples to see what happens when backpressure over network boundaries happens, and Hystrix is involved in the data flow.

Also note that features like semaphore/thread-pool rejection are analogous to Observable.onBackpressureDrop.

This is marked as 'discussion' as I am not sure what the relationship between Hystrix and backpressure is yet. If anyone on the community has ideas, or better yet code, I'd love your input.

[agentgt]

Well one of the problem is that Hystrix isn't really a streaming/pub/sub/reactive library but more of a command/request/reply/blocking command pattern. That is Hystrix Observables are generally just Single (in theory you could do streaming). Generally backpressure is not needed when you want a single item.

The problem when you start doing streaming is resource management. Request/Reply doesn't have this problem because once the reply is filled you are done and resources can be cleaned up (there is only one endpoint you have to worry about not responding). Thus if you want to start investigating I think that resource management is starting point. I would imagine this would require a significant addition to Hystrix to manage this (albeit I would be highly interested in this feature and in theory you could use the same principles ie semphores and timeouts).

TLDR You are now worrying about two endpoints instead of one (ie in reactive-stream parlance the subscriber needs to call request along with the publisher actually fulfilling that request).

A library to investigate that sort of deals with the whole reactive-streams management rather well is reactor ( I believe rxjava2 even borrowed some code from them).

[mattrjacobs]

I generally agree with your points for HystrixCommand. Have you seen HystrixObservableCommand? This is intended to wrap calls like async HTTP calls, that may return streams (like RxNetty). In a case where we do I/O that supports backpressure, I'd like Hystrix to not get in the way of expressing that backpressure.

I'll check out reactor - thanks for the link!

[smaldini]

Ha sorry the snapshot site is pending update (goes down when update takes place for a couple minutes). Meanwhile there is still http://projectreactor.io & https://github.com/reactor/reactor-core.

Nevertheless I would be interested in providing any assistance if you guys want to return something like HystrixFluxCommand (Flux being our Rx lite Publisher implementation for N data) and HystrixMonoCommand (Mono being our Rx lite Publisher implementation for 1 data).

FYI
Our subscribeOn equivalent -> publishOn : http://next.projectreactor.io/core/docs/api/reactor/core/publisher/Flux.html#publishOn-java.util.concurrent.Callable-

[agentgt]

As far as my understanding goes the HystrixObservableCommand has Hystrix only participate in the subscription thread (ie subscribeOn) which is good and bad. Its good for fake streams that are blocking but bad for long open reactive push streams.

Its bad because in the reactive-streams model the subscribe phase should be nonblocking. That is if your blocking on subscribe your doing it wrong (see my recent question here reactive-streams/reactive-streams-jvm#276 ).

So using a thread ExecutionIsolationStrategy for a true reactive command is a complete waste. Its also fairly risky because you might want to have thousands of open streams (ie some chat Netty like server). What is useful are probably semaphores at maybe a fairly high count in this case.

So yeah Hystrix does get in the way if you naive and use the the thread policy.

My concern isn't that Hystrix has an impact but rather it probably has little impact. For example I would be interested in having some fault tolerance to prevent hundreds of thousands of open streams and from my quick pass at the code I don't think Hystrix does anything in that regard. Hystrix will I suppose prevent the subscription (it doesn't prevent the total subscriptions but rather too many subscribing at once which again is not useful really) but what about the cleanup of all those open streams and/or timeout?

In the AbstractCommand code it appears all the current timeouts, semaphores, etc are all based on the publisher being naughty. The issue though with backpressure fault tolerance is what if the subscribers stops requesting for more data or become unresponsive? Unlike RxJava 1.0 the reactive-streams model and thus RxJava 2.0 put a lot more demands on making proper subscribers (there is a discussion somewhere in the reactive-streams issues about this)...

TLDR if your talking about backpressure you have to remember the subscriber requests for more data and is the one that is more likely to do something bad. Hystrix only protects you from a bad publisher.

EDIT I guess for rxjava 1.0 the publisher and subscriber generally communicate on the same thread (that is request(n) is typically done on the subscribe. So long as you don't observeOn than Hystrix would work fine with backpressure. But if you observerOn and rely on request Hystrix will probably not do what you want (ie it will timeout for a client) as the timeout appears to be for the whole stream and not for each request (see com.netflix.hystrix.AbstractCommand.HystrixObservableTimeoutOperator.call(...).new Subscriber() {...}.isNotTimedOut()). Actually its even worse because its only for one item 😮

[agentgt]

I tested Hystrix with a couple unit tests with backpressure and it appears it doesn't support it at all.

There is some RxJava operator that overrides the Producer.request to always use Long.MAX_VALUE on the returned Observable.

        TestHystrixObservableCommand<Integer> command = getCommand(isolationStrategy, executionResult, 50, AbstractTestHystrixCommand.FallbackResult.UNIMPLEMENTED, 1000000);
        long observedCommandDuration = 0;

            final CountDownLatch latch = new CountDownLatch(1);
            command.toObservable().subscribe(new Subscriber<Integer>() {
                @Override
                public void onStart() {
                    request(1);
                }
                @Override
                public void onCompleted() {
                    latch.countDown();
                }

                @Override
                public void onError(Throwable e) {
                    throw new UnsupportedOperationException("onError is not yet supported");
                }

                @Override
                public void onNext(Integer t) {
                    out.println("" + t);
                    //N.B. no future request but Hystrix ignores this.
                }
            });
            latch.await(5, TimeUnit.MINUTES);

Consequently your probably better off not even getting Hystrix involved if your using real reactive streams and just let the endpoints deal with timeout/circuit breaking etc (ie nioSelector to nioSelector aka Netty deals with the overload itself).

I sort of googled around to see what sort of fault tolerance libraries there are for streams and I decided that there really isn't a Hystrix of streams but this probably should be a separate library because so much of this library assumes Single like behavior and not streams.

[anton-antonov]

Here is the good example why Hystrix commands should support Backpressure:

Example with THREAD execution strategy but it is quite the same for SEMAPHORE strategy

    private class NumberHystrixCommand extends HystrixCommand<Integer> {

        private int number;

        public NumberHystrixCommand(int number) {
            super(Setter.withGroupKey(HystrixCommandGroupKey.Factory.asKey("NumberThread"))
                .andCommandPropertiesDefaults(HystrixCommandProperties.Setter()
                    .withExecutionIsolationStrategy(HystrixCommandProperties.ExecutionIsolationStrategy.THREAD)
                    .withExecutionTimeoutInMilliseconds(1000)
                    .withRequestCacheEnabled(false))
                .andThreadPoolPropertiesDefaults(HystrixThreadPoolProperties.Setter()
                    .withCoreSize(10)
                    .withMaxQueueSize(10)
                    .withQueueSizeRejectionThreshold(10)));
            this.number = number;
        }

        @Override
        protected Integer run() throws Exception {
            //emulate networking
            Thread.sleep(100);
            return this.number + 100;
        }
    }

    @Test
    public void testBackpressure() throws Exception {
        Observable.range(1, 1000)
            .map(NumberHystrixCommand::new)
            .flatMap(cmd -> cmd.toObservable())
            .doOnNext(System.out::println)
            .toBlocking()
        .subscribe();
    }

I believe this is a quite normal use-case for cmd.toObservable(). But unfrotunately, and obvious, it fails with Caused by: java.util.concurrent.RejectedExecutionException: Rejected command because thread-pool queueSize is at rejection threshold.

For this example there's easy workaround:

@Test
    public void testBackpressure() throws Exception {
        AbstractConfiguration configInstance = ConfigurationManager.getConfigInstance();
        int coreSize = configInstance.getInteger("hystrix.threadpool.NumberThread.coreSize", 10);
        int queueSize = configInstance.getInteger("hystrix.threadpool.NumberThread.queueSizeRejectionThreshold", 10)  - 1;
        int parallelism = coreSize + queueSize;
        Observable.range(1, 1000)
            .map(NumberHystrixCommand::new)
            .flatMap(cmd -> cmd.toObservable(), parallelism)
            .doOnNext(System.out::println)
            .toBlocking()
            .subscribe();
    }

There's a more interesting case with HystrixCollapser where parallel executions collapsed to single batch command. For HystrixCollapser use-case workaround is more complicated but still quite easy (assuming that we have NumberCollapser in the example with maxRequestsInBatch=10):

@Test
    public void testBackpressure() throws Exception {
        AbstractConfiguration configInstance = ConfigurationManager.getConfigInstance();
        int coreSize = configInstance.getInteger("hystrix.threadpool.NumberThread.coreSize", 10);
        int queueSize = configInstance.getInteger("hystrix.threadpool.NumberThread.queueSizeRejectionThreshold", 10)  - 1;
        int batchSize = configInstance.getInteger("hystrix.collapser.NumberCollapser.maxRequestsInBatch", 10);
        int parallelism = (coreSize + queueSize) * batchSize;
        Observable.range(1, 1000)
            .map(NumberHystrixCommand::new)
            .flatMap(cmd -> cmd.toObservable(), parallelism)
            .doOnNext(System.out::println)
            .toBlocking()
            .subscribe();
    }

I think that such "parallelism" could be supported out-of-the-box for source Observables, which support Backpressure (like Observable.from(Iterable), range(...), etc. ).
It won't affect blocking usages as soon as it will request(N), where N > 1, and blocking call new NumberHystrixCommand(10).execute() produces a single value always. With a parallel execution of blocking calls (in multiple threads, aka web requests) it will probably exceed the limits, and in this case, it's OK to throw RejectedExecutionException, but it does already, thus nothing to be done here.

[mattrjacobs]

I followed your first 2 examples, and both make sense to me. I didn't quite follow your third example - are you asking for something new in the source observable produced by Hystrix or something else?

[anton-antonov]

@mattrjacobs Third example is about collapser use-case when we're using HystrixCollapser (not HystrixCommand) and, in that case, "parralelism" is multiplied by batchSize.
Thus it possible to call much more "top" collapser commands while "internal" batch commands still satisfy thread pool rejection policies

[agentgt]

@anton-antonov I am sorry but I too might still be confused.

I suppose I somewhat understand in that you want to be able to create a stream of commands with backpressure but this is an entirely different feature than Hystrix supporting a stream of objects from a command with backpressure support. And I think you are right given the rx.Single nature of the library that your last example is more likely.

That is you want something like a HystrixCommand<HystrixCommand<Integer>> with the parent command some how being smart about the inner command(s) which would be fairly complicated.

I still think some new sort of interface (or even library) something like HystrixCommandStream<Integer> might be better and would have entirely different semantics on thread pool creation and/or rejection... it might even block.

[anton-antonov]

@agentgt Thanks. I think, I've got the idea. In order to support backpressure, we'd need to build an outer wrapper around HystrixCommand which actually "will know" about command configuration and won't allow calling a child command more than it's allowed by thread pool policies.

[agentgt]

@anton-antonov Bingo. And I think that it really should be a separate library. One that helps you write correct and fault tolerant reactive-stream publishers/subscribers.

To be even more candid I would seriously consider @Deprecating toObservable() having new methods that return rx.Single aka a cold Promise. Furthermore I would also deprecate HystrixObservableCommand as I am not sure it really provides any value. I would love @mattrjacobs input on how they use that at Netflix but it seems like an unnecessary wrapper around a true streaming Observable (for example from a Netty publisher). Just use the Observable directly.

Even more so if Hystrix were to get rewritten from scratch as a Command library it should really not have its entire core depend on rx.Java. It is the wrong abstraction for the Command pattern aka Request/Reply. In large part I think (and I'm probably wrong) Chris just chose Observable because there wasn't any good compose-able cold promise libraries that were good in Java at the time.

TLDR it is really hard and confusing to developers to merge Request/Reply with Pub/Sub in one library and particularly in one interface. The cognitive load and confusion is just too high.

[mattrjacobs]

@anton-antonov Yeah, I think that's right.

@agentgt We're still in the early days of transitioning to non-blocking architectures within Netflix, so it's hard to see how it will play out precisely. I do agree that rx.Observable is overkill for the request-reply case.

The intended use case of HystrixObservableCommand is to wrap work that's already async (a Netty HTTP client is a good example). In this case, we're not tying up a thread, so the thread-pool isolation isn't used. I still think that concepts like fallback, timeout, semaphore rejection are potentially useful.

One idea I'm excited about (but haven't started work on yet) is #1174. I think that forcing an entire object upon consumers and then making them choose how to configure everything about it is not ideal. If they could pick and choose composable behaviors, the cognitive load would be lighter. I think we're saying the same thing.

Also, I agree with your point about distinguishing the type of work that is wrapped by Hystrix. I think the 3 major cases I've seen are:

1. Request-Reply with Blocking I/O
2. Request-Reply with Nonblocking I/O
3. Request-Stream (or subscription) with Nonblocking I/O

Case 1 is handled today by HystrixCommand. It exposes T execute(), j.u.c.Future queue(), and rx.Observable toObservable(). It lacks a single-valued, composable Future (j.u.c.CompletableFuture, rx.Single and reactor.core.publisher.Mono come to mind).

Case 2 is handled today by HystrixObservableCommand. It exposes rx.Observable toObservable(). It lacks the same single-valued Future as Case 1. HystrixObservableCommand wraps an rx.Observable. It's likely more appropriate to wrap a single-valued Future. That's tracked by #602

Case 3 is handled today by HystrixObservableCommand. Here, it's truly appropriate to wrap an rx.Observable and get one out the other side. In this case, the semantics of everything is different. Should there be timeouts between emissions, only between subscription and the first? I don't have enough experience with this yet to make a decision. Any thoughts you have here would be helpful