third-party/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 com.google.common.collect.Sets;
  19 import io.atomix.storage.journal.index.JournalIndex;
  20 import io.atomix.storage.journal.index.SparseJournalIndex;
  21 import io.atomix.utils.serializer.Namespace;
  22
  23 import java.io.File;
  24 import java.io.IOException;
  25 import java.nio.MappedByteBuffer;
  26 import java.nio.channels.FileChannel;
  27 import java.nio.file.Files;
  28 import java.nio.file.StandardOpenOption;
  29 import java.util.Set;
  30 import java.util.concurrent.atomic.AtomicInteger;
  31
  32 import static com.google.common.base.MoreObjects.toStringHelper;
  33 import static com.google.common.base.Preconditions.checkState;
  34
  35 /**
  36  * Log segment.
  37  *
  38  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  39  */
  40 public class JournalSegment<E> implements AutoCloseable {
  41   private final JournalSegmentFile file;
  42   private final JournalSegmentDescriptor descriptor;
  43   private final StorageLevel storageLevel;
  44   private final int maxEntrySize;
  45   private final JournalIndex index;
  46   private final Namespace namespace;
  47   private final MappableJournalSegmentWriter<E> writer;
  48   private final Set<MappableJournalSegmentReader<E>> readers = Sets.newConcurrentHashSet();
  49   private final AtomicInteger references = new AtomicInteger();
  50   private boolean open = true;
  51
  52   public JournalSegment(
  53       JournalSegmentFile file,
  54       JournalSegmentDescriptor descriptor,
  55       StorageLevel storageLevel,
  56       int maxEntrySize,
  57       double indexDensity,
  58       Namespace namespace) {
  59     this.file = file;
  60     this.descriptor = descriptor;
  61     this.storageLevel = storageLevel;
  62     this.maxEntrySize = maxEntrySize;
  63     this.index = new SparseJournalIndex(indexDensity);
  64     this.namespace = namespace;
  65     this.writer = new MappableJournalSegmentWriter<>(openChannel(file.file()), this, maxEntrySize, index, namespace);
  66   }
  67
  68   private FileChannel openChannel(File file) {
  69     try {
  70       return FileChannel.open(file.toPath(), StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
  71     } catch (IOException e) {
  72       throw new StorageException(e);
  73     }
  74   }
  75
  76   /**
  77    * Returns the segment ID.
  78    *
  79    * @return The segment ID.
  80    */
  81   public long id() {
  82     return descriptor.id();
  83   }
  84
  85   /**
  86    * Returns the segment version.
  87    *
  88    * @return The segment version.
  89    */
  90   public long version() {
  91     return descriptor.version();
  92   }
  93
  94   /**
  95    * Returns the segment's starting index.
  96    *
  97    * @return The segment's starting index.
  98    */
  99   public long index() {
 100     return descriptor.index();
 101   }
 102
 103   /**
 104    * Returns the last index in the segment.
 105    *
 106    * @return The last index in the segment.
 107    */
 108   public long lastIndex() {
 109     return writer.getLastIndex();
 110   }
 111
 112   /**
 113    * Returns the size of the segment.
 114    *
 115    * @return the size of the segment
 116    */
 117   public int size() {
 118     return writer.size();
 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<>(
 213         openChannel(file.file()), this, maxEntrySize, index, 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   }
 257
 258   /**
 259    * Deletes the segment.
 260    */
 261   public void delete() {
 262     try {
 263       Files.deleteIfExists(file.file().toPath());
 264     } catch (IOException e) {
 265       throw new StorageException(e);
 266     }
 267   }
 268
 269   @Override
 270   public String toString() {
 271     return toStringHelper(this)
 272         .add("id", id())
 273         .add("version", version())
 274         .add("index", index())
 275         .toString();
 276   }
 277 }