From 29f8023d07d2d41b2f0071e0deda58e82a340e8c Mon Sep 17 00:00:00 2001 From: MishaDemianenko Date: Fri, 2 Jun 2017 01:53:31 +0200 Subject: [PATCH] Use transaction counters to wait for all transactions completion on shutdown Use transaction statistics counters to wait for all transaction to complete before NeoStoreDataSource will try to acquire transactional log file lock to prevent deadlock on shutdown --- .../org/neo4j/kernel/NeoStoreDataSource.java | 16 +++++++++++---- .../org/neo4j/kernel/impl/api/Kernel.java | 15 ++++++++++---- .../neo4j/kernel/NeoStoreDataSourceTest.java | 20 +++++++++++++++++++ .../impl/api/integrationtest/KernelIT.java | 2 -- .../test/rule/NeoStoreDataSourceRule.java | 11 ++++++++-- 5 files changed, 52 insertions(+), 12 deletions(-) diff --git a/community/kernel/src/main/java/org/neo4j/kernel/NeoStoreDataSource.java b/community/kernel/src/main/java/org/neo4j/kernel/NeoStoreDataSource.java index b6338e1c98e8..38d2b69eae7b 100644 --- a/community/kernel/src/main/java/org/neo4j/kernel/NeoStoreDataSource.java +++ b/community/kernel/src/main/java/org/neo4j/kernel/NeoStoreDataSource.java @@ -98,7 +98,7 @@ import org.neo4j.kernel.impl.storemigration.monitoring.VisibleMigrationProgressMonitor; import org.neo4j.kernel.impl.storemigration.participant.StoreMigrator; import org.neo4j.kernel.impl.transaction.TransactionHeaderInformationFactory; -import org.neo4j.kernel.impl.transaction.TransactionMonitor; +import org.neo4j.kernel.impl.transaction.TransactionStats; import org.neo4j.kernel.impl.transaction.log.BatchingTransactionAppender; import org.neo4j.kernel.impl.transaction.log.LogFile; import org.neo4j.kernel.impl.transaction.log.LogFileInformation; @@ -280,7 +280,7 @@ boolean applicable( DiagnosticsPhase phase ) private final LockService lockService; private final IndexingService.Monitor indexingServiceMonitor; private final FileSystemAbstraction fs; - private final TransactionMonitor transactionMonitor; + private final TransactionStats transactionMonitor; private final DatabaseHealth databaseHealth; private final PhysicalLogFile.Monitor physicalLogMonitor; private final TransactionHeaderInformationFactory transactionHeaderInformationFactory; @@ -330,7 +330,7 @@ public NeoStoreDataSource( TransactionEventHandlers transactionEventHandlers, IndexingService.Monitor indexingServiceMonitor, FileSystemAbstraction fs, - TransactionMonitor transactionMonitor, + TransactionStats transactionMonitor, DatabaseHealth databaseHealth, PhysicalLogFile.Monitor physicalLogMonitor, TransactionHeaderInformationFactory transactionHeaderInformationFactory, @@ -905,13 +905,21 @@ private void awaitAllTransactionsClosed() // Only wait for committed transactions to be applied if the kernel is healthy (i.e. no panic) // otherwise if there has been a panic transactions will not be applied properly anyway. TransactionIdStore txIdStore = getDependencyResolver().resolveDependency( TransactionIdStore.class ); - while ( databaseHealth.isHealthy() && + while ( databaseHealth.isHealthy() && getNumberOfOngoingTransactions() > 0 && !txIdStore.closedTransactionIdIsOnParWithOpenedTransactionId() ) { LockSupport.parkNanos( 10_000_000 ); // 10 ms } } + private long getNumberOfOngoingTransactions() + { + return transactionMonitor.getNumberOfStartedTransactions() - + transactionMonitor.getNumberOfCommittedTransactions() - + transactionMonitor.getNumberOfRolledBackTransactions() - + transactionMonitor.getNumberOfTerminatedTransactions(); + } + private Lifecycle lifecycleToTriggerCheckPointOnShutdown() { // Write new checkpoint in the log only if the kernel is healthy. diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/Kernel.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/Kernel.java index be619e8d5e00..e53b9298cf33 100644 --- a/community/kernel/src/main/java/org/neo4j/kernel/impl/api/Kernel.java +++ b/community/kernel/src/main/java/org/neo4j/kernel/impl/api/Kernel.java @@ -93,13 +93,20 @@ public KernelTransaction newTransaction( KernelTransaction.Type type, SecurityCo } @Override - public KernelTransaction newTransaction( KernelTransaction.Type type, SecurityContext securityContext, long timeout ) throws - TransactionFailureException + public KernelTransaction newTransaction( KernelTransaction.Type type, SecurityContext securityContext, long timeout ) + throws TransactionFailureException { health.assertHealthy( TransactionFailureException.class ); - KernelTransaction transaction = transactions.newInstance( type, securityContext, timeout ); transactionMonitor.transactionStarted(); - return transaction; + try + { + return transactions.newInstance( type, securityContext, timeout ); + } + catch ( Throwable t ) + { + transactionMonitor.transactionFinished( false, false ); + throw t; + } } @Override diff --git a/community/kernel/src/test/java/org/neo4j/kernel/NeoStoreDataSourceTest.java b/community/kernel/src/test/java/org/neo4j/kernel/NeoStoreDataSourceTest.java index bcb20b7d2bfd..bf4e238dfbcb 100644 --- a/community/kernel/src/test/java/org/neo4j/kernel/NeoStoreDataSourceTest.java +++ b/community/kernel/src/test/java/org/neo4j/kernel/NeoStoreDataSourceTest.java @@ -21,6 +21,7 @@ import org.junit.Rule; import org.junit.Test; +import org.mockito.Mockito; import java.io.File; import java.io.IOException; @@ -36,6 +37,7 @@ import org.neo4j.kernel.impl.logging.SimpleLogService; import org.neo4j.kernel.impl.store.id.IdGeneratorFactory; import org.neo4j.kernel.impl.store.id.configuration.CommunityIdTypeConfigurationProvider; +import org.neo4j.kernel.impl.transaction.TransactionStats; import org.neo4j.kernel.impl.transaction.log.PhysicalLogFiles; import org.neo4j.kernel.impl.transaction.log.entry.LogEntryVersion; import org.neo4j.kernel.impl.transaction.log.entry.LogHeader; @@ -60,6 +62,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import static org.neo4j.helpers.collection.MapUtil.stringMap; @@ -273,6 +276,23 @@ public void shouldAlwaysShutdownLifeEvenWhenCheckPointingFails() throws Exceptio } } + @Test + public void checkTransactionStatsWhenStopDataSource() throws IOException + { + NeoStoreDataSource dataSource = + dsRule.getDataSource( dir.graphDbDir(), fs.get(), pageCacheRule.getPageCache( fs ), emptyMap() ); + TransactionStats transactionMonitor = dsRule.getTransactionMonitor(); + dataSource.start(); + + Mockito.verifyZeroInteractions(transactionMonitor); + + dataSource.stop(); + verify( transactionMonitor, times( 2 ) ).getNumberOfTerminatedTransactions(); + verify( transactionMonitor, times( 2 ) ).getNumberOfRolledBackTransactions(); + verify( transactionMonitor, times( 2 ) ).getNumberOfStartedTransactions(); + verify( transactionMonitor, times( 2 ) ).getNumberOfCommittedTransactions(); + } + private NeoStoreDataSource neoStoreDataSourceWithLogFilesContainingLowestTxId( PhysicalLogFiles files ) { DependencyResolver resolver = mock( DependencyResolver.class ); diff --git a/community/kernel/src/test/java/org/neo4j/kernel/impl/api/integrationtest/KernelIT.java b/community/kernel/src/test/java/org/neo4j/kernel/impl/api/integrationtest/KernelIT.java index 205b56e46e0a..92999b2e2594 100644 --- a/community/kernel/src/test/java/org/neo4j/kernel/impl/api/integrationtest/KernelIT.java +++ b/community/kernel/src/test/java/org/neo4j/kernel/impl/api/integrationtest/KernelIT.java @@ -66,8 +66,6 @@ public class KernelIT extends KernelIntegrationTest { - // TODO: Split this into area-specific tests, see PropertyIT. - @Test public void mixingBeansApiWithKernelAPI() throws Exception { diff --git a/community/kernel/src/test/java/org/neo4j/test/rule/NeoStoreDataSourceRule.java b/community/kernel/src/test/java/org/neo4j/test/rule/NeoStoreDataSourceRule.java index 2325b25b2efb..b355f00d3db8 100644 --- a/community/kernel/src/test/java/org/neo4j/test/rule/NeoStoreDataSourceRule.java +++ b/community/kernel/src/test/java/org/neo4j/test/rule/NeoStoreDataSourceRule.java @@ -57,7 +57,7 @@ import org.neo4j.kernel.impl.store.id.configuration.CommunityIdTypeConfigurationProvider; import org.neo4j.kernel.impl.store.id.configuration.IdTypeConfigurationProvider; import org.neo4j.kernel.impl.transaction.TransactionHeaderInformationFactory; -import org.neo4j.kernel.impl.transaction.TransactionMonitor; +import org.neo4j.kernel.impl.transaction.TransactionStats; import org.neo4j.kernel.impl.transaction.log.PhysicalLogFile; import org.neo4j.kernel.impl.util.JobScheduler; import org.neo4j.kernel.internal.DatabaseHealth; @@ -75,6 +75,7 @@ public class NeoStoreDataSourceRule extends ExternalResource { private NeoStoreDataSource dataSource; + private TransactionStats transactionMonitor; public NeoStoreDataSource getDataSource( File storeDir, FileSystemAbstraction fs, PageCache pageCache, Map additionalConfig, DatabaseHealth databaseHealth ) @@ -108,13 +109,14 @@ public NeoStoreDataSource getDataSource( File storeDir, FileSystemAbstraction fs JobScheduler jobScheduler = mock( JobScheduler.class, RETURNS_MOCKS ); Monitors monitors = new Monitors(); + transactionMonitor = mock( TransactionStats.class ); dataSource = new NeoStoreDataSource( storeDir, config, idGeneratorFactory, IdReuseEligibility.ALWAYS, idConfigurationProvider, logService, mock( JobScheduler.class, RETURNS_MOCKS ), mock( TokenNameLookup.class ), dependencyResolverForNoIndexProvider(), mock( PropertyKeyTokenHolder.class ), mock( LabelTokenHolder.class ), mock( RelationshipTypeTokenHolder.class ), locksFactory, mock( SchemaWriteGuard.class ), mock( TransactionEventHandlers.class ), IndexingService.NO_MONITOR, - fs, mock( TransactionMonitor.class ), databaseHealth, + fs, transactionMonitor, databaseHealth, mock( PhysicalLogFile.Monitor.class ), TransactionHeaderInformationFactory.DEFAULT, new StartupStatisticsProvider(), null, new CommunityCommitProcessFactory(), mock( InternalAutoIndexing.class ), pageCache, @@ -126,6 +128,11 @@ fs, mock( TransactionMonitor.class ), databaseHealth, return dataSource; } + public TransactionStats getTransactionMonitor() + { + return transactionMonitor; + } + public NeoStoreDataSource getDataSource( File storeDir, FileSystemAbstraction fs, PageCache pageCache, Map additionalConfig ) {