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:
The close, write failure, and receiver failure paths now remove pending
requests under the client lock. Future completion and metrics updates happen
after releasing the lock because completing a `CompletableFuture` may
synchronously invoke callbacks, including close().
I used `remove(key, entry)` instead of snapshot followed by clear(), since
timeout and response removal intentionally remain lock-free. Conditional
removal gives each request a single terminal owner while ensuring failed
entries are removed from the map.
--
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]
