@Test public void readByteBufferPerformanceTest() throws IOException, InterruptedException, TimeoutException { long maxSize = 1L * 1024 * 1024 * 1024; int logSize = 1024; int batchSize = 10 * 1024; PreloadBufferPool bufferPool = PreloadBufferPool.getInstance(); try (PositioningStore<ByteBuffer> store = prepareStore(bufferPool)) { ByteBuffer buffer = MessageTestUtils.createMessage(new byte[logSize]); long writeSize = write(store, maxSize, buffer); readByteBuffer(store, batchSize, writeSize); } }
public ByteBuffer readByteBuffer(long position, int length) throws IOException { checkReadPosition(position); try { StoreFile<T> storeFile = storeFileMap.floorEntry(position).getValue(); int relPosition = (int) (position - storeFile.position()); ByteBuffer byteBuffer = storeFile.readByteBuffer(relPosition, length); byteBuffer.limit(byteBuffer.position() + serializer.trim(byteBuffer, length)); return byteBuffer; } catch (Throwable t) { logger.warn("Exception on readByteBuffer position {} of store {}, " + "leftPosition: {}, rightPosition: {}, flushPosition: {}.", position, base.getAbsolutePath(), Format.formatWithComma(left()), Format.formatWithComma(right()), Format.formatWithComma(flushPosition()), t); throw t; } }
PositioningStore seable { { public ByteBuffer readByteBuffer(long position, int length) throws IOException { checkReadPosition(position); try { StoreFile<T> storeFile = storeFileMap.floorEntry(position).getValue(); int relPosition = (int) (position - storeFile.position()); ByteBuffer byteBuffer = storeFile.readByteBuffer(relPosition, length); byteBuffer.limit(byteBuffer.position() + serializer.trim(byteBuffer, length)); return byteBuffer; } catch (Throwable t) { logger.warn("Exception on readByteBuffer position {} of store {}, " + "leftPosition: {}, rightPosition: {}, flushPosition: {}.", position, base.getAbsolutePath(), Format.formatWithComma(left()), Format.formatWithComma(right()), Format.formatWithComma(flushPosition()), t); throw t; } } }
PositioningStore seable { { public ByteBuffer readByteBuffer(long position, int length) throws IOException { checkReadPosition(position); try { StoreFile<T> storeFile = storeFileMap.floorEntry(position).getValue(); int relPosition = (int) (position - storeFile.position()); ByteBuffer byteBuffer = storeFile.readByteBuffer(relPosition, length); byteBuffer.limit(byteBuffer.position() + serializer.trim(byteBuffer, length)); return byteBuffer; } catch (Throwable t) { logger.warn("Exception on readByteBuffer position {} of store {}, " + "leftPosition: {}, rightPosition: {}, flushPosition: {}.", position, base.getAbsolutePath(), Format.formatWithComma(left()), Format.formatWithComma(right()), Format.formatWithComma(flushPosition()), t); throw t; } } PositioningStore(File base, Config config, PreloadBufferPool bufferPool, LogSerializer<T> serializer); }
PositioningStore seable { { public ByteBuffer readByteBuffer(long position, int length) throws IOException { checkReadPosition(position); try { StoreFile<T> storeFile = storeFileMap.floorEntry(position).getValue(); int relPosition = (int) (position - storeFile.position()); ByteBuffer byteBuffer = storeFile.readByteBuffer(relPosition, length); byteBuffer.limit(byteBuffer.position() + serializer.trim(byteBuffer, length)); return byteBuffer; } catch (Throwable t) { logger.warn("Exception on readByteBuffer position {} of store {}, " + "leftPosition: {}, rightPosition: {}, flushPosition: {}.", position, base.getAbsolutePath(), Format.formatWithComma(left()), Format.formatWithComma(right()), Format.formatWithComma(flushPosition()), t); throw t; } } PositioningStore(File base, Config config, PreloadBufferPool bufferPool, LogSerializer<T> serializer); long left(); long right(); long flushPosition(); void clear(long position); void setRight(long position); void recover(); long toLogStart(long position); long appendByteBuffer(ByteBuffer byteBuffer); long append(T t); long append(final List<T> ts); boolean flush(); boolean isDiskFull(); T read(long position); T read(long position, int length); List<T> batchRead(long position, int count); ByteBuffer readByteBuffer(long position, int length); long physicalSize(); long physicalDeleteTo(long position); boolean isClean(); long physicalDeleteLeftFile(); long position(long position, int offsetCount); File base(); @Override void close(); byte[] readBytes(long position, int length); int fileCount(); int meetMinStoreFile(long minIndexedPhysicalPosition); }
PositioningStore seable { { public ByteBuffer readByteBuffer(long position, int length) throws IOException { checkReadPosition(position); try { StoreFile<T> storeFile = storeFileMap.floorEntry(position).getValue(); int relPosition = (int) (position - storeFile.position()); ByteBuffer byteBuffer = storeFile.readByteBuffer(relPosition, length); byteBuffer.limit(byteBuffer.position() + serializer.trim(byteBuffer, length)); return byteBuffer; } catch (Throwable t) { logger.warn("Exception on readByteBuffer position {} of store {}, " + "leftPosition: {}, rightPosition: {}, flushPosition: {}.", position, base.getAbsolutePath(), Format.formatWithComma(left()), Format.formatWithComma(right()), Format.formatWithComma(flushPosition()), t); throw t; } } PositioningStore(File base, Config config, PreloadBufferPool bufferPool, LogSerializer<T> serializer); long left(); long right(); long flushPosition(); void clear(long position); void setRight(long position); void recover(); long toLogStart(long position); long appendByteBuffer(ByteBuffer byteBuffer); long append(T t); long append(final List<T> ts); boolean flush(); boolean isDiskFull(); T read(long position); T read(long position, int length); List<T> batchRead(long position, int count); ByteBuffer readByteBuffer(long position, int length); long physicalSize(); long physicalDeleteTo(long position); boolean isClean(); long physicalDeleteLeftFile(); long position(long position, int offsetCount); File base(); @Override void close(); byte[] readBytes(long position, int length); int fileCount(); int meetMinStoreFile(long minIndexedPhysicalPosition);  }