[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072580281
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PageMemoryStorageUtils.java:
##
@@ -0,0 +1,125 @@
+/*
+ * 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.storage.pagememory;
+
+import java.util.function.Supplier;
+import org.apache.ignite.internal.pagememory.PageMemory;
+import org.apache.ignite.internal.storage.StorageClosedException;
+import org.apache.ignite.internal.storage.StorageException;
+import org.apache.ignite.internal.storage.StorageRebalanceException;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.lang.IgniteStringFormatter;
+
+/**
+ * Helper class for {@link PageMemory}-based storages.
+ */
+public class PageMemoryStorageUtils {
+/**
+ * Runs a function under a busyLock, if it was not possible to
acquire(busy) busyLock throws an exception depending on
+ * {@link StorageState}.
+ *
+ * @param Type of the returned value.
+ * @param busyLock Busy lock.
+ * @param supplier Function.
+ * @param storageInfoSupplier Storage state supplier.
+ * @param storageStateSupplier Storage information supplier, for example
in the format "table=user, partitionId=1".
+ * @return Value.
+ * @throws StorageClosedException If the storage is closed.
+ * @throws StorageRebalanceException If storage is in the process of
rebalancing.
+ * @throws StorageException For other {@link StorageState}.
+ */
+public static V inBusyLock(
+IgniteSpinBusyLock busyLock,
+Supplier supplier,
+Supplier storageStateSupplier,
Review Comment:
I tried to change a little.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072512178
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+GroupPartitionId groupPartitionId =
createGroupPartitionId(mvPartitionStorage.partitionId());
-@Override
-CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
-int partitionId = mvPartitionStorage.partitionId();
+return destroyPartitionPhysically(groupPartitionId).thenAccept(unused
-> {
+TableView tableView = tableConfig.value();
-// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
-// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+PersistentPageMemory pageMemory = dataRegion.pageMemory();
-mvPartitionStorage.close();
+int partitionId = groupPartitionId.getPartitionId();
+
+PartitionMeta meta = getOrCreatePartitionMeta(groupPartitionId,
ensurePartitionFilePageStore(tableView, groupPartitionId));
+
+inCheckpointLock(() -> {
+RowVersionFreeList rowVersionFreeList =
createRowVersionFreeList(tableView, partitionId, pageMemory, meta);
-int tableId = tableCfg.tableId().value();
+IndexColumnsFreeList indexColumnsFreeList
+= createIndexColumnsFreeList(tableView, partitionId,
rowVersionFreeList, pageMemory, meta);
-GroupPartitionId groupPartitionId = new GroupPartitionId(tableId,
partitionId);
+VersionChainTree versionChainTree =
createVersionChainTree(tableView, partitionId, rowVersionFreeList, pageMemory,
meta);
+IndexMetaTree indexMetaTree = createIndexMetaTree(tableView,
partitionId, rowVersionFreeList, pageMemory, meta);
+
+((PersistentPageMemoryMvPartitionStorage)
mvPartitionStorage).updateDataStructuresOnRebalance(
+meta,
+rowVersionFreeList,
+indexColumnsFreeList,
+versionChainTree,
+indexMetaTree
+);
+
+return null;
+});
+});
+}
+
+private CompletableFuture
destroyPartitionPhysically(GroupPartitionId groupPartitionId) {
dataRegion.filePageStoreManager().getStore(groupPartitionId).markToDestroy();
-dataRegion.pageMemory().invalidate(tableId, partitionId);
+dataRegion.pageMemory().invalidate(groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId());
Review Comment:
Created a ticket: https://issues.apache.org/jira/browse/IGNITE-18565
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072510827
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -396,4 +372,56 @@ private int compareRows(SortedIndexRowKey key1,
SortedIndexRowKey key2) {
);
}
}
+
+/**
+ * Prepares storage for rebalancing.
+ *
+ * Stops ongoing index operations.
+ *
+ * @throws StorageRebalanceException If there was an error when starting
the rebalance.
+ */
+public void startRebalance() {
+if (!state.compareAndSet(StorageState.RUNNABLE,
StorageState.REBALANCE)) {
+throwExceptionDependingOnStorageStateOnRebalance(state.get(),
createStorageInfo());
+}
+
+// Stops ongoing operations on the storage.
Review Comment:
I tried to modify it.
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java:
##
@@ -203,19 +187,68 @@ public void destroy() throws StorageException {
* Closes the hash index storage.
*/
public void close() {
-if (!CLOSED.compareAndSet(this, false, true)) {
+if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
+StorageState state = this.state.get();
+
+assert state == StorageState.CLOSED : state;
+
return;
}
-closeBusyLock.block();
+busyLock.block();
hashIndexTree.close();
}
/**
- * Throws an exception that the storage is already closed.
+ * Prepares storage for rebalancing.
+ *
+ * Stops ongoing index operations.
+ *
+ * @throws StorageRebalanceException If there was an error when starting
the rebalance.
+ */
+public void startRebalance() {
+if (!state.compareAndSet(StorageState.RUNNABLE,
StorageState.REBALANCE)) {
+throwExceptionDependingOnStorageStateOnRebalance(state.get(),
createStorageInfo());
+}
+
+// Stops ongoing operations on the storage.
Review Comment:
I tried to modify it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072480981
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java:
##
@@ -365,4 +315,70 @@ private void syncMetadataOnCheckpoint(@Nullable Executor
executor) throws Ignite
});
}
}
+
+@Override
+public void lastAppliedOnRebalance(long lastAppliedIndex, long
lastAppliedTerm) throws StorageException {
+throwExceptionIfStorageNotInProgressOfRebalance(state.get(),
this::createStorageInfo);
+
+lastApplied0(lastAppliedIndex, lastAppliedTerm);
+
+persistedIndex = lastAppliedIndex;
+}
+
+/**
+ * Updates the internal data structures of the storage and its indexes on
rebalance.
+ *
+ * @param meta Partition meta.
+ * @param rowVersionFreeList Free list for {@link RowVersion}.
+ * @param indexFreeList Free list fot {@link IndexColumns}.
+ * @param versionChainTree Table tree for {@link VersionChain}.
+ * @param indexMetaTree Tree that contains SQL indexes' metadata.
+ * @throws StorageRebalanceException If the storage is not in the process
of rebalancing.
+ */
+public void updateDataStructuresOnRebalance(
+PartitionMeta meta,
+RowVersionFreeList rowVersionFreeList,
+IndexColumnsFreeList indexFreeList,
+VersionChainTree versionChainTree,
+IndexMetaTree indexMetaTree
+) {
+throwExceptionIfStorageNotInProgressOfRebalance(state.get(),
this::createStorageInfo);
+
+this.meta = meta;
+
+this.rowVersionFreeList.close();
Review Comment:
I'll try to close them at the beginning of the rebalancing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072480105
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -396,4 +372,56 @@ private int compareRows(SortedIndexRowKey key1,
SortedIndexRowKey key2) {
);
}
}
+
+/**
+ * Prepares storage for rebalancing.
+ *
+ * Stops ongoing index operations.
+ *
+ * @throws StorageRebalanceException If there was an error when starting
the rebalance.
+ */
+public void startRebalance() {
+if (!state.compareAndSet(StorageState.RUNNABLE,
StorageState.REBALANCE)) {
+throwExceptionDependingOnStorageStateOnRebalance(state.get(),
createStorageInfo());
+}
+
+// Stops ongoing operations on the storage.
+busyLock.block();
+busyLock.unblock();
+}
+
+/**
+ * Completes the rebalancing of the storage.
+ *
+ * @throws StorageRebalanceException If there is an error while completing
the storage rebalance.
+ */
+public void completeRebalance() {
+if (!state.compareAndSet(StorageState.REBALANCE,
StorageState.RUNNABLE)) {
+throwExceptionDependingOnStorageStateOnRebalance(state.get(),
createStorageInfo());
+}
+}
+
+/**
+ * Updates the internal data structures of the storage on rebalance.
+ *
+ * @param freeList Free list to store index columns.
+ * @param sortedIndexTree Sorted index tree instance.
+ * @throws StorageRebalanceException If the storage is not in the process
of rebalancing.
+ */
+public void updateDataStructuresOnRebalance(IndexColumnsFreeList freeList,
SortedIndexTree sortedIndexTree) {
+throwExceptionIfStorageNotInProgressOfRebalance(state.get(),
this::createStorageInfo);
+
+this.freeList = freeList;
+
+this.sortedIndexTree.close();
Review Comment:
I'll try to close them at the beginning of the rebalancing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072470190
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java:
##
@@ -203,19 +187,68 @@ public void destroy() throws StorageException {
* Closes the hash index storage.
*/
public void close() {
-if (!CLOSED.compareAndSet(this, false, true)) {
+if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
+StorageState state = this.state.get();
+
+assert state == StorageState.CLOSED : state;
+
return;
}
-closeBusyLock.block();
+busyLock.block();
hashIndexTree.close();
}
/**
- * Throws an exception that the storage is already closed.
+ * Prepares storage for rebalancing.
+ *
+ * Stops ongoing index operations.
+ *
+ * @throws StorageRebalanceException If there was an error when starting
the rebalance.
+ */
+public void startRebalance() {
+if (!state.compareAndSet(StorageState.RUNNABLE,
StorageState.REBALANCE)) {
+throwExceptionDependingOnStorageStateOnRebalance(state.get(),
createStorageInfo());
+}
+
+// Stops ongoing operations on the storage.
+busyLock.block();
+busyLock.unblock();
+}
+
+/**
+ * Completes the rebalancing of the storage.
+ *
+ * @throws StorageRebalanceException If there is an error while completing
the storage rebalance.
*/
-private void throwStorageClosedException() {
-throw new StorageClosedException();
+public void completeRebalance() {
+if (!state.compareAndSet(StorageState.REBALANCE,
StorageState.RUNNABLE)) {
+throwExceptionDependingOnStorageStateOnRebalance(state.get(),
createStorageInfo());
+}
+}
+
+/**
+ * Updates the internal data structures of the storage on rebalance.
+ *
+ * @param freeList Free list to store index columns.
+ * @param hashIndexTree Hash index tree instance.
+ * @throws StorageRebalanceException If the storage is not in the process
of rebalancing.
+ */
+public void updateDataStructuresOnRebalance(IndexColumnsFreeList freeList,
HashIndexTree hashIndexTree) {
+throwExceptionIfStorageNotInProgressOfRebalance(state.get(),
this::createStorageInfo);
+
+this.freeList = freeList;
+
+this.hashIndexTree.close();
Review Comment:
I'll try to close them at the beginning of the rebalancing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072447536
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java:
##
@@ -987,24 +909,174 @@ public boolean hasNext() {
ReadResult result = findLatestRowVersion(chain);
if (result.isEmpty() && !result.isWriteIntent()) {
-continue;
+return null;
}
nextRead = result;
currentChain = chain;
return true;
-} finally {
-closeBusyLock.leaveBusy();
+});
+
+if (hasNext != null) {
+return hasNext;
}
}
}
}
+private class ScanVersionsCursor implements Cursor {
+final RowId rowId;
+
+@Nullable
+private Boolean hasNext;
+
+@Nullable
+private VersionChain versionChain;
+
+@Nullable
+private RowVersion rowVersion;
+
+private ScanVersionsCursor(RowId rowId) {
+this.rowId = rowId;
+}
+
+@Override
+public void close() {
+// No-op.
+}
+
+@Override
+public boolean hasNext() {
+return busy(() -> {
+advanceIfNeeded();
+
+return hasNext;
+});
+}
+
+@Override
+public ReadResult next() {
+return busy(() -> {
+advanceIfNeeded();
+
+if (!hasNext) {
+throw new NoSuchElementException();
+}
+
+hasNext = null;
+
+return
rowVersionToResultNotFillingLastCommittedTs(versionChain, rowVersion);
+});
+}
+
+private void advanceIfNeeded() {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
AbstractPageMemoryMvPartitionStorage.this::createStorageInfo);
+
+if (hasNext != null) {
+return;
+}
+
+if (versionChain == null) {
+try {
+versionChain = versionChainTree.findOne(new
VersionChainKey(rowId));
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException(e);
+}
+
+rowVersion = versionChain == null ? null :
readRowVersion(versionChain.headLink(), ALWAYS_LOAD_VALUE);
+} else {
+rowVersion = !rowVersion.hasNextLink() ? null :
readRowVersion(rowVersion.nextLink(), ALWAYS_LOAD_VALUE);
+}
+
+hasNext = rowVersion != null;
+}
+}
+
+@Override
+public void close() {
+if (!state.compareAndSet(StorageState.RUNNABLE, StorageState.CLOSED)) {
+StorageState state = this.state.get();
+
+assert state == StorageState.CLOSED : state;
+
+return;
+}
+
+busyLock.block();
+
+versionChainTree.close();
Review Comment:
Makes sense, I'll do it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072431782
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+GroupPartitionId groupPartitionId =
createGroupPartitionId(mvPartitionStorage.partitionId());
-@Override
-CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
-int partitionId = mvPartitionStorage.partitionId();
+return destroyPartitionPhysically(groupPartitionId).thenAccept(unused
-> {
+TableView tableView = tableConfig.value();
-// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
-// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+PersistentPageMemory pageMemory = dataRegion.pageMemory();
-mvPartitionStorage.close();
+int partitionId = groupPartitionId.getPartitionId();
+
+PartitionMeta meta = getOrCreatePartitionMeta(groupPartitionId,
ensurePartitionFilePageStore(tableView, groupPartitionId));
+
+inCheckpointLock(() -> {
+RowVersionFreeList rowVersionFreeList =
createRowVersionFreeList(tableView, partitionId, pageMemory, meta);
-int tableId = tableCfg.tableId().value();
+IndexColumnsFreeList indexColumnsFreeList
+= createIndexColumnsFreeList(tableView, partitionId,
rowVersionFreeList, pageMemory, meta);
-GroupPartitionId groupPartitionId = new GroupPartitionId(tableId,
partitionId);
+VersionChainTree versionChainTree =
createVersionChainTree(tableView, partitionId, rowVersionFreeList, pageMemory,
meta);
+IndexMetaTree indexMetaTree = createIndexMetaTree(tableView,
partitionId, rowVersionFreeList, pageMemory, meta);
+
+((PersistentPageMemoryMvPartitionStorage)
mvPartitionStorage).updateDataStructuresOnRebalance(
+meta,
+rowVersionFreeList,
+indexColumnsFreeList,
+versionChainTree,
+indexMetaTree
+);
+
+return null;
+});
+});
+}
+
+private CompletableFuture
destroyPartitionPhysically(GroupPartitionId groupPartitionId) {
dataRegion.filePageStoreManager().getStore(groupPartitionId).markToDestroy();
-dataRegion.pageMemory().invalidate(tableId, partitionId);
+dataRegion.pageMemory().invalidate(groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId());
return
dataRegion.checkpointManager().onPartitionDestruction(groupPartitionId)
.thenAccept(unused ->
dataRegion.partitionMetaManager().removeMeta(groupPartitionId))
.thenCompose(unused ->
dataRegion.filePageStoreManager().destroyPartition(groupPartitionId));
}
+
+private GroupPartitionId createGroupPartitionId(int partitionId) {
+return new GroupPartitionId(tableConfig.tableId().value(),
partitionId);
+}
+
+private V inCheckpointLock(Supplier supplier) {
+CheckpointTimeoutLock checkpointTimeoutLock =
dataRegion.checkpointManager().checkpointTimeoutLock();
+
+checkpointTimeoutLock.checkpointReadLock();
+
+try {
+return supplier.get();
+} finally {
+checkpointTimeoutLock.checkpointReadUnlock();
+}
+}
+
+private
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072415322
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+GroupPartitionId groupPartitionId =
createGroupPartitionId(mvPartitionStorage.partitionId());
-@Override
-CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
-int partitionId = mvPartitionStorage.partitionId();
+return destroyPartitionPhysically(groupPartitionId).thenAccept(unused
-> {
+TableView tableView = tableConfig.value();
-// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
-// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+PersistentPageMemory pageMemory = dataRegion.pageMemory();
-mvPartitionStorage.close();
+int partitionId = groupPartitionId.getPartitionId();
+
+PartitionMeta meta = getOrCreatePartitionMeta(groupPartitionId,
ensurePartitionFilePageStore(tableView, groupPartitionId));
+
+inCheckpointLock(() -> {
+RowVersionFreeList rowVersionFreeList =
createRowVersionFreeList(tableView, partitionId, pageMemory, meta);
-int tableId = tableCfg.tableId().value();
+IndexColumnsFreeList indexColumnsFreeList
+= createIndexColumnsFreeList(tableView, partitionId,
rowVersionFreeList, pageMemory, meta);
-GroupPartitionId groupPartitionId = new GroupPartitionId(tableId,
partitionId);
+VersionChainTree versionChainTree =
createVersionChainTree(tableView, partitionId, rowVersionFreeList, pageMemory,
meta);
+IndexMetaTree indexMetaTree = createIndexMetaTree(tableView,
partitionId, rowVersionFreeList, pageMemory, meta);
+
+((PersistentPageMemoryMvPartitionStorage)
mvPartitionStorage).updateDataStructuresOnRebalance(
+meta,
+rowVersionFreeList,
+indexColumnsFreeList,
+versionChainTree,
+indexMetaTree
+);
+
+return null;
+});
+});
+}
+
+private CompletableFuture
destroyPartitionPhysically(GroupPartitionId groupPartitionId) {
dataRegion.filePageStoreManager().getStore(groupPartitionId).markToDestroy();
-dataRegion.pageMemory().invalidate(tableId, partitionId);
+dataRegion.pageMemory().invalidate(groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId());
return
dataRegion.checkpointManager().onPartitionDestruction(groupPartitionId)
.thenAccept(unused ->
dataRegion.partitionMetaManager().removeMeta(groupPartitionId))
.thenCompose(unused ->
dataRegion.filePageStoreManager().destroyPartition(groupPartitionId));
}
+
+private GroupPartitionId createGroupPartitionId(int partitionId) {
+return new GroupPartitionId(tableConfig.tableId().value(),
partitionId);
+}
+
+private V inCheckpointLock(Supplier supplier) {
+CheckpointTimeoutLock checkpointTimeoutLock =
dataRegion.checkpointManager().checkpointTimeoutLock();
+
+checkpointTimeoutLock.checkpointReadLock();
+
+try {
+return supplier.get();
+} finally {
+checkpointTimeoutLock.checkpointReadUnlock();
+}
+}
+
+private
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072414034
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+GroupPartitionId groupPartitionId =
createGroupPartitionId(mvPartitionStorage.partitionId());
-@Override
-CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
-int partitionId = mvPartitionStorage.partitionId();
+return destroyPartitionPhysically(groupPartitionId).thenAccept(unused
-> {
+TableView tableView = tableConfig.value();
-// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
-// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+PersistentPageMemory pageMemory = dataRegion.pageMemory();
-mvPartitionStorage.close();
+int partitionId = groupPartitionId.getPartitionId();
+
+PartitionMeta meta = getOrCreatePartitionMeta(groupPartitionId,
ensurePartitionFilePageStore(tableView, groupPartitionId));
+
+inCheckpointLock(() -> {
+RowVersionFreeList rowVersionFreeList =
createRowVersionFreeList(tableView, partitionId, pageMemory, meta);
-int tableId = tableCfg.tableId().value();
+IndexColumnsFreeList indexColumnsFreeList
+= createIndexColumnsFreeList(tableView, partitionId,
rowVersionFreeList, pageMemory, meta);
-GroupPartitionId groupPartitionId = new GroupPartitionId(tableId,
partitionId);
+VersionChainTree versionChainTree =
createVersionChainTree(tableView, partitionId, rowVersionFreeList, pageMemory,
meta);
+IndexMetaTree indexMetaTree = createIndexMetaTree(tableView,
partitionId, rowVersionFreeList, pageMemory, meta);
+
+((PersistentPageMemoryMvPartitionStorage)
mvPartitionStorage).updateDataStructuresOnRebalance(
+meta,
+rowVersionFreeList,
+indexColumnsFreeList,
+versionChainTree,
+indexMetaTree
+);
+
+return null;
+});
+});
+}
+
+private CompletableFuture
destroyPartitionPhysically(GroupPartitionId groupPartitionId) {
dataRegion.filePageStoreManager().getStore(groupPartitionId).markToDestroy();
-dataRegion.pageMemory().invalidate(tableId, partitionId);
+dataRegion.pageMemory().invalidate(groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId());
return
dataRegion.checkpointManager().onPartitionDestruction(groupPartitionId)
.thenAccept(unused ->
dataRegion.partitionMetaManager().removeMeta(groupPartitionId))
.thenCompose(unused ->
dataRegion.filePageStoreManager().destroyPartition(groupPartitionId));
}
+
+private GroupPartitionId createGroupPartitionId(int partitionId) {
+return new GroupPartitionId(tableConfig.tableId().value(),
partitionId);
+}
+
+private V inCheckpointLock(Supplier supplier) {
+CheckpointTimeoutLock checkpointTimeoutLock =
dataRegion.checkpointManager().checkpointTimeoutLock();
+
+checkpointTimeoutLock.checkpointReadLock();
+
+try {
+return supplier.get();
+} finally {
+checkpointTimeoutLock.checkpointReadUnlock();
+}
+}
+
+private
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072346248
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -96,49 +98,22 @@ public boolean isVolatile() {
@Override
protected void finishDestruction() {
-dataRegion.pageMemory().onGroupDestroyed(tableCfg.tableId().value());
+
dataRegion.pageMemory().onGroupDestroyed(tableConfig.tableId().value());
}
@Override
public PersistentPageMemoryMvPartitionStorage createMvPartitionStorage(int
partitionId) {
-CompletableFuture partitionDestroyFuture =
partitionIdDestroyFutureMap.get(partitionId);
+waitPartitionToBeDestroyed(partitionId);
-if (partitionDestroyFuture != null) {
-try {
-// Time is chosen randomly (long enough) so as not to call
#join().
-partitionDestroyFuture.get(10, TimeUnit.SECONDS);
-} catch (Exception e) {
-throw new StorageException("Error waiting for the destruction
of the previous version of the partition: " + partitionId, e);
-}
-}
+TableView tableView = tableConfig.value();
-TableView tableView = tableCfg.value();
+GroupPartitionId groupPartitionId =
createGroupPartitionId(partitionId);
-GroupPartitionId groupPartitionId = new
GroupPartitionId(tableView.tableId(), partitionId);
+PartitionMeta meta =
getOrCreatePartitionMetaWithRecreatePartitionPageStoreIfRebalanceNotCompleted(groupPartitionId);
Review Comment:
I didn't put it in a separate lock block, but moved it outside of it.
This is not an error since we are reading the `PartitionMeta` directly from
the file, not using `PageMemory`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072334426
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+GroupPartitionId groupPartitionId =
createGroupPartitionId(mvPartitionStorage.partitionId());
-@Override
-CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
-int partitionId = mvPartitionStorage.partitionId();
+return destroyPartitionPhysically(groupPartitionId).thenAccept(unused
-> {
+TableView tableView = tableConfig.value();
-// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
-// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+PersistentPageMemory pageMemory = dataRegion.pageMemory();
-mvPartitionStorage.close();
+int partitionId = groupPartitionId.getPartitionId();
+
+PartitionMeta meta = getOrCreatePartitionMeta(groupPartitionId,
ensurePartitionFilePageStore(tableView, groupPartitionId));
+
+inCheckpointLock(() -> {
+RowVersionFreeList rowVersionFreeList =
createRowVersionFreeList(tableView, partitionId, pageMemory, meta);
-int tableId = tableCfg.tableId().value();
+IndexColumnsFreeList indexColumnsFreeList
+= createIndexColumnsFreeList(tableView, partitionId,
rowVersionFreeList, pageMemory, meta);
-GroupPartitionId groupPartitionId = new GroupPartitionId(tableId,
partitionId);
+VersionChainTree versionChainTree =
createVersionChainTree(tableView, partitionId, rowVersionFreeList, pageMemory,
meta);
+IndexMetaTree indexMetaTree = createIndexMetaTree(tableView,
partitionId, rowVersionFreeList, pageMemory, meta);
+
+((PersistentPageMemoryMvPartitionStorage)
mvPartitionStorage).updateDataStructuresOnRebalance(
+meta,
+rowVersionFreeList,
+indexColumnsFreeList,
+versionChainTree,
+indexMetaTree
+);
+
+return null;
+});
+});
+}
+
+private CompletableFuture
destroyPartitionPhysically(GroupPartitionId groupPartitionId) {
dataRegion.filePageStoreManager().getStore(groupPartitionId).markToDestroy();
-dataRegion.pageMemory().invalidate(tableId, partitionId);
+dataRegion.pageMemory().invalidate(groupPartitionId.getGroupId(),
groupPartitionId.getPartitionId());
Review Comment:
I think we need to fix this in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072329420
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
Review Comment:
It is not expected to be overridden, it overrides method
`org.apache.ignite.internal.storage.pagememory.AbstractPageMemoryTableStorage#clearStorageAndUpdateDataStructures`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072329420
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -405,42 +373,148 @@ private IndexMetaTree createIndexMetaTree(
}
@Override
-public CompletableFuture startRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
destroyMvPartitionStorage(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
+// It is enough for us to close the partition storage and its indexes
(do not destroy). Prepare the data region, checkpointer, and
+// compactor to remove the partition, and then simply delete the
partition file and its delta files.
+mvPartitionStorage.close();
-@Override
-public CompletableFuture abortRebalancePartition(int partitionId) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
+return
destroyPartitionPhysically(createGroupPartitionId(mvPartitionStorage.partitionId()));
}
@Override
-public CompletableFuture finishRebalancePartition(int partitionId,
long lastAppliedIndex, long lastAppliedTerm) {
-// TODO: IGNITE-18029 Implement
-throw new UnsupportedOperationException();
-}
+CompletableFuture
clearStorageAndUpdateDataStructures(AbstractPageMemoryMvPartitionStorage
mvPartitionStorage) {
Review Comment:
It is not expected to be overridden.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072328103
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/PersistentPageMemoryTableStorage.java:
##
@@ -210,7 +177,8 @@ private FilePageStore
ensurePartitionFilePageStore(TableView tableView, GroupPar
/**
* Returns id of the last started checkpoint, or {@code null} if no
checkpoints were started yet.
*/
-public @Nullable UUID lastCheckpointId() {
+@Nullable
+private UUID lastCheckpointId() {
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072279893
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/PersistentPageMemoryMvPartitionStorage.java:
##
@@ -216,51 +197,41 @@ public void lastApplied(long lastAppliedIndex, long
lastAppliedTerm) throws Stor
@Override
public long persistedIndex() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-return persistedIndex;
-} finally {
-closeBusyLock.leaveBusy();
-}
+return busy(() -> persistedIndex);
}
@Override
@Nullable
public RaftGroupConfiguration committedGroupConfiguration() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-replicationProtocolGroupConfigReadWriteLock.readLock().lock();
-
+return busy(() -> {
Review Comment:
Tried to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072276069
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java:
##
@@ -800,67 +732,65 @@ public PartitionTimestampCursor scan(HybridTimestamp
timestamp) throws StorageEx
} else {
return new TimestampCursor(treeCursor, timestamp);
}
-} finally {
-closeBusyLock.leaveBusy();
-}
+});
}
@Override
public @Nullable RowId closestRowId(RowId lowerBound) throws
StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
this::createStorageInfo);
-try (Cursor cursor = versionChainTree.find(new
VersionChainKey(lowerBound), null)) {
-return cursor.hasNext() ? cursor.next().rowId() : null;
-} catch (Exception e) {
-throw new StorageException("Error occurred while trying to read a
row id", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+try (Cursor cursor = versionChainTree.find(new
VersionChainKey(lowerBound), null)) {
+return cursor.hasNext() ? cursor.next().rowId() : null;
+} catch (Exception e) {
+throw new StorageException("Error occurred while trying to
read a row id", e);
+}
+});
}
@Override
public long rowsCount() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
this::createStorageInfo);
-try {
-return versionChainTree.size();
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Error occurred while fetching the
size.", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+try {
+return versionChainTree.size();
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Error occurred while fetching the
size", e);
+}
+});
}
private abstract class BasePartitionTimestampCursor implements
PartitionTimestampCursor {
-protected final Cursor treeCursor;
+final Cursor treeCursor;
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072275714
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java:
##
@@ -157,35 +141,31 @@ protected AbstractPageMemoryMvPartitionStorage(
* Starts a partition by initializing its internal structures.
*/
public void start() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+busy(() -> {
Review Comment:
Tried to fix it.
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/mv/AbstractPageMemoryMvPartitionStorage.java:
##
@@ -528,20 +516,16 @@ private static byte[] rowBytes(@Nullable BinaryRow row) {
@Override
public @Nullable BinaryRow addWrite(RowId rowId, @Nullable BinaryRow row,
UUID txId, UUID commitTableId, int commitPartitionId)
throws TxIdMismatchException, StorageException {
-assert rowId.partitionId() == partitionId : rowId;
-
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+assert rowId.partitionId() == partitionId : rowId;
-try {
VersionChain currentChain = findVersionChain(rowId);
if (currentChain == null) {
RowVersion newVersion = insertRowVersion(row, NULL_LINK);
-VersionChain versionChain =
VersionChain.createUncommitted(rowId, txId, commitTableId, commitPartitionId,
newVersion.link(),
-NULL_LINK);
+VersionChain versionChain =
VersionChain.createUncommitted(rowId, txId, commitTableId, commitPartitionId,
+newVersion.link(), NULL_LINK);
Review Comment:
Fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072257607
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -298,76 +276,74 @@ public void close() {
@Override
public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-advanceIfNeeded();
+return busy(() -> {
+try {
+advanceIfNeeded();
-return hasNext;
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Error while advancing the cursor",
e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return hasNext;
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Error while advancing the
cursor", e);
+}
+});
}
@Override
public IndexRow next() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-advanceIfNeeded();
+return busy(() -> {
+try {
+advanceIfNeeded();
-boolean hasNext = this.hasNext;
+boolean hasNext = this.hasNext;
-if (!hasNext) {
-throw new NoSuchElementException();
-}
+if (!hasNext) {
+throw new NoSuchElementException();
+}
-this.hasNext = null;
+this.hasNext = null;
-return toIndexRowImpl(treeRow);
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Error while advancing the cursor",
e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return toIndexRowImpl(treeRow);
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Error while advancing the
cursor", e);
+}
+});
}
@Override
public @Nullable IndexRow peek() {
-if (hasNext != null) {
-if (hasNext) {
-return toIndexRowImpl(treeRow);
+return busy(() -> {
Review Comment:
Same, but why not use a generic template for this class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072257176
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -298,76 +276,74 @@ public void close() {
@Override
public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-advanceIfNeeded();
+return busy(() -> {
Review Comment:
Same, but why not use a generic template for this class?
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -298,76 +276,74 @@ public void close() {
@Override
public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-advanceIfNeeded();
+return busy(() -> {
+try {
+advanceIfNeeded();
-return hasNext;
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Error while advancing the cursor",
e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return hasNext;
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Error while advancing the
cursor", e);
+}
+});
}
@Override
public IndexRow next() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
-
-try {
-advanceIfNeeded();
+return busy(() -> {
Review Comment:
Same, but why not use a generic template for this class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072256909
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -247,28 +233,20 @@ public void close() {
@Override
public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
-try {
return cursor.hasNext();
-} finally {
-closeBusyLock.leaveBusy();
-}
+});
}
@Override
public R next() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
Review Comment:
Same, but why not use a generic template for this class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072251321
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -247,28 +233,20 @@ public void close() {
@Override
public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
-try {
return cursor.hasNext();
-} finally {
-closeBusyLock.leaveBusy();
-}
+});
Review Comment:
Same, but why not use a generic template for this class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072251321
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -247,28 +233,20 @@ public void close() {
@Override
public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemorySortedIndexStorage.this::createStorageInfo);
-try {
return cursor.hasNext();
-} finally {
-closeBusyLock.leaveBusy();
-}
+});
Review Comment:
Also, but why not use a generic template for this class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072249467
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java:
##
@@ -97,100 +93,88 @@ public HashIndexDescriptor indexDescriptor() {
@Override
public Cursor get(BinaryTuple key) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
this::createStorageInfo);
-try {
-IndexColumns indexColumns = new IndexColumns(partitionId,
key.byteBuffer());
+try {
+IndexColumns indexColumns = new IndexColumns(partitionId,
key.byteBuffer());
-HashIndexRow lowerBound = new HashIndexRow(indexColumns,
lowestRowId);
-HashIndexRow upperBound = new HashIndexRow(indexColumns,
highestRowId);
+HashIndexRow lowerBound = new HashIndexRow(indexColumns,
lowestRowId);
+HashIndexRow upperBound = new HashIndexRow(indexColumns,
highestRowId);
-Cursor cursor = hashIndexTree.find(lowerBound,
upperBound);
+Cursor cursor = hashIndexTree.find(lowerBound,
upperBound);
-return new Cursor<>() {
-@Override
-public void close() {
-cursor.close();
-}
-
-@Override
-public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
+return new Cursor() {
+@Override
+public void close() {
+cursor.close();
}
-try {
-return cursor.hasNext();
-} finally {
-closeBusyLock.leaveBusy();
-}
-}
+@Override
+public boolean hasNext() {
+return busy(() -> {
+
throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemoryHashIndexStorage.this::createStorageInfo);
-@Override
-public RowId next() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
+return cursor.hasNext();
+});
}
-try {
-return cursor.next().rowId();
-} finally {
-closeBusyLock.leaveBusy();
+@Override
+public RowId next() {
+return busy(() -> {
+
throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemoryHashIndexStorage.this::createStorageInfo);
+
+return cursor.next().rowId();
+});
}
-}
-};
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Failed to create scan cursor", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+};
+} catch (Throwable e) {
+throw new StorageException("Failed to create scan cursor", e);
+}
+});
}
@Override
public void put(IndexRow row) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+busy(() -> {
+try {
+IndexColumns indexColumns = new IndexColumns(partitionId,
row.indexColumns().byteBuffer());
-try {
-IndexColumns indexColumns = new IndexColumns(partitionId,
row.indexColumns().byteBuffer());
+HashIndexRow hashIndexRow = new HashIndexRow(indexColumns,
row.rowId());
-HashIndexRow hashIndexRow = new HashIndexRow(indexColumns,
row.rowId());
+var insert = new InsertHashIndexRowInvokeClosure(hashIndexRow,
freeList, hashIndexTree.inlineSize());
-var insert = new InsertHashIndexRowInvokeClosure(hashIndexRow,
freeList, hashIndexTree.inlineSize());
+hashIndexTree.invoke(hashIndexRow, null, insert);
-hashIndexTree.invoke(hashIndexRow, null, insert);
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Failed to put value into index", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return null;
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Failed to put value into index",
e);
+}
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072249794
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -105,71 +102,65 @@ public SortedIndexDescriptor indexDescriptor() {
@Override
public Cursor get(BinaryTuple key) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
Review Comment:
Tried to fix it.
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -105,71 +102,65 @@ public SortedIndexDescriptor indexDescriptor() {
@Override
public Cursor get(BinaryTuple key) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
this::createStorageInfo);
-try {
-SortedIndexRowKey lowerBound = toSortedIndexRow(key, lowestRowId);
+try {
+SortedIndexRowKey lowerBound = toSortedIndexRow(key,
lowestRowId);
-SortedIndexRowKey upperBound = toSortedIndexRow(key, highestRowId);
+SortedIndexRowKey upperBound = toSortedIndexRow(key,
highestRowId);
-return convertCursor(sortedIndexTree.find(lowerBound, upperBound),
SortedIndexRow::rowId);
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Failed to create scan cursor", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return convertCursor(sortedIndexTree.find(lowerBound,
upperBound), SortedIndexRow::rowId);
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Failed to create scan cursor", e);
+}
+});
}
@Override
public void put(IndexRow row) {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+busy(() -> {
Review Comment:
Tried to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072250057
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/sorted/PageMemorySortedIndexStorage.java:
##
@@ -105,71 +102,65 @@ public SortedIndexDescriptor indexDescriptor() {
@Override
public Cursor get(BinaryTuple key) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
this::createStorageInfo);
-try {
-SortedIndexRowKey lowerBound = toSortedIndexRow(key, lowestRowId);
+try {
+SortedIndexRowKey lowerBound = toSortedIndexRow(key,
lowestRowId);
-SortedIndexRowKey upperBound = toSortedIndexRow(key, highestRowId);
+SortedIndexRowKey upperBound = toSortedIndexRow(key,
highestRowId);
-return convertCursor(sortedIndexTree.find(lowerBound, upperBound),
SortedIndexRow::rowId);
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Failed to create scan cursor", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return convertCursor(sortedIndexTree.find(lowerBound,
upperBound), SortedIndexRow::rowId);
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Failed to create scan cursor", e);
+}
+});
}
@Override
public void put(IndexRow row) {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+busy(() -> {
+try {
+SortedIndexRow sortedIndexRow =
toSortedIndexRow(row.indexColumns(), row.rowId());
-try {
-SortedIndexRow sortedIndexRow =
toSortedIndexRow(row.indexColumns(), row.rowId());
+var insert = new
InsertSortedIndexRowInvokeClosure(sortedIndexRow, freeList,
sortedIndexTree.inlineSize());
-var insert = new InsertSortedIndexRowInvokeClosure(sortedIndexRow,
freeList, sortedIndexTree.inlineSize());
+sortedIndexTree.invoke(sortedIndexRow, null, insert);
-sortedIndexTree.invoke(sortedIndexRow, null, insert);
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Failed to put value into index", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+return null;
+} catch (IgniteInternalCheckedException e) {
+throw new StorageException("Failed to put value into index",
e);
+}
+});
}
@Override
public void remove(IndexRow row) {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+busy(() -> {
Review Comment:
Tried to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072249213
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java:
##
@@ -97,100 +93,88 @@ public HashIndexDescriptor indexDescriptor() {
@Override
public Cursor get(BinaryTuple key) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
Review Comment:
Tried to fix it.
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/index/hash/PageMemoryHashIndexStorage.java:
##
@@ -97,100 +93,88 @@ public HashIndexDescriptor indexDescriptor() {
@Override
public Cursor get(BinaryTuple key) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+return busy(() -> {
+throwExceptionIfStorageInProgressOfRebalance(state.get(),
this::createStorageInfo);
-try {
-IndexColumns indexColumns = new IndexColumns(partitionId,
key.byteBuffer());
+try {
+IndexColumns indexColumns = new IndexColumns(partitionId,
key.byteBuffer());
-HashIndexRow lowerBound = new HashIndexRow(indexColumns,
lowestRowId);
-HashIndexRow upperBound = new HashIndexRow(indexColumns,
highestRowId);
+HashIndexRow lowerBound = new HashIndexRow(indexColumns,
lowestRowId);
+HashIndexRow upperBound = new HashIndexRow(indexColumns,
highestRowId);
-Cursor cursor = hashIndexTree.find(lowerBound,
upperBound);
+Cursor cursor = hashIndexTree.find(lowerBound,
upperBound);
-return new Cursor<>() {
-@Override
-public void close() {
-cursor.close();
-}
-
-@Override
-public boolean hasNext() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
+return new Cursor() {
+@Override
+public void close() {
+cursor.close();
}
-try {
-return cursor.hasNext();
-} finally {
-closeBusyLock.leaveBusy();
-}
-}
+@Override
+public boolean hasNext() {
+return busy(() -> {
+
throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemoryHashIndexStorage.this::createStorageInfo);
-@Override
-public RowId next() {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
+return cursor.hasNext();
+});
}
-try {
-return cursor.next().rowId();
-} finally {
-closeBusyLock.leaveBusy();
+@Override
+public RowId next() {
+return busy(() -> {
+
throwExceptionIfStorageInProgressOfRebalance(state.get(),
PageMemoryHashIndexStorage.this::createStorageInfo);
+
+return cursor.next().rowId();
+});
}
-}
-};
-} catch (IgniteInternalCheckedException e) {
-throw new StorageException("Failed to create scan cursor", e);
-} finally {
-closeBusyLock.leaveBusy();
-}
+};
+} catch (Throwable e) {
+throw new StorageException("Failed to create scan cursor", e);
+}
+});
}
@Override
public void put(IndexRow row) throws StorageException {
-if (!closeBusyLock.enterBusy()) {
-throwStorageClosedException();
-}
+busy(() -> {
Review Comment:
Tried to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072225481
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/AbstractPageMemoryTableStorage.java:
##
@@ -268,12 +377,52 @@ private void checkPartitionId(int partitionId) {
int partitions = mvPartitions.length();
if (partitionId < 0 || partitionId >= partitions) {
-throw new IllegalArgumentException(S.toString(
-"Unable to access partition with id outside of configured
range",
-"table", tableCfg.value().name(), false,
-"partitionId", partitionId, false,
-"partitions", partitions, false
+throw new IllegalArgumentException(IgniteStringFormatter.format(
+"Unable to access partition with id outside of configured
range: [table={}, partitionId={}, partitions={}]",
+getTableName(),
+partitionId,
+partitions
));
}
}
+
+/**
+ * Returns multi-versioned partition storage without using {@link
#busyLock}.
+ *
+ * @param partitionId Partition ID.
+ * @return {@code Null} if there is no storage.
+ */
+@Nullable
+AbstractPageMemoryMvPartitionStorage
getMvPartitionStorageWithoutBusyLock(int partitionId) {
+checkPartitionId(partitionId);
+
+return mvPartitions.get(partitionId);
+}
+
+/**
+ * Returns multi-versioned partition storage, if it doesn't exist it will
throw an exception from the
+ * {@code missingStorageExceptionFunction}, without using {@link
#busyLock}.
+ *
+ * @param partitionId Partition ID.
+ * @param missingStorageExceptionFunction Function to create an exception
if the store is missing.
+ */
+AbstractPageMemoryMvPartitionStorage getMvPartitionStorageWithoutBusyLock(
Review Comment:
Replaced with `null` check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072202522
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/AbstractPageMemoryTableStorage.java:
##
@@ -268,12 +377,52 @@ private void checkPartitionId(int partitionId) {
int partitions = mvPartitions.length();
if (partitionId < 0 || partitionId >= partitions) {
-throw new IllegalArgumentException(S.toString(
-"Unable to access partition with id outside of configured
range",
-"table", tableCfg.value().name(), false,
-"partitionId", partitionId, false,
-"partitions", partitions, false
+throw new IllegalArgumentException(IgniteStringFormatter.format(
+"Unable to access partition with id outside of configured
range: [table={}, partitionId={}, partitions={}]",
+getTableName(),
+partitionId,
+partitions
));
}
}
+
+/**
+ * Returns multi-versioned partition storage without using {@link
#busyLock}.
+ *
+ * @param partitionId Partition ID.
+ * @return {@code Null} if there is no storage.
+ */
+@Nullable
+AbstractPageMemoryMvPartitionStorage
getMvPartitionStorageWithoutBusyLock(int partitionId) {
+checkPartitionId(partitionId);
+
+return mvPartitions.get(partitionId);
+}
+
+/**
+ * Returns multi-versioned partition storage, if it doesn't exist it will
throw an exception from the
+ * {@code missingStorageExceptionFunction}, without using {@link
#busyLock}.
+ *
+ * @param partitionId Partition ID.
+ * @param missingStorageExceptionFunction Function to create an exception
if the store is missing.
+ */
+AbstractPageMemoryMvPartitionStorage getMvPartitionStorageWithoutBusyLock(
+int partitionId,
+Function
missingStorageExceptionFunction
+) {
+AbstractPageMemoryMvPartitionStorage mvPartitionStorage =
getMvPartitionStorageWithoutBusyLock(partitionId);
+
+if (mvPartitionStorage == null) {
+throw
missingStorageExceptionFunction.apply(IgniteStringFormatter.format("Partition
ID {} does not exist", partitionId));
+}
+
+return mvPartitionStorage;
+}
+
+/**
+ * Returns table name.
+ */
+public String getTableName() {
Review Comment:
Agree.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072201651
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/AbstractPageMemoryTableStorage.java:
##
@@ -167,86 +203,79 @@ public CompletableFuture destroy() {
@Override
public AbstractPageMemoryMvPartitionStorage getOrCreateMvPartition(int
partitionId) throws StorageException {
-AbstractPageMemoryMvPartitionStorage partition =
getMvPartition(partitionId);
+return inBusyLock(busyLock, () -> {
+AbstractPageMemoryMvPartitionStorage partition =
getMvPartitionStorageWithoutBusyLock(partitionId);
-if (partition != null) {
-return partition;
-}
+if (partition != null) {
+return partition;
+}
-partition = createMvPartitionStorage(partitionId);
+partition = createMvPartitionStorage(partitionId);
-partition.start();
+partition.start();
-mvPartitions.set(partitionId, partition);
+mvPartitions.set(partitionId, partition);
-return partition;
+return partition;
+});
}
@Override
public @Nullable AbstractPageMemoryMvPartitionStorage getMvPartition(int
partitionId) {
-assert started : "Storage has not started yet";
-
-checkPartitionId(partitionId);
-
-return mvPartitions.get(partitionId);
+return inBusyLock(busyLock, () ->
getMvPartitionStorageWithoutBusyLock(partitionId));
}
@Override
public CompletableFuture destroyPartition(int partitionId) {
-assert started : "Storage has not started yet";
+return inBusyLock(busyLock, () -> {
Review Comment:
Tried to fix it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r107218
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/AbstractPageMemoryTableStorage.java:
##
@@ -18,61 +18,89 @@
package org.apache.ignite.internal.storage.pagememory;
import static java.util.concurrent.CompletableFuture.completedFuture;
+import static
org.apache.ignite.internal.storage.MvPartitionStorage.REBALANCE_IN_PROGRESS;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReferenceArray;
+import java.util.function.Function;
import org.apache.ignite.internal.pagememory.DataRegion;
import org.apache.ignite.internal.pagememory.PageMemory;
+import org.apache.ignite.internal.pagememory.freelist.FreeList;
+import org.apache.ignite.internal.pagememory.reuse.ReuseList;
+import org.apache.ignite.internal.pagememory.tree.BplusTree;
import org.apache.ignite.internal.schema.configuration.TableConfiguration;
import org.apache.ignite.internal.schema.configuration.TableView;
import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.StorageException;
+import org.apache.ignite.internal.storage.StorageRebalanceException;
import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.index.HashIndexStorage;
import org.apache.ignite.internal.storage.index.SortedIndexStorage;
import
org.apache.ignite.internal.storage.pagememory.mv.AbstractPageMemoryMvPartitionStorage;
-import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.lang.IgniteStringFormatter;
import org.jetbrains.annotations.Nullable;
/**
* Abstract table storage implementation based on {@link PageMemory}.
*/
public abstract class AbstractPageMemoryTableStorage implements MvTableStorage
{
-protected final TableConfiguration tableCfg;
+protected static final VarHandle CLOSED;
-protected TablesConfiguration tablesConfiguration;
+static {
+try {
+CLOSED =
MethodHandles.lookup().findVarHandle(AbstractPageMemoryTableStorage.class,
"closed", boolean.class);
+} catch (ReflectiveOperationException e) {
+throw new ExceptionInInitializerError(e);
+}
+}
+
+protected final TableConfiguration tableConfig;
-protected volatile boolean started;
+protected final TablesConfiguration tablesConfig;
protected volatile
AtomicReferenceArray mvPartitions;
-protected final ConcurrentMap>
partitionIdDestroyFutureMap = new ConcurrentHashMap<>();
+protected final ConcurrentMap>
destroyFutureByPartitionId = new ConcurrentHashMap<>();
+
+protected final ConcurrentMap>
rebalanceFutureByPartitionId = new ConcurrentHashMap<>();
+
+/** Busy lock to stop synchronously. */
+protected final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+/** To avoid double closure. */
+@SuppressWarnings("unused")
+protected volatile boolean closed;
Review Comment:
Returned it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072185181
##
modules/storage-page-memory/src/main/java/org/apache/ignite/internal/storage/pagememory/AbstractPageMemoryTableStorage.java:
##
@@ -18,61 +18,89 @@
package org.apache.ignite.internal.storage.pagememory;
import static java.util.concurrent.CompletableFuture.completedFuture;
+import static
org.apache.ignite.internal.storage.MvPartitionStorage.REBALANCE_IN_PROGRESS;
+import static org.apache.ignite.internal.util.IgniteUtils.inBusyLock;
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReferenceArray;
+import java.util.function.Function;
import org.apache.ignite.internal.pagememory.DataRegion;
import org.apache.ignite.internal.pagememory.PageMemory;
+import org.apache.ignite.internal.pagememory.freelist.FreeList;
+import org.apache.ignite.internal.pagememory.reuse.ReuseList;
+import org.apache.ignite.internal.pagememory.tree.BplusTree;
import org.apache.ignite.internal.schema.configuration.TableConfiguration;
import org.apache.ignite.internal.schema.configuration.TableView;
import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.storage.MvPartitionStorage;
import org.apache.ignite.internal.storage.StorageException;
+import org.apache.ignite.internal.storage.StorageRebalanceException;
import org.apache.ignite.internal.storage.engine.MvTableStorage;
import org.apache.ignite.internal.storage.index.HashIndexStorage;
import org.apache.ignite.internal.storage.index.SortedIndexStorage;
import
org.apache.ignite.internal.storage.pagememory.mv.AbstractPageMemoryMvPartitionStorage;
-import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.lang.IgniteStringFormatter;
import org.jetbrains.annotations.Nullable;
/**
* Abstract table storage implementation based on {@link PageMemory}.
*/
public abstract class AbstractPageMemoryTableStorage implements MvTableStorage
{
-protected final TableConfiguration tableCfg;
+protected static final VarHandle CLOSED;
Review Comment:
Returned it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072184092
##
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java:
##
@@ -811,8 +885,8 @@ private void checkForPresenceRows(
) {
for (IgniteTuple3 row : rows) {
assertThat(
-
getAll(mvPartitionStorage.scanVersions(row.get1())).stream().map(ReadResult::binaryRow).collect(toList()),
-containsInAnyOrder(row.get2())
+
toListOfByteArrays(mvPartitionStorage.scanVersions(row.get1())),
+containsInAnyOrder(row.get2().bytes())
Review Comment:
There are no bugs, inside it is checked that if these are arrays, then it is
checked by their lengths and elements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072181314
##
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java:
##
@@ -548,6 +573,51 @@ public void testDestroyTableStorage() throws Exception {
assertThat(tableStorage.destroy(), willCompleteSuccessfully());
}
+/**
+ * Checks that if we restart the storages after a crash in the middle of a
rebalance, the storages will be empty.
+ */
+@Test
+public void testRestartStoragesAfterFailOnMiddleOfRebalance() {
Review Comment:
Fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
[GitHub] [ignite-3] tkalkirill commented on a diff in pull request #1506: IGNITE-18029 Implementation of a full rebalance for PersistentPageMemoryMvPartitionStorage on receiver
tkalkirill commented on code in PR #1506:
URL: https://github.com/apache/ignite-3/pull/1506#discussion_r1072180599
##
modules/storage-api/src/testFixtures/java/org/apache/ignite/internal/storage/AbstractMvTableStorageTest.java:
##
@@ -88,25 +92,46 @@ public abstract class AbstractMvTableStorageTest extends
BaseMvStoragesTest {
/** Partition id for 1 storage. */
protected static final int PARTITION_ID_1 = 1 << 8;
-private MvTableStorage tableStorage;
+protected MvTableStorage tableStorage;
private TableIndexView sortedIdx;
private TableIndexView hashIdx;
+private StorageEngine storageEngine;
+
/**
* Initializes the internal structures needed for tests.
*
* This method *MUST* always be called in either subclass' constructor
or setUp method.
*/
-protected final void initialize(MvTableStorage tableStorage,
TablesConfiguration tablesCfg) {
-createTestTable(tableStorage.configuration());
-createTestIndexes(tablesCfg);
+protected final void initialize(StorageEngine storageEngine,
TablesConfiguration tablesConfig) {
+createTestTable(getTableConfig(tablesConfig));
+createTestIndexes(tablesConfig);
+
+this.storageEngine = storageEngine;
Review Comment:
Makes sense, fixed it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscr...@ignite.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
