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

ARTEMIS-1700 Fixed deadlock in paging state 

This closes #1894
This commit is contained in:
Clebert Suconic 2018-02-26 13:13:50 -05:00
parent b69790b5ff
commit ecf4110b1b
5 changed files with 26 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
artemis-commons/src/main/java/org/apache/activemq/artemis/utils/actors/ArtemisExecutor.java
View File

@ -50,6 +50,23 @@ public interface ArtemisExecutor extends Executor {
return 0;
}
/** To be used to flush an executor from a different thread.
* WARNING: Do not call this within the executor. That would be stoopid ;)
*
* @param timeout
* @param unit
* @return
*/
default boolean flush(long timeout, TimeUnit unit) {
CountDownLatch latch = new CountDownLatch(1);
execute(latch::countDown);
try {
return latch.await(timeout, unit);
} catch (Exception e) {
return false;
}
}
/**
* It will wait the current execution (if there is one) to finish
* but will not complete any further executions

9
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
View File

@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.artemis.core.filter.Filter;
@ -38,7 +39,6 @@ import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.server.ActiveMQServerLogger;
import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.utils.FutureLatch;
import org.apache.activemq.artemis.utils.SoftValueHashMap;
import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
import org.jboss.logging.Logger;
@ -244,12 +244,9 @@ public class PageCursorProviderImpl implements PageCursorProvider {
}
private void waitForFuture() {
FutureLatch future = new FutureLatch();
if (!executor.flush(10, TimeUnit.SECONDS)) {
ActiveMQServerLogger.LOGGER.timedOutStoppingPagingCursor(executor);
executor.execute(future);
while (!future.await(10000)) {
ActiveMQServerLogger.LOGGER.timedOutStoppingPagingCursor(future, executor);
}
}

6
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
View File

@ -28,6 +28,7 @@ import java.util.Map.Entry;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@ -54,7 +55,6 @@ import org.apache.activemq.artemis.core.transaction.Transaction;
import org.apache.activemq.artemis.core.transaction.TransactionOperationAbstract;
import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
import org.apache.activemq.artemis.utils.FutureLatch;
import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
import org.apache.activemq.artemis.utils.collections.ConcurrentHashSet;
import org.jboss.logging.Logger;
@ -703,9 +703,7 @@ final class PageSubscriptionImpl implements PageSubscription {
@Override
public void flushExecutors() {
FutureLatch future = new FutureLatch();
executor.execute(future);
while (!future.await(1000)) {
if (!executor.flush(10, TimeUnit.SECONDS)) {
ActiveMQServerLogger.LOGGER.timedOutFlushingExecutorsPagingCursor(this);
}
}

5
artemis-server/src/main/java/org/apache/activemq/artemis/core/server/ActiveMQServerLogger.java
View File

@ -61,7 +61,6 @@ import org.apache.activemq.artemis.core.server.cluster.impl.ClusterConnectionImp
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
import org.apache.activemq.artemis.core.server.impl.ServerSessionImpl;
import org.apache.activemq.artemis.core.server.management.Notification;
import org.apache.activemq.artemis.utils.FutureLatch;
import org.jboss.logging.BasicLogger;
import org.jboss.logging.Logger;
import org.jboss.logging.annotations.Cause;
@ -512,8 +511,8 @@ public interface ActiveMQServerLogger extends BasicLogger {
void problemUndeployingNode(@Cause Exception e, Node node);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222022, value = "Timed out waiting for paging cursor to stop {0} {1}", format = Message.Format.MESSAGE_FORMAT)
void timedOutStoppingPagingCursor(FutureLatch future, Executor executor);
@Message(id = 222022, value = "Timed out waiting for paging cursor to stop {0}", format = Message.Format.MESSAGE_FORMAT)
void timedOutStoppingPagingCursor(Executor executor);
@LogMessage(level = Logger.Level.WARN)
@Message(id = 222023, value = "problem cleaning page address {0}", format = Message.Format.MESSAGE_FORMAT)

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

@ -2121,9 +2121,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
protected PagingStoreFactory getPagingStoreFactory() throws Exception {
if (configuration.getStoreConfiguration() != null && configuration.getStoreConfiguration().getStoreType() == StoreConfiguration.StoreType.DATABASE) {
DatabaseStorageConfiguration dbConf = (DatabaseStorageConfiguration) configuration.getStoreConfiguration();
return new PagingStoreFactoryDatabase(dbConf, storageManager, configuration.getJournalBufferTimeout_NIO(), scheduledPool, executorFactory, false, shutdownOnCriticalIO);
return new PagingStoreFactoryDatabase(dbConf, storageManager, configuration.getJournalBufferTimeout_NIO(), scheduledPool, ioExecutorFactory, false, shutdownOnCriticalIO);
}
return new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), configuration.getJournalBufferTimeout_NIO(), scheduledPool, executorFactory, configuration.isJournalSyncNonTransactional(), shutdownOnCriticalIO);
return new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), configuration.getJournalBufferTimeout_NIO(), scheduledPool, ioExecutorFactory, configuration.isJournalSyncNonTransactional(), shutdownOnCriticalIO);
}
/**