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