From 36946923d6210e0ddb7ff2a4a54512b53a935a40 Mon Sep 17 00:00:00 2001 From: echonesis Date: Wed, 9 Sep 2026 15:17:39 +0800 Subject: [PATCH 1/3] HDDS-16335. Change XceiverClientShortCircuit to support concurrent access --- .../hdds/scm/XceiverClientShortCircuit.java | 291 ++++++++++++------ .../hdds/scm/TestXceiverClientManagerSC.java | 279 +++++++++++++++++ 2 files changed, 482 insertions(+), 88 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java index a69bc1453c19..f527182de5e0 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java @@ -34,6 +34,7 @@ import java.net.InetSocketAddress; import java.net.SocketTimeoutException; import java.nio.channels.ClosedChannelException; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; @@ -94,6 +95,7 @@ public class XceiverClientShortCircuit extends XceiverClientSpi { private final DomainSocketFactory domainSocketFactory; private DomainSocket domainSocket; private final AtomicBoolean isDomainSocketOpen = new AtomicBoolean(false); + // Protects connection state, counters, and RequestEntry.sentTimeNs. private final Lock lock = new ReentrantLock(); private final int bufferSize; private final ByteString clientId = ByteString.copyFrom(UUID.randomUUID().toString().getBytes(UTF_8)); @@ -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 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())); + if (Thread.currentThread() != readDaemon) { + try { + readDaemon.join(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } } @Override public boolean isClosed() { - return closed; + lock.lock(); + try { + return closed; + } finally { + lock.unlock(); + } } @Override @@ -295,7 +346,12 @@ private XceiverClientReply sendCommandWithTraceID( Objects.requireNonNull(ioException); String message = "Failed to execute command {}"; if (LOG.isDebugEnabled()) { - LOG.debug(message + " on the datanode {} {}.", request, dn, domainSocket, ioException); + lock.lock(); + try { + LOG.debug(message + " on the datanode {} {}.", request, dn, domainSocket, ioException); + } finally { + lock.unlock(); + } } throw ioException; } @@ -305,10 +361,16 @@ private XceiverClientReply sendCommandWithTraceID( @VisibleForTesting public XceiverClientReply sendCommandInternal(ContainerCommandRequestProto request) throws IOException, InterruptedException { - checkOpen(); final CompletableFuture replyFuture = new CompletableFuture<>(); - RequestEntry entry = new RequestEntry(request, replyFuture); + final RequestKey key = new RequestKey(request.getClientId(), request.getCallId()); + TimerTask task = new TimerTask() { + @Override + public void run() { + requestTimeout(key); + } + }; + RequestEntry entry = new RequestEntry(request, replyFuture, task); sendRequest(entry); return new XceiverClientReply(replyFuture); } @@ -320,13 +382,18 @@ public XceiverClientReply sendCommandAsync( throw new UnsupportedOperationException("Operation Not supported for " + DomainSocketFactory.FEATURE + " client"); } - public synchronized void checkOpen() throws IOException { - if (closed) { - throw new IOException("DomainSocket is not connected."); - } - - if (!isDomainSocketOpen.get()) { - throw new IOException(domainSocket.toString() + " is not open."); + public void checkOpen() throws IOException { + lock.lock(); + try { + if (closed || domainSocket == null) { + throw new IOException("DomainSocket is not connected."); + } + // isOpen() may remain true after EOF, so also check the receiver's state. + if (!domainSocket.isOpen() || !isDomainSocketOpen.get()) { + throw new IOException(domainSocket + " is not open."); + } + } finally { + lock.unlock(); } } @@ -360,7 +427,13 @@ public static Logger getLogger() { } void requestTimeout(RequestKey requestKey) { - final RequestEntry entry = sentRequests.remove(requestKey); + final RequestEntry entry; + lock.lock(); + try { + entry = sentRequests.remove(requestKey); + } finally { + lock.unlock(); + } if (entry != null) { LOG.warn("Timeout to receive response for command {}", entry.getRequest()); ContainerProtos.Type type = entry.getRequest().getCmdType(); @@ -369,30 +442,25 @@ void requestTimeout(RequestKey requestKey) { } } - void sendRequest(RequestEntry entry) { + void sendRequest(RequestEntry entry) throws IOException { ContainerCommandRequestProto request = entry.getRequest(); + IOException failure = null; + List pending = null; + lock.lock(); try { + checkOpen(); final RequestKey key = new RequestKey(request.getClientId(), request.getCallId()); - TimerTask task = new TimerTask() { - @Override - public void run() { - requestTimeout(key); - } - }; - entry.setTimerTask(task); - timer.schedule(task, readTimeoutMs); sentRequests.put(key, entry); ContainerProtos.Type type = request.getCmdType(); metrics.incrPendingContainerOpsMetrics(type); - byte[] bytes = request.toByteArray(); - if (bytes.length != request.getSerializedSize()) { - throw new IOException("Serialized request " + request.getCmdType() - + " size mismatch, byte array size " + bytes.length + - ", serialized size " + request.getSerializedSize()); - } - - lock.lock(); + timer.schedule(entry.getTimerTask(), readTimeoutMs); try { + byte[] bytes = request.toByteArray(); + if (bytes.length != request.getSerializedSize()) { + throw new IOException("Serialized request " + request.getCmdType() + + " size mismatch, byte array size " + bytes.length + + ", serialized size " + request.getSerializedSize()); + } DataOutputStream dataOut = new DataOutputStream(new BufferedOutputStream(domainSocket.getOutputStream(), bufferSize)); // send version number @@ -402,14 +470,22 @@ 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()); } @@ -417,12 +493,17 @@ public void run() { @Override public String toString() { - final StringBuilder b = - new StringBuilder(getClass().getSimpleName()) - .append('[').append(" DomainSocket: ").append(domainSocket.toString()) - .append(" Pipeline: ").append(pipeline.toString()) - .append(" ]"); - return b.toString(); + lock.lock(); + try { + final StringBuilder b = + new StringBuilder(getClass().getSimpleName()) + .append('[').append(" DomainSocket: ").append(domainSocket) + .append(" Pipeline: ").append(pipeline.toString()) + .append(" ]"); + return b.toString(); + } finally { + lock.unlock(); + } } /** @@ -431,12 +512,29 @@ public String toString() { public class ReceiveResponseTask implements Runnable { @Override public void run() { - long timerTaskCancelledCount = 0; - do { + final DomainSocket socket; + final Timer responseTimer; + lock.lock(); + try { + socket = domainSocket; + responseTimer = timer; Thread.currentThread().setName(prefix + "-ReceiveResponse"); + } finally { + lock.unlock(); + } + long timerTaskCancelledCount = 0; + while (true) { + lock.lock(); + try { + if (!isDomainSocketOpen.get()) { + return; + } + } finally { + lock.unlock(); + } RequestEntry entry = null; try { - DataInputStream dataIn = new DataInputStream(domainSocket.getInputStream()); + DataInputStream dataIn = new DataInputStream(socket.getInputStream()); final short version = dataIn.readShort(); if (version != DATA_TRANSFER_VERSION) { throw new IOException("Version Mismatch (Expected: " + @@ -450,7 +548,14 @@ public void run() { if (LOG.isDebugEnabled()) { LOG.debug("received response {} callId {}", type, responseProto.getCallId()); } - entry = sentRequests.remove(new RequestKey(responseProto.getClientId(), responseProto.getCallId())); + final long sentTimeNs; + lock.lock(); + try { + entry = sentRequests.remove(new RequestKey(responseProto.getClientId(), responseProto.getCallId())); + sentTimeNs = entry == null ? 0 : entry.getSentTimeNs(); + } finally { + lock.unlock(); + } if (entry == null) { // This could be two cases // 1. there is bug in the code @@ -464,7 +569,7 @@ public void run() { timerTaskCancelledCount++; // purge timer every 1000 cancels if (timerTaskCancelledCount == 1000) { - timer.purge(); + responseTimer.purge(); timerTaskCancelledCount = 0; } } @@ -481,7 +586,7 @@ public void run() { // read FS from domainSocket FileInputStream[] fis = new FileInputStream[1]; byte[] buf = new byte[1]; - int ret = domainSocket.recvFileInputStreams(fis, buf, 0, buf.length); + int ret = socket.recvFileInputStreams(fis, buf, 0, buf.length); if (ret == -1) { throw new IOException("failed to get a file descriptor from datanode " + dn + " for peer is shutdown."); @@ -511,7 +616,7 @@ 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()) { @@ -519,26 +624,38 @@ public void run() { "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 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); } - sentRequests.values().forEach(i -> i.fail(e)); + pending.forEach(i -> i.fail(e)); break; } - } while (isDomainSocketOpen.get()); + } } } @@ -583,13 +700,15 @@ static class RequestEntry { private final ContainerCommandRequestProto request; private final CompletableFuture future; private final long createTimeNs; + // Accessed under the enclosing client's lock. private long sentTimeNs; - private TimerTask timerTask; + private final TimerTask timerTask; RequestEntry(ContainerCommandRequestProto requestProto, - CompletableFuture future) { + CompletableFuture future, TimerTask timerTask) { this.request = requestProto; this.future = future; + this.timerTask = timerTask; this.createTimeNs = System.nanoTime(); } @@ -613,10 +732,6 @@ public void setSentTimeNs() { sentTimeNs = System.nanoTime(); } - public void setTimerTask(TimerTask task) { - timerTask = task; - } - public TimerTask getTimerTask() { return timerTask; } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java index ff6e49b57ad5..4ed611b9132d 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java @@ -18,25 +18,55 @@ package org.apache.hadoop.hdds.scm; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import java.io.File; import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; import org.apache.hadoop.hdds.scm.XceiverClientManager.ScmClientConfig; import org.apache.hadoop.hdds.scm.container.common.helpers.ContainerWithPipeline; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolClientSideTranslatorPB; +import org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls; +import org.apache.hadoop.hdds.scm.storage.DomainSocketFactory; import org.apache.hadoop.io.IOUtils; import org.apache.hadoop.net.unix.DomainSocket; import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.container.common.SCMTestUtils; +import org.apache.ozone.test.GenericTestUtils; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.MockedStatic; /** * Test for short-circuit enabled XceiverClientManager. @@ -51,6 +81,7 @@ public class TestXceiverClientManagerSC { private static OzoneConfiguration config; private static MiniOzoneCluster cluster; + private static ContainerWithPipeline echoContainer; private static StorageContainerLocationProtocolClientSideTranslatorPB storageContainerLocationClient; @TempDir @@ -70,6 +101,20 @@ public static void init() throws Exception { cluster.waitForClusterToBeReady(); storageContainerLocationClient = cluster .getStorageContainerLocationClient(); + echoContainer = storageContainerLocationClient.allocateContainer(HddsProtos.ReplicationType.RATIS, + HddsProtos.ReplicationFactor.THREE, OzoneConsts.OZONE); + try (XceiverClientManager clientManager = new XceiverClientManager(config, + config.getObject(ScmClientConfig.class), null)) { + XceiverClientSpi client = clientManager.acquireClient(echoContainer.getPipeline()); + try { + ContainerProtocolCalls.createContainer(client, echoContainer.getContainerInfo().getContainerID(), null); + } finally { + clientManager.releaseClient(client, false); + } + } + GenericTestUtils.waitFor(() -> cluster.getHddsDatanodes().stream().allMatch(dn -> dn.getDatanodeStateMachine() + .getContainer().getContainerSet().getContainer(echoContainer.getContainerInfo().getContainerID()) != null), + 100, 30000); } @AfterAll @@ -106,4 +151,238 @@ public void testAllocateShortCircuitClient() throws IOException { assertTrue(client3 instanceof XceiverClientGrpc); } } + + @Test + public void testConcurrentConnectAndRequests() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + ExecutorService executor = Executors.newFixedThreadPool(8); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + CountDownLatch start = new CountDownLatch(1); + List> connections = new ArrayList<>(); + for (int i = 0; i < 8; i++) { + connections.add(executor.submit(() -> { + assertTrue(start.await(30, TimeUnit.SECONDS)); + client.connect(); + return null; + })); + } + start.countDown(); + for (Future connection : connections) { + connection.get(30, TimeUnit.SECONDS); + } + client.connect(); + assertFalse(client.isClosed()); + client.checkOpen(); + assertNotNull(client.toString()); + + CountDownLatch send = new CountDownLatch(1); + List> requests = new ArrayList<>(); + for (int i = 0; i < 32; i++) { + ContainerCommandRequestProto request = echoRequest(client, 0); + requests.add(executor.submit(() -> { + assertTrue(send.await(30, TimeUnit.SECONDS)); + assertEchoResponse(request, client.sendCommand(request)); + return null; + })); + } + send.countDown(); + for (Future request : requests) { + request.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + + @Test + public void testMultiplePendingRequestsAndCloseFromResponse() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + client.connect(); + List requests = new ArrayList<>(); + List> responses = new ArrayList<>(); + for (int i = 0; i < 8; i++) { + ContainerCommandRequestProto request = echoRequest(client, i == 0 ? 1000 : 0); + requests.add(request); + responses.add(client.sendCommandInternal(request).getResponse()); + } + assertFalse(responses.get(0).isDone()); + for (int i = 0; i < requests.size(); i++) { + assertEchoResponse(requests.get(i), responses.get(i).get(30, TimeUnit.SECONDS)); + } + client.sendCommandInternal(echoRequest(client, 1000)).getResponse() + .thenRun(client::close).get(30, TimeUnit.SECONDS); + assertTrue(client.isClosed()); + } + } + + @Test + public void testConcurrentCloseWithPendingRequests() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + ExecutorService executor = Executors.newFixedThreadPool(8); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + client.connect(); + CompletableFuture pending = + client.sendCommandInternal(echoRequest(client, 1000)).getResponse(); + CountDownLatch start = new CountDownLatch(1); + List> closes = new ArrayList<>(); + for (int i = 0; i < 8; i++) { + closes.add(executor.submit(() -> { + assertTrue(start.await(30, TimeUnit.SECONDS)); + client.close(); + return null; + })); + } + start.countDown(); + for (Future close : closes) { + close.get(30, TimeUnit.SECONDS); + } + assertThrows(ExecutionException.class, () -> pending.get(30, TimeUnit.SECONDS)); + assertTrue(client.isClosed()); + assertThrows(IOException.class, client::connect); + assertThrows(IOException.class, client::checkOpen); + assertThrows(IOException.class, () -> client.sendCommandInternal(echoRequest(client, 0))); + assertNotNull(client.toString()); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + + @Test + public void testConnectAndSendRacingWithClose() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + ExecutorService executor = Executors.newFixedThreadPool(3); + try { + for (int i = 0; i < 10; i++) { + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + if (i % 2 == 0) { + client.connect(); + } + CountDownLatch start = new CountDownLatch(1); + Future connect = executor.submit(() -> { + assertTrue(start.await(30, TimeUnit.SECONDS)); + try { + client.connect(); + } catch (IOException expected) { + assertTrue(client.isClosed()); + } + return null; + }); + Future send = executor.submit(() -> { + assertTrue(start.await(30, TimeUnit.SECONDS)); + ContainerCommandRequestProto request = echoRequest(client, 0); + try { + assertEchoResponse(request, client.sendCommandInternal(request).getResponse().get(30, TimeUnit.SECONDS)); + } catch (IOException expected) { + // The connection may not have opened yet, or close may have won the lock. + } catch (ExecutionException expected) { + assertTrue(expected.getCause() instanceof IOException); + } + return null; + }); + Future close = executor.submit(() -> { + assertTrue(start.await(30, TimeUnit.SECONDS)); + client.close(); + return null; + }); + start.countDown(); + connect.get(30, TimeUnit.SECONDS); + send.get(30, TimeUnit.SECONDS); + close.get(30, TimeUnit.SECONDS); + assertTrue(client.isClosed()); + assertThrows(IOException.class, client::connect); + } + } + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + + @Test + public void testCloseBeforeConnect() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + assertFalse(client.isClosed()); + assertNotNull(client.toString()); + assertThrows(IOException.class, client::checkOpen); + assertThrows(IOException.class, () -> client.sendCommandInternal(echoRequest(client, 0))); + client.close(); + client.close(); + assertTrue(client.isClosed()); + assertThrows(IOException.class, client::connect); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testConnectFailureCannotReconnect(boolean throwException) throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + DomainSocketFactory factory = mock(DomainSocketFactory.class); + if (throwException) { + when(factory.createSocket(anyInt(), anyInt(), any())).thenThrow(new IOException("Connection failed")); + } + try (MockedStatic mocked = mockStatic(DomainSocketFactory.class)) { + mocked.when(() -> DomainSocketFactory.getInstance(config)).thenReturn(factory); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + assertThrows(IOException.class, client::connect); + assertTrue(client.isClosed()); + assertNotNull(client.toString()); + assertThrows(IOException.class, client::connect); + assertThrows(IOException.class, client::checkOpen); + verify(factory, times(1)).createSocket(anyInt(), anyInt(), any()); + } + } + } + + @Test + public void testReceiverFailureCannotReconnect() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + OzoneConfiguration timeoutConfig = new OzoneConfiguration(config); + timeoutConfig.setTimeDuration(OzoneConfigKeys.OZONE_CLIENT_READ_TIMEOUT, 200, TimeUnit.MILLISECONDS); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, timeoutConfig, pipeline.getClosestNode())) { + client.connect(); + GenericTestUtils.waitFor(() -> { + try { + client.checkOpen(); + return false; + } catch (IOException expected) { + return true; + } + }, 50, 30000); + assertFalse(client.isClosed()); + assertNotNull(client.toString()); + assertThrows(IOException.class, client::connect); + assertThrows(IOException.class, () -> client.sendCommandInternal(echoRequest(client, 0))); + } + } + + private static ContainerCommandRequestProto echoRequest(XceiverClientShortCircuit client, int sleepTimeMs) { + return ContainerCommandRequestProto.newBuilder() + .setCmdType(ContainerProtos.Type.Echo) + .setContainerID(echoContainer.getContainerInfo().getContainerID()) + .setDatanodeUuid(client.getDn().getUuidString()) + .setClientId(client.getClientId()) + .setCallId(client.getCallId()) + .setEcho(ContainerProtos.EchoRequestProto.newBuilder().setReadOnly(true).setSleepTimeMs(sleepTimeMs) + .setPayloadSizeResp(1024)) + .build(); + } + + private static void assertEchoResponse(ContainerCommandRequestProto request, ContainerCommandResponseProto response) { + assertEquals(ContainerProtos.Result.SUCCESS, response.getResult()); + assertEquals(request.getClientId(), response.getClientId()); + assertEquals(request.getCallId(), response.getCallId()); + assertEquals(1024, response.getEcho().getPayload().size()); + } + } From 871db07ddb52674e1fb8d21f781ebfb2e7e79643 Mon Sep 17 00:00:00 2001 From: echonesis Date: Thu, 10 Sep 2026 13:46:11 +0800 Subject: [PATCH 2/3] fix: allow responses and timeouts during blocked short-circuit writes --- .../hdds/scm/XceiverClientShortCircuit.java | 38 +--- .../hdds/scm/TestXceiverClientManagerSC.java | 176 ++++++++++++++---- 2 files changed, 151 insertions(+), 63 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java index f527182de5e0..80fb8f0afc29 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java @@ -95,7 +95,7 @@ public class XceiverClientShortCircuit extends XceiverClientSpi { private final DomainSocketFactory domainSocketFactory; private DomainSocket domainSocket; private final AtomicBoolean isDomainSocketOpen = new AtomicBoolean(false); - // Protects connection state, counters, and RequestEntry.sentTimeNs. + // Protects connection state and counters. private final Lock lock = new ReentrantLock(); private final int bufferSize; private final ByteString clientId = ByteString.copyFrom(UUID.randomUUID().toString().getBytes(UTF_8)); @@ -427,13 +427,7 @@ public static Logger getLogger() { } void requestTimeout(RequestKey requestKey) { - final RequestEntry entry; - lock.lock(); - try { - entry = sentRequests.remove(requestKey); - } finally { - lock.unlock(); - } + final RequestEntry entry = sentRequests.remove(requestKey); if (entry != null) { LOG.warn("Timeout to receive response for command {}", entry.getRequest()); ContainerProtos.Type type = entry.getRequest().getCmdType(); @@ -523,15 +517,7 @@ public void run() { lock.unlock(); } long timerTaskCancelledCount = 0; - while (true) { - lock.lock(); - try { - if (!isDomainSocketOpen.get()) { - return; - } - } finally { - lock.unlock(); - } + while (isDomainSocketOpen.get()) { RequestEntry entry = null; try { DataInputStream dataIn = new DataInputStream(socket.getInputStream()); @@ -548,14 +534,8 @@ public void run() { if (LOG.isDebugEnabled()) { LOG.debug("received response {} callId {}", type, responseProto.getCallId()); } - final long sentTimeNs; - lock.lock(); - try { - entry = sentRequests.remove(new RequestKey(responseProto.getClientId(), responseProto.getCallId())); - sentTimeNs = entry == null ? 0 : entry.getSentTimeNs(); - } finally { - lock.unlock(); - } + entry = sentRequests.remove(new RequestKey(responseProto.getClientId(), responseProto.getCallId())); + final long sentTimeNs = entry == null ? 0 : entry.getSentTimeNs(); if (entry == null) { // This could be two cases // 1. there is bug in the code @@ -700,8 +680,7 @@ static class RequestEntry { private final ContainerCommandRequestProto request; private final CompletableFuture future; private final long createTimeNs; - // Accessed under the enclosing client's lock. - private long sentTimeNs; + private final AtomicLong sentTimeNs; private final TimerTask timerTask; RequestEntry(ContainerCommandRequestProto requestProto, @@ -710,6 +689,7 @@ static class RequestEntry { this.future = future; this.timerTask = timerTask; this.createTimeNs = System.nanoTime(); + this.sentTimeNs = new AtomicLong(createTimeNs); } public ContainerCommandRequestProto getRequest() { @@ -725,11 +705,11 @@ public long getCreateTimeNs() { } public long getSentTimeNs() { - return sentTimeNs; + return sentTimeNs.get(); } public void setSentTimeNs() { - sentTimeNs = System.nanoTime(); + sentTimeNs.set(System.nanoTime()); } public TimerTask getTimerTask() { diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java index 4ed611b9132d..d56001213fcc 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java @@ -30,6 +30,8 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.io.DataInputStream; +import java.io.DataOutputStream; import java.io.File; import java.io.IOException; import java.util.ArrayList; @@ -59,6 +61,7 @@ import org.apache.hadoop.ozone.OzoneConsts; import org.apache.hadoop.ozone.container.common.SCMTestUtils; import org.apache.ozone.test.GenericTestUtils; +import org.apache.ratis.thirdparty.com.google.protobuf.ByteString; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -79,6 +82,20 @@ @Timeout(300) public class TestXceiverClientManagerSC { + private static final int CONCURRENT_OPERATION_COUNT = 8; + private static final int REQUEST_COUNT = CONCURRENT_OPERATION_COUNT * 4; + private static final int RACE_ITERATION_COUNT = 10; + private static final int TEST_TIMEOUT_SECONDS = 30; + private static final int TEST_TIMEOUT_MILLIS = + Math.toIntExact(TimeUnit.SECONDS.toMillis(TEST_TIMEOUT_SECONDS)); + private static final int NON_BLOCKING_TIMEOUT_SECONDS = 5; + private static final int WAIT_INTERVAL_MILLIS = 100; + private static final int DELAYED_ECHO_MILLIS = + Math.toIntExact(TimeUnit.SECONDS.toMillis(1)); + private static final int RECEIVER_READ_TIMEOUT_MILLIS = 200; + private static final int BLOCKED_WRITE_PAYLOAD_SIZE = + Math.toIntExact(8 * OzoneConsts.MB); + private static OzoneConfiguration config; private static MiniOzoneCluster cluster; private static ContainerWithPipeline echoContainer; @@ -114,7 +131,7 @@ public static void init() throws Exception { } GenericTestUtils.waitFor(() -> cluster.getHddsDatanodes().stream().allMatch(dn -> dn.getDatanodeStateMachine() .getContainer().getContainerSet().getContainer(echoContainer.getContainerInfo().getContainerID()) != null), - 100, 30000); + WAIT_INTERVAL_MILLIS, TEST_TIMEOUT_MILLIS); } @AfterAll @@ -155,21 +172,21 @@ public void testAllocateShortCircuitClient() throws IOException { @Test public void testConcurrentConnectAndRequests() throws Exception { Pipeline pipeline = echoContainer.getPipeline(); - ExecutorService executor = Executors.newFixedThreadPool(8); + ExecutorService executor = Executors.newFixedThreadPool(CONCURRENT_OPERATION_COUNT); try (XceiverClientShortCircuit client = new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { CountDownLatch start = new CountDownLatch(1); List> connections = new ArrayList<>(); - for (int i = 0; i < 8; i++) { + for (int i = 0; i < CONCURRENT_OPERATION_COUNT; i++) { connections.add(executor.submit(() -> { - assertTrue(start.await(30, TimeUnit.SECONDS)); + assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); client.connect(); return null; })); } start.countDown(); for (Future connection : connections) { - connection.get(30, TimeUnit.SECONDS); + connection.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); } client.connect(); assertFalse(client.isClosed()); @@ -178,21 +195,87 @@ public void testConcurrentConnectAndRequests() throws Exception { CountDownLatch send = new CountDownLatch(1); List> requests = new ArrayList<>(); - for (int i = 0; i < 32; i++) { + for (int i = 0; i < REQUEST_COUNT; i++) { ContainerCommandRequestProto request = echoRequest(client, 0); requests.add(executor.submit(() -> { - assertTrue(send.await(30, TimeUnit.SECONDS)); + assertTrue(send.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); assertEchoResponse(request, client.sendCommand(request)); return null; })); } send.countDown(); for (Future request : requests) { - request.get(30, TimeUnit.SECONDS); + request.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } + } + + @Test + public void testResponseAndTimeoutWhileWriteIsBlocked() throws Exception { + Pipeline pipeline = echoContainer.getPipeline(); + DomainSocketFactory factory = mock(DomainSocketFactory.class); + DomainSocket[] sockets = DomainSocket.socketpair(); + DomainSocket clientSocket = sockets[0]; + DomainSocket peerSocket = sockets[1]; + clientSocket.setAttribute(DomainSocket.SEND_BUFFER_SIZE, + config.getObject(OzoneClientConfig.class).getShortCircuitBufferSize()); + when(factory.createSocket(anyInt(), anyInt(), any())).thenReturn(clientSocket); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try (MockedStatic mocked = mockStatic(DomainSocketFactory.class)) { + mocked.when(() -> DomainSocketFactory.getInstance(config)).thenReturn(factory); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + client.connect(); + DataInputStream peerIn = new DataInputStream(peerSocket.getInputStream()); + DataOutputStream peerOut = new DataOutputStream(peerSocket.getOutputStream()); + + ContainerCommandRequestProto responseRequest = echoRequest(client, 0); + CompletableFuture responseFuture = + client.sendCommandInternal(responseRequest).getResponse(); + assertEquals(responseRequest, readRequest(peerIn)); + + ContainerCommandRequestProto timeoutRequest = echoRequest(client, 0); + CompletableFuture timeoutFuture = + client.sendCommandInternal(timeoutRequest).getResponse(); + assertEquals(timeoutRequest, readRequest(peerIn)); + + ContainerCommandRequestProto blockedRequest = echoRequest(client, 0).toBuilder() + .setEcho(ContainerProtos.EchoRequestProto.newBuilder() + .setReadOnly(true) + .setPayload(ByteString.copyFrom(new byte[BLOCKED_WRITE_PAYLOAD_SIZE]))) + .build(); + Future blockedSend = + executor.submit(() -> client.sendCommandInternal(blockedRequest)); + assertEquals(OzoneClientConfig.DATA_TRANSFER_VERSION, peerIn.readShort()); + assertEquals(ContainerProtos.Type.Echo.getNumber(), peerIn.readShort()); + assertFalse(blockedSend.isDone()); + + sendEchoResponse(responseRequest, peerOut); + assertEchoResponse(responseRequest, + responseFuture.get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + + Future timeout = executor.submit(() -> + client.requestTimeout(new XceiverClientShortCircuit.RequestKey( + timeoutRequest.getClientId(), timeoutRequest.getCallId()))); + timeout.get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS); + assertThrows(ExecutionException.class, + () -> timeoutFuture.get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertFalse(blockedSend.isDone()); + + peerSocket.close(); + assertThrows(ExecutionException.class, + () -> blockedSend.get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS).getResponse() + .get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + } finally { + IOUtils.cleanupWithLogger(null, peerSocket, clientSocket); } } finally { executor.shutdownNow(); - assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + assertTrue(executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); } } @@ -204,17 +287,17 @@ public void testMultiplePendingRequestsAndCloseFromResponse() throws Exception { client.connect(); List requests = new ArrayList<>(); List> responses = new ArrayList<>(); - for (int i = 0; i < 8; i++) { - ContainerCommandRequestProto request = echoRequest(client, i == 0 ? 1000 : 0); + for (int i = 0; i < CONCURRENT_OPERATION_COUNT; i++) { + ContainerCommandRequestProto request = echoRequest(client, i == 0 ? DELAYED_ECHO_MILLIS : 0); requests.add(request); responses.add(client.sendCommandInternal(request).getResponse()); } assertFalse(responses.get(0).isDone()); for (int i = 0; i < requests.size(); i++) { - assertEchoResponse(requests.get(i), responses.get(i).get(30, TimeUnit.SECONDS)); + assertEchoResponse(requests.get(i), responses.get(i).get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); } - client.sendCommandInternal(echoRequest(client, 1000)).getResponse() - .thenRun(client::close).get(30, TimeUnit.SECONDS); + client.sendCommandInternal(echoRequest(client, DELAYED_ECHO_MILLIS)).getResponse() + .thenRun(client::close).get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); assertTrue(client.isClosed()); } } @@ -222,26 +305,27 @@ public void testMultiplePendingRequestsAndCloseFromResponse() throws Exception { @Test public void testConcurrentCloseWithPendingRequests() throws Exception { Pipeline pipeline = echoContainer.getPipeline(); - ExecutorService executor = Executors.newFixedThreadPool(8); + ExecutorService executor = Executors.newFixedThreadPool(CONCURRENT_OPERATION_COUNT); try (XceiverClientShortCircuit client = new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { client.connect(); CompletableFuture pending = - client.sendCommandInternal(echoRequest(client, 1000)).getResponse(); + client.sendCommandInternal(echoRequest(client, DELAYED_ECHO_MILLIS)).getResponse(); CountDownLatch start = new CountDownLatch(1); List> closes = new ArrayList<>(); - for (int i = 0; i < 8; i++) { + for (int i = 0; i < CONCURRENT_OPERATION_COUNT; i++) { closes.add(executor.submit(() -> { - assertTrue(start.await(30, TimeUnit.SECONDS)); + assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); client.close(); return null; })); } start.countDown(); for (Future close : closes) { - close.get(30, TimeUnit.SECONDS); + close.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); } - assertThrows(ExecutionException.class, () -> pending.get(30, TimeUnit.SECONDS)); + assertThrows(ExecutionException.class, + () -> pending.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); assertTrue(client.isClosed()); assertThrows(IOException.class, client::connect); assertThrows(IOException.class, client::checkOpen); @@ -249,7 +333,7 @@ public void testConcurrentCloseWithPendingRequests() throws Exception { assertNotNull(client.toString()); } finally { executor.shutdownNow(); - assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + assertTrue(executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); } } @@ -258,7 +342,7 @@ public void testConnectAndSendRacingWithClose() throws Exception { Pipeline pipeline = echoContainer.getPipeline(); ExecutorService executor = Executors.newFixedThreadPool(3); try { - for (int i = 0; i < 10; i++) { + for (int i = 0; i < RACE_ITERATION_COUNT; i++) { try (XceiverClientShortCircuit client = new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { if (i % 2 == 0) { @@ -266,7 +350,7 @@ public void testConnectAndSendRacingWithClose() throws Exception { } CountDownLatch start = new CountDownLatch(1); Future connect = executor.submit(() -> { - assertTrue(start.await(30, TimeUnit.SECONDS)); + assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); try { client.connect(); } catch (IOException expected) { @@ -275,10 +359,11 @@ public void testConnectAndSendRacingWithClose() throws Exception { return null; }); Future send = executor.submit(() -> { - assertTrue(start.await(30, TimeUnit.SECONDS)); + assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); ContainerCommandRequestProto request = echoRequest(client, 0); try { - assertEchoResponse(request, client.sendCommandInternal(request).getResponse().get(30, TimeUnit.SECONDS)); + assertEchoResponse(request, + client.sendCommandInternal(request).getResponse().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); } catch (IOException expected) { // The connection may not have opened yet, or close may have won the lock. } catch (ExecutionException expected) { @@ -287,21 +372,21 @@ public void testConnectAndSendRacingWithClose() throws Exception { return null; }); Future close = executor.submit(() -> { - assertTrue(start.await(30, TimeUnit.SECONDS)); + assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); client.close(); return null; }); start.countDown(); - connect.get(30, TimeUnit.SECONDS); - send.get(30, TimeUnit.SECONDS); - close.get(30, TimeUnit.SECONDS); + connect.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + send.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + close.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); assertTrue(client.isClosed()); assertThrows(IOException.class, client::connect); } } } finally { executor.shutdownNow(); - assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS)); + assertTrue(executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); } } @@ -347,7 +432,8 @@ public void testConnectFailureCannotReconnect(boolean throwException) throws Exc public void testReceiverFailureCannotReconnect() throws Exception { Pipeline pipeline = echoContainer.getPipeline(); OzoneConfiguration timeoutConfig = new OzoneConfiguration(config); - timeoutConfig.setTimeDuration(OzoneConfigKeys.OZONE_CLIENT_READ_TIMEOUT, 200, TimeUnit.MILLISECONDS); + timeoutConfig.setTimeDuration(OzoneConfigKeys.OZONE_CLIENT_READ_TIMEOUT, + RECEIVER_READ_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS); try (XceiverClientShortCircuit client = new XceiverClientShortCircuit(pipeline, timeoutConfig, pipeline.getClosestNode())) { client.connect(); @@ -358,7 +444,7 @@ public void testReceiverFailureCannotReconnect() throws Exception { } catch (IOException expected) { return true; } - }, 50, 30000); + }, WAIT_INTERVAL_MILLIS, TEST_TIMEOUT_MILLIS); assertFalse(client.isClosed()); assertNotNull(client.toString()); assertThrows(IOException.class, client::connect); @@ -374,15 +460,37 @@ private static ContainerCommandRequestProto echoRequest(XceiverClientShortCircui .setClientId(client.getClientId()) .setCallId(client.getCallId()) .setEcho(ContainerProtos.EchoRequestProto.newBuilder().setReadOnly(true).setSleepTimeMs(sleepTimeMs) - .setPayloadSizeResp(1024)) + .setPayloadSizeResp(Math.toIntExact(OzoneConsts.KB))) + .build(); + } + + private static ContainerCommandRequestProto readRequest(DataInputStream input) throws IOException { + assertEquals(OzoneClientConfig.DATA_TRANSFER_VERSION, input.readShort()); + assertEquals(ContainerProtos.Type.Echo.getNumber(), input.readShort()); + return ContainerCommandRequestProto.parseDelimitedFrom(input); + } + + private static void sendEchoResponse(ContainerCommandRequestProto request, DataOutputStream output) + throws IOException { + ContainerCommandResponseProto response = ContainerCommandResponseProto.newBuilder() + .setCmdType(ContainerProtos.Type.Echo) + .setResult(ContainerProtos.Result.SUCCESS) + .setClientId(request.getClientId()) + .setCallId(request.getCallId()) + .setEcho(ContainerProtos.EchoResponseProto.newBuilder() + .setPayload(ByteString.copyFrom(new byte[request.getEcho().getPayloadSizeResp()]))) .build(); + output.writeShort(OzoneClientConfig.DATA_TRANSFER_VERSION); + output.writeShort(ContainerProtos.Type.Echo.getNumber()); + response.writeDelimitedTo(output); + output.flush(); } private static void assertEchoResponse(ContainerCommandRequestProto request, ContainerCommandResponseProto response) { assertEquals(ContainerProtos.Result.SUCCESS, response.getResult()); assertEquals(request.getClientId(), response.getClientId()); assertEquals(request.getCallId(), response.getCallId()); - assertEquals(1024, response.getEcho().getPayload().size()); + assertEquals(request.getEcho().getPayloadSizeResp(), response.getEcho().getPayload().size()); } } From d5b0ad3f6eb21fc72f1ed934439a13cf3c04663b Mon Sep 17 00:00:00 2001 From: echonesis Date: Mon, 14 Sep 2026 11:51:32 +0800 Subject: [PATCH 3/3] fix: account for all pending short-circuit requests on failure --- .../hdds/scm/XceiverClientShortCircuit.java | 46 ++++++++---- .../hdds/scm/TestXceiverClientManagerSC.java | 73 ++++++++++++------- 2 files changed, 78 insertions(+), 41 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java index 80fb8f0afc29..a569a4303156 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientShortCircuit.java @@ -201,11 +201,11 @@ public void close() { } readDaemon.interrupt(); } - pending = new ArrayList<>(sentRequests.values()); + pending = removePendingRequests(); } finally { lock.unlock(); } - pending.forEach(entry -> entry.fail(new ClosedChannelException())); + failRequests(pending, new ClosedChannelException()); if (Thread.currentThread() != readDaemon) { try { readDaemon.join(); @@ -430,9 +430,7 @@ void requestTimeout(RequestKey requestKey) { final RequestEntry entry = sentRequests.remove(requestKey); if (entry != null) { LOG.warn("Timeout to receive response for command {}", entry.getRequest()); - ContainerProtos.Type type = entry.getRequest().getCmdType(); - metrics.decrPendingContainerOpsMetrics(type); - entry.getFuture().completeExceptionally(new TimeoutException("Timeout to receive response")); + failRequest(entry, new TimeoutException("Timeout to receive response")); } } @@ -467,7 +465,7 @@ void sendRequest(RequestEntry entry) throws IOException { } catch (IOException e) { isDomainSocketOpen.set(false); failure = e; - pending = new ArrayList<>(sentRequests.values()); + pending = removePendingRequests(); } finally { entry.setSentTimeNs(); requestSent++; @@ -477,14 +475,32 @@ void sendRequest(RequestEntry entry) throws IOException { } 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()); + failRequests(pending, failure); } } + private List removePendingRequests() { + // The caller holds the lock and has prevented new requests from being registered. + List pending = new ArrayList<>(); + sentRequests.forEach((key, entry) -> { + if (sentRequests.remove(key, entry)) { + pending.add(entry); + } + }); + return pending; + } + + private void failRequests(List requests, Throwable failure) { + requests.forEach(entry -> failRequest(entry, failure)); + } + + private void failRequest(RequestEntry entry, Throwable failure) { + entry.fail(failure); + ContainerProtos.Type type = entry.getRequest().getCmdType(); + metrics.decrPendingContainerOpsMetrics(type); + metrics.addContainerOpsLatency(type, System.nanoTime() - entry.getCreateTimeNs()); + } + @Override public String toString() { lock.lock(); @@ -585,7 +601,7 @@ public void run() { LOG.warn("Failed to handle short-circuit information exchange", e); // disable docket socket for a while domainSocketFactory.disableShortCircuit(); - entry.getFuture().completeExceptionally(e); + failRequest(entry, e); continue; } } @@ -625,14 +641,14 @@ public void run() { LOG.error("{} failed after send {} requests and received {} responses", socket, requestSent, responseReceived, e); } - pending = new ArrayList<>(sentRequests.values()); + pending = removePendingRequests(); } finally { lock.unlock(); } if (entry != null) { - entry.getFuture().completeExceptionally(e); + failRequest(entry, e); } - pending.forEach(i -> i.fail(e)); + failRequests(pending, e); break; } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java index d56001213fcc..1efbe1ef3b01 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientManagerSC.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.times; @@ -216,6 +217,7 @@ public void testConcurrentConnectAndRequests() throws Exception { @Test public void testResponseAndTimeoutWhileWriteIsBlocked() throws Exception { Pipeline pipeline = echoContainer.getPipeline(); + XceiverClientMetrics metrics = mock(XceiverClientMetrics.class); DomainSocketFactory factory = mock(DomainSocketFactory.class); DomainSocket[] sockets = DomainSocket.socketpair(); DomainSocket clientSocket = sockets[0]; @@ -225,8 +227,10 @@ public void testResponseAndTimeoutWhileWriteIsBlocked() throws Exception { when(factory.createSocket(anyInt(), anyInt(), any())).thenReturn(clientSocket); ExecutorService executor = Executors.newFixedThreadPool(2); - try (MockedStatic mocked = mockStatic(DomainSocketFactory.class)) { - mocked.when(() -> DomainSocketFactory.getInstance(config)).thenReturn(factory); + try (MockedStatic mockedFactory = mockStatic(DomainSocketFactory.class); + MockedStatic mockedManager = mockStatic(XceiverClientManager.class)) { + mockedFactory.when(() -> DomainSocketFactory.getInstance(config)).thenReturn(factory); + mockedManager.when(XceiverClientManager::getXceiverClientMetrics).thenReturn(metrics); try (XceiverClientShortCircuit client = new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { client.connect(); @@ -243,6 +247,11 @@ public void testResponseAndTimeoutWhileWriteIsBlocked() throws Exception { client.sendCommandInternal(timeoutRequest).getResponse(); assertEquals(timeoutRequest, readRequest(peerIn)); + ContainerCommandRequestProto failureRequest = echoRequest(client, 0); + CompletableFuture failureFuture = + client.sendCommandInternal(failureRequest).getResponse(); + assertEquals(failureRequest, readRequest(peerIn)); + ContainerCommandRequestProto blockedRequest = echoRequest(client, 0).toBuilder() .setEcho(ContainerProtos.EchoRequestProto.newBuilder() .setReadOnly(true) @@ -270,6 +279,11 @@ public void testResponseAndTimeoutWhileWriteIsBlocked() throws Exception { assertThrows(ExecutionException.class, () -> blockedSend.get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS).getResponse() .get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertThrows(ExecutionException.class, + () -> failureFuture.get(NON_BLOCKING_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + verify(metrics, times(4)).incrPendingContainerOpsMetrics(ContainerProtos.Type.Echo); + verify(metrics, times(4)).decrPendingContainerOpsMetrics(ContainerProtos.Type.Echo); + verify(metrics, times(4)).addContainerOpsLatency(any(), anyLong()); } finally { IOUtils.cleanupWithLogger(null, peerSocket, clientSocket); } @@ -306,31 +320,38 @@ public void testMultiplePendingRequestsAndCloseFromResponse() throws Exception { public void testConcurrentCloseWithPendingRequests() throws Exception { Pipeline pipeline = echoContainer.getPipeline(); ExecutorService executor = Executors.newFixedThreadPool(CONCURRENT_OPERATION_COUNT); - try (XceiverClientShortCircuit client = - new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { - client.connect(); - CompletableFuture pending = - client.sendCommandInternal(echoRequest(client, DELAYED_ECHO_MILLIS)).getResponse(); - CountDownLatch start = new CountDownLatch(1); - List> closes = new ArrayList<>(); - for (int i = 0; i < CONCURRENT_OPERATION_COUNT; i++) { - closes.add(executor.submit(() -> { - assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); - client.close(); - return null; - })); - } - start.countDown(); - for (Future close : closes) { - close.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + XceiverClientMetrics metrics = mock(XceiverClientMetrics.class); + try (MockedStatic mockedManager = mockStatic(XceiverClientManager.class)) { + mockedManager.when(XceiverClientManager::getXceiverClientMetrics).thenReturn(metrics); + try (XceiverClientShortCircuit client = + new XceiverClientShortCircuit(pipeline, config, pipeline.getClosestNode())) { + client.connect(); + CompletableFuture pending = + client.sendCommandInternal(echoRequest(client, DELAYED_ECHO_MILLIS)).getResponse(); + CountDownLatch start = new CountDownLatch(1); + List> closes = new ArrayList<>(); + for (int i = 0; i < CONCURRENT_OPERATION_COUNT; i++) { + closes.add(executor.submit(() -> { + assertTrue(start.await(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + client.close(); + return null; + })); + } + start.countDown(); + for (Future close : closes) { + close.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + assertThrows(ExecutionException.class, + () -> pending.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); + assertTrue(client.isClosed()); + assertThrows(IOException.class, client::connect); + assertThrows(IOException.class, client::checkOpen); + assertThrows(IOException.class, () -> client.sendCommandInternal(echoRequest(client, 0))); + assertNotNull(client.toString()); + verify(metrics).incrPendingContainerOpsMetrics(ContainerProtos.Type.Echo); + verify(metrics).decrPendingContainerOpsMetrics(ContainerProtos.Type.Echo); + verify(metrics).addContainerOpsLatency(any(), anyLong()); } - assertThrows(ExecutionException.class, - () -> pending.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS)); - assertTrue(client.isClosed()); - assertThrows(IOException.class, client::connect); - assertThrows(IOException.class, client::checkOpen); - assertThrows(IOException.class, () -> client.sendCommandInternal(echoRequest(client, 0))); - assertNotNull(client.toString()); } finally { executor.shutdownNow(); assertTrue(executor.awaitTermination(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS));