You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sashapolo (via GitHub)" <gi...@apache.org> on 2026/03/26 15:07:17 UTC
[PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
sashapolo opened a new pull request, #7880:
URL: https://github.com/apache/ignite-3/pull/7880
https://issues.apache.org/jira/browse/IGNITE-26295
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020245246
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -137,29 +207,43 @@ void cleanupLeftoverFiles() throws IOException {
}
@VisibleForTesting
- void runCompaction(SegmentFile segmentFile) throws IOException {
- LOG.info("Compacting segment file [path = {}].", segmentFile.path());
+ long logSizeBytes() {
+ return logSizeBytes.get();
+ }
+
+ @VisibleForTesting
+ void runCompaction(FileProperties segmentFileProperties) throws IOException {
+ Path segmentFilePath = segmentFilesDir.resolve(SegmentFile.fileName(segmentFileProperties));
+
+ LOG.info("Compacting segment file [path = {}].", segmentFilePath);
+ // TODO: Skip non-compactible files, see https://issues.apache.org/jira/browse/IGNITE-28417.
Long2ObjectMap<IndexFileMeta> segmentFileDescription
- = indexFileManager.describeSegmentFile(segmentFile.fileProperties().ordinal());
+ = indexFileManager.describeSegmentFile(segmentFileProperties.ordinal());
boolean canRemoveSegmentFile = segmentFileDescription.isEmpty();
- Path indexFilePath = indexFileManager.indexFilePath(segmentFile.fileProperties());
+ Path indexFilePath = indexFileManager.indexFilePath(segmentFileProperties);
long logSizeDelta;
if (canRemoveSegmentFile) {
- indexFileManager.onIndexFileRemoved(segmentFile.fileProperties());
+ indexFileManager.onIndexFileRemoved(segmentFileProperties);
- logSizeDelta = Files.size(segmentFile.path()) + Files.size(indexFilePath);
+ logSizeDelta = Files.size(segmentFilePath) + Files.size(indexFilePath);
} else {
- logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ SegmentFile segmentFile = SegmentFile.openExisting(segmentFilePath, false);
+
+ try {
+ logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ } finally {
+ segmentFile.close();
+ }
}
// Remove the previous generation of the segment file and its index. This is safe to do, because we rely on the file system
// guarantees that other threads reading from the segment file will still be able to do that even if the file is deleted.
- Files.delete(segmentFile.path());
+ Files.delete(segmentFilePath);
Review Comment:
AFAIK on Windows this scenario is not possible at all, removing an open file should throw an exception unless you use a specific flag
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo merged PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020354216
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -71,11 +86,66 @@ class RaftLogGarbageCollector {
private final IndexFileManager indexFileManager;
+ private final long softLimitBytes;
+
+ private final SegmentFileCompactionStrategy strategy;
+
+ private final FailureProcessor failureProcessor;
+
private final AtomicLong logSizeBytes = new AtomicLong();
- RaftLogGarbageCollector(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ private final Thread gcThread;
+
+ RaftLogGarbageCollector(
+ String nodeName,
+ Path segmentFilesDir,
+ IndexFileManager indexFileManager,
+ long softLimitBytes,
+ SegmentFileCompactionStrategy strategy,
+ FailureProcessor failureProcessor
+ ) {
this.segmentFilesDir = segmentFilesDir;
this.indexFileManager = indexFileManager;
+ this.softLimitBytes = softLimitBytes;
+ this.strategy = strategy;
+ this.failureProcessor = failureProcessor;
+
+ gcThread = new IgniteThread(nodeName, "segstore-gc", new GcTask());
+ }
+
+ void start() throws IOException {
+ initLogSizeFromDisk();
+
+ gcThread.start();
+ }
+
+ void stop() {
+ gcThread.interrupt();
Review Comment:
I still don't understand the problem, sorry. When interrupting a thread that is doing something with a channel, we get a `ClosedByInterruptException` and we try to handle this exception. What are the other implications?
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020296876
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/MostGarbageFirstCompactionStrategy.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.raft.storage.segstore;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+
+/**
+ * Compaction strategy that prioritizes files with the most dead entries, fully-deletable files first.
+ */
+class MostGarbageFirstCompactionStrategy implements SegmentFileCompactionStrategy {
+ private final Path segmentFilesDir;
+
+ private final IndexFileManager indexFileManager;
+
+ MostGarbageFirstCompactionStrategy(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ this.segmentFilesDir = segmentFilesDir;
+ this.indexFileManager = indexFileManager;
+ }
+
+ @Override
+ public Stream<FileProperties> selectCandidates() throws IOException {
+ var scores = new Object2LongOpenHashMap<FileProperties>();
+
+ Comparator<FileProperties> comparator =
+ Comparator.<FileProperties>comparingLong(props -> scores.computeIfAbsent(props, this::score))
+ .thenComparing(Comparator.naturalOrder());
+
+ return Files.list(segmentFilesDir)
+ .filter(p -> !p.getFileName().toString().endsWith(".tmp"))
+ .map(SegmentFile::fileProperties)
+ .filter(props -> Files.exists(indexFileManager.indexFilePath(props)))
+ .sorted(comparator);
+ }
+
+ private long score(FileProperties props) {
+ Long2ObjectMap<IndexFileMeta> description = indexFileManager.describeSegmentFile(props.ordinal());
+
+ if (description.isEmpty()) {
+ return -1; // fully deletable — highest priority
Review Comment:
I see, didn't connect the dots
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020333210
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogCheckpointer.java:
##########
@@ -62,14 +63,18 @@ class RaftLogCheckpointer {
private final FailureProcessor failureProcessor;
+ private final LongConsumer beforeIndexFileCreated;
+
RaftLogCheckpointer(
String nodeName,
IndexFileManager indexFileManager,
FailureProcessor failureProcessor,
- int maxQueueSize
+ int maxQueueSize,
+ LongConsumer beforeIndexFileCreated
Review Comment:
sure, will do
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020310259
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -137,29 +207,43 @@ void cleanupLeftoverFiles() throws IOException {
}
@VisibleForTesting
- void runCompaction(SegmentFile segmentFile) throws IOException {
- LOG.info("Compacting segment file [path = {}].", segmentFile.path());
+ long logSizeBytes() {
+ return logSizeBytes.get();
+ }
+
+ @VisibleForTesting
+ void runCompaction(FileProperties segmentFileProperties) throws IOException {
+ Path segmentFilePath = segmentFilesDir.resolve(SegmentFile.fileName(segmentFileProperties));
+
+ LOG.info("Compacting segment file [path = {}].", segmentFilePath);
+ // TODO: Skip non-compactible files, see https://issues.apache.org/jira/browse/IGNITE-28417.
Long2ObjectMap<IndexFileMeta> segmentFileDescription
- = indexFileManager.describeSegmentFile(segmentFile.fileProperties().ordinal());
+ = indexFileManager.describeSegmentFile(segmentFileProperties.ordinal());
boolean canRemoveSegmentFile = segmentFileDescription.isEmpty();
- Path indexFilePath = indexFileManager.indexFilePath(segmentFile.fileProperties());
+ Path indexFilePath = indexFileManager.indexFilePath(segmentFileProperties);
long logSizeDelta;
if (canRemoveSegmentFile) {
- indexFileManager.onIndexFileRemoved(segmentFile.fileProperties());
+ indexFileManager.onIndexFileRemoved(segmentFileProperties);
- logSizeDelta = Files.size(segmentFile.path()) + Files.size(indexFilePath);
+ logSizeDelta = Files.size(segmentFilePath) + Files.size(indexFilePath);
} else {
- logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ SegmentFile segmentFile = SegmentFile.openExisting(segmentFilePath, false);
+
+ try {
+ logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ } finally {
+ segmentFile.close();
+ }
}
// Remove the previous generation of the segment file and its index. This is safe to do, because we rely on the file system
// guarantees that other threads reading from the segment file will still be able to do that even if the file is deleted.
- Files.delete(segmentFile.path());
+ Files.delete(segmentFilePath);
Review Comment:
Does that mean that this operation can fail with an exception? I wonder how we work on Windows in general, I bet there are many places where we don't handle it in a right way
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3026929590
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
+ Path indexFilesDir = indexFileManager.indexFilesDir();
+
+ try (Stream<Path> files = Stream.concat(Files.list(segmentFilesDir), Files.list(indexFilesDir))) {
+ long logSizeOnDisk = files
+ .mapToLong(path -> {
+ try {
+ return Files.size(path);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ })
+ .sum();
+
+ logSizeBytes.addAndGet(logSizeOnDisk);
+ }
+ }
+
+ private class GcTask implements Runnable {
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ runGcCycle();
+ } catch (ClosedByInterruptException e) {
+ return;
+ } catch (IOException e) {
+ failureProcessor.process(new FailureContext(CRITICAL_ERROR, e));
+ }
+
+ LockSupport.park();
+ }
+ }
+
+ private void runGcCycle() throws IOException {
+ if (logSizeBytes.get() < softLimitBytes) {
+ return;
+ }
+
+ LOG.info("Starting Log Storage GC cycle.");
Review Comment:
https://issues.apache.org/jira/browse/IGNITE-28432
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020077046
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -71,11 +86,66 @@ class RaftLogGarbageCollector {
private final IndexFileManager indexFileManager;
+ private final long softLimitBytes;
+
+ private final SegmentFileCompactionStrategy strategy;
+
+ private final FailureProcessor failureProcessor;
+
private final AtomicLong logSizeBytes = new AtomicLong();
- RaftLogGarbageCollector(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ private final Thread gcThread;
+
+ RaftLogGarbageCollector(
+ String nodeName,
+ Path segmentFilesDir,
+ IndexFileManager indexFileManager,
+ long softLimitBytes,
+ SegmentFileCompactionStrategy strategy,
+ FailureProcessor failureProcessor
+ ) {
this.segmentFilesDir = segmentFilesDir;
this.indexFileManager = indexFileManager;
+ this.softLimitBytes = softLimitBytes;
+ this.strategy = strategy;
+ this.failureProcessor = failureProcessor;
+
+ gcThread = new IgniteThread(nodeName, "segstore-gc", new GcTask());
+ }
+
+ void start() throws IOException {
+ initLogSizeFromDisk();
+
+ gcThread.start();
+ }
+
+ void stop() {
+ gcThread.interrupt();
Review Comment:
When exactly do we stop this component? Are you sure that you won't interrupt any file channel that's still in use?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -137,29 +207,43 @@ void cleanupLeftoverFiles() throws IOException {
}
@VisibleForTesting
- void runCompaction(SegmentFile segmentFile) throws IOException {
- LOG.info("Compacting segment file [path = {}].", segmentFile.path());
+ long logSizeBytes() {
+ return logSizeBytes.get();
+ }
+
+ @VisibleForTesting
+ void runCompaction(FileProperties segmentFileProperties) throws IOException {
+ Path segmentFilePath = segmentFilesDir.resolve(SegmentFile.fileName(segmentFileProperties));
+
+ LOG.info("Compacting segment file [path = {}].", segmentFilePath);
+ // TODO: Skip non-compactible files, see https://issues.apache.org/jira/browse/IGNITE-28417.
Long2ObjectMap<IndexFileMeta> segmentFileDescription
- = indexFileManager.describeSegmentFile(segmentFile.fileProperties().ordinal());
+ = indexFileManager.describeSegmentFile(segmentFileProperties.ordinal());
boolean canRemoveSegmentFile = segmentFileDescription.isEmpty();
- Path indexFilePath = indexFileManager.indexFilePath(segmentFile.fileProperties());
+ Path indexFilePath = indexFileManager.indexFilePath(segmentFileProperties);
long logSizeDelta;
if (canRemoveSegmentFile) {
- indexFileManager.onIndexFileRemoved(segmentFile.fileProperties());
+ indexFileManager.onIndexFileRemoved(segmentFileProperties);
- logSizeDelta = Files.size(segmentFile.path()) + Files.size(indexFilePath);
+ logSizeDelta = Files.size(segmentFilePath) + Files.size(indexFilePath);
} else {
- logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ SegmentFile segmentFile = SegmentFile.openExisting(segmentFilePath, false);
+
+ try {
+ logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ } finally {
+ segmentFile.close();
+ }
}
// Remove the previous generation of the segment file and its index. This is safe to do, because we rely on the file system
// guarantees that other threads reading from the segment file will still be able to do that even if the file is deleted.
- Files.delete(segmentFile.path());
+ Files.delete(segmentFilePath);
Review Comment:
> because we rely on the file system guarantees that other threads reading from the segment file will still be able to do that even if the file is deleted.
That's not true in general. NTFS would not do that, as far as I know
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/IndexFileManager.java:
##########
@@ -504,6 +505,21 @@ private static byte[] payload(SegmentInfo segmentInfo) {
return payloadBuffer.array();
}
+ /**
+ * Computes the size in bytes that the index file for the given {@code indexMemTable} will occupy on disk.
+ */
+ static long computeIndexFileSize(ReadModeIndexMemTable indexMemTable) {
+ long total = headerSize(indexMemTable.numGroups());
+
+ Iterator<Entry<Long, SegmentInfo>> it = indexMemTable.iterator();
+
+ while (it.hasNext()) {
+ total += payloadSize(it.next().getValue());
+ }
+
+ return total;
Review Comment:
Can we make `ReadModeIndexMemTable` implement `Iterable<Entry<Long, SegmentInfo>>`? This way you would just use `foreach`
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
+ Path indexFilesDir = indexFileManager.indexFilesDir();
+
+ try (Stream<Path> files = Stream.concat(Files.list(segmentFilesDir), Files.list(indexFilesDir))) {
+ long logSizeOnDisk = files
+ .mapToLong(path -> {
+ try {
+ return Files.size(path);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ })
+ .sum();
+
+ logSizeBytes.addAndGet(logSizeOnDisk);
+ }
+ }
+
+ private class GcTask implements Runnable {
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ runGcCycle();
+ } catch (ClosedByInterruptException e) {
+ return;
+ } catch (IOException e) {
+ failureProcessor.process(new FailureContext(CRITICAL_ERROR, e));
+ }
+
+ LockSupport.park();
+ }
+ }
+
+ private void runGcCycle() throws IOException {
+ if (logSizeBytes.get() < softLimitBytes) {
+ return;
+ }
+
+ LOG.info("Starting Log Storage GC cycle.");
+
+ try (Stream<FileProperties> candidates = strategy.selectCandidates()) {
+ Iterator<FileProperties> it = candidates.iterator();
+
+ do {
+ if (!it.hasNext()) {
+ LOG.warn(
+ "Log size is above the soft limit but there are no files to compact "
+ + "[current log size = {} bytes, soft limit = {} bytes].",
Review Comment:
Please follow the guidelines, names here should include no spaces
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogCheckpointer.java:
##########
@@ -62,14 +63,18 @@ class RaftLogCheckpointer {
private final FailureProcessor failureProcessor;
+ private final LongConsumer beforeIndexFileCreated;
+
RaftLogCheckpointer(
String nodeName,
IndexFileManager indexFileManager,
FailureProcessor failureProcessor,
- int maxQueueSize
+ int maxQueueSize,
+ LongConsumer beforeIndexFileCreated
Review Comment:
Were there any other ways of doing it? Currently I don't even understand the purpose of this closure, such patterns make code more difficult in my opinion. Maybe we should use explicit interface instead of a generic `LongConsumer`, what do you think?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/MostGarbageFirstCompactionStrategy.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.raft.storage.segstore;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+
+/**
+ * Compaction strategy that prioritizes files with the most dead entries, fully-deletable files first.
+ */
+class MostGarbageFirstCompactionStrategy implements SegmentFileCompactionStrategy {
+ private final Path segmentFilesDir;
+
+ private final IndexFileManager indexFileManager;
+
+ MostGarbageFirstCompactionStrategy(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ this.segmentFilesDir = segmentFilesDir;
+ this.indexFileManager = indexFileManager;
+ }
+
+ @Override
+ public Stream<FileProperties> selectCandidates() throws IOException {
+ var scores = new Object2LongOpenHashMap<FileProperties>();
+
+ Comparator<FileProperties> comparator =
+ Comparator.<FileProperties>comparingLong(props -> scores.computeIfAbsent(props, this::score))
+ .thenComparing(Comparator.naturalOrder());
+
+ return Files.list(segmentFilesDir)
+ .filter(p -> !p.getFileName().toString().endsWith(".tmp"))
+ .map(SegmentFile::fileProperties)
+ .filter(props -> Files.exists(indexFileManager.indexFilePath(props)))
+ .sorted(comparator);
+ }
+
+ private long score(FileProperties props) {
+ Long2ObjectMap<IndexFileMeta> description = indexFileManager.describeSegmentFile(props.ordinal());
+
+ if (description.isEmpty()) {
+ return -1; // fully deletable — highest priority
Review Comment:
```suggestion
return -1; // Fully deletable — highest priority.
```
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
+ Path indexFilesDir = indexFileManager.indexFilesDir();
+
+ try (Stream<Path> files = Stream.concat(Files.list(segmentFilesDir), Files.list(indexFilesDir))) {
+ long logSizeOnDisk = files
+ .mapToLong(path -> {
+ try {
+ return Files.size(path);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ })
+ .sum();
+
+ logSizeBytes.addAndGet(logSizeOnDisk);
+ }
+ }
+
+ private class GcTask implements Runnable {
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ runGcCycle();
+ } catch (ClosedByInterruptException e) {
+ return;
+ } catch (IOException e) {
+ failureProcessor.process(new FailureContext(CRITICAL_ERROR, e));
+ }
+
+ LockSupport.park();
+ }
+ }
+
+ private void runGcCycle() throws IOException {
+ if (logSizeBytes.get() < softLimitBytes) {
+ return;
+ }
+
+ LOG.info("Starting Log Storage GC cycle.");
Review Comment:
I would also include log storage name. I suspect that metastorage, CMG and diztribution zones will have their own independent threads. Well, maybe we should identify them by a thread name, not by a message, could you please make sure that they're differentiable? Thank you!
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
Review Comment:
We're sure that at this point all `tmp` files are deleted, is that correct?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
+ Path indexFilesDir = indexFileManager.indexFilesDir();
+
+ try (Stream<Path> files = Stream.concat(Files.list(segmentFilesDir), Files.list(indexFilesDir))) {
+ long logSizeOnDisk = files
+ .mapToLong(path -> {
+ try {
+ return Files.size(path);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ })
+ .sum();
+
+ logSizeBytes.addAndGet(logSizeOnDisk);
+ }
+ }
+
+ private class GcTask implements Runnable {
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ runGcCycle();
+ } catch (ClosedByInterruptException e) {
+ return;
+ } catch (IOException e) {
+ failureProcessor.process(new FailureContext(CRITICAL_ERROR, e));
+ }
+
+ LockSupport.park();
+ }
+ }
+
+ private void runGcCycle() throws IOException {
+ if (logSizeBytes.get() < softLimitBytes) {
+ return;
+ }
+
+ LOG.info("Starting Log Storage GC cycle.");
+
+ try (Stream<FileProperties> candidates = strategy.selectCandidates()) {
+ Iterator<FileProperties> it = candidates.iterator();
+
+ do {
+ if (!it.hasNext()) {
+ LOG.warn(
+ "Log size is above the soft limit but there are no files to compact "
+ + "[current log size = {} bytes, soft limit = {} bytes].",
Review Comment:
```suggestion
+ "[currentLogSize={}, softLimit={}]",
```
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/MostGarbageFirstCompactionStrategy.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.raft.storage.segstore;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+
+/**
+ * Compaction strategy that prioritizes files with the most dead entries, fully-deletable files first.
+ */
+class MostGarbageFirstCompactionStrategy implements SegmentFileCompactionStrategy {
+ private final Path segmentFilesDir;
+
+ private final IndexFileManager indexFileManager;
+
+ MostGarbageFirstCompactionStrategy(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ this.segmentFilesDir = segmentFilesDir;
+ this.indexFileManager = indexFileManager;
+ }
+
+ @Override
+ public Stream<FileProperties> selectCandidates() throws IOException {
+ var scores = new Object2LongOpenHashMap<FileProperties>();
+
+ Comparator<FileProperties> comparator =
+ Comparator.<FileProperties>comparingLong(props -> scores.computeIfAbsent(props, this::score))
+ .thenComparing(Comparator.naturalOrder());
+
+ return Files.list(segmentFilesDir)
+ .filter(p -> !p.getFileName().toString().endsWith(".tmp"))
+ .map(SegmentFile::fileProperties)
+ .filter(props -> Files.exists(indexFileManager.indexFilePath(props)))
+ .sorted(comparator);
+ }
+
+ private long score(FileProperties props) {
+ Long2ObjectMap<IndexFileMeta> description = indexFileManager.describeSegmentFile(props.ordinal());
+
+ if (description.isEmpty()) {
+ return -1; // fully deletable — highest priority
Review Comment:
What if we replace it with something like `props.ordinal() - Long.MAX_VALUE`? This way the sorting will provide a natural order from lowest to highest ordinal for all "empty" files, ideally producing a GC process without any gaps in ordinals
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020217377
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -71,11 +86,66 @@ class RaftLogGarbageCollector {
private final IndexFileManager indexFileManager;
+ private final long softLimitBytes;
+
+ private final SegmentFileCompactionStrategy strategy;
+
+ private final FailureProcessor failureProcessor;
+
private final AtomicLong logSizeBytes = new AtomicLong();
- RaftLogGarbageCollector(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ private final Thread gcThread;
+
+ RaftLogGarbageCollector(
+ String nodeName,
+ Path segmentFilesDir,
+ IndexFileManager indexFileManager,
+ long softLimitBytes,
+ SegmentFileCompactionStrategy strategy,
+ FailureProcessor failureProcessor
+ ) {
this.segmentFilesDir = segmentFilesDir;
this.indexFileManager = indexFileManager;
+ this.softLimitBytes = softLimitBytes;
+ this.strategy = strategy;
+ this.failureProcessor = failureProcessor;
+
+ gcThread = new IgniteThread(nodeName, "segstore-gc", new GcTask());
+ }
+
+ void start() throws IOException {
+ initLogSizeFromDisk();
+
+ gcThread.start();
+ }
+
+ void stop() {
+ gcThread.interrupt();
Review Comment:
> Are you sure that you won't interrupt any file channel that's still in use?
Could you please elaborate? Why is that a bad thing?
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020299077
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogCheckpointer.java:
##########
@@ -62,14 +63,18 @@ class RaftLogCheckpointer {
private final FailureProcessor failureProcessor;
+ private final LongConsumer beforeIndexFileCreated;
+
RaftLogCheckpointer(
String nodeName,
IndexFileManager indexFileManager,
FailureProcessor failureProcessor,
- int maxQueueSize
+ int maxQueueSize,
+ LongConsumer beforeIndexFileCreated
Review Comment:
Using custom interface instead of a generic `Consumer` would look nicer
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3015764520
##########
modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/segstore/AbstractCompactionStrategyTest.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.raft.storage.segstore;
+
+import static org.apache.ignite.internal.raft.storage.segstore.IndexFileManager.indexFileName;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
Review Comment:
It is used, what are you talking about?
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3027218949
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -71,11 +86,66 @@ class RaftLogGarbageCollector {
private final IndexFileManager indexFileManager;
+ private final long softLimitBytes;
+
+ private final SegmentFileCompactionStrategy strategy;
+
+ private final FailureProcessor failureProcessor;
+
private final AtomicLong logSizeBytes = new AtomicLong();
- RaftLogGarbageCollector(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ private final Thread gcThread;
+
+ RaftLogGarbageCollector(
+ String nodeName,
+ Path segmentFilesDir,
+ IndexFileManager indexFileManager,
+ long softLimitBytes,
+ SegmentFileCompactionStrategy strategy,
+ FailureProcessor failureProcessor
+ ) {
this.segmentFilesDir = segmentFilesDir;
this.indexFileManager = indexFileManager;
+ this.softLimitBytes = softLimitBytes;
+ this.strategy = strategy;
+ this.failureProcessor = failureProcessor;
+
+ gcThread = new IgniteThread(nodeName, "segstore-gc", new GcTask());
+ }
+
+ void start() throws IOException {
+ initLogSizeFromDisk();
+
+ gcThread.start();
+ }
+
+ void stop() {
+ gcThread.interrupt();
Review Comment:
The channel itself gets closed too
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020215847
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogCheckpointer.java:
##########
@@ -62,14 +63,18 @@ class RaftLogCheckpointer {
private final FailureProcessor failureProcessor;
+ private final LongConsumer beforeIndexFileCreated;
+
RaftLogCheckpointer(
String nodeName,
IndexFileManager indexFileManager,
FailureProcessor failureProcessor,
- int maxQueueSize
+ int maxQueueSize,
+ LongConsumer beforeIndexFileCreated
Review Comment:
This closure is needed to notify the GC that an index file is going to be saved on disk. I don't like the current approach much either, do you have any other suggestions?
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020218359
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
Review Comment:
yes
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020357801
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -137,29 +207,43 @@ void cleanupLeftoverFiles() throws IOException {
}
@VisibleForTesting
- void runCompaction(SegmentFile segmentFile) throws IOException {
- LOG.info("Compacting segment file [path = {}].", segmentFile.path());
+ long logSizeBytes() {
+ return logSizeBytes.get();
+ }
+
+ @VisibleForTesting
+ void runCompaction(FileProperties segmentFileProperties) throws IOException {
+ Path segmentFilePath = segmentFilesDir.resolve(SegmentFile.fileName(segmentFileProperties));
+
+ LOG.info("Compacting segment file [path = {}].", segmentFilePath);
+ // TODO: Skip non-compactible files, see https://issues.apache.org/jira/browse/IGNITE-28417.
Long2ObjectMap<IndexFileMeta> segmentFileDescription
- = indexFileManager.describeSegmentFile(segmentFile.fileProperties().ordinal());
+ = indexFileManager.describeSegmentFile(segmentFileProperties.ordinal());
boolean canRemoveSegmentFile = segmentFileDescription.isEmpty();
- Path indexFilePath = indexFileManager.indexFilePath(segmentFile.fileProperties());
+ Path indexFilePath = indexFileManager.indexFilePath(segmentFileProperties);
long logSizeDelta;
if (canRemoveSegmentFile) {
- indexFileManager.onIndexFileRemoved(segmentFile.fileProperties());
+ indexFileManager.onIndexFileRemoved(segmentFileProperties);
- logSizeDelta = Files.size(segmentFile.path()) + Files.size(indexFilePath);
+ logSizeDelta = Files.size(segmentFilePath) + Files.size(indexFilePath);
} else {
- logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ SegmentFile segmentFile = SegmentFile.openExisting(segmentFilePath, false);
+
+ try {
+ logSizeDelta = compactSegmentFile(segmentFile, indexFilePath, segmentFileDescription);
+ } finally {
+ segmentFile.close();
+ }
}
// Remove the previous generation of the segment file and its index. This is safe to do, because we rely on the file system
// guarantees that other threads reading from the segment file will still be able to do that even if the file is deleted.
- Files.delete(segmentFile.path());
+ Files.delete(segmentFilePath);
Review Comment:
> Does that mean that this operation can fail with an exception?
yes, and this is a bug in this case, I'll try to test it on a Windows machine (though not with a high priority)
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020255063
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/MostGarbageFirstCompactionStrategy.java:
##########
@@ -0,0 +1,71 @@
+/*
+ * 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.raft.storage.segstore;
+
+import it.unimi.dsi.fastutil.longs.Long2ObjectMap;
+import it.unimi.dsi.fastutil.objects.Object2LongOpenHashMap;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Comparator;
+import java.util.stream.Stream;
+
+/**
+ * Compaction strategy that prioritizes files with the most dead entries, fully-deletable files first.
+ */
+class MostGarbageFirstCompactionStrategy implements SegmentFileCompactionStrategy {
+ private final Path segmentFilesDir;
+
+ private final IndexFileManager indexFileManager;
+
+ MostGarbageFirstCompactionStrategy(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ this.segmentFilesDir = segmentFilesDir;
+ this.indexFileManager = indexFileManager;
+ }
+
+ @Override
+ public Stream<FileProperties> selectCandidates() throws IOException {
+ var scores = new Object2LongOpenHashMap<FileProperties>();
+
+ Comparator<FileProperties> comparator =
+ Comparator.<FileProperties>comparingLong(props -> scores.computeIfAbsent(props, this::score))
+ .thenComparing(Comparator.naturalOrder());
+
+ return Files.list(segmentFilesDir)
+ .filter(p -> !p.getFileName().toString().endsWith(".tmp"))
+ .map(SegmentFile::fileProperties)
+ .filter(props -> Files.exists(indexFileManager.indexFilePath(props)))
+ .sorted(comparator);
+ }
+
+ private long score(FileProperties props) {
+ Long2ObjectMap<IndexFileMeta> description = indexFileManager.describeSegmentFile(props.ordinal());
+
+ if (description.isEmpty()) {
+ return -1; // fully deletable — highest priority
Review Comment:
I already do that but in a less elegant way, see `thenComparing(naturalOrder)`. Should we bother changing 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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "sashapolo (via GitHub)" <gi...@apache.org>.
sashapolo commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020249598
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -286,6 +370,69 @@ private static boolean isLogIndexInRange(long index, IndexFileMeta indexFileMeta
return index >= indexFileMeta.firstLogIndexInclusive() && index < indexFileMeta.lastLogIndexExclusive();
}
+ private void initLogSizeFromDisk() throws IOException {
+ Path indexFilesDir = indexFileManager.indexFilesDir();
+
+ try (Stream<Path> files = Stream.concat(Files.list(segmentFilesDir), Files.list(indexFilesDir))) {
+ long logSizeOnDisk = files
+ .mapToLong(path -> {
+ try {
+ return Files.size(path);
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ })
+ .sum();
+
+ logSizeBytes.addAndGet(logSizeOnDisk);
+ }
+ }
+
+ private class GcTask implements Runnable {
+ @Override
+ public void run() {
+ while (!Thread.currentThread().isInterrupted()) {
+ try {
+ runGcCycle();
+ } catch (ClosedByInterruptException e) {
+ return;
+ } catch (IOException e) {
+ failureProcessor.process(new FailureContext(CRITICAL_ERROR, e));
+ }
+
+ LockSupport.park();
+ }
+ }
+
+ private void runGcCycle() throws IOException {
+ if (logSizeBytes.get() < softLimitBytes) {
+ return;
+ }
+
+ LOG.info("Starting Log Storage GC cycle.");
Review Comment:
Good idea, I'll add it in a separate PR
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3020305635
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java:
##########
@@ -71,11 +86,66 @@ class RaftLogGarbageCollector {
private final IndexFileManager indexFileManager;
+ private final long softLimitBytes;
+
+ private final SegmentFileCompactionStrategy strategy;
+
+ private final FailureProcessor failureProcessor;
+
private final AtomicLong logSizeBytes = new AtomicLong();
- RaftLogGarbageCollector(Path segmentFilesDir, IndexFileManager indexFileManager) {
+ private final Thread gcThread;
+
+ RaftLogGarbageCollector(
+ String nodeName,
+ Path segmentFilesDir,
+ IndexFileManager indexFileManager,
+ long softLimitBytes,
+ SegmentFileCompactionStrategy strategy,
+ FailureProcessor failureProcessor
+ ) {
this.segmentFilesDir = segmentFilesDir;
this.indexFileManager = indexFileManager;
+ this.softLimitBytes = softLimitBytes;
+ this.strategy = strategy;
+ this.failureProcessor = failureProcessor;
+
+ gcThread = new IgniteThread(nodeName, "segstore-gc", new GcTask());
+ }
+
+ void start() throws IOException {
+ initLogSizeFromDisk();
+
+ gcThread.start();
+ }
+
+ void stop() {
+ gcThread.interrupt();
Review Comment:
Interruption of the thread may interrupt the channel that it currently uses, and all other threads that use it won't be able to do so. So, we have to be careful while interrupting any threads that can do any disk IO, just in case
--
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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-26295 Integrate Raft Log GC into the new storage [ignite-3]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #7880:
URL: https://github.com/apache/ignite-3/pull/7880#discussion_r3015720997
##########
modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGcSoftLimitTest.java:
##########
@@ -0,0 +1,225 @@
+/*
+ * 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.raft.storage.segstore;
+
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.randomBytes;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.runRace;
+import static org.awaitility.Awaitility.await;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
+import org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
+import org.apache.ignite.internal.failure.NoOpFailureManager;
+import org.apache.ignite.internal.lang.RunnableX;
+import org.apache.ignite.internal.raft.configuration.LogStorageConfiguration;
+import org.apache.ignite.internal.raft.configuration.RaftConfiguration;
+import org.apache.ignite.internal.testframework.IgniteAbstractTest;
+import org.apache.ignite.raft.jraft.entity.LogEntry;
+import org.apache.ignite.raft.jraft.entity.LogId;
+import org.apache.ignite.raft.jraft.entity.codec.LogEntryEncoder;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.RepeatedTest;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+/**
+ * Tests for {@link RaftLogGarbageCollector} soft-limit behavior.
+ */
+@ExtendWith(ConfigurationExtension.class)
+class RaftLogGcSoftLimitTest extends IgniteAbstractTest {
+ private static final int FILE_SIZE = 200;
+
+ private static final long SMALL_SOFT_LIMIT = 2 * FILE_SIZE;
+
+ private static final long GROUP_ID_1 = 1000;
+
+ private static final long GROUP_ID_2 = 2000;
+
+ private static final int STRIPES = 10;
+
+ private static final String NODE_NAME = "test";
+
+ @InjectConfiguration
+ private RaftConfiguration raftConfiguration;
+
+ @InjectConfiguration(
+ value = "mock.segmentFileSizeBytes=" + FILE_SIZE + ", mock.softLogSizeLimitBytes=" + SMALL_SOFT_LIMIT,
+ validate = false
+ )
+ private LogStorageConfiguration storageConfiguration;
+
+ private SegmentFileManager fileManager;
+
+ private RaftLogGarbageCollector garbageCollector;
+
+ @BeforeEach
+ void setUp() throws IOException {
+ fileManager = new SegmentFileManager(
+ NODE_NAME,
+ workDir,
+ STRIPES,
+ new NoOpFailureManager(),
+ raftConfiguration,
+ storageConfiguration
+ );
+
+ fileManager.start();
+
+ garbageCollector = fileManager.garbageCollector();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (fileManager != null) {
+ fileManager.close();
+ }
+ }
+
+ @Test
+ void testGcTriggeredWhenSoftLimitExceeded() throws Exception {
+ List<byte[]> batches = createRandomData(FILE_SIZE / 4, 10);
+ for (int i = 0; i < batches.size(); i++) {
+ appendBytes(GROUP_ID_1, i, batches.get(i));
+ }
+
+ await().until(() -> indexFiles(fileManager), hasSize(4));
+
+ assertThat(totalLogSizeFromDisk(fileManager), is(greaterThanOrEqualTo(SMALL_SOFT_LIMIT)));
+
+ // Truncate all GROUP_ID_1 entries so every old file becomes fully deletable.
+ fileManager.truncatePrefix(GROUP_ID_1, batches.size() - 1);
+
+ // Write one GROUP_ID_2 batch to force a rollover so the truncation record ends up in a
+ // completed (checkpointable) file.
+ appendBytes(GROUP_ID_2, batches.size() + 1, createRandomData(FILE_SIZE / 4, 1).get(0));
+
+ await().until(() -> garbageCollector.logSizeBytes(), is(lessThanOrEqualTo(SMALL_SOFT_LIMIT)));
+ }
+
+ @RepeatedTest(10)
+ void testConcurrentWritesAndAutoGc() throws Exception {
+ int numBatches = 20;
+ List<byte[]> batches = createRandomData(FILE_SIZE / 8, numBatches);
+
+ // Write initial data from both groups.
+ for (int i = 0; i < numBatches; i++) {
+ appendBytes(GROUP_ID_1, i, batches.get(i));
+ appendBytes(GROUP_ID_2, i, batches.get(i));
+ }
+
+ // Wait for checkpoints so the GC has candidates to compact.
+ await().until(() -> indexFiles(fileManager), hasSize(greaterThanOrEqualTo(1)));
+
+ var writerDone = new AtomicBoolean(false);
+
+ RunnableX writerTask = () -> {
+ try {
+ for (int i = numBatches; i < numBatches * 3; i++) {
+ appendBytes(GROUP_ID_1, i, batches.get(i % numBatches));
+ appendBytes(GROUP_ID_2, i, batches.get(i % numBatches));
+ fileManager.truncatePrefix(GROUP_ID_1, i - 1);
+ }
+ } finally {
+ writerDone.set(true);
+ }
+ };
+
+ RunnableX readerTask = () -> {
+ while (!writerDone.get()) {
+ for (int i = 0; i < numBatches; i++) {
+ int index = i;
+ fileManager.getEntry(GROUP_ID_2, i, bs -> {
+ if (bs != null) {
+ assertThat(bs, is(batches.get(index)));
+ }
Review Comment:
In `readerTask`, missing entries are silently ignored (`if (bs != null) { ... }`). Since `GROUP_ID_2` entries are never truncated in this test, `bs` should always be non-null; allowing null would let data-loss regressions slip through. Consider asserting non-null for the expected indices (or explicitly documenting why null is acceptable here).
```suggestion
assertThat("Log entry for GROUP_ID_2 at index " + index + " must not be null",
bs, is(batches.get(index)));
```
##########
modules/raft/src/test/java/org/apache/ignite/internal/raft/storage/segstore/AbstractCompactionStrategyTest.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * 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.raft.storage.segstore;
+
+import static org.apache.ignite.internal.raft.storage.segstore.IndexFileManager.indexFileName;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.contains;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
Review Comment:
Unused import `java.util.List` in this test file. If the build enforces Checkstyle/Spotless for unused imports, this will fail CI; please remove 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-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org