third-party/atomix/storage/src/main/java/io/atomix/storage/journal/MappedJournalSegmentReader.java

   1 /*
   2  * Copyright 2017-present Open Networking Foundation
   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 io.atomix.storage.journal.index.JournalIndex;
  19 import io.atomix.storage.journal.index.Position;
  20 import io.atomix.utils.serializer.Namespace;
  21
  22 import java.nio.BufferUnderflowException;
  23 import java.nio.ByteBuffer;
  24 import java.util.NoSuchElementException;
  25 import java.util.zip.CRC32;
  26
  27 /**
  28  * Log segment reader.
  29  *
  30  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  31  */
  32 class MappedJournalSegmentReader<E> implements JournalReader<E> {
  33   private final ByteBuffer buffer;
  34   private final int maxEntrySize;
  35   private final JournalIndex index;
  36   private final Namespace namespace;
  37   private final long firstIndex;
  38   private Indexed<E> currentEntry;
  39   private Indexed<E> nextEntry;
  40
  41   MappedJournalSegmentReader(
  42       ByteBuffer buffer,
  43       JournalSegment<E> segment,
  44       int maxEntrySize,
  45       JournalIndex index,
  46       Namespace namespace) {
  47     this.buffer = buffer.slice();
  48     this.maxEntrySize = maxEntrySize;
  49     this.index = index;
  50     this.namespace = namespace;
  51     this.firstIndex = segment.index();
  52     reset();
  53   }
  54
  55   @Override
  56   public long getFirstIndex() {
  57     return firstIndex;
  58   }
  59
  60   @Override
  61   public long getCurrentIndex() {
  62     return currentEntry != null ? currentEntry.index() : 0;
  63   }
  64
  65   @Override
  66   public Indexed<E> getCurrentEntry() {
  67     return currentEntry;
  68   }
  69
  70   @Override
  71   public long getNextIndex() {
  72     return currentEntry != null ? currentEntry.index() + 1 : firstIndex;
  73   }
  74
  75   @Override
  76   public void reset(long index) {
  77     reset();
  78     Position position = this.index.lookup(index - 1);
  79     if (position != null) {
  80       currentEntry = new Indexed<>(position.index() - 1, null, 0);
  81       buffer.position(position.position());
  82       readNext();
  83     }
  84     while (getNextIndex() < index && hasNext()) {
  85       next();
  86     }
  87   }
  88
  89   @Override
  90   public void reset() {
  91     buffer.position(JournalSegmentDescriptor.BYTES);
  92     currentEntry = null;
  93     nextEntry = null;
  94     readNext();
  95   }
  96
  97   @Override
  98   public boolean hasNext() {
  99     // If the next entry is null, check whether a next entry exists.
 100     if (nextEntry == null) {
 101       readNext();
 102     }
 103     return nextEntry != null;
 104   }
 105
 106   @Override
 107   public Indexed<E> next() {
 108     if (!hasNext()) {
 109       throw new NoSuchElementException();
 110     }
 111
 112     // Set the current entry to the next entry.
 113     currentEntry = nextEntry;
 114
 115     // Reset the next entry to null.
 116     nextEntry = null;
 117
 118     // Read the next entry in the segment.
 119     readNext();
 120
 121     // Return the current entry.
 122     return currentEntry;
 123   }
 124
 125   /**
 126    * Reads the next entry in the segment.
 127    */
 128   private void readNext() {
 129     // Compute the index of the next entry in the segment.
 130     final long index = getNextIndex();
 131
 132     // Mark the buffer so it can be reset if necessary.
 133     buffer.mark();
 134
 135     try {
 136       // Read the length of the entry.
 137       final int length = buffer.getInt();
 138
 139       // If the buffer length is zero then return.
 140       if (length <= 0 || length > maxEntrySize) {
 141         buffer.reset();
 142         nextEntry = null;
 143         return;
 144       }
 145
 146       // Read the checksum of the entry.
 147       long checksum = buffer.getInt() & 0xFFFFFFFFL;
 148
 149       // Compute the checksum for the entry bytes.
 150       final CRC32 crc32 = new CRC32();
 151       ByteBuffer slice = buffer.slice();
 152       slice.limit(length);
 153       crc32.update(slice);
 154
 155       // If the stored checksum equals the computed checksum, return the entry.
 156       if (checksum == crc32.getValue()) {
 157         slice.rewind();
 158         E entry = namespace.deserialize(slice);
 159         nextEntry = new Indexed<>(index, entry, length);
 160         buffer.position(buffer.position() + length);
 161       } else {
 162         buffer.reset();
 163         nextEntry = null;
 164       }
 165     } catch (BufferUnderflowException e) {
 166       buffer.reset();
 167       nextEntry = null;
 168     }
 169   }
 170
 171   @Override
 172   public void close() {
 173     // Do nothing. The writer is responsible for cleaning the mapped buffer.
 174   }
 175 }