From 12ec70f270def0fe142c60476a2b93817ac67fb1 Mon Sep 17 00:00:00 2001 From: "Jalpreet Singh Nanda (:imjalpreet)" Date: Wed, 5 Jul 2023 13:46:00 +0530 Subject: [PATCH 1/3] Stop computing queued queries in QueryManagerStats#trackQueryStats The count should not be incremented at this stage since both DispatchQuery and QueryExecution are not in QueuedState. --- .../java/com/facebook/presto/execution/QueryManagerStats.java | 2 -- 1 file changed, 2 deletions(-) diff --git a/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java b/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java index 2705c0655a50d..7797dfd3b2e24 100644 --- a/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java +++ b/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java @@ -59,14 +59,12 @@ public class QueryManagerStats public void trackQueryStats(DispatchQuery managedQueryExecution) { submittedQueries.update(1); - queuedQueries.incrementAndGet(); managedQueryExecution.addStateChangeListener(new StatisticsListener(managedQueryExecution)); } public void trackQueryStats(QueryExecution managedQueryExecution) { submittedQueries.update(1); - queuedQueries.incrementAndGet(); managedQueryExecution.addStateChangeListener(new StatisticsListener()); managedQueryExecution.addFinalQueryInfoListener(finalQueryInfo -> queryFinished(new BasicQueryInfo(finalQueryInfo))); } From 86f46cfdd08da593131fd73398eb56f25f9c1eab Mon Sep 17 00:00:00 2001 From: "Jalpreet Singh Nanda (:imjalpreet)" Date: Thu, 6 Jul 2023 19:32:00 +0530 Subject: [PATCH 2/3] Fix computation of QueuedQueries JMX metric Implement tracking of Queued QueryState in QueryManagerStats#StatisticsListener to increment the count based on state change Fixes prestodb#19929 --- .../presto/execution/QueryManagerStats.java | 33 +++++++++++++--- .../presto/tests/TestQueryManager.java | 38 +++++++++++++++++++ 2 files changed, 65 insertions(+), 6 deletions(-) diff --git a/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java b/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java index 7797dfd3b2e24..ba7fb4aeafbe1 100644 --- a/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java +++ b/presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java @@ -28,6 +28,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Supplier; +import static com.facebook.presto.execution.QueryState.QUEUED; import static com.facebook.presto.execution.QueryState.RUNNING; import static com.facebook.presto.spi.StandardErrorCode.ABANDONED_QUERY; import static com.facebook.presto.spi.StandardErrorCode.USER_CANCELED; @@ -73,7 +74,6 @@ private void queryStarted() { startedQueries.update(1); runningQueries.incrementAndGet(); - queryDequeued(); } private void queryStopped() @@ -81,6 +81,11 @@ private void queryStopped() runningQueries.decrementAndGet(); } + private void queryQueued() + { + queuedQueries.incrementAndGet(); + } + private void queryDequeued() { queuedQueries.decrementAndGet(); @@ -143,6 +148,8 @@ private class StatisticsListener private boolean stopped; @GuardedBy("this") private boolean started; + @GuardedBy("this") + private boolean queued; public StatisticsListener() { @@ -167,16 +174,30 @@ public void stateChanged(QueryState newValue) if (started) { queryStopped(); } - else { + else if (queued) { queryDequeued(); } finalQueryInfoSupplier.get() .ifPresent(QueryManagerStats.this::queryFinished); + return; } - else if (newValue.ordinal() >= RUNNING.ordinal()) { - if (!started) { - started = true; - queryStarted(); + + if (newValue.ordinal() == QUEUED.ordinal()) { + if (!queued) { + queued = true; + queryQueued(); + } + } + else if (newValue.ordinal() > QUEUED.ordinal()) { + if (queued) { + queryDequeued(); + queued = false; + } + if (newValue.ordinal() >= RUNNING.ordinal()) { + if (!started) { + started = true; + queryStarted(); + } } } } diff --git a/presto-tests/src/test/java/com/facebook/presto/tests/TestQueryManager.java b/presto-tests/src/test/java/com/facebook/presto/tests/TestQueryManager.java index e7e87d72e1c24..27eb2814a95ae 100644 --- a/presto-tests/src/test/java/com/facebook/presto/tests/TestQueryManager.java +++ b/presto-tests/src/test/java/com/facebook/presto/tests/TestQueryManager.java @@ -32,6 +32,8 @@ import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; +import java.util.List; + import static com.facebook.presto.SessionTestUtils.TEST_SESSION; import static com.facebook.presto.execution.QueryState.FAILED; import static com.facebook.presto.execution.QueryState.QUEUED; @@ -49,6 +51,7 @@ import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertNotEquals; import static org.testng.Assert.assertNotNull; +import static org.testng.Assert.assertTrue; import static org.testng.Assert.fail; @Test(singleThreaded = true) @@ -226,4 +229,39 @@ public void testQueryOutputSizeExceeded() assertEquals(queryInfo.getErrorCode(), EXCEEDED_OUTPUT_SIZE_LIMIT.toErrorCode()); } } + + @Test + public void testQueryCountMetrics() + throws Exception + { + DispatchManager dispatchManager = queryRunner.getCoordinator().getDispatchManager(); + // Create a total of 10 queries to test concurrency limit and + // ensure that some queries are queued as concurrency limit is 3 + createQueries(dispatchManager, 10); + + List queries = dispatchManager.getQueries(); + long queuedQueryCount = dispatchManager.getStats().getQueuedQueries(); + long runningQueryCount = dispatchManager.getStats().getRunningQueries(); + + assertEquals(queuedQueryCount, + queries.stream().filter(basicQueryInfo -> basicQueryInfo.getState() == QUEUED).count()); + assertEquals(runningQueryCount, + queries.stream().filter(basicQueryInfo -> basicQueryInfo.getState() == RUNNING).count()); + + Stopwatch stopwatch = Stopwatch.createStarted(); + + long oldQueuedQueryCount = queuedQueryCount; + + // Assert that number of queued queries are decreasing with time and + // number of running queries are always <= 3 (max concurrency limit) + while (dispatchManager.getStats().getQueuedQueries() + dispatchManager.getStats().getRunningQueries() > 0 + && stopwatch.elapsed().toMillis() < 60000) { + assertTrue(dispatchManager.getStats().getQueuedQueries() <= oldQueuedQueryCount); + assertTrue(dispatchManager.getStats().getRunningQueries() <= 3); + + oldQueuedQueryCount = dispatchManager.getStats().getQueuedQueries(); + + Thread.sleep(100); + } + } } From 5ec241b6ea6a33b3725aae3398ca3928a8d3c10d Mon Sep 17 00:00:00 2001 From: "Jalpreet Singh Nanda (:imjalpreet)" Date: Fri, 7 Jul 2023 21:51:00 +0530 Subject: [PATCH 3/3] Expose QueryManagerStats in JMX under DispatchManager namespace --- .../main/java/com/facebook/presto/server/CoordinatorModule.java | 1 + 1 file changed, 1 insertion(+) diff --git a/presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java b/presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java index 07b5ad2703a07..60489142b810c 100644 --- a/presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java +++ b/presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java @@ -189,6 +189,7 @@ protected void setup(Binder binder) // dispatcher binder.bind(DispatchManager.class).in(Scopes.SINGLETON); + newExporter(binder).export(DispatchManager.class).withGeneratedName(); binder.bind(FailedDispatchQueryFactory.class).in(Scopes.SINGLETON); binder.bind(DispatchExecutor.class).in(Scopes.SINGLETON); newExporter(binder).export(DispatchExecutor.class).withGeneratedName();