atomix-storage/src/main/java/io/atomix/storage/journal/JournalSegmentReader.java

   1 /*
   2  * Copyright (c) 2024 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  * http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16 package io.atomix.storage.journal;
  17
  18 import static com.google.common.base.Verify.verify;
  19 import static java.util.Objects.requireNonNull;
  20
  21 import io.netty.buffer.ByteBuf;
  22 import io.netty.buffer.Unpooled;
  23 import org.eclipse.jdt.annotation.Nullable;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 final class JournalSegmentReader {
  28     private static final Logger LOG = LoggerFactory.getLogger(JournalSegmentReader.class);
  29
  30     private final JournalSegment segment;
  31     private final FileReader fileReader;
  32     private final int maxSegmentSize;
  33     private final int maxEntrySize;
  34
  35     private int position;
  36
  37     JournalSegmentReader(final JournalSegment segment, final FileReader fileReader, final int maxEntrySize) {
  38         this.segment = requireNonNull(segment);
  39         this.fileReader = requireNonNull(fileReader);
  40         maxSegmentSize = segment.file().maxSize();
  41         this.maxEntrySize = maxEntrySize;
  42     }
  43
  44     /**
  45      * Return the current position.
  46      *
  47      * @return current position.
  48      */
  49     int position() {
  50         return position;
  51     }
  52
  53     /**
  54      * Set the file position.
  55      *
  56      * @param position new position
  57      */
  58     void setPosition(final int position) {
  59         verify(position >= JournalSegmentDescriptor.BYTES && position < maxSegmentSize,
  60             "Invalid position %s", position);
  61         this.position = position;
  62         fileReader.invalidateCache();
  63     }
  64
  65     /**
  66      * Invalidate any cache that is present, so that the next read is coherent with the backing file.
  67      */
  68     void invalidateCache() {
  69         fileReader.invalidateCache();
  70     }
  71
  72     /**
  73      * Reads the next binary data block.
  74      *
  75      * @return The binary data, or {@code null}
  76      */
  77     @Nullable ByteBuf readBytes() {
  78         // Check if there is enough in the buffer remaining
  79         final int remaining = maxSegmentSize - position - SegmentEntry.HEADER_BYTES;
  80         if (remaining < 0) {
  81             // Not enough space in the segment, there can never be another entry
  82             return null;
  83         }
  84
  85         // Calculate maximum entry length not exceeding file size nor maxEntrySize
  86         final var maxLength = Math.min(remaining, maxEntrySize);
  87         final var buffer = fileReader.read(position, maxLength + SegmentEntry.HEADER_BYTES);
  88
  89         // Read the entry length
  90         final var length = buffer.getInt(0);
  91         if (length < 1 || length > maxLength) {
  92             // Invalid length, make sure next read re-tries
  93             invalidateCache();
  94             return null;
  95         }
  96
  97         // Read the entry checksum
  98         final int checksum = buffer.getInt(Integer.BYTES);
  99
 100         // Slice off the entry's bytes
 101         final var entryBuffer = buffer.slice(SegmentEntry.HEADER_BYTES, length);
 102         // If the stored checksum does not equal the computed checksum, do not proceed further
 103         final var computed = SegmentEntry.computeChecksum(entryBuffer);
 104         if (checksum != computed) {
 105             LOG.warn("Expected checksum {}, computed {}", Integer.toHexString(checksum), Integer.toHexString(computed));
 106             invalidateCache();
 107             return null;
 108         }
 109
 110         // update position
 111         position += SegmentEntry.HEADER_BYTES + length;
 112
 113         // rewind and return
 114         return Unpooled.wrappedBuffer(entryBuffer.rewind());
 115     }
 116
 117     /**
 118      * Close this reader.
 119      */
 120     void close() {
 121         segment.closeReader(this);
 122     }
 123 }