Aggregations can be bottlenecked on ChildMemoryCircuitBreaker.limit()

[hackerwin7]

In the method of ChildMemoryCircuitBreaker.addEstimateBytesAndMaybeBreak(), the limit() method would do while loop atomically compare and set the used value.

when we test the high qps query to ES-node, the major cost is ChildMemoryCircuitBreaker.limit() which cause to AggregationPhase.preProcess()'s cost is extremely high

If data node receive high qps complicated search request body, and data-node will be busy to create aggregator, and it will lead to SumAggregator.<init>, lead to BigArrays.newDoubleArray() and lead to BigArrays.adjustBreaker(), finally call breaker.addEstimateBytesAndMaybeBreak and ChildMemoryCircuitBreaker.limit()

[elasticmachine]

Pinging @elastic/es-perf (:Performance)

[elasticmachine]

Pinging @elastic/es-search (:Search/Search)

[hackerwin7]

a flame graph, here it is:

the most cost is ChildMemoryCircuitBreaker.limit().

[elasticmachine]

Pinging @elastic/es-analytics-geo (:Analytics/Aggregations)

[elasticmachine]

Pinging @elastic/es-core-infra (:Core/Infra/Circuit Breakers)

[nik9000]

I wonder if we should consolidate how we grab bytes from the breaker. Or if we should grab a big chunk whenever we grab any and then only come back when we need more. Sort of like a TLAB. For the most part BigArrays is designed to oversize its allocations so for big requests they amortize to happening infrequently. But for many small requests we grab breaker byte very frequently.

[nik9000]

@hackerwin7 could you try running this against master or 7.x? Could you share the search and reproduction?

It looks like the search is a terms which contains another terms which contains a sum. I've cut down on the number of aggregator allocations in 7.x and master over at #56487. It may not help if the cardinality of the sum agg has is ultimately pretty low. The flame graph shows that the initial allocation is what is getting wedged which I haven't changed. But there should be fewer allocations.

[hackerwin7]

@nik9000 Thanks for your reply.

@hackerwin7 could you try running this against master or 7.x? Could you share the search and reproduction?

we test it at 7.6.2, it seems that a high QPS of complicated of aggregations search (which maybe build many aggregators), the main cost is create Aggregator and the BigArrays.adjustBreaker(), in the end call ChildMemoryCircuitBreaker.limit().

I've cut down on the number of aggregator allocations in 7.x and master over at #56487

reduce aggregator allocations maybe a way of reduce the race condition of limit() (loop set atomic value), I will test this in master branch and check the effects

I think in high QPS and many much more aggregators search , according to this flame graph, can we reduce the frequency of call to limit() to reduce race condition. maybe we can batched the limit() or addEstimateBytesAndMaybeBreak(), when added bytes reach to a buff size, we do a real loop set atomic value logic.

[danielmitterdorfer]

I wonder if we should consolidate how we grab bytes from the breaker

I agree that the right approach is to reduce the number of calls to the circuit breaker instead of optimizing the circuit breaker's #limit() method which appears to be the bottleneck because of the high number of calls to it (2.6 million calls in the flame graph that you present).

I also think that the issue title mentioning a "race condition" is misleading because - in my understanding - this issue is about inefficient behavior.

[lqbilbo]

I wonder if we should consolidate how we grab bytes from the breaker

I agree that the right approach is to reduce the number of calls to the circuit breaker instead of optimizing the circuit breaker's #limit() method which appears to be the bottleneck because of the high number of calls to it (2.6 million calls in the flame graph that you present).

I also think that the issue title mentioning a "race condition" is misleading because - in my understanding - this issue is about inefficient behavior.

@danielmitterdorfer Hi Daniel, if we add a setting indices.breaker.memory_usage_estimates.enable in ClusterSettings which can control if or not estimating memory usage during aggregation phase dynamically, we can reduce the number of calls of AggregatorBase#addRequestCircuitBreakerBytes and BigArrays#adjustBreaker when the switch is closed. Do you think it's a good way?

For example:

protected long addRequestCircuitBreakerBytes(long bytes) {
        // Only use the potential to circuit break if bytes are being incremented
        if (isMemoryUsageEstimatesEnabled()) {   // we add a switch
            if (bytes > 0) {
                this.breakerService
                    .getBreaker(CircuitBreaker.REQUEST)
                    .addEstimateBytesAndMaybeBreak(bytes, "<agg [" + name + "]>");
            } else {
                this.breakerService
                    .getBreaker(CircuitBreaker.REQUEST)
                    .addWithoutBreaking(bytes);
            }
        }
        this.requestBytesUsed += bytes;
        return requestBytesUsed;
    }

and

