PipelineBus.sendEvents locks on the sender PipelineOutput impacting overall throughput
- Dominant language
- Java
- Stars
- 14.9k
- Forks
- 3.5k
- Avg merge
- 1d 4h
- Merged PRs (30d)
- 88
Description
Currently (up to 8.14.0 at the moment of writing), the PipelineBus class [has a lock on the sender output](https://github.com/elastic/logstash/blob/v8.13.4/logstash-core/src/main/java/org/logstash/plugins/pipeline/PipelineBus.java#L53-L58):
```java
public void sendEvents(final PipelineOutput sender,
final Collection events,
final boolean ensureDelivery) {
if (events.isEmpty()) return; // This can happen on pipeline shutdown or in some other situations
synchronized (sender) {
// .....
addressesToInputs.forEach((address, addressState) -> {
// .....
PipelineInput input = addressState.getInput(); // Save on calls to getInput since it's volatile
// .....
lastResponse = input.internalReceive(clones);
```
`internalReceive` will call Queue.write that mainly does 3 steps:
```
public long write(Queueable element) throws IOException {
//...
byte[] data = element.serialize();
lock.lock();
this.headPage.write( .. );
```
This means that when there are multiple writers to the same pipeline and serialization + pagewrite take a long time, most threads will spend time waiting for 1 thread that is writing an event, which can be seen with the simple pipelines.yml:
```yaml
- pipeline.id: source
config.string: "input { java_generator {} } output { pipeline { send_to => [dest1] } }"
- pipeline.id: dest
config.string: "input { pipeline { address => dest1 } } output { null {} }"
```
And `queue.type: persisted` in the logstash.yml. This will cause all but one of the workers of the upstream pipeline to be blocked at any given time:
```
❯ curl -s -XGET 'localhost:9600/_node/hot_threads?human=true&threads=30&stacktrace_size=10' | grep "thread name.*source.*worker" -A 1 | grep -v "\-\-"
12.17 % of cpu usage, state: blocked, thread name: '[source]>worker2', thread id: 54
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
12.12 % of cpu usage, state: blocked, thread name: '[source]>worker5', thread id: 60
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.96 % of cpu usage, state: blocked, thread name: '[source]>worker0', thread id: 51
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.93 % of cpu usage, state: blocked, thread name: '[source]>worker4', thread id: 58
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.89 % of cpu usage, state: runnable, thread name: '[source]>worker3', thread id: 56
java.base@17.0.10/jdk.internal.misc.Unsafe.unpark(Native Method)
11.84 % of cpu usage, state: blocked, thread name: '[source]>worker7', thread id: 63
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.82 % of cpu usage, state: blocked, thread name: '[source]>worker6', thread id: 62
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.66 % of cpu usage, state: blocked, thread name: '[source]>worker9', thread id: 68
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.57 % of cpu usage, state: blocked, thread name: '[source]>worker1', thread id: 52
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
11.48 % of cpu usage, state: blocked, thread name: '[source]>worker8', thread id: 65
app//org.logstash.plugins.pipeline.PipelineBus.sendEvents(PipelineBus.java:58)
```
This was introduced by https://github.com/elastic/logstash/pull/10872 to ensure proper order during pipeline shutdown. However it should be possible to improve concurrency by having a readwritelock that allows read access to the sender object during event processing, but uses the write lock for every other operation.
Contributor guide
Research direction
Start in logstash-core/src/main/java/org/logstash/plugins/pipeline/PipelineBus.java at sendEvents and review the synchronization introduced by pull request 10872. Reproduce the contention with the shown pipelines.yml and queue.type: persisted, then inspect the source pipeline workers with the hot_threads endpoint. Done means event delivery preserves pipeline shutdown ordering while avoiding unnecessary serialization and page-write contention between senders.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100