Explicitly cleanup SqlTask on worker when no longer needed #24729

losipiuk · 2025-01-16T17:13:58Z

Currently SqlTask objects are removed from SqlTaskManager.tasks map (cache) after timeout (15 minutes by default). Even though the object is not huge, we observed increased memory pressure up to OOM on busy clusters.

With this PR entries are dropped form SqlTaskManager as soon as they are no longer needed, when coordinator will no longer query for the information

Note: I this code is prone to race condition if last worker to acknowlege task results retransmits acknowlegement. There is a chance that a this point task is already cleaned by cleanup request from coordinator, and entry will get resurected in tasks cache map. This should be extremly rare though, and not cause any problems.

wendigo · 2025-01-16T17:15:55Z

core/trino-main/src/main/java/io/trino/server/TaskResource.java

+    @POST
+    @Path("{taskId}/cleanup")
+    public void cleanupTask(
+            @PathParam("taskId") TaskId taskId)


nit: wrapping not needed

core/trino-main/src/main/java/io/trino/server/remotetask/RemoteTaskCleaner.java

core/trino-main/src/main/java/io/trino/server/remotetask/DynamicFiltersFetcher.java

sopel39 · 2025-01-16T17:28:55Z

core/trino-main/src/main/java/io/trino/server/remotetask/DynamicFiltersFetcher.java

@@ -124,6 +128,7 @@ public synchronized void updateDynamicFiltersVersionAndFetchIfNecessary(long new
    private synchronized void stop()


this method is not called normally. only in erronous situations. You cannot rely on it to be executed.

The external entry point for DynamicFiltersFetcher request is really updateDynamicFiltersVersionAndFetchIfNecessary.
So the conditions for DynamicFiltersFetcher being completed are:

there is no pending query

all places which could call updateDynamicFiltersVersionAndFetchIfNecessary are stopped (currenty it's continuous fetcher)

@sopel39 @raunaqmorarka Yeah - you are right. Do you have quick clue if there is natural place where we know that DynFilFetcher finished its work. The way it is called is kinda convoluted

Yes:

there is no pending request from DynamicFilterFetcher

all places which could call updateDynamicFiltersVersionAndFetchIfNecessary are stopped (it's only ContinuousTaskStatusFetcher atm IIRC)

PTAL now :)

core/trino-main/src/main/java/io/trino/server/remotetask/RemoteTaskCleaner.java

sopel39

Please add testing (with DF) to TestHttpRemoteTask

sopel39 · 2025-01-16T22:51:26Z

core/trino-main/src/main/java/io/trino/server/remotetask/RemoteTaskCleaner.java

+import static java.util.Objects.requireNonNull;
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+public class RemoteTaskCleaner


Do we need a separate service for this? Cannot HttpRemoteTask do cleanup itself?

It can - just having separate class is cleaner I think. I can put that as inner class too but I do not like that too much.

losipiuk · 2025-01-17T15:10:00Z

@sopel39 PTAL

losipiuk · 2025-01-21T14:30:09Z

@sopel39 PTAL

@sopel39 PTAL

core/trino-main/src/main/java/io/trino/execution/SqlTaskManager.java

sopel39 · 2025-01-23T16:26:58Z

core/trino-main/src/main/java/io/trino/server/remotetask/ContinuousTaskStatusFetcher.java

@@ -121,6 +124,8 @@ public synchronized void stop()
            future.cancel(true);
            future = null;
        }
+        remoteTaskCleaner.markTaskStatusFetcherStopped(taskStatus.get().getState());
+        dynamicFiltersFetcher.noMoreUpdateRequests();


could it be just DynamicFiltersFetcher#stop method?

no - as we want to wait for results of last after last DFF.updateDynamicFiltersVersionAndFetchIfNecessary call

sopel39 · 2025-01-23T16:34:55Z

core/trino-main/src/main/java/io/trino/server/remotetask/ContinuousTaskStatusFetcher.java

@@ -253,7 +258,11 @@ void updateTaskStatus(TaskStatus newValue)
            onFail.accept(new TrinoException(REMOTE_TASK_MISMATCH, format("%s (%s)", REMOTE_TASK_MISMATCH_ERROR, HostAddress.fromUri(getTaskStatus().getSelf()))));
        }

