sashapolo commented on code in PR #7610: URL: https://github.com/apache/ignite-3/pull/7610#discussion_r2846997217
########## modules/raft/src/main/java/org/apache/ignite/internal/raft/storage/segstore/RaftLogGarbageCollector.java: ########## @@ -0,0 +1,317 @@ +/* + * 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 java.lang.Math.toIntExact; +import static java.nio.file.StandardOpenOption.CREATE_NEW; +import static java.nio.file.StandardOpenOption.WRITE; +import static org.apache.ignite.internal.raft.storage.segstore.IndexFileManager.indexFileProperties; +import static org.apache.ignite.internal.raft.storage.segstore.SegmentFileManager.HEADER_RECORD; +import static org.apache.ignite.internal.raft.storage.segstore.SegmentPayload.CRC_SIZE_BYTES; +import static org.apache.ignite.internal.raft.storage.segstore.SegmentPayloadParser.endOfSegmentReached; +import static org.apache.ignite.internal.raft.storage.segstore.SegmentPayloadParser.validateSegmentFileHeader; +import static org.apache.ignite.internal.util.IgniteUtils.atomicMoveFile; + +import it.unimi.dsi.fastutil.longs.Long2ObjectOpenHashMap; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.channels.WritableByteChannel; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Iterator; +import java.util.concurrent.atomic.AtomicLong; +import java.util.stream.Stream; +import org.apache.ignite.internal.close.ManuallyCloseable; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.raft.storage.segstore.GroupInfoProvider.GroupInfo; +import org.apache.ignite.internal.raft.util.VarlenEncoder; +import org.apache.ignite.raft.jraft.storage.LogStorage; +import org.jetbrains.annotations.VisibleForTesting; + +/** + * Garbage Collector for Raft log segment files. + * + * <p>The garbage collector performs compaction of segment files by removing truncated log entries and creating new generations + * of segment files. This process reclaims disk space occupied by log entries that have been truncated via + * {@link LogStorage#truncatePrefix} or {@link LogStorage#truncateSuffix} operations. + * + * <h2>Compaction Process</h2> + * When a segment file is selected for compaction, the GC: + * <ol> + * <li>Copies non-truncated entries to a new segment file with an incremented generation number</li> + * <li>Creates a new index file for the new generation</li> + * <li>Atomically replaces the old segment file with the new one</li> + * <li>Deletes the old segment file and its index file</li> + * </ol> + */ +class RaftLogGarbageCollector { + private static final IgniteLogger LOG = Loggers.forClass(RaftLogGarbageCollector.class); + + private static final String TMP_FILE_SUFFIX = ".tmp"; + + private final Path segmentFilesDir; + + private final IndexFileManager indexFileManager; + + private final GroupInfoProvider groupInfoProvider; + + private final AtomicLong logSize = new AtomicLong(); + + RaftLogGarbageCollector( + Path segmentFilesDir, + IndexFileManager indexFileManager, + GroupInfoProvider groupInfoProvider + ) { + this.segmentFilesDir = segmentFilesDir; + this.indexFileManager = indexFileManager; + this.groupInfoProvider = groupInfoProvider; + } + + void cleanupLeftoverFiles() throws IOException { + FileProperties prevFileProperties = null; + + try (Stream<Path> segmentFiles = Files.list(segmentFilesDir)) { + Iterator<Path> it = segmentFiles.sorted().iterator(); + + while (it.hasNext()) { + Path segmentFile = it.next(); + + if (segmentFile.getFileName().toString().endsWith(TMP_FILE_SUFFIX)) { + LOG.info("Deleting temporary segment file: {}.", segmentFile); + + Files.delete(segmentFile); + } else { + FileProperties fileProperties = SegmentFile.fileProperties(segmentFile); + + if (prevFileProperties != null && prevFileProperties.ordinal() == fileProperties.ordinal()) { + Path prevPath = segmentFilesDir.resolve(SegmentFile.fileName(prevFileProperties)); + + LOG.info("Deleting segment file {} because it has a higher generation version.", prevPath); + + Files.delete(prevPath); + } + + prevFileProperties = fileProperties; + } + } + } + + // Do the same routine but for index files. + prevFileProperties = null; + + try (Stream<Path> indexFiles = Files.list(indexFileManager.indexFilesDir())) { + Iterator<Path> it = indexFiles.sorted().iterator(); + + while (it.hasNext()) { + Path indexFile = it.next(); + + FileProperties fileProperties = indexFileProperties(indexFile); + + // Temporary index fils are not created by the GC, they are created by the index manager and are cleaned up by it. Review Comment: This literally happens on the next line, am I missing something? -- 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]
