Throttle Network Reads on Memory Pressure

[original-brownbear]

The Problem

Currently, when a node experiences heavy load we continue to read from connections as fast as possible in most cases (recovery, snapshots, etc. have throughput limits, but essentially all other requests don't). This creates two issues:

Higher than Necessary Memory Use

• We're reading messages and partially deserializing that we know we can't process

Since we are never throttling reads connections will if constantly writing to a node that can't handle the throughput can eventually lead to the node running out of heap-memory (without the real memory circuit breaker in place). This is a result of requests that run on a thread-pool without a size limit (generic pool) queuing up in de-serialized form and without limit.

Also, without throttling reads the network buffer memory (currently that's on heap as well) used by a node is proportional to the product of the average message rate per connection, average message size and the number of connections in the ideal case. Under heavy load with the circuit breaker interrupting requests, this might not hold if connections retry requests. So currently, under heavy load, heap pressure can result in increased additional heap pressure from network buffers and thus sub-optimal scaling.
Under heavy load, every message read and then failed by the circuit-breaker costs the buffer pool memory to buffer it in as well as some heap-memory for de-serializing the type/action of message and such.

Liveness Issues due to Real Memory Circuit Breaker

• Heavy load by one type of request can prevent other requests from executing reliably

Since under heavy load request currently get interrupted by the real memory circuit breaker, all connections currently share a single
memory pool. This introduces liveness issues like e.g.:

• Heavy search load on a data-node can lead to the circuit breaker killing a replication request leading to the replica failing.
  • You could technically argue that a node limited by available memory and under sustained heavy search load can't hold a stable replica
• Search phases interfering with each other Prioritize FetchPhase over QueryPhase Under High Search Workload #39891

Under heavy load, the system does not fairly distribute resources across connections and thus a single misbehaving connection can
monopolize a node. This behavior makes the system less efficient in its memory use under heavy load than it is without memory pressure due to wasted allocations on interrupt and retrying.

Suggested Solution

To improve behavior under heavy load we should globally limit network buffer memory use by not reading from connections when no memory is available to deal with a message anyway. We can't simply introduce a global limit on network buffer memory use as it would simply move the liveness issues mentioned above to a different layer.
Instead, we could roughly (implementation details in the sub-points are a little rough):

1. Limit the amount of network buffer memory available per connection and stop reading once no memory is available
   • For this to be efficient we should move the full deserialization of messages off of the IO thread (for messages that fork off to another pool). This way we save a lot of heap memory because we don't have to store serialized requests/responses on-heap as part of tasks waiting to be executed and thus only have de-serialized messages live on-heap for as long as they're needed on the heap. Connections that keep sending messages below the per-connection memory limit and that are handled on the IO thread would always be live. Also, we could make sure to only release the memory associated with a message once its corresponding task was complete instead of when the message was de-serialized to get smoother behavior for e.g. heavy search requests on coordinating nodes.
2. Have a global reserve pool for all connections to allocate from if additional memory beyond the per-connection limit is required and fairly wait for allocations of memory from it
   • Larger messages that exceed the per-connection limit would be throttled when having to wait for memory from the reserve pool without breaking liveness for other connections.
3. Interrupt messages (like we currently do with the circuit breaker kicks in) when allocating enough memory from the global pool takes too long to communicate back-pressure to the connection.
   • The timeout of this interruption could be made dependent on the action of the message. E.g. Search requests could be aggressively interrupted after a short wait because the cost of retrying is low for the client, search responses or replica requests could get a higher timeout due to the cost/waste associated with interrupting these.

[elasticmachine]

Pinging @elastic/es-distributed

[original-brownbear]

One somewhat interesting reproducer of the issues from not throttling reads is the following:

Create a 3-node (can be on the same host) cluster with -Xmx=256m for each node (just default configuration otherwise).

Run:

esrally --track=http_logs --target-hosts=127.0.0.1:9200 --track-params="bulk_size:50,number_of_replicas:3,number_of_shards:3,bulk_indexing_clients:16

This should work fine (as in no shard failures). If it doesn't because you have few CPU cores, lower the number of clients.

Then crank up the number of clients a lot and make the requests a little larger:

esrally --track=http_logs --target-hosts=127.0.0.1:9200 --track-params="bulk_size:150,number_of_replicas:3,number_of_shards:3,bulk_indexing_clients:100

This will start producing a lot of the following:

"shard":1,
"primary":false,
"current_state":"unassigned",
"unassigned_info":{"reason":"ALLOCATION_FAILED","at":"2019-07-18T06:18:55.601Z",
"failed_allocation_attempts":1,
"details":"failed shard on node [YzwjFCTfSeKMOZLhWZNmYg]: failed to perform indices:data/write/bulk[s] on replica [logs-201998][1], node[YzwjFCTfSeKMOZLhWZNmYg]
...
failure RemoteTransportException[[pc-168][127.0.0.1:9302][indices:data/write/bulk[s][r]]]; nested: CircuitBreakingException[[parent] Data too large, data for [<transport_request>] would be [246898276/235.4mb], which is larger than the limit of [246546432/235.1mb], real usage: [246865928/235.4mb], new bytes reserved: [32348/31.5kb], usages [request=213720/208.7kb, fielddata=0/0b, in_flight_requests=6576290/6.2mb, accounting=3210276/3mb]];

and have a lower indexing rate than the example running smaller requests at lower parallelism. This IMO is a bug, just because a user is cranking up parallelism the indexing rate shouldn't be lower for requests that individually easily fit into memory (in this case it's like 32k for each bulk request apparently ...).

[original-brownbear]

Another (expected) finding of running the above test with ever larger bulk sizes that IMO is a clear bug:

• Increasing the bulk size to something like 1500 which is about 300kb per bulk request it seems actually is capable of killing a node's process (with -Xmx256m) reliably since the circuit breaker doesn't kick in for network reads. So at 100 clients in parallel we have 30MB of in-flight bulk requests that could simultaneously arrive before the breaker has any chance to catch them -> we OOM.

[original-brownbear]

One thought I had on how to make incremental progress here:

Can't we simply implement throttling on memory pressure just for the BULK connections and keep the behavior for all other connections as is. Concretely, how about this:

• Non bulk channels behave like they currently do:
  • Never throttle
  • Circuit break all messages that allow it
• Bulk channels do this instead:
  • Throttle reads instead of circuit breaking for replication requests. So when the circuit breaker would've failed, auto-read (Netty) is set to false (we'd have to add a similar thing to NIO to turn off read interest)
  • Re-enable reads once the circuit breaker drops below a certain watermark (need to add some hook for that)
  • To fix the liveness issue here, we could define an amount of memory that we to be below the circuit breaker watermark that we require per channel and then reactivate reading in fair order roughly so:

0. We set a limit of 4 MB of free heap to allow reading from a channel
1. Channels A, B and C turn off auto-reads as a result of running into the circuit breaker on a request in order (A then B then C)
2. Memory usage drops 4 MB below the breaker watermark -> re-enable reading from A
3. Next read from A trips breaker again
4. Memory drops 4 MB below watermark -> re-enable reading from B

...

If we appropriately set limits here, it seems like we should be able to safely turn off the circuit breaker for replica requests?