diff --git a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectReader.java b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectReader.java index 707bf307c05..29c1c52fcb5 100644 --- a/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectReader.java +++ b/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/storage/directentrylogger/DirectReader.java @@ -157,7 +157,7 @@ private int readBytesIntoBuf(ByteBuf buf, long offset, int size) throws IOExcept .kv("offset", offset) .kv("size", size).toString()); } - return nativeBuffer.readByteBuf(buf, offsetInBuffer, size); + return nativeBuffer.readByteBuf(buf, offsetInBuffer, sizeInBuffer); } } @@ -207,6 +207,9 @@ void readBlock(long offset) throws IOException { long bytesToRead = Math.min(blockSize, bytesAvailable); long bytesOutstanding = bytesToRead; long bytesRead = -1; + // A failed read may still overwrite nativeBuffer, so invalidate the + // cached block before loading a new one. + clearCache(); try { while (true) { long readSize = blockSize - bufferOffset; @@ -226,8 +229,21 @@ void readBlock(long offset) throws IOException { if ((bytesOutstanding - bytesRead) <= 0) { break; } - bytesOutstanding -= bytesRead & Buffer.ALIGNMENT; - bufferOffset += bytesRead & Buffer.ALIGNMENT; + long alignedBytesRead = bytesRead & ~(Buffer.ALIGNMENT - 1L); + if (alignedBytesRead <= 0) { + readBlockStats.registerFailedEvent(System.nanoTime() - startNs, TimeUnit.NANOSECONDS); + throw new IOException(exMsg("Short read did not make aligned progress") + .kv("requestedBytes", readSize) + .kv("offset", blockStart + bufferOffset) + .kv("expectedBytes", Math.min(blockSize, bytesAvailable)) + .kv("bytesOutstanding", bytesOutstanding) + .kv("bufferOffset", bufferOffset) + .kv("bytesRead", bytesRead) + .kv("file", filename) + .kv("fd", fd).toString()); + } + bytesOutstanding -= alignedBytesRead; + bufferOffset += alignedBytesRead; } } catch (NativeIOException ne) { readBlockStats.registerFailedEvent(System.nanoTime() - startNs, TimeUnit.NANOSECONDS); diff --git a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/directentrylogger/TestDirectReader.java b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/directentrylogger/TestDirectReader.java index 04cd4a0c429..b98f8cf7567 100644 --- a/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/directentrylogger/TestDirectReader.java +++ b/bookkeeper-server/src/test/java/org/apache/bookkeeper/bookie/storage/directentrylogger/TestDirectReader.java @@ -392,8 +392,7 @@ public void testPartialRead() throws Exception { NativeIOImpl nativeIO = new NativeIOImpl() { @Override public long pread(int fd, long buf, long size, long offset) throws NativeIOException { - long read = super.pread(fd, buf, size, offset); - return Math.min(read, Buffer.ALIGNMENT); // force only less than a buffer read + return super.pread(fd, buf, Math.min(size, Buffer.ALIGNMENT), offset); } @Override @@ -445,6 +444,85 @@ buffers, new NativeIOImpl(), Logger.get(TestDirectReader.class))) { } } + @Test + public void testPartialReadProgressesByAlignedBytes() throws Exception { + File ledgerDir = tmpDirs.createNew("partialReadAligned", "logs"); + + writeFileWithPattern(ledgerDir, 1234, 0xbeefcafe, 1, 1 << 20); + + class ShortReadNativeIO extends NativeIOImpl { + int calls; + + @Override + public long pread(int fd, long buf, long size, long offset) throws NativeIOException { + if (calls == 0) { + assertThat(offset, equalTo(0L)); + } else if (calls == 1) { + assertThat(offset, equalTo((long) Buffer.ALIGNMENT * 2)); + } + calls++; + + return super.pread(fd, buf, Math.min(size, Buffer.ALIGNMENT * 2L), offset); + } + } + + ShortReadNativeIO nativeIO = new ShortReadNativeIO(); + try (LogReader reader = new DirectReader(1234, logFilename(ledgerDir, 1234), + ByteBufAllocator.DEFAULT, + nativeIO, Buffer.ALIGNMENT * 4, + 1 << 20, opLogger)) { + ByteBuf bb = reader.readBufferAt(0, Buffer.ALIGNMENT * 4); + try { + for (int block = 0; block < 4; block++) { + for (int i = 0; i < Buffer.ALIGNMENT / Integer.BYTES; i++) { + assertThat(bb.readInt(), equalTo(0xbeefcafe + block)); + } + } + assertThat(bb.readableBytes(), equalTo(0)); + } finally { + bb.release(); + } + } + assertThat(nativeIO.calls, equalTo(2)); + } + + @Test + public void testFailedPartialReadInvalidatesCachedBlock() throws Exception { + File ledgerDir = tmpDirs.createNew("partialReadCache", "logs"); + + writeFileWithPattern(ledgerDir, 1234, 0xbeefcafe, 1, 1 << 20); + + class FailingShortReadNativeIO extends NativeIOImpl { + int calls; + + @Override + public long pread(int fd, long buf, long size, long offset) throws NativeIOException { + calls++; + if (calls == 2) { + return super.pread(fd, buf, Buffer.ALIGNMENT * 2L, offset); + } else if (calls == 3) { + return 0; + } + return super.pread(fd, buf, size, offset); + } + } + + FailingShortReadNativeIO nativeIO = new FailingShortReadNativeIO(); + try (LogReader reader = new DirectReader(1234, logFilename(ledgerDir, 1234), + ByteBufAllocator.DEFAULT, + nativeIO, Buffer.ALIGNMENT * 4, + 1 << 20, opLogger)) { + assertThat(reader.readIntAt(0), equalTo(0xbeefcafe)); + + IOException exception = Assertions.assertThrows( + IOException.class, () -> reader.readIntAt(Buffer.ALIGNMENT * 4L)); + Assertions.assertFalse(exception instanceof EOFException); + + assertThat(reader.readIntAt(0), equalTo(0xbeefcafe)); + } + assertThat(nativeIO.calls, equalTo(4)); + } + @Test public void testLargeEntry() throws Exception { File ledgerDir = tmpDirs.createNew("largeEntries", "logs");