Skip to content

Fix computation of QueuedQueries JMX metrics #19930

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tdcmeehan merged 3 commits into from
Jul 27, 2023
Merged

Fix computation of QueuedQueries JMX metrics #19930

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
12ec70f
Stop computing queued queries in QueryManagerStats#trackQueryStats
imjalpreet Jul 5, 2023
86f46cf
Fix computation of QueuedQueries JMX metric
imjalpreet Jul 6, 2023
5ec241b
Expose QueryManagerStats in JMX under DispatchManager namespace
imjalpreet Jul 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 27 additions & 8 deletions presto-main/src/main/java/com/facebook/presto/execution/QueryManagerStats.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

import static com.facebook.presto.execution.QueryState.QUEUED;
Copy link
Copy Markdown
Contributor

@yingsu00 yingsu00 Jul 3, 2023

Choose a reason for hiding this comment

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

I actually think trackQueryStats() methods in QueryManagerStats should NOT do queuedQueries.incrementAndGet();. Having a query submitted doesn't mean it's already queued. @tdcmeehan Do you know why they were done this way?

Copy link
Copy Markdown
Member Author

@imjalpreet imjalpreet Jul 4, 2023

Choose a reason for hiding this comment

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

I actually think trackQueryStats() methods in QueryManagerStats should NOT do queuedQueries.incrementAndGet();. Having a query submitted doesn't mean it's already queued.

I agree, that doesn't look right for both the methods. In the case of DispatchQuery, it is queued after the call to trackQueryStats, and in the case of SqlQueryManager, the query is not even in the queued state when trackQueryStats is called. It is called after waiting for resources finishes and query execution is about to start.

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;
Expand Down Expand Up @@ -59,14 +60,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)));
}
Expand All @@ -75,14 +74,18 @@ private void queryStarted()
{
startedQueries.update(1);
runningQueries.incrementAndGet();
queryDequeued();
}

private void queryStopped()
{
runningQueries.decrementAndGet();
}

private void queryQueued()
{
queuedQueries.incrementAndGet();
}

private void queryDequeued()
{
queuedQueries.decrementAndGet();
Expand Down Expand Up @@ -145,6 +148,8 @@ private class StatisticsListener
private boolean stopped;
@GuardedBy("this")
private boolean started;
@GuardedBy("this")
private boolean queued;

public StatisticsListener()
{
Expand All @@ -169,16 +174,30 @@ public void stateChanged(QueryState newValue)
if (started) {
queryStopped();
}
else {
else if (queued) {
queryDequeued();
}
finalQueryInfoSupplier.get()
.ifPresent(QueryManagerStats.this::queryFinished);
return;
}

if (newValue.ordinal() == QUEUED.ordinal()) {
if (!queued) {
queued = true;
queryQueued();
}
}
else if (newValue.ordinal() >= RUNNING.ordinal()) {
if (!started) {
started = true;
queryStarted();
else if (newValue.ordinal() > QUEUED.ordinal()) {
if (queued) {
queryDequeued();
queued = false;
}
if (newValue.ordinal() >= RUNNING.ordinal()) {
if (!started) {
started = true;
queryStarted();
}
}
}
}
Expand Down
1 change: 1 addition & 0 deletions presto-main/src/main/java/com/facebook/presto/server/CoordinatorModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,7 @@ protected void setup(Binder binder)

// dispatcher
binder.bind(DispatchManager.class).in(Scopes.SINGLETON);
newExporter(binder).export(DispatchManager.class).withGeneratedName();
Copy link
Copy Markdown
Member Author

@imjalpreet imjalpreet Jul 2, 2023

Choose a reason for hiding this comment

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

For now, I have exposed the DispatchManager QueryManagerStats in a separate namespace. But the stats object exposed via this class has more accurate stats(compared to SqlQueryManager) for all metrics including Queued Query and Running Query count.

If we decide to expose only one of them, I would recommend we should expose the DispatchManager stats object. To make the change backwards compatible we can expose it under the same namespace as before so it remains transparent to the users.

binder.bind(FailedDispatchQueryFactory.class).in(Scopes.SINGLETON);
binder.bind(DispatchExecutor.class).in(Scopes.SINGLETON);
newExporter(binder).export(DispatchExecutor.class).withGeneratedName();
Expand Down
38 changes: 38 additions & 0 deletions presto-tests/src/test/java/com/facebook/presto/tests/TestQueryManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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)
Expand Down Expand Up @@ -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<BasicQueryInfo> 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);
}
}
}