This is an automated email from the ASF dual-hosted git repository. dakirily pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
commit 804858330a356d51f8ef6b2215623294b9be0529 Author: Daniil Kirilyuk <[email protected]> AuthorDate: Wed Sep 16 12:40:20 2026 +0200 QPID-8757: [Broker-J] WebSocket idle checker queues unbounded tick jobs while a connection is writing, exhausting the broker heap --- broker-plugins/websocket/pom.xml | 7 + .../transport/websocket/WebSocketProvider.java | 335 +++++++++++-- .../transport/websocket/WebSocketProviderTest.java | 548 +++++++++++++++++++++ .../websocket/WebSocketFrameTransport.java | 37 +- .../websocket/WebSocketIdleTimeoutTest.java | 101 ++++ 5 files changed, 976 insertions(+), 52 deletions(-) diff --git a/broker-plugins/websocket/pom.xml b/broker-plugins/websocket/pom.xml index e237ce4f86..3761ba8b1e 100644 --- a/broker-plugins/websocket/pom.xml +++ b/broker-plugins/websocket/pom.xml @@ -52,6 +52,13 @@ <artifactId>jetty-ee11-websocket-jetty-server</artifactId> </dependency> + <!-- test dependencies --> + <dependency> + <groupId>org.apache.qpid</groupId> + <artifactId>qpid-test-utils</artifactId> + <scope>test</scope> + </dependency> + </dependencies> <build> diff --git a/broker-plugins/websocket/src/main/java/org/apache/qpid/server/transport/websocket/WebSocketProvider.java b/broker-plugins/websocket/src/main/java/org/apache/qpid/server/transport/websocket/WebSocketProvider.java index 5b363c091d..712f077d5b 100644 --- a/broker-plugins/websocket/src/main/java/org/apache/qpid/server/transport/websocket/WebSocketProvider.java +++ b/broker-plugins/websocket/src/main/java/org/apache/qpid/server/transport/websocket/WebSocketProvider.java @@ -29,11 +29,15 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.Objects; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.ThreadFactory; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.LongSupplier; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLEngine; @@ -89,12 +93,17 @@ class WebSocketProvider implements AcceptingTransport { private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketProvider.class); private static final String AMQP_WEBSOCKET_SUBPROTOCOL = "amqp"; + // Defensive backoff for tickers that remain overdue or fail. Normal scheduling uses the ticker's deadline. + private static final int IMMEDIATE_OVERDUE_TICK_RETRY_LIMIT = 1; + private static final long INITIAL_TICK_RETRY_BACKOFF_MILLIS = 100L; + private static final long MAXIMUM_TICK_RETRY_BACKOFF_MILLIS = 1_000L; private final Transport _transport; private final SslContextFactory.Server _sslContextFactory; private final AmqpPort<?> _port; private final Broker<?> _broker; private final MultiVersionProtocolEngineFactory _factory; + private final LongSupplier _currentTimeSupplier; private Server _server; @@ -102,17 +111,29 @@ class WebSocketProvider implements AcceptingTransport private final WebSocketIdleTimeoutChecker _idleTimeoutChecker = new WebSocketIdleTimeoutChecker(); private final AtomicBoolean _closed = new AtomicBoolean(); + private boolean _tickSubmissionFailureReported; WebSocketProvider(final Transport transport, final SSLContext sslContext, final AmqpPort<?> port, final Set<Protocol> supported, final Protocol defaultSupportedProtocolReply) + { + this(transport, sslContext, port, supported, defaultSupportedProtocolReply, System::currentTimeMillis); + } + + WebSocketProvider(final Transport transport, + final SSLContext sslContext, + final AmqpPort<?> port, + final Set<Protocol> supported, + final Protocol defaultSupportedProtocolReply, + final LongSupplier currentTimeSupplier) { _transport = transport; _sslContextFactory = transport == Transport.WSS ? createSslContextFactory(port) : null; _port = port; _broker = ((Broker<?>) port.getParent()); + _currentTimeSupplier = Objects.requireNonNull(currentTimeSupplier, "Current time supplier must not be null"); _factory = new MultiVersionProtocolEngineFactory( _broker, @@ -362,8 +383,7 @@ class WebSocketProvider implements AcceptingTransport } _protocolEngine.setNetworkConnection(_connectionWrapper); _protocolEngine.setWorkListener(object -> _server.getThreadPool().execute(() -> _connectionWrapper.doWork())); - _activeConnections.add(_connectionWrapper); - _idleTimeoutChecker.wakeup(); + registerConnection(_connectionWrapper); } @OnWebSocketMessage @@ -478,13 +498,12 @@ class WebSocketProvider implements AcceptingTransport { _protocolEngine.closed(); } - _activeConnections.remove(_connectionWrapper); - _idleTimeoutChecker.wakeup(); + unregisterConnection(_connectionWrapper); _netInputBuffer.dispose(); } } - private class ConnectionWrapper implements ServerNetworkConnection, ByteBufferSender + class ConnectionWrapper implements ServerNetworkConnection, ByteBufferSender { private final Session _connection; private final SocketAddress _localAddress; @@ -493,6 +512,12 @@ class WebSocketProvider implements AcceptingTransport private final MultiVersionProtocolEngine _protocolEngine; private final ThreadPool _threadPool; private final Runnable _tickJob; + // Remains set while the job is queued, blocked on this wrapper's monitor, or executing. + private final AtomicBoolean _tickOutstanding = new AtomicBoolean(); + private final AtomicInteger _consecutiveTickRetries = new AtomicInteger(); + + // Per-connection eligibility prevents unrelated checker wakeups from bypassing retry backoff. + private volatile long _tickNotBeforeTime; private Certificate _certificate; private long _maxWriteIdleMillis; @@ -511,14 +536,94 @@ class WebSocketProvider implements AcceptingTransport _threadPool = threadPool; _tickJob = () -> { - synchronized (ConnectionWrapper.this) + long nextTickTime = 0L; + boolean tickSucceeded = false; + try + { + nextTickTime = processTick(); + tickSucceeded = true; + } + finally { - protocolEngine.getAggregateTicker().tick(System.currentTimeMillis()); - doWrite(); + final long currentTime = _currentTimeSupplier.getAsLong(); + final long tickRetryTime = tickSucceeded + ? processTickResult(nextTickTime, currentTime) + : deferTick(currentTime, false); + _tickOutstanding.set(false); + _idleTimeoutChecker.tickCompleted(tickRetryTime); } }; } + private long processTick() + { + synchronized (this) + { + final Ticker ticker = _protocolEngine.getAggregateTicker(); + ticker.tick(_currentTimeSupplier.getAsLong()); + doWrite(); + + final long currentTime = _currentTimeSupplier.getAsLong(); + final long timeToNextTick = ticker.getTimeToNextTick(currentTime); + return timeToNextTick <= 0L ? currentTime : currentTime + timeToNextTick; + } + } + + private long processTickResult(final long nextTickTime, final long currentTime) + { + if (nextTickTime > currentTime) + { + resetTickRetry(); + return nextTickTime; + } + + return deferTick(currentTime, true); + } + + private long deferTick(final long currentTime, final boolean immediateRetryAllowed) + { + final int retryCount = _consecutiveTickRetries.incrementAndGet(); + final int backoffRetryCount = immediateRetryAllowed + ? retryCount - IMMEDIATE_OVERDUE_TICK_RETRY_LIMIT + : retryCount; + final long retryDelay = backoffRetryCount <= 0 ? 0L : calculateTickRetryBackoff(backoffRetryCount); + final long retryTime = currentTime > Long.MAX_VALUE - retryDelay + ? Long.MAX_VALUE + : currentTime + retryDelay; + _tickNotBeforeTime = retryTime; + return retryTime; + } + + private long calculateTickRetryBackoff(final int retryCount) + { + long retryDelay = INITIAL_TICK_RETRY_BACKOFF_MILLIS; + for (int i = 1; i < retryCount && retryDelay < MAXIMUM_TICK_RETRY_BACKOFF_MILLIS; i++) + { + retryDelay = Math.min(retryDelay * 2L, MAXIMUM_TICK_RETRY_BACKOFF_MILLIS); + } + return retryDelay; + } + + private void resetTickRetry() + { + _consecutiveTickRetries.set(0); + _tickNotBeforeTime = 0L; + } + + long getTickRetryDelay(final long currentTime) + { + final long tickNotBeforeTime = _tickNotBeforeTime; + return tickNotBeforeTime > currentTime ? tickNotBeforeTime - currentTime : 0L; + } + + private void tickNoLongerOverdue() + { + if (_tickNotBeforeTime != 0L) + { + resetTickRetry(); + } + } + @Override public ByteBufferSender getSender() { @@ -692,20 +797,141 @@ class WebSocketProvider implements AcceptingTransport { _protocolEngine.setIOThread(null); } - } - - public void tick() + boolean tryScheduleTick(final long currentTime) { - _threadPool.execute(_tickJob); + if (getTickRetryDelay(currentTime) > 0L) + { + return false; + } + + if (_tickOutstanding.compareAndSet(false, true)) + { + boolean submissionAttempted = false; + boolean submitted = false; + try + { + // Completion publishes its retry deadline before releasing the outstanding guard. + if (getTickRetryDelay(_currentTimeSupplier.getAsLong()) > 0L) + { + return false; + } + + submissionAttempted = true; + _threadPool.execute(_tickJob); + submitted = true; + return true; + } + finally + { + if (!submitted) + { + if (submissionAttempted) + { + final long tickRetryTime = deferTick(_currentTimeSupplier.getAsLong(), false); + _tickOutstanding.set(false); + _idleTimeoutChecker.tickCompleted(tickRetryTime); + } + else + { + _tickOutstanding.set(false); + } + } + } + } + return false; } } + void registerConnection(final ConnectionWrapper connection) + { + _activeConnections.add(connection); + _idleTimeoutChecker.wakeup(); + } + private void unregisterConnection(final ConnectionWrapper connection) + { + _activeConnections.remove(connection); + _idleTimeoutChecker.wakeup(); + } + + long scheduleDueConnections(final long currentTime) + { + long timeToNextTick = Long.MAX_VALUE; + RejectedExecutionException submissionFailure = null; + boolean dueConnectionFound = false; + boolean tickSubmissionSucceeded = false; + for (final ConnectionWrapper connection : _activeConnections) + { + final ProtocolEngine engine = connection._protocolEngine; + final Ticker ticker = engine.getAggregateTicker(); + final long timeToTick = ticker.getTimeToNextTick(currentTime); + if (timeToTick <= 0) + { + dueConnectionFound = true; + final long retryDelay = connection.getTickRetryDelay(currentTime); + if (retryDelay > 0L) + { + timeToNextTick = Math.min(timeToNextTick, retryDelay); + } + else + { + try + { + final boolean tickScheduled = connection.tryScheduleTick(currentTime); + tickSubmissionSucceeded |= tickScheduled; + if (!tickScheduled) + { + final long updatedRetryDelay = connection.getTickRetryDelay(currentTime); + if (updatedRetryDelay > 0L) + { + timeToNextTick = Math.min(timeToNextTick, updatedRetryDelay); + } + } + } + catch (final RejectedExecutionException e) + { + if (submissionFailure == null) + { + submissionFailure = e; + } + final long submissionRetryDelay = connection.getTickRetryDelay(currentTime); + timeToNextTick = Math.min(timeToNextTick, submissionRetryDelay); + } + } + } + else + { + connection.tickNoLongerOverdue(); + if (timeToTick < timeToNextTick) + { + timeToNextTick = timeToTick; + } + } + } + if (submissionFailure == null) + { + if (tickSubmissionSucceeded || !dueConnectionFound) + { + _tickSubmissionFailureReported = false; + } + } + else if (!_closed.get() && !_tickSubmissionFailureReported) + { + _tickSubmissionFailureReported = true; + LOGGER.warn("Failed to schedule WebSocket connection idle timeout processing; " + + "repeated failures will not be reported until scheduling recovers", submissionFailure); + } + return timeToNextTick; + } private class WebSocketIdleTimeoutChecker extends Thread { + // Prevents a notification between scanning the tickers and entering wait() from being lost. + private long _wakeupSequence; + // Earliest deadline reported by a tick job that completed after the current scan began. + private long _tickCompletionTime = Long.MAX_VALUE; public WebSocketIdleTimeoutChecker() { @@ -715,51 +941,66 @@ class WebSocketProvider implements AcceptingTransport @Override public void run() { - while(!_closed.get()) + while (!_closed.get()) { - ConnectionWrapper connectionToTick = null; - long currentTime = System.currentTimeMillis(); - synchronized (this) + final long wakeupSequence = beginScan(); + final long timeToNextTick = scheduleDueConnections(_currentTimeSupplier.getAsLong()); + try { - long nextTick = Long.MAX_VALUE; - for(ConnectionWrapper connection : _activeConnections) - { - ProtocolEngine engine = connection._protocolEngine; - final Ticker ticker = engine.getAggregateTicker(); - long tick = ticker.getTimeToNextTick(currentTime); - if(tick <= 0) - { - connectionToTick = connection; - nextTick = -1; - break; - } - else if(tick < nextTick) - { - nextTick = tick; - } - } - if(nextTick > 0) - { - try - { - wait(nextTick); - } - catch (InterruptedException e) - { - Thread.currentThread().interrupt(); - break; - } - } + awaitNextTick(wakeupSequence, timeToNextTick); + } + catch (final InterruptedException e) + { + Thread.currentThread().interrupt(); + break; } - if(connectionToTick != null) + } + } + + private synchronized long beginScan() + { + _tickCompletionTime = Long.MAX_VALUE; + return _wakeupSequence; + } + + private synchronized void awaitNextTick(final long wakeupSequence, final long timeToNextTick) + throws InterruptedException + { + final long currentTime = _currentTimeSupplier.getAsLong(); + final long scanTime = timeToNextTick == Long.MAX_VALUE + ? Long.MAX_VALUE + : currentTime + Math.max(0L, timeToNextTick); + while (!_closed.get() && wakeupSequence == _wakeupSequence) + { + final long wakeupTime = Math.min(scanTime, _tickCompletionTime); + if (wakeupTime == Long.MAX_VALUE) { - connectionToTick.tick(); + wait(); } + else + { + final long waitTime = wakeupTime - _currentTimeSupplier.getAsLong(); + if (waitTime <= 0L) + { + break; + } + wait(waitTime); + } + } + } + + private synchronized void tickCompleted(final long nextTickTime) + { + if (nextTickTime < _tickCompletionTime) + { + _tickCompletionTime = nextTickTime; + notifyAll(); } } private synchronized void wakeup() { + _wakeupSequence++; notifyAll(); } } diff --git a/broker-plugins/websocket/src/test/java/org/apache/qpid/server/transport/websocket/WebSocketProviderTest.java b/broker-plugins/websocket/src/test/java/org/apache/qpid/server/transport/websocket/WebSocketProviderTest.java new file mode 100644 index 0000000000..d791b5be02 --- /dev/null +++ b/broker-plugins/websocket/src/test/java/org/apache/qpid/server/transport/websocket/WebSocketProviderTest.java @@ -0,0 +1,548 @@ +/* + * + * 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.websocket; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.net.InetSocketAddress; +import java.util.ArrayDeque; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.FutureTask; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import org.eclipse.jetty.util.thread.ThreadPool; +import org.eclipse.jetty.websocket.api.Session; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.apache.qpid.server.model.Broker; +import org.apache.qpid.server.model.Protocol; +import org.apache.qpid.server.model.Transport; +import org.apache.qpid.server.model.port.AmqpPort; +import org.apache.qpid.server.transport.AggregateTicker; +import org.apache.qpid.server.transport.MultiVersionProtocolEngine; +import org.apache.qpid.server.transport.network.Ticker; +import org.apache.qpid.test.utils.UnitTestBase; + +class WebSocketProviderTest extends UnitTestBase +{ + private static final int SUBMISSION_COUNT = 100_000; + private static final long INITIAL_TIME = 1_000_000L; + + private TestThreadPool _threadPool; + private WebSocketProvider _provider; + private AtomicLong _currentTime; + + @BeforeEach + @SuppressWarnings({"rawtypes", "unchecked"}) + public void setUp() + { + _threadPool = new TestThreadPool(); + _currentTime = new AtomicLong(INITIAL_TIME); + + final Broker<?> broker = mock(Broker.class); + final AmqpPort<?> port = mock(AmqpPort.class); + when(port.getParent()).thenReturn((Broker) broker); + + _provider = new WebSocketProvider(Transport.WS, null, port, Set.of(Protocol.AMQP_1_0), + Protocol.AMQP_1_0, _currentTime::get); + } + + @Test + public void testTickSubmissionsAreCoalescedWhileQueued() + { + final WebSocketProvider.ConnectionWrapper connection = createConnection(new MutableTicker(-1)); + + int accepted = 0; + for (int i = 0; i < SUBMISSION_COUNT; i++) + { + if (tryScheduleTick(connection)) + { + accepted++; + } + } + + assertEquals(1, accepted); + assertEquals(1, _threadPool.getQueueSize()); + } + + @Test + public void testTickRemainsCoalescedUntilJobCompletes() + { + final MutableTicker ticker = new MutableTicker(-1); + final WebSocketProvider.ConnectionWrapper connection = createConnection(ticker); + ticker.setOnTick(() -> assertFalse(tryScheduleTick(connection))); + + assertTrue(tryScheduleTick(connection)); + _threadPool.runNext(); + + assertEquals(1, ticker.getTickCount()); + assertEquals(0, _threadPool.getQueueSize()); + + assertTrue(tryScheduleTick(connection)); + assertEquals(1, _threadPool.getQueueSize()); + _threadPool.runNext(); + assertEquals(2, ticker.getTickCount()); + } + + @Test + public void testCompletionBackoffCannotBeBypassedBetweenEligibilityCheckAndClaim() throws Exception + { + final MutableTicker ticker = new MutableTicker(-1); + final BlockingConnectionWrapper connection = createBlockingConnection(ticker); + + assertTrue(tryScheduleTick(connection)); + _threadPool.runNext(); + + final CountDownLatch tickStarted = new CountDownLatch(1); + final CountDownLatch allowTickCompletion = new CountDownLatch(1); + ticker.setOnTick(() -> + { + tickStarted.countDown(); + await(allowTickCompletion, "Timed out waiting to complete the tick job"); + }); + + assertTrue(tryScheduleTick(connection)); + final FutureTask<Void> tickTask = new FutureTask<>(() -> + { + _threadPool.runNext(); + return null; + }); + startDaemonThread(tickTask, "tick"); + + final FutureTask<Boolean> schedulingTask = new FutureTask<>(() -> tryScheduleTick(connection)); + try + { + await(tickStarted, "Timed out waiting for the tick job to start"); + connection.blockNextRetryDelayRead(); + startDaemonThread(schedulingTask, "schedule"); + connection.awaitRetryDelayRead(); + + allowTickCompletion.countDown(); + tickTask.get(10L, TimeUnit.SECONDS); + connection.continueRetryDelayRead(); + } + finally + { + allowTickCompletion.countDown(); + connection.continueRetryDelayRead(); + } + + assertFalse(schedulingTask.get(10L, TimeUnit.SECONDS)); + assertEquals(0, _threadPool.getQueueSize()); + + _currentTime.addAndGet(100L); + assertTrue(tryScheduleTick(connection)); + assertEquals(1, _threadPool.getQueueSize()); + } + + @Test + public void testTickRefreshesNextDeadlineAfterProcessing() + { + final MutableTicker ticker = new MutableTicker(-1); + final WebSocketProvider.ConnectionWrapper connection = createConnection(ticker); + ticker.setOnTick(() -> ticker.setTimeToNextTick(37_000)); + + assertTrue(tryScheduleTick(connection)); + _threadPool.runNext(); + + assertEquals(1, ticker.getTickCount()); + assertEquals(1, ticker.getTimeToNextTickCallCount()); + } + + @Test + public void testSubmissionFailureDoesNotLeaveTickOutstanding() + { + final WebSocketProvider.ConnectionWrapper connection = createConnection(new MutableTicker(-1)); + _threadPool.rejectNext(); + + assertThrows(RejectedExecutionException.class, () -> tryScheduleTick(connection)); + + assertFalse(tryScheduleTick(connection)); + _currentTime.addAndGet(100L); + assertTrue(tryScheduleTick(connection)); + assertEquals(1, _threadPool.getQueueSize()); + } + + @Test + public void testSubmissionErrorDoesNotLeaveTickOutstanding() + { + final WebSocketProvider.ConnectionWrapper connection = createConnection(new MutableTicker(-1)); + _threadPool.failNextWithError(); + + assertThrows(AssertionError.class, () -> tryScheduleTick(connection)); + + assertFalse(tryScheduleTick(connection)); + _currentTime.addAndGet(100L); + assertTrue(tryScheduleTick(connection)); + assertEquals(1, _threadPool.getQueueSize()); + } + + @Test + public void testTickFailureDoesNotLeaveTickOutstanding() + { + final MutableTicker ticker = new MutableTicker(-1); + final WebSocketProvider.ConnectionWrapper connection = createConnection(ticker); + ticker.setOnTick(() -> + { + throw new IllegalStateException("Ticker failure"); + }); + + assertTrue(tryScheduleTick(connection)); + assertThrows(IllegalStateException.class, _threadPool::runNext); + + ticker.setOnTick(() -> { }); + assertFalse(tryScheduleTick(connection)); + _currentTime.addAndGet(100L); + assertTrue(tryScheduleTick(connection)); + assertEquals(1, _threadPool.getQueueSize()); + } + + @Test + public void testEveryDueConnectionIsScheduled() + { + final WebSocketProvider.ConnectionWrapper first = createConnection(new MutableTicker(-1)); + final WebSocketProvider.ConnectionWrapper second = createConnection(new MutableTicker(0)); + final WebSocketProvider.ConnectionWrapper future = createConnection(new MutableTicker(37_000)); + _provider.registerConnection(first); + _provider.registerConnection(second); + _provider.registerConnection(future); + + final long timeToNextTick = scheduleDueConnections(); + + assertEquals(37_000, timeToNextTick); + assertEquals(2, _threadPool.getQueueSize()); + } + + @Test + public void testPendingDueConnectionIsNotResubmitted() + { + final WebSocketProvider.ConnectionWrapper connection = createConnection(new MutableTicker(-1)); + _provider.registerConnection(connection); + + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + + assertEquals(1, _threadPool.getQueueSize()); + } + + @Test + public void testRejectedSubmissionDoesNotPreventOtherDueConnectionsFromBeingScheduled() + { + final WebSocketProvider.ConnectionWrapper first = createConnection(new MutableTicker(-1)); + final WebSocketProvider.ConnectionWrapper second = createConnection(new MutableTicker(-1)); + _provider.registerConnection(first); + _provider.registerConnection(second); + _threadPool.rejectNext(); + + final long timeToNextTick = scheduleDueConnections(); + + assertEquals(100L, timeToNextTick); + assertEquals(1, _threadPool.getQueueSize()); + assertEquals(100L, scheduleDueConnections()); + assertEquals(1, _threadPool.getQueueSize()); + assertFalse(tryScheduleTick(first)); + _currentTime.addAndGet(100L); + assertTrue(tryScheduleTick(first)); + assertEquals(2, _threadPool.getQueueSize()); + } + + @Test + public void testOverdueTickRetryBackoffCannotBeBypassedByRepeatedScans() + { + final MutableTicker ticker = new MutableTicker(-1); + final WebSocketProvider.ConnectionWrapper connection = createConnection(ticker); + _provider.registerConnection(connection); + + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + _threadPool.runNext(); + + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + _threadPool.runNext(); + + for (int i = 0; i < 10; i++) + { + assertEquals(100L, scheduleDueConnections()); + } + assertEquals(0, _threadPool.getQueueSize()); + + final long[] retryDelays = {100L, 200L, 400L, 800L, 1_000L, 1_000L}; + for (final long retryDelay : retryDelays) + { + assertEquals(retryDelay, scheduleDueConnections()); + _currentTime.addAndGet(retryDelay); + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + _threadPool.runNext(); + } + + assertEquals(1_000L, scheduleDueConnections()); + assertEquals(8, ticker.getTickCount()); + assertEquals(0, _threadPool.getQueueSize()); + } + + @Test + public void testFutureTickerDeadlineClearsRetryBackoff() + { + final MutableTicker ticker = new MutableTicker(-1); + final WebSocketProvider.ConnectionWrapper connection = createConnection(ticker); + _provider.registerConnection(connection); + + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + _threadPool.runNext(); + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + _threadPool.runNext(); + assertEquals(100L, scheduleDueConnections()); + + ticker.setTimeToNextTick(37_000); + assertEquals(37_000L, scheduleDueConnections()); + + ticker.setTimeToNextTick(-1); + assertEquals(Long.MAX_VALUE, scheduleDueConnections()); + assertEquals(1, _threadPool.getQueueSize()); + } + + private boolean tryScheduleTick(final WebSocketProvider.ConnectionWrapper connection) + { + return connection.tryScheduleTick(_currentTime.get()); + } + + private long scheduleDueConnections() + { + return _provider.scheduleDueConnections(_currentTime.get()); + } + + private WebSocketProvider.ConnectionWrapper createConnection(final Ticker ticker) + { + final AggregateTicker aggregateTicker = new AggregateTicker(); + aggregateTicker.addTicker(ticker); + + final MultiVersionProtocolEngine protocolEngine = mock(MultiVersionProtocolEngine.class); + when(protocolEngine.getAggregateTicker()).thenReturn(aggregateTicker); + + final InetSocketAddress address = new InetSocketAddress("localhost", 0); + return _provider.new ConnectionWrapper(mock(Session.class), address, address, protocolEngine, _threadPool); + } + + private BlockingConnectionWrapper createBlockingConnection(final Ticker ticker) + { + final AggregateTicker aggregateTicker = new AggregateTicker(); + aggregateTicker.addTicker(ticker); + + final MultiVersionProtocolEngine protocolEngine = mock(MultiVersionProtocolEngine.class); + when(protocolEngine.getAggregateTicker()).thenReturn(aggregateTicker); + + final InetSocketAddress address = new InetSocketAddress("localhost", 0); + return new BlockingConnectionWrapper(_provider, mock(Session.class), address, address, protocolEngine, + _threadPool); + } + + private void startDaemonThread(final FutureTask<?> task, final String name) + { + final Thread thread = new Thread(task, getTestName() + "-" + name); + thread.setDaemon(true); + thread.start(); + } + + private static void await(final CountDownLatch latch, final String timeoutMessage) + { + try + { + if (!latch.await(10L, TimeUnit.SECONDS)) + { + throw new AssertionError(timeoutMessage); + } + } + catch (final InterruptedException e) + { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } + } + + private static final class BlockingConnectionWrapper extends WebSocketProvider.ConnectionWrapper + { + private final AtomicBoolean _blockNextRetryDelayRead = new AtomicBoolean(); + private final CountDownLatch _retryDelayRead = new CountDownLatch(1); + private final CountDownLatch _continueRetryDelayRead = new CountDownLatch(1); + + private BlockingConnectionWrapper(final WebSocketProvider provider, + final Session connection, + final InetSocketAddress localAddress, + final InetSocketAddress remoteAddress, + final MultiVersionProtocolEngine protocolEngine, + final ThreadPool threadPool) + { + provider.super(connection, localAddress, remoteAddress, protocolEngine, threadPool); + } + + @Override + long getTickRetryDelay(final long currentTime) + { + final long retryDelay = super.getTickRetryDelay(currentTime); + if (_blockNextRetryDelayRead.compareAndSet(true, false)) + { + _retryDelayRead.countDown(); + await(_continueRetryDelayRead, "Timed out waiting to continue the retry-delay read"); + } + return retryDelay; + } + + private void blockNextRetryDelayRead() + { + _blockNextRetryDelayRead.set(true); + } + + private void awaitRetryDelayRead() + { + await(_retryDelayRead, "Timed out waiting for the retry-delay read"); + } + + private void continueRetryDelayRead() + { + _continueRetryDelayRead.countDown(); + } + } + + private static final class MutableTicker implements Ticker + { + private final AtomicInteger _tickCount = new AtomicInteger(); + private final AtomicInteger _timeToNextTickCallCount = new AtomicInteger(); + private volatile int _timeToNextTick; + private Runnable _onTick = () -> { }; + + private MutableTicker(final int timeToNextTick) + { + _timeToNextTick = timeToNextTick; + } + + @Override + public int getTimeToNextTick(final long currentTime) + { + _timeToNextTickCallCount.incrementAndGet(); + return _timeToNextTick; + } + + @Override + public int tick(final long currentTime) + { + _tickCount.incrementAndGet(); + _onTick.run(); + return _timeToNextTick; + } + + private void setOnTick(final Runnable onTick) + { + _onTick = onTick; + } + + private void setTimeToNextTick(final int timeToNextTick) + { + _timeToNextTick = timeToNextTick; + } + + private int getTickCount() + { + return _tickCount.get(); + } + + private int getTimeToNextTickCallCount() + { + return _timeToNextTickCallCount.get(); + } + } + + private static final class TestThreadPool implements ThreadPool + { + private final Queue<Runnable> _jobs = new ArrayDeque<>(); + private boolean _rejectNext; + private boolean _failNextWithError; + + @Override + public void execute(final Runnable job) + { + if (_rejectNext) + { + _rejectNext = false; + throw new RejectedExecutionException("Rejected for test"); + } + if (_failNextWithError) + { + _failNextWithError = false; + throw new AssertionError("Failed for test"); + } + _jobs.add(job); + } + + @Override + public void join() + { + } + + @Override + public int getThreads() + { + return 0; + } + + @Override + public int getIdleThreads() + { + return 0; + } + + @Override + public boolean isLowOnThreads() + { + return false; + } + + private void rejectNext() + { + _rejectNext = true; + } + + private void failNextWithError() + { + _failNextWithError = true; + } + + private int getQueueSize() + { + return _jobs.size(); + } + + private void runNext() + { + _jobs.remove().run(); + } + } +} diff --git a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketFrameTransport.java b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketFrameTransport.java index 12b62e5c62..93ff8674dc 100644 --- a/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketFrameTransport.java +++ b/systests/protocol-tests-amqp-1-0/src/main/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketFrameTransport.java @@ -37,12 +37,14 @@ import io.netty.channel.SimpleChannelInboundHandler; import io.netty.handler.codec.http.DefaultHttpHeaders; import io.netty.handler.codec.http.FullHttpResponse; import io.netty.handler.codec.http.HttpClientCodec; +import io.netty.handler.codec.http.HttpHeaderNames; import io.netty.handler.codec.http.HttpObjectAggregator; import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker; import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory; import io.netty.handler.codec.http.websocketx.WebSocketFrame; import io.netty.handler.codec.http.websocketx.WebSocketVersion; +import io.netty.handler.codec.http.websocketx.extensions.compression.WebSocketClientCompressionHandler; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -52,20 +54,28 @@ import org.apache.qpid.tests.utils.BrokerAdmin; public class WebSocketFrameTransport extends FrameTransport { private static final Logger LOGGER = LoggerFactory.getLogger(WebSocketFrameTransport.class); + private static final int MAX_DECOMPRESSED_PAYLOAD_SIZE = 65_536; private final WebSocketFramingOutputHandler _webSocketFramingOutputHandler = new WebSocketFramingOutputHandler(); private final WebSocketDeframingInputHandler _webSocketDeframingInputHandler = new WebSocketDeframingInputHandler(); private final WebSocketClientHandler _webSocketClientHandler; + private final boolean _compressionEnabled; public WebSocketFrameTransport(final BrokerAdmin brokerAdmin) + { + this(brokerAdmin, false); + } + + WebSocketFrameTransport(final BrokerAdmin brokerAdmin, final boolean compressionEnabled) { super(brokerAdmin, BrokerAdmin.PortType.ANONYMOUS_AMQPWS); - URI uri = URI.create(String.format("tcp://%s:%d/", - getBrokerAddress().getHostString(), - getBrokerAddress().getPort())); + _compressionEnabled = compressionEnabled; + final URI uri = URI.create(String.format("tcp://%s:%d/", + getBrokerAddress().getHostString(), + getBrokerAddress().getPort())); _webSocketClientHandler = new WebSocketClientHandler( WebSocketClientHandshakerFactory.newHandshaker( - uri, WebSocketVersion.V13, "amqp", false, new DefaultHttpHeaders())); + uri, WebSocketVersion.V13, "amqp", compressionEnabled, new DefaultHttpHeaders())); } @Override @@ -73,6 +83,10 @@ public class WebSocketFrameTransport extends FrameTransport { pipeline.addLast(new HttpClientCodec()); pipeline.addLast(new HttpObjectAggregator(65536)); + if (_compressionEnabled) + { + pipeline.addLast(new WebSocketClientCompressionHandler(MAX_DECOMPRESSED_PAYLOAD_SIZE)); + } pipeline.addLast(_webSocketClientHandler); pipeline.addLast(_webSocketFramingOutputHandler); pipeline.addLast(_webSocketDeframingInputHandler); @@ -93,6 +107,11 @@ public class WebSocketFrameTransport extends FrameTransport return this; } + String getNegotiatedExtensions() + { + return _webSocketClientHandler.getNegotiatedExtensions(); + } + private static class WebSocketFramingOutputHandler extends ChannelOutboundHandlerAdapter { private boolean _splitFrames; @@ -169,6 +188,7 @@ public class WebSocketFrameTransport extends FrameTransport private final WebSocketClientHandshaker _handshaker; private ChannelPromise _handshakeFuture; + private volatile String _negotiatedExtensions; WebSocketClientHandler(final WebSocketClientHandshaker handshaker) { @@ -180,6 +200,11 @@ public class WebSocketFrameTransport extends FrameTransport return _handshakeFuture; } + String getNegotiatedExtensions() + { + return _negotiatedExtensions; + } + @Override public void handlerAdded(final ChannelHandlerContext ctx) { @@ -199,7 +224,9 @@ public class WebSocketFrameTransport extends FrameTransport if (!_handshaker.isHandshakeComplete()) { // web socket client connected - _handshaker.finishHandshake(ch, (FullHttpResponse) msg); + final FullHttpResponse response = (FullHttpResponse) msg; + _negotiatedExtensions = response.headers().get(HttpHeaderNames.SEC_WEBSOCKET_EXTENSIONS); + _handshaker.finishHandshake(ch, response); _handshakeFuture.setSuccess(); return; } diff --git a/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketIdleTimeoutTest.java b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketIdleTimeoutTest.java new file mode 100644 index 0000000000..af28c6557a --- /dev/null +++ b/systests/protocol-tests-amqp-1-0/src/test/java/org/apache/qpid/tests/protocol/v1_0/extensions/websocket/WebSocketIdleTimeoutTest.java @@ -0,0 +1,101 @@ +/* + * + * 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.tests.protocol.v1_0.extensions.websocket; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.is; +import static org.junit.jupiter.api.Assumptions.assumeTrue; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import org.apache.qpid.server.model.port.AmqpPort; +import org.apache.qpid.server.protocol.v1_0.type.transport.Open; +import org.apache.qpid.tests.protocol.ChannelClosedResponse; +import org.apache.qpid.tests.protocol.SpecificationTest; +import org.apache.qpid.tests.protocol.v1_0.EmptyResponse; +import org.apache.qpid.tests.protocol.v1_0.Interaction; +import org.apache.qpid.tests.utils.BrokerAdmin; +import org.apache.qpid.tests.utils.BrokerAdminUsingTestBase; +import org.apache.qpid.tests.utils.BrokerSpecific; +import org.apache.qpid.tests.utils.ConfigItem; + +@BrokerSpecific(kind = BrokerAdmin.KIND_BROKER_J) +@ConfigItem(name = AmqpPort.HEART_BEAT_DELAY, value = WebSocketIdleTimeoutTest.IDLE_SECONDS) +public class WebSocketIdleTimeoutTest extends BrokerAdminUsingTestBase +{ + static final String IDLE_SECONDS = "1"; + private static final int IDLE_TIMEOUT_MILLIS = Integer.parseInt(IDLE_SECONDS) * 1000; + + @BeforeEach + public void setUp() + { + assumeTrue(getBrokerAdmin().isWebSocketSupported(), "Broker support for AMQP over WebSocket is required"); + assumeTrue(getBrokerAdmin().isAnonymousSupported(), "Broker support for anonymous open is required"); + } + + @Test + @SpecificationTest(section = "2.4.5", + description = "If the idle timeout threshold is exceeded, a peer SHOULD try to close the connection.") + public void brokerClosesIdleWebSocketConnection() throws Exception + { + try (WebSocketFrameTransport transport = new WebSocketFrameTransport(getBrokerAdmin()).connect()) + { + final Interaction interaction = transport.newInteraction(); + final Open responseOpen = interaction + .openContainerId("testContainerId") + .negotiateOpen() + .getLatestResponse(Open.class); + assertThat(responseOpen.getIdleTimeOut().intValue(), is(equalTo(IDLE_TIMEOUT_MILLIS))); + + // TODO: The broker ought to send a close performative, but currently closes the transport directly. + interaction.consumeResponse().getLatestResponse(ChannelClosedResponse.class); + } + } + + @Test + @SpecificationTest(section = "2.4.5", + description = "A peer with nothing to send MAY send an empty frame to prevent idle timeout.") + public void compressedWebSocketReceivesIdleFrames() throws Exception + { + try (WebSocketFrameTransport transport = new WebSocketFrameTransport(getBrokerAdmin(), true).connect()) + { + assertThat(transport.getNegotiatedExtensions(), containsString("permessage-deflate")); + + final Interaction interaction = transport.newInteraction(); + final Open responseOpen = interaction + .openContainerId("testContainerId") + .openIdleTimeOut(IDLE_TIMEOUT_MILLIS) + .negotiateOpen() + .getLatestResponse(Open.class); + assertThat(responseOpen.getIdleTimeOut().intValue(), is(equalTo(IDLE_TIMEOUT_MILLIS))); + + interaction.consumeResponse(EmptyResponse.class) + .emptyFrame(); + interaction.consumeResponse(EmptyResponse.class) + .emptyFrame(); + + interaction.doCloseConnection(); + } + } +} --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
