You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2024/09/09 08:34:34 UTC
(ignite-3) branch main updated: IGNITE-22609 Add busy lock to raft group listeners (#4350)
This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 14e557713e IGNITE-22609 Add busy lock to raft group listeners (#4350)
14e557713e is described below
commit 14e557713e1e52650ab2e0a6cdcd16b7f213000d
Author: Ivan Bessonov <be...@gmail.com>
AuthorDate: Mon Sep 9 11:34:29 2024 +0300
IGNITE-22609 Add busy lock to raft group listeners (#4350)
---
.../management/raft/CmgRaftGroupListener.java | 29 ++++++++++++++++++
.../server/raft/MetaStorageListener.java | 26 ++++++++++++++++-
.../replicator/ZonePartitionRaftListener.java | 17 ++++++++++-
.../internal/raft/service/RaftGroupListener.java | 8 +++++
.../jraft/rpc/impl/ActionRequestProcessor.java | 34 +++++++++-------------
.../table/distributed/raft/PartitionListener.java | 17 +++++++++++
6 files changed, 109 insertions(+), 22 deletions(-)
diff --git a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListener.java b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListener.java
index 61a51aec6d..acd87cb1b4 100644
--- a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListener.java
+++ b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/raft/CmgRaftGroupListener.java
@@ -54,6 +54,7 @@ import org.apache.ignite.internal.raft.ReadCommand;
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.network.ClusterNode;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -64,6 +65,8 @@ import org.jetbrains.annotations.TestOnly;
public class CmgRaftGroupListener implements RaftGroupListener {
private static final IgniteLogger LOG = Loggers.forClass(CmgRaftGroupListener.class);
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
private final ClusterStateStorageManager storageManager;
private final LogicalTopology logicalTopology;
@@ -96,6 +99,18 @@ public class CmgRaftGroupListener implements RaftGroupListener {
@Override
public void onRead(Iterator<CommandClosure<ReadCommand>> iterator) {
+ if (!busyLock.enterBusy()) {
+ iterator.forEachRemaining(clo -> clo.result(new ShutdownException()));
+ }
+
+ try {
+ onReadBusy(iterator);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private void onReadBusy(Iterator<CommandClosure<ReadCommand>> iterator) {
while (iterator.hasNext()) {
CommandClosure<ReadCommand> clo = iterator.next();
@@ -125,6 +140,18 @@ public class CmgRaftGroupListener implements RaftGroupListener {
@Override
public void onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
+ if (!busyLock.enterBusy()) {
+ iterator.forEachRemaining(clo -> clo.result(new ShutdownException()));
+ }
+
+ try {
+ onWriteBusy(iterator);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private void onWriteBusy(Iterator<CommandClosure<WriteCommand>> iterator) {
while (iterator.hasNext()) {
CommandClosure<WriteCommand> clo = iterator.next();
@@ -292,6 +319,8 @@ public class CmgRaftGroupListener implements RaftGroupListener {
@Override
public void onShutdown() {
+ busyLock.block();
+
// Raft storage lifecycle is managed by outside components.
}
diff --git a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java
index 8e4780e58b..5c3cfaa852 100644
--- a/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java
+++ b/modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/server/raft/MetaStorageListener.java
@@ -45,6 +45,7 @@ import org.apache.ignite.internal.raft.service.BeforeApplyHandler;
import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.util.Cursor;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.jetbrains.annotations.Nullable;
/**
@@ -52,6 +53,8 @@ import org.jetbrains.annotations.Nullable;
* TODO: IGNITE-14693 Implement Meta storage exception handling logic.
*/
public class MetaStorageListener implements RaftGroupListener, BeforeApplyHandler {
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
private final MetaStorageWriteHandler writeHandler;
/** Storage. */
@@ -72,6 +75,18 @@ public class MetaStorageListener implements RaftGroupListener, BeforeApplyHandle
@Override
public void onRead(Iterator<CommandClosure<ReadCommand>> iter) {
+ if (!busyLock.enterBusy()) {
+ iter.forEachRemaining(clo -> clo.result(new ShutdownException()));
+ }
+
+ try {
+ onReadBusy(iter);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private void onReadBusy(Iterator<CommandClosure<ReadCommand>> iter) {
while (iter.hasNext()) {
CommandClosure<ReadCommand> clo = iter.next();
@@ -157,7 +172,15 @@ public class MetaStorageListener implements RaftGroupListener, BeforeApplyHandle
@Override
public void onWrite(Iterator<CommandClosure<WriteCommand>> iter) {
- iter.forEachRemaining(writeHandler::handleWriteCommand);
+ if (!busyLock.enterBusy()) {
+ iter.forEachRemaining(clo -> clo.result(new ShutdownException()));
+ }
+
+ try {
+ iter.forEachRemaining(writeHandler::handleWriteCommand);
+ } finally {
+ busyLock.leaveBusy();
+ }
}
@Override
@@ -180,5 +203,6 @@ public class MetaStorageListener implements RaftGroupListener, BeforeApplyHandle
@Override
public void onShutdown() {
+ busyLock.block();
}
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionRaftListener.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionRaftListener.java
index 2895a0b3c0..98994e74f7 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionRaftListener.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/ZonePartitionRaftListener.java
@@ -35,6 +35,7 @@ import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
import org.apache.ignite.internal.tx.TransactionResult;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
/**
* RAFT listener for the zone partition.
@@ -42,6 +43,8 @@ import org.apache.ignite.internal.tx.TransactionResult;
public class ZonePartitionRaftListener implements RaftGroupListener {
private static final IgniteLogger LOG = Loggers.forClass(ZonePartitionRaftListener.class);
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
@Override
public void onRead(Iterator<CommandClosure<ReadCommand>> iterator) {
iterator.forEachRemaining((CommandClosure<? extends ReadCommand> clo) -> {
@@ -53,6 +56,18 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
@Override
public void onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
+ if (!busyLock.enterBusy()) {
+ iterator.forEachRemaining(clo -> clo.result(new ShutdownException()));
+ }
+
+ try {
+ onWriteBusy(iterator);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private void onWriteBusy(Iterator<CommandClosure<WriteCommand>> iterator) {
iterator.forEachRemaining((CommandClosure<? extends WriteCommand> clo) -> {
Command command = clo.command();
@@ -96,6 +111,6 @@ public class ZonePartitionRaftListener implements RaftGroupListener {
@Override
public void onShutdown() {
- // No-op
+ busyLock.block();
}
}
diff --git a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/RaftGroupListener.java b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/RaftGroupListener.java
index 6ca04df0eb..d1696d06fe 100644
--- a/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/RaftGroupListener.java
+++ b/modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/RaftGroupListener.java
@@ -27,6 +27,14 @@ import org.apache.ignite.internal.raft.WriteCommand;
* A listener for replication group events.
*/
public interface RaftGroupListener {
+ /**
+ * Exception type that should be used by {@link #onRead(Iterator)} and {@link #onWrite(Iterator)} to complete their closures, in cases
+ * when these methods have been invoked on a listener that's already shut down.
+ */
+ class ShutdownException extends RuntimeException {
+ private static final long serialVersionUID = 5354137682604995519L;
+ }
+
/**
* The callback to apply read commands.
*
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java
index 5255c727f4..407746cca9 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java
@@ -34,6 +34,7 @@ import org.apache.ignite.internal.raft.server.impl.JraftServerImpl.DelegatingSta
import org.apache.ignite.internal.raft.service.BeforeApplyHandler;
import org.apache.ignite.internal.raft.service.CommandClosure;
import org.apache.ignite.internal.raft.service.RaftGroupListener;
+import org.apache.ignite.internal.raft.service.RaftGroupListener.ShutdownException;
import org.apache.ignite.raft.jraft.Closure;
import org.apache.ignite.raft.jraft.Node;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
@@ -183,13 +184,7 @@ public class ActionRequestProcessor implements RpcProcessor<ActionRequest> {
new CommandClosureImpl<>(command) {
@Override
public void result(Serializable res) {
- if (res instanceof Throwable) {
- sendSMError(rpcCtx, (Throwable)res, false);
-
- return;
- }
-
- rpcCtx.sendResponse(factory.actionResponse().result(res).build());
+ sendResponse(res, rpcCtx);
}
@Override
@@ -221,13 +216,7 @@ public class ActionRequestProcessor implements RpcProcessor<ActionRequest> {
}
@Override public void result(Serializable res) {
- if (res instanceof Throwable) {
- sendSMError(rpcCtx, (Throwable)res, false);
-
- return;
- }
-
- rpcCtx.sendResponse(factory.actionResponse().result(res).build());
+ sendResponse(res, rpcCtx);
}
}).iterator());
}
@@ -251,12 +240,7 @@ public class ActionRequestProcessor implements RpcProcessor<ActionRequest> {
}
@Override public void result(Serializable res) {
- if (res instanceof Throwable) {
- sendSMError(rpcCtx, (Throwable)res, false);
-
- return;
- }
- rpcCtx.sendResponse(factory.actionResponse().result(res).build());
+ sendResponse(res, rpcCtx);
}
}).iterator());
}
@@ -266,6 +250,16 @@ public class ActionRequestProcessor implements RpcProcessor<ActionRequest> {
}
}
+ private void sendResponse(Serializable res, RpcContext rpcCtx) {
+ if (res instanceof ShutdownException) {
+ rpcCtx.sendResponse(factory.errorResponse().errorCode(RaftError.ESHUTDOWN.getNumber()).build());
+ } else if (res instanceof Throwable) {
+ sendSMError(rpcCtx, (Throwable)res, false);
+ } else {
+ rpcCtx.sendResponse(factory.actionResponse().result(res).build());
+ }
+ }
+
/** {@inheritDoc} */
@Override public String interest() {
return ActionRequest.class.getName();
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
index 91fa878221..2249068f9d 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java
@@ -86,6 +86,7 @@ import org.apache.ignite.internal.tx.TxStateMeta;
import org.apache.ignite.internal.tx.UpdateCommandResult;
import org.apache.ignite.internal.tx.message.VacuumTxStatesCommand;
import org.apache.ignite.internal.tx.storage.state.TxStateStorage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.PendingComparableValuesTracker;
import org.apache.ignite.internal.util.TrackerClosedException;
import org.jetbrains.annotations.Nullable;
@@ -101,6 +102,8 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
/** Undefined value for {@link #minActiveTxBeginTime}. */
private static final long UNDEFINED_MIN_TX_TIME = 0L;
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
/** Transaction manager. */
private final TxManager txManager;
@@ -181,6 +184,18 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
@Override
public void onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
+ if (!busyLock.enterBusy()) {
+ iterator.forEachRemaining(clo -> clo.result(new ShutdownException()));
+ }
+
+ try {
+ onWriteBusy(iterator);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ private void onWriteBusy(Iterator<CommandClosure<WriteCommand>> iterator) {
iterator.forEachRemaining((CommandClosure<? extends WriteCommand> clo) -> {
Command command = clo.command();
@@ -564,6 +579,8 @@ public class PartitionListener implements RaftGroupListener, BeforeApplyHandler
@Override
public void onShutdown() {
+ busyLock.block();
+
storage.close();
}