sashapolo commented on code in PR #6799:
URL: https://github.com/apache/ignite-3/pull/6799#discussion_r2435711867
##########
modules/core/src/test/java/org/apache/ignite/internal/util/IgniteUtilsTest.java:
##########
@@ -217,4 +218,11 @@ void testByteBufferToByteArray() {
ByteBuffer smallDirectBuffer =
bigDirectBuffer.position(1).limit(4).slice();
assertArrayEquals(new byte[] {1, 2, 3},
byteBufferToByteArray(smallDirectBuffer));
}
+
+ @Test
+ void test() {
Review Comment:
???
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/compaction/Compactor.java:
##########
@@ -92,6 +94,9 @@ public class Compactor extends IgniteWorker {
private final PartitionDestructionLockManager
partitionDestructionLockManager;
+ /** Latch for pausing and resuming the compaction, {@code null} if there
was no pause. */
+ private final AtomicReference<CountDownLatch> pauseLatchRef = new
AtomicReference<>();
Review Comment:
I'm not sure whether I like this approach for pausing and resuming. I would
suggest to use `wait/notify` or a Lock with a Condition.
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/compaction/Compactor.java:
##########
@@ -396,6 +403,8 @@ void mergeDeltaFileToMainFile(
long pageOffset = deltaFilePageStore.pageOffset(pageIndex);
+ pauseCompactionIfNeeded();
Review Comment:
How did you identify which places are suitable for calling
`pauseCompactionIfNeeded()`? What if we just called it once per every cycle
iteration?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java:
##########
@@ -346,6 +346,8 @@ void doCheckpoint() throws IgniteInternalCheckedException {
Checkpoint chp = null;
try {
+ compactor.pause();
Review Comment:
It is very weird that we pause the compactor here and then resume it in two
different places. Why is it written that way?
##########
modules/page-memory/src/main/java/org/apache/ignite/internal/pagememory/persistence/compaction/Compactor.java:
##########
@@ -485,4 +500,45 @@ private DeltaFileForCompaction(
this.deltaFilePageStoreIo = deltaFilePageStoreIo;
}
}
+
+ /**
+ * Pauses the compactor until it is resumed or compactor or stopped. It is
expected that this method will not be called multiple times
+ * in parallel and subsequent calls will strictly be calls after {@link
#resume}.
+ */
+ public void pause() {
+ boolean casResult = pauseLatchRef.compareAndSet(null, new
CountDownLatch(1));
+
+ assert casResult : "It is expected that there will be no parallel
pause, resume or previous one has ended: " + pauseLatchRef.get();
+ }
+
+ /** Resumes the compactor if it was paused. It is expected that this
method will not be called multiple times in parallel. */
+ public void resume() {
+ CountDownLatch latch = pauseLatchRef.get();
+
+ boolean casResult = pauseLatchRef.compareAndSet(latch, null);
Review Comment:
This can be replaced with `getAndSet`. We will lose the `casResult` but you
are checking for a very small race interval here anyway.
--
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]
