Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

This closes #1624

This commit is contained in:
Justin Bertram 2017-10-31 08:33:44 -05:00
parent 3bdeed80c3 f8a0eec2cf
commit f8a4e0c964
12 changed files with 196 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalAnalyzerImpl.java
View File

@ -147,6 +147,8 @@ public class CriticalAnalyzerImpl implements CriticalAnalyzer {
logger.warn(e.getMessage(), e);
}
}
actions.clear();
}
@Override

4
artemis-commons/src/main/java/org/apache/activemq/artemis/utils/critical/CriticalComponentImpl.java
View File

@ -25,6 +25,10 @@ public class CriticalComponentImpl implements CriticalComponent {
private final CriticalMeasure[] measures;
private final CriticalAnalyzer analyzer;
public CriticalAnalyzer getCriticalAnalyzer() {
return analyzer;
}
public CriticalComponentImpl(CriticalAnalyzer analyzer, int numberOfPaths) {
if (analyzer == null) {
analyzer = EmptyCriticalAnalyzer.getInstance();

12
artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/AbstractSequentialFileFactory.java
View File

@ -33,6 +33,8 @@ import org.apache.activemq.artemis.api.core.ActiveMQInterruptedException;
import org.apache.activemq.artemis.core.io.buffer.TimedBuffer;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
import org.apache.activemq.artemis.utils.critical.EmptyCriticalAnalyzer;
import org.jboss.logging.Logger;
/**
@ -74,11 +76,17 @@ public abstract class AbstractSequentialFileFactory implements SequentialFileFac
final int bufferTimeout,
final int maxIO,
final boolean logRates,
final IOCriticalErrorListener criticalErrorListener) {
final IOCriticalErrorListener criticalErrorListener,
CriticalAnalyzer criticalAnalyzer) {
this.journalDir = journalDir;
if (criticalAnalyzer == null) {
criticalAnalyzer = EmptyCriticalAnalyzer.getInstance();
}
if (buffered && bufferTimeout > 0) {
timedBuffer = new TimedBuffer(bufferSize, bufferTimeout, logRates);
timedBuffer = new TimedBuffer(criticalAnalyzer, bufferSize, bufferTimeout, logRates);
criticalAnalyzer.add(timedBuffer);
} else {
timedBuffer = null;
}

12
artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFileFactory.java
View File

@ -35,6 +35,7 @@ import org.apache.activemq.artemis.jlibaio.LibaioFile;
import org.apache.activemq.artemis.jlibaio.SubmitInfo;
import org.apache.activemq.artemis.jlibaio.util.CallbackCache;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
import org.jboss.logging.Logger;
public final class AIOSequentialFileFactory extends AbstractSequentialFileFactory {
@ -56,11 +57,11 @@ public final class AIOSequentialFileFactory extends AbstractSequentialFileFactor
private static final String AIO_TEST_FILE = ".aio-test";
public AIOSequentialFileFactory(final File journalDir, int maxIO) {
this(journalDir, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, maxIO, false, null);
this(journalDir, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, maxIO, false, null, null);
}
public AIOSequentialFileFactory(final File journalDir, final IOCriticalErrorListener listener, int maxIO) {
this(journalDir, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, maxIO, false, listener);
this(journalDir, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, maxIO, false, listener, null);
}
public AIOSequentialFileFactory(final File journalDir,
@ -68,7 +69,7 @@ public final class AIOSequentialFileFactory extends AbstractSequentialFileFactor
final int bufferTimeout,
final int maxIO,
final boolean logRates) {
this(journalDir, bufferSize, bufferTimeout, maxIO, logRates, null);
this(journalDir, bufferSize, bufferTimeout, maxIO, logRates, null, null);
}
public AIOSequentialFileFactory(final File journalDir,
@ -76,8 +77,9 @@ public final class AIOSequentialFileFactory extends AbstractSequentialFileFactor
final int bufferTimeout,
final int maxIO,
final boolean logRates,
final IOCriticalErrorListener listener) {
super(journalDir, true, bufferSize, bufferTimeout, maxIO, logRates, listener);
final IOCriticalErrorListener listener,
final CriticalAnalyzer analyzer) {
super(journalDir, true, bufferSize, bufferTimeout, maxIO, logRates, listener, analyzer);
callbackPool = new CallbackCache<>(maxIO);
if (logger.isTraceEnabled()) {
logger.trace("New AIO File Created");

16
artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/buffer/TimedBuffer.java
View File

@ -32,9 +32,14 @@ import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.journal.EncodingSupport;
import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
import org.apache.activemq.artemis.utils.critical.CriticalComponentImpl;
import org.jboss.logging.Logger;
public final class TimedBuffer {
public final class TimedBuffer extends CriticalComponentImpl {
protected static final int CRITICAL_PATHS = 1;
protected static final int CRITICAL_PATH_FLUSH = 0;
private static final Logger logger = Logger.getLogger(TimedBuffer.class);
@ -99,7 +104,8 @@ public final class TimedBuffer {
// Public --------------------------------------------------------
public TimedBuffer(final int size, final int timeout, final boolean logRates) {
public TimedBuffer(CriticalAnalyzer analyzer, final int size, final int timeout, final boolean logRates) {
super(analyzer, CRITICAL_PATHS);
bufferSize = size;
this.logRates = logRates;
@ -286,6 +292,8 @@ public final class TimedBuffer {
throw new IllegalStateException("TimedBuffer is not started");
}
enterCritical(CRITICAL_PATH_FLUSH);
try {
if (!delayFlush && buffer.writerIndex() > 0) {
int pos = buffer.writerIndex();
@ -299,7 +307,6 @@ public final class TimedBuffer {
//perform memcpy under the hood due to the off heap buffer
buffer.getBytes(0, bufferToFlush);
bufferObserver.flushBuffer(bufferToFlush, pendingSync, callbacks);
stopSpin();
@ -319,6 +326,9 @@ public final class TimedBuffer {
} else {
return false;
}
} finally {
leaveCritical(CRITICAL_PATH_FLUSH);
}
}
}

4
artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/mapped/MappedSequentialFileFactory.java
View File

@ -40,7 +40,9 @@ public final class MappedSequentialFileFactory extends AbstractSequentialFileFac
final int bufferTimeout,
IOCriticalErrorListener criticalErrorListener) {
super(directory, buffered, bufferSize, bufferTimeout, 1, false, criticalErrorListener);
// at the moment we only use the critical analyzer on the timed buffer
// MappedSequentialFile is not using any buffering, hence we just pass in null
super(directory, buffered, bufferSize, bufferTimeout, 1, false, criticalErrorListener, null);
this.capacity = capacity;
this.setDatasync(true);

12
artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/nio/NIOSequentialFileFactory.java
View File

@ -27,6 +27,7 @@ import org.apache.activemq.artemis.core.io.AbstractSequentialFileFactory;
import org.apache.activemq.artemis.core.io.IOCriticalErrorListener;
import org.apache.activemq.artemis.core.io.SequentialFile;
import org.apache.activemq.artemis.utils.Env;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
public final class NIOSequentialFileFactory extends AbstractSequentialFileFactory {
@ -42,7 +43,7 @@ public final class NIOSequentialFileFactory extends AbstractSequentialFileFactor
}
public NIOSequentialFileFactory(final File journalDir, final IOCriticalErrorListener listener, final int maxIO) {
this(journalDir, false, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, maxIO, false, listener);
this(journalDir, false, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, maxIO, false, listener, null);
}
public NIOSequentialFileFactory(final File journalDir, final boolean buffered, final int maxIO) {
@ -53,7 +54,7 @@ public final class NIOSequentialFileFactory extends AbstractSequentialFileFactor
final boolean buffered,
final IOCriticalErrorListener listener,
final int maxIO) {
this(journalDir, buffered, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, maxIO, false, listener);
this(journalDir, buffered, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, maxIO, false, listener, null);
}
public NIOSequentialFileFactory(final File journalDir,
@ -62,7 +63,7 @@ public final class NIOSequentialFileFactory extends AbstractSequentialFileFactor
final int bufferTimeout,
final int maxIO,
final boolean logRates) {
this(journalDir, buffered, bufferSize, bufferTimeout, maxIO, logRates, null);
this(journalDir, buffered, bufferSize, bufferTimeout, maxIO, logRates, null, null);
}
public NIOSequentialFileFactory(final File journalDir,
@ -71,8 +72,9 @@ public final class NIOSequentialFileFactory extends AbstractSequentialFileFactor
final int bufferTimeout,
final int maxIO,
final boolean logRates,
final IOCriticalErrorListener listener) {
super(journalDir, buffered, bufferSize, bufferTimeout, maxIO, logRates, listener);
final IOCriticalErrorListener listener,
final CriticalAnalyzer analyzer) {
super(journalDir, buffered, bufferSize, bufferTimeout, maxIO, logRates, listener, analyzer);
this.bufferPooling = true;
this.bytesPool = new ThreadLocal<>();
}

4
artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/JournalTptBenchmark.java
View File

@ -66,10 +66,10 @@ public class JournalTptBenchmark {
.setDatasync(dataSync);
break;
case Nio:
factory = new NIOSequentialFileFactory(tmpDirectory, true, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, 1, false, null).setDatasync(dataSync);
factory = new NIOSequentialFileFactory(tmpDirectory, true, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, 1, false, null, null).setDatasync(dataSync);
break;
case Aio:
factory = new AIOSequentialFileFactory(tmpDirectory, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, 500, false, null).setDatasync(dataSync);
factory = new AIOSequentialFileFactory(tmpDirectory, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, 500, false, null, null).setDatasync(dataSync);
//disable it when using directly the same buffer: ((AIOSequentialFileFactory)factory).disableBufferReuse();
if (!LibaioContext.isLoaded()) {
throw new IllegalStateException("lib AIO not loaded!");

4
artemis-journal/src/test/java/org/apache/activemq/artemis/core/io/SequentialFileTptBenchmark.java
View File

@ -59,10 +59,10 @@ public class SequentialFileTptBenchmark {
factory = new MappedSequentialFileFactory(tmpDirectory, fileSize, true, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, null).setDatasync(dataSync);
break;
case Nio:
factory = new NIOSequentialFileFactory(tmpDirectory, true, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, 1, false, null).setDatasync(dataSync);
factory = new NIOSequentialFileFactory(tmpDirectory, true, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_NIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_NIO, 1, false, null, null).setDatasync(dataSync);
break;
case Aio:
factory = new AIOSequentialFileFactory(tmpDirectory, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, 500, false, null).setDatasync(dataSync);
factory = new AIOSequentialFileFactory(tmpDirectory, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_SIZE_AIO, ArtemisConstants.DEFAULT_JOURNAL_BUFFER_TIMEOUT_AIO, 500, false, null, null).setDatasync(dataSync);
//disable it when using directly the same buffer: ((AIOSequentialFileFactory)factory).disableBufferReuse();
if (!LibaioContext.isLoaded()) {
throw new IllegalStateException("lib AIO not loaded!");

4
artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
View File

@ -133,11 +133,11 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
case NIO:
ActiveMQServerLogger.LOGGER.journalUseNIO();
journalFF = new NIOSequentialFileFactory(config.getJournalLocation(), true, config.getJournalBufferSize_NIO(), config.getJournalBufferTimeout_NIO(), config.getJournalMaxIO_NIO(), config.isLogJournalWriteRate(), criticalErrorListener);
journalFF = new NIOSequentialFileFactory(config.getJournalLocation(), true, config.getJournalBufferSize_NIO(), config.getJournalBufferTimeout_NIO(), config.getJournalMaxIO_NIO(), config.isLogJournalWriteRate(), criticalErrorListener, getCriticalAnalyzer());
break;
case ASYNCIO:
ActiveMQServerLogger.LOGGER.journalUseAIO();
journalFF = new AIOSequentialFileFactory(config.getJournalLocation(), config.getJournalBufferSize_AIO(), config.getJournalBufferTimeout_AIO(), config.getJournalMaxIO_AIO(), config.isLogJournalWriteRate(), criticalErrorListener);
journalFF = new AIOSequentialFileFactory(config.getJournalLocation(), config.getJournalBufferSize_AIO(), config.getJournalBufferTimeout_AIO(), config.getJournalMaxIO_AIO(), config.isLogJournalWriteRate(), criticalErrorListener, getCriticalAnalyzer());
break;
case MAPPED:
ActiveMQServerLogger.LOGGER.journalUseMAPPED();

110
tests/extra-tests/src/test/java/org/apache/activemq/artemis/tests/extras/byteman/CriticalAnalyzerFaultInjectionTest.java Normal file
View File

@ -0,0 +1,110 @@
/*
* 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.activemq.artemis.tests.extras.byteman;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import org.apache.activemq.artemis.api.core.RoutingType;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.config.Configuration;
import org.apache.activemq.artemis.core.server.JournalType;
import org.apache.activemq.artemis.core.server.impl.AddressInfo;
import org.apache.activemq.artemis.tests.util.JMSTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzerPolicy;
import org.jboss.byteman.contrib.bmunit.BMRule;
import org.jboss.byteman.contrib.bmunit.BMRules;
import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@RunWith(BMUnitRunner.class)
public class CriticalAnalyzerFaultInjectionTest extends JMSTestBase {
// Critical Analyzer Settings
private static long CHECK_PERIOD = 100;
private static long TIMEOUT = 1000;
public static long TEST_TIMEOUT = 5000;
private SimpleString address = SimpleString.toSimpleString("faultInjectionTestAddress");
@Override
@Before
public void setUp() throws Exception {
super.setUp();
server.addAddressInfo(new AddressInfo(address, RoutingType.ANYCAST));
server.createQueue(address, RoutingType.ANYCAST, address, null, true, false);
conn = nettyCf.createConnection();
}
/*
Checks every 100ms timesout after 3000ms. Test should wait no longer than 3100s + Shutdown time.
*/
@Override
protected Configuration createDefaultConfig(boolean netty) throws Exception {
return super.createDefaultConfig(netty)
.setCriticalAnalyzerPolicy(CriticalAnalyzerPolicy.SHUTDOWN)
.setCriticalAnalyzer(true)
.setCriticalAnalyzerCheckPeriod(CHECK_PERIOD)
.setCriticalAnalyzerTimeout(TIMEOUT)
.setJournalType(JournalType.NIO);
}
@Override
public boolean usePersistence() {
return true;
}
@BMRules(
rules = {@BMRule(
name = "Sync file data hangs",
targetClass = "org.apache.activemq.artemis.core.io.nio.NIOSequentialFile",
targetMethod = "sync",
targetLocation = "ENTRY",
action = "org.apache.activemq.artemis.tests.extras.byteman.CriticalAnalyzerFaultInjectionTest.methodHang();")})
@Test(timeout = 60000)
public void testSlowDiskSync() throws Exception {
sendConsumeDurableMessage();
Wait.waitFor(() -> !server.isStarted(), WAIT_TIMEOUT * 5);
assertFalse(server.isStarted());
}
private void sendConsumeDurableMessage() throws Exception {
try {
Session s = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
Queue jmsQueue = s.createQueue(address.toString());
MessageProducer p = s.createProducer(jmsQueue);
p.setDeliveryMode(DeliveryMode.PERSISTENT);
conn.start();
p.send(s.createTextMessage("payload"));
} catch (JMSException expected) {
} finally {
if (conn != null) {
conn.close();
}
}
}
public static void methodHang() throws InterruptedException {
Thread.sleep(TEST_TIMEOUT);
}
}

10
tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/journal/impl/TimedBufferTest.java
View File

@ -87,7 +87,7 @@ public class TimedBufferTest extends ActiveMQTestBase {
}
}
TimedBuffer timedBuffer = new TimedBuffer(100, TimedBufferTest.ONE_SECOND_IN_NANOS, false);
TimedBuffer timedBuffer = new TimedBuffer(null, 100, TimedBufferTest.ONE_SECOND_IN_NANOS, false);
timedBuffer.start();
@ -155,7 +155,7 @@ public class TimedBufferTest extends ActiveMQTestBase {
}
}
TimedBuffer timedBuffer = new TimedBuffer(100, TimedBufferTest.ONE_SECOND_IN_NANOS / 2, false);
TimedBuffer timedBuffer = new TimedBuffer(null, 100, TimedBufferTest.ONE_SECOND_IN_NANOS / 2, false);
timedBuffer.start();
@ -393,7 +393,7 @@ public class TimedBufferTest extends ActiveMQTestBase {
//it is optimistic: the timeout and the blockingDeviceFlushTime are a perfect match
final long deviceTime = timeout;
final int bufferSize = Env.osPageSize();
final TimedBuffer timedBuffer = new TimedBuffer(bufferSize, (int) timeout, false);
final TimedBuffer timedBuffer = new TimedBuffer(null, bufferSize, (int) timeout, false);
timedBuffer.start();
try (NonBlockingObserver observer = new NonBlockingObserver(bufferSize, deviceTime)) {
timedBuffer.setObserver(observer);
@ -434,7 +434,7 @@ public class TimedBufferTest extends ActiveMQTestBase {
//it is optimistic: the timeout and the blockingDeviceFlushTime are a perfect match
final long deviceTime = timeout;
final int bufferSize = Env.osPageSize();
final TimedBuffer timedBuffer = new TimedBuffer(bufferSize, (int) timeout, false);
final TimedBuffer timedBuffer = new TimedBuffer(null, bufferSize, (int) timeout, false);
timedBuffer.start();
try (BlockingObserver observer = new BlockingObserver(bufferSize, deviceTime)) {
timedBuffer.setObserver(observer);
@ -489,7 +489,7 @@ public class TimedBufferTest extends ActiveMQTestBase {
}
}
TimedBuffer timedBuffer = new TimedBuffer(100, TimedBufferTest.ONE_SECOND_IN_NANOS / 10, false);
TimedBuffer timedBuffer = new TimedBuffer(null, 100, TimedBufferTest.ONE_SECOND_IN_NANOS / 10, false);
timedBuffer.start();