natas_hw
/
mps-coderules
mirror of https://github.com/JetBrains/mps-coderules
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor incremental logic in ConstraintsProcessing into separate facade (MPSCR-71) 

Relations between stages and entry points into incremental
algorithm are now much clearer.
This commit is contained in:
Grigorii Kirgizov 2020-09-01 13:32:33 +03:00
parent da27c7ccd4
commit 0df40a65f1
4 changed files with 162 additions and 104 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ConstraintsProcessing.kt
View File

@ -49,26 +49,12 @@ internal class ConstraintsProcessing(
) : StoreAwareJournalImpl(journal, logicalState), IncrSpecHolder {
private val journalIndex: MatchJournal.Index = journal.index()
private var incrementalProcessing: IncrementalProcessing = NonIncrementalProcessing()
private val occurrenceContractObserver: OccurrenceContractObserver? =
if (ispec.assertLevel().assertContracts()) OccurrenceContractObserver(logicalState, ispec) else null
inner class ProgramStateCleaner{
fun erase(occurrence: Occurrence) {
dispatchingFront = dispatchingFront.forget(occurrence)
occurrence.clearLogicalState(logicalState)
}
fun erase(match: RuleMatchEx) {
dispatchingFront = dispatchingFront.forget(match)
}
}
private val stateCleaner = ProgramStateCleaner()
fun activateContinue(controller: Controller, activeOcc: Occurrence, parent: MatchJournal.MatchChunk): FeedbackStatus {
assert(activeOcc.stored)
@ -81,62 +67,10 @@ internal class ConstraintsProcessing(
return processActivated(controller, activeOcc, parent, FeedbackStatus.NORMAL())
}
private lateinit var invalidator: InvalidationStage
private lateinit var adder: AdditionStage
private lateinit var postponer: PostponeMatchesStage
private lateinit var continuator: ContinueOccurrencesStage
fun invalidateAndAddRules(controller: Controller, rulesDiff: RulesDiff): FeedbackStatus {
val ruleOrdering = RuleOrdering(ruleIndex)
continuator = ContinueOccurrencesStage(ispec, journalIndex)
invalidator = InvalidationStage(ispec, rulesDiff.removed, continuator, stateCleaner, trace)
adder = AdditionStage(ispec, rulesDiff.added, continuator, ruleOrdering, ruleIndex, trace)
postponer = PostponeMatchesStage(ispec, this, journalIndex, ruleOrdering)
val cursor = this.cursor
var status: FeedbackStatus = FeedbackStatus.NORMAL()
while (true) {
invalidator.run(cursor)
val postponedMatches = postponer.onNext(cursor)
adder.receive(postponedMatches)
adder.onNext(cursor)
status = continuator.run(controller, cursor)
if (!status.operational) break
// continuator may request invalidating more chunks
val haveChanges = invalidator.run(cursor)
if (cursor.atEnd()) break
if (!haveChanges) cursor.next()
}
return status
}
private fun InvalidationStage.run(cursor: RemovingJournalIterator): Boolean {
var haveChanges = false
while (this.onNext(cursor)) {
cursor.removeNext()
haveChanges = true
}
return haveChanges
}
private fun ContinueOccurrencesStage.run(controller: Controller, chunkReader: ChunkReader): FeedbackStatus {
var status: FeedbackStatus = FeedbackStatus.NORMAL()
val parentChunk = parentChunk()
for (continuedOcc in this.onNext(chunkReader)) {
if (continuedOcc.stored) {
status = activateContinue(controller, continuedOcc, parentChunk)
if (!status.operational) break
}
}
return status
fun runIncrementally(controller: Controller, rulesDiff: RulesDiff): Pair<FeedbackStatus, FeedbackKeySet> {
incrementalProcessing = IncrementalProcessingImpl(ispec, this, rulesDiff, ProgramStateCleaner(), ruleIndex, trace)
val status = incrementalProcessing.run(this, controller)
return status to incrementalProcessing.invalidatedFeedback()
}
/**
@ -161,8 +95,6 @@ internal class ConstraintsProcessing(
ruleMatcher.rule().isPrincipal || ruleMatcher.probe().hasOccurrences()
}
fun invalidatedFeedback(): FeedbackKeySet = invalidator.invalidatedFeedback()
/**
* Called to update the state with the currently active constraint occurrence.
* Calls the controller to process matches (if any) that were triggered.
@ -176,6 +108,7 @@ internal class ConstraintsProcessing(
logActivation(active)
active.revive(logicalState)
// fixme: move to incremental facade
occurrenceContractObserver?.onActivated(active)
}
assert(active.alive)
@ -187,7 +120,7 @@ internal class ConstraintsProcessing(
}
val matches = dispatchingFront.matches().toList()
val currentMatches = postponeFutureMatches(active, matches)
val currentMatches = incrementalProcessing.processOccurrenceMatches(active, matches)
val outStatus = currentMatches.fold(inStatus) { status, match ->
// TODO: paranoid check. should be isAlive() instead
@ -227,24 +160,11 @@ internal class ConstraintsProcessing(
}
}
.also { trace.trigger(match) }
.also { continueReplacedHeads(match, parent) }
.also { incrementalProcessing.processMatch(match) }
.also { accept(controller, match) }
.then { controller.processBody(match, parent, it) }
.also { trace.finish(match) }
private fun continueReplacedHeads(match: RuleMatchEx, parent: MatchJournal.MatchChunk) {
if (requiresIncrementalProcessing(match)) {
val invalidJustifications = match.matchHeadReplaced().filter { it.isPrincipal }
invalidator.receive(invalidJustifications)
}
}
private fun postponeFutureMatches(active: Occurrence, matches: List<RuleMatchEx>) =
if (requiresIncrementalProcessing(active)) {
postponer.postponeFutureMatches(matches)
} else matches
private fun accept(controller: Controller, match: RuleMatchEx) {
profiler.profile("logMatch") {
@ -288,6 +208,17 @@ internal class ConstraintsProcessing(
}
inner class ProgramStateCleaner{
fun erase(occurrence: Occurrence) {
dispatchingFront = dispatchingFront.forget(occurrence)
occurrence.clearLogicalState(logicalState)
}
fun erase(match: RuleMatchEx) {
dispatchingFront = dispatchingFront.forget(match)
}
}
/**
* Encapsulates logic for deriving [Evidence] and [Justifications] for a new [Occurrence].
*/
@ -319,11 +250,4 @@ internal class ConstraintsProcessing(
}
}
private fun requiresIncrementalProcessing(match: RuleMatchEx) = !isFront() && ispec.ability().allowed() && match.isPrincipal
private fun requiresIncrementalProcessing(occ: Occurrence) = !isFront() && ispec.ability().allowed() && occ.isPrincipal
private fun MatchJournal.MatchChunk.dependsOnAny(utags: Iterable<Any>): Boolean =
utags.contains(this.ruleUniqueTag) || utags.any { utag -> dependsOnRule(utag) }
}

10
reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ControllerImpl.kt
View File

@ -54,14 +54,8 @@ internal class ControllerImpl (
return storeView()
}
fun incrLaunch(constraint: Constraint, rulesDiff: RulesDiff): Pair<FeedbackStatus, FeedbackKeySet> {
val status = profiler.profile<FeedbackStatus>("journalTraverse") {
processing.invalidateAndAddRules(this, rulesDiff)
}
return status to processing.invalidatedFeedback()
}
fun incrLaunch(rulesDiff: RulesDiff): Pair<FeedbackStatus, FeedbackKeySet> =
processing.runIncrementally(this, rulesDiff)
fun activate(constraint: Constraint) : FeedbackStatus {
// FIXME noLogicalContext

2
reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/EvaluationSessionImpl.kt
View File

@ -81,7 +81,7 @@ internal class EvaluationSessionImpl private constructor (
val controller = ControllerImpl(supervisor, processing, incrementality, trace, profiler)
logicalState.init(controller)
val status2tags = controller.incrLaunch(main, rulesDiff)
val status2tags = controller.incrLaunch(rulesDiff)
newToken = processing.endSession()
status = status2tags.first
invalidFeedbackKeys = status2tags.second

140
reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/IncrementalProcessing.kt Normal file
View File

@ -0,0 +1,140 @@
/*
* Copyright 2014-2020 JetBrains s.r.o.
*
* Licensed 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 jetbrains.mps.logic.reactor.core.internal
import jetbrains.mps.logic.reactor.core.*
import jetbrains.mps.logic.reactor.evaluation.EvaluationTrace
import jetbrains.mps.logic.reactor.program.IncrementalSpec
/**
* Facade for incremental processing
*/
internal interface IncrementalProcessing {
fun invalidatedFeedback(): FeedbackKeySet
fun run(processing: ConstraintsProcessing, controller: Controller): FeedbackStatus
fun processMatch(match: RuleMatchEx)
fun processOccurrenceMatches(active: Occurrence, matches: List<RuleMatchEx>): List<RuleMatchEx>
}
internal class NonIncrementalProcessing: IncrementalProcessing {
override fun invalidatedFeedback(): FeedbackKeySet = emptySet()
override fun run(processing: ConstraintsProcessing, controller: Controller): FeedbackStatus = FeedbackStatus.NORMAL()
override fun processMatch(match: RuleMatchEx) {}
override fun processOccurrenceMatches(active: Occurrence, matches: List<RuleMatchEx>): List<RuleMatchEx> = matches
}
internal class IncrementalProcessingImpl(
override val ispec: IncrementalSpec,
val journal: MatchJournal,
rulesDiff: RulesDiff,
stateCleaner: ConstraintsProcessing.ProgramStateCleaner,
ruleIndex: RuleIndex,
trace: EvaluationTrace
): IncrementalProcessing, IncrSpecHolder {
private val journalIndex = journal.index()
private val ruleOrdering = RuleOrdering(ruleIndex)
private val continuator = ContinueOccurrencesStage(ispec, journalIndex)
private val invalidator = InvalidationStage(ispec, rulesDiff.removed, continuator, stateCleaner, trace)
private val adder = AdditionStage(ispec, rulesDiff.added, continuator, ruleOrdering, ruleIndex, trace)
private val postponer = PostponeMatchesStage(ispec, journal, journalIndex, ruleOrdering)
override fun invalidatedFeedback(): FeedbackKeySet = invalidator.invalidatedFeedback()
override fun processMatch(match: RuleMatchEx) =
continueReplacedHeadsImpl(match)
override fun processOccurrenceMatches(active: Occurrence, matches: List<RuleMatchEx>) =
postponeFutureMatchesImpl(active, matches)
override fun run(processing: ConstraintsProcessing, controller: Controller): FeedbackStatus {
var status: FeedbackStatus = FeedbackStatus.NORMAL()
val cursor = journal.cursor
while (true) {
invalidate(cursor)
val postponedMatches = postponer.onNext(cursor)
adder.receive(postponedMatches)
adder.onNext(cursor)
status = runContinued(processing, controller, cursor)
if (!status.operational) break
// continuator may request invalidating more chunks
val haveChanges = invalidate(cursor)
if (cursor.atEnd()) break
if (!haveChanges) cursor.next()
}
return status
}
private fun continueReplacedHeadsImpl(match: RuleMatchEx) {
if (requiresIncrementalProcessing(match)) {
val invalidJustifications = match.matchHeadReplaced().filter { it.isPrincipal }
invalidator.receive(invalidJustifications)
}
}
private fun postponeFutureMatchesImpl(active: Occurrence, matches: List<RuleMatchEx>) =
if (requiresIncrementalProcessing(active)) {
postponer.postponeFutureMatches(matches)
} else matches
private fun invalidate(cursor: RemovingJournalIterator): Boolean {
var haveChanges = false
while (invalidator.onNext(cursor)) {
cursor.removeNext()
haveChanges = true
}
return haveChanges
}
private fun runContinued(processing: ConstraintsProcessing, controller: Controller, chunkReader: ChunkReader): FeedbackStatus {
var status: FeedbackStatus = FeedbackStatus.NORMAL()
val parentChunk = journal.parentChunk()
for (continuedOcc in continuator.onNext(chunkReader)) {
if (continuedOcc.stored) {
status = processing.activateContinue(controller, continuedOcc, parentChunk)
if (!status.operational) break
}
}
return status
}
private fun requiresIncrementalProcessing(match: RuleMatchEx) = !journal.isFront() && ispec.ability().allowed() && match.isPrincipal
private fun requiresIncrementalProcessing(occ: Occurrence) = !journal.isFront() && ispec.ability().allowed() && occ.isPrincipal
}