Skip to content

IGNITE-16657 [Native Persistence 3.0] Implement partitions chunks merger #972

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
SammyVimes merged 14 commits into from
Aug 4, 2022
Merged

IGNITE-16657 [Native Persistence 3.0] Implement partitions chunks merger #972

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
5d4c166
IGNITE-16657 wip
tkalkirill Aug 2, 2022
e5a72e0
Merge branch 'ai-main' into ignite-16657
tkalkirill Aug 2, 2022
a6feb93
IGNITE-16657 wip
tkalkirill Aug 2, 2022
93075af
IGNITE-16657 wip
tkalkirill Aug 3, 2022
f4d433a
Merge branch 'ai-main' into ignite-16657
tkalkirill Aug 3, 2022
e1c0462
IGNITE-16657 wip
tkalkirill Aug 3, 2022
9b954f0
IGNITE-16657 wip
tkalkirill Aug 3, 2022
8bd2430
IGNITE-16657 wip
tkalkirill Aug 3, 2022
8d81875
Merge branch 'ai-main' into ignite-16657
tkalkirill Aug 3, 2022
f48eebb
IGNITE-16657 wip
tkalkirill Aug 3, 2022
1c00eae
IGNITE-16657 wip
tkalkirill Aug 3, 2022
165e1f8
IGNITE-16657 wip
tkalkirill Aug 3, 2022
d548d31
Merge branch 'ai-main' into ignite-16657
tkalkirill Aug 4, 2022
183991e
IGNITE-16657 after review #1.0
tkalkirill Aug 4, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion ...ite/internal/pagememory/configuration/schema/PageMemoryCheckpointConfigurationSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,12 @@ public class PageMemoryCheckpointConfigurationSchema {
/** Number of checkpoint threads. */
@Range(min = 1)
@Value(hasDefault = true)
public int threads = 4;
public int checkpointThreads = 4;

/** Number of threads to compact delta files. */
@Range(min = 1)
@Value(hasDefault = true)
public int compactionThreads = 4;

/** Timeout for checkpoint read lock acquisition in milliseconds. */
@Range(min = 0)
Expand Down
28 changes: 27 additions & 1 deletion .../java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import org.apache.ignite.internal.pagememory.persistence.PartitionMetaManager;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
import org.apache.ignite.internal.pagememory.persistence.checkpoint.CheckpointDirtyPages.CheckpointDirtyPagesView;
import org.apache.ignite.internal.pagememory.persistence.compaction.Compactor;
import org.apache.ignite.internal.pagememory.persistence.store.DeltaFilePageStoreIo;
import org.apache.ignite.internal.pagememory.persistence.store.FilePageStore;
import org.apache.ignite.internal.pagememory.persistence.store.FilePageStoreManager;
Expand Down Expand Up @@ -76,6 +77,9 @@ public class CheckpointManager {
/** File page store manager. */
private final FilePageStoreManager filePageStoreManager;

/** Delta file compactor. */
private final Compactor compactor;

/**
* Constructor.
*
Expand Down Expand Up @@ -133,6 +137,15 @@ public CheckpointManager(
pageSize
);

compactor = new Compactor(
Loggers.forClass(Compactor.class),
igniteInstanceName,
workerListener,
checkpointConfig.compactionThreads(),
filePageStoreManager,
pageSize
);

checkpointer = new Checkpointer(
Loggers.forClass(Checkpoint.class),
igniteInstanceName,
Expand All @@ -141,6 +154,7 @@ public CheckpointManager(
checkpointWorkflow,
checkpointPagesWriterFactory,
filePageStoreManager,
compactor,
checkpointConfig
);

Expand All @@ -162,6 +176,8 @@ public void start() {
checkpointer.start();

checkpointTimeoutLock.start();

compactor.start();
}

/**
Expand All @@ -171,7 +187,8 @@ public void stop() throws Exception {
IgniteUtils.closeAll(
checkpointTimeoutLock::stop,
checkpointer::stop,
checkpointWorkflow::stop
checkpointWorkflow::stop,
compactor::stop
);
}

Expand Down Expand Up @@ -285,4 +302,13 @@ static int[] pageIndexesForDeltaFilePageStore(CheckpointDirtyPagesView partition

return pageIndexes;
}

/**
* Adds the number of delta files to compact.
*
* @param count Number of delta files.
*/
public void addDeltaFileCountForCompaction(int count) {
compactor.addDeltaFiles(count);
}
}
12 changes: 6 additions & 6 deletions ...a/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,8 +75,8 @@ public class CheckpointPagesWriter implements Runnable {
/** Future which should be finished when all pages would be written. */
private final CompletableFuture<?> doneFut;

/** Some action which will be executed every time before page will be written. */
private final Runnable beforePageWrite;
/** Update heartbeat callback. */
private final Runnable updateHeartbeat;

/** Thread local with buffers for the checkpoint threads. Each buffer represent one page for durable memory. */
private final ThreadLocal<ByteBuffer> threadBuf;
Expand All @@ -103,7 +103,7 @@ public class CheckpointPagesWriter implements Runnable {
* @param writePageIds Queue of dirty page IDs to write.
* @param updatedPartitions Updated partitions.
* @param doneFut Done future.
* @param beforePageWrite Action to be performed before every page write.
* @param updateHeartbeat Update heartbeat callback.
* @param log Logger.
* @param threadBuf Thread local byte buffer.
* @param checkpointProgress Checkpoint progress.
Expand All @@ -118,7 +118,7 @@ public class CheckpointPagesWriter implements Runnable {
IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePageIds,
ConcurrentMap<GroupPartitionId, LongAdder> updatedPartitions,
CompletableFuture<?> doneFut,
Runnable beforePageWrite,
Runnable updateHeartbeat,
ThreadLocal<ByteBuffer> threadBuf,
CheckpointProgressImpl checkpointProgress,
WriteDirtyPage pageWriter,
Expand All @@ -131,7 +131,7 @@ public class CheckpointPagesWriter implements Runnable {
this.writePageIds = writePageIds;
this.updatedPartitions = updatedPartitions;
this.doneFut = doneFut;
this.beforePageWrite = beforePageWrite;
this.updateHeartbeat = updateHeartbeat;
this.threadBuf = threadBuf;
this.checkpointProgress = checkpointProgress;
this.pageWriter = pageWriter;
Expand Down Expand Up @@ -183,7 +183,7 @@ private IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> writePa
AtomicBoolean writeMetaPage = new AtomicBoolean();

while (!shutdownNow.getAsBoolean() && writePageIds.next(queueResult)) {
beforePageWrite.run();
updateHeartbeat.run();

FullPageId fullId = queueResult.getValue();

Expand Down
6 changes: 3 additions & 3 deletions ...pache/ignite/internal/pagememory/persistence/checkpoint/CheckpointPagesWriterFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ public class CheckpointPagesWriterFactory {
* @param dirtyPageIdQueue Checkpoint dirty page ID queue to write.
* @param updatedPartitions Updated partitions.
* @param doneWriteFut Write done future.
* @param beforePageWrite Before page write callback.
* @param updateHeartbeat Update heartbeat callback.
* @param checkpointProgress Current checkpoint data.
* @param shutdownNow Checker of stop operation.
*/
Expand All @@ -100,7 +100,7 @@ CheckpointPagesWriter build(
IgniteConcurrentMultiPairQueue<PersistentPageMemory, FullPageId> dirtyPageIdQueue,
ConcurrentMap<GroupPartitionId, LongAdder> updatedPartitions,
CompletableFuture<?> doneWriteFut,
Runnable beforePageWrite,
Runnable updateHeartbeat,
CheckpointProgressImpl checkpointProgress,
// TODO: IGNITE-16993 Consider a lock replacement
BooleanSupplier shutdownNow
Expand All @@ -111,7 +111,7 @@ CheckpointPagesWriter build(
dirtyPageIdQueue,
updatedPartitions,
doneWriteFut,
beforePageWrite,
updateHeartbeat,
threadBuf,
checkpointProgress,
dirtyPageWriter,
Expand Down
19 changes: 11 additions & 8 deletions .../main/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/Checkpointer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,6 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.LongAdder;
Expand All @@ -48,6 +47,7 @@
import org.apache.ignite.internal.pagememory.configuration.schema.PageMemoryCheckpointView;
import org.apache.ignite.internal.pagememory.persistence.GroupPartitionId;
import org.apache.ignite.internal.pagememory.persistence.PersistentPageMemory;
import org.apache.ignite.internal.pagememory.persistence.compaction.Compactor;
import org.apache.ignite.internal.pagememory.persistence.store.DeltaFilePageStoreIo;
import org.apache.ignite.internal.pagememory.persistence.store.FilePageStore;
import org.apache.ignite.internal.pagememory.persistence.store.FilePageStoreManager;
Expand Down Expand Up @@ -139,6 +139,9 @@ public class Checkpointer extends IgniteWorker {
/** File page store manager. */
private final FilePageStoreManager filePageStoreManager;

/** Delta file compactor. */
private final Compactor compactor;

/**
* Constructor.
*
Expand All @@ -149,6 +152,7 @@ public class Checkpointer extends IgniteWorker {
* @param checkpointWorkFlow Implementation of checkpoint.
* @param factory Page writer factory.
* @param filePageStoreManager File page store manager.
* @param compactor Delta file compactor.
* @param checkpointConfig Checkpoint configuration.
*/
Checkpointer(
Expand All @@ -159,6 +163,7 @@ public class Checkpointer extends IgniteWorker {
CheckpointWorkflow checkpointWorkFlow,
CheckpointPagesWriterFactory factory,
FilePageStoreManager filePageStoreManager,
Compactor compactor,
PageMemoryCheckpointConfiguration checkpointConfig
) {
super(log, igniteInstanceName, "checkpoint-thread", workerListener);
Expand All @@ -168,10 +173,11 @@ public class Checkpointer extends IgniteWorker {
this.checkpointWorkflow = checkpointWorkFlow;
this.checkpointPagesWriterFactory = factory;
this.filePageStoreManager = filePageStoreManager;
this.compactor = compactor;

scheduledCheckpointProgress = new CheckpointProgressImpl(MILLISECONDS.toNanos(nextCheckpointInterval()));

int checkpointWritePageThreads = checkpointConfig.threads().value();
int checkpointWritePageThreads = checkpointConfig.checkpointThreads().value();

if (checkpointWritePageThreads > 1) {
checkpointWritePagesPool = new ThreadPoolExecutor(
Expand Down Expand Up @@ -405,12 +411,7 @@ boolean writePages(
if (pageWritePool == null) {
write.run();
} else {
try {
pageWritePool.execute(write);
} catch (RejectedExecutionException ignore) {
// Run the task synchronously.
write.run();
}
pageWritePool.execute(write);
}
}

Expand All @@ -431,6 +432,8 @@ boolean writePages(

syncUpdatedPageStores(updatedPartitions);

compactor.addDeltaFiles(updatedPartitions.size());

if (shutdownNow.getAsBoolean()) {
currentCheckpointProgress.fail(new NodeStoppingException("Node is stopping."));

Expand Down
Loading