ibessonov commented on code in PR #7189:
URL: https://github.com/apache/ignite-3/pull/7189#discussion_r2601671767
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java:
##########
@@ -404,23 +405,44 @@ public void switchWriteIntents(
// However, we still need to run `onApplication` if it is not null,
e.g. called in TxCleanupCommand handler in PartitionListener
// to update indexes. In this case it should be executed under
`runConsistently`.
if (!pendingRowIds.isEmpty() || onApplication != null) {
- storage.runConsistently(locker -> {
- pendingRowIds.forEach(locker::lock);
-
- // Here we don't need to check for mismatch of the transaction
that created the write intent and commits it. Since the
- // commit can happen in #handleUpdate and #handleUpdateAll.
- if (commit) {
- performCommitWrite(txId, pendingRowIds, commitTimestamp);
- } else {
- performAbortWrite(txId, pendingRowIds, indexIds);
- }
-
- if (onApplication != null) {
- onApplication.run();
- }
-
- return null;
- });
+ Iterator<RowId> pendingRowIdsIterator = pendingRowIds.iterator();
+ boolean finished = false;
+ while (!finished) {
+ finished = storage.runConsistently(locker -> {
+ int modificationsCount = 0;
+ boolean shouldRelease = false;
+ while (pendingRowIdsIterator.hasNext()) {
+ RowId rowId = pendingRowIdsIterator.next();
+ locker.lock(rowId);
+
+ // Here we don't need to check for mismatch of the
transaction that created the write intent and commits it.
+ // Since the commit can happen in #handleUpdate and
#handleUpdateAll.
+ if (commit) {
+ storage.commitWrite(rowId, commitTimestamp, txId);
+ modificationsCount++;
+ } else {
+ performAbortWrite(txId, rowId, indexIds);
+ }
+
+ shouldRelease = locker.shouldRelease();
+ if (shouldRelease) {
+ break;
+ }
+ }
+ if (commit) {
+ modificationCounter.updateValue(modificationsCount,
commitTimestamp);
+ }
+ if (shouldRelease) {
Review Comment:
```suggestion
}
if (shouldRelease) {
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java:
##########
@@ -404,23 +405,44 @@ public void switchWriteIntents(
// However, we still need to run `onApplication` if it is not null,
e.g. called in TxCleanupCommand handler in PartitionListener
// to update indexes. In this case it should be executed under
`runConsistently`.
if (!pendingRowIds.isEmpty() || onApplication != null) {
- storage.runConsistently(locker -> {
- pendingRowIds.forEach(locker::lock);
-
- // Here we don't need to check for mismatch of the transaction
that created the write intent and commits it. Since the
- // commit can happen in #handleUpdate and #handleUpdateAll.
- if (commit) {
- performCommitWrite(txId, pendingRowIds, commitTimestamp);
- } else {
- performAbortWrite(txId, pendingRowIds, indexIds);
- }
-
- if (onApplication != null) {
- onApplication.run();
- }
-
- return null;
- });
+ Iterator<RowId> pendingRowIdsIterator = pendingRowIds.iterator();
+ boolean finished = false;
+ while (!finished) {
+ finished = storage.runConsistently(locker -> {
+ int modificationsCount = 0;
+ boolean shouldRelease = false;
+ while (pendingRowIdsIterator.hasNext()) {
+ RowId rowId = pendingRowIdsIterator.next();
+ locker.lock(rowId);
+
+ // Here we don't need to check for mismatch of the
transaction that created the write intent and commits it.
+ // Since the commit can happen in #handleUpdate and
#handleUpdateAll.
+ if (commit) {
+ storage.commitWrite(rowId, commitTimestamp, txId);
+ modificationsCount++;
+ } else {
+ performAbortWrite(txId, rowId, indexIds);
+ }
+
+ shouldRelease = locker.shouldRelease();
+ if (shouldRelease) {
+ break;
+ }
+ }
+ if (commit) {
Review Comment:
```suggestion
}
if (commit) {
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/StorageUpdateHandler.java:
##########
@@ -454,24 +476,37 @@ private void performCommitWrite(UUID txId, Set<RowId>
pendingRowIds, HybridTimes
*/
private void performAbortWrite(UUID txId, Set<RowId> pendingRowIds,
@Nullable List<Integer> indexIds) {
Review Comment:
I believe the second usage of this method only passes `Set.of(rowId)`, which
means that the method is redundant
--
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]