void adjustBreaker(final long delta, final boolean isDataAlreadyCreated) {
        if (isMemoryUsageEstimatesEnabled()) {         // we add a switch
            if (this.breakerService != null) {
                CircuitBreaker breaker = this.breakerService.getBreaker(breakerName);
                if (this.checkBreaker) {
                    // checking breaker means potentially tripping, but it doesn't
                    // have to if the delta is negative
                    if (delta > 0) {
                        try {
                            breaker.addEstimateBytesAndMaybeBreak(delta, "<reused_arrays>");
                        } catch (CircuitBreakingException e) {
                            if (isDataAlreadyCreated) {
                                // since we've already created the data, we need to
                                // add it so closing the stream re-adjusts properly
                                breaker.addWithoutBreaking(delta);
                            }
                            // re-throw the original exception
                            throw e;
                        }
                    } else {
                        breaker.addWithoutBreaking(delta);
                    }
                } else {
                    // even if we are not checking the breaker, we need to adjust
                    // its' totals, so add without breaking
                    breaker.addWithoutBreaking(delta);
                }
            }
        }
    }

Because we only remain an abstract function in SearchContext and a volatile variable in BigArrays, and we assign the switch value onto BigArrays's variable only while AggregatorBase initial build up. We have controlled the scale of the influence. Just like:

protected final SearchContext context;

protected AggregatorBase(String name, AggregatorFactories factories, SearchContext context, Aggregator parent,
            List<PipelineAggregator> pipelineAggregators, Map<String, Object> metaData) throws IOException {
        this.name = name;
        this.pipelineAggregators = pipelineAggregators;
        this.metaData = metaData;
        this.parent = parent;
        this.context = context;
        this.breakerService = context.bigArrays().breakerService();
        // if we disable memory usage estimates, bigArrays doesn't need to adjust memory size.
        context.bigArrays().enableMemoryUsageEstimates(context.enableMemoryUsageEstimates());
        ...
    }

And we have tried a lot of test cases. They show that bucket and metrics aggregation in our query phase consume few memory usage but the number of calls is large.

After I reviewed the pull request #46751 and #57042, they still can't fix this problem. So if needed, I'll show you the code.

Wish your reply!

[nik9000]

I'm curious if this is still a big problem in 7.9.

I'm a bit worried that mutating BigArrays would be dangerous because its shared by the system. We could certainly fork a fresh one for each request but at that point it might be cleaner to have the forking process reserve some chunk up front for the request.

I dunno. As much fun as it is to attack bottlenecks it isn't clear to me that this is both still a big bottleneck.

[lqbilbo]

I'm curious if this is still a big problem in 7.9.

I'm a bit worried that mutating BigArrays would be dangerous because its shared by the system. We could certainly fork a fresh one for each request but at that point it might be cleaner to have the forking process reserve some chunk up front for the request.

@nik9000 Thanks for your reply. After I saw your reply, I ran some test cases in 7.9, but as a regret, the new version still doesn't solve the problem. I agree with you that we shouldn't mutate BigArrays, so I put the Settings into BreakerSettings and I put the condition phase: if (this.breakerService != null && this.breakerService.isMemoryUsageEstimatesEnabled()) into BigArrays#adjustBreaker. And isMemoryUsageEstimatesEnabled() is only False when AggregateBase set it. So it will not affect other part of the whole system.

I dunno. As much fun as it is to attack bottlenecks it isn't clear to me that this is both still a big bottleneck.

We really face the problem just as high QPS, the CPU is almost full and it will affect the stability of our Elasticsearch cluster. So we want to attack the bottleneck while facing with such a business scene.

[nik9000]

@lqbilbo. OK. I'm going to add this to my list of things to think about some more too then. I'm not really sold on the idea of not checking the breaker on allocation but its certainly something. I want to play with pre-allocating the a bunch of byes up front and see what that looks like for us. It might not be as big a change as it sounds.... But it'll take me a week or two before I can get to it what with Thanksgiving here in the US and me spending most of my time on the initial push for runtime fields.

Regardless of how we fix this I'd like to know more about your tests. Ultimately I'm going to have to reproduce this with rally and/or with our jmh benchmarks. Can you tell me what you were doing to get this to show up? I have a suspicion I'll be able to see this really easy with JMH but with rally I'd need to set up a bunch of machines to run a realistic cluster and generate load. Like, what kinds of searches are you sending? To how many shards? On how many nodes? More shards on single nodes'll all step on each other.....

[nik9000]

I had a look this morning. I made a JMH benchmark that builds six sum aggs as fast as its silicon heart can manage. The measurements I get out of it are kind of funny, especially when I run them on a GCP node with 60 cores. There, I see that building these SumAggregators on all the threads clocks in at about 9 microseconds. Not super duper fast, but, like, fine.

Now, with the circuit breaker on I see lots of contention and building the SumAggregators clocks in a 280 microseconds or 620 microseconds. I get both measurements and I'm not 100% sure why, but they are fairly consistent. Measurement mysteries aside, the circuit breaker costs about 600 microseconds in 60 threads. When I run the benchmark with a single thread I see that the circuit breakers adding 1.2 microseconds to building the aggregators. So contention is certainly killing us here. I mean, the flame graph said it, but I wanted to confirm.

