Skip to content

Dart: Wait for a merge buffer if none is available.#18075

Merged
gianm merged 6 commits into
apache:masterfrom
gianm:msq-dart-await-buffer
Jun 15, 2025
Merged

Dart: Wait for a merge buffer if none is available.#18075
gianm merged 6 commits into
apache:masterfrom
gianm:msq-dart-await-buffer

Conversation

@gianm

@gianm gianm commented Jun 4, 2025

Copy link
Copy Markdown
Contributor

For Dart-only workloads, it is not possible for merge buffers to be exhausted, because the number of concurrent queries is capped at the number of merge buffers by DartWorkerMemoryManagementModule. However, When native and Dart queries are run on the same Historical, it is possible for buffers to be held by native queries. In this case, Dart workers must wait for buffers to become available.

This waiting now happens in the worker runner thread, and lasts for the configured query timeout. At this time, the waiting is not interrupted when a cancellation request arrives from the controller. This should be addressed in another patch.

@github-actions github-actions Bot added Area - Batch Ingestion Area - MSQ For multi stage queries - https://github.com/apache/druid/issues/12262 labels Jun 4, 2025
For Dart-only workloads, it is not possible for merge buffers to be
exhausted, because the number of concurrent queries is capped at the
number of merge buffers by DartWorkerMemoryManagementModule. However,
When native and Dart queries are run on the same Historical, it is
possible for buffers to be held by native queries. In this case, Dart
workers must wait for buffers to become available.

This waiting now happens in the worker runner thread, and lasts for the
configured query timeout. At this time, the waiting is not interrupted
when a cancellation request arrives from the controller. This should be
addressed in another patch.
@gianm
gianm force-pushed the msq-dart-await-buffer branch from d3b60b1 to eecf2c6 Compare June 4, 2025 06:08
gianm added a commit to gianm/druid that referenced this pull request Jun 8, 2025
This patch simplifies the worker lifecycle and focuses on using interruption
instead of various flags to manage cancellation. This should enable workers to
cancel themselves more quickly, and is particularly important when they may
be waiting for a resource to become available (such as merge buffers as in apache#18075).

Changes:

1) Introduce WorkerRunRef to manage worker execution and cancellation. This class
   encapsulates the logic for running a worker in an executor and canceling it
   via thread interruption.

2) Move periodic controller status checking out of IndexerWorkerContext and into
   a dedicated PeriodicControllerChecker class.

3) Remove stop(), awaitStop(), and controllerFailed() from the Worker interface.
   Workers are now canceled via interruption only.

4) Remove controllerAlive tracking from WorkerImpl. It should no longer be necessary
   now that workers are interrupted on controller failure.
* Acquire buffers for a {@link Worker}.
*/
ResourceHolder<ProcessingBuffersSet> acquire(int poolSize);
ResourceHolder<ProcessingBuffersSet> acquire(int poolSize, long timeoutMillis);

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

note: I think the usage of Duration would be usefull

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IMO a long makes sense because there's only one call site, and it passes queryDef.getContext().getTimeout() which is a long.

Comment on lines +62 to +64
/**
* Acquire buffers if a particular stages needs them; otherwise, returns null.
*/

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

note: instead of this apidoc + @Nullable annotations and the if-s handling the null: it would be usefull to return a ResourceHolder which's get throws the exception that there is no buffer

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair enough, I made this change.

@gianm
gianm merged commit bb63387 into apache:master Jun 15, 2025
73 of 74 checks passed
@gianm
gianm deleted the msq-dart-await-buffer branch June 15, 2025 05:10
jtuglu1 pushed a commit to jtuglu1/druid that referenced this pull request Jun 17, 2025
* Dart: Wait for a merge buffer if none is available.

For Dart-only workloads, it is not possible for merge buffers to be
exhausted, because the number of concurrent queries is capped at the
number of merge buffers by DartWorkerMemoryManagementModule. However,
When native and Dart queries are run on the same Historical, it is
possible for buffers to be held by native queries. In this case, Dart
workers must wait for buffers to become available.

This waiting now happens in the worker runner thread, and lasts for the
configured query timeout. At this time, the waiting is not interrupted
when a cancellation request arrives from the controller. This should be
addressed in another patch.

* Add test coverage for DartProcessingBuffersProvider.

* Changes from review.

* Add tests.

* More tests.
gianm added a commit that referenced this pull request Jun 27, 2025
* MSQ: Use interrupts for worker cancellation.

This patch simplifies the worker lifecycle and focuses on using interruption
instead of various flags to manage cancellation. This should enable workers to
cancel themselves more quickly, and is particularly important when they may
be waiting for a resource to become available (such as merge buffers as in #18075).

Changes:

1) Introduce WorkerRunRef to manage worker execution and cancellation. This class
   encapsulates the logic for running a worker in an executor and canceling it
   via thread interruption.

2) Move periodic controller status checking out of IndexerWorkerContext and into
   a dedicated PeriodicControllerChecker class.

3) Remove stop(), awaitStop(), and controllerFailed() from the Worker interface.
   Workers are now canceled via interruption only.

4) Remove controllerAlive tracking from WorkerImpl. It should no longer be necessary
   now that workers are interrupted on controller failure.

* Style fixes.
@capistrant capistrant added this to the 34.0.0 milestone Jul 22, 2025
riovic918data pushed a commit to riovic918data/druid that referenced this pull request Jun 12, 2026
* Dart: Wait for a merge buffer if none is available.

For Dart-only workloads, it is not possible for merge buffers to be
exhausted, because the number of concurrent queries is capped at the
number of merge buffers by DartWorkerMemoryManagementModule. However,
When native and Dart queries are run on the same Historical, it is
possible for buffers to be held by native queries. In this case, Dart
workers must wait for buffers to become available.

This waiting now happens in the worker runner thread, and lasts for the
configured query timeout. At this time, the waiting is not interrupted
when a cancellation request arrives from the controller. This should be
addressed in another patch.

* Add test coverage for DartProcessingBuffersProvider.

* Changes from review.

* Add tests.

* More tests.
riovic918data pushed a commit to riovic918data/druid that referenced this pull request Jun 12, 2026
* MSQ: Use interrupts for worker cancellation.

This patch simplifies the worker lifecycle and focuses on using interruption
instead of various flags to manage cancellation. This should enable workers to
cancel themselves more quickly, and is particularly important when they may
be waiting for a resource to become available (such as merge buffers as in apache#18075).

Changes:

1) Introduce WorkerRunRef to manage worker execution and cancellation. This class
   encapsulates the logic for running a worker in an executor and canceling it
   via thread interruption.

2) Move periodic controller status checking out of IndexerWorkerContext and into
   a dedicated PeriodicControllerChecker class.

3) Remove stop(), awaitStop(), and controllerFailed() from the Worker interface.
   Workers are now canceled via interruption only.

4) Remove controllerAlive tracking from WorkerImpl. It should no longer be necessary
   now that workers are interrupted on controller failure.

* Style fixes.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Area - Batch Ingestion Area - MSQ For multi stage queries - https://github.com/apache/druid/issues/12262

Projects

None yet

Development

Successfully merging this pull request may close these issues.

4 participants