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