diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java index 04a47ad3aa4..4d70d3f6cf1 100644 --- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java +++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/QueueImpl.java @@ -2077,7 +2077,7 @@ private int iterQueue(final String operationName, try { Transaction tx = new TransactionImpl(storageManager); - synchronized (this) { + synchronized (QueueImpl.this) { // ensure all messages are moved from intermediateMessageReferences so that they can be seen by the iterator doInternalPoll(); @@ -2223,7 +2223,7 @@ public void destroyPaging() throws Exception { } @Override - public synchronized boolean deleteReference(final long messageID) throws Exception { + public boolean deleteReference(final long messageID) throws Exception { return iterQueue("deleteReference", DEFAULT_FLUSH_LIMIT, null, new QueueIterateAction(messageID) { @Override public boolean actMessage(Transaction tx, MessageReference ref) throws Exception { diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java index 09ddd9006bc..d421fd8d8a7 100644 --- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java +++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/management/ManagementWithPagingServerTest.java @@ -20,17 +20,28 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; import org.apache.activemq.artemis.api.core.RoutingType; +import org.apache.activemq.artemis.core.server.Queue; import org.apache.activemq.artemis.json.JsonArray; import org.apache.activemq.artemis.json.JsonNumber; import org.apache.activemq.artemis.json.JsonObject; import org.apache.activemq.artemis.json.JsonValue; + +import java.lang.invoke.MethodHandles; +import java.lang.management.ManagementFactory; +import java.lang.management.ThreadInfo; +import java.lang.management.ThreadMXBean; import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import org.apache.activemq.artemis.api.core.ActiveMQBuffer; import org.apache.activemq.artemis.api.core.JsonUtil; @@ -51,16 +62,21 @@ import org.apache.activemq.artemis.utils.RandomUtil; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * This class contains tests for core management functionalities that are affected by a server in paging mode. */ public class ManagementWithPagingServerTest extends ManagementTestBase { + private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + private ActiveMQServer server; private ClientSession session1; private ClientSession session2; private ServerLocator locator; + private ClientSessionFactory sf; @Test public void testListMessagesAsJSON() throws Exception { @@ -72,15 +88,16 @@ public void testListMessagesAsJSON() throws Exception { QueueControl queueControl = createManagementControl(address, queue); int num = 1000; - SenderThread sender = new SenderThread(address, num, 0); - - ReceiverThread receiver = new ReceiverThread(queue, num, 0); + SenderRunnable sender = new SenderRunnable(address, num, 0); + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 0); - //kick off sender - sender.start(); + ExecutorService executorService = Executors.newFixedThreadPool(1); + runAfter(executorService::shutdownNow); + runAfter(sender::stop); + runAfter(receiver::stop); - //wait for all messages sent - sender.join(); + executorService.execute(sender); + sender.waitDone(); assertNull(sender.getError()); long count = queueControl.countMessages(null); @@ -98,9 +115,8 @@ public void testListMessagesAsJSON() throws Exception { } assertEquals(num, array.size()); - //kick off receiver - receiver.start(); - receiver.join(); + executorService.execute(receiver); + receiver.waitDone(); assertNull(receiver.getError()); result = queueControl.listMessagesAsJSON(null); @@ -154,10 +170,12 @@ public void testListMessagesAsJSONWithFilter() throws Exception { long n = queueControl.countMessages(filter); assertEquals(num / 2, n); - //drain out messages - ReceiverThread receiver = new ReceiverThread(queue, num, 1); - receiver.start(); - receiver.join(); + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 1); + ExecutorService executorService = Executors.newFixedThreadPool(1); + runAfter(executorService::shutdownNow); + runAfter(receiver::stop); + executorService.execute(receiver); + receiver.waitDone(); } //In this test, the management api listMessageAsJSon is called while @@ -174,30 +192,29 @@ public void testListMessagesAsJSONWhilePagingOnGoing() throws Exception { QueueControl queueControl = createManagementControl(address, queue); int num = 1000; - SenderThread sender = new SenderThread(address, num, 1); + SenderRunnable sender = new SenderRunnable(address, num, 1); + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 2); + ManagementRunnable console = new ManagementRunnable(queueControl); - ReceiverThread receiver = new ReceiverThread(queue, num, 2); + ExecutorService executorService = Executors.newFixedThreadPool(2); + runAfter(executorService::shutdownNow); + runAfter(sender::stop); + runAfter(receiver::stop); + runAfter(console::stop); - ManagementThread console = new ManagementThread(queueControl); + executorService.execute(sender); + executorService.execute(console); - //kick off sender - sender.start(); - - //kick off jmx client - console.start(); - - //wait for all messages sent - sender.join(); + sender.waitDone(); assertNull(sender.getError()); - //kick off receiver - receiver.start(); + executorService.execute(receiver); - receiver.join(); + receiver.waitDone(); assertNull(receiver.getError()); - console.exit(); - console.join(); + console.stop(); + console.waitDone(); assertNull(console.getError()); } @@ -300,7 +317,7 @@ public void testMoveMessageWhilstPagingAndConsuming() throws Exception { QueueControl otherQueueControl = createManagementControl(otherAddress, otherQueue); - int num = 1000; + int num = 10; ClientProducer producer = session1.createProducer(address); for (int i = 0; i < num; i++) { @@ -308,20 +325,92 @@ public void testMoveMessageWhilstPagingAndConsuming() throws Exception { producer.send(message); } - ManagementCopyThread console = new ManagementCopyThread(queueControl, otherQueue.toString()); - ReceiverThread receiver = new ReceiverThread(queue, num, 0); - console.start(); - receiver.start(); + ExecutorService executorService = Executors.newFixedThreadPool(2); + runAfter(executorService::shutdownNow); - receiver.join(); - console.stop = true; - console.join(); + ManagementCopyRunnable console = new ManagementCopyRunnable(queue.toString(), queueControl, otherQueue.toString()); + runAfter(console::stop); + + ReceiverRunnable receiver = new ReceiverRunnable(queue, num, 0); + runAfter(receiver::stop); + executorService.execute(receiver); + executorService.execute(console); + + receiver.waitDone(); + + console.stop(); + console.waitDone(); Map[] messages = otherQueueControl.listMessages(null); assertEquals(messages.length, console.copiedMessages); } + /** + * Reproduction for ARTEMIS-6179: QueueImpl#deleteReference() is still {@code synchronized} and calls + * iterQueue(), which locks depageLock -- while QueueImpl#depage() locks depageLock first and then enters a + * synchronized(this) block. Racing QueueControl#removeMessage() (-> deleteReference()) against depaging + * (triggered here by the ReceiverThread acking messages) can deadlock the two threads against each other. + * This mirrors testMoveMessageWhilstPagingAndConsuming(), which caught the equivalent bug for copyReference() + * (ARTEMIS-5376), replacing copyMessage with removeMessage. Detection uses the JVM's own deadlock detector + * (ThreadMXBean) instead of a fixed timeout, since a hang here is the failure itself. + */ + @Test + public void testRemoveMessageWhilstPagingAndConsuming() throws Exception { + final int messagesPerIteration = 100; + final int maxIterations = 2; + + SimpleString address = RandomUtil.randomUUIDSimpleString(); + SimpleString queue = RandomUtil.randomUUIDSimpleString(); + + session1.createQueue(QueueConfiguration.of(queue).setAddress(address)); + + QueueControl queueControl = createManagementControl(address, queue); + + ThreadMXBean threadMXBean = ManagementFactory.getThreadMXBean(); + + // A single remover is sufficient: QueueControlImpl#removeMessage() is serialized broker-wide through + // ActiveMQServerImpl's managementLock, so extra concurrent removers would just contend on that lock + // without increasing the odds of racing depage(). + ManagementRemoveThread remover = new ManagementRemoveThread(queueControl); + + ExecutorService executorService = Executors.newFixedThreadPool(3); + runAfter(executorService::shutdownNow); + runAfter(remover::stop); + + executorService.execute(remover); + + long[] deadlockedIds = null; + + for (int iteration = 0; iteration < maxIterations; iteration++) { + logger.info("iteration : {}", iteration); + SenderRunnable sender = new SenderRunnable(address, messagesPerIteration, 0); + ReceiverRunnable receiver = new ReceiverRunnable(queue, messagesPerIteration, 0); + + executorService.execute(sender); + executorService.execute(receiver); + + sender.waitDone(); + receiver.waitDone(); + + deadlockedIds = threadMXBean.findDeadlockedThreads(); + + if (deadlockedIds != null) { + StringBuilder sb = new StringBuilder("Deadlock detected between removeMessage() and depage():\n"); + for (long id : deadlockedIds) { + ThreadInfo info = threadMXBean.getThreadInfo(id, Integer.MAX_VALUE); + sb.append(info).append('\n'); + } + fail(sb.toString()); + } + } + + remover.stop(); + remover.waitDone(); + assertNull(remover.getError()); + + } + @Override @BeforeEach public void setUp() throws Exception { @@ -338,84 +427,115 @@ public void setUp() throws Exception { server.start(); locator = createInVMNonHALocator().setBlockOnNonDurableSend(false).setConsumerWindowSize(0); - ClientSessionFactory sf = createSessionFactory(locator); + sf = createSessionFactory(locator); session1 = sf.createSession(false, true, false); session1.start(); session2 = sf.createSession(false, true, false); session2.start(); } - private class SenderThread extends Thread { + protected abstract class AbstractRunnable implements Runnable { + protected volatile boolean stop = false; + protected volatile Exception error = null; + protected final CountDownLatch done = new CountDownLatch(1); + + public boolean waitDone() throws Exception { + return done.await(10, TimeUnit.SECONDS); + } + + protected void done() { + done.countDown(); + } + + public void stop() { + stop = true; + } + + public Exception getError() { + return this.error; + } + } + + private class SenderRunnable extends AbstractRunnable { private SimpleString address; private int num; private long delay; private volatile Exception error = null; + ClientSession sessionSender; - private SenderThread(SimpleString address, int num, long delay) { + private SenderRunnable(SimpleString address, int num, long delay) throws Exception { this.address = address; this.num = num; this.delay = delay; + sessionSender = sf.createSession(false, false, false); } @Override public void run() { - ClientProducer producer; + try { + ClientProducer producer; - byte[] body = new byte[128]; - ByteBuffer bb = ByteBuffer.wrap(body); - for (int j = 1; j <= 128; j++) { - bb.put(getSamplebyte(j)); - } + byte[] body = new byte[128]; + ByteBuffer bb = ByteBuffer.wrap(body); + for (int j = 1; j <= 128; j++) { + bb.put(getSamplebyte(j)); + } - try { - producer = session1.createProducer(address); - - for (int i = 0; i < num; i++) { - ClientMessage message = session1.createMessage(true); - message.setPriority((byte) 1); - ActiveMQBuffer buffer = message.getBodyBuffer(); - buffer.writeBytes(body); - producer.send(message); - try { - Thread.sleep(delay); - } catch (InterruptedException e) { - //ignore + try { + producer = sessionSender.createProducer(address); + + for (int i = 0; i < num && !stop; i++) { + ClientMessage message = sessionSender.createMessage(true); + message.setPriority((byte) 1); + ActiveMQBuffer buffer = message.getBodyBuffer(); + buffer.writeBytes(body); + producer.send(message); + if ((i + 1) % 100 == 0) { + sessionSender.commit(); + } + try { + Thread.sleep(delay); + } catch (InterruptedException e) { + //ignore + } } + sessionSender.commit(); + } catch (Exception e) { + error = e; } - } catch (Exception e) { - error = e; + } finally { + done(); } } - public Exception getError() { - return this.error; - } } - private class ReceiverThread extends Thread { + private class ReceiverRunnable extends AbstractRunnable { private SimpleString queue; private int num; private long delay; - private volatile Exception error = null; + private ClientSession sessionConsumer; - private ReceiverThread(SimpleString queue, int num, long delay) { + private ReceiverRunnable(SimpleString queue, int num, long delay) throws Exception { this.queue = queue; this.num = num; this.delay = delay; + this.sessionConsumer = sf.createSession(false, true, false); } @Override public void run() { ClientConsumer consumer; try { - consumer = session2.createConsumer(queue); + consumer = sessionConsumer.createConsumer(queue); + sessionConsumer.start(); - for (int i = 0; i < num; i++) { + for (int i = 0; i < num && !stop; i++) { ClientMessage message = consumer.receive(5000); message.acknowledge(); - session2.commit(); + sessionConsumer.commit(); try { Thread.sleep(delay); } catch (InterruptedException e) { @@ -424,21 +544,17 @@ public void run() { } } catch (Exception e) { error = e; + } finally { + done(); } } - - public Exception getError() { - return this.error; - } } - private class ManagementThread extends Thread { + private class ManagementRunnable extends AbstractRunnable { private QueueControl queueControl; - private volatile boolean stop = false; - private Exception error = null; - private ManagementThread(QueueControl queueControl) { + private ManagementRunnable(QueueControl queueControl) { this.queueControl = queueControl; } @@ -456,55 +572,70 @@ public void run() { } } catch (Exception e) { error = e; + } finally { + done(); } } - - public Exception getError() { - return error; - } - - public void exit() { - stop = true; - } } - private class ManagementCopyThread extends Thread { + private class ManagementCopyRunnable extends AbstractRunnable { private QueueControl queueControl; private String queue; - private volatile boolean stop = false; + private String originalQueue; + Queue targetQueue; int copiedMessages = 0; - private Exception error = null; - private ManagementCopyThread(QueueControl queueControl, String queue) { + private ManagementCopyRunnable(String originalQueue, QueueControl queueControl, String queue) { this.queueControl = queueControl; this.queue = queue; + this.originalQueue = originalQueue; + targetQueue = server.locateQueue(queue); } @Override public void run() { try { Random random = new Random(System.currentTimeMillis()); - while (!stop) { + while (!stop && copiedMessages < 100) { long messageID = random.nextInt(1000); boolean copied = queueControl.copyMessage(messageID, queue); - System.out.println("messageID = " + messageID); if (copied) { copiedMessages++; + logger.info("Copied message ID {}, totalCopied so far = {}", messageID, copiedMessages); + targetQueue.forEach(r -> { + logger.info("containing {}", r.getMessage()); + }); } } } catch (Exception e) { error = e; + } finally { + done(); } } + } - public Exception getError() { - return error; + private class ManagementRemoveThread extends AbstractRunnable { + + private QueueControl queueControl; + + private ManagementRemoveThread(QueueControl queueControl) { + this.queueControl = queueControl; } - public void exit() { - stop = true; + @Override + public void run() { + try { + while (!stop) { + queueControl.removeMessage(0); + } + } catch (Exception e) { + error = e; + } finally { + done(); + } } } }