ExternalVendorCode/corda
1
0
Fork 0
mirror of https://github.com/corda/corda.git synced 2024-12-22 06:17:55 +00:00
Code Issues Actions 2 Packages Projects Releases Wiki Activity

Database.isolatedTransaction removal

Browse Source
This commit is contained in:
szymonsztuka 2017-08-09 09:31:48 +01:00 committed by GitHub
parent 1d965b1785
commit a3ffd92544
5 changed files with 13 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
node/src/main/kotlin/net/corda/node/internal/AbstractNode.kt
View File

@ -772,7 +772,7 @@ abstract class AbstractNode(open val configuration: NodeConfiguration,
// the KMS is meant for derived temporary keys used in transactions, and we're not supposed to sign things with
// the identity key. But the infrastructure to make that easy isn't here yet.
override val keyManagementService by lazy { makeKeyManagementService(identityService) }
override val schedulerService by lazy { NodeSchedulerService(this, unfinishedSchedules = busyNodeLatch) }
override val schedulerService by lazy { NodeSchedulerService(this, unfinishedSchedules = busyNodeLatch, serverThread = serverThread) }
override val identityService by lazy {
val trustStore = KeyStoreWrapper(configuration.trustStoreFile, configuration.trustStorePassword)
val caKeyStore = KeyStoreWrapper(configuration.nodeKeystore, configuration.keyStorePassword)

9
node/src/main/kotlin/net/corda/node/services/events/NodeSchedulerService.kt
View File

@ -42,7 +42,8 @@ import javax.annotation.concurrent.ThreadSafe
@ThreadSafe
class NodeSchedulerService(private val services: ServiceHubInternal,
private val schedulerTimerExecutor: Executor = Executors.newSingleThreadExecutor(),
private val unfinishedSchedules: ReusableLatch = ReusableLatch())
private val unfinishedSchedules: ReusableLatch = ReusableLatch(),
private val serverThread: AffinityExecutor)
: SchedulerService, SingletonSerializeAsToken() {
companion object {
@ -156,13 +157,10 @@ class NodeSchedulerService(private val services: ServiceHubInternal,
}
private fun onTimeReached(scheduledState: ScheduledStateRef) {
serverThread.fetchFrom {
services.database.transaction {
val scheduledFlow = getScheduledFlow(scheduledState)
if (scheduledFlow != null) {
// TODO Because the flow is executed asynchronously, there is a small window between this tx we're in
// committing and the flow's first checkpoint when it starts in which we can lose the flow if the node
// goes down.
// See discussion in https://github.com/corda/corda/pull/639#discussion_r115257437
val future = services.startFlow(scheduledFlow, FlowInitiator.Scheduled(scheduledState)).resultFuture
future.then {
unfinishedSchedules.countDown()
@ -170,6 +168,7 @@ class NodeSchedulerService(private val services: ServiceHubInternal,
}
}
}
}
private fun getScheduledFlow(scheduledState: ScheduledStateRef): FlowLogic<*>? {
val scheduledActivity = getScheduledActivity(scheduledState)

6
node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineManager.kt
View File

@ -468,11 +468,7 @@ class StateMachineManager(val serviceHub: ServiceHubInternal,
fun <T> add(logic: FlowLogic<T>, flowInitiator: FlowInitiator): FlowStateMachineImpl<T> {
// TODO: Check that logic has @Suspendable on its call method.
executor.checkOnThread()
// We swap out the parent transaction context as using this frequently leads to a deadlock as we wait
// on the flow completion future inside that context. The problem is that any progress checkpoints are
// unable to acquire the table lock and move forward till the calling transaction finishes.
// Committing in line here on a fresh context ensure we can progress.
val fiber = database.isolatedTransaction {
val fiber = database.transaction {
val fiber = createFiber(logic, flowInitiator)
updateCheckpoint(fiber)
fiber

9
node/src/main/kotlin/net/corda/node/utilities/CordaPersistence.kt
View File

@ -42,15 +42,6 @@ class CordaPersistence(var dataSource: HikariDataSource, databaseProperties: Pro
return ctx?.connection ?: throw IllegalStateException("Was expecting to find database transaction: must wrap calling code within a transaction.")
}
fun <T> isolatedTransaction(block: DatabaseTransaction.() -> T): T {
val context = DatabaseTransactionManager.setThreadLocalTx(null)
return try {
transaction(block)
} finally {
DatabaseTransactionManager.restoreThreadLocalTx(context)
}
}
fun <T> transaction(statement: DatabaseTransaction.() -> T): T {
DatabaseTransactionManager.dataSource = this
return transaction(transactionIsolationLevel, 3, statement)

2
node/src/test/kotlin/net/corda/node/services/events/NodeSchedulerServiceTest.kt
View File

@ -97,8 +97,8 @@ class NodeSchedulerServiceTest : SingletonSerializeAsToken() {
override val vaultService: VaultService = NodeVaultService(this, dataSourceProps, makeTestDatabaseProperties())
override val testReference = this@NodeSchedulerServiceTest
}
scheduler = NodeSchedulerService(services, schedulerGatedExecutor)
smmExecutor = AffinityExecutor.ServiceAffinityExecutor("test", 1)
scheduler = NodeSchedulerService(services, schedulerGatedExecutor, serverThread = smmExecutor)
val mockSMM = StateMachineManager(services, DBCheckpointStorage(), smmExecutor, database)
mockSMM.changes.subscribe { change ->
if (change is StateMachineManager.Change.Removed && mockSMM.allStateMachines.isEmpty()) {