You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sk...@apache.org on 2025/05/16 08:01:02 UTC
(ignite-3) branch main updated: IGNITE-25356 Switch to NaiveAsyncReadWriteLock (#5798)
This is an automated email from the ASF dual-hosted git repository.
sk0x50 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 122116ee204 IGNITE-25356 Switch to NaiveAsyncReadWriteLock (#5798)
122116ee204 is described below
commit 122116ee2048d9509dd954f749200654cc6ea3d6
Author: Slava Koptilin <sl...@gmail.com>
AuthorDate: Fri May 16 11:00:56 2025 +0300
IGNITE-25356 Switch to NaiveAsyncReadWriteLock (#5798)
---
.../internal/table/distributed/TableManager.java | 249 ++++++++++++---------
.../distributed/storage/NullMvTableStorage.java | 6 +-
.../distributed/storage/NullStorageEngine.java | 2 +-
3 files changed, 147 insertions(+), 110 deletions(-)
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 9e967f0a055..787e4f66792 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -87,7 +87,6 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.locks.StampedLock;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -145,6 +144,7 @@ import org.apache.ignite.internal.network.TopologyService;
import org.apache.ignite.internal.network.serialization.MessageSerializationRegistry;
import org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEvent;
import org.apache.ignite.internal.partition.replicator.LocalPartitionReplicaEventParameters;
+import org.apache.ignite.internal.partition.replicator.NaiveAsyncReadWriteLock;
import org.apache.ignite.internal.partition.replicator.PartitionReplicaLifecycleManager;
import org.apache.ignite.internal.partition.replicator.ReliableCatalogVersions;
import org.apache.ignite.internal.partition.replicator.ReplicaTableProcessor;
@@ -435,7 +435,7 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
/** Mapping zone identifier to a collection of tables related to the zone. */
private final Map<Integer, Set<TableImpl>> tablesPerZone = new HashMap<>();
/** Locks to synchronize an access to the {@link #tablesPerZone}. */
- private final Map<Integer, StampedLock> tablesPerZoneLocks = new ConcurrentHashMap<>();
+ private final Map<Integer, NaiveAsyncReadWriteLock> tablesPerZoneLocks = new ConcurrentHashMap<>();
/** Configuration of rebalance retries delay. */
private final SystemDistributedConfigurationPropertyHolder<Integer> rebalanceRetryDelayConfiguration;
@@ -702,46 +702,50 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
return inBusyLockAsync(busyLock, () -> {
ZonePartitionId zonePartitionId = parameters.zonePartitionId();
- StampedLock zoneLock = tablesPerZoneLocks.computeIfAbsent(zonePartitionId.zoneId(), id -> new StampedLock());
+ NaiveAsyncReadWriteLock zoneLock = tablesPerZoneLocks.computeIfAbsent(
+ zonePartitionId.zoneId(),
+ id -> new NaiveAsyncReadWriteLock());
- long stamp = zoneLock.readLock();
+ CompletableFuture<Long> readLockAcquisitionFuture = zoneLock.readLock();
try {
- Set<TableImpl> zoneTables = zoneTablesRawSet(zonePartitionId.zoneId());
-
- int partitionIndex = zonePartitionId.partitionId();
-
- PartitionSet singlePartitionIdSet = PartitionSet.of(partitionIndex);
-
- CompletableFuture<?>[] futures = zoneTables.stream()
- .map(tbl -> inBusyLockAsync(busyLock, () -> {
- return getOrCreatePartitionStorages(tbl, singlePartitionIdSet)
- .thenRunAsync(() -> inBusyLock(busyLock, () -> {
- localPartsByTableId.compute(
- tbl.tableId(),
- (tableId, oldPartitionSet) -> extendPartitionSet(oldPartitionSet, partitionIndex)
- );
-
- lowWatermark.getLowWatermarkSafe(lwm ->
- registerIndexesToTable(
- tbl,
- catalogService,
- singlePartitionIdSet,
- tbl.schemaView(),
- lwm
- )
- );
-
- preparePartitionResourcesAndLoadToZoneReplica(tbl, zonePartitionId, parameters.onRecovery());
- }), ioExecutor)
- // If the table is already closed, it's not a problem (probably the node is stopping).
- .exceptionally(ignoreTableClosedException());
- }))
- .toArray(CompletableFuture[]::new);
+ return readLockAcquisitionFuture.thenCompose(stamp -> {
+ Set<TableImpl> zoneTables = zoneTablesRawSet(zonePartitionId.zoneId());
+
+ int partitionIndex = zonePartitionId.partitionId();
+
+ PartitionSet singlePartitionIdSet = PartitionSet.of(partitionIndex);
+
+ CompletableFuture<?>[] futures = zoneTables.stream()
+ .map(tbl -> inBusyLockAsync(busyLock, () -> {
+ return getOrCreatePartitionStorages(tbl, singlePartitionIdSet)
+ .thenRunAsync(() -> inBusyLock(busyLock, () -> {
+ localPartsByTableId.compute(
+ tbl.tableId(),
+ (tableId, oldPartitionSet) -> extendPartitionSet(oldPartitionSet, partitionIndex)
+ );
+
+ lowWatermark.getLowWatermarkSafe(lwm ->
+ registerIndexesToTable(
+ tbl,
+ catalogService,
+ singlePartitionIdSet,
+ tbl.schemaView(),
+ lwm
+ )
+ );
+
+ preparePartitionResourcesAndLoadToZoneReplica(tbl, zonePartitionId, parameters.onRecovery());
+ }), ioExecutor)
+ // If the table is already closed, it's not a problem (probably the node is stopping).
+ .exceptionally(ignoreTableClosedException());
+ }))
+ .toArray(CompletableFuture[]::new);
- return allOf(futures).thenAccept(unused -> zoneLock.unlockRead(stamp));
+ return allOf(futures).whenComplete((unused, t) -> zoneLock.unlockRead(stamp));
+ });
} catch (Throwable t) {
- zoneLock.unlockRead(stamp);
+ readLockAcquisitionFuture.whenComplete((stamp, ex) -> zoneLock.unlockRead(stamp));
return failedFuture(t);
}
@@ -764,22 +768,22 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
ZonePartitionId zonePartitionId = parameters.zonePartitionId();
- StampedLock zoneLock = tablesPerZoneLocks.computeIfAbsent(zonePartitionId.zoneId(), id -> new StampedLock());
+ NaiveAsyncReadWriteLock zoneLock = tablesPerZoneLocks.computeIfAbsent(
+ zonePartitionId.zoneId(),
+ id -> new NaiveAsyncReadWriteLock());
- long stamp = zoneLock.readLock();
+ CompletableFuture<Long> readLockAcquisitionFuture = zoneLock.readLock();
try {
- CompletableFuture<?>[] futures = zoneTablesRawSet(zonePartitionId.zoneId()).stream()
- .map(this::stopTablePartitions)
- .toArray(CompletableFuture[]::new);
-
- return allOf(futures).thenApply(unused -> {
- zoneLock.unlockRead(stamp);
+ return readLockAcquisitionFuture.thenCompose(stamp -> {
+ CompletableFuture<?>[] futures = zoneTablesRawSet(zonePartitionId.zoneId()).stream()
+ .map(this::stopTablePartitions)
+ .toArray(CompletableFuture[]::new);
- return false;
+ return allOf(futures).whenComplete((v, t) -> zoneLock.unlockRead(stamp)).thenApply(v -> false);
});
} catch (Throwable t) {
- zoneLock.unlockRead(stamp);
+ readLockAcquisitionFuture.whenComplete((stamp, ex) -> zoneLock.unlockRead(stamp));
return failedFuture(t);
}
@@ -792,28 +796,32 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
ZonePartitionId zonePartitionId = parameters.zonePartitionId();
- StampedLock zoneLock = tablesPerZoneLocks.computeIfAbsent(zonePartitionId.zoneId(), id -> new StampedLock());
+ NaiveAsyncReadWriteLock zoneLock = tablesPerZoneLocks.computeIfAbsent(
+ zonePartitionId.zoneId(),
+ id -> new NaiveAsyncReadWriteLock());
- long stamp = zoneLock.readLock();
+ CompletableFuture<Long> readLockAcquisitionFuture = zoneLock.readLock();
try {
- return inBusyLockAsync(busyLock, () -> {
- CompletableFuture<?>[] futures = zoneTablesRawSet(zonePartitionId.zoneId()).stream()
- .map(table -> supplyAsync(
- () -> inBusyLockAsync(
- busyLock,
- () -> stopAndDestroyTablePartition(
- new TablePartitionId(table.tableId(), zonePartitionId.partitionId()),
- parameters.causalityToken()
- )
- ),
- ioExecutor))
- .toArray(CompletableFuture[]::new);
+ return readLockAcquisitionFuture.thenCompose(stamp -> {
+ return inBusyLockAsync(busyLock, () -> {
+ CompletableFuture<?>[] futures = zoneTablesRawSet(zonePartitionId.zoneId()).stream()
+ .map(table -> supplyAsync(
+ () -> inBusyLockAsync(
+ busyLock,
+ () -> stopAndDestroyTablePartition(
+ new TablePartitionId(table.tableId(), zonePartitionId.partitionId()),
+ parameters.causalityToken()
+ )
+ ),
+ ioExecutor))
+ .toArray(CompletableFuture[]::new);
- return allOf(futures);
- }).thenApply((unused) -> false);
+ return allOf(futures).whenComplete((v, t) -> zoneLock.unlockRead(stamp));
+ }).thenApply((unused) -> false);
+ });
} catch (Throwable t) {
- zoneLock.unlockRead(stamp);
+ readLockAcquisitionFuture.whenComplete((stamp, ex) -> zoneLock.unlockRead(stamp));
return failedFuture(t);
}
@@ -2854,40 +2862,44 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
int partitions = internalTable.partitions();
- StampedLock zoneLock = tablesPerZoneLocks.computeIfAbsent(internalTable.zoneId(), id -> new StampedLock());
+ NaiveAsyncReadWriteLock zoneLock = tablesPerZoneLocks.computeIfAbsent(
+ internalTable.zoneId(),
+ id -> new NaiveAsyncReadWriteLock());
- long stamp = zoneLock.writeLock();
+ CompletableFuture<Long> writeLockAcquisitionFuture = zoneLock.writeLock();
try {
- CompletableFuture<?>[] stopReplicaAndDestroyFutures = new CompletableFuture<?>[partitions];
+ return writeLockAcquisitionFuture.thenCompose(stamp -> {
+ CompletableFuture<?>[] stopReplicaAndDestroyFutures = new CompletableFuture<?>[partitions];
+
+ // TODO https://issues.apache.org/jira/browse/IGNITE-19170 Partitions should be stopped on the assignments change
+ // event triggered by zone drop or alter. Stop replica asynchronously, out of metastorage event pipeline.
+ for (int partitionId = 0; partitionId < partitions; partitionId++) {
+ CompletableFuture<Void> resourcesUnloadFuture;
+
+ if (enabledColocation) {
+ resourcesUnloadFuture = partitionReplicaLifecycleManager.unloadTableResourcesFromZoneReplica(
+ new ZonePartitionId(internalTable.zoneId(), partitionId),
+ internalTable.tableId()
+ );
+ } else {
+ resourcesUnloadFuture = nullCompletedFuture();
+ }
- // TODO https://issues.apache.org/jira/browse/IGNITE-19170 Partitions should be stopped on the assignments change
- // event triggered by zone drop or alter. Stop replica asynchronously, out of metastorage event pipeline.
- for (int partitionId = 0; partitionId < partitions; partitionId++) {
- CompletableFuture<Void> resourcesUnloadFuture;
+ var tablePartitionId = new TablePartitionId(internalTable.tableId(), partitionId);
- if (enabledColocation) {
- resourcesUnloadFuture = partitionReplicaLifecycleManager.unloadTableResourcesFromZoneReplica(
- new ZonePartitionId(internalTable.zoneId(), partitionId),
- internalTable.tableId()
- );
- } else {
- resourcesUnloadFuture = nullCompletedFuture();
+ stopReplicaAndDestroyFutures[partitionId] = resourcesUnloadFuture
+ .thenCompose(v -> stopAndDestroyTablePartition(tablePartitionId, table));
}
- var tablePartitionId = new TablePartitionId(internalTable.tableId(), partitionId);
-
- stopReplicaAndDestroyFutures[partitionId] = resourcesUnloadFuture
- .thenCompose(v -> stopAndDestroyTablePartition(tablePartitionId, table));
- }
-
- return allOf(stopReplicaAndDestroyFutures).whenComplete((res, th) -> {
- tablesPerZone.getOrDefault(internalTable.zoneId(), emptySet()).remove(table);
+ return allOf(stopReplicaAndDestroyFutures).whenComplete((res, th) -> {
+ tablesPerZone.getOrDefault(internalTable.zoneId(), emptySet()).remove(table);
- zoneLock.unlockWrite(stamp);
+ zoneLock.unlockWrite(stamp);
+ });
});
} catch (Throwable t) {
- zoneLock.unlockWrite(stamp);
+ writeLockAcquisitionFuture.whenComplete((stamp, ex) -> zoneLock.unlockWrite(stamp));
throw t;
}
@@ -3270,16 +3282,29 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*
* @param zoneId Zone identifier.
* @return Set of tables.
+ * @throws IgniteInternalException If failed to acquire a read lock for the zone or current thread was interrupted while waiting.
*/
- public Set<TableImpl> zoneTables(int zoneId) {
- StampedLock zoneLock = tablesPerZoneLocks.computeIfAbsent(zoneId, id -> new StampedLock());
+ public Set<TableImpl> zoneTables(int zoneId) throws IgniteInternalException {
+ NaiveAsyncReadWriteLock zoneLock = tablesPerZoneLocks.computeIfAbsent(zoneId, id -> new NaiveAsyncReadWriteLock());
- long stamp = zoneLock.readLock();
+ CompletableFuture<Long> readLockAcquisitionFuture = zoneLock.readLock();
try {
- return Set.copyOf(zoneTablesRawSet(zoneId));
- } finally {
- zoneLock.unlockRead(stamp);
+ return readLockAcquisitionFuture.thenApply(stamp -> {
+ Set<TableImpl> res = Set.copyOf(zoneTablesRawSet(zoneId));
+
+ zoneLock.unlockRead(stamp);
+
+ return res;
+ }).get();
+ } catch (Throwable t) {
+ readLockAcquisitionFuture.whenComplete((stamp, ex) -> zoneLock.unlockRead(stamp));
+
+ if (t instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+
+ throw new IgniteInternalException(INTERNAL_ERR, "Failed to acquire a read lock for zone [zoneId=" + zoneId + ']', t);
}
}
@@ -3299,23 +3324,35 @@ public class TableManager implements IgniteTablesInternal, IgniteComponent {
*
* @param zoneId Zone identifier.
* @param table Table to add.
+ * @throws IgniteInternalException If failed to acquire a write lock for the zone or current thread was interrupted while waiting.
*/
- private void addTableToZone(int zoneId, TableImpl table) {
- StampedLock rwLock = tablesPerZoneLocks.computeIfAbsent(zoneId, id -> new StampedLock());
+ private void addTableToZone(int zoneId, TableImpl table) throws IgniteInternalException {
+ NaiveAsyncReadWriteLock zoneLock = tablesPerZoneLocks.computeIfAbsent(zoneId, id -> new NaiveAsyncReadWriteLock());
+
+ CompletableFuture<Long> writeLockAcquisitionFuture = zoneLock.writeLock();
- long stamp = rwLock.writeLock();
try {
- tablesPerZone.compute(zoneId, (id, tables) -> {
- if (tables == null) {
- tables = new HashSet<>();
- }
+ writeLockAcquisitionFuture.thenAccept(stamp -> {
+ tablesPerZone.compute(zoneId, (id, tables) -> {
+ if (tables == null) {
+ tables = new HashSet<>();
+ }
- tables.add(table);
+ tables.add(table);
- return tables;
- });
- } finally {
- rwLock.unlockWrite(stamp);
+ return tables;
+ });
+
+ zoneLock.unlockWrite(stamp);
+ }).get();
+ } catch (Throwable t) {
+ writeLockAcquisitionFuture.whenComplete((stamp, ex) -> zoneLock.unlockWrite(stamp));
+
+ if (t instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+
+ throw new IgniteInternalException(INTERNAL_ERR, "Failed to acquire a write lock for zone [zoneId=" + zoneId + ']', t);
}
}
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullMvTableStorage.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullMvTableStorage.java
index 9a7f6e42271..085ff61799e 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullMvTableStorage.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullMvTableStorage.java
@@ -72,7 +72,7 @@ public class NullMvTableStorage implements MvTableStorage {
@Override
public CompletableFuture<Void> destroyIndex(int indexId) {
return throwNoStorageEngineException("Table uses an unknown storage profile or engine, so current node either should not receive "
- + "any index destruction events, or storage profile addition is not handled property");
+ + "any index destruction events, or storage profile addition is not handled properly");
}
@Override
@@ -83,7 +83,7 @@ public class NullMvTableStorage implements MvTableStorage {
@Override
public CompletableFuture<Void> destroy() {
return throwNoStorageEngineException("Table uses an unknown storage profile or engine, so current node either should not receive "
- + "any table destruction events, or storage profile addition is not handled property");
+ + "any table destruction events, or storage profile addition is not handled properly");
}
@Override
@@ -123,7 +123,7 @@ public class NullMvTableStorage implements MvTableStorage {
private <T> T throwNoPartitionsException() {
return throwNoStorageEngineException("Table uses an unknown storage profile or engine, so current node either should not receive "
- + "any assignments, or storage profile addition is not handled property");
+ + "any assignments, or storage profile addition is not handled properly");
}
private <T> T throwNoStorageEngineException(String errorMessage) {
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullStorageEngine.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullStorageEngine.java
index 1ab3e1de83d..e1cb5a98c19 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullStorageEngine.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/NullStorageEngine.java
@@ -65,7 +65,7 @@ public class NullStorageEngine implements StorageEngine {
private static <T> T throwNoEngineException(@Nullable Integer tableId) {
throw new StorageException("Table uses an unknown storage profile or engine, so current node either should not receive "
- + "any assignments, or storage profile addition is not handled property"
+ + "any assignments, or storage profile addition is not handled properly"
+ (tableId == null ? "" : " [tableId=" + tableId + "]"));
}
}