diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java index 075ab08fe5a6..e7c259ce464a 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/MultipartInputStream.java @@ -174,7 +174,9 @@ public synchronized void seek(long pos) throws IOException { } // Reset the previous partStream's position - partStreams.get(prevPartIndex).seek(0); + if (prevPartIndex != partIndex) { + partStreams.get(prevPartIndex).seek(0); + } // Reset all the partStreams above the partIndex. We do this to reset // any previous reads which might have updated the higher part diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java index c15cd338908d..b99d88ad3053 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamBlockInputStream.java @@ -87,7 +87,7 @@ public class StreamBlockInputStream extends BlockExtendedInputStream { private XceiverClientFactory xceiverClientFactory; private XceiverClientGrpc xceiverClient; - private ByteBuffer buffer; + private ReadBuffer readBuffer; private long position = 0; private long requestedLength = 0; private StreamingReader streamingReader; @@ -116,6 +116,8 @@ public StreamBlockInputStream( this.responseDataSize = config.getStreamReadResponseDataSize(); this.readTimeout = config.getStreamReadTimeout(); this.readTimeoutNanos = readTimeout.toNanos(); + + LOG.debug("{}: new StreamBlockInputStream", name); } @Override @@ -136,11 +138,12 @@ public synchronized long getPos() { @Override public synchronized int read() throws IOException { checkOpen(); - if (!dataAvailableToRead(1, true)) { + final boolean preRead = true; + if (!dataAvailableToRead(1, preRead)) { return EOF; } - int value = buffer.get(); - advancePosition(1); + final int value = readBuffer.getByteBuffer().get(); + advancePosition(1, preRead); return value; } @@ -162,12 +165,14 @@ synchronized int readFully(ByteBuffer targetBuf, boolean preRead) throws IOExcep if (!dataAvailableToRead(targetBuf.remaining(), preRead)) { break; } + + final ByteBuffer buffer = readBuffer.getByteBuffer(); int toCopy = Math.min(buffer.remaining(), targetBuf.remaining()); ByteBuffer tmpBuf = buffer.duplicate(); tmpBuf.limit(tmpBuf.position() + toCopy); targetBuf.put(tmpBuf); buffer.position(tmpBuf.position()); - advancePosition(toCopy); + advancePosition(toCopy, preRead); read += toCopy; } return read > 0 ? read : EOF; @@ -177,30 +182,31 @@ private synchronized boolean dataAvailableToRead(int length, boolean preRead) th if (position >= blockLength) { return false; } + while (true) { try { initialize(); - if (bufferHasRemaining()) { - return true; + if (!hasRemaining(readBuffer)) { + readBuffer = streamingReader.read(length, preRead); } - buffer = streamingReader.read(length, preRead); retries = 0; - return bufferHasRemaining(); + return hasRemaining(readBuffer); } catch (IOException ex) { handleExceptions(ex); } } } - private synchronized void advancePosition(long delta) { + private synchronized void advancePosition(long delta, boolean preRead) { + LOG.trace("{}: advance {} -> {}", getName(streamingReader), position, position + delta); position += delta; - if (position >= blockLength && streamingReader != null) { - closeStream(); + if (preRead && position >= blockLength) { + closeReader("advancePosition"); } } - private synchronized boolean bufferHasRemaining() { - return buffer != null && buffer.hasRemaining(); + private static boolean hasRemaining(ReadBuffer read) { + return read != null && read.getByteBuffer().hasRemaining(); } @Override @@ -220,10 +226,42 @@ public synchronized void seek(long pos) throws IOException { if (pos == position) { return; } - LOG.debug("{}: seek {} -> {}", this, position, pos); - closeStream(); + LOG.debug("{}: seek {} -> {}", getName(streamingReader), position, pos); + readBuffer = reuseReadBuffer(readBuffer, pos); position = pos; - requestedLength = pos; + if (readBuffer == null) { + // Only rewind the request high-watermark when the buffered (already requested/served) data cannot be reused; + // otherwise we would re-request data that is still buffered. + requestedLength = pos; + } + } + + static ReadBuffer reuseReadBuffer(ReadBuffer previous, long blockOffset) { + if (previous != null) { + final ByteBuffer buffer = getByteBuffer(previous.getProto(), blockOffset); + if (buffer != null && buffer.hasRemaining()) { + previous.getByteBuffer().position(buffer.position()); + Preconditions.assertSame(buffer.remaining(), previous.getByteBuffer().remaining(), "remaining"); + return previous; + } + } + return null; + } + + static ByteBuffer getByteBuffer(ReadBlockResponseProto proto, long blockOffset) { + final ByteBuffer buffer = proto.getData().asReadOnlyByteBuffer(); + // Adjust buffer position since the server always returns data starting at checksum boundary. + final long protoOffset = proto.getOffset(); + if (blockOffset < protoOffset) { + // This can happen after seek, just drop it for now + // TODO: consider to cache the proto, which will be useful when seeking back. + return null; + } + final long offset = blockOffset - protoOffset; + if (offset > 0) { + buffer.position(Math.toIntExact(Math.min(offset, buffer.limit()))); + } + return buffer; } @Override @@ -238,19 +276,15 @@ public synchronized void unbuffer() { releaseClient(); } - private synchronized void closeStream() { + private synchronized void closeReader(String reason) { + readBuffer = null; if (streamingReader == null) { - buffer = null; return; } final StreamingReader reader = streamingReader; streamingReader = null; - buffer = null; - - if (LOG.isDebugEnabled()) { - LOG.debug("Closing {}", reader); - } + LOG.debug("{} closeReader for {}", getName(reader), reason); reader.onCompleted(); @@ -305,6 +339,7 @@ private synchronized void initialize() throws IOException { try { acquireClient(); final StreamingReader reader = new StreamingReader(); + LOG.debug("{}: new StreamingReader", getName(reader)); xceiverClient.initStreamRead(blockID, reader, failedStreamingDatanodes); streamingReader = reader; } catch (IOException ioe) { @@ -373,7 +408,7 @@ private void recordFailedStreamingDatanode() { protected synchronized void releaseClient() { if (xceiverClientFactory != null && xceiverClient != null) { - closeStream(); + closeReader("releaseClient"); xceiverClientFactory.releaseClientForReadData(xceiverClient, false); xceiverClient = null; } @@ -413,6 +448,35 @@ public Duration getReadTimeout() { return readTimeout; } + private Object getName(StreamingReader reader) { + return reader != null ? reader : name; + } + + static class ReadBuffer { + private final ReadBlockResponseProto proto; + private final ByteBuffer buffer; + + ReadBuffer(ReadBlockResponseProto proto, ByteBuffer buffer) { + this.proto = proto; + this.buffer = buffer; + } + + ReadBlockResponseProto getProto() { + return proto; + } + + ByteBuffer getByteBuffer() { + return buffer; + } + + @Override + public String toString() { + return "ReadBuffer: offset=" + proto.getOffset() + + ", dataSize=" + proto.getData().size() + + ", buffer=" + buffer; + } + } + /** * Implementation of a StreamObserver used to received and buffer streaming GRPC reads. */ @@ -462,61 +526,43 @@ ReadBlockResponseProto poll() throws IOException { } final long elapsedNanos = System.nanoTime() - startTime; - if (elapsedNanos >= readTimeoutNanos) { - setFailedAndThrow(new TimeoutIOException( - "Timed out waiting for response after " + readTimeout)); + if (elapsedNanos >= readTimeoutNanos && !future.isDone()) { + final TimeoutIOException e = new TimeoutIOException( + this + ": Failed to poll a response, timed out " + readTimeout); + if (setFailed(e)) { + throw e; + } return null; } } } - private ByteBuffer read(int length, boolean preRead) throws IOException { + private ReadBuffer read(int length, boolean preRead) throws IOException { checkError(); if (future.isDone()) { // Don't return null while items remain in the queue. onNext() may have delivered items just before // onCompleted() fired. - return responseQueue.isEmpty() ? null : readFromQueue(); + if (responseQueue.isEmpty()) { + return null; + } + } else { + // send gRPC onNext(..) + readBlock(length, preRead); } - readBlock(length, preRead); - + // poll buffer from queue while (true) { - final ByteBuffer buf = readFromQueue(); - if (buf == null) { - return null; // Stream ended + final ReadBlockResponseProto proto = poll(); + if (proto == null) { + return null; } - if (buf.hasRemaining()) { - return buf; + final ByteBuffer buffer = getByteBuffer(proto, getPos()); + final ReadBuffer read = buffer != null ? new ReadBuffer(proto, buffer) : null; + if (hasRemaining(read)) { + LOG.debug("{}: read(length={}, preRead={}) returns {}", name, length, preRead, read); + return read; } - // buf is empty: the server aligned its response to a checksum boundary - // before our current position and all bytes were skipped. Fetch the next - // response, which should start at or after our position. - } - } - - ByteBuffer readFromQueue() throws IOException { - final ReadBlockResponseProto readBlock = poll(); - if (readBlock == null) { - return null; // Stream ended } - // The server always returns data starting from the last checksum boundary. Therefore if the reader position is - // ahead of the position we received from the server, we need to adjust the buffer position accordingly. - final ByteString data = readBlock.getData(); - final ByteBuffer dataBuffer = data.asReadOnlyByteBuffer(); - final long blockOffset = readBlock.getOffset(); - final long pos = getPos(); - if (pos < blockOffset) { - // This should not happen, and if it does, we have a bug. - setFailedAndThrow(new IllegalStateException( - this + ": out of order, position " + pos + " < block offset " + blockOffset)); - } - final long offset = pos - blockOffset; - if (offset > 0) { - dataBuffer.position(Math.toIntExact(Math.min(offset, dataBuffer.limit()))); - } - LOG.debug("{}: return response positon {}, length {} (block offset {}, length {})", - name, pos, dataBuffer.remaining(), blockOffset, data.size()); - return dataBuffer; } private void releaseResources() { @@ -575,12 +621,6 @@ StreamingReadResponse getResponse() { return response.get(); } - private void setFailedAndThrow(T throwable) throws T { - if (setFailed(throwable)) { - throw throwable; - } - } - private boolean setFailed(Throwable throwable) { final boolean completed = future.completeExceptionally(throwable); if (!completed) { @@ -608,9 +648,9 @@ private void setCompleted() { } private void offerToQueue(ReadBlockResponseProto item) { - if (LOG.isDebugEnabled()) { + if (LOG.isTraceEnabled()) { final ContainerProtos.ChecksumData checksumData = item.getChecksumData(); - LOG.debug("{}: enqueue response offset {}, length {}, numChecksums {}, bytesPerChecksum={}", + LOG.trace("{}: enqueue response offset {}, length {}, numChecksums {}, bytesPerChecksum={}", name, item.getOffset(), item.getData().size(), checksumData.getChecksumsList().size(), checksumData.getBytesPerChecksum()); } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java index 97b802e78b1a..2d4ce5095aae 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamBlockInputStream.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.nio.ByteBuffer; +import java.util.Random; import java.util.concurrent.ThreadLocalRandom; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -61,7 +62,7 @@ public class TestStreamBlockInputStream extends InputStreamTests { GenericTestUtils.setLogLevel(LoggerFactory.getLogger("SCMHATransactionMonitor"), Level.ERROR); GenericTestUtils.setLogLevel(GrpcXceiverService.class, Level.ERROR); -// GenericTestUtils.setLogLevel(LoggerFactory.getLogger(StreamBlockInputStream.class), Level.TRACE); +// GenericTestUtils.setLogLevel(StreamBlockInputStream.class, Level.DEBUG); // GenericTestUtils.setLogLevel(LoggerFactory.getLogger(XceiverClientGrpc.class), Level.TRACE); } @@ -81,7 +82,7 @@ void testReadKey() throws Exception { OzoneConfiguration conf = cluster.getConf(); runTestReadKey(DATA_LENGTH, false, conf); - for (int i = 0; i < 3; i++) { + for (int i = 0; i < 2; i++) { final int keyLength = DATA_LENGTH + ThreadLocalRandom.current().nextInt(DATA_LENGTH); runTestReadKey(keyLength, true, conf); } @@ -185,13 +186,25 @@ void assertData(int pos, int length, ByteBuffer buffer) { } @Test - void testAll() throws Exception { + void testAllWithPreRead() throws Exception { + runTestAll(true); + } + + @Test + void testAllWithoutPreRead() throws Exception { + runTestAll(false); + } + + void runTestAll(boolean preRead) throws Exception { try (MiniOzoneCluster cluster = newCluster()) { cluster.waitForClusterToBeReady(); OzoneConfiguration conf = cluster.getConf(); OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); clientConfig.setStreamReadBlock(true); + if (!preRead) { + clientConfig.setStreamReadPreReadSize(0); + } OzoneConfiguration copy = new OzoneConfiguration(conf); copy.setFromObject(clientConfig); String keyName = getNewKeyName(); @@ -250,15 +263,38 @@ private void testReadKeyFully(String key) throws Exception { } } + void assertSeekRead(KeyInputStream in, int position) throws IOException { + in.seek(position); + int b = in.read(); + assertEquals(inputData[position], (byte) b, "Read data is not same as written data at index " + position); + } + + private void runTestSeek(KeyInputStream in, int seekSize, Random random) throws IOException { + LOG.info("runTestSeek: seekSize={}", seekSize); + for (int i = 0; i < 100; i++) { + int position = random.nextInt(seekSize); + assertSeekRead(in, position); + } + + for (int position = 0; position < DATA_LENGTH; position += random.nextInt(seekSize)) { + assertSeekRead(in, position); + } + + for (int position = DATA_LENGTH - 1; position >= 0; position -= random.nextInt(seekSize)) { + assertSeekRead(in, position); + } + assertSeekRead(in, 0); + } + private void testSeek(String key) throws IOException { - java.util.Random random = new java.util.Random(); + final Random random = new Random(); try (KeyInputStream keyInputStream = bucket.getKeyInputStream(key)) { - for (int i = 0; i < 100; i++) { - int position = random.nextInt(DATA_LENGTH); - keyInputStream.seek(position); - int b = keyInputStream.read(); - assertEquals(inputData[position], (byte) b, "Read data is not same as written data at index " + position); - } + runTestSeek(keyInputStream, CHUNK_SIZE / 8, random); + runTestSeek(keyInputStream, CHUNK_SIZE, random); + runTestSeek(keyInputStream, BLOCK_SIZE, random); + runTestSeek(keyInputStream, DATA_LENGTH, random); + + // error cases StreamBlockInputStream blockStream = (StreamBlockInputStream) keyInputStream.getPartStreams().get(0); long length = blockStream.getLength(); blockStream.seek(10); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java index ee56eb38825e..1838b0d928bb 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/read/TestStreamRead.java @@ -74,6 +74,7 @@ public class TestStreamRead { GenericTestUtils.setLogLevel(LoggerFactory.getLogger("ExpiredContainerReplicaOpScrubber"), Level.ERROR); GenericTestUtils.setLogLevel(LoggerFactory.getLogger("SCMHATransactionMonitor"), Level.ERROR); GenericTestUtils.setLogLevel(LoggerFactory.getLogger(CodecBuffer.class), Level.ERROR); +// GenericTestUtils.setLogLevel(StreamBlockInputStream.class, Level.DEBUG); } static final int CHUNK_SIZE = 1 << 20; // 1MB