@Test public void positionTest() throws Exception { PositioningStore.Config config = new PositioningStore.Config(); VirtualThreadExecutor virtualThreadPool = new VirtualThreadExecutor(500, 100, 10, 1000, 4); PreloadBufferPool bufferPool = PreloadBufferPool.getInstance(); PositioningStore<ByteBuffer> store = new PositioningStore<>(logBase, new PositioningStore.Config(), bufferPool, new StoreMessageSerializer(1024 * 1024)); store.recover(); int count = 10; String writeMessage = "Hello, world!"; List<String> bodyList = MessageTestUtils.createBodyList(writeMessage, count); List<ByteBuffer> writeMessages = MessageTestUtils.createMessages(bodyList); long writePosition = store.append(writeMessages); long position = store.position(writePosition, 3); Assert.assertEquals(writePosition, position); position = store.position(position, -1); long expectPosition = IntStream.range(0, 9).mapToLong(i -> writeMessages.get(i).capacity()).sum(); Assert.assertEquals(expectPosition, position); position = position + 8; position = store.position(position, -5); expectPosition = IntStream.range(0, 4).mapToLong(i -> writeMessages.get(i).capacity()).sum(); Assert.assertEquals(expectPosition, position); position = position - 7; position = store.position(position, -3); expectPosition = 0; Assert.assertEquals(expectPosition, position); position = store.position(position, 5); expectPosition = IntStream.range(0, 5).mapToLong(i -> writeMessages.get(i).capacity()).sum(); Assert.assertEquals(expectPosition, position); position += 7; position = store.position(position, 2); expectPosition = IntStream.range(0, 7).mapToLong(i -> writeMessages.get(i).capacity()).sum(); Assert.assertEquals(expectPosition, position); position -= 7; position = store.position(position, 0); expectPosition = IntStream.range(0, 6).mapToLong(i -> writeMessages.get(i).capacity()).sum(); Assert.assertEquals(expectPosition, position); position = store.position(position, 10); expectPosition = IntStream.range(0, 10).mapToLong(i -> writeMessages.get(i).capacity()).sum(); Assert.assertEquals(expectPosition, position); }
public long position(long position, int offsetCount) { long t0 = SystemClock.now(); int offset = 0; long pos = position; if (pos < left()) { pos = left(); } else if (pos > right()) { pos = right(); } else if (left() < pos && pos < right()) { pos = toLogStart(position); } if (offsetCount > 0) { while (offset < offsetCount && pos < right()) { pos = toLogTail(pos); offset++; } } else if (offsetCount < 0) { while (offset > offsetCount && pos > left()) { pos = toLogStart(pos - 1); offset--; } } long t1 = SystemClock.now(); if(t1 - t0 > 300) { logger.info("Find log position takes: {}ms, store: {}", t1 - t0, base.getAbsolutePath()); } return pos; }
PositioningStore seable { { public long position(long position, int offsetCount) { long t0 = SystemClock.now(); int offset = 0; long pos = position; if (pos < left()) { pos = left(); } else if (pos > right()) { pos = right(); } else if (left() < pos && pos < right()) { pos = toLogStart(position); } if (offsetCount > 0) { while (offset < offsetCount && pos < right()) { pos = toLogTail(pos); offset++; } } else if (offsetCount < 0) { while (offset > offsetCount && pos > left()) { pos = toLogStart(pos - 1); offset--; } } long t1 = SystemClock.now(); if(t1 - t0 > 300) { logger.info("Find log position takes: {}ms, store: {}", t1 - t0, base.getAbsolutePath()); } return pos; } }
PositioningStore seable { { public long position(long position, int offsetCount) { long t0 = SystemClock.now(); int offset = 0; long pos = position; if (pos < left()) { pos = left(); } else if (pos > right()) { pos = right(); } else if (left() < pos && pos < right()) { pos = toLogStart(position); } if (offsetCount > 0) { while (offset < offsetCount && pos < right()) { pos = toLogTail(pos); offset++; } } else if (offsetCount < 0) { while (offset > offsetCount && pos > left()) { pos = toLogStart(pos - 1); offset--; } } long t1 = SystemClock.now(); if(t1 - t0 > 300) { logger.info("Find log position takes: {}ms, store: {}", t1 - t0, base.getAbsolutePath()); } return pos; } PositioningStore(File base, Config config, PreloadBufferPool bufferPool, LogSerializer<T> serializer); }
PositioningStore seable { { public long position(long position, int offsetCount) { long t0 = SystemClock.now(); int offset = 0; long pos = position; if (pos < left()) { pos = left(); } else if (pos > right()) { pos = right(); } else if (left() < pos && pos < right()) { pos = toLogStart(position); } if (offsetCount > 0) { while (offset < offsetCount && pos < right()) { pos = toLogTail(pos); offset++; } } else if (offsetCount < 0) { while (offset > offsetCount && pos > left()) { pos = toLogStart(pos - 1); offset--; } } long t1 = SystemClock.now(); if(t1 - t0 > 300) { logger.info("Find log position takes: {}ms, store: {}", t1 - t0, base.getAbsolutePath()); } return pos; } 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 long position(long position, int offsetCount) { long t0 = SystemClock.now(); int offset = 0; long pos = position; if (pos < left()) { pos = left(); } else if (pos > right()) { pos = right(); } else if (left() < pos && pos < right()) { pos = toLogStart(position); } if (offsetCount > 0) { while (offset < offsetCount && pos < right()) { pos = toLogTail(pos); offset++; } } else if (offsetCount < 0) { while (offset > offsetCount && pos > left()) { pos = toLogStart(pos - 1); offset--; } } long t1 = SystemClock.now(); if(t1 - t0 > 300) { logger.info("Find log position takes: {}ms, store: {}", t1 - t0, base.getAbsolutePath()); } return pos; } 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);  }