echonesis commented on code in PR #11225:
URL: https://github.com/apache/ozone/pull/11225#discussion_r4001882618
##########
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:
Thanks for mentioning.
`close()` now removes and collects pending entries under the client lock,
then fails and accounts for each request outside the lock. Conditional removal
ensures that repeated or concurrent `close()` calls cannot account for the same
request more than once.
--
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]
