From 0df40a65f12e269b57e054e41e0b01082d500fc9 Mon Sep 17 00:00:00 2001 From: Grigorii Kirgizov Date: Tue, 1 Sep 2020 13:32:33 +0300 Subject: [PATCH] Refactor incremental logic in ConstraintsProcessing into separate facade (MPSCR-71) Relations between stages and entry points into incremental algorithm are now much clearer. --- .../core/internal/ConstraintsProcessing.kt | 114 +++----------- .../reactor/core/internal/ControllerImpl.kt | 10 +- .../core/internal/EvaluationSessionImpl.kt | 2 +- .../core/internal/IncrementalProcessing.kt | 140 ++++++++++++++++++ 4 files changed, 162 insertions(+), 104 deletions(-) create mode 100644 reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/IncrementalProcessing.kt diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ConstraintsProcessing.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ConstraintsProcessing.kt index 3e81d37c..8f440054 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ConstraintsProcessing.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ConstraintsProcessing.kt @@ -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 { + 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) = - 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): Boolean = - utags.contains(this.ruleUniqueTag) || utags.any { utag -> dependsOnRule(utag) } - } diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ControllerImpl.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ControllerImpl.kt index b5d63c11..4b24c9f1 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ControllerImpl.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ControllerImpl.kt @@ -54,14 +54,8 @@ internal class ControllerImpl ( return storeView() } - fun incrLaunch(constraint: Constraint, rulesDiff: RulesDiff): Pair { - - val status = profiler.profile("journalTraverse") { - processing.invalidateAndAddRules(this, rulesDiff) - } - - return status to processing.invalidatedFeedback() - } + fun incrLaunch(rulesDiff: RulesDiff): Pair = + processing.runIncrementally(this, rulesDiff) fun activate(constraint: Constraint) : FeedbackStatus { // FIXME noLogicalContext diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/EvaluationSessionImpl.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/EvaluationSessionImpl.kt index aa37c5a9..597454aa 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/EvaluationSessionImpl.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/EvaluationSessionImpl.kt @@ -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 diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/IncrementalProcessing.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/IncrementalProcessing.kt new file mode 100644 index 00000000..879dc4c7 --- /dev/null +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/IncrementalProcessing.kt @@ -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): List +} + + +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): List = 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) = + 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) = + 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 + +} +