This is an automated email from the ASF dual-hosted git repository.
korlov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new c2afff94d8 IGNITE-19412 Sql. Memory leak in MailboxRegistryImpl (#2043)
c2afff94d8 is described below
commit c2afff94d80f479a4e6099ab6e126efe9b232a17
Author: korlov42 <[email protected]>
AuthorDate: Thu May 11 16:49:19 2023 +0300
IGNITE-19412 Sql. Memory leak in MailboxRegistryImpl (#2043)
---
.../sql/engine/exec/ExecutionServiceImpl.java | 159 ++++++++++++---------
.../sql/engine/exec/ExecutionServiceImplTest.java | 81 ++++++++++-
2 files changed, 174 insertions(+), 66 deletions(-)
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
index 06b75885a3..062d53155a 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
@@ -218,7 +218,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
BaseQueryContext ctx,
MultiStepPlan plan
) {
- DistributedQueryManager queryManager = new
DistributedQueryManager(ctx);
+ DistributedQueryManager queryManager = new
DistributedQueryManager(true, ctx);
DistributedQueryManager old = queryManagerMap.put(ctx.queryId(),
queryManager);
@@ -427,6 +427,8 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
* A convenient class that manages the initialization and termination of
distributed queries.
*/
private class DistributedQueryManager {
+ private final boolean coordinator;
+
private final BaseQueryContext ctx;
private final CompletableFuture<Void> cancelFut = new
CompletableFuture<>();
@@ -441,9 +443,9 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
private volatile Long rootFragmentId = null;
-
- private DistributedQueryManager(BaseQueryContext ctx) {
+ private DistributedQueryManager(boolean coordinator, BaseQueryContext
ctx) {
this.ctx = ctx;
+ this.coordinator = coordinator;
var root = new CompletableFuture<AsyncRootNode<RowT,
List<Object>>>();
@@ -456,6 +458,10 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
this.root = root;
}
+ private DistributedQueryManager(BaseQueryContext ctx) {
+ this(false, ctx);
+ }
+
private List<AbstractNode<?>> localFragments() {
return List.copyOf(localFragments);
}
@@ -660,32 +666,32 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
for (String nodeName : fragmentDesc.nodeNames()) {
CompletableFuture<Void> resultOfSending =
sendFragment(nodeName, fragment, fragmentDesc, attributes);
- resultsOfFragmentSending.add(resultOfSending);
-
- resultOfSending.handle((ignored, t) -> {
- if (t == null) {
- return null;
- }
+ resultsOfFragmentSending.add(
+ resultOfSending.handle((ignored, t) -> {
+ if (t == null) {
+ return null;
+ }
- // if we were unable to send a request, then
no need
- // to wait for the remote node to complete
initialization
+ // if we were unable to send a
request, then no need
+ // to wait for the remote node to
complete initialization
- CompletableFuture<?> completionFuture =
remoteFragmentInitCompletion.get(
- new RemoteFragmentKey(nodeName,
fragment.fragmentId())
- );
+ CompletableFuture<?> completionFuture
= remoteFragmentInitCompletion.get(
+ new
RemoteFragmentKey(nodeName, fragment.fragmentId())
+ );
- if (completionFuture != null) {
- completionFuture.complete(null);
- }
+ if (completionFuture != null) {
+ completionFuture.complete(null);
+ }
- throw ExceptionUtils.withCauseAndCode(
- IgniteInternalException::new,
- Sql.INTERNAL_ERR,
- format("Unable to send fragment
[targetNode={}, fragmentId={}, cause={}]",
- nodeName,
fragment.fragmentId(), t.getMessage()),
- t
- );
- });
+ throw ExceptionUtils.withCauseAndCode(
+ IgniteInternalException::new,
+ Sql.INTERNAL_ERR,
+ format("Unable to send
fragment [targetNode={}, fragmentId={}, cause={}]",
+ nodeName,
fragment.fragmentId(), t.getMessage()),
+ t
+ );
+ })
+ );
}
}
@@ -812,56 +818,23 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
start
.thenCompose(tmp -> {
- Map<String, List<CompletableFuture<?>>>
requestsPerNode = new HashMap<>();
- for (Map.Entry<RemoteFragmentKey,
CompletableFuture<Void>> entry : remoteFragmentInitCompletion.entrySet()) {
-
requestsPerNode.computeIfAbsent(entry.getKey().nodeName(), key -> new
ArrayList<>()).add(entry.getValue());
- }
-
- List<CompletableFuture<?>> cancelFuts = new
ArrayList<>();
- for (Map.Entry<String, List<CompletableFuture<?>>>
entry : requestsPerNode.entrySet()) {
- String nodeId = entry.getKey();
-
- cancelFuts.add(
-
CompletableFuture.allOf(entry.getValue().toArray(new CompletableFuture[0]))
- .handle((none2, t) -> {
- // Some fragments may be
completed exceptionally, and that is fine.
- // Here we need to make sure
that all query initialisation requests
- // have been processed before
sending a cancellation request. That's
- // why we just ignore any
exception.
-
- return null;
- })
- .thenCompose(ignored ->
messageService.send(
- nodeId,
- FACTORY.queryCloseMessage()
-
.queryId(ctx.queryId())
- .build()
- ))
- );
- }
-
- if (cancel) {
- ExecutionCancelledException ex = new
ExecutionCancelledException();
-
- for (AbstractNode<?> node : localFragments) {
- node.context().execute(() -> node.onError(ex),
node::onError);
- }
- }
+ CompletableFuture<Void> cancelResult = coordinator
+ ? awaitFragmentInitialisationAndClose()
+ : closeLocalFragments();
- var compoundCancelFut =
CompletableFuture.allOf(cancelFuts.toArray(new CompletableFuture[0]));
- var finalStepFut = compoundCancelFut.whenComplete((r,
e) -> {
+ var finalStepFut = cancelResult.whenComplete((r, e) ->
{
queryManagerMap.remove(ctx.queryId());
try {
ctx.cancel().cancel();
- } catch (Exception ex) {
+ } catch (Exception ignored) {
// NO-OP
}
cancelFut.complete(null);
});
- return compoundCancelFut.thenCombine(finalStepFut,
(none1, none2) -> null);
+ return cancelResult.thenCombine(finalStepFut, (none1,
none2) -> null);
});
start.completeAsync(() -> null, taskExecutor);
@@ -869,6 +842,64 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
return cancelFut.thenApply(Function.identity());
}
+ private CompletableFuture<Void> closeLocalFragments() {
+ ExecutionCancelledException ex = new ExecutionCancelledException();
+
+ List<CompletableFuture<?>> localFragmentCompletions = new
ArrayList<>();
+ for (AbstractNode<?> node : localFragments) {
+ if (node.context().isCancelled()) {
+ continue;
+ }
+
+ localFragmentCompletions.add(
+ node.context().submit(() -> node.onError(ex),
node::onError)
+ );
+ }
+
+ return CompletableFuture.allOf(
+ localFragmentCompletions.toArray(new CompletableFuture[0])
+ );
+ }
+
+ private CompletableFuture<Void> awaitFragmentInitialisationAndClose() {
+ Map<String, List<CompletableFuture<?>>> requestsPerNode = new
HashMap<>();
+ for (Map.Entry<RemoteFragmentKey, CompletableFuture<Void>> entry :
remoteFragmentInitCompletion.entrySet()) {
+ requestsPerNode.computeIfAbsent(entry.getKey().nodeName(), key
-> new ArrayList<>()).add(entry.getValue());
+ }
+
+ List<CompletableFuture<?>> cancelFuts = new ArrayList<>();
+ for (Map.Entry<String, List<CompletableFuture<?>>> entry :
requestsPerNode.entrySet()) {
+ String nodeId = entry.getKey();
+
+ cancelFuts.add(
+ CompletableFuture.allOf(entry.getValue().toArray(new
CompletableFuture[0]))
+ .handle((none2, t) -> {
+ // Some fragments may be completed
exceptionally, and that is fine.
+ // Here we need to make sure that all
query initialisation requests
+ // have been processed before sending a
cancellation request. That's
+ // why we just ignore any exception.
+
+ return null;
+ })
+ .thenCompose(ignored -> {
+ // for local fragments don't send the
message, just close the fragments
+ if (localNode.name().equals(nodeId)) {
+ return closeLocalFragments();
+ }
+
+ return messageService.send(
+ nodeId,
+ FACTORY.queryCloseMessage()
+ .queryId(ctx.queryId())
+ .build()
+ );
+ })
+ );
+ }
+
+ return CompletableFuture.allOf(cancelFuts.toArray(new
CompletableFuture[0]));
+ }
+
/**
* Synchronously closes the tree's execution iterator.
*
diff --git
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index 305b4b17d7..a9d34fcde6 100644
---
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -39,9 +39,12 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
+import java.util.Collections;
+import java.util.IdentityHashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
@@ -64,7 +67,9 @@ import
org.apache.ignite.internal.sql.engine.AsyncCursor.BatchedResult;
import org.apache.ignite.internal.sql.engine.QueryCancel;
import
org.apache.ignite.internal.sql.engine.exec.ExecutionServiceImplTest.TestCluster.TestNode;
import org.apache.ignite.internal.sql.engine.exec.ddl.DdlCommandHandler;
+import org.apache.ignite.internal.sql.engine.exec.rel.Inbox;
import org.apache.ignite.internal.sql.engine.exec.rel.Node;
+import org.apache.ignite.internal.sql.engine.exec.rel.Outbox;
import org.apache.ignite.internal.sql.engine.exec.rel.ScanNode;
import org.apache.ignite.internal.sql.engine.framework.NoOpTransaction;
import org.apache.ignite.internal.sql.engine.framework.TestTable;
@@ -129,6 +134,8 @@ public class ExecutionServiceImplTest {
private final IgniteSchema schema = new IgniteSchema("PUBLIC",
Map.of(table.name(), table), null, -1);
+ private final List<CapturingMailboxRegistry> mailboxes = new ArrayList<>();
+
private TestCluster testCluster;
private List<ExecutionServiceImpl<?>> executionServices;
private PrepareService prepareService;
@@ -146,6 +153,12 @@ public class ExecutionServiceImplTest {
@AfterEach
public void tearDown() throws Exception {
prepareService.stop();
+
+ for (CapturingMailboxRegistry mailbox : mailboxes) {
+ assertTrue(waitForCondition(mailbox::empty, TIMEOUT_IN_MS));
+ }
+
+ mailboxes.clear();
}
/**
@@ -427,7 +440,7 @@ public class ExecutionServiceImplTest {
AsyncCursor<List<Object>> cursor = execService.executePlan(tx, plan,
ctx);
// Wait till the query fails due to nodes' unavailability.
- assertThat(cursor.closeAsync(), willThrow(hasProperty("message",
containsString("Connection refused to ")), 10, TimeUnit.SECONDS));
+ assertThat(cursor.closeAsync(), willThrow(hasProperty("message",
containsString("Unable to send fragment")), 10, TimeUnit.SECONDS));
// Let the root fragment be executed.
queryFailedLatch.countDown();
@@ -455,7 +468,8 @@ public class ExecutionServiceImplTest {
node.dataset(dataPerNode.get(nodeName));
var messageService = node.messageService();
- var mailboxRegistry = new MailboxRegistryImpl();
+ var mailboxRegistry = new CapturingMailboxRegistry(new
MailboxRegistryImpl());
+ mailboxes.add(mailboxRegistry);
var exchangeService = new ExchangeServiceImpl(mailboxRegistry,
messageService);
@@ -729,4 +743,67 @@ public class ExecutionServiceImplTest {
}
};
}
+
+ private static class CapturingMailboxRegistry implements MailboxRegistry {
+ private final MailboxRegistry delegate;
+
+ private final Set<Inbox<?>> inboxes = Collections.newSetFromMap(new
IdentityHashMap<>());
+ private final Set<Outbox<?>> outboxes = Collections.newSetFromMap(new
IdentityHashMap<>());
+
+ CapturingMailboxRegistry(MailboxRegistry delegate) {
+ this.delegate = delegate;
+ }
+
+ boolean empty() {
+ return inboxes.isEmpty() && outboxes.isEmpty();
+ }
+
+ @Override
+ public void start() {
+ delegate.start();
+ }
+
+ @Override
+ public void stop() throws Exception {
+ delegate.stop();
+ }
+
+ @Override
+ public void register(Inbox<?> inbox) {
+ delegate.register(inbox);
+
+ inboxes.add(inbox);
+ }
+
+ @Override
+ public void register(Outbox<?> outbox) {
+ delegate.register(outbox);
+
+ outboxes.add(outbox);
+ }
+
+ @Override
+ public void unregister(Inbox<?> inbox) {
+ delegate.unregister(inbox);
+
+ inboxes.remove(inbox);
+ }
+
+ @Override
+ public void unregister(Outbox<?> outbox) {
+ delegate.unregister(outbox);
+
+ outboxes.remove(outbox);
+ }
+
+ @Override
+ public CompletableFuture<Outbox<?>> outbox(UUID qryId, long
exchangeId) {
+ return delegate.outbox(qryId, exchangeId);
+ }
+
+ @Override
+ public Inbox<?> inbox(UUID qryId, long exchangeId) {
+ return delegate.inbox(qryId, exchangeId);
+ }
+ }
}
