rpuch commented on code in PR #1325:
URL: https://github.com/apache/ignite-3/pull/1325#discussion_r1030075512
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManager.java:
##########
@@ -230,78 +269,86 @@ public long allocatePage(int grpId, int partId, byte
flags) throws IgniteInterna
}
/**
- * Initializing the file page stores for a group.
+ * Initialization of the page storage for the group partition.
*
* @param tableName Table name.
* @param tableId Integer table id.
- * @param partitions Partition number, must be greater than {@code 0} and
less {@link PageIdAllocator#MAX_PARTITION_ID}.
+ * @param partitionId Partition ID, must be between {@code 0} (inclusive)
and {@link PageIdAllocator#MAX_PARTITION_ID} (inclusive).
* @throws IgniteInternalCheckedException If failed.
*/
- public void initialize(String tableName, int tableId, int partitions)
throws IgniteInternalCheckedException {
- assert partitions > 0 && partitions < MAX_PARTITION_ID : partitions;
+ public void initialize(String tableName, int tableId, int partitionId)
throws IgniteInternalCheckedException {
+ assert partitionId >= 0 && partitionId <= MAX_PARTITION_ID :
partitionId;
- initGroupDirLock.lock(tableId);
+ stripedLock.lock(tableId + partitionId);
try {
- if (!groupPageStores.containsPageStores(tableId)) {
- List<FilePageStore> partitionFilePageStores =
createFilePageStores(tableId, partitions);
+ if (!groupPageStores.contains(tableId, partitionId)) {
+ Path tableWorkDir = ensureGroupWorkDir(tableId);
+
+ ByteBuffer buffer = allocateBuffer(pageSize);
+
+ try {
+ Path partFilePath =
tableWorkDir.resolve(String.format(PART_FILE_TEMPLATE, partitionId));
- List<FilePageStore> old = groupPageStores.put(tableId,
partitionFilePageStores);
+ Path[] partDeltaFiles =
findPartitionDeltaFiles(tableWorkDir, partitionId);
- assert old == null : tableName;
+ FilePageStore filePageStore =
filePageStoreFactory.createPageStore(buffer.rewind(), partFilePath,
partDeltaFiles);
+
+ FilePageStore previous = groupPageStores.put(tableId,
partitionId, filePageStore);
+
+ assert previous == null : IgniteStringFormatter.format(
+ "Parallel creation is not allowed: [tableName={},
tableId={}, partitionId={}]",
+ tableName,
+ tableId,
+ partitionId
+ );
+ } finally {
+ freeBuffer(buffer);
+ }
}
} catch (IgniteInternalCheckedException e) {
// TODO: IGNITE-16899 By analogy with 2.0, fail a node
throw e;
} finally {
- initGroupDirLock.unlock(tableId);
+ stripedLock.unlock(tableId + partitionId);
}
}
/**
- * Returns collection of related partition file page stores for group.
+ * Returns the group partition page stores.
*
* @param grpId Group ID.
*/
- public @Nullable List<FilePageStore> getStores(int grpId) {
+ public @Nullable GroupPageStores<FilePageStore> getStores(int grpId) {
return groupPageStores.get(grpId);
}
/**
- * Returns all page stores of all groups.
+ * Returns view for all page stores of all groups.
*/
- public Collection<List<FilePageStore>> allPageStores() {
- return groupPageStores.allPageStores();
+ public Collection<GroupPageStores<FilePageStore>> allPageStores() {
+ return groupPageStores.getAll();
}
/**
* Returns partition file page store for the corresponding parameters.
*
- * @param grpId Group ID.
- * @param partId Partition ID, from {@code 0} to {@link
PageIdAllocator#MAX_PARTITION_ID} (inclusive).
- * @throws IgniteInternalCheckedException If group or partition with the
given ID was not created.
+ * @param groupId Group ID.
+ * @param partitionId Partition ID, from {@code 0} (inclusive) to {@link
PageIdAllocator#MAX_PARTITION_ID} (inclusive).
*/
- public FilePageStore getStore(int grpId, int partId) throws
IgniteInternalCheckedException {
- assert partId >= 0 && partId <= MAX_PARTITION_ID : partId;
+ public @Nullable FilePageStore getStore(int groupId, int partitionId) {
Review Comment:
Please specify (in the javadoc) the conditions under which this method
returns `null`. Does it happen if, and only if, the partition is destroyed?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/store/FilePageStoreManager.java:
##########
@@ -230,78 +269,86 @@ public long allocatePage(int grpId, int partId, byte
flags) throws IgniteInterna
}
/**
- * Initializing the file page stores for a group.
+ * Initialization of the page storage for the group partition.
*
* @param tableName Table name.
* @param tableId Integer table id.
- * @param partitions Partition number, must be greater than {@code 0} and
less {@link PageIdAllocator#MAX_PARTITION_ID}.
+ * @param partitionId Partition ID, must be between {@code 0} (inclusive)
and {@link PageIdAllocator#MAX_PARTITION_ID} (inclusive).
* @throws IgniteInternalCheckedException If failed.
*/
- public void initialize(String tableName, int tableId, int partitions)
throws IgniteInternalCheckedException {
- assert partitions > 0 && partitions < MAX_PARTITION_ID : partitions;
+ public void initialize(String tableName, int tableId, int partitionId)
throws IgniteInternalCheckedException {
+ assert partitionId >= 0 && partitionId <= MAX_PARTITION_ID :
partitionId;
- initGroupDirLock.lock(tableId);
+ stripedLock.lock(tableId + partitionId);
try {
- if (!groupPageStores.containsPageStores(tableId)) {
- List<FilePageStore> partitionFilePageStores =
createFilePageStores(tableId, partitions);
+ if (!groupPageStores.contains(tableId, partitionId)) {
+ Path tableWorkDir = ensureGroupWorkDir(tableId);
+
+ ByteBuffer buffer = allocateBuffer(pageSize);
+
+ try {
+ Path partFilePath =
tableWorkDir.resolve(String.format(PART_FILE_TEMPLATE, partitionId));
- List<FilePageStore> old = groupPageStores.put(tableId,
partitionFilePageStores);
+ Path[] partDeltaFiles =
findPartitionDeltaFiles(tableWorkDir, partitionId);
- assert old == null : tableName;
+ FilePageStore filePageStore =
filePageStoreFactory.createPageStore(buffer.rewind(), partFilePath,
partDeltaFiles);
+
+ FilePageStore previous = groupPageStores.put(tableId,
partitionId, filePageStore);
+
+ assert previous == null : IgniteStringFormatter.format(
+ "Parallel creation is not allowed: [tableName={},
tableId={}, partitionId={}]",
+ tableName,
+ tableId,
+ partitionId
+ );
+ } finally {
+ freeBuffer(buffer);
+ }
}
} catch (IgniteInternalCheckedException e) {
// TODO: IGNITE-16899 By analogy with 2.0, fail a node
throw e;
} finally {
- initGroupDirLock.unlock(tableId);
+ stripedLock.unlock(tableId + partitionId);
}
}
/**
- * Returns collection of related partition file page stores for group.
+ * Returns the group partition page stores.
*
* @param grpId Group ID.
*/
- public @Nullable List<FilePageStore> getStores(int grpId) {
+ public @Nullable GroupPageStores<FilePageStore> getStores(int grpId) {
return groupPageStores.get(grpId);
}
/**
- * Returns all page stores of all groups.
+ * Returns view for all page stores of all groups.
*/
- public Collection<List<FilePageStore>> allPageStores() {
- return groupPageStores.allPageStores();
+ public Collection<GroupPageStores<FilePageStore>> allPageStores() {
+ return groupPageStores.getAll();
}
/**
* Returns partition file page store for the corresponding parameters.
*
- * @param grpId Group ID.
- * @param partId Partition ID, from {@code 0} to {@link
PageIdAllocator#MAX_PARTITION_ID} (inclusive).
- * @throws IgniteInternalCheckedException If group or partition with the
given ID was not created.
+ * @param groupId Group ID.
+ * @param partitionId Partition ID, from {@code 0} (inclusive) to {@link
PageIdAllocator#MAX_PARTITION_ID} (inclusive).
*/
- public FilePageStore getStore(int grpId, int partId) throws
IgniteInternalCheckedException {
- assert partId >= 0 && partId <= MAX_PARTITION_ID : partId;
+ public @Nullable FilePageStore getStore(int groupId, int partitionId) {
Review Comment:
Also, this method is called 3 times in this class (in `read()`, `write()`
and `allocate()`), and the result is never checked for `null`, but methods are
invoked on the result right off the bat. I suggest adding a method like
`getRequiredStore()` that would call `getStore()`, throw if the result is null
(with a proper explanation) and then return non-null result.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java:
##########
@@ -452,9 +456,21 @@ private void syncUpdatedPageStores(
return;
}
- fsyncDeltaFilePageStoreOnCheckpointThread(entry.getKey(),
entry.getValue());
+ GroupPartitionId partitionId = entry.getKey();
+
+ FilePageStore filePageStore =
filePageStoreManager.getStore(partitionId.getGroupId(),
partitionId.getPartitionId());
+
+ if (filePageStore == null ||
filePageStore.isMarkedToDestroy()) {
Review Comment:
Let's replace
```
if (COND) continue;
REST;
```
with
```
if (!COND) {
REST
}
```
After this transformation, lines 459-473 will become identical to lines
498-510, so a method could be extracted to avoid code duplication.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
