[
https://issues.apache.org/jira/browse/QPID-8772?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18130748#comment-18130748
]
ASF GitHub Bot commented on QPID-8772:
--------------------------------------
dakirily commented on code in PR #453:
URL: https://github.com/apache/qpid-broker-j/pull/453#discussion_r4357226513
##########
broker-core/src/test/java/org/apache/qpid/server/transport/TlsTestContext.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.server.transport;
+
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+
+import org.apache.qpid.test.utils.tls.KeyCertificatePair;
+import org.apache.qpid.test.utils.tls.TlsResourceBuilder;
+
+final class TlsTestContext
+{
+ private TlsTestContext()
+ {
Review Comment:
Please provide a comment explaining private empty constructor
##########
broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionWriteTest.java:
##########
@@ -0,0 +1,507 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.server.transport;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionWriteTest extends UnitTestBase
+{
+ private static final int NETWORK_BUFFER_SIZE = 1024;
+
+ private final AtomicBoolean _hasWork = new AtomicBoolean();
+ private final AtomicBoolean _blocked = new AtomicBoolean();
+ private final AtomicInteger _writableBytes = new AtomicInteger();
+ private final List<Boolean> _blockedTransitions = new ArrayList<>();
+ private ProtocolEngine _engine;
+ private SocketChannel _channel;
+ private NonBlockingConnection _connection;
+
+ @BeforeEach
+ void setUp() throws Exception
+ {
+ _hasWork.set(false);
+ _blocked.set(false);
+ _writableBytes.set(0);
+ _blockedTransitions.clear();
+ final Socket socket = mock(Socket.class);
+ when(socket.getRemoteSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 1000));
+ when(socket.getLocalSocketAddress()).thenReturn(new
InetSocketAddress("localhost", 5672));
+ _channel = mock(SocketChannel.class);
+ when(_channel.socket()).thenReturn(socket);
+ when(_channel.write(any(ByteBuffer[].class))).thenAnswer(invocation ->
+ {
+ final ByteBuffer[] buffers = invocation.getArgument(0);
+ final int offset = 0;
+ final int length = buffers.length;
+ long written = 0;
+ for (int i = offset; i < offset + length; i++)
+ {
+ final int size = Math.min(buffers[i].remaining(),
_writableBytes.get());
+ buffers[i].position(buffers[i].position() + size);
+ _writableBytes.addAndGet(-size);
+ written += size;
+ }
+ return written;
+ });
+
+ _engine = mock(ProtocolEngine.class);
+ final AggregateTicker ticker = mock(AggregateTicker.class);
+
when(ticker.getTimeToNextTick(anyLong())).thenReturn(Integer.MAX_VALUE);
+ when(_engine.getAggregateTicker()).thenReturn(ticker);
+ when(_engine.processPendingIterator()).thenAnswer(invocation ->
Collections.emptyIterator());
+ when(_engine.hasWork()).thenAnswer(invocation -> _hasWork.get());
+ when(_engine.isTransportBlockedForWriting()).thenAnswer(invocation ->
_blocked.get());
+ doAnswer(invocation ->
+ {
+ _hasWork.set(false);
+ return null;
+ }).when(_engine).clearWork();
+ doAnswer(invocation ->
+ {
+ _hasWork.set(true);
+ return null;
+ }).when(_engine).notifyWork();
+ doAnswer(invocation ->
+ {
+ final boolean blocked = invocation.getArgument(0);
+ if (_blocked.getAndSet(blocked) != blocked)
+ {
+ _blockedTransitions.add(blocked);
+ if (!blocked)
+ {
+ _engine.notifyWork();
+ }
+ }
+ return null;
+ }).when(_engine).setTransportBlockedForWriting(anyBoolean());
+
+ final AmqpPort<?> port = mock(AmqpPort.class);
+ when(port.getNetworkBufferSize()).thenReturn(NETWORK_BUFFER_SIZE);
+ when(port.getContextValue(Integer.class,
AmqpPort.FINAL_WRITE_THRESHOLD)).thenReturn(1);
+ when(port.getContextValue(Long.class,
AmqpPort.FINAL_WRITE_TIMEOUT)).thenReturn(0L);
+ when(port.getContextValue(Integer.class,
AmqpPort.MAX_GATHERING_WRITE_BUFFERS)).thenReturn(1024);
+ final Broker<?> broker = mock(Broker.class);
+ when(broker.getEventLogger()).thenReturn(mock(EventLogger.class));
+ doReturn(broker).when(port).getParent();
+ _connection = new NonBlockingConnection(_channel, _engine,
Set.of(TransportEncryption.NONE),
+ () -> { }, mock(NetworkConnectionScheduler.class), port);
+ _connection.setSelectionTask(mock(SelectorThread.SelectionTask.class));
+ }
+
+ @AfterEach
+ void tearDown()
+ {
+ _writableBytes.set(Integer.MAX_VALUE);
+ _connection.close();
+ _connection.doWork();
+ }
+
+ @Test
+ void blockedOutputWaitsForSocketReadiness()
+ {
+ send(128);
+
+ assertFalse(_connection.doWork());
+
+ assertTrue(_connection.wantsWrite());
+ assertTrue(_connection.wantsRead(), "Completed protocol work must
allow incoming traffic during a blocked write");
+ assertFalse(_connection.isStateChanged(), "Pending output alone must
not reschedule the connection");
+ assertEquals(List.of(true), _blockedTransitions);
+
+ _engine.notifyWork();
+ _connection.doWork();
+ assertEquals(List.of(true), _blockedTransitions, "A failed retry must
not briefly resume consumers");
+ assertFalse(_connection.isStateChanged());
+
+ _writableBytes.set(32);
+ _connection.doWork();
+ assertTrue(_connection.wantsWrite(), "Partial progress must retain
write interest");
+ assertEquals(List.of(true), _blockedTransitions);
+ assertFalse(_connection.isStateChanged());
+
+ _writableBytes.set(96);
+ _connection.doWork();
+ assertEquals(List.of(true, false), _blockedTransitions);
+ assertTrue(_connection.isStateChanged(), "Unblocking must resume
pending protocol output");
+ assertTrue(_connection.wantsRead());
+ assertFalse(_connection.wantsWrite());
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, NETWORK_BUFFER_SIZE / 4})
+ void pendingIteratorIsRescheduledWhenSecondWriteDrainsOutput(final int
firstWriteBytes) throws Exception
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final List<Runnable> tasks = List.of(() -> send(2 *
NETWORK_BUFFER_SIZE), completed::incrementAndGet);
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+ final AtomicInteger writeCalls = new AtomicInteger();
+ doAnswer(invocation ->
+ {
+ final ByteBuffer[] buffers = invocation.getArgument(0);
+ int writableBytes = writeCalls.getAndIncrement() == 0 ?
firstWriteBytes : Integer.MAX_VALUE;
+ long written = 0;
+ for (final ByteBuffer buffer : buffers)
+ {
+ final int size = Math.min(buffer.remaining(), writableBytes);
+ buffer.position(buffer.position() + size);
+ writableBytes -= size;
+ written += size;
+ }
+ return written;
+ }).when(_channel).write(any(ByteBuffer[].class));
+
+ assertFalse(_connection.doWork());
+
+ assertEquals(2, writeCalls.get(), "The second write must drain output
after the iterator is suspended");
+ assertEquals(0, completed.get(), "The iterator still has runnable work
after the second write");
+ assertTrue(_connection.wantsRead());
+ assertFalse(_connection.wantsWrite());
+ assertFalse(_connection.isPartialRead());
+ assertFalse(_blocked.get());
+ assertTrue(_blockedTransitions.isEmpty(), "There is no backpressure
transition to supply a wake-up");
+ assertTrue(_connection.isStateChanged(),
+ "Pending protocol work must be rescheduled when output drains
in the same pass");
+
+ assertFalse(_connection.doWork());
+ assertEquals(1, completed.get(), "The retained iterator must resume
exactly once");
+ assertFalse(_connection.isStateChanged(), "An exhausted iterator must
not keep rescheduling");
+ verify(_engine).processPendingIterator();
+ }
+
+ @Test
+ void pendingIteratorResumesAfterOutputDrains()
+ {
+ final AtomicInteger completed = new AtomicInteger();
+ final List<Runnable> tasks = List.of(() -> send(NETWORK_BUFFER_SIZE),
completed::incrementAndGet);
+ when(_engine.processPendingIterator()).thenReturn(tasks.iterator());
+
+ _connection.doWork();
+
+ assertEquals(0, completed.get());
+ assertTrue(_connection.wantsWrite());
+ assertFalse(_connection.wantsRead(), "Suspended protocol work must
continue to apply read backpressure");
+ assertFalse(_connection.isStateChanged(), "A suspended iterator must
wait for its buffered output to drain");
+
+ _writableBytes.set(NETWORK_BUFFER_SIZE);
+ _connection.doWork();
+
+ assertEquals(1, completed.get(), "Suspended work must resume exactly
once");
+ assertFalse(_connection.wantsWrite());
+ }
+
+ @ParameterizedTest
+ @CsvSource({"false, 768", "true, 768", "false, 1023", "true, 1023"})
+ void partialWriteBelowBufferLimitResumesWorkAndReads(final boolean
pendingTask,
+ final int
remainingOutput) throws Exception
Review Comment:
Please fix the parameter indentation
##########
broker-core/src/main/java/org/apache/qpid/server/transport/NonBlockingConnection.java:
##########
@@ -278,19 +279,26 @@ public boolean doWork()
if (processPendingComplete)
{
_pendingIterator = null;
- _protocolEngine.setTransportBlockedForWriting(false);
+ }
+
+ if (processPendingComplete ||
_delegate.needsReadForWrite())
+ {
final boolean dataRead = doRead();
-
_protocolEngine.setTransportBlockedForWriting(!doWrite());
+ doWrite();
- if (!_fullyWritten || dataRead ||
(_delegate.needsWork() && _delegate.getNetInputBuffer().position() != 0))
+ if (_fullyWritten &&
+ (dataRead || (_delegate.needsWork() &&
_delegate.getNetInputBuffer().position() != 0)))
Review Comment:
if expression on lines 289-290 should be on one line
##########
broker-core/src/test/java/org/apache/qpid/server/transport/NonBlockingConnectionSchedulingTest.java:
##########
@@ -0,0 +1,455 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.qpid.server.transport;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocket;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import org.apache.qpid.server.bytebuffer.QpidByteBuffer;
+import org.apache.qpid.server.logging.EventLogger;
+import org.apache.qpid.server.model.Broker;
+import org.apache.qpid.server.model.port.AmqpPort;
+import org.apache.qpid.server.transport.network.TransportEncryption;
+import org.apache.qpid.server.util.Action;
+import org.apache.qpid.test.utils.UnitTestBase;
+
+class NonBlockingConnectionSchedulingTest extends UnitTestBase
+{
+ private static final int TIMEOUT_MILLIS = 10000;
+
+ private final List<Thread> _threads = new CopyOnWriteArrayList<>();
+ private final List<TestConnection> _connections = new ArrayList<>();
+ private NetworkConnectionScheduler _scheduler;
+ private SSLContext _sslContext;
+
+ @BeforeAll
+ void createTlsContext() throws Exception
+ {
+ _sslContext = TlsTestContext.create();
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {2, 4})
+ void blockedSocketReturnsWorkersAndRecoversPendingOutput(final int
workers) throws Exception
+ {
+ startScheduler(workers);
+ final TestConnection blocked = createConnection(1024);
+ final TestConnection independent = createConnection(1024);
+ final byte[] payload = new byte[2 * 1024 * 1024];
+ Arrays.fill(payload, (byte) 42);
+ final byte[] marker = new byte[]{1, 2, 3};
+ for (int i = 0; i < 3; i++)
+ {
+ blocked.submit(List.of(() -> blocked.send(payload), () ->
blocked.send(marker)));
+ blocked.awaitWriteInterest();
+
+ independent.submit(List.of(() -> independent.send(marker)));
+ assertArrayEquals(marker,
independent._client.getInputStream().readNBytes(marker.length),
+ "An independent connection must progress while the first
socket remains blocked");
+
+ assertArrayEquals(payload,
blocked._client.getInputStream().readNBytes(payload.length));
+ assertArrayEquals(marker,
blocked._client.getInputStream().readNBytes(marker.length),
+ "The suspended protocol iterator must resume after socket
capacity becomes available");
+ }
+ }
+
+ @ParameterizedTest
+ @CsvSource({"TLSv1.2, 2", "TLSv1.2, 4", "TLSv1.3, 2", "TLSv1.3, 4"})
+ void blockedTlsSocketReturnsWorkersAndRecoversPendingOutput(final String
protocol, final int workers) throws Exception
+ {
+ startScheduler(workers);
+ final TestConnection blocked = createTlsConnection(protocol);
+ final TestConnection independent = createTlsConnection(protocol);
+ final byte[] payload = new byte[2 * 1024 * 1024];
+ for (int i = 0; i < payload.length; i++)
+ {
+ payload[i] = (byte) i;
+ }
+ final byte[] marker = new byte[]{11, 22, 33};
+ for (int round = 0; round < 3; round++)
+ {
+ blocked.submit(List.of(() -> blocked.send(payload), () ->
blocked.send(marker)));
+ blocked.awaitWriteInterest();
+ assertTrue(blocked._blocked.get());
+
+ independent.submit(List.of(() -> independent.send(marker)));
+ assertArrayEquals(marker,
independent._client.getInputStream().readNBytes(marker.length),
+ "An independent TLS connection must progress while the
first socket is blocked");
+
+ blocked._client.getOutputStream().write(42 + round);
+ assertEquals(Integer.valueOf(42 + round),
blocked._received.poll(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS),
+ "TLS input must still be decrypted while encrypted output
is blocked");
+ blocked.awaitWriteInterest();
+ assertTrue(blocked._blocked.get(), "Reading peer input must not
briefly resume blocked consumers");
+
+ assertArrayEquals(payload,
blocked._client.getInputStream().readNBytes(payload.length));
+ assertArrayEquals(marker,
blocked._client.getInputStream().readNBytes(marker.length),
+ "Pending output must resume without losing or reordering
application bytes");
+ blocked.awaitSelection(false);
+ assertFalse(blocked._blocked.get(), "Draining encrypted output
must resume consumers");
+ }
+ }
+
+ @Test
+ void incomingHeartbeatsPreventReadIdleTimeoutWhileOutputIsBlocked() throws
Exception
+ {
+ startScheduler(2);
+ final TestConnection blocked = createConnection(4 * 1024 * 1024);
+ blocked.submit(List.of(() -> blocked.send(new byte[2 * 1024 * 1024])));
+ blocked.awaitWriteInterest();
+
+ final int readIdleMillis = 2000;
+ final AtomicLong currentTime = new AtomicLong();
+ final AtomicLong lastReadTime = new AtomicLong();
+ final BlockingQueue<Integer> heartbeats = new LinkedBlockingQueue<>();
+ final AtomicBoolean readerIdle = new AtomicBoolean();
+ when(blocked._engine.getLastReadTime()).thenAnswer(invocation ->
lastReadTime.get());
+ doAnswer(invocation ->
+ {
+ final QpidByteBuffer input = invocation.getArgument(0);
+ lastReadTime.set(currentTime.get());
+ while (input.hasRemaining())
+ {
+ heartbeats.add((int) input.get());
+ }
+ return null;
+ }).when(blocked._engine).received(any());
+ doAnswer(invocation ->
+ {
+ readerIdle.set(true);
+ blocked._connection.close();
+ return null;
+ }).when(blocked._engine).readerIdle();
+
+ // Keep wall-clock scheduling timestamps out of the idle calculation
so only the test advances time.
+ blocked._ticker.addTicker(new
ServerIdleReadTimeoutTicker(mock(ServerNetworkConnection.class),
+ blocked._engine, readIdleMillis)
+ {
+ @Override
+ public int getTimeToNextTick(final long ignored)
+ {
+ return super.getTimeToNextTick(currentTime.get());
+ }
+ });
+ blocked.awaitWriteInterest();
+
+ for (int i = 0; i < 8; i++)
+ {
+ currentTime.addAndGet(readIdleMillis / 4);
+ blocked._client.getOutputStream().write(42);
+ assertEquals(Integer.valueOf(42), heartbeats.poll(TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS),
+ "Incoming heartbeats must be consumed while the peer is
not reading output");
+ blocked.awaitWriteInterest();
+ assertFalse(readerIdle.get(), "A peer sending heartbeats must not
be considered idle");
+ }
+
+ assertTrue(blocked._connection.wantsWrite(), "Output must remain
blocked throughout the heartbeat checks");
+ currentTime.addAndGet(readIdleMillis - 1);
+ blocked.awaitWriteInterest();
+ assertFalse(readerIdle.get(), "The connection must remain open until
the read deadline");
+
+ currentTime.incrementAndGet();
+ blocked._connection.getSelectionTask().wakeup();
+ blocked._closed.get(TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
+ assertTrue(readerIdle.get(), "The read timeout must still close the
connection after heartbeats stop");
+ }
+
+ private void startScheduler(final int workers)
+ {
+ _scheduler = new NetworkConnectionScheduler(getTestName(), 1, workers,
1L, task ->
+ {
+ final Thread thread = new Thread(task, getTestName() + "-" +
_threads.size());
+ _threads.add(thread);
+ return thread;
+ });
+ _scheduler.start();
+ }
+
+ private TestConnection createConnection(final int networkBufferSize)
throws Exception
+ {
+ final TestConnection connection = new TestConnection(_scheduler,
networkBufferSize);
+ _connections.add(connection);
+ return connection;
+ }
+
+ private TestConnection createTlsConnection(final String protocol) throws
Exception
+ {
+ final SSLSocket client = (SSLSocket)
_sslContext.getSocketFactory().createSocket();
+ client.setEnabledProtocols(new String[]{protocol});
+ final TestConnection connection = new TestConnection(_scheduler, 64 *
1024, client, _sslContext);
+ _connections.add(connection);
+ client.startHandshake();
+ assertEquals(protocol, client.getSession().getProtocol());
+ return connection;
+ }
+
+ @AfterEach
+ void tearDown() throws Exception
+ {
+ if (_scheduler != null)
+ {
+ try
+ {
+ for (final TestConnection connection : _connections)
+ {
+ connection._connection.close();
+ }
+ for (final TestConnection connection : _connections)
+ {
+ connection._closed.get(TIMEOUT_MILLIS,
TimeUnit.MILLISECONDS);
+ }
+ }
+ finally
+ {
+ for (final TestConnection connection : _connections)
+ {
+ connection._client.close();
+ connection._channel.close();
+ }
+ _scheduler.close();
+ try
+ {
+ for (final Thread thread : _threads)
+ {
+ thread.join(TIMEOUT_MILLIS);
+ assertFalse(thread.isAlive(), "Scheduler worker did
not terminate: " + thread.getName());
+ }
+ }
+ finally
+ {
+ for (final Thread thread : _threads)
+ {
+ thread.interrupt();
+ }
+ for (final Thread thread : _threads)
+ {
+ thread.join(TIMEOUT_MILLIS);
+ }
+ }
+ }
+ }
+ }
+
+ private static final class TestConnection
+ {
+ private final AtomicBoolean _hasWork = new AtomicBoolean();
+ private final AtomicBoolean _blocked = new AtomicBoolean();
+ private final AtomicReference<Iterator<Runnable>> _pending =
+ new AtomicReference<>(Collections.emptyIterator());
+ private final AtomicReference<SelectionInterest> _selectionInterest =
new AtomicReference<>();
+ private final BlockingQueue<Integer> _received = new
LinkedBlockingQueue<>();
+ private final AggregateTicker _ticker = new AggregateTicker()
+ {
+ @Override
+ public void resetModified()
+ {
+ super.resetModified();
+ // The selector calls this after examining a registered
connection's ticker.
+ signalSelection();
+ }
+ };
+ private final CompletableFuture<Void> _closed = new
CompletableFuture<>();
+ private final Socket _client;
+ private final SocketChannel _channel;
+ private final ProtocolEngine _engine = mock(ProtocolEngine.class);
+ private final NonBlockingConnection _connection;
+
+ private TestConnection(final NetworkConnectionScheduler scheduler,
final int networkBufferSize) throws Exception
+ {
+ this(scheduler, networkBufferSize, new Socket(), null);
+ }
+
+ private TestConnection(final NetworkConnectionScheduler scheduler,
final int networkBufferSize,
Review Comment:
Please place each method parameter on a separate line
> [Broker-J] Excessive CPU usage from continuous rescheduling of blocked
> TCP/TLS writes
> -------------------------------------------------------------------------------------
>
> Key: QPID-8772
> URL: https://issues.apache.org/jira/browse/QPID-8772
> Project: Qpid
> Issue Type: Bug
> Components: Broker-J
> Affects Versions: qpid-java-broker-10.1.1
> Reporter: Kato Kaito
> Priority: Minor
> Fix For: qpid-java-broker-10.2.0
>
> Attachments: qpid-8772-jmh-main.zip
>
>
> Deployments with many concurrent AMQP connections can experience write
> backpressure when some clients consume data slowly or communicate over
> constrained network paths, such as VPNs or TCP proxies. In these conditions,
> outbound data remains buffered while the broker waits for clients to receive
> it.
> The native TCP/TLS transport can repeatedly reschedule a connection with
> pending output without waiting for the socket to become writable. This causes
> repeated write attempts even when the connection cannot make progress. A
> stale partial-read indication can also keep a connection scheduled when
> subsequent processing skips reads.
> This behavior can consume significant CPU despite limited network throughput
> and occupy I/O workers needed by other connections. The impact is
> particularly relevant when several slow connections share an I/O pool with
> otherwise healthy clients.
> TLS requires an additional distinction: pending application output may be
> waiting for peer handshake data rather than socket write capacity. In this
> case, write readiness cannot advance the connection; the transport must allow
> reads and wait for incoming data.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
