From 0da629ae5a7342ad52b7c0a2cfcf74880890351c Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Fri, 16 Oct 2026 10:55:02 +0300 Subject: [PATCH 1/6] IGNITE-29173 Add kill lock command and locks control after jobs and services --- docs/_docs/distributed-locks.adoc | 12 + .../sql-reference/operational-commands.adoc | 33 + .../CommandHandlerParsingTest.java | 21 + .../util/KillCommandsControlShTest.java | 48 ++ .../internal/management/kill/KillCommand.java | 3 +- .../management/kill/KillLockCommand.java | 66 ++ .../management/kill/KillLockCommandArg.java | 66 ++ .../management/kill/KillLockTask.java | 100 +++ .../processors/cache/CacheLockImpl.java | 36 +- .../cache/GridCacheMvccManager.java | 119 +++ .../processors/job/GridJobWorker.java | 38 + .../service/ClientServiceInvokeRequest.java | 31 + .../service/IgniteServiceProcessor.java | 26 + .../distributed/ExplicitLockKillTest.java | 746 ++++++++++++++++++ .../ignite/testframework/LogListener.java | 2 +- .../junits/GridAbstractTest.java | 6 +- .../testsuites/IgniteCacheTestSuite14.java | 2 + ...mmandHandlerClusterByClassTest_help.output | 7 + ...ndlerClusterByClassWithSSLTest_help.output | 7 + 19 files changed, 1364 insertions(+), 5 deletions(-) create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommand.java create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommandArg.java create mode 100644 modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockTask.java create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java diff --git a/docs/_docs/distributed-locks.adoc b/docs/_docs/distributed-locks.adoc index ab3fca21eb2ba..04b17fde4d800 100644 --- a/docs/_docs/distributed-locks.adoc +++ b/docs/_docs/distributed-locks.adoc @@ -52,6 +52,18 @@ In Ignite, locks are supported only for the `TRANSACTIONAL` atomicity mode, whic `CacheConfiguration.atomicityMode` parameter. ==== +== Lock Ownership + +Explicit locks are bound to the thread that acquired them: the lock must be released by the same thread. Locks acquired by a compute job, +by the `Service.execute()` method, or by a service method invoked remotely or from a thin client, and not released by the time the method +returns are released forcibly and a warning is logged, since otherwise they would be inherited by the next task executed by the same pool +thread or never released at all. Locks acquired by a service method invoked locally through a service proxy belong to the calling thread +and are not released. +Thus, a lock can't be held across job continuations (`ComputeJobContinuation.holdcc()`) or failovers. + +If a lock is released externally (for example, it is killed with the link:sql-reference/operational-commands#kill-lock[KILL LOCK] command), +`Lock.unlock()` throws `IllegalStateException` to signal that the code protected by the lock may have been executed without the lock held. + == Locks and Transactions Explicit locks are not transactional and cannot not be used from within transactions (exception will be thrown). diff --git a/docs/_docs/sql-reference/operational-commands.adoc b/docs/_docs/sql-reference/operational-commands.adoc index 97f56ed422d47..6a3c8abbee293 100644 --- a/docs/_docs/sql-reference/operational-commands.adoc +++ b/docs/_docs/sql-reference/operational-commands.adoc @@ -424,3 +424,36 @@ control.bat --kill CLIENT connection_id [--node-id node_id] Value differs on nodes for the same client. You can always find parameter values with the link:monitoring-metrics/system-views#client_connections[CLIENT_CONNECTIONS] view. * `node_id` - corresponds to the node_id to drop connection from. + +== KILL LOCK + +The `KILL LOCK` command allows you to cancel an explicit cache lock acquired via the `Cache.lock(key)` API. + +Note that killing a lock may affect the execution of the application code that relies on the lock, so the command must be used with caution. + +The command releases the lock on all its keys. If the lock is not acquired yet (the thread is still waiting for it), the lock request is +cancelled and the waiting `Lock.lock()` call fails with `CacheException`. Other threads holding or waiting for the same keys are not affected. +The subsequent `Lock.unlock()` call of the thread that has acquired the killed lock fails with `IllegalStateException`. + +[tabs] +-- + +tab:Unix[] +[source,bash] +---- +./control.sh --kill LOCK xid [--node-id node_id] +---- + +tab:Windows[] +[source,bash] +---- +control.bat --kill LOCK xid [--node-id node_id] +---- + +-- + +=== Parameters + +* `xid` - the lock id (mandatory) that can be retrieved via the link:monitoring-metrics/system-views#cache_explicit_locks[CACHE_EXPLICIT_LOCKS] view. +* `node_id` - the node id (optional). The id of the node where the lock is acquired. If specified, the request is sent only to this node. Otherwise, all nodes are checked. + diff --git a/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java b/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java index 2e5b918f7d1ac..9ed84b06acd5a 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/internal/commandline/CommandHandlerParsingTest.java @@ -75,6 +75,8 @@ import org.apache.ignite.internal.management.kill.KillAllCommand; import org.apache.ignite.internal.management.kill.KillAllCommandArg; import org.apache.ignite.internal.management.kill.KillCommand; +import org.apache.ignite.internal.management.kill.KillLockCommand; +import org.apache.ignite.internal.management.kill.KillLockCommandArg; import org.apache.ignite.internal.management.meta.MetaCommand; import org.apache.ignite.internal.management.meta.MetaRemoveCommand; import org.apache.ignite.internal.management.meta.MetaUpdateCommand; @@ -94,6 +96,7 @@ import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.T2; import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.ListeningTestLogger; import org.apache.ignite.testframework.LogListener; @@ -557,6 +560,8 @@ else if (cmd.getClass() == MetaRemoveCommand.class) cmdText = F.concat(cmdText, "--typeId", "1"); else if (cmd.getClass() == KillAllCommand.class) cmdText = F.concat(cmdText, "SQL"); + else if (cmd.getClass() == KillLockCommand.class) + cmdText = F.concat(cmdText, IgniteUuid.randomUuid().toString()); args = parseArgs(asList(cmdText)); @@ -729,6 +734,22 @@ public void testKillArguments() { assertParseArgsThrows("String representation of \"java.util.UUID\" is exepected", IllegalArgumentException.class, "--kill", "continuous", UUID.randomUUID().toString(), "not_a_uuid"); + // Lock command format errors. + assertParseArgsThrows("Argument xid required.", "--kill", "lock"); + + assertParseArgsThrows("Invalid UUID string: not_a_uuid", IllegalArgumentException.class, + "--kill", "lock", "not_a_uuid"); + + // Lock command parse success. + IgniteUuid xid = IgniteUuid.randomUuid(); + UUID nodeId = UUID.randomUUID(); + + KillLockCommandArg lockArg = (KillLockCommandArg)parseArgs(asList( + "--kill", "lock", xid.toString(), "--node-id", nodeId.toString())).commandArg(); + + assertEquals(xid, lockArg.xid()); + assertEquals(nodeId, lockArg.nodeId()); + // Kill all command format errors. assertParseArgsThrows("Argument target required.", "--kill", "all"); assertParseArgsThrows("Can't parse value 'unknown'", "--kill", "all", "unknown"); diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsControlShTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsControlShTest.java index e195c5ba00144..17322d9bfb536 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsControlShTest.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/KillCommandsControlShTest.java @@ -22,6 +22,7 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.locks.Lock; import java.util.function.Consumer; import org.apache.ignite.IgniteCache; import org.apache.ignite.cache.CacheAtomicityMode; @@ -38,6 +39,7 @@ import org.apache.ignite.internal.util.typedef.T3; import org.apache.ignite.lang.IgnitePredicate; import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.spi.systemview.view.CacheExplicitLockView; import org.apache.ignite.spi.systemview.view.ComputeJobView; import org.apache.ignite.spi.systemview.view.SystemView; import org.apache.ignite.testframework.GridTestUtils; @@ -46,10 +48,12 @@ import org.junit.Test; import static java.util.Arrays.asList; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_CONNECTION_FAILED; import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_INVALID_ARGUMENTS; import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK; import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_UNEXPECTED_ERROR; +import static org.apache.ignite.internal.processors.cache.GridCacheMvccManager.CACHE_EXPLICIT_LOCKS_VIEW; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.AbstractSnapshotSelfTest.doSnapshotCancellationTest; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METRICS; import static org.apache.ignite.internal.processors.job.GridJobProcessor.JOBS_VIEW; @@ -165,6 +169,50 @@ public void testCancelSQLQuery() { }); } + /** @throws Exception If failed. */ + @Test + public void testKillLock() throws Exception { + int key = 0; + + Lock lock = client.cache(DEFAULT_CACHE_NAME).lock(key); + + // Explicit lock is bound to the thread and stays held after the thread finishes. + long threadId = GridTestUtils.runAsync(() -> { + lock.lock(); + + return Thread.currentThread().getId(); + }).get(); + + CacheExplicitLockView view = F.find(client.context().systemView().view(CACHE_EXPLICIT_LOCKS_VIEW), null, + v -> v.threadId() == threadId); + + assertNotNull(view); + + IgniteUuid xid = view.xid(); + + injectTestSystemOut(); + + assertEquals(EXIT_CODE_OK, execute("--kill", "lock", xid.toString())); + + Lock srvLock = srvs.get(0).cache(DEFAULT_CACHE_NAME).lock(key); + + assertTrue(srvLock.tryLock(getTestTimeout(), MILLISECONDS)); + + srvLock.unlock(); + + assertContains(log, testOut.toString(), + "Lock with xid " + xid + " has been killed on node " + client.localNode().id() + '.'); + + String notFoundMsg = "Lock with xid " + xid + " is not found."; + + assertNotContains(log, testOut.toString(), notFoundMsg); + + // The lock is already killed. + assertEquals(EXIT_CODE_OK, execute("--kill", "lock", xid.toString())); + + assertContains(log, testOut.toString(), notFoundMsg); + } + /** @throws Exception If failed. */ @Test public void testCancelContinuousQuery() throws Exception { diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java index 0abc330d95295..aac7531cb49fc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillCommand.java @@ -33,7 +33,8 @@ public KillCommand() { new KillContinuousCommand(), new KillClientCommand(), new KillSnapshotCommand(), - new KillConsistencyCommand() + new KillConsistencyCommand(), + new KillLockCommand() ); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommand.java new file mode 100644 index 0000000000000..983ab5a172479 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommand.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.management.kill; + +import java.util.Collection; +import java.util.UUID; +import java.util.function.Consumer; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.management.api.CommandUtils; +import org.apache.ignite.internal.management.api.ComputeCommand; + +/** + * Kill explicit lock command. + */ +public class KillLockCommand implements ComputeCommand { + /** {@inheritDoc} */ + @Override public String description() { + return "Kill explicit lock by xid"; + } + + /** {@inheritDoc} */ + @Override public Class argClass() { + return KillLockCommandArg.class; + } + + /** {@inheritDoc} */ + @Override public Class taskClass() { + return KillLockTask.class; + } + + /** {@inheritDoc} */ + @Override public Collection nodes(Collection nodes, KillLockCommandArg arg) { + return CommandUtils.nodeOrAll(arg.nodeId(), nodes); + } + + /** {@inheritDoc} */ + @Override public void printResult(KillLockCommandArg arg, UUID res, Consumer printer) { + if (res == null) { + printer.accept("Lock with xid " + arg.xid() + " is not found" + + (arg.nodeId() == null ? "" : " on node " + arg.nodeId()) + '.'); + } + else + printer.accept("Lock with xid " + arg.xid() + " has been killed on node " + res + '.'); + } + + /** {@inheritDoc} */ + @Override public String confirmationPrompt(KillLockCommandArg arg) { + return "Warning: the command will kill explicit lock with xid " + arg.xid() + + ". Killing a lock may affect the execution of application code that relies on it."; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommandArg.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommandArg.java new file mode 100644 index 0000000000000..6e903b2f913ee --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockCommandArg.java @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.management.kill; + +import java.util.UUID; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.apache.ignite.internal.management.api.Argument; +import org.apache.ignite.internal.management.api.CliConfirmArgument; +import org.apache.ignite.internal.management.api.Positional; +import org.apache.ignite.lang.IgniteUuid; + +/** + * Arguments for --kill lock command. + */ +@CliConfirmArgument +public class KillLockCommandArg extends IgniteDataTransferObject { + /** */ + private static final long serialVersionUID = 0; + + /** */ + @Order(0) + @Positional + @Argument(description = "Lock identifier (xid from CACHE_EXPLICIT_LOCKS view)") + IgniteUuid xid; + + /** */ + @Order(1) + @Argument(description = "Node id the lock is acquired on", optional = true) + UUID nodeId; + + /** */ + public IgniteUuid xid() { + return xid; + } + + /** */ + public void xid(IgniteUuid xid) { + this.xid = xid; + } + + /** */ + public UUID nodeId() { + return nodeId; + } + + /** */ + public void nodeId(UUID nodeId) { + this.nodeId = nodeId; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockTask.java new file mode 100644 index 0000000000000..326724bf196e1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/kill/KillLockTask.java @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.management.kill; + +import java.util.List; +import java.util.UUID; +import org.apache.ignite.IgniteException; +import org.apache.ignite.compute.ComputeJobResult; +import org.apache.ignite.internal.binary.BinaryUtils; +import org.apache.ignite.internal.processors.cache.GridCacheExplicitLockSpan; +import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate; +import org.apache.ignite.internal.processors.cache.GridCacheMvccManager; +import org.apache.ignite.internal.processors.task.GridInternal; +import org.apache.ignite.internal.visor.VisorJob; +import org.apache.ignite.internal.visor.VisorMultiNodeTask; +import org.jetbrains.annotations.Nullable; + +/** + * Task to kill explicit cache locks. + */ +@GridInternal +public class KillLockTask extends VisorMultiNodeTask { + /** */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override protected VisorJob job(KillLockCommandArg arg) { + return new KillLockJob(arg, debug); + } + + /** {@inheritDoc} */ + @Nullable @Override protected UUID reduce0(List results) throws IgniteException { + IgniteException err = null; + + for (ComputeJobResult res : results) { + // The lock is held on one node only, failures on other nodes don't matter if it is killed. + if (res.getException() == null) { + if (res.getData() != null) + return res.getData(); + } + else if (err == null) + err = new IgniteException("Failed to kill lock on node: " + res.getNode().id(), res.getException()); + } + + if (err != null) + throw err; + + return null; + } + + /** + * Job to kill explicit lock on node. Returns ID of the node if the lock is found and killed. + */ + private static class KillLockJob extends VisorJob { + /** */ + private static final long serialVersionUID = 0L; + + /** + * Create job with specified argument. + * + * @param arg Job argument. + * @param debug Flag indicating whether debug information should be printed into node log. + */ + protected KillLockJob(KillLockCommandArg arg, boolean debug) { + super(arg, debug); + } + + /** {@inheritDoc} */ + @Nullable @Override protected UUID run(KillLockCommandArg arg) throws IgniteException { + GridCacheMvccManager mvcc = ignite.context().cache().context().mvcc(); + + for (GridCacheExplicitLockSpan span : mvcc.activeExplicitLocks()) { + for (GridCacheMvccCandidate cand : span.candidates()) { + if (arg.xid().equals(BinaryUtils.asIgniteUuid(cand.version()))) { + // Xid identifies the lock of one thread, all keys of the lock are released at once. + return mvcc.releaseExplicitLock(cand.threadId(), cand.version()) ? + ignite.localNode().id() : null; + } + } + } + + return null; + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java index c8ac3470270c7..6bcb094ea9e88 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java @@ -34,6 +34,10 @@ * */ class CacheLockImpl implements Lock { + /** */ + private static final String LOCK_CANCELLED_MSG = "Failed to acquire lock, the lock request has been cancelled " + + "(was the lock killed?)."; + /** Gateway. */ private final GridCacheGateway gate; @@ -76,7 +80,8 @@ class CacheLockImpl implements Lock { try { checkTx(); - delegate.lockAll(keys, 0); + if (!delegate.lockAll(keys, 0)) + throw new CacheException(LOCK_CANCELLED_MSG); incrementLockCounter(); } @@ -104,7 +109,9 @@ private void incrementLockCounter() { /** {@inheritDoc} */ @Override public void lockInterruptibly() throws InterruptedException { - tryLock(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + // The lock can't be timed out, so it is not acquired only if it is cancelled. + if (!tryLock(Long.MAX_VALUE, TimeUnit.MILLISECONDS)) + throw new CacheException(LOCK_CANCELLED_MSG); } /** {@inheritDoc} */ @@ -192,6 +199,8 @@ private void incrementLockCounter() { assert cntr > 0; + boolean held = lockedByCurrentThread(); + cntr--; if (cntr == 0) { @@ -213,6 +222,12 @@ private void incrementLockCounter() { } delegate.unlockAll(keys); + + // The lock state of the instance is reset above, so it stays usable after the exception. + if (!held) { + throw new IllegalStateException("Failed to unlock keys, the lock has been released externally " + + "(was the lock killed?)."); + } } catch (IgniteCheckedException e) { throw new CacheException(e.getMessage(), e); @@ -222,6 +237,23 @@ private void incrementLockCounter() { } } + /** + * @return {@code True} if all the keys are still locked by the current thread, {@code false} if the lock + * has been released externally. + */ + private boolean lockedByCurrentThread() { + GridCacheContext cctx = delegate.context(); + + long threadId = Thread.currentThread().getId(); + + for (K key : keys) { + if (cctx.mvcc().explicitLock(threadId, cctx.txKey(cctx.toCacheKeyObject(key))) == null) + return false; + } + + return true; + } + /** {@inheritDoc} */ @NotNull @Override public Condition newCondition() { throw new UnsupportedOperationException(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java index 2a82a6a6ed45f..fda2e031abc45 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java @@ -48,6 +48,8 @@ import org.apache.ignite.internal.processors.cache.distributed.GridCacheMappedVersion; import org.apache.ignite.internal.processors.cache.distributed.GridDistributedCacheEntry; import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishFuture; +import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedLockFuture; +import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockFuture; import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx; import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; @@ -1119,6 +1121,123 @@ public GridCacheMvccCandidate removeExplicitLock(long threadId, return span != null ? span.topologyVersion() : null; } + /** + * @param threadId Thread ID. + * @return Versions of explicit locks acquired or awaited by the thread. + */ + public Set explicitLockVersions(long threadId) { + GridCacheExplicitLockSpan span = pendingExplicit.get(threadId); + + if (span == null) + return Collections.emptySet(); + + Set vers = new HashSet<>(); + + for (GridCacheMvccCandidate cand : span.candidates()) + vers.add(cand.version()); + + return vers; + } + + /** + * Forcibly releases explicit locks of the thread except the given ones. + *

