2 * Copyright 2017-2022 Open Networking Foundation and others. All rights reserved.
3 * Copyright (c) 2024 PANTHEON.tech, s.r.o.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
17 package io.atomix.storage.journal;
19 import static io.atomix.storage.journal.SegmentEntry.HEADER_BYTES;
21 import com.esotericsoftware.kryo.KryoException;
22 import io.atomix.storage.journal.index.JournalIndex;
23 import java.io.IOException;
24 import java.nio.BufferOverflowException;
25 import java.nio.ByteBuffer;
26 import java.nio.MappedByteBuffer;
27 import java.nio.channels.FileChannel;
28 import java.util.zip.CRC32;
29 import org.eclipse.jdt.annotation.NonNull;
34 * The format of an entry in the log is as follows:
36 * <li>64-bit index</li>
37 * <li>8-bit boolean indicating whether a term change is contained in the entry</li>
38 * <li>64-bit optional term</li>
39 * <li>32-bit signed entry length, including the entry type ID</li>
40 * <li>8-bit signed entry type ID</li>
41 * <li>n-bit entry bytes</li>
44 * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
46 final class MappedJournalSegmentWriter<E> extends JournalSegmentWriter<E> {
47 private final @NonNull MappedByteBuffer mappedBuffer;
48 private final JournalSegmentReader<E> reader;
49 private final ByteBuffer buffer;
51 MappedJournalSegmentWriter(final FileChannel channel, final JournalSegment<E> segment, final int maxEntrySize,
52 final JournalIndex index, final JournalSerdes namespace) {
53 super(channel, segment, maxEntrySize, index, namespace);
55 mappedBuffer = mapBuffer(channel, maxSegmentSize);
56 buffer = mappedBuffer.slice();
57 reader = new JournalSegmentReader<>(segment, new MappedFileReader(segment.file().file().toPath(), mappedBuffer),
58 maxEntrySize, namespace);
62 MappedJournalSegmentWriter(final JournalSegmentWriter<E> previous) {
65 mappedBuffer = mapBuffer(channel, maxSegmentSize);
66 buffer = mappedBuffer.slice();
67 reader = new JournalSegmentReader<>(segment, new MappedFileReader(segment.file().file().toPath(), mappedBuffer),
68 maxEntrySize, namespace);
71 private static @NonNull MappedByteBuffer mapBuffer(final FileChannel channel, final int maxSegmentSize) {
73 return channel.map(FileChannel.MapMode.READ_WRITE, 0, maxSegmentSize);
74 } catch (IOException e) {
75 throw new StorageException(e);
80 @NonNull MappedByteBuffer buffer() {
85 MappedJournalSegmentWriter<E> toMapped() {
90 DiskJournalSegmentWriter<E> toFileChannel() {
92 return new DiskJournalSegmentWriter<>(this);
96 JournalSegmentReader<E> reader() {
101 @SuppressWarnings("unchecked")
102 <T extends E> Indexed<T> append(final T entry) {
103 // Store the entry index.
104 final long index = getNextIndex();
106 // Serialize the entry.
107 final int bodyPosition = currentPosition + HEADER_BYTES;
108 final int avail = maxSegmentSize - bodyPosition;
110 throw new BufferOverflowException();
113 final var entryBytes = buffer.slice(bodyPosition, Math.min(avail, maxEntrySize));
115 namespace.serialize(entry, entryBytes);
116 } catch (KryoException e) {
117 if (entryBytes.capacity() != maxEntrySize) {
118 // We have not provided enough capacity, signal to roll to next segment
119 throw new BufferOverflowException();
122 // Just reset the buffer. There's no need to zero the bytes since we haven't written the length or checksum.
123 throw new StorageException.TooLarge("Entry size exceeds maximum allowed bytes (" + maxEntrySize + ")");
126 final int length = entryBytes.position();
128 // Compute the checksum for the entry.
129 final var crc32 = new CRC32();
130 crc32.update(entryBytes.flip());
132 // Create a single byte[] in memory for the entire entry and write it as a batch to the underlying buffer.
133 buffer.putInt(currentPosition, length).putInt(currentPosition + Integer.BYTES, (int) crc32.getValue());
135 // Update the last entry with the correct index/term/length.
136 Indexed<E> indexedEntry = new Indexed<>(index, entry, length);
137 lastEntry = indexedEntry;
138 this.index.index(index, currentPosition);
140 currentPosition = currentPosition + HEADER_BYTES + length;
141 return (Indexed<T>) indexedEntry;
145 void writeEmptyHeader(final int position) {
146 // Note: we issue a single putLong() instead of two putInt()s.
147 buffer.putLong(position, 0L);
152 mappedBuffer.force();
159 BufferCleaner.freeBuffer(mappedBuffer);
160 } catch (IOException e) {
161 throw new StorageException(e);