I hacked together a prototype that only dips into the circuit breaker once for the whole request. Presumably we could make a slight overestimate of the memory we'll be needing for a "nearly empty" aggregation and allocate all of that up front. That dropped the construction to 170 microseconds or 85 microseconds. Same deal with the two consistent and different measurements. Something must be wrong with my test, but it shows us the order of magnitude. Anyway, 170 microseconds is a lot more than 9 microseconds, but its still a fair savings percentage wise.

What kind of numbers were you seeing, @lqbilbo? What kind of aggs are you running and how many cores do you have? What kind of CPU are you running? I haven't tried ARM. When I ran this on my desktop which has 12 cores I didn't really see much contention. So I imagine you are running with far more cores.

How fast are the requests you are measuring? 600 microseconds is, like, not all the much time. If you end up going to disk for any of your requests the 600 microseconds is a drop in the bucket. Still, lower CPU load is lower CPU load. And we're wasting a bunch of time here if you are running with dozens of CPUs. So I still think this is worth me looking more into.

[lqbilbo]

I had a look this morning. I made a JMH benchmark that builds six sum aggs as fast as its silicon heart can manage. The measurements I get out of it are kind of funny, especially when I run them on a GCP node with 60 cores. There, I see that building these SumAggregators on all the threads clocks in at about 9 microseconds. Not super duper fast, but, like, fine.

How fast are the requests you are measuring? 600 microseconds is, like, not all the much time. If you end up going to disk for any of your requests the 600 microseconds is a drop in the bucket. Still, lower CPU load is lower CPU load. And we're wasting a bunch of time here if you are running with dozens of CPUs. So I still think this is worth me looking more into.

We ran tests on a test cluster with 48core and 256g memory size on each node. And the aggs is a query phase with one Date histogram filter, one terms filter and about sixty sum aggs. Our cluster is the product environment cluster which all boxes of it has a high hdd space, or you can say they are for logging scenario, and we build up raid6 for a higher I/O read speed. The cluster configuration is three client nodes, 24 data&master nodes. And we ran nearly three thousand requests per second and the cpu load at highest point is about 60 percent, not too high as we know the nearly 100% CPU load would lead to a full search queue. and this is meaningless. Also, the pct99 is just 600 microseconds as your test show.

I hacked together a prototype that only dips into the circuit breaker once for the whole request. Presumably we could make a slight overestimate of the memory we'll be needing for a "nearly empty" aggregation and allocate all of that up front. That dropped the construction to 170 microseconds or 85 microseconds. Same deal with the two consistent and different measurements. Something must be wrong with my test, but it shows us the order of magnitude. Anyway, 170 microseconds is a lot more than 9 microseconds, but its still a fair savings percentage wise.

So if I can follow your measurements and run the tests with dipping circuit breaker for the whole request into once estimate memory or allocate all aggregation consume before all this happened?

[nik9000]

We ran tests on a test cluster with 48core and 256g memory size on each node. And the aggs is a query phase with one Date histogram filter, one terms filter and about sixty sum aggs. Our cluster is the product environment cluster which all boxes of it has a high hdd space, or you can say they are for logging scenario, and we build up raid6 for a higher I/O read speed. The cluster configuration is three client nodes, 24 data&master nodes. And we ran nearly three thousand requests per second and the cpu load at highest point is about 60 percent, not too high as we know the nearly 100% CPU load would lead to a full search queue. and this is meaningless. Also, the pct99 is just 600 microseconds as your test show.

Perfect.

Is the index being written to? If it isn't, you might be able to get caching by separating your aggs from your top_hits searches. But if you queries that vary then you won't get any cache hits anyway. One thing that is useful - if two requests come in with the same cache key and we're caching the request then we merge those request together.

So if I can follow your measurements and run the tests with dipping circuit breaker for the whole request into once estimate memory or allocate all aggregation consume before all this happened?

I really did just hack together the benchmark to see if I could reproduce your issue and to see if my hunch that we could pre-allocate would help. It isn't really useful outside of theorizing. Be that as it may, I did a sixty sums version on my 12 core machine and got:

Benchmark                                            (breaker)  Mode  Cnt    Score    Error  Units
AggConstructionContentionBenchmark.termsSixtySums         noop  avgt   10   81.853 ±  2.188  us/op
AggConstructionContentionBenchmark.termsSixtySums         real  avgt   10  297.595 ± 13.411  us/op
AggConstructionContentionBenchmark.termsSixtySums  preallocate  avgt   10  110.816 ± 10.993  us/op

Which looks to me like pre-allocating could do the trick.

[nik9000]

For those following along at home the contention gets much, much worse when you have a ton of cpus all trying. The preallocation helps quite a lot in that case so its ultimately what we merged in #66895 to close this.