-        dynamicFiltersFetcher.updateDynamicFiltersVersionAndFetchIfNecessary(newValue.getDynamicFiltersVersion());
+        synchronized (this) {
+            if (running) {


Can it happen in non-aborted query?

I think that's still wrong, because last DF update could be potentially received after ContinuousTaskStatusFetcher#stop was called. We need that last request for DFs to happen, otherwise we loose filtering opportunity.

You mean that running can get true bacasue stop() is called as a result of listener due to taskStatus.setIf above?

Maybe we should just move

dynamicFiltersFetcher.updateDynamicFiltersVersionAndFetchIfNecessary(newValue.getDynamicFiltersVersion());

to the top of updateTaskStatus. WDYT?

sopel39 · 2025-01-23T16:35:54Z

core/trino-main/src/main/java/io/trino/server/remotetask/DynamicFiltersFetcher.java

@@ -113,17 +118,38 @@ public synchronized void start()

    public synchronized void updateDynamicFiltersVersionAndFetchIfNecessary(long newDynamicFiltersVersion)
    {
+        if (noMoreUpdateRequests) {


again, we need that last update to happen to fetch final DFs.

Just make dynamic filters fetcher report when it's finished.

see my previous comment.

What do you exactly mean by updateDynamicFiltersVersionAndFetchIfNecessary?
LIke return future from updateDynamicFiltersVersionAndFetchIfNecessary and mark it completed when relevant processing is completed?

return future from updateDynamicFiltersVersionAndFetchIfNecessary and mark it completed when relevant processing is completed?

I tried that in other branch - and it gets extremly ugly and complidcated.

I mean essentially that the fetchers are not considered closed if there is still a pending request. It may happens that last ContinuousTaskStatusFetcher.java request could trigger DF fetch request, so the end condition is:

ContinuousTaskStatusFetcher is REALLY closed (no pending request)

DF doesn't have any pending request.

I think if you just put checkpoints in right places accounting for the above you should be fine

sopel39 · 2025-01-23T16:37:45Z

core/trino-main/src/main/java/io/trino/server/remotetask/ContinuousTaskStatusFetcher.java

@@ -121,6 +124,8 @@ public synchronized void stop()
            future.cancel(true);
            future = null;
        }
+        remoteTaskCleaner.markTaskStatusFetcherStopped(taskStatus.get().getState());


I would postpone that only after pending request is finished, so that last DF update can be processed

how is this one related to pending requst in DF fetcher?

Currently SqlTask objects are removed from SqlTaskManager.tasks map (cache) after timeout (15 minutes by default). Even though the object is not huge, we observed increased memory pressure up to OOM on busy clusters. With this PR entries are dropped form SqlTaskManager as soon as they are no longer needed, when coordinator will no longer query for the information

losipiuk · 2025-01-28T20:11:09Z

@sopel39 ?

losipiuk · 2025-01-28T20:15:24Z

@sopel39 ?

actually wait - sth is messed up

cla-bot bot added the cla-signed label Jan 16, 2025

losipiuk mentioned this pull request Jan 16, 2025

Explicitly cleanup SqlTask on worker when no longer needed #24728

Closed

losipiuk requested review from sopel39, wendigo, findepi and pettyjamesm January 16, 2025 17:14

wendigo approved these changes Jan 16, 2025

View reviewed changes

sopel39 reviewed Jan 16, 2025

View reviewed changes

pettyjamesm reviewed Jan 16, 2025

View reviewed changes

core/trino-main/src/main/java/io/trino/server/remotetask/RemoteTaskCleaner.java Outdated Show resolved Hide resolved

wendigo reviewed Jan 16, 2025

View reviewed changes

core/trino-main/src/main/java/io/trino/server/remotetask/RemoteTaskCleaner.java Show resolved Hide resolved

wendigo reviewed Jan 16, 2025

View reviewed changes

core/trino-main/src/main/java/io/trino/server/remotetask/RemoteTaskCleaner.java Outdated Show resolved Hide resolved

sopel39 reviewed Jan 16, 2025

View reviewed changes

losipiuk force-pushed the lo/fsdfsarwqr branch from 5045fc2 to 362f673 Compare January 17, 2025 10:51

sopel39 mentioned this pull request Jan 17, 2025

Reduce task.info.max-age to 5m #24736

Merged

losipiuk force-pushed the lo/fsdfsarwqr branch from 362f673 to 2b798b6 Compare January 17, 2025 15:09

sopel39 reviewed Jan 23, 2025

View reviewed changes

losipiuk force-pushed the lo/fsdfsarwqr branch from 2b798b6 to 7ebaa40 Compare January 28, 2025 12:46

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Explicitly cleanup SqlTask on worker when no longer needed #24729

Explicitly cleanup SqlTask on worker when no longer needed #24729

losipiuk commented Jan 16, 2025

wendigo Jan 16, 2025

sopel39 Jan 16, 2025

sopel39 Jan 16, 2025

losipiuk Jan 17, 2025

sopel39 Jan 17, 2025

losipiuk Jan 28, 2025

sopel39 left a comment •

edited

Loading

sopel39 Jan 16, 2025

losipiuk Jan 17, 2025

losipiuk commented Jan 17, 2025

losipiuk commented Jan 21, 2025

sopel39 Jan 23, 2025

losipiuk Jan 24, 2025

sopel39 Jan 23, 2025

losipiuk Jan 24, 2025

sopel39 Jan 23, 2025

losipiuk Jan 24, 2025 •

edited

Loading

losipiuk Jan 24, 2025

sopel39 Jan 24, 2025 •

edited

Loading

sopel39 Jan 23, 2025

losipiuk Jan 24, 2025

losipiuk commented Jan 28, 2025

losipiuk commented Jan 28, 2025

		@@ -124,6 +128,7 @@ public synchronized void updateDynamicFiltersVersionAndFetchIfNecessary(long new
		private synchronized void stop()

Explicitly cleanup SqlTask on worker when no longer needed #24729

Are you sure you want to change the base?

Explicitly cleanup SqlTask on worker when no longer needed #24729

Conversation

losipiuk commented Jan 16, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sopel39 left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

losipiuk commented Jan 17, 2025

losipiuk commented Jan 21, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

losipiuk Jan 24, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sopel39 Jan 24, 2025 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

losipiuk commented Jan 28, 2025

losipiuk commented Jan 28, 2025

sopel39 left a comment •

edited

Loading

losipiuk Jan 24, 2025 •

edited

Loading

sopel39 Jan 24, 2025 •

edited

Loading