+ * The method is usually called when a job or a service has finished, and the current thread can be interrupted + * at this moment (for example, the job or the service has been cancelled). Locks are removed locally first and + * then unlock requests are sent to primary nodes, so the interruption flag is cleared for the time of the release: + * otherwise a request can fail to be sent and the lock is left on the primary node with no way to release it. + * + * @param threadId Thread ID. + * @param exclude Versions of locks to keep. + * @return Versions of released locks. + */ + public Collection releaseExplicitLocks(long threadId, Set exclude) { + Set vers = explicitLockVersions(threadId); + + if (vers.isEmpty()) + return Collections.emptySet(); + + vers.removeAll(exclude); + + if (vers.isEmpty()) + return vers; + + boolean interrupted = Thread.interrupted(); + + try { + for (GridCacheVersion ver : vers) + releaseExplicitLock(threadId, ver); + } + finally { + if (interrupted) + Thread.currentThread().interrupt(); + } + + return vers; + } + + /** + * Forcibly releases explicit lock acquired or awaited by the thread on all keys of the lock. + *

+ * The lock future is cancelled first: if the lock is not acquired yet (or acquired on a part of the keys only), + * the future undoes the lock itself, otherwise cancellation has no effect and the acquired lock is removed after. + * Decision can't be made by the {@link GridCacheMvccCandidate#owner()} flag of the candidate since the lock can + * be concurrently acquired. + * + * @param threadId Thread ID. + * @param ver Lock version. + * @return {@code True} if the lock was found. + */ + public boolean releaseExplicitLock(long threadId, GridCacheVersion ver) { + GridCacheExplicitLockSpan span = pendingExplicit.get(threadId); + + if (span == null) + return false; + + // Candidates of reentries have the same version, each of them is removed by a separate call. + List cands = new ArrayList<>(); + + for (GridCacheMvccCandidate cand : span.candidates()) { + if (cand.version().equals(ver)) + cands.add(cand); + } + + if (cands.isEmpty()) + return false; + + Collection> futs = futuresForVersion(ver); + + if (futs != null) { + for (GridCacheVersionedFuture fut : futs) { + if (fut instanceof GridNearLockFuture || fut instanceof GridDhtColocatedLockFuture) { + try { + fut.cancel(); + } + catch (IgniteCheckedException e) { + U.error(log, "Failed to cancel explicit lock future: " + fut, e); + } + } + } + } + + for (GridCacheMvccCandidate cand : cands) { + GridCacheContext cacheCtx = cctx.cacheContext(cand.key().cacheId()); + + // Cache context may be absent if the cache was stopped concurrently. + if (cacheCtx == null) + continue; + + Collection keys = Collections.singleton(cand.key().key()); + + if (cacheCtx.isNear()) + cacheCtx.nearTx().removeLocks(ver, keys); + else if (cacheCtx.isColocated()) + cacheCtx.colocated().removeLocks(threadId, ver, keys); + } + + return true; + } + /** {@inheritDoc} */ @Override public void printMemoryStats() { X.println(">>> "); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java index 7dfc10746d294..306e4dd2ca2dc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.Collection; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.atomic.AtomicBoolean; @@ -49,7 +50,9 @@ import org.apache.ignite.internal.managers.communication.CommunicationMarshalling; import org.apache.ignite.internal.managers.deployment.GridDeployment; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.internal.processors.cache.GridCacheMvccManager; import org.apache.ignite.internal.processors.cache.distributed.dht.GridReservable; +import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.query.GridQueryProcessor; import org.apache.ignite.internal.processors.security.SecurityContext; import org.apache.ignite.internal.processors.security.SecurityUtils; @@ -584,6 +587,10 @@ private void execute0(boolean skipNtf) { else { res = U.wrapThreadLoader(dep.classLoader(), new Callable() { @Nullable @Override public Object call() { + // The job can be executed by a thread that already holds explicit locks. + Set locksBefore = locksManager().explicitLockVersions( + Thread.currentThread().getId()); + try { if (internal && ctx.config().isPeerClassLoadingEnabled()) ctx.job().internal(true); @@ -593,6 +600,8 @@ private void execute0(boolean skipNtf) { finally { if (internal && ctx.config().isPeerClassLoadingEnabled()) ctx.job().internal(false); + + releaseLeftExplicitLocks(locksBefore); } } }); @@ -673,6 +682,35 @@ else if (log.isDebugEnabled()) } } + /** @return Locks manager. */ + private GridCacheMvccManager locksManager() { + return ctx.cache().context().mvcc(); + } + + /** + * Releases explicit locks acquired but not released by the job. Locks are bound to the thread, so locks left + * unreleased would be inherited by the next job executed by the thread or never released if the thread is + * terminated. A held job (see {@link org.apache.ignite.compute.ComputeJobContinuation#holdcc()}) may be continued + * by another thread, so locks can't be held across continuation calls either. + * + * @param locksBefore Versions of explicit locks held by the thread before the job execution. + */ + private void releaseLeftExplicitLocks(Set locksBefore) { + try { + Collection released = locksManager().releaseExplicitLocks( + Thread.currentThread().getId(), locksBefore); + + if (!released.isEmpty()) { + U.warn(log, "Job finished without releasing explicit locks, the locks have been released forcibly " + + "[taskName=" + ses.getTaskName() + ", jobId=" + ses.getJobId() + ", lockVers=" + released + ']'); + } + } + catch (Exception e) { + U.error(log, "Failed to release explicit locks left by job [taskName=" + ses.getTaskName() + + ", jobId=" + ses.getJobId() + ']', e); + } + } + /** * Handles {@link Throwable} generic exception for task * deployment and execution. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java index 3c10a8231fde9..9b62f13ed07ee 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java @@ -21,6 +21,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.Map; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -31,6 +32,7 @@ import org.apache.ignite.internal.binary.BinaryReaderEx; import org.apache.ignite.internal.binary.BinaryUtils; import org.apache.ignite.internal.cluster.ClusterGroupAdapter; +import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.platform.PlatformNativeException; import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext; import org.apache.ignite.internal.processors.platform.client.ClientObjectResponse; @@ -42,6 +44,7 @@ import org.apache.ignite.internal.processors.service.GridServiceProxy; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.services.Service; import org.apache.ignite.services.ServiceDescriptor; @@ -199,6 +202,34 @@ public ClientServiceInvokeRequest(BinaryReaderEx reader, ClientProtocolContext p catch (Throwable e) { throw new IgniteException(e); } + finally { + releaseLeftExplicitLocks(ctx); + } + } + + /** + * Releases explicit locks acquired but not released by the service method invoked locally (a service deployed + * on the local node is invoked in the client connector thread). Locks are bound to the thread, so locks left + * unreleased would be inherited by the next request processed by the thread. The client connector thread can't + * acquire explicit locks outside of service invocations, so all locks held by the thread are released. + * + * @param ctx Connection context. + */ + private void releaseLeftExplicitLocks(ClientConnectionContext ctx) { + try { + Collection released = ctx.kernalContext().cache().context().mvcc() + .releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); + + if (!released.isEmpty()) { + U.warn(ctx.kernalContext().log(getClass()), "Service method invocation finished without releasing " + + "explicit locks, the locks have been released forcibly [name=" + name + ", method=" + methodName + + ", lockVers=" + released + ']'); + } + } + catch (Exception e) { + U.error(ctx.kernalContext().log(getClass()), "Failed to release explicit locks left by service method " + + "[name=" + name + ", method=" + methodName + ']', e); + } } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java index 2c6f7f9a466eb..4f6b89032a31b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java @@ -62,6 +62,7 @@ import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch; import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest; import org.apache.ignite.internal.processors.cache.ValidationOnNodeJoinUtils; +import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage; import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState; import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport; @@ -1381,6 +1382,8 @@ else if (ctxs.size() < assignCnt) { throw (Error)e; } finally { + releaseLeftExplicitLocks(srvcCtx); + // Suicide. exe.shutdownNow(); } @@ -1389,6 +1392,29 @@ else if (ctxs.size() < assignCnt) { } } + /** + * Releases explicit locks acquired but not released by the service execution. Locks are bound to the thread, + * so locks left unreleased by the terminating service thread would never be released. + * + * @param srvcCtx Service context. + */ + private void releaseLeftExplicitLocks(ServiceContextImpl srvcCtx) { + try { + Collection released = ctx.cache().context().mvcc() + .releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); + + if (!released.isEmpty()) { + U.warn(log, "Service execution finished without releasing explicit locks, the locks have been " + + "released forcibly [name=" + srvcCtx.name() + ", execId=" + srvcCtx.executionId() + + ", lockVers=" + released + ']'); + } + } + catch (Exception e) { + U.error(log, "Failed to release explicit locks left by service [name=" + srvcCtx.name() + + ", execId=" + srvcCtx.executionId() + ']', e); + } + } + /** * @param cfg Service configuration. * @param svcCtx Service context to be injected into the service. diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java new file mode 100644 index 0000000000000..88b447baafb10 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java @@ -0,0 +1,746 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cache.distributed; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.locks.Lock; +import javax.cache.CacheException; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.Ignition; +import org.apache.ignite.client.IgniteClient; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.ClientConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.configuration.NearCacheConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.management.kill.KillLockCommandArg; +import org.apache.ignite.internal.management.kill.KillLockTask; +import org.apache.ignite.internal.processors.cache.IgniteCacheProxy; +import org.apache.ignite.internal.processors.job.GridJobWorker; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.G; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.internal.visor.VisorTaskArgument; +import org.apache.ignite.lang.IgniteRunnable; +import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.resources.IgniteInstanceResource; +import org.apache.ignite.services.Service; +import org.apache.ignite.spi.communication.tcp.internal.ConnectionClientPool; +import org.apache.ignite.spi.systemview.view.CacheExplicitLockView; +import org.apache.ignite.spi.systemview.view.SystemView; +import org.apache.ignite.testframework.ListeningTestLogger; +import org.apache.ignite.testframework.LogListener; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; +import static org.apache.ignite.internal.processors.cache.GridCacheMvccManager.CACHE_EXPLICIT_LOCKS_VIEW; +import static org.apache.ignite.internal.processors.odbc.ClientListenerProcessor.CLIENT_LISTENER_PORT; +import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** + * Tests forcible release of explicit locks: kill of a lock and release of locks left by jobs and services. + */ +@RunWith(Parameterized.class) +public class ExplicitLockKillTest extends GridCommonAbstractTest { + /** */ + private static final long TIMEOUT = 10_000L; + + /** */ + private static final int SRV_CNT = 3; + + /** */ + private static final String UNLOCK_ERR_MSG = "the lock has been released externally"; + + /** */ + private static final String LOCK_ERR_MSG = "the lock request has been cancelled"; + + /** Near cache flag. */ + @Parameterized.Parameter + public boolean near; + + /** */ + private final ListeningTestLogger testLog = new ListeningTestLogger(log); + + /** Lock owner node. */ + private IgniteEx client; + + /** Executors of threads the locks are bound to. */ + private final List executors = new ArrayList<>(); + + /** */ + @Parameterized.Parameters(name = "near={0}") + public static Collection parameters() { + return Arrays.asList(false, true); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + if (getTestIgniteInstanceName(0).equals(igniteInstanceName)) + cfg.setGridLogger(testLog); + + CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAtomicityMode(TRANSACTIONAL) + .setBackups(1); + + if (near) + ccfg.setNearConfiguration(new NearCacheConfiguration<>()); + + return cfg.setCacheConfiguration(ccfg); + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + clearStaticLog(GridJobWorker.class); + + startGrids(SRV_CNT); + + client = startClientGrid(SRV_CNT); + + if (near) + client.getOrCreateNearCache(DEFAULT_CACHE_NAME, new NearCacheConfiguration<>()); + + awaitPartitionMapExchange(); + + for (Ignite node : G.allGrids()) + assertEquals(near, ((IgniteCacheProxy)node.cache(DEFAULT_CACHE_NAME)).context().isNear()); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + for (ExecutorService exec : executors) + exec.shutdownNow(); + + executors.clear(); + + testLog.clearListeners(); + + stopAllGrids(); + + super.afterTest(); + } + + /** */ + @Test + public void testKillAcquiredLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + ExecutorService exec = createExecutor(); + + long threadId = threadId(exec); + + Lock lock = cache().lock(key); + + exec.submit(lock::lock).get(); + + IgniteUuid xid = awaitXid(threadId); + + assertTrue(killLock(xid)); + + // The lock is already killed. + assertFalse(killLock(xid)); + + assertLockFree(key); + + assertThrowsAnyCause(log, () -> exec.submit(lock::unlock).get(), IllegalStateException.class, UNLOCK_ERR_MSG); + + // The lock instance is still usable after the failed unlock. + exec.submit(() -> { + lock.lock(); + lock.unlock(); + }).get(); + + assertLockFree(key); + } + + /** */ + @Test + public void testKillUnknownLock() throws Exception { + assertFalse(killLock(IgniteUuid.randomUuid())); + } + + /** */ + @Test + public void testKillReentrantLock() throws Exception { + int key1 = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + int key2 = primaryKey(grid(2).cache(DEFAULT_CACHE_NAME)); + + ExecutorService exec = createExecutor(); + + long threadId = threadId(exec); + + Lock lock = cache().lock(key1); + + // Key 1 is reentered by both locks while key 2 is locked with another lock version. + Lock lockAll = cache().lockAll(Arrays.asList(key1, key2)); + + exec.submit(() -> { + lock.lock(); + lock.lock(); + }).get(); + + IgniteUuid xid = awaitXid(threadId); + + exec.submit(lockAll::lock).get(); + + assertTrue(killLock(xid)); + + assertLockFree(key1); + assertLockHeld(key2); + + // Key 2 is released even though the unlock fails due to the killed key 1. + assertThrowsAnyCause(log, () -> exec.submit(lockAll::unlock).get(), IllegalStateException.class, + UNLOCK_ERR_MSG); + + assertLockFree(key2); + + for (int i = 0; i < 2; i++) { + assertThrowsAnyCause(log, () -> exec.submit(lock::unlock).get(), IllegalStateException.class, + UNLOCK_ERR_MSG); + } + + assertTrue(threadLocks(threadId).isEmpty()); + } + + /** */ + @Test + public void testKillPendingLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + ExecutorService holderExec = createExecutor(); + ExecutorService waiterExec1 = createExecutor(); + ExecutorService waiterExec2 = createExecutor(); + + long waiterThreadId1 = threadId(waiterExec1); + long waiterThreadId2 = threadId(waiterExec2); + + Lock holderLock = cache().lock(key); + + holderExec.submit(holderLock::lock).get(); + + Lock waiterLock1 = cache().lock(key); + Lock waiterLock2 = cache().lock(key); + + Future waitFut1 = waiterExec1.submit(waiterLock1::lock); + Future waitFut2 = waiterExec2.submit(() -> { + waiterLock2.lockInterruptibly(); + + return null; + }); + + IgniteUuid waiterXid1 = awaitXid(waiterThreadId1); + IgniteUuid waiterXid2 = awaitXid(waiterThreadId2); + + assertFalse(waitFut1.isDone()); + assertFalse(waitFut2.isDone()); + + assertTrue(killLock(waiterXid1)); + + assertThrowsAnyCause(log, () -> waitFut1.get(TIMEOUT, MILLISECONDS), CacheException.class, LOCK_ERR_MSG); + + assertTrue(killLock(waiterXid2)); + + assertThrowsAnyCause(log, () -> waitFut2.get(TIMEOUT, MILLISECONDS), CacheException.class, LOCK_ERR_MSG); + + assertTrue(threadLocks(waiterThreadId1).isEmpty()); + assertTrue(threadLocks(waiterThreadId2).isEmpty()); + + // The lock owner is not affected. + assertLockHeld(key); + + holderExec.submit(holderLock::unlock).get(); + + assertLockFree(key); + } + + /** */ + @Test + public void testKillPartiallyAcquiredLock() throws Exception { + int key1 = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + int key2 = primaryKey(grid(2).cache(DEFAULT_CACHE_NAME)); + + ExecutorService holderExec = createExecutor(); + ExecutorService waiterExec = createExecutor(); + + long waiterThreadId = threadId(waiterExec); + + Lock holderLock = cache().lock(key1); + + holderExec.submit(holderLock::lock).get(); + + // Keys are locked one by one in the given order, so key 2 is acquired while key 1 is awaited. + Lock waiterLock = cache().lockAll(Arrays.asList(key2, key1)); + + Future waitFut = waiterExec.submit(waiterLock::lock); + + IgniteUuid waiterXid = awaitXid(waiterThreadId); + + assertTrue(waitForCondition(() -> lockedInCluster(key2), TIMEOUT)); + + assertFalse(waitFut.isDone()); + + assertTrue(killLock(waiterXid)); + + assertThrowsAnyCause(log, () -> waitFut.get(TIMEOUT, MILLISECONDS), CacheException.class, + LOCK_ERR_MSG); + + assertLockFree(key2); + assertLockHeld(key1); + + holderExec.submit(holderLock::unlock).get(); + + assertLockFree(key1); + } + + /** + * Internal jobs mapped to the local node are executed synchronously in the thread that started the task, so locks + * held by the thread before the job execution must not be released. + */ + @Test + public void testLocalInternalJobKeepsLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + LogListener lsnr = LogListener.matches("Job finished without releasing explicit locks").build(); + + testLog.registerListener(lsnr); + + // Size task is internal and is mapped to data nodes, so the size job is executed in the user thread. + IgniteCache cache = grid(0).cache(DEFAULT_CACHE_NAME); + + ExecutorService exec = createExecutor(); + + Lock lock = cache.lock(key); + + exec.submit(() -> { + lock.lock(); + + try { + cache.size(); + + assertTrue(cache.isLocalLocked(key, true)); + } + finally { + lock.unlock(); + } + }).get(); + + assertLockFree(key); + + // The same for the size job executed in the thread of the job holding the lock. + client.compute(client.cluster().forNodeId(grid(0).localNode().id())).run(new LockSizeJob(key)); + + assertLockFree(key); + + assertFalse(lsnr.check()); + } + + /** */ + @Test + public void testJobLeftLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + LogListener lsnr = LogListener.matches("Job finished without releasing explicit locks").build(); + + testLog.registerListener(lsnr); + + client.compute(client.cluster().forNodeId(grid(0).localNode().id())).run(new LockJob(key)); + + assertTrue(lsnr.check()); + + assertLockFree(key); + + // The lock acquired and released by the job is not reported. + LogListener lsnr2 = LogListener.matches("Job finished without releasing explicit locks").build(); + + testLog.registerListener(lsnr2); + + client.compute(client.cluster().forNodeId(grid(0).localNode().id())).run(new LockUnlockJob(key)); + + assertFalse(lsnr2.check()); + } + + /** + * Locks left by a cancelled job are released while the job thread is interrupted. Lock is removed locally first and + * then the unlock request is sent to the primary node, so the request must not fail due to the interruption. + */ + @Test + public void testInterruptedJobLeftLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + LogListener lsnr = LogListener.matches("Job finished without releasing explicit locks").build(); + + testLog.registerListener(lsnr); + + // The job is executed locally, so its result is not affected by the interruption of the job thread. + grid(0).compute(grid(0).cluster().forLocal()).run(new InterruptedLockJob(key, grid(1).localNode().id())); + + assertTrue(lsnr.check()); + + assertLockFree(key); + } + + /** */ + @Test + public void testServiceLeftLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + LogListener lsnr = LogListener.matches("Service execution finished without releasing explicit locks").build(); + + testLog.registerListener(lsnr); + + grid(0).services(grid(0).cluster().forLocal()).deployNodeSingleton("lock-service", new LockService(key)); + + assertTrue(lsnr.check(TIMEOUT)); + + assertLockFree(key); + } + + /** */ + @Test + public void testServiceMethodLeftLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + // A remote service method is invoked by the service proxy as a compute job (ServiceProxyCallable), so the left + // lock is released and reported by the job worker. + LogListener lsnr = LogListener.matches("Job finished without releasing explicit locks").build(); + + testLog.registerListener(lsnr); + + grid(0).services(grid(0).cluster().forLocal()).deployNodeSingleton("lock-method-service", + new LockMethodServiceImpl()); + + client.services().serviceProxy("lock-method-service", LockMethodService.class, false).lock(key); + + assertTrue(lsnr.check()); + + assertLockFree(key); + } + + /** */ + @Test + public void testThinClientServiceMethodLeftLock() throws Exception { + int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + LogListener lsnr = LogListener.matches("Service method invocation finished without releasing explicit locks") + .build(); + + testLog.registerListener(lsnr); + + grid(0).services(grid(0).cluster().forLocal()).deployNodeSingleton("lock-method-service", + new LockMethodServiceImpl()); + + // The thin client is connected only to the node the service is deployed on, so the method is invoked locally + // in the client connector thread. + try (IgniteClient thinClient = Ignition.startClient(new ClientConfiguration() + .setAddresses("127.0.0.1:" + grid(0).localNode().attribute(CLIENT_LISTENER_PORT)) + .setPartitionAwarenessEnabled(false) + .setClusterDiscoveryEnabled(false)) + ) { + thinClient.services().serviceProxy("lock-method-service", LockMethodService.class).lock(key); + } + + assertTrue(lsnr.check()); + + assertLockFree(key); + } + + /** @return Cache of the lock owner node. */ + private IgniteCache cache() { + return client.cache(DEFAULT_CACHE_NAME); + } + + /** @return Single thread executor the locks are bound to. */ + private ExecutorService createExecutor() { + ExecutorService exec = Executors.newSingleThreadExecutor(); + + executors.add(exec); + + return exec; + } + + /** @return ID of the executor thread. */ + private static long threadId(ExecutorService exec) throws Exception { + return exec.submit(() -> Thread.currentThread().getId()).get(); + } + + /** + * @param threadId Thread ID. + * @return Explicit locks of the thread. + */ + private List threadLocks(long threadId) { + SystemView view = client.context().systemView().view(CACHE_EXPLICIT_LOCKS_VIEW); + + return U.arrayList(view.iterator(), view.size(), v -> v.threadId() == threadId); + } + + /** + * @param threadId Thread ID. + * @return Xid of the lock acquired or awaited by the thread. + */ + private IgniteUuid awaitXid(long threadId) throws Exception { + assertTrue(waitForCondition(() -> !threadLocks(threadId).isEmpty(), TIMEOUT)); + + return threadLocks(threadId).get(0).xid(); + } + + /** + * @param xid Lock xid. + * @return {@code True} if the lock was found and killed on the lock owner (client) node. + */ + private boolean killLock(IgniteUuid xid) throws Exception { + KillLockCommandArg arg = new KillLockCommandArg(); + + arg.xid(xid); + + VisorTaskArgument taskArg = + new VisorTaskArgument<>(F.viewReadOnly(grid(0).cluster().nodes(), ClusterNode::id), arg, false); + + // Lock owner is a client node, so the task is executed on all nodes, not only on servers. + UUID nodeId = grid(0).compute(grid(0).cluster()).execute(KillLockTask.class, taskArg).result(); + + if (nodeId == null) + return false; + + assertEquals(client.localNode().id(), nodeId); + + return true; + } + + /** @param key Key that must be free in the cluster. */ + private void assertLockFree(int key) throws Exception { + Lock lock = grid(SRV_CNT - 1).cache(DEFAULT_CACHE_NAME).lock(key); + + // Lock release is not synchronous with the kill or job completion, so wait for it. + assertTrue(lock.tryLock(TIMEOUT, MILLISECONDS)); + + lock.unlock(); + } + + /** @param key Key that must be locked in the cluster. */ + private void assertLockHeld(int key) { + assertTrue(lockedInCluster(key)); + } + + /** + * @param key Key. + * @return {@code True} if the key is locked in the cluster. + */ + private boolean lockedInCluster(int key) { + Lock lock = grid(0).cache(DEFAULT_CACHE_NAME).lock(key); + + if (!lock.tryLock()) + return true; + + lock.unlock(); + + return false; + } + + /** Job that locks a key and does not unlock it. */ + private static class LockJob implements IgniteRunnable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + private final int key; + + /** @param key Key. */ + LockJob(int key) { + this.key = key; + } + + /** {@inheritDoc} */ + @Override public void run() { + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); + } + } + + /** Job that gets cache size while holding a lock and checks that the lock is still held after that. */ + private static class LockSizeJob implements IgniteRunnable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + private final int key; + + /** @param key Key. */ + LockSizeJob(int key) { + this.key = key; + } + + /** {@inheritDoc} */ + @Override public void run() { + IgniteCache cache = ignite.cache(DEFAULT_CACHE_NAME); + + Lock lock = cache.lock(key); + + lock.lock(); + + try { + cache.size(); + + assertTrue(cache.isLocalLocked(key, true)); + } + finally { + // Fails if the lock is released by the size job. + lock.unlock(); + } + } + } + + /** + * Job that locks a key and finishes with the interrupted thread as a cancelled job does. Connections to the primary + * node are closed, so the unlock request requires a new connection to be established by the interrupted thread. + */ + private static class InterruptedLockJob implements IgniteRunnable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + private final int key; + + /** */ + private final UUID primaryId; + + /** + * @param key Key. + * @param primaryId Primary node ID for the key. + */ + InterruptedLockJob(int key, UUID primaryId) { + this.key = key; + this.primaryId = primaryId; + } + + /** {@inheritDoc} */ + @Override public void run() { + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); + + ConnectionClientPool connPool = U.field(ignite.configuration().getCommunicationSpi(), "clientPool"); + + try { + connPool.forceCloseConnection(primaryId); + } + catch (IgniteCheckedException e) { + throw new IgniteException(e); + } + + Thread.currentThread().interrupt(); + } + } + + /** Job that locks and unlocks a key. */ + private static class LockUnlockJob implements IgniteRunnable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + private final int key; + + /** @param key Key. */ + LockUnlockJob(int key) { + this.key = key; + } + + /** {@inheritDoc} */ + @Override public void run() { + Lock lock = ignite.cache(DEFAULT_CACHE_NAME).lock(key); + + lock.lock(); + lock.unlock(); + } + } + + /** */ + public interface LockMethodService { + /** @param key Key to lock and not unlock. */ + public void lock(int key); + } + + /** Service with a method that locks a key and does not unlock it. */ + public static class LockMethodServiceImpl implements LockMethodService, Service { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** {@inheritDoc} */ + @Override public void lock(int key) { + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); + } + } + + /** Service that locks a key in {@link #execute()} and does not unlock it. */ + private static class LockService implements Service { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + private final int key; + + /** @param key Key. */ + LockService(int key) { + this.key = key; + } + + /** {@inheritDoc} */ + @Override public void execute() { + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/LogListener.java b/modules/core/src/test/java/org/apache/ignite/testframework/LogListener.java index 3c62e12163e20..7255b240b82cc 100644 --- a/modules/core/src/test/java/org/apache/ignite/testframework/LogListener.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/LogListener.java @@ -81,7 +81,7 @@ public boolean check(long millis) throws InterruptedException { if (check()) return true; - Thread.sleep(1000); + Thread.sleep(200); } return check(); diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java index 563b246fb8d81..4944464fe01ea 100755 --- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/GridAbstractTest.java @@ -3067,7 +3067,11 @@ private void clearStaticClassLog(Class cls) { assertNotNull(cls); ((AtomicReference)getFieldValue(cls, "logRef")).set(null); - setFieldValue(cls, "log", null); + + Field logField = U.findField(cls, "log"); + + if (logField != null && Modifier.isStatic(logField.getModifiers())) + setFieldValue(cls, "log", null); } /** diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite14.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite14.java index 76aafeb28bf54..b48b4ef05953e 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite14.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite14.java @@ -44,6 +44,7 @@ import org.apache.ignite.internal.processors.cache.distributed.CacheLoadingConcurrentGridStartSelfTest; import org.apache.ignite.internal.processors.cache.distributed.CacheLockReleaseNodeLeaveTest; import org.apache.ignite.internal.processors.cache.distributed.CacheTxLoadingConcurrentGridStartSelfTestAllowOverwrite; +import org.apache.ignite.internal.processors.cache.distributed.ExplicitLockKillTest; import org.apache.ignite.internal.processors.cache.distributed.GridCachePartitionedNearDisabledTxMultiThreadedSelfTest; import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheClientNodePartitionsExchangeTest; import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheServerNodeConcurrentStart; @@ -129,6 +130,7 @@ public static List> suite(Collection ignoredTests) { GridTestUtils.addTestIfNeeded(suite, CacheLockReleaseNodeLeaveTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, ExplicitLockCancelOnNodeStopTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, ExplicitLockDumpTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, ExplicitLockKillTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCachePartitionedNestedTxTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCachePartitionedTxConcurrentGetTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCachePartitionedTxReadTest.class, ignoredTests); diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output index ed26dbae2dbf8..fcae8c518d034 100644 --- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output +++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassTest_help.output @@ -228,6 +228,13 @@ This utility can do the following commands: Kill consistency task: control.(sh|bat) --kill consistency + Kill explicit lock by xid: + control.(sh|bat) --kill lock xid [--node-id node_id] [--yes] + + Parameters: + xid - Lock identifier (xid from CACHE_EXPLICIT_LOCKS view). + --node-id node_id - Node id the lock is acquired on. + Create cluster snapshot: control.(sh|bat) --snapshot create snapshot_name [--dest path] [--sync] [--incremental] diff --git a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output index 1173917caff49..301b5563e656f 100644 --- a/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output +++ b/modules/core/src/test/resources/org.apache.ignite.util/GridCommandHandlerClusterByClassWithSSLTest_help.output @@ -228,6 +228,13 @@ This utility can do the following commands: Kill consistency task: control.(sh|bat) --kill consistency + Kill explicit lock by xid: + control.(sh|bat) --kill lock xid [--node-id node_id] [--yes] + + Parameters: + xid - Lock identifier (xid from CACHE_EXPLICIT_LOCKS view). + --node-id node_id - Node id the lock is acquired on. + Create cluster snapshot: control.(sh|bat) --snapshot create snapshot_name [--dest path] [--sync] [--incremental] From f47a109ec815693f851de4dca5f5ce7ffaa8ef2e Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Wed, 28 Oct 2026 21:13:53 +0300 Subject: [PATCH 2/6] IGNITE-29173 Add kill lock command and locks control after jobs and services --- .../processors/cache/CacheLockImpl.java | 23 +------------------ .../processors/cache/GridCacheProxyImpl.java | 4 ++-- .../processors/cache/IgniteInternalCache.java | 4 +++- .../distributed/dht/GridDhtCacheAdapter.java | 4 +++- .../dht/colocated/GridDhtColocatedCache.java | 12 +++++++--- .../distributed/near/GridNearAtomicCache.java | 4 ++-- .../near/GridNearTransactionalCache.java | 19 +++++++++++---- 7 files changed, 34 insertions(+), 36 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java index 6bcb094ea9e88..4478bf98b374b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheLockImpl.java @@ -199,8 +199,6 @@ private void incrementLockCounter() { assert cntr > 0; - boolean held = lockedByCurrentThread(); - cntr--; if (cntr == 0) { @@ -221,10 +219,8 @@ private void incrementLockCounter() { cctx.shared().txMetrics().onExplicitLockRelease(U.nanosToMillis(holdTimeNanos)); } - delegate.unlockAll(keys); - // The lock state of the instance is reset above, so it stays usable after the exception. - if (!held) { + if (!delegate.unlockAll(keys)) { throw new IllegalStateException("Failed to unlock keys, the lock has been released externally " + "(was the lock killed?)."); } @@ -237,23 +233,6 @@ private void incrementLockCounter() { } } - /** - * @return {@code True} if all the keys are still locked by the current thread, {@code false} if the lock - * has been released externally. - */ - private boolean lockedByCurrentThread() { - GridCacheContext cctx = delegate.context(); - - long threadId = Thread.currentThread().getId(); - - for (K key : keys) { - if (cctx.mvcc().explicitLock(threadId, cctx.txKey(cctx.toCacheKeyObject(key))) == null) - return false; - } - - return true; - } - /** {@inheritDoc} */ @NotNull @Override public Condition newCondition() { throw new UnsupportedOperationException(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java index 21c1fa361fe4a..fa2958188fcec 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java @@ -1288,11 +1288,11 @@ public IgniteInternalCache delegate() { } /** {@inheritDoc} */ - @Override public void unlockAll(@Nullable Collection keys) throws IgniteCheckedException { + @Override public boolean unlockAll(@Nullable Collection keys) throws IgniteCheckedException { CacheOperationContext prev = gate.enter(opCtx); try { - delegate.unlockAll(keys); + return delegate.unlockAll(keys); } finally { gate.leave(prev); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java index 925ac9c3bf511..9e975387e9544 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java @@ -1330,9 +1330,11 @@ public boolean lock(K key, long timeout) * which will acquire explicit locks for relevant cache operations. * * @param keys Keys to unlock. + * @return {@code True} if all the keys were locked by the current thread, {@code false} if the lock on any of + * the keys isn't held by the current thread (for example, it has been released externally). * @throws IgniteCheckedException If unlock execution resulted in error. */ - public void unlockAll(@Nullable Collection keys) throws IgniteCheckedException; + public boolean unlockAll(@Nullable Collection keys) throws IgniteCheckedException; /** * Checks if any node owns a lock for this key. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java index ec2e09483c0d1..b95778107c99b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java @@ -1536,8 +1536,10 @@ private void updateTtl( } /** {@inheritDoc} */ - @Override public void unlockAll(Collection keys) { + @Override public boolean unlockAll(Collection keys) { assert false; + + return false; } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java index 6c4217a21197f..5f6d6c13e8302 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java @@ -683,9 +683,11 @@ else if (!skipVals && ctx.statisticsEnabled()) } /** {@inheritDoc} */ - @Override public void unlockAll(Collection keys) { + @Override public boolean unlockAll(Collection keys) { if (keys.isEmpty()) - return; + return true; + + boolean locked = true; try { GridCacheVersion ver = null; @@ -759,10 +761,12 @@ else if (log.isDebugEnabled()) log.debug("Current thread still owns lock (or there are no other nodes)" + " [lock=" + lock + ", curThreadId=" + Thread.currentThread().getId() + ']'); } + else + locked = false; } if (ver == null) - return; + return locked; if (!locKeys.isEmpty()) removeLocks(ctx.localNodeId(), ver, locKeys, true); @@ -793,6 +797,8 @@ else if (log.isDebugEnabled()) catch (IgniteCheckedException ex) { U.error(log, "Failed to unlock the lock for keys: " + keys, ex); } + + return locked; } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java index ee9509af199ce..96a5d71e67bac 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java @@ -602,8 +602,8 @@ private void processNearAtomicUpdateResponse( } /** {@inheritDoc} */ - @Override public void unlockAll(@Nullable Collection keys) throws IgniteCheckedException { - dht.unlockAll(keys); + @Override public boolean unlockAll(@Nullable Collection keys) throws IgniteCheckedException { + return dht.unlockAll(keys); } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java index 8c06cd6fc0f71..27dad4cac6a25 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java @@ -351,9 +351,11 @@ protected boolean evictNearEntry(GridCacheEntryEx e, GridCacheVersion obsoleteVe } /** {@inheritDoc} */ - @Override public void unlockAll(Collection keys) { + @Override public boolean unlockAll(Collection keys) { if (keys.isEmpty()) - return; + return true; + + boolean locked = true; try { GridCacheVersion ver = null; @@ -370,8 +372,11 @@ protected boolean evictNearEntry(GridCacheEntryEx e, GridCacheVersion obsoleteVe GridDistributedCacheEntry entry = peekExx(cacheKey); - if (entry == null) + if (entry == null) { + locked = false; + break; // While. + } try { GridCacheMvccCandidate cand = entry.candidate(ctx.nodeId(), Thread.currentThread().getId()); @@ -387,7 +392,7 @@ protected boolean evictNearEntry(GridCacheEntryEx e, GridCacheVersion obsoleteVe Collection affNodes = CU.affinityNodes(ctx, cand.topologyVersion()); if (F.isEmpty(affNodes)) - return; + return locked; keyCnt = (int)Math.ceil((double)keys.size() / affNodes.size()); @@ -439,6 +444,8 @@ else if (log.isDebugEnabled()) " [lock=" + rmv + ", curThreadId=" + Thread.currentThread().getId() + ']'); } } + else + locked = false; assert !topVer.equals(AffinityTopologyVersion.NONE) || cand == null; @@ -454,7 +461,7 @@ else if (log.isDebugEnabled()) } if (ver == null) - return; + return locked; for (Map.Entry mapping : map.entrySet()) { ClusterNode n = mapping.getKey(); @@ -471,6 +478,8 @@ else if (!F.isEmpty(req.keys())) catch (IgniteCheckedException ex) { U.error(log, "Failed to unlock the lock for keys: " + keys, ex); } + + return locked; } /** From 8a827c9013374c43c11c796341d4af32c0456be6 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Thu, 29 Oct 2026 20:16:43 +0300 Subject: [PATCH 3/6] IGNITE-29173 Add kill lock command and locks control after jobs and services --- .../sql-reference/operational-commands.adoc | 4 +- .../service/IgniteServiceProcessor.java | 14 ++++-- .../distributed/ExplicitLockKillTest.java | 47 +++++++++++++++---- 3 files changed, 51 insertions(+), 14 deletions(-) diff --git a/docs/_docs/sql-reference/operational-commands.adoc b/docs/_docs/sql-reference/operational-commands.adoc index 6a3c8abbee293..d8dd459cbc8bf 100644 --- a/docs/_docs/sql-reference/operational-commands.adoc +++ b/docs/_docs/sql-reference/operational-commands.adoc @@ -441,13 +441,13 @@ The subsequent `Lock.unlock()` call of the thread that has acquired the killed l tab:Unix[] [source,bash] ---- -./control.sh --kill LOCK xid [--node-id node_id] +./control.sh --kill LOCK xid [--node-id node_id] [--yes] ---- tab:Windows[] [source,bash] ---- -control.bat --kill LOCK xid [--node-id node_id] +control.bat --kill LOCK xid [--node-id node_id] [--yes] ---- -- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java index 4f6b89032a31b..c9994bdbac630 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java @@ -1337,6 +1337,9 @@ else if (ctxs.size() < assignCnt) { throw new IgniteCheckedException("Error occured during service initialization: " + "[locId=" + ctx.localNodeId() + ", name=" + name + ']', e); } + finally { + releaseLeftExplicitLocks(srvcCtx, "init"); + } if (log.isInfoEnabled()) log.info("Starting service instance [name=" + srvcCtx.name() + ", execId=" + @@ -1382,7 +1385,7 @@ else if (ctxs.size() < assignCnt) { throw (Error)e; } finally { - releaseLeftExplicitLocks(srvcCtx); + releaseLeftExplicitLocks(srvcCtx, "execute"); // Suicide. exe.shutdownNow(); @@ -1397,15 +1400,16 @@ else if (ctxs.size() < assignCnt) { * so locks left unreleased by the terminating service thread would never be released. * * @param srvcCtx Service context. + * @param method Method name. */ - private void releaseLeftExplicitLocks(ServiceContextImpl srvcCtx) { + private void releaseLeftExplicitLocks(ServiceContextImpl srvcCtx, String method) { try { Collection released = ctx.cache().context().mvcc() .releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); if (!released.isEmpty()) { - U.warn(log, "Service execution finished without releasing explicit locks, the locks have been " + - "released forcibly [name=" + srvcCtx.name() + ", execId=" + srvcCtx.executionId() + + U.warn(log, "Service execution (" + method + " method) finished without releasing explicit locks, " + + "the locks have been released forcibly [name=" + srvcCtx.name() + ", execId=" + srvcCtx.executionId() + ", lockVers=" + released + ']'); } } @@ -1535,6 +1539,8 @@ private void cancel(ServiceContextImpl ctx) { throw e; } finally { + releaseLeftExplicitLocks(ctx, "cancel"); + try { this.ctx.resource().cleanup(srvc); } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java index 88b447baafb10..c0bc8c36a2ac3 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/ExplicitLockKillTest.java @@ -31,6 +31,7 @@ import org.apache.ignite.IgniteCache; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteServices; import org.apache.ignite.Ignition; import org.apache.ignite.client.IgniteClient; import org.apache.ignite.cluster.ClusterNode; @@ -418,15 +419,25 @@ public void testInterruptedJobLeftLock() throws Exception { public void testServiceLeftLock() throws Exception { int key = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); - LogListener lsnr = LogListener.matches("Service execution finished without releasing explicit locks").build(); + for (String method : List.of("init", "execute", "cancel")) { + LogListener lsnr = LogListener + .matches("Service execution (" + method + " method) finished without releasing explicit locks") + .build(); - testLog.registerListener(lsnr); + testLog.registerListener(lsnr); - grid(0).services(grid(0).cluster().forLocal()).deployNodeSingleton("lock-service", new LockService(key)); + IgniteServices svcs = grid(0).services(grid(0).cluster().forLocal()); - assertTrue(lsnr.check(TIMEOUT)); + String svcName = "lock-" + method + "-service"; - assertLockFree(key); + svcs.deployNodeSingleton(svcName, new LockService(key, method)); + + svcs.cancel(svcName); + + assertTrue(lsnr.check(TIMEOUT)); + + assertLockFree(key); + } } /** */ @@ -733,14 +744,34 @@ private static class LockService implements Service { /** */ private final int key; - /** @param key Key. */ - LockService(int key) { + /** */ + private final String method; + + /** + * @param key Key. + * @param method Method which locks the key. + */ + LockService(int key, String method) { this.key = key; + this.method = method; + } + + /** {@inheritDoc} */ + @Override public void init() { + if ("init".equals(method)) + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); } /** {@inheritDoc} */ @Override public void execute() { - ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); + if ("execute".equals(method)) + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); + } + + /** {@inheritDoc} */ + @Override public void cancel() { + if ("cancel".equals(method)) + ignite.cache(DEFAULT_CACHE_NAME).lock(key).lock(); } } } From 95dd3d344ebbc5fc8cb40a2f35e72e45af930d17 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Thu, 29 Oct 2026 20:46:51 +0300 Subject: [PATCH 4/6] IGNITE-29173 Add kill lock command and locks control after jobs and services --- .../internal/processors/job/GridJobWorker.java | 13 +++++++++---- .../client/service/ClientServiceInvokeRequest.java | 7 +++++-- .../processors/service/IgniteServiceProcessor.java | 7 +++++-- 3 files changed, 19 insertions(+), 8 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java index 306e4dd2ca2dc..7f5df12e0a475 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/job/GridJobWorker.java @@ -19,6 +19,7 @@ import java.util.ArrayList; import java.util.Collection; +import java.util.Collections; import java.util.Map; import java.util.Set; import java.util.UUID; @@ -588,8 +589,10 @@ private void execute0(boolean skipNtf) { res = U.wrapThreadLoader(dep.classLoader(), new Callable() { @Nullable @Override public Object call() { // The job can be executed by a thread that already holds explicit locks. - Set locksBefore = locksManager().explicitLockVersions( - Thread.currentThread().getId()); + GridCacheMvccManager locksMgr = locksManager(); + + Set locksBefore = locksMgr == null ? Collections.emptySet() : + locksMgr.explicitLockVersions(Thread.currentThread().getId()); try { if (internal && ctx.config().isPeerClassLoadingEnabled()) @@ -697,8 +700,10 @@ private GridCacheMvccManager locksManager() { */ private void releaseLeftExplicitLocks(Set locksBefore) { try { - Collection released = locksManager().releaseExplicitLocks( - Thread.currentThread().getId(), locksBefore); + GridCacheMvccManager locksMgr = locksManager(); + + Collection released = locksMgr == null ? Collections.emptySet() : + locksMgr.releaseExplicitLocks(Thread.currentThread().getId(), locksBefore); if (!released.isEmpty()) { U.warn(log, "Job finished without releasing explicit locks, the locks have been released forcibly " + diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java index 9b62f13ed07ee..ea41283ff262a 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceInvokeRequest.java @@ -32,6 +32,7 @@ import org.apache.ignite.internal.binary.BinaryReaderEx; import org.apache.ignite.internal.binary.BinaryUtils; import org.apache.ignite.internal.cluster.ClusterGroupAdapter; +import org.apache.ignite.internal.processors.cache.GridCacheMvccManager; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.platform.PlatformNativeException; import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext; @@ -217,8 +218,10 @@ public ClientServiceInvokeRequest(BinaryReaderEx reader, ClientProtocolContext p */ private void releaseLeftExplicitLocks(ClientConnectionContext ctx) { try { - Collection released = ctx.kernalContext().cache().context().mvcc() - .releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); + GridCacheMvccManager locksMgr = ctx.kernalContext().cache().context().mvcc(); + + Collection released = locksMgr == null ? Collections.emptySet() : + locksMgr.releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); if (!released.isEmpty()) { U.warn(ctx.kernalContext().log(getClass()), "Service method invocation finished without releasing " + diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java index c9994bdbac630..6fc28cc66dd29 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java @@ -61,6 +61,7 @@ import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch; import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest; +import org.apache.ignite.internal.processors.cache.GridCacheMvccManager; import org.apache.ignite.internal.processors.cache.ValidationOnNodeJoinUtils; import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage; @@ -1404,8 +1405,10 @@ else if (ctxs.size() < assignCnt) { */ private void releaseLeftExplicitLocks(ServiceContextImpl srvcCtx, String method) { try { - Collection released = ctx.cache().context().mvcc() - .releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); + GridCacheMvccManager locksMgr = ctx.cache().context().mvcc(); + + Collection released = locksMgr == null ? Collections.emptySet() : + locksMgr.releaseExplicitLocks(Thread.currentThread().getId(), Collections.emptySet()); if (!released.isEmpty()) { U.warn(log, "Service execution (" + method + " method) finished without releasing explicit locks, " + From 3630b98162eb92e873d1d939d41f9289e7897583 Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Thu, 29 Oct 2026 22:03:51 +0300 Subject: [PATCH 5/6] IGNITE-29173 Add kill lock command and locks control after jobs and services --- .../internal/processors/cache/GridCacheMvccManager.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java index fda2e031abc45..0d31a9ca1f6dc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java @@ -1229,8 +1229,11 @@ public boolean releaseExplicitLock(long threadId, GridCacheVersion ver) { Collection keys = Collections.singleton(cand.key().key()); - if (cacheCtx.isNear()) + if (cacheCtx.isNear()) { cacheCtx.nearTx().removeLocks(ver, keys); + + removeExplicitLock(cand); + } else if (cacheCtx.isColocated()) cacheCtx.colocated().removeLocks(threadId, ver, keys); } From 84f4eca32d0e9e707a1971b2eebfcfd4f0cbee6e Mon Sep 17 00:00:00 2001 From: Aleksey Plekhanov Date: Fri, 30 Oct 2026 09:42:49 +0300 Subject: [PATCH 6/6] IGNITE-29173 Add kill lock command and locks control after jobs and services --- ...ignite.internal.management.kill.KillLockCommandArg.json | 7 +++++++ 1 file changed, 7 insertions(+) create mode 100644 modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.kill.KillLockCommandArg.json diff --git a/modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.kill.KillLockCommandArg.json b/modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.kill.KillLockCommandArg.json new file mode 100644 index 0000000000000..5f2de58296e46 --- /dev/null +++ b/modules/wire-compatibility-check/wire/idto/org.apache.ignite.internal.management.kill.KillLockCommandArg.json @@ -0,0 +1,7 @@ +{ + "class": "org.apache.ignite.internal.management.kill.KillLockCommandArg", + "fields": [ + {"order": 0, "type": "org.apache.ignite.lang.IgniteUuid", "name": "xid"}, + {"order": 1, "type": "java.util.UUID", "name": "nodeId"} + ] +}