ss77892 commented on code in PR #11080:
URL: https://github.com/apache/ozone/pull/11080#discussion_r4168527932
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientGrpc.java:
##########
@@ -603,52 +611,73 @@ public void initStreamRead(BlockID blockID,
StreamingReaderSpi streamObserver) t
*/
public void initStreamRead(BlockID blockID, StreamingReaderSpi
streamObserver,
Set<DatanodeID> excludedDatanodes) throws IOException {
- final List<DatanodeDetails> datanodeList = sortDatanodes(null,
ContainerProtos.Type.ReadBlock);
- IOException lastException = null;
- for (DatanodeDetails dn : datanodeList) {
- if (excludedDatanodes.contains(dn.getID())) {
- LOG.debug("Skipping excluded datanode {} (uuid={}) for initStreamRead
{}",
- dn, dn.getUuidString(), blockID.getContainerBlockID());
- continue;
- }
- try {
- checkOpen(dn);
- semaphore.acquire();
- XceiverClientProtocolServiceStub stub =
dnChannelInfoMap.get(dn.getID()).getStub();
- if (stub == null) {
- throw new IOException("Failed to get gRPC stub for DataNode: " + dn);
+ acquireStreamPermit(blockID);
+ boolean started = false;
+ try {
+ final List<DatanodeDetails> datanodeList = sortDatanodes(null,
ContainerProtos.Type.ReadBlock);
+ IOException lastException = null;
+ for (DatanodeDetails dn : datanodeList) {
+ if (excludedDatanodes.contains(dn.getID())) {
+ LOG.debug("Skipping excluded datanode {} (uuid={}) for
initStreamRead {}",
+ dn, dn.getUuidString(), blockID.getContainerBlockID());
+ continue;
}
- LOG.debug("initStreamRead {} on datanode {}",
blockID.getContainerBlockID(), dn);
- StreamObserver<ContainerCommandRequestProto> requestObserver = stub
- .withDeadlineAfter(timeout, TimeUnit.SECONDS)
- .send(streamObserver);
- streamObserver.setStreamingReadResponse(new StreamingReadResponse(dn,
- (ClientCallStreamObserver<ContainerCommandRequestProto>)
requestObserver));
- return;
- } catch (IOException e) {
- LOG.error("Failed to start streaming read to DataNode {}", dn, e);
- semaphore.release();
- lastException = e;
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- throw new IOException("Interrupted initStreamRead to " + dn + " for "
+ blockID, e);
+ try {
+ checkOpen(dn);
+ XceiverClientProtocolServiceStub stub =
dnChannelInfoMap.get(dn.getID()).getStub();
+ if (stub == null) {
+ throw new IOException("Failed to get gRPC stub for DataNode: " +
dn);
+ }
+ LOG.debug("initStreamRead {} on datanode {}",
blockID.getContainerBlockID(), dn);
+ // No deadline: it would bound the entire long-lived streaming call.
Per-request timeliness is
+ // enforced by streamReadTimeout in streamRead() and
StreamingReader.poll().
+ StreamObserver<ContainerCommandRequestProto> requestObserver =
stub.send(streamObserver);
Review Comment:
> I may be missing an existing safeguard, but could long-lived streams keep
server-side files open for an extended period? The current limits don’t seem to
apply across the whole datanode. Is there another server-side limit or cleanup
mechanism?
There was no safeguard: with the deadline gone, an idle stream would keep
its block file open until the client closed the stream or the connection
dropped. maxConnectionIdle(15m) doesn't help, because it only applies to
connections with no active calls.
The update adds hdds.datanode.stream.read.file.idle.timeout (default 1m).
GrpcXceiverService closes the block file of any stream that has been idle
longer than that. The gRPC stream stays open and the next ReadBlock reopens the
file. A lock around each request keeps the file from being closed under an
in-flight read. This bounds descriptors held by idle streams, which is the case
the deadline removal creates. Tests are in TestGrpcXceiverService.
There is still no datanode-wide cap on concurrent streaming reads, but that
is not new: before this change, active streams were limited only by the
client's 30s deadline, which killed healthy streams too. I'd like to track a
datanode-wide limit on concurrent streams separately, so this PR stays focused
on the deadline regression.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
