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

   1 /*
   2  * Copyright 2017-2022 Open Networking Foundation and others.  All rights reserved.
   3  * Copyright (c) 2024 PANTHEON.tech, s.r.o. and others.
   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 static java.util.Objects.requireNonNull;
  20
  21 import java.io.File;
  22
  23 /**
  24  * Segmented journal.
  25  */
  26 public final class SegmentedJournal<E> implements Journal<E> {
  27     private final SegmentedByteBufJournal journal;
  28     private final SegmentedJournalWriter<E> writer;
  29     private final ByteBufMapper<E> mapper;
  30
  31     public SegmentedJournal(final SegmentedByteBufJournal journal, final ByteBufMapper<E> mapper) {
  32         this.journal = requireNonNull(journal, "journal is required");
  33         this.mapper = requireNonNull(mapper, "mapper cannot be null");
  34         writer = new SegmentedJournalWriter<>(journal.writer(), mapper);
  35     }
  36
  37     @Override
  38     public long lastIndex() {
  39         return journal.lastIndex();
  40     }
  41
  42     @Override
  43     public JournalWriter<E> writer() {
  44         return writer;
  45     }
  46
  47     @Override
  48     public JournalReader<E> openReader(final long index) {
  49         return openReader(index, JournalReader.Mode.ALL);
  50     }
  51
  52     /**
  53      * Opens a new journal reader with the given reader mode.
  54      *
  55      * @param index The index from which to begin reading entries.
  56      * @param mode The mode in which to read entries.
  57      * @return The journal reader.
  58      */
  59     @Override
  60     public JournalReader<E> openReader(final long index, final JournalReader.Mode mode) {
  61         final var byteReader = switch (mode) {
  62             case ALL -> journal.openReader(index);
  63             case COMMITS -> journal.openCommitsReader(index);
  64         };
  65         return new SegmentedJournalReader<>(byteReader, mapper);
  66     }
  67
  68     @Override
  69     public void close() {
  70         journal.close();
  71     }
  72
  73     /**
  74      * Compacts the journal up to the given index.
  75      *
  76      * <p>
  77      * The semantics of compaction are not specified by this interface.
  78      *
  79      * @param index The index up to which to compact the journal.
  80      */
  81     public void compact(final long index) {
  82         journal.compact(index);
  83     }
  84
  85     /**
  86      * Returns a new segmented journal builder.
  87      *
  88      * @return A new segmented journal builder.
  89      */
  90     public static <E> Builder<E> builder() {
  91         return new Builder<>();
  92     }
  93
  94     public static final class Builder<E> {
  95         private final SegmentedByteBufJournal.Builder byteJournalBuilder = SegmentedByteBufJournal.builder();
  96         private ByteBufMapper<E> mapper;
  97
  98         private Builder() {
  99             // on purpose
 100         }
 101
 102         /**
 103          * Sets the journal name.
 104          *
 105          * @param name The journal name.
 106          * @return The journal builder.
 107          */
 108         public Builder<E> withName(final String name) {
 109             byteJournalBuilder.withName(name);
 110             return this;
 111         }
 112
 113         /**
 114          * Sets the journal storage level.
 115          *
 116          * <p>
 117          * The storage level indicates how individual entries will be persisted in the journal.
 118          *
 119          * @param storageLevel The log storage level.
 120          * @return The journal builder.
 121          */
 122         public Builder<E> withStorageLevel(final StorageLevel storageLevel) {
 123             byteJournalBuilder.withStorageLevel(storageLevel);
 124             return this;
 125         }
 126
 127         /**
 128          * Sets the journal storage directory.
 129          *
 130          * <p>
 131          * The journal will write segment files into the provided directory.
 132          *
 133          * @param directory The journal storage directory.
 134          * @return The journal builder.
 135          * @throws NullPointerException If the {@code directory} is {@code null}
 136          */
 137         public Builder<E> withDirectory(final String directory) {
 138             byteJournalBuilder.withDirectory(directory);
 139             return this;
 140         }
 141
 142         /**
 143          * Sets the journal storage directory.
 144          *
 145          * <p>
 146          * The journal will write segment files into the provided directory.
 147          *
 148          * @param directory The journal storage directory.
 149          * @return The journal builder.
 150          * @throws NullPointerException If the {@code directory} is {@code null}
 151          */
 152         public Builder<E> withDirectory(final File directory) {
 153             byteJournalBuilder.withDirectory(directory);
 154             return this;
 155         }
 156
 157         /**
 158          * Sets the journal namespace.
 159          *
 160          * @param namespace The journal serializer.
 161          * @return The journal builder.
 162          * @deprecated due to serialization refactoring, use {@link Builder#withMapper(ByteBufMapper)} instead
 163          */
 164         @Deprecated(forRemoval = true, since = "9.0.3")
 165         public Builder<E> withNamespace(final JournalSerdes namespace) {
 166             return withMapper(requireNonNull(namespace, "namespace cannot be null").toMapper());
 167         }
 168
 169         /**
 170          * Sets journal serializer.
 171          *
 172          * @param mapper Journal serializer
 173          * @return The journal builder
 174          */
 175         public Builder<E> withMapper(final ByteBufMapper<E> mapper) {
 176             this.mapper = requireNonNull(mapper);
 177             return this;
 178         }
 179
 180         /**
 181          * Sets the maximum segment size in bytes.
 182          *
 183          * <p>
 184          * The maximum segment size dictates when journal should roll over to new segments. As entries are written
 185          * to a journal segment, once the size of the segment surpasses the configured maximum segment size, the
 186          * journal will create a new segment and append new entries to that segment.
 187          *
 188          * <p>
 189          * By default, the maximum segment size is 32M.
 190          *
 191          * @param maxSegmentSize The maximum segment size in bytes.
 192          * @return The storage builder.
 193          * @throws IllegalArgumentException If the {@code maxSegmentSize} is not positive
 194          */
 195         public Builder<E> withMaxSegmentSize(final int maxSegmentSize) {
 196             byteJournalBuilder.withMaxSegmentSize(maxSegmentSize);
 197             return this;
 198         }
 199
 200         /**
 201          * Sets the maximum entry size in bytes.
 202          *
 203          * @param maxEntrySize the maximum entry size in bytes
 204          * @return the storage builder
 205          * @throws IllegalArgumentException if the {@code maxEntrySize} is not positive
 206          */
 207         public Builder<E> withMaxEntrySize(final int maxEntrySize) {
 208             byteJournalBuilder.withMaxEntrySize(maxEntrySize);
 209             return this;
 210         }
 211
 212         /**
 213          * Sets the maximum number of entries per segment.
 214          *
 215          * @param maxEntriesPerSegment The maximum number of entries allowed per segment.
 216          * @return The journal builder.
 217          * @deprecated since 3.0.2, no longer used
 218          */
 219         @Deprecated
 220         public Builder<E> withMaxEntriesPerSegment(final int maxEntriesPerSegment) {
 221             // ignore
 222             return this;
 223         }
 224
 225         /**
 226          * Sets the journal index density.
 227          *
 228          * <p>
 229          * The index density is the frequency at which the position of entries written to the journal will be recorded
 230          * in an in-memory index for faster seeking.
 231          *
 232          * @param indexDensity the index density
 233          * @return the journal builder
 234          * @throws IllegalArgumentException if the density is not between 0 and 1
 235          */
 236         public Builder<E> withIndexDensity(final double indexDensity) {
 237             byteJournalBuilder.withIndexDensity(indexDensity);
 238             return this;
 239         }
 240
 241         /**
 242          * Enables flushing buffers to disk when entries are committed to a segment.
 243          *
 244          * <p>
 245          * When flush-on-commit is enabled, log entry buffers will be automatically flushed to disk each time an
 246          * entry is committed in a given segment.
 247          *
 248          * @return The journal builder.
 249          */
 250         public Builder<E> withFlushOnCommit() {
 251             return withFlushOnCommit(true);
 252         }
 253
 254         /**
 255          * Enables flushing buffers to disk when entries are committed to a segment.
 256          *
 257          * <p>
 258          * When flush-on-commit is enabled, log entry buffers will be automatically flushed to disk each time an
 259          * entry is committed in a given segment.
 260          *
 261          * @param flushOnCommit Whether to flush buffers to disk when entries are committed to a segment.
 262          * @return The journal builder.
 263          */
 264         public Builder<E> withFlushOnCommit(final boolean flushOnCommit) {
 265             byteJournalBuilder.withFlushOnCommit(flushOnCommit);
 266             return this;
 267         }
 268
 269         /**
 270          * Build the {@link SegmentedJournal}.
 271          *
 272          * @return {@link SegmentedJournal} instance.
 273          */
 274         public SegmentedJournal<E> build() {
 275             return new SegmentedJournal<>(byteJournalBuilder.build(), mapper);
 276         }
 277     }
 278 }