From 2504b354f03144ee71595e5d374af56d9ed8fbaf Mon Sep 17 00:00:00 2001 From: Grigorii Kirgizov Date: Wed, 16 Dec 2020 13:00:08 +0300 Subject: [PATCH] Implement fallback from incremental to default processing on unhandled cases (MPSCR-67) --- .../mps/logic/reactor/core/Feedback.kt | 1 + .../core/internal/EvaluationSessionImpl.kt | 93 +++++++++++++++---- .../internal/PrincipalObserverDispatcher.kt | 9 +- .../core/internal/ProcessingStrategy.kt | 12 ++- reactor/Test/test/TestIncrementalProgram.kt | 65 ++----------- 5 files changed, 102 insertions(+), 78 deletions(-) diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/Feedback.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/Feedback.kt index 02ee78c9..27bb7008 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/Feedback.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/Feedback.kt @@ -56,5 +56,6 @@ abstract class Feedback : EvaluationFeedback() { } typealias FeedbackKeySet = Set +typealias MutableFeedbackKeySet = MutableSet internal val RuleMatch.feedbackKey: Any get() = System.identityHashCode(this) 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 280306fc..16cf984d 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 @@ -73,29 +73,44 @@ internal class EvaluationSessionImpl private constructor ( @Suppress("UNCHECKED_CAST") override fun parameter(key: ParameterKey): T? = params ?.get(key) as T + private fun SessionToken?.hasBindingObservers() = + (this as? SessionTokenImpl)?.principalObservers?.isNotEmpty() ?: false + private fun launch(token: SessionToken?, store: OccurrenceStore, main: Constraint): EvaluationResult { val sessionProcessing: ProcessingSession = with(incrementality) { when { - ability().allowed() && incrLevel() == IncrementalSpec.IncrLevel.Full -> - IncrementalProcessingSession() - ability().allowed() && incrLevel() == IncrementalSpec.IncrLevel.Preamble -> - PreambleProcessingSession() - else -> - DefaultProcessingSession() + ability().allowed() -> when { + // it's a too restrictive place for fallback, happens almost always + //token.hasBindingObservers() -> FallbackProcessingSession() + incrLevel() == IncrementalSpec.IncrLevel.Full -> IncrementalProcessingSession() + incrLevel() == IncrementalSpec.IncrLevel.Preamble -> PreambleProcessingSession() + else -> DefaultProcessingSession() + } + else -> DefaultProcessingSession() } } - with (sessionProcessing) { - val session = + val session = + with(sessionProcessing) { if (token != null) { (token as? SessionTokenImpl)?.setStore(store) nextSession(token) } else { firstSession() } + } - return runSession(session, main) + try { + return sessionProcessing.runSession(session, main) + + // fixme: exception is a dirty way to abort incremental processing. + } catch (e: IncrementalContractViolationException) { + with(FallbackProcessingSession(session)) { + val fallbackSession = + if (token != null) nextSession(token) else firstSession() + return runSession(fallbackSession, main) + } } } @@ -115,12 +130,7 @@ internal class EvaluationSessionImpl private constructor ( val logicalState = LogicalState() val dispatchingFront = Dispatcher(ruleIndex).front() - // todo: remove this option - val principalObservers: PrincipalObserverDispatcher = - if (incrementality.assertLevel().assertContracts()) - LogicalBindObserverDispatcher() - else - PrincipalObserverDispatcher.EMPTY + val principalObservers = LogicalBindObserverDispatcher() val processingStrategy = GroundProcessing(incrementality, principalObservers) val processing = ConstraintsProcessing(dispatchingFront, journal, logicalState, incrementality, trace, profiler) @@ -142,10 +152,61 @@ internal class EvaluationSessionImpl private constructor ( return EvaluationResultImpl(newToken, status, strategy.invalidatedFeedback(), strategy.invalidatedRules()) } - private fun SessionParts.run(main: Constraint): FeedbackStatus = strategy.run(processing, controller, main) + protected fun SessionParts.run(main: Constraint): FeedbackStatus = strategy.run(processing, controller, main) } + /** + * Same as [DefaultProcessingSession], but also returns + * [EvaluationResult.invalidFeedbackKeys] & [EvaluationResult.invalidRules] + * to signify that results of a previous session must be cleared. + */ + inner class FallbackProcessingSession(private val abortedSession: SessionParts?): DefaultProcessingSession() { + + constructor(): this(null) + + val invalidatedFeedback: MutableFeedbackKeySet = mutableSetOf() + val invalidatedRules: MutableList = mutableListOf() + + + override fun nextSession(token: SessionToken): SessionParts { + + invalidateToken(token.journalView as MatchJournal.View) + invalidateAborted() + + return super.nextSession(token) + } + + override fun runSession(session: SessionParts, main: Constraint): EvaluationResult = with(session) { + val status = run(main) + controller.shutDown() + val newToken = endSession(session) + + invalidatedFeedback.addAll(strategy.invalidatedFeedback()) + invalidatedRules.addAll(strategy.invalidatedRules()) + + return EvaluationResultImpl(newToken, status, invalidatedFeedback, invalidatedRules) + } + + private fun invalidateAborted() { + if(abortedSession != null) { + with (abortedSession) { + invalidatedFeedback.addAll(strategy.invalidatedFeedback()) + invalidatedRules.addAll(strategy.invalidatedRules()) + } + } + } + + private fun invalidateToken(view: MatchJournal.View) { + for(chunk in view.chunks) { + if (chunk is MatchJournal.MatchChunk) { + invalidatedFeedback.add(chunk.match.feedbackKey) + invalidatedRules.add(chunk.ruleUniqueTag) + } + } + } + } + open inner class IncrementalProcessingSession(): DefaultProcessingSession() { override fun nextSession(token: SessionToken): SessionParts { val tkn = token as SessionTokenImpl diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/PrincipalObserverDispatcher.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/PrincipalObserverDispatcher.kt index 362afa74..705a0935 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/PrincipalObserverDispatcher.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/PrincipalObserverDispatcher.kt @@ -37,12 +37,15 @@ interface PrincipalObserverDispatcher { fun isTriggered(occ: Occurrence): Boolean fun removeTriggered(occ: Occurrence): Boolean + fun isEmpty(): Boolean + fun isNotEmpty() = !isEmpty() companion object EMPTY : PrincipalObserverDispatcher { override fun onActivated(occ: Occurrence, observable: LogicalStateObservable) { } override fun onInvalidated(occ: Occurrence, observable: LogicalStateObservable) { } override fun setTriggerReceiver(receiver: ObserverTriggeredHandler) { } override fun clearTriggerReceiver() { } + override fun isEmpty(): Boolean = true override fun isObserving(occ: Occurrence): Boolean = false override fun isTriggered(occ: Occurrence): Boolean = false override fun removeTriggered(occ: Occurrence): Boolean = false @@ -85,6 +88,8 @@ internal class LogicalBindObserverDispatcher : PrincipalObserverDispatcher { observing.remove(occ.identity)?.removeObservers(observable) } + override fun isEmpty(): Boolean = triggered.isEmpty() //&& observing.isEmpty() + override fun isObserving(occ: Occurrence): Boolean = observing.containsKey(occ.identity) override fun isTriggered(occ: Occurrence): Boolean = triggered.contains(occ.identity) @@ -99,10 +104,6 @@ internal class LogicalBindObserverDispatcher : PrincipalObserverDispatcher { // then remember it for later requests triggered.add(source.identity) } - -// checkContract(false) { -// "$logical can't be unified because it's used in principal occurrence $source" -// } } private inner class LogicalBindObserver( diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStrategy.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStrategy.kt index 53e22fff..1a869477 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStrategy.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStrategy.kt @@ -19,6 +19,7 @@ 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.Constraint +import jetbrains.mps.logic.reactor.program.IncrementalContractViolationException import jetbrains.mps.logic.reactor.program.IncrementalSpec import jetbrains.mps.logic.reactor.program.Rule @@ -192,7 +193,7 @@ internal class IncrementalProcessing( override fun offerMatch(match: RuleMatchEx) = !match.allHeads().filter { it.isPrincipal }.toList().let { if (it.isNotEmpty()) { - rewinder.receive(it.asSequence()) + rewinder.receiveStrict(it.asSequence(), match) } else false } @@ -250,13 +251,20 @@ internal class IncrementalProcessing( postponer.postponeFutureMatches(matches) } else matches + private fun RewindVolatileOccurrencesStage.receiveStrict(occs: Sequence, match: RuleMatchEx) = + rewinder.receive(occs).also { + if (ispec.assertLevel() == IncrementalSpec.AssertLevel.AssertContracts) { + throw IncrementalContractViolationException( + "Incremental processing can't guarantee correctness for match ${match.rule().uniqueTag()}" + ) + } + } private fun RewindVolatileOccurrencesStage.rewind(cursor: ChunkReader): Boolean = maybeReset(cursor)?.let { rewindPos -> // This is the only place where journal can be reset to past // fixme: modifying not through cursor; not the cleanest way posTracker.rewind(rewindPos) - // fixme: modifying not through cursor; not the cleanest way // Some stages on rewind need to take actions (e.g. clear internal state) continuator.onRewind(cursor) diff --git a/reactor/Test/test/TestIncrementalProgram.kt b/reactor/Test/test/TestIncrementalProgram.kt index 7c3861d5..4959a8a2 100644 --- a/reactor/Test/test/TestIncrementalProgram.kt +++ b/reactor/Test/test/TestIncrementalProgram.kt @@ -1479,7 +1479,7 @@ class TestIncrementalProgram { } @Test - fun observerReactivatesFutureOccurrence() { + fun observerCantReactivateFutureOccurrence() { val progSpec = MockIncrProgSpec( setOf("main", "bindVar", "produceBound", "eliminateBound"), setOf(sym1("foo")) @@ -1519,7 +1519,7 @@ class TestIncrementalProgram { // Insert rule before produceBound: test whether // occurrence reactivation due to var bind can occur // before this occurrence is actually activated, - // which leads, of course, to inconsistent state. + // which would lead, of course, to inconsistent state. builder.insertRulesAt(1, rule("bindVar", headKept( @@ -1530,60 +1530,13 @@ class TestIncrementalProgram { )) ).relaunch("test reactivation", progSpec, evalRes.token()) { result -> - // Currently incremental processing can't handle such rule dependencies - // stemming from incremental changes to logicals, - // so "eliminateBound" rule won't be reexecuted. - // But reactivation of hasBound shouldn't pass either, - // because 'hasBound' isn't in store at the point of tried reactivation. - result.storeView().constraintSymbols() shouldBe setOf(sym1("foo"), sym1("hasBound")) - // NB: this is correct if incremental processing // takes into account rule dependencies due to logicals. - // result.storeView().constraintSymbols() shouldBe setOf(sym1("foo"), sym0("unexpected")) + result.storeView().constraintSymbols() shouldBe setOf(sym1("foo")) } } } - @Ignore("feature is superseded") - @Test(expected = EvaluationFailureException::class) - fun violatePrincipalLogicalContract() { - val progSpec = MockIncrProgSpec( - setOf("main", "produceBound"), - setOf(sym0("main"), sym1("foo")) - ).withContractChecks() - - val (X, Y) = metaLogical("X", "Y") - programWithRules( - rule("main", - headReplaced( - pconstraint("main") - ), - body( - pconstraint("foo", X) - )), - rule("produceBound", - headKept( - pconstraint("foo", X) - ), - body( - statement({ x, y -> eq(x,y) }, X, Y), - constraint("hasBound", Y) - )), - rule("bindVar", - headKept( - constraint("hasBound", Y) - ), - body( - statement({ y -> y.set(42) }, Y) - )) - ).launch("violate contract assertion", progSpec) { result -> - - //NB: this check isn't supposed to be even run because of exception - result.storeView().constraintSymbols() shouldBe setOf(sym1("foo"), sym1("hasBound")) - - } - } - @Test fun inference_insertLogicalUni_2() { val progSpec = MockIncrProgSpec( @@ -1639,7 +1592,7 @@ class TestIncrementalProgram { val progSpec = MockIncrProgSpec( setOf("main", "bindVar", "beforeBind"), setOf(sym0("main"), sym1("foo")) - ).withContractChecks() + ) val X = metaLogical("X") @@ -1694,7 +1647,7 @@ class TestIncrementalProgram { val progSpec = MockIncrProgSpec( setOf("main", "bindVar", "checkFresh"), setOf(sym0("main"), sym1("foo")) - ).withContractChecks() + ) val X = metaLogical("X") @@ -1745,7 +1698,7 @@ class TestIncrementalProgram { val progSpec = MockIncrProgSpec( setOf("main", "produceBound", "beforeProduceBound"), setOf(sym0("main"), sym1("foo")) - ).withContractChecks() + ) val (X, Y) = metaLogical("X", "Y") @@ -1805,7 +1758,7 @@ class TestIncrementalProgram { val progSpec = MockIncrProgSpec( setOf("main", "bindVar", "beforeBind"), setOf(sym0("main"), sym1("foo")) - ).withContractChecks() + ) val X = metaLogical("X") @@ -1858,7 +1811,7 @@ class TestIncrementalProgram { val progSpec = MockIncrProgSpec( setOf("main", "produceBound", "beforeProduceBound", "runFoo"), setOf(sym0("main"), sym1("foo"), sym0("runFoo")) - ).withContractChecks() + ) val (X, Y) = metaLogical("X", "Y") @@ -1928,7 +1881,7 @@ class TestIncrementalProgram { val progSpec = MockIncrProgSpec( setOf("main", "produceBound", "uniVars", "bindVar", "checkFresh"), setOf(sym0("main"), sym1("foo"), sym1("bar")) - ).withContractChecks() + ) val (X, Y) = metaLogical("X", "Y")