atomix-storage/src/main/java/io/atomix/storage/journal/JournalSegment.java

   1 /*
   2  * Copyright 2017-2022 Open Networking Foundation and others.  All rights reserved.
   3  * Copyright (c) 2024 PANTHEON.tech, s.r.o.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  * http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17 package io.atomix.storage.journal;
  18
  19 import com.google.common.base.MoreObjects;
  20 import io.atomix.storage.journal.index.JournalIndex;
  21 import io.atomix.storage.journal.index.SparseJournalIndex;
  22 import java.io.IOException;
  23 import java.nio.channels.FileChannel;
  24 import java.nio.file.Files;
  25 import java.nio.file.StandardOpenOption;
  26 import java.util.Set;
  27 import java.util.concurrent.ConcurrentHashMap;
  28 import java.util.concurrent.atomic.AtomicInteger;
  29
  30 /**
  31  * Log segment.
  32  *
  33  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  34  */
  35 final class JournalSegment<E> implements AutoCloseable {
  36   private final JournalSegmentFile file;
  37   private final JournalSegmentDescriptor descriptor;
  38   private final StorageLevel storageLevel;
  39   private final int maxEntrySize;
  40   private final JournalIndex index;
  41   private final JournalSerdes namespace;
  42   private final Set<JournalSegmentReader<E>> readers = ConcurrentHashMap.newKeySet();
  43   private final AtomicInteger references = new AtomicInteger();
  44   private final FileChannel channel;
  45
  46   private JournalSegmentWriter<E> writer;
  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 = switch (storageLevel) {
  69         case DISK -> new DiskJournalSegmentWriter<>(channel, this, maxEntrySize, index, namespace);
  70         case MAPPED -> new MappedJournalSegmentWriter<>(channel, this, maxEntrySize, index, namespace).toFileChannel();
  71     };
  72   }
  73
  74   /**
  75    * Returns the segment's starting index.
  76    *
  77    * @return The segment's starting index.
  78    */
  79   long index() {
  80     return descriptor.index();
  81   }
  82
  83   /**
  84    * Returns the last index in the segment.
  85    *
  86    * @return The last index in the segment.
  87    */
  88   long lastIndex() {
  89     return writer.getLastIndex();
  90   }
  91
  92   /**
  93    * Returns the size of the segment.
  94    *
  95    * @return the size of the segment
  96    */
  97   int size() {
  98     try {
  99       return (int) channel.size();
 100     } catch (IOException e) {
 101       throw new StorageException(e);
 102     }
 103   }
 104
 105   /**
 106    * Returns the segment file.
 107    *
 108    * @return The segment file.
 109    */
 110   JournalSegmentFile file() {
 111     return file;
 112   }
 113
 114   /**
 115    * Returns the segment descriptor.
 116    *
 117    * @return The segment descriptor.
 118    */
 119   JournalSegmentDescriptor descriptor() {
 120     return descriptor;
 121   }
 122
 123   /**
 124    * Acquires a reference to the log segment.
 125    */
 126   private void acquire() {
 127     if (references.getAndIncrement() == 0 && storageLevel == StorageLevel.MAPPED) {
 128       writer = writer.toMapped();
 129     }
 130   }
 131
 132   /**
 133    * Releases a reference to the log segment.
 134    */
 135   private void release() {
 136     if (references.decrementAndGet() == 0) {
 137       if (storageLevel == StorageLevel.MAPPED) {
 138         writer = writer.toFileChannel();
 139       }
 140       if (!open) {
 141         finishClose();
 142       }
 143     }
 144   }
 145
 146   /**
 147    * Acquires a reference to the segment writer.
 148    *
 149    * @return The segment writer.
 150    */
 151   JournalSegmentWriter<E> acquireWriter() {
 152     checkOpen();
 153     acquire();
 154
 155     return writer;
 156   }
 157
 158   /**
 159    * Releases the reference to the segment writer.
 160    */
 161   void releaseWriter() {
 162       release();
 163   }
 164
 165   /**
 166    * Creates a new segment reader.
 167    *
 168    * @return A new segment reader.
 169    */
 170   JournalSegmentReader<E> createReader() {
 171     checkOpen();
 172     acquire();
 173
 174     final var buffer = writer.buffer();
 175     final var reader = buffer == null
 176       ? new DiskJournalSegmentReader<>(channel, this, maxEntrySize, index, namespace)
 177         : new MappedJournalSegmentReader<>(buffer, this, maxEntrySize, index, namespace);
 178     readers.add(reader);
 179     return reader;
 180   }
 181
 182   /**
 183    * Closes a segment reader.
 184    *
 185    * @param reader the closed segment reader
 186    */
 187   void closeReader(JournalSegmentReader<E> reader) {
 188     if (readers.remove(reader)) {
 189       release();
 190     }
 191   }
 192
 193   /**
 194    * Checks whether the segment is open.
 195    */
 196   private void checkOpen() {
 197     if (!open) {
 198       throw new IllegalStateException("Segment not open");
 199     }
 200   }
 201
 202   /**
 203    * Returns a boolean indicating whether the segment is open.
 204    *
 205    * @return indicates whether the segment is open
 206    */
 207   public boolean isOpen() {
 208     return open;
 209   }
 210
 211   /**
 212    * Closes the segment.
 213    */
 214   @Override
 215   public void close() {
 216     if (!open) {
 217       return;
 218     }
 219
 220     open = false;
 221     readers.forEach(JournalSegmentReader::close);
 222     if (references.get() == 0) {
 223       finishClose();
 224     }
 225   }
 226
 227   private void finishClose() {
 228     writer.close();
 229     try {
 230       channel.close();
 231     } catch (IOException e) {
 232       throw new StorageException(e);
 233     }
 234   }
 235
 236   /**
 237    * Deletes the segment.
 238    */
 239   void delete() {
 240     try {
 241       Files.deleteIfExists(file.file().toPath());
 242     } catch (IOException e) {
 243       throw new StorageException(e);
 244     }
 245   }
 246
 247   @Override
 248   public String toString() {
 249     return MoreObjects.toStringHelper(this)
 250         .add("id", descriptor.id())
 251         .add("version", descriptor.version())
 252         .add("index", index())
 253         .toString();
 254   }
 255 }