echonesis commented on code in PR #11225:
URL: https://github.com/apache/ozone/pull/11225#discussion_r4001887215
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java:
##########
@@ -511,34 +596,46 @@ public void run() {
}
long currentTime = System.nanoTime();
long endToEndCost = currentTime - entry.getCreateTimeNs();
- long sentCost = entry.getSentTimeNs() - entry.getCreateTimeNs();
+ long sentCost = sentTimeNs - entry.getCreateTimeNs();
long receiveCost = processStartTime - receiveStartTime;
long processCost = currentTime - processStartTime;
if (LOG.isDebugEnabled()) {
LOG.debug("Executed command {} {}:{} on datanode {}, end-to-end {}
ns, sent {} ns, receive {} ns, " +
"process {} ns", type,
entry.getRequest().getClientId().toStringUtf8(),
entry.getRequest().getCallId(), dn, endToEndCost, sentCost,
receiveCost, processCost);
}
- responseReceived++;
+ lock.lock();
+ try {
+ responseReceived++;
+ } finally {
+ lock.unlock();
+ }
metrics.decrPendingContainerOpsMetrics(type);
metrics.addContainerOpsLatency(type, endToEndCost);
- } catch (SocketTimeoutException | EOFException |
ClosedChannelException e) {
- isDomainSocketOpen.set(false);
- LOG.info("{} receiveResponseTask is closed after send {} requests
and received {} responses, due to {}",
- domainSocket.toString(), requestSent, responseReceived,
e.getClass().getName(), e);
- // fail all requests pending responses
- sentRequests.values().forEach(i -> i.fail(e));
} catch (Throwable e) {
- isDomainSocketOpen.set(false);
- LOG.error("{} failed after send {} requests and received {}
responses",
- domainSocket.toString(), requestSent, responseReceived, e);
+ final List<RequestEntry> pending;
+ lock.lock();
+ try {
+ isDomainSocketOpen.set(false);
+ if (e instanceof SocketTimeoutException || e instanceof
EOFException
+ || e instanceof ClosedChannelException) {
+ LOG.info("{} receiveResponseTask is closed after send {}
requests and received {} responses, due to {}",
+ socket, requestSent, responseReceived,
e.getClass().getName(), e);
+ } else {
+ LOG.error("{} failed after send {} requests and received {}
responses",
+ socket, requestSent, responseReceived, e);
+ }
+ pending = new ArrayList<>(sentRequests.values());
+ } finally {
+ lock.unlock();
+ }
if (entry != null) {
entry.getFuture().completeExceptionally(e);
Review Comment:
In the current flow, entry can only become non-null through
`sentRequests.remove(...)`, so it is already absent from the pending snapshot
created later in the same catch block.
There is still a possible overlap with a stale snapshot created by another
close or failure path. I will replace snapshot-based failure handling with
explicit ownership: the receiver handles the entry it already removed, while
the common removal helper handles only entries it can conditionally remove from
`sentRequests`. This prevents duplicate completion and metrics accounting
across all competing terminal paths.
--
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]
