third-party/atomix/storage/src/main/java/io/atomix/storage/journal/FileChannelJournalSegmentWriter.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 com.esotericsoftware.kryo.KryoException;
  19 import io.atomix.storage.StorageException;
  20 import io.atomix.storage.journal.index.JournalIndex;
  21 import io.atomix.utils.serializer.Namespace;
  22
  23 import java.io.IOException;
  24 import java.nio.BufferOverflowException;
  25 import java.nio.BufferUnderflowException;
  26 import java.nio.ByteBuffer;
  27 import java.nio.channels.FileChannel;
  28 import java.util.zip.CRC32;
  29 import java.util.zip.Checksum;
  30
  31 /**
  32  * Segment writer.
  33  * <p>
  34  * The format of an entry in the log is as follows:
  35  * <ul>
  36  * <li>64-bit index</li>
  37  * <li>8-bit boolean indicating whether a term change is contained in the entry</li>
  38  * <li>64-bit optional term</li>
  39  * <li>32-bit signed entry length, including the entry type ID</li>
  40  * <li>8-bit signed entry type ID</li>
  41  * <li>n-bit entry bytes</li>
  42  * </ul>
  43  *
  44  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  45  */
  46 class FileChannelJournalSegmentWriter<E> implements JournalWriter<E> {
  47   private final FileChannel channel;
  48   private final JournalSegment segment;
  49   private final int maxEntrySize;
  50   private final JournalIndex index;
  51   private final Namespace namespace;
  52   private final ByteBuffer memory;
  53   private final long firstIndex;
  54   private Indexed<E> lastEntry;
  55
  56   FileChannelJournalSegmentWriter(
  57       FileChannel channel,
  58       JournalSegment segment,
  59       int maxEntrySize,
  60       JournalIndex index,
  61       Namespace namespace) {
  62     this.channel = channel;
  63     this.segment = segment;
  64     this.maxEntrySize = maxEntrySize;
  65     this.index = index;
  66     this.memory = ByteBuffer.allocate((maxEntrySize + Integer.BYTES + Integer.BYTES) * 2);
  67     memory.limit(0);
  68     this.namespace = namespace;
  69     this.firstIndex = segment.index();
  70     reset(0);
  71   }
  72
  73   @Override
  74   public void reset(long index) {
  75     long nextIndex = firstIndex;
  76
  77     // Clear the buffer indexes.
  78     try {
  79       channel.position(JournalSegmentDescriptor.BYTES);
  80       memory.clear().flip();
  81
  82       // Record the current buffer position.
  83       long position = channel.position();
  84
  85       // Read more bytes from the segment if necessary.
  86       if (memory.remaining() < maxEntrySize) {
  87         memory.clear();
  88         channel.read(memory);
  89         channel.position(position);
  90         memory.flip();
  91       }
  92
  93       // Read the entry length.
  94       memory.mark();
  95       int length = memory.getInt();
  96
  97       // If the length is non-zero, read the entry.
  98       while (0 < length && length <= maxEntrySize && (index == 0 || nextIndex <= index)) {
  99
 100         // Read the checksum of the entry.
 101         final long checksum = memory.getInt() & 0xFFFFFFFFL;
 102
 103         // Compute the checksum for the entry bytes.
 104         final Checksum crc32 = new CRC32();
 105         crc32.update(memory.array(), memory.position(), length);
 106
 107         // If the stored checksum equals the computed checksum, return the entry.
 108         if (checksum == crc32.getValue()) {
 109           int limit = memory.limit();
 110           memory.limit(memory.position() + length);
 111           final E entry = namespace.deserialize(memory);
 112           memory.limit(limit);
 113           lastEntry = new Indexed<>(nextIndex, entry, length);
 114           this.index.index(nextIndex, (int) position);
 115           nextIndex++;
 116         } else {
 117           break;
 118         }
 119
 120         // Update the current position for indexing.
 121         position = channel.position() + memory.position();
 122
 123         // Read more bytes from the segment if necessary.
 124         if (memory.remaining() < maxEntrySize) {
 125           channel.position(position);
 126           memory.clear();
 127           channel.read(memory);
 128           channel.position(position);
 129           memory.flip();
 130         }
 131
 132         memory.mark();
 133         length = memory.getInt();
 134       }
 135
 136       // Reset the buffer to the previous mark.
 137       channel.position(channel.position() + memory.reset().position());
 138     } catch (BufferUnderflowException e) {
 139       try {
 140         channel.position(channel.position() + memory.reset().position());
 141       } catch (IOException e2) {
 142         throw new StorageException(e2);
 143       }
 144     } catch (IOException e) {
 145       throw new StorageException(e);
 146     }
 147   }
 148
 149   @Override
 150   public long getLastIndex() {
 151     return lastEntry != null ? lastEntry.index() : segment.index() - 1;
 152   }
 153
 154   @Override
 155   public Indexed<E> getLastEntry() {
 156     return lastEntry;
 157   }
 158
 159   @Override
 160   public long getNextIndex() {
 161     if (lastEntry != null) {
 162       return lastEntry.index() + 1;
 163     } else {
 164       return firstIndex;
 165     }
 166   }
 167
 168   /**
 169    * Returns the size of the underlying buffer.
 170    *
 171    * @return The size of the underlying buffer.
 172    */
 173   public long size() {
 174     try {
 175       return channel.position();
 176     } catch (IOException e) {
 177       throw new StorageException(e);
 178     }
 179   }
 180
 181   /**
 182    * Returns a boolean indicating whether the segment is empty.
 183    *
 184    * @return Indicates whether the segment is empty.
 185    */
 186   public boolean isEmpty() {
 187     return lastEntry == null;
 188   }
 189
 190   /**
 191    * Returns a boolean indicating whether the segment is full.
 192    *
 193    * @return Indicates whether the segment is full.
 194    */
 195   public boolean isFull() {
 196     return size() >= segment.descriptor().maxSegmentSize()
 197         || getNextIndex() - firstIndex >= segment.descriptor().maxEntries();
 198   }
 199
 200   /**
 201    * Returns the first index written to the segment.
 202    */
 203   public long firstIndex() {
 204     return firstIndex;
 205   }
 206
 207   @Override
 208   @SuppressWarnings("unchecked")
 209   public void append(Indexed<E> entry) {
 210     final long nextIndex = getNextIndex();
 211
 212     // If the entry's index is greater than the next index in the segment, skip some entries.
 213     if (entry.index() > nextIndex) {
 214       throw new IndexOutOfBoundsException("Entry index is not sequential");
 215     }
 216
 217     // If the entry's index is less than the next index, truncate the segment.
 218     if (entry.index() < nextIndex) {
 219       truncate(entry.index() - 1);
 220     }
 221     append(entry.entry());
 222   }
 223
 224   @Override
 225   @SuppressWarnings("unchecked")
 226   public <T extends E> Indexed<T> append(T entry) {
 227     // Store the entry index.
 228     final long index = getNextIndex();
 229
 230     try {
 231       // Serialize the entry.
 232       memory.clear();
 233       memory.position(Integer.BYTES + Integer.BYTES);
 234       try {
 235         namespace.serialize(entry, memory);
 236       } catch (KryoException e) {
 237         throw new StorageException.TooLarge("Entry size exceeds maximum allowed bytes (" + maxEntrySize + ")");
 238       }
 239       memory.flip();
 240
 241       final int length = memory.limit() - (Integer.BYTES + Integer.BYTES);
 242
 243       // Ensure there's enough space left in the buffer to store the entry.
 244       long position = channel.position();
 245       if (segment.descriptor().maxSegmentSize() - position < length + Integer.BYTES + Integer.BYTES) {
 246         throw new BufferOverflowException();
 247       }
 248
 249       // If the entry length exceeds the maximum entry size then throw an exception.
 250       if (length > maxEntrySize) {
 251         throw new StorageException.TooLarge("Entry size " + length + " exceeds maximum allowed bytes (" + maxEntrySize + ")");
 252       }
 253
 254       // Compute the checksum for the entry.
 255       final Checksum crc32 = new CRC32();
 256       crc32.update(memory.array(), Integer.BYTES + Integer.BYTES, memory.limit() - (Integer.BYTES + Integer.BYTES));
 257       final long checksum = crc32.getValue();
 258
 259       // Create a single byte[] in memory for the entire entry and write it as a batch to the underlying buffer.
 260       memory.putInt(0, length);
 261       memory.putInt(Integer.BYTES, (int) checksum);
 262       channel.write(memory);
 263
 264       // Update the last entry with the correct index/term/length.
 265       Indexed<E> indexedEntry = new Indexed<>(index, entry, length);
 266       this.lastEntry = indexedEntry;
 267       this.index.index(index, (int) position);
 268       return (Indexed<T>) indexedEntry;
 269     } catch (IOException e) {
 270       throw new StorageException(e);
 271     }
 272   }
 273
 274   @Override
 275   public void commit(long index) {
 276
 277   }
 278
 279   @Override
 280   @SuppressWarnings("unchecked")
 281   public void truncate(long index) {
 282     // If the index is greater than or equal to the last index, skip the truncate.
 283     if (index >= getLastIndex()) {
 284       return;
 285     }
 286
 287     // Reset the last entry.
 288     lastEntry = null;
 289
 290     try {
 291       // Truncate the index.
 292       this.index.truncate(index);
 293
 294       if (index < segment.index()) {
 295         channel.position(JournalSegmentDescriptor.BYTES);
 296         channel.write(zero());
 297         channel.position(JournalSegmentDescriptor.BYTES);
 298       } else {
 299         // Reset the writer to the given index.
 300         reset(index);
 301
 302         // Zero entries after the given index.
 303         long position = channel.position();
 304         channel.write(zero());
 305         channel.position(position);
 306       }
 307     } catch (IOException e) {
 308       throw new StorageException(e);
 309     }
 310   }
 311
 312   /**
 313    * Returns a zeroed out byte buffer.
 314    */
 315   private ByteBuffer zero() {
 316     memory.clear();
 317     for (int i = 0; i < memory.limit(); i++) {
 318       memory.put(i, (byte) 0);
 319     }
 320     return memory;
 321   }
 322
 323   @Override
 324   public void flush() {
 325     try {
 326       if (channel.isOpen()) {
 327         channel.force(true);
 328       }
 329     } catch (IOException e) {
 330       throw new StorageException(e);
 331     }
 332   }
 333
 334   @Override
 335   public void close() {
 336     flush();
 337   }
 338 }