atomix-storage/src/main/java/io/atomix/storage/journal/JournalSegment.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.SparseJournalIndex;
  20 import java.io.IOException;
  21 import java.nio.MappedByteBuffer;
  22 import java.nio.channels.FileChannel;
  23 import java.nio.file.Files;
  24 import java.nio.file.StandardOpenOption;
  25 import java.util.Set;
  26 import java.util.concurrent.ConcurrentHashMap;
  27 import java.util.concurrent.atomic.AtomicInteger;
  28
  29 import static com.google.common.base.MoreObjects.toStringHelper;
  30 import static com.google.common.base.Preconditions.checkState;
  31
  32 /**
  33  * Log segment.
  34  *
  35  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  36  */
  37 final class JournalSegment<E> implements AutoCloseable {
  38   private final JournalSegmentFile file;
  39   private final JournalSegmentDescriptor descriptor;
  40   private final StorageLevel storageLevel;
  41   private final int maxEntrySize;
  42   private final JournalIndex index;
  43   private final JournalSerdes namespace;
  44   private final MappableJournalSegmentWriter<E> writer;
  45   private final Set<MappableJournalSegmentReader<E>> readers = ConcurrentHashMap.newKeySet();
  46   private final AtomicInteger references = new AtomicInteger();
  47   private final FileChannel channel;
  48   private boolean open = true;
  49
  50   JournalSegment(
  51       JournalSegmentFile file,
  52       JournalSegmentDescriptor descriptor,
  53       StorageLevel storageLevel,
  54       int maxEntrySize,
  55       double indexDensity,
  56       JournalSerdes namespace) {
  57     this.file = file;
  58     this.descriptor = descriptor;
  59     this.storageLevel = storageLevel;
  60     this.maxEntrySize = maxEntrySize;
  61     this.namespace = namespace;
  62     index = new SparseJournalIndex(indexDensity);
  63     try {
  64       channel = FileChannel.open(file.file().toPath(),
  65         StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
  66     } catch (IOException e) {
  67       throw new StorageException(e);
  68     }
  69     writer = new MappableJournalSegmentWriter<>(channel, this, maxEntrySize, index, namespace);
  70   }
  71
  72   /**
  73    * Returns the segment ID.
  74    *
  75    * @return The segment ID.
  76    */
  77   public long id() {
  78     return descriptor.id();
  79   }
  80
  81   /**
  82    * Returns the segment version.
  83    *
  84    * @return The segment version.
  85    */
  86   public long version() {
  87     return descriptor.version();
  88   }
  89
  90   /**
  91    * Returns the segment's starting index.
  92    *
  93    * @return The segment's starting index.
  94    */
  95   public long index() {
  96     return descriptor.index();
  97   }
  98
  99   /**
 100    * Returns the last index in the segment.
 101    *
 102    * @return The last index in the segment.
 103    */
 104   public long lastIndex() {
 105     return writer.getLastIndex();
 106   }
 107
 108   /**
 109    * Returns the size of the segment.
 110    *
 111    * @return the size of the segment
 112    */
 113   public int size() {
 114     try {
 115       return (int) channel.size();
 116     } catch (IOException e) {
 117       throw new StorageException(e);
 118     }
 119   }
 120
 121   /**
 122    * Returns the segment file.
 123    *
 124    * @return The segment file.
 125    */
 126   public JournalSegmentFile file() {
 127     return file;
 128   }
 129
 130   /**
 131    * Returns the segment descriptor.
 132    *
 133    * @return The segment descriptor.
 134    */
 135   public JournalSegmentDescriptor descriptor() {
 136     return descriptor;
 137   }
 138
 139   /**
 140    * Returns a boolean value indicating whether the segment is empty.
 141    *
 142    * @return Indicates whether the segment is empty.
 143    */
 144   public boolean isEmpty() {
 145     return length() == 0;
 146   }
 147
 148   /**
 149    * Returns the segment length.
 150    *
 151    * @return The segment length.
 152    */
 153   public long length() {
 154     return writer.getNextIndex() - index();
 155   }
 156
 157   /**
 158    * Acquires a reference to the log segment.
 159    */
 160   void acquire() {
 161     if (references.getAndIncrement() == 0 && open) {
 162       map();
 163     }
 164   }
 165
 166   /**
 167    * Releases a reference to the log segment.
 168    */
 169   void release() {
 170     if (references.decrementAndGet() == 0 && open) {
 171       unmap();
 172     }
 173   }
 174
 175   /**
 176    * Maps the log segment into memory.
 177    */
 178   private void map() {
 179     if (storageLevel == StorageLevel.MAPPED) {
 180       MappedByteBuffer buffer = writer.map();
 181       readers.forEach(reader -> reader.map(buffer));
 182     }
 183   }
 184
 185   /**
 186    * Unmaps the log segment from memory.
 187    */
 188   private void unmap() {
 189     if (storageLevel == StorageLevel.MAPPED) {
 190       writer.unmap();
 191       readers.forEach(reader -> reader.unmap());
 192     }
 193   }
 194
 195   /**
 196    * Returns the segment writer.
 197    *
 198    * @return The segment writer.
 199    */
 200   public MappableJournalSegmentWriter<E> writer() {
 201     checkOpen();
 202     return writer;
 203   }
 204
 205   /**
 206    * Creates a new segment reader.
 207    *
 208    * @return A new segment reader.
 209    */
 210   MappableJournalSegmentReader<E> createReader() {
 211     checkOpen();
 212     MappableJournalSegmentReader<E> reader = new MappableJournalSegmentReader<>(channel, this, maxEntrySize, index,
 213         namespace);
 214     MappedByteBuffer buffer = writer.buffer();
 215     if (buffer != null) {
 216       reader.map(buffer);
 217     }
 218     readers.add(reader);
 219     return reader;
 220   }
 221
 222   /**
 223    * Closes a segment reader.
 224    *
 225    * @param reader the closed segment reader
 226    */
 227   void closeReader(MappableJournalSegmentReader<E> reader) {
 228     readers.remove(reader);
 229   }
 230
 231   /**
 232    * Checks whether the segment is open.
 233    */
 234   private void checkOpen() {
 235     checkState(open, "Segment not open");
 236   }
 237
 238   /**
 239    * Returns a boolean indicating whether the segment is open.
 240    *
 241    * @return indicates whether the segment is open
 242    */
 243   public boolean isOpen() {
 244     return open;
 245   }
 246
 247   /**
 248    * Closes the segment.
 249    */
 250   @Override
 251   public void close() {
 252     unmap();
 253     writer.close();
 254     readers.forEach(reader -> reader.close());
 255     open = false;
 256     try {
 257       channel.close();
 258     } catch (IOException e) {
 259       throw new StorageException(e);
 260     }
 261   }
 262
 263   /**
 264    * Deletes the segment.
 265    */
 266   public void delete() {
 267     try {
 268       Files.deleteIfExists(file.file().toPath());
 269     } catch (IOException e) {
 270       throw new StorageException(e);
 271     }
 272   }
 273
 274   @Override
 275   public String toString() {
 276     return toStringHelper(this)
 277         .add("id", id())
 278         .add("version", version())
 279         .add("index", index())
 280         .toString();
 281   }
 282 }