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      * @param index entry index
  76      * @return The binary data, or {@code null}
  77      */
  78     @Nullable ByteBuf readBytes(final long index) {
  79         // Check if there is enough in the buffer remaining
  80         final int remaining = maxSegmentSize - position - SegmentEntry.HEADER_BYTES;
  81         if (remaining < 0) {
  82             // Not enough space in the segment, there can never be another entry
  83             return null;
  84         }
  85
  86         // Calculate maximum entry length not exceeding file size nor maxEntrySize
  87         final var maxLength = Math.min(remaining, maxEntrySize);
  88         final var buffer = fileReader.read(position, maxLength + SegmentEntry.HEADER_BYTES);
  89
  90         // Read the entry length
  91         final var length = buffer.getInt(0);
  92         if (length < 1 || length > maxLength) {
  93             // Invalid length, make sure next read re-tries
  94             invalidateCache();
  95             return null;
  96         }
  97
  98         // Read the entry checksum
  99         final int checksum = buffer.getInt(Integer.BYTES);
 100
 101         // Slice off the entry's bytes
 102         final var entryBuffer = buffer.slice(SegmentEntry.HEADER_BYTES, length);
 103         // If the stored checksum does not equal the computed checksum, do not proceed further
 104         final var computed = SegmentEntry.computeChecksum(entryBuffer);
 105         if (checksum != computed) {
 106             LOG.warn("Expected checksum {}, computed {}", Integer.toHexString(checksum), Integer.toHexString(computed));
 107             invalidateCache();
 108             return null;
 109         }
 110
 111         // update position
 112         position += SegmentEntry.HEADER_BYTES + length;
 113
 114         // rewind and return
 115         return Unpooled.buffer(length).writeBytes(entryBuffer.rewind());
 116     }
 117
 118     /**
 119      * Close this reader.
 120      */
 121     void close() {
 122         segment.closeReader(this);
 123     }
 124 }