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:
The current receiver entry had already been removed from sentRequests, so it
was not normally part of the snapshot taken later in the same catch block.
However, snapshot-based failure handling could still overlap with another close
or failure path.
I replaced the snapshots with explicit removal ownership. The receiver
accounts for the entry it removed, while the common failure path handles only
entries it can conditionally remove from sentRequests. This avoids duplicate
future completion and metrics accounting across 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]
