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

   1 /*
   2  * Copyright (c) 2024 PANTHEON.tech, s.r.o. and others.  All rights reserved.
   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 static java.util.Objects.requireNonNull;
  19
  20 import io.netty.buffer.AbstractReferenceCountedByteBuf;
  21 import io.netty.buffer.ByteBuf;
  22 import io.netty.buffer.ByteBufAllocator;
  23 import io.netty.buffer.ByteBufUtil;
  24 import io.netty.util.internal.PlatformDependent;
  25 import java.io.Flushable;
  26 import java.io.IOException;
  27 import java.io.InputStream;
  28 import java.io.OutputStream;
  29 import java.io.UncheckedIOException;
  30 import java.nio.ByteBuffer;
  31 import java.nio.ByteOrder;
  32 import java.nio.MappedByteBuffer;
  33 import java.nio.channels.FileChannel;
  34 import java.nio.channels.FileChannel.MapMode;
  35 import java.nio.channels.GatheringByteChannel;
  36 import java.nio.channels.ScatteringByteChannel;
  37 import org.eclipse.jdt.annotation.NonNullByDefault;
  38
  39 /**
  40  * A {@link ByteBuf} backed by a {@link MappedByteBuffer}.
  41  */
  42 final class MappedByteBuf extends AbstractReferenceCountedByteBuf implements Flushable {
  43     private final ByteBufAllocator alloc;
  44
  45     private MappedByteBuffer byteBuffer;
  46     private ByteBuffer internalNio;
  47
  48     private MappedByteBuf(final ByteBufAllocator alloc, final MappedByteBuffer byteBuffer) {
  49         super(byteBuffer.limit());
  50         this.alloc = requireNonNull(alloc);
  51         this.byteBuffer = requireNonNull(byteBuffer);
  52     }
  53
  54     @NonNullByDefault
  55     static MappedByteBuf of(final JournalSegmentFile file) throws IOException {
  56         return new MappedByteBuf(file.allocator(), file.channel().map(MapMode.READ_WRITE, 0, file.maxSize()));
  57     }
  58
  59     @Override
  60     @SuppressWarnings("checkstyle:avoidHidingCauseException")
  61     public void flush() throws IOException {
  62         ensureAccessible();
  63         try {
  64             byteBuffer.force();
  65         } catch (UncheckedIOException e) {
  66             throw e.getCause();
  67         }
  68     }
  69
  70     @Override
  71     protected void deallocate() {
  72         final var local = byteBuffer;
  73         if (local != null) {
  74             byteBuffer = null;
  75             PlatformDependent.freeDirectBuffer(local);
  76         }
  77     }
  78
  79     @Override
  80     public ByteBufAllocator alloc() {
  81         return alloc;
  82     }
  83
  84     @Override
  85     public boolean isContiguous() {
  86         return true;
  87     }
  88
  89     @Override
  90     public boolean hasArray() {
  91         return false;
  92     }
  93
  94     @Override
  95     public byte[] array() {
  96         throw new UnsupportedOperationException();
  97     }
  98
  99     @Override
 100     public int arrayOffset() {
 101         throw new UnsupportedOperationException();
 102     }
 103
 104     @Override
 105     public boolean hasMemoryAddress() {
 106         return false;
 107     }
 108
 109     @Override
 110     public long memoryAddress() {
 111         throw new UnsupportedOperationException();
 112     }
 113
 114     @Override
 115     public int capacity() {
 116         return maxCapacity();
 117     }
 118
 119     @Override
 120     public ByteBuf capacity(final int newCapacity) {
 121         throw new UnsupportedOperationException("capacity cannot be set");
 122     }
 123
 124     @Override
 125     @Deprecated
 126     public ByteOrder order() {
 127         return ByteOrder.BIG_ENDIAN;
 128     }
 129
 130     @Override
 131     public ByteBuf unwrap() {
 132         return null;
 133     }
 134
 135     @Override
 136     public ByteBuf copy(final int index, final int length) {
 137         ensureAccessible();
 138         return alloc.heapBuffer(length).writeBytes(byteBuffer.slice(index, length));
 139     }
 140
 141     @Override
 142     public boolean isDirect() {
 143         return true;
 144     }
 145
 146     @Override
 147     public int nioBufferCount() {
 148         return 1;
 149     }
 150
 151     @Override
 152     public ByteBuffer nioBuffer(final int index, final int length) {
 153         checkIndex(index, length);
 154         return byteBuffer.slice(index, length);
 155     }
 156
 157     @Override
 158     public ByteBuffer internalNioBuffer(final int index, final int length) {
 159         checkIndex(index, length);
 160         return internalNio().position(index).limit(index + length);
 161     }
 162
 163     private ByteBuffer internalNio() {
 164         var local = internalNio;
 165         if (local == null) {
 166             internalNio = local = byteBuffer.duplicate();
 167         }
 168         return local;
 169     }
 170
 171     @Override
 172     public ByteBuffer[] nioBuffers(final int index, final int length) {
 173         return new ByteBuffer[] { nioBuffer(index, length) };
 174     }
 175
 176     @Override
 177     public byte getByte(final int index) {
 178         ensureAccessible();
 179         return _getByte(index);
 180     }
 181
 182     @Override
 183     protected byte _getByte(final int index) {
 184         return byteBuffer.get(index);
 185     }
 186
 187     @Override
 188     public short getShort(final int index) {
 189         ensureAccessible();
 190         return _getShort(index);
 191     }
 192
 193     @Override
 194     protected short _getShort(final int index) {
 195         return byteBuffer.getShort(index);
 196     }
 197
 198     @Override
 199     protected short _getShortLE(final int index) {
 200         return ByteBufUtil.swapShort(byteBuffer.getShort(index));
 201     }
 202
 203     @Override
 204     public int getUnsignedMedium(final int index) {
 205         ensureAccessible();
 206         return _getUnsignedMedium(index);
 207     }
 208
 209     @Override
 210     protected int _getUnsignedMedium(final int index) {
 211         return (_getByte(index) & 0xff) << 16 | (_getByte(index + 1) & 0xff) << 8 | _getByte(index + 2) & 0xff;
 212     }
 213
 214     @Override
 215     protected int _getUnsignedMediumLE(final int index) {
 216         return _getByte(index) & 0xff | (_getByte(index + 1) & 0xff) << 8 | (_getByte(index + 2) & 0xff) << 16;
 217     }
 218
 219     @Override
 220     public int getInt(final int index) {
 221         ensureAccessible();
 222         return _getInt(index);
 223     }
 224
 225     @Override
 226     protected int _getInt(final int index) {
 227         return byteBuffer.getInt(index);
 228     }
 229
 230     @Override
 231     protected int _getIntLE(final int index) {
 232         return ByteBufUtil.swapInt(byteBuffer.getInt(index));
 233     }
 234
 235     @Override
 236     public long getLong(final int index) {
 237         ensureAccessible();
 238         return _getLong(index);
 239     }
 240
 241     @Override
 242     protected long _getLong(final int index) {
 243         return byteBuffer.getLong(index);
 244     }
 245
 246     @Override
 247     protected long _getLongLE(final int index) {
 248         return ByteBufUtil.swapLong(byteBuffer.getLong(index));
 249     }
 250
 251     @Override
 252     public ByteBuf setByte(final int index, final int value) {
 253         ensureAccessible();
 254         _setByte(index, value);
 255         return this;
 256     }
 257
 258     @Override
 259     protected void _setByte(final int index, final int value) {
 260         byteBuffer.put(index, (byte) value);
 261     }
 262
 263     @Override
 264     public ByteBuf setShort(final int index, final int value) {
 265         ensureAccessible();
 266         _setShort(index, value);
 267         return this;
 268     }
 269
 270     @Override
 271     protected void _setShort(final int index, final int value) {
 272         byteBuffer.putShort(index, (short) value);
 273     }
 274
 275     @Override
 276     protected void _setShortLE(final int index, final int value) {
 277         byteBuffer.putShort(index, ByteBufUtil.swapShort((short) value));
 278     }
 279
 280     @Override
 281     public ByteBuf setMedium(final int index, final int value) {
 282         ensureAccessible();
 283         _setMedium(index, value);
 284         return this;
 285     }
 286
 287     @Override
 288     protected void _setMedium(final int index, final int value) {
 289         setByte(index, (byte) (value >>> 16));
 290         setByte(index + 1, (byte) (value >>> 8));
 291         setByte(index + 2, (byte) value);
 292     }
 293
 294     @Override
 295     protected void _setMediumLE(final int index, final int value) {
 296         setByte(index, (byte) value);
 297         setByte(index + 1, (byte) (value >>> 8));
 298         setByte(index + 2, (byte) (value >>> 16));
 299     }
 300
 301     @Override
 302     public ByteBuf setInt(final int index, final int value) {
 303         ensureAccessible();
 304         _setInt(index, value);
 305         return this;
 306     }
 307
 308     @Override
 309     protected void _setInt(final int index, final int value) {
 310         byteBuffer.putInt(index, value);
 311     }
 312
 313     @Override
 314     protected void _setIntLE(final int index, final int value) {
 315         byteBuffer.putInt(index, ByteBufUtil.swapInt(value));
 316     }
 317
 318     @Override
 319     public ByteBuf setLong(final int index, final long value) {
 320         ensureAccessible();
 321         _setLong(index, value);
 322         return this;
 323     }
 324
 325     @Override
 326     protected void _setLong(final int index, final long value) {
 327         byteBuffer.putLong(index, value);
 328     }
 329
 330     @Override
 331     protected void _setLongLE(final int index, final long value) {
 332         byteBuffer.putLong(index, ByteBufUtil.swapLong(value));
 333     }
 334
 335     @Override
 336     public ByteBuf getBytes(final int index, final ByteBuf dst, final int dstIndex, final int length) {
 337         checkDstIndex(index, length, dstIndex, dst.capacity());
 338         if (dst.hasArray()) {
 339             byteBuffer.get(index, dst.array(), dst.arrayOffset() + dstIndex, length);
 340         } else {
 341             dst.setBytes(dstIndex, this, index, length);
 342         }
 343         return this;
 344     }
 345
 346     @Override
 347     public ByteBuf getBytes(final int index, final byte[] dst, final int dstIndex, final int length) {
 348         checkDstIndex(index, length, dstIndex, dst.length);
 349         byteBuffer.get(index, dst, dstIndex, length);
 350         return this;
 351     }
 352
 353     @Override
 354     public ByteBuf getBytes(final int index, final ByteBuffer dst) {
 355         final var remaining = dst.remaining();
 356         checkIndex(index, remaining);
 357         dst.put(byteBuffer.slice(index, remaining));
 358         return this;
 359     }
 360
 361     @Override
 362     public ByteBuf getBytes(final int index, final OutputStream out, final int length) throws IOException {
 363         throw new UnsupportedOperationException();
 364     }
 365
 366     @Override
 367     public int getBytes(final int index, final GatheringByteChannel out, final int length) throws IOException {
 368         throw new UnsupportedOperationException();
 369     }
 370
 371     @Override
 372     public int getBytes(final int index, final FileChannel out, final long position, final int length)
 373             throws IOException {
 374         throw new UnsupportedOperationException();
 375     }
 376
 377     @Override
 378     public ByteBuf setBytes(final int index, final ByteBuf src, final int srcIndex, final int length) {
 379         checkSrcIndex(index, length, srcIndex, src.capacity());
 380         src.getBytes(srcIndex, this, index, length);
 381         return this;
 382     }
 383
 384     @Override
 385     public ByteBuf setBytes(final int index, final byte[] src, final int srcIndex, final int length) {
 386         checkSrcIndex(index, length, srcIndex, src.length);
 387         byteBuffer.put(index, src, srcIndex, length);
 388         return this;
 389     }
 390
 391     @Override
 392     public ByteBuf setBytes(final int index, final ByteBuffer src) {
 393         ensureAccessible();
 394         byteBuffer.put(index, src, src.position(), src.remaining());
 395         return this;
 396     }
 397
 398     @Override
 399     public int setBytes(final int index, final InputStream in, final int length) throws IOException {
 400         throw new UnsupportedOperationException();
 401     }
 402
 403     @Override
 404     public int setBytes(final int index, final ScatteringByteChannel in, final int length) throws IOException {
 405         throw new UnsupportedOperationException();
 406     }
 407
 408     @Override
 409     public int setBytes(final int index, final FileChannel in, final long position, final int length)
 410             throws IOException {
 411         throw new UnsupportedOperationException();
 412     }
 413 }