Parallel Hystrix. Improving Distributed Application Performance
About a year ago, our team rewrote the backend of a little-known application with 5 million users using the “latency and fault tolerance” Hystrix . This made it possible to significantly increase the reliability of the application during crashes or delays in downstream systems (there are about 10 of them, which is not a lot for a serious system), it provided a wonderful tool (Hystrix Dashboard) for monitoring the load of external systems (now we know who the brake is), it allowed us to optimize the sizes of various pools in the application. However, the problem of the lengthy processing of individual heavy queries remained, the solution of which this article is devoted to.
All the goodies were achieved with minimal effort and very small code changes - Hystrix provides a synchronous programming model, as a result, only a small http client call code was transferred from application streams to hystrix thread pools with as few problems as possible (I do not believe that these problems can be avoid completely).
Consider a simple example - the CommandHelloWorld command, which retrieves very important data and client code that uses it with a delay.
It turned out to be a very simple solution with controlled runtime. Of course, over time, any initial advantage will turn into disadvantages. There are clients who need to provide a list of 200 elements, for the formation of each element you need to contact the external system once (or more than once) (sometimes more than one). Not all external systems support batches, not every code is so easy to parallelize without errors - after spending 20 ms per trip to the external system, we get a hefty 4 seconds for the entire request, during which the user suffered, and the pool of tomcat threads did not receive its stream back, at the same time, the histrix thread pool was practically not filled with requests. It's time to fix it.
Almost the entire application was built according to the following pattern :
Of course, theoretically, there is a parallel operation in java stream, but in fact it has extremely poor management of the thread pools in which the work will be performed, so it was decided to refuse to use it. It would be necessary to have some kind of understandable solution that would not break the original Pattern and not break the mosk to the team. As a result, 2 working solutions were proposed - based on reactivex.io and an add-on for java stream.
A demo can be found on the github , all working examples are placed in tests.
It should be noted that inside the histrix uses the reactive (this is a term, not the fact that it is fast) library of "asynchronous data sources" reactivex.io . The article is not a guide to this unapproachable topic, but one elegant solution will be shown. Unfortunately, I am not familiar with the well-established Russian translation of the term observable, therefore I will call it the source. And so, wanting not to break the Pattern , we will act in this way:
On the one hand, the last operation is not as simple as it seems, but since we know that all external connections are controlled by the histrix, and therefore are strictly limited in time, and the application’s own code does nothing really, we can assume that the data source for the specified period the timeout must raise the value or the exception will be thrown by the histrix before the timeout expires (1 second). Therefore, we will use a simple collector function:
We will create two commands and sources using the native API:
And so, consider a simple case of processing a list of their 6 elements:
Everything works out remarkably in parallel for approximately 500ms, the program logs confirm the simultaneous use of histrix streams. As a side effect, items are randomly listed in the list. This is the price of reactivity.
Let's try to increase the size of the list to 49 - and get a regular bummer:
Here is such a funny log exhaust:
The fact is that by default, the histrix creates a pool of 10 threads with a zero queue. When subscribing to a source, all its elements are emitted very quickly, instantly filling the entire pool, even the creation of a cold source did not help. We do not need such a histrix. It is required to reasonably limit the greed of one client.
The solution turned out to be quite simple:
The code turned out to be surprisingly compact, although it took a long time to understand its work:
Let's see a fragment of the logs:
It can be seen from the application logs that the process of launching new teams has stopped after starting the 7th team. The result was a very simple code modification, which allows you to run queries to external systems with the desired degree of parallelism without clogging the entire pool.
A rare person understand rx. Even if he says the opposite - ask to write the code above in your own words. But java 8 already has stream and future, the histrix seems to be able to work with the future as a native, let's try to start parallel processing with their help. We will create the future like this:
We are trying to process a list of 49 elements:
and again a familiar problem.
In this case, the order of creating the commands is controlled by the splitter, that is, it is generally completely opaque and dangerous. We will try to solve the problem in 2 stages, while maintaining a fairly familiar technique for working with the stream:
1) map the source data stream into the stream futures. Moreover, with the control of really launched tasks. For this, we need an intermediate executor with the degree of parallelism we need
2) turn the stream of futures into the stream of values. keeping in mind that each futures is ultimately executed by histrix and has a guaranteed lead time.
To implement the first step, we will make a separate parallelWarp operator for converting a user-defined function, for the second step we will have to write a waitStream function that receives and returns streams:
the record turned out almost familiar to users of streams. Let's see what's under the hood, this is the last piece of code for today:
The waitStream method is very simple, only error handling ruined it. The parallelWarp operator is extremely simple and probably has a special name among adherents of functional programming. New histrix commands are created only by the internal executor, which has the degree of parallelism we need. Pruflink:
With this approach, we needed an additional pool of threads for each pool of histrix threads, but the output list kept order. Which approach will benefit in the application - time will tell.
I repeat that all examples can be seen in tests on the github . I will be glad to the famous habro-criticism.
Why it was easy to insert a histrix
All the goodies were achieved with minimal effort and very small code changes - Hystrix provides a synchronous programming model, as a result, only a small http client call code was transferred from application streams to hystrix thread pools with as few problems as possible (I do not believe that these problems can be avoid completely).
Consider a simple example - the CommandHelloWorld command, which retrieves very important data and client code that uses it with a delay.
public class CommandHelloWorld extends HystrixCommand {
@Override
protected String run() throws InterruptedException {
// вот тут происходит очень важное обращение во внешнюю систему
// все это работает в пуле потоков хистрикса
Thread.sleep(random.nextInt(500));
return "Hello " + name + "!";
}
}
// клиентский код
String executeCommand(String str) {
// вызов зависает на время, не большее чем установленный таймаут.
return new CommandHelloWorld(str).execute();
}
It turned out to be a very simple solution with controlled runtime. Of course, over time, any initial advantage will turn into disadvantages. There are clients who need to provide a list of 200 elements, for the formation of each element you need to contact the external system once (or more than once) (sometimes more than one). Not all external systems support batches, not every code is so easy to parallelize without errors - after spending 20 ms per trip to the external system, we get a hefty 4 seconds for the entire request, during which the user suffered, and the pool of tomcat threads did not receive its stream back, at the same time, the histrix thread pool was practically not filled with requests. It's time to fix it.
How do we arrange the histrix
Almost the entire application was built according to the following pattern :
- we get a list with the source data
- form stream from it
- we filter / map, with a histrix including
- collect from stream list
Of course, theoretically, there is a parallel operation in java stream, but in fact it has extremely poor management of the thread pools in which the work will be performed, so it was decided to refuse to use it. It would be necessary to have some kind of understandable solution that would not break the original Pattern and not break the mosk to the team. As a result, 2 working solutions were proposed - based on reactivex.io and an add-on for java stream.
A demo can be found on the github , all working examples are placed in tests.
Option 1. Add reactivity
It should be noted that inside the histrix uses the reactive (this is a term, not the fact that it is fast) library of "asynchronous data sources" reactivex.io . The article is not a guide to this unapproachable topic, but one elegant solution will be shown. Unfortunately, I am not familiar with the well-established Russian translation of the term observable, therefore I will call it the source. And so, wanting not to break the Pattern , we will act in this way:
- we get a list with the source data
- form a source from it
- filter / map source, with histrix including
- collect from the source list
On the one hand, the last operation is not as simple as it seems, but since we know that all external connections are controlled by the histrix, and therefore are strictly limited in time, and the application’s own code does nothing really, we can assume that the data source for the specified period the timeout must raise the value or the exception will be thrown by the histrix before the timeout expires (1 second). Therefore, we will use a simple collector function:
static List toList(Observable observable) {
return observable.timeout(1, TimeUnit.SECONDS).toList().toBlocking().single();
}
We will create two commands and sources using the native API:
/**
* создает горячую хистриксную команду-источник
*/
static Observable executeCommand(String str) {
LOG.info("Hot Hystrix command created: {}", str);
return new CommandHelloWorld(str).observe();
}
/**
* создает холодную хистриксную команду-источник
*/
static Observable executeCommandDelayed(String str) {
LOG.info("Cold Hystrix command created: {}", str);
return new CommandHelloWorld(str).toObservable();
}
And so, consider a simple case of processing a list of their 6 elements:
public void testNaive() {
List source = IntStream.range(1, 7).boxed().collect(Collectors.toList());
Observable observable = Observable.from(source)
.flatMap(elem -> executeCommand(elem.toString()));
toList(observable).forEach(el ->LOG.info("List element: {}", el));
}
Everything works out remarkably in parallel for approximately 500ms, the program logs confirm the simultaneous use of histrix streams. As a side effect, items are randomly listed in the list. This is the price of reactivity.
Let's try to increase the size of the list to 49 - and get a regular bummer:
public void testStupid() {
List source = IntStream.range(1, 50).boxed().collect(Collectors.toList());
Observable observable = Observable.from(source)
.flatMap(elem -> executeCommand(elem.toString()));
toList(observable).forEach(el ->LOG.info("List element: {}", el));
}
Here is such a funny log exhaust:
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 1
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 2
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 3
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 4
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 5
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 6
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 7
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 8
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 9
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 10
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 11
[hystrix-ExampleGroup-5] INFO org.silentpom.CommandHelloWorld - Command start: 5
[hystrix-ExampleGroup-2] INFO org.silentpom.CommandHelloWorld - Command start: 2
[hystrix-ExampleGroup-9] INFO org.silentpom.CommandHelloWorld - Command start: 9
[hystrix-ExampleGroup-8] INFO org.silentpom.CommandHelloWorld - Command start: 8
[hystrix-ExampleGroup-4] INFO org.silentpom.CommandHelloWorld - Command start: 4
[hystrix-ExampleGroup-1] INFO org.silentpom.CommandHelloWorld - Command start: 1
[hystrix-ExampleGroup-6] INFO org.silentpom.CommandHelloWorld - Command start: 6
[hystrix-ExampleGroup-10] INFO org.silentpom.CommandHelloWorld - Command start: 10
[hystrix-ExampleGroup-3] INFO org.silentpom.CommandHelloWorld - Command start: 3
[main] ERROR org.silentpom.RxHystrixTest - Ooops
com.netflix.hystrix.exception.HystrixRuntimeException: CommandHelloWorld could not be queued for execution and no fallback available.
The fact is that by default, the histrix creates a pool of 10 threads with a zero queue. When subscribing to a source, all its elements are emitted very quickly, instantly filling the entire pool, even the creation of a cold source did not help. We do not need such a histrix. It is required to reasonably limit the greed of one client.
The solution turned out to be quite simple:
- For starters, we will not use flatMap so that subscribing to the source does not cause the creation of all commands. And create a double source using the map method.
- group these sources by the window method - we get a triple source!
- it's time to strictly organize your triple sources - release them one by one with the concatMap method
- each double source is computed in parallel by the flatMap method
The code turned out to be surprisingly compact, although it took a long time to understand its work:
public void testWindow() {
List source = IntStream.range(1, 50).boxed().collect(Collectors.toList());
Observable observable = Observable.from(source)
.map(elem -> executeCommandDelayed(elem.toString()))
.window(7)
.concatMap(window -> window.flatMap(x -> x));
toList(observable).forEach(el ->LOG.info("List element: {}", el));
}
Let's see a fragment of the logs:
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 20
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 21
[hystrix-ExampleGroup-3] INFO org.silentpom.CommandHelloWorld - Command start: 3
[hystrix-ExampleGroup-7] INFO org.silentpom.CommandHelloWorld - Command start: 7
[hystrix-ExampleGroup-5] INFO org.silentpom.CommandHelloWorld - Command start: 5
[hystrix-ExampleGroup-4] INFO org.silentpom.CommandHelloWorld - Command start: 4
[hystrix-ExampleGroup-2] INFO org.silentpom.CommandHelloWorld - Command start: 2
[hystrix-ExampleGroup-6] INFO org.silentpom.CommandHelloWorld - Command start: 6
[hystrix-ExampleGroup-1] INFO org.silentpom.CommandHelloWorld - Command start: 1
[hystrix-ExampleGroup-3] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 3
[hystrix-ExampleGroup-6] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 6
[hystrix-ExampleGroup-2] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 2
[hystrix-ExampleGroup-7] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 7
[hystrix-ExampleGroup-1] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 1
[hystrix-ExampleGroup-5] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 5
[hystrix-ExampleGroup-4] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 4
[hystrix-ExampleGroup-8] INFO org.silentpom.CommandHelloWorld - Command start: 8
[hystrix-ExampleGroup-5] INFO org.silentpom.CommandHelloWorld - Command start: 11
[hystrix-ExampleGroup-1] INFO org.silentpom.CommandHelloWorld - Command start: 12
It can be seen from the application logs that the process of launching new teams has stopped after starting the 7th team. The result was a very simple code modification, which allows you to run queries to external systems with the desired degree of parallelism without clogging the entire pool.
Option 2. See reactivity! We have executives
A rare person understand rx. Even if he says the opposite - ask to write the code above in your own words. But java 8 already has stream and future, the histrix seems to be able to work with the future as a native, let's try to start parallel processing with their help. We will create the future like this:
// просим хистрикс запустить команду и вернуть фьючу
static Future executeCommandDelayed(String str) {
LOG.info("Direct Hystrix command created: {}", str);
return new CommandHelloWorld(str).queue();
}
// по старинке синхронно вызываем хистрикс, все в ручную
static String executeCommand(String str) {
LOG.info("Direct Hystrix command created: {}", str);
return new CommandHelloWorld(str).execute();
}
We are trying to process a list of 49 elements:
public void testStupid() {
IntStream.range(1, 50).boxed().map(
value -> executeCommandDelayed(value.toString())
).collect(Collectors.toList())
.forEach(el -> LOG.info("List element (FUTURE): {}", el.toString()));
}
and again a familiar problem.
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 2
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 3
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 4
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 5
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 6
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 7
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 8
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 9
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 10
[main] INFO org.silentpom.stream.ParallelAsyncServiceTest - Direct Hystrix command created: 11
[main] ERROR org.silentpom.stream.ParallelAsyncServiceTest - Ooops
com.netflix.hystrix.exception.HystrixRuntimeException: CommandHelloWorld could not be queued for execution and no fallback available.
In this case, the order of creating the commands is controlled by the splitter, that is, it is generally completely opaque and dangerous. We will try to solve the problem in 2 stages, while maintaining a fairly familiar technique for working with the stream:
1) map the source data stream into the stream futures. Moreover, with the control of really launched tasks. For this, we need an intermediate executor with the degree of parallelism we need
2) turn the stream of futures into the stream of values. keeping in mind that each futures is ultimately executed by histrix and has a guaranteed lead time.
To implement the first step, we will make a separate parallelWarp operator for converting a user-defined function, for the second step we will have to write a waitStream function that receives and returns streams:
public void testSmart() {
service.waitStream(
IntStream.range(1, 50).boxed().map(
service.parallelWarp(
value -> executeCommand(value.toString())
)
)
).collect(Collectors.toList())
.forEach(el -> LOG.info("List element: {}", el));
}the record turned out almost familiar to users of streams. Let's see what's under the hood, this is the last piece of code for today:
// по традиции threadSize = 7
public ParallelAsyncService(int threadSize) {
ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
.setNameFormat("parallel-async-thread-%d").build();
// создаем промежуточный экзекутор, который будет создавать команды хистрикса
executorService = Executors.newFixedThreadPool(threadSize, namedThreadFactory);
}
/**
* Maps user function T -> Ret to function T -> Future. Adds task to executor service
* @param mapper user function
* @param user function argument
* @param user function result
* @return function to future
*/
public Function> parallelWarp(Function mapper) {
return (T t) -> {
LOG.info("Submitting task to inner executor");
Future future = executorService.submit(() -> {
LOG.info("Sending task to hystrix");
return mapper.apply(t);
});
return future;
};
}
/**
* waits all futures in stream and rethrow exception if occured
* @param futureStream stream of futures
* @param type
* @return stream of results
*/
public Stream waitStream(Stream> futureStream) {
List> futures = futureStream.collect(Collectors.toList());
// wait all futures one by one.
for (Future future : futures) {
try {
future.get();
} catch (InterruptedException e) {
throw new RuntimeException(e);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof RuntimeException) {
throw (RuntimeException) cause;
}
throw new RuntimeException(e);
}
}
// all futures have completed, it is safe to call get
return futures.stream().map(
future -> {
try {
return future.get();
} catch (Exception e) {
e.printStackTrace();
return null; // не должен вызываться вообще
}
}
);
The waitStream method is very simple, only error handling ruined it. The parallelWarp operator is extremely simple and probably has a special name among adherents of functional programming. New histrix commands are created only by the internal executor, which has the degree of parallelism we need. Pruflink:
main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 18
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 19
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 20
[main] INFO org.silentpom.RxHystrix - Cold Hystrix command created: 21
[hystrix-ExampleGroup-4] INFO org.silentpom.CommandHelloWorld - Command start: 4
[hystrix-ExampleGroup-1] INFO org.silentpom.CommandHelloWorld - Command start: 1
[hystrix-ExampleGroup-2] INFO org.silentpom.CommandHelloWorld - Command start: 2
[hystrix-ExampleGroup-3] INFO org.silentpom.CommandHelloWorld - Command start: 3
[hystrix-ExampleGroup-5] INFO org.silentpom.CommandHelloWorld - Command start: 5
[hystrix-ExampleGroup-6] INFO org.silentpom.CommandHelloWorld - Command start: 6
[hystrix-ExampleGroup-7] INFO org.silentpom.CommandHelloWorld - Command start: 7
[hystrix-ExampleGroup-2] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 2
[hystrix-ExampleGroup-5] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 5
[hystrix-ExampleGroup-3] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 3
[hystrix-ExampleGroup-7] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 7
[hystrix-ExampleGroup-6] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 6
[hystrix-ExampleGroup-4] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 4
[hystrix-ExampleGroup-1] INFO org.silentpom.CommandHelloWorld - Command calculation finished: 1
[hystrix-ExampleGroup-4] INFO org.silentpom.CommandHelloWorld - Command start: 11
[hystrix-ExampleGroup-6] INFO org.silentpom.CommandHelloWorld - Command start: 12
[hystrix-ExampleGroup-8] INFO org.silentpom.CommandHelloWorld - Command start: 8
[hystrix-ExampleGroup-7] INFO org.silentpom.CommandHelloWorld - Command start: 13
[hystrix-ExampleGroup-9] INFO org.silentpom.CommandHelloWorld - Command start: 9
With this approach, we needed an additional pool of threads for each pool of histrix threads, but the output list kept order. Which approach will benefit in the application - time will tell.
I repeat that all examples can be seen in tests on the github . I will be glad to the famous habro-criticism.
