echonesis commented on code in PR #11225:
URL: https://github.com/apache/ozone/pull/11225#discussion_r4001891085
##########
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());
Review Comment:
I will make one ordering adjustment: while holding the client lock, the code
will conditionally remove and collect all pending entries; after releasing the
lock, it will fail their futures and update metrics. Completing a
`CompletableFuture` may synchronously run dependent callbacks, including
callbacks that call `close()`, so that work should remain outside the client
lock.
A plain snapshot followed by clear() is not sufficient because response and
timeout removal intentionally remain lock-free so they can proceed while a
socket write is blocked. Using `remove(key, entry)` establishes exactly one
owner for each request while still leaving the map empty after the terminal
transition.
--
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]
