atomix-storage/src/main/java/io/atomix/storage/journal/SegmentedByteBufWriter.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 io.netty.buffer.ByteBuf;
  22
  23 /**
  24  * A {@link ByteBufWriter} implementation.
  25  */
  26 final class SegmentedByteBufWriter implements ByteBufWriter {
  27     private final SegmentedByteBufJournal journal;
  28
  29     private JournalSegment currentSegment;
  30     private JournalSegmentWriter currentWriter;
  31
  32     SegmentedByteBufWriter(final SegmentedByteBufJournal journal) {
  33         this.journal = requireNonNull(journal);
  34         currentSegment = journal.lastSegment();
  35         currentWriter = currentSegment.acquireWriter();
  36     }
  37
  38     @Override
  39     public long nextIndex() {
  40         return currentWriter.nextIndex();
  41     }
  42
  43     @Override
  44     public void commit(final long index) {
  45         if (index > journal.getCommitIndex()) {
  46             journal.setCommitIndex(index);
  47             if (journal.isFlushOnCommit()) {
  48                 flush();
  49             }
  50         }
  51     }
  52
  53     @Override
  54     public long append(final ByteBuf bytes) {
  55         final var position = currentWriter.append(bytes);
  56         return position != null ? position.index() : appendToNextSegment(bytes);
  57     }
  58
  59     //  Slow path: we do not have enough capacity
  60     private long appendToNextSegment(final ByteBuf bytes) {
  61         currentWriter.flush();
  62         currentSegment.releaseWriter();
  63         currentSegment = journal.createNextSegment();
  64         currentWriter = currentSegment.acquireWriter();
  65         return currentWriter.append(bytes).index();
  66     }
  67
  68     @Override
  69     public void reset(final long index) {
  70         final var commitIndex = journal.getCommitIndex();
  71         if (index <= commitIndex) {
  72             // also catches index == 0, which is not a valid next index
  73             throw new IndexOutOfBoundsException("Cannot reset to: " + index + ", committed index: " + commitIndex);
  74         }
  75
  76         final var lastIndex = currentSegment.lastIndex();
  77         final var prevIndex = index - 1;
  78         if (prevIndex == lastIndex) {
  79             // already at the correct position: no-op
  80             return;
  81         }
  82         if (prevIndex > lastIndex) {
  83             // cannot seek past last written entry
  84             throw new IndexOutOfBoundsException("Cannot reset to: " + index + ", lastIndex: " + lastIndex);
  85         }
  86
  87         // move back:
  88         // 1. delete all segments with first indexes greater than the given index.
  89         while (prevIndex < currentSegment.firstIndex() && currentSegment != journal.firstSegment()) {
  90             currentSegment.releaseWriter();
  91             journal.removeSegment(currentSegment);
  92             currentSegment = journal.lastSegment();
  93             currentWriter = currentSegment.acquireWriter();
  94         }
  95         // 2. truncate the current index.
  96         currentWriter.truncate(prevIndex);
  97
  98         // 3. reset segment readers.
  99         journal.resetTail(index);
 100         journal.resetHead(index);
 101     }
 102
 103     @Override
 104     public void flush() {
 105         currentWriter.flush();
 106     }
 107 }