Improve cancellability in TransportTasksAction #96279
Conversation
Each `TransportTasksAction` fans-out to multiple nodes, accumulates responses and retains them until all the nodes have responded, and then converts the responses into a final result. Similarly to elastic#92987 and elastic#93484, we should accumulate the responses in a structure that doesn't require so much copying later on, and should drop the received responses if the task is cancelled while some nodes' responses are still pending.
DaveCTurner
added
>bug
:Distributed Coordination/Task Management
elasticsearchmachine
added
the
Team:Distributed (Obsolete)
|
Pinging @elastic/es-distributed (Team:Distributed) |
|
Hi @DaveCTurner, I've created a changelog YAML for you. |
In a busy cluster the list-tasks API may retain information about a very large number of tasks while waiting for all nodes to respond. This commit makes the API cancellable so that unnecessary partial results can be released earlier. Relates elastic#96279, which implements the early-release functionality.
|
|
||
| reachabilityChecker.ensureUnreachable(); | ||
|
|
||
| while (true) { |
There was a problem hiding this comment.
Would it make sense to use while (taskResponseListeners.peek() != null) here instead of manually breaking out of the loop?
There was a problem hiding this comment.
That'd work, but I'd rather just do a single read, avoiding having to reason about the relationship between peek() and poll().
| return; | ||
| } | ||
|
|
||
| logger.debug(Strings.format("failed to execute on node [{}]", nodeId), e); |
There was a problem hiding this comment.
I was wondering if can perform formatting lazily by using a supplier : logger.debug(() -> Strings.format(
There was a problem hiding this comment.
Yes; it's not very common to get here, but I'll do that.
| @Override | ||
| public void onResponse(NodeTasksResponse nodeResponse) { | ||
| synchronized (taskResponses) { | ||
| taskResponses.addAll(nodeResponse.results); |
There was a problem hiding this comment.
I've run some tests and have seen a lot of cases where nodeResponse.results is empty. Would it make sense to perform an isEmpty check on nodeResponse.results before acquiring the lock on taskResponses?
There was a problem hiding this comment.
Yes that makes sense.
| private void nodeOperation(CancellableTask task, NodeTaskRequest nodeTaskRequest, ActionListener<NodeTasksResponse> listener) { | ||
| TasksRequest request = nodeTaskRequest.tasksRequest; | ||
| processTasks(request, ActionListener.wrap(tasks -> nodeOperation(task, listener, request, tasks), listener::onFailure)); | ||
| final var taskResponses = new ArrayList<TaskResponse>(); |
There was a problem hiding this comment.
Does it make sense to pre-size taskResponses based on the amount of the nodes from which we collect responses?
There was a problem hiding this comment.
I don't think the number of nodes would be a particularly useful estimate for this - as you say, many nodes return nothing, but sometimes they will return thousands of results. These are pretty low-throughput APIs so I think the default sizing is ok.
DaveCTurner
added
auto-merge-without-approval
|
@elasticmachine update branch |
We have this somewhat-complex pattern in 3 places already, and elastic#96279 will introduce a couple more, so this commit extracts it as a dedicated utility. Relates elastic#92987 Relates elastic#93484
|
Yeah it is a bit convoluted isn't it? I extracted a utility (and added more commentary) in #96373, and will update this PR to use it once that's merged. |
DaveCTurner
added
the
auto-merge-without-approval
In a busy cluster the list-tasks API may retain information about a very large number of tasks while waiting for all nodes to respond. This commit makes the API cancellable so that unnecessary partial results can be released earlier. Relates #96279, which implements the early-release functionality.
5 participants
Each
TransportTasksActionfans-out to multiple nodes, accumulates responses and retains them until all the nodes have responded, and then converts the responses into a final result.Similarly to #92987 and #93484, we should accumulate the responses in a structure that doesn't require so much copying later on, and should drop the received responses if the task is cancelled while some nodes' responses are still pending.