Gargi-jais11 commented on code in PR #11225:
URL: https://github.com/apache/ozone/pull/11225#discussion_r3988185528
##########
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 generic catch (Throwable e) block,
if entry != null you are calling the
entry.getFuture().completeExceptionally(e) and then pending.forEach(i ->
i.fail(e)). If entry is still in sentRequests, it gets failed twice.
CompletableFuture ignores the second completion, so this is harmless but
redundant.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java:
##########
@@ -132,46 +134,95 @@ public XceiverClientShortCircuit(Pipeline pipeline,
ConfigurationSource config,
*/
@Override
public void connect() throws IOException {
- // Even the in & out stream has returned EOFException,
domainSocket.isOpen() is still true.
- if (domainSocket != null && domainSocket.isOpen() &&
isDomainSocketOpen.get()) {
- return;
+ lock.lock();
+ try {
+ if (closed) {
+ throw new IOException("DomainSocket is closed.");
+ }
+ if (domainSocket != null) {
+ checkOpen();
+ return;
+ }
+ boolean connected = false;
+ try {
+ domainSocket = domainSocketFactory.createSocket(readTimeoutMs,
writeTimeoutMs, dnAddr);
+ if (domainSocket == null) {
+ throw new IOException("DomainSocket is not available for " + dn);
+ }
+ prefix = XceiverClientShortCircuit.class.getSimpleName() + "-" +
domainSocket;
+ timer = new Timer(prefix + "-Timer");
+ isDomainSocketOpen.set(true);
+ readDaemon.start();
+ connected = true;
+ LOG.info("{} is started", prefix);
+ } finally {
+ if (!connected) {
+ closed = true;
+ isDomainSocketOpen.set(false);
+ if (timer != null) {
+ timer.cancel();
+ }
+ if (domainSocket != null) {
+ try {
+ domainSocket.close();
+ } catch (IOException e) {
+ LOG.warn("Failed to close domain socket for datanode {}", dn, e);
+ }
+ }
+ }
+ }
+ } finally {
+ lock.unlock();
}
- domainSocket = domainSocketFactory.createSocket(readTimeoutMs,
writeTimeoutMs, dnAddr);
- isDomainSocketOpen.set(true);
- prefix = XceiverClientShortCircuit.class.getSimpleName() + "-" +
domainSocket.toString();
- timer = new Timer(prefix + "-Timer");
- readDaemon.start();
- LOG.info("{} is started", prefix);
}
/**
* Close the DomainSocket.
*/
@Override
- public synchronized void close() {
- closed = true;
- timer.cancel();
- if (domainSocket != null) {
- try {
+ public void close() {
+ final List<RequestEntry> pending;
+ lock.lock();
+ try {
+ if (!closed) {
+ closed = true;
isDomainSocketOpen.set(false);
- domainSocket.close();
- LOG.info("{} is closed for {} with {} requests sent and {} responses
received",
- domainSocket.toString(), dn, requestSent, responseReceived);
- } catch (IOException e) {
- LOG.warn("Failed to close domain socket for datanode {}", dn, e);
+ if (timer != null) {
+ timer.cancel();
+ }
+ if (domainSocket != null) {
+ try {
+ domainSocket.close();
+ LOG.info("{} is closed for {} with {} requests sent and {}
responses received",
+ domainSocket, dn, requestSent, responseReceived);
+ } catch (IOException e) {
+ LOG.warn("Failed to close domain socket for datanode {}", dn, e);
+ }
+ }
+ readDaemon.interrupt();
}
+ pending = new ArrayList<>(sentRequests.values());
+ } finally {
+ lock.unlock();
}
- readDaemon.interrupt();
- try {
- readDaemon.join();
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
+ pending.forEach(entry -> entry.fail(new ClosedChannelException()));
Review Comment:
here when the close fails, we are not tracking any metrics failures.
##########
hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java:
##########
@@ -402,27 +464,40 @@ public void run() {
// send request body
request.writeDelimitedTo(dataOut);
dataOut.flush();
+ } catch (IOException e) {
+ isDomainSocketOpen.set(false);
+ failure = e;
+ pending = new ArrayList<>(sentRequests.values());
} finally {
- lock.unlock();
entry.setSentTimeNs();
requestSent++;
}
- } catch (IOException e) {
- LOG.error("Failed to send command {}", request, e);
- entry.getFuture().completeExceptionally(e);
+ } finally {
+ lock.unlock();
+ }
+ if (failure != null) {
+ LOG.error("Failed to send command {}", request, failure);
+ for (RequestEntry requestEntry : pending) {
+ requestEntry.fail(failure);
+ }
metrics.decrPendingContainerOpsMetrics(request.getCmdType());
metrics.addContainerOpsLatency(request.getCmdType(), System.nanoTime() -
entry.getCreateTimeNs());
Review Comment:
In sendRequest(), when the write faills we are currently fail all entries in
pending, but `metrics.decrPendingContainerOpsMetrics()` and
`addContainerOpsLatency()` are only called once for the current request, so
other in-flight requests will leave infleted pending metrics.
--
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]
