[Concurrent Segment Search] Make MultiBucketConsumerService thread safe to use across slices during search

[sohami]

Is your feature request related to a problem? Please describe.
With concurrent segment search we will need to make the MultiBucketConsumer thread safe as it will be shared across different slices of the shard segments. Ref here.

Describe the solution you'd like
Make MultiBucketConsumer thread safe. We also need to see if the usage will change with Concurrent flow. For example: currently if both global and non global aggregations are present in a request, it keep tracks of the buckets across both of those and then reset after collection is completed for both aggregations. Given with concurrent segment search reduce is happening separately for both Aggregation types we will need to ensure that reset only happens in postProcess for concurrent model i.e. when both aggregation has completed the reduce phase.

[reta]

@sohami we should be very careful to not introduce any contention here, defeating the purpose of searching concurrently. I have not looked into code deeply but have you considered the approach to allow overfetching? We do that for normal search as a limitation (for exactly same reasons)

[sohami]

@reta With initial look this is what I have found.

MultiBucketConsumer is used in 2 stages:

1. Final Reduce stage on coordinator side when the actual number of buckets collected is compared against the limit. I guess this is what you are referring as overfetching since shards can still collect buckets more than the limit and it is finally checked in coordinator side ? Ref here
2. On shards, where it doesn't check for bucket collection limit but does periodic check to see if request or parent level CircuitBreaker trips or not based on current usage. Ref here)

With concurrent search the reduce of aggregation will also happen on shards but on single thread during reduce phase. Also this is partial reduce where the bucket consumer is not consulted. So there is no change needed here and will keep the behavior as is.

Currently MultiBucketConsumer has 2 variables count and callCount which are non atomic.

• count is used for case 1 which is final reduce phase and that happens on single thread, so I don't see any changes are needed for that.
• callCount which will be incremented by different slices in concurrent case and will need to be made thread safe. This is used to trigger the CB to see if limit is breached with a value of 0

For callCount I am thinking below:

Make callCount an AtomicInteger also keep another AtomicBoolean variable which will be set when the CB trips for the first time. For each call to accept(value), increment and get the current value and compare that against the 1024 or check if new boolean variable is set. In either case it will call CB to check for memory pressure. This way if the first slice thread trips then all the other slice threads will also trip with CB check and fail all the slice threads for that request. Without the boolean variable one of the slice can trip whereas other slice can continue running and will trip only once the callCount has been increased by 1024 and so on.

[reta]

@sohami thanks a lot for the research,

My main concern is how much the contention would impact the concurrent search, from this perspective adding more synchronization would definitely have larger (negative) impact. I would suggest to use LongAccumulator instead of AtomicInteger (you could check the Javadocs why it is preferable) and, if possible, drop the usage of the AtomicBoolean. Regarding the tripping the other slices (if we drop AtomicBoolean), we could consider using LongAccumulator for that by adding some very large number to record the fact that one of the slices had overflow (over the limit), just an idea, I am sure there are more clever options to look at as well.

[neetikasinghal]

I will be working on this issue

[neetikasinghal]

callCount variable needs to be made thread safe as there is a periodic check on the CircuitBreaker in the MultiConsumerService that checks if the CB has tripped. With Concurrent Search, for each shard, there are multiple slices running on different threads and having one instance of MultiBucketConsumer. Multiple threads within the same instance of MultiBucketConsumer will try to access the callCount variable, and as of current logic that checks for CB trip after every 1024 calls to the accept function, there is possibility that if the callCount is not made thread safe, the CB trip check might never happen.

In order to make MultiBucketConsumer thread safe, these are the following options(neetikasinghal@5153c45):

Option 1
Having synchronized block in the method, which makes variable callCount as thread safe.

Reference to the code pointer: link

Average time taken per test for call to the accept function(Total test runs=1000, number of threads per test=6): 1.098ms
Test code: link

Pros

• This would essentially trip the circuit breaker for the other threads if one of the threads has already tripped the CB.
• Least time taken in the test runs across different options.
• There is one exception thrown per shard by the MultiBucketConsumer.

Cons

• Under high contention, expected throughput of this can be lower.

Option 2
Initializing callCount as a LongAdder variable making it thread safe.

Reference to the code pointer: link

Average time taken per test for call to the accept function(Total test runs=1000, number of threads per test=6): 1.559ms

Pros

• Under high contention, expected throughput of this is significantly higher
• There is one exception thrown per shard by the MultiBucketConsumer.

Cons

• This doesn't trip the CB on any other threads if one of the thread already has a circuit breaker.
• Most time taken in the test runs across different options.
• Can have a higher memory consumption due to the use of LongAdder

Option 3
Initializing callCount as a LongAdder variable making it thread safe and also having an additional Atomic variable that tries to trip the CB for the other threads in case for one of the threads CB has already tripped.

Reference to the code pointer: link

Average time taken per test for call to the accept function(Total test runs=1000, number of threads per test=6): 1.211ms

Pros

• There is one exception thrown per shard by the MultiBucketConsumer.

Cons

• Some threads might also trip CB after the CB has tripped but this doesn't guarantee that all the threads would essentially trip CB after the first thread tripped as there can be a thread that could come for a check in the interval between circuitBreakerTripped3 variable is set vs get.

Option 4
Make MultiBucketConsumer specific to per slice, this would mean that instances of MultiBucketConsumer are specific to each of the thread which means every thread would maintain the callCount variable, hence callCount variable wouldn't need be thread safe.

Pros

• This would not need adding any synchronization as mentioned in the above options.

Cons

• There will be multiple exceptions thrown per shard by the MultiBucketConsumer which would be a change in the current behavior with normal search.