Open jsvd opened 1 month ago
Impact on 7.2.0 -> 7.2.1 can be easily observed. For the following pipelines.yaml:
- pipeline.id: source
config.string: "input { generator { count => 20000000 } } output { pipeline { send_to => destination } }"
- pipeline.id: destination
queue.type: persisted
config.string: "input { pipeline { address => destination } } output { null {} }"
we can observe the following numbers:
7.2.0 - 1m 46s
❯ curl -s localhost:9600/_node/stats/pipelines/destination | jq .pipelines.destination.events
{
"duration_in_millis": 771,
"filtered": 20000000,
"in": 20000000,
"out": 20000000,
"queue_push_duration_in_millis": 90072
}
/tmp/logstash-7.2.1
❯ curl -s localhost:9600/_node/stats/pipelines/source | jq .pipelines.source.events
{
"duration_in_millis": 866418,
"filtered": 20000000,
"in": 20000000,
"out": 20000000,
"queue_push_duration_in_millis": 7082
}
7.2.1 - 2m 38s
❯ curl -s localhost:9600/_node/stats/pipelines/destination | jq .pipelines.destination.events
{
"duration_in_millis": 565,
"filtered": 20000000,
"in": 20000000,
"out": 20000000
"queue_push_duration_in_millis": 3720,
}
/tmp/logstash-7.2.0
❯ curl -s localhost:9600/_node/stats/pipelines/source | jq .pipelines.source.events
{
"duration_in_millis": 1354787,
"filtered": 20000000,
"in": 20000000,
"out": 20000000
"queue_push_duration_in_millis": 12592,
}
Currently (up to 8.14.0 at the moment of writing), the PipelineBus class has a lock on the sender output:
internalReceive
will call Queue.write that mainly does 3 steps: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:
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: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.