Introduce processing strategy for reusing occurrences from cache

CachedOccurrencesProcessing simply activates occurrences
passed to it as input before proceeding with 'main'.
Use it instead of PreambleProcessing (that works with journal).

This should help if 'main' doesn't lead to duplicate activations
of these occurrences. So coderules program must be aware of
occurrence cache and should not include rules for cached occ-s.

reactor/Core/src/jetbrains/mps/logic/reactor/core/Controller.kt

@@ -16,6 +16,7 @@
 
 package jetbrains.mps.logic.reactor.core
 
+import jetbrains.mps.logic.reactor.core.internal.ConstraintsProcessing
 import jetbrains.mps.logic.reactor.core.internal.FeedbackStatus
 import jetbrains.mps.logic.reactor.core.internal.MatchJournal
 import jetbrains.mps.logic.reactor.evaluation.PredicateInvocation

reactor/Core/src/jetbrains/mps/logic/reactor/core/Dispatcher.kt

@@ -23,6 +23,8 @@ typealias DispatchingFrontState = Map<Any, RuleMatcher>
 
 internal fun emptyFrontState(): DispatchingFrontState = emptyMap()
 
+internal fun DispatchingFrontState.resetLookup() = apply { values.forEach(RuleMatcher::resetRuleLookup) }
+
 /**
  * A front-end interface to [RuleMatcher].
  * 

reactor/Core/src/jetbrains/mps/logic/reactor/core/RuleMatcher.kt

@@ -41,6 +41,8 @@ interface RuleMatcher {
 
     fun setRuleLookup(ruleLookup: RuleLookup): Unit
 
+    fun resetRuleLookup(): Unit
+
 }
 
 fun createRuleMatcher(lookup: RuleLookup, tag: Any): RuleMatcher = ReteRuleMatcherImpl(lookup, tag)

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

@@ -78,6 +78,23 @@ internal class ConstraintsProcessing(
         return processActivated(controller, activeOcc, parent, FeedbackStatus.NORMAL())
     }
 
+    fun evaluate(controller: Controller, prototype: Occurrence, inStatus: FeedbackStatus) : FeedbackStatus =
+        profiler.profile<FeedbackStatus>("activate_${prototype.constraint().symbol()}") {
+            // fixme: ensure justifications are tracked (incremented) correctly in processing & creator
+            // NB: provide new justifications instead of occ.justifications
+            with(JustifiedOccurrenceCreator(evidence(), initialChunk().justifications())) {
+
+                prototype.constraint.occurrence(
+                    controller.logicalStateObservable(), prototype.arguments(), prototype.logicalContext(), prototype.ruleUniqueTag()
+                ).let { occ ->
+                    trace.activate(occ)
+                    processActivated(controller, occ, initialChunk(), inStatus)
+                }
+
+            }
+        }
+
+
     /**
      * Called to update the state with the currently active constraint occurrence.
      * Calls the controller to process matches (if any) that were triggered.
@@ -105,14 +122,7 @@ internal class ConstraintsProcessing(
         val matches = dispatchingFront.matches().toList()
         val currentMatches = incrementalProcessing.processOccurrenceMatches(active, matches)
 
-        val outStatus = currentMatches.fold(inStatus) { status, match ->
-            // TODO: paranoid check. should be isAlive() instead
-            // FIXME: move this check elsewhere
-            if (status.operational && active.stored && match.allStored()) {
-                assert(match.allHeads().contains(active))
-                processMatch(controller, match, parent, status)
-            } else status
-        }
+        val outStatus = processMatches(controller, active, currentMatches, parent, inStatus)
 
         // TODO: should be isAlive()
         if (active.stored) {
@@ -122,6 +132,16 @@ internal class ConstraintsProcessing(
         return outStatus
     }
 
+    fun processMatches(controller: Controller, active: Occurrence, matches: List<RuleMatchEx>, parent: MatchJournal.MatchChunk, inStatus: FeedbackStatus) : FeedbackStatus =
+        matches.fold(inStatus) { status, match ->
+            // TODO: paranoid check. should be isAlive() instead
+            // FIXME: move this check elsewhere
+            if (status.operational && active.stored && match.allStored()) {
+                assert(match.allHeads().contains(active))
+                processMatch(controller, match, parent, status)
+            } else status
+        }
+
 
     private inline fun FeedbackStatus.then(action: (FeedbackStatus) -> FeedbackStatus) : FeedbackStatus =
         if (operational) action(this) else this
@@ -205,10 +225,10 @@ internal class ConstraintsProcessing(
     /**
      * Encapsulates logic for deriving [Evidence] and [Justifications] for a new [Occurrence].
      */
-    inner class JustifiedOccurrenceCreator {
-        val savedEvidence: Evidence = evidence()
-        val savedJustifications: Justifications = justifications()
-
+    inner class JustifiedOccurrenceCreator(
+        private val savedEvidence: Evidence = evidence(),
+        private val savedJustifications: Justifications = justifications()
+    ) {
         fun Constraint.occurrence(
             observable: LogicalStateObservable,
             arguments: List<*>,

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

@@ -213,11 +213,14 @@ internal class ControllerImpl (
 
         return context.currentStatus()
     }
-    
+
     private fun activateConstraint(constraint: Constraint, parent: MatchJournal.MatchChunk, creator: ConstraintsProcessing.JustifiedOccurrenceCreator, context: Context) : Boolean {
         val args = supervisor.instantiateArguments(constraint.arguments(), context.logicalContext, context)
-        return context.eval { status ->
+        return activateConstraint(constraint, args, parent, creator, context)
+    }
 
+    private fun activateConstraint(constraint: Constraint, args: List<*>, parent: MatchJournal.MatchChunk, creator: ConstraintsProcessing.JustifiedOccurrenceCreator, context: Context) : Boolean =
+        context.eval { status ->
             profiler.profile<FeedbackStatus>("activate_${constraint.symbol()}") {
 
                 with(creator) {
@@ -229,7 +232,6 @@ internal class ControllerImpl (
 
             }
         }
-    }
 
     private fun askPredicate(predicate: Predicate, context: Context) : Boolean =
         profiler.profile<Boolean>("ask_${predicate.symbol()}") {

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

@@ -28,6 +28,11 @@ import java.util.*
  */
 
 
+internal typealias OccurrenceStore = Collection<Occurrence>
+
+internal fun emptyStore(): OccurrenceStore = emptyList()
+
+
 internal interface ProcessingSession {
 
     fun firstSession(): SessionParts
@@ -67,7 +72,7 @@ internal class EvaluationSessionImpl private constructor (
     @Suppress("UNCHECKED_CAST")
     override fun <T : Any> parameter(key: ParameterKey<T>): T? = params ?.get(key) as T
 
-    private fun launch(token: SessionToken?, main: Constraint): EvaluationResult {
+    private fun launch(token: SessionToken?, store: OccurrenceStore, main: Constraint): EvaluationResult {
         val sessionProcessing: ProcessingSession =
             with(incrementality) {
                 when {
@@ -82,10 +87,12 @@ internal class EvaluationSessionImpl private constructor (
 
         with (sessionProcessing) {
             val session =
-                if (token != null)
+                if (token != null) {
+                    (token as? SessionTokenImpl)?.setStore(store)
                     nextSession(token)
-                else
+                } else {
                     firstSession()
+                }
 
             return runSession(session, main)
         }
@@ -119,6 +126,7 @@ internal class EvaluationSessionImpl private constructor (
         override fun endSession(session: SessionParts): SessionToken = with(session) {
             SessionTokenImpl(
                 journal.view(),
+                emptyList(),
                 ruleIndex.toRules(),
                 emptyFrontState(),
                 logicalState,
@@ -161,8 +169,8 @@ internal class EvaluationSessionImpl private constructor (
             val histView = journal.view()
             processing.resetStore() // clear observers
             // todo: need clearing occurrenceContractObservers? (MPSCR-66)
-            val principalState = sessionState(frontState)
-            return SessionTokenImpl(histView, ruleIndex.toRules(), principalState, logicalState, ruleIndex)
+            val principalState = sessionState(frontState).resetLookup()
+            return SessionTokenImpl(histView, emptyList(), ruleIndex.toRules(), principalState, logicalState, ruleIndex)
         }
 
         /**
@@ -176,65 +184,72 @@ internal class EvaluationSessionImpl private constructor (
     }
 
     inner class PreambleProcessingSession(): IncrementalProcessingSession() {
-        private var tkn: SessionToken? = null
+        private var inputStore: OccurrenceStore = emptyStore()
 
         override fun nextSession(token: SessionToken): SessionParts {
             val tkn = token as SessionTokenImpl
             val logicalState = LogicalState()
 
+            // fixme: ensure why ruleIndex.update here leads to errors
+            // val ruleIndex = tkn.ruleIndex.apply{ updateIndexFromRules(program.rules()) }
             val ruleIndex = RuleIndex(program.rules())
-            val journal = MatchJournalImpl(incrementality, tkn.journalView as MatchJournal.View)
-            val front = Dispatcher(ruleIndex, tkn.getFrontState()).front()
+            val front = Dispatcher(ruleIndex).front() // new dispatcher front
+            val journal = MatchJournalImpl(incrementality) // new journal
             val processing = ConstraintsProcessing(front, journal, logicalState, incrementality, trace, profiler)
 
-            val processingStrategy = PreambleProcessing(
-                incrementality, journal, program.newRules(), ruleIndex, trace
-            )
+            this.inputStore = tkn.principalStore
+            val processingStrategy = CachedOccurrencesProcessing(incrementality, inputStore)
             processing.setStrategy(processingStrategy)
 
             val controller = ControllerImpl(supervisor, processing, incrementality, trace, profiler)
 
-            this.tkn = tkn
             return SessionParts(program.preambleInfo(), ruleIndex, journal, logicalState, controller, processing, processingStrategy)
         }
 
-        // Compute Preamble token only once, after firstSession()
         override fun endSession(session: SessionParts): SessionToken = with(session) {
-            if (tkn == null) { // if it is a first run
+            val histView = journal.view()
+            val outputOccurrences = histView.filterOccurrences(inputStore)
+            // todo: make output store in other processing strategies?
 
-                val histView = getPreamble(session)
+            processing.resetStore() // clear observers
+            // todo: need clearing occurrenceContractObservers? (MPSCR-66)
+            logicalState.reset()
+            outputOccurrences.forEach{ it.terminate(logicalState) }
 
-                processing.resetStore() // clear observers
-                // todo: need clearing occurrenceContractObservers? (MPSCR-66)
+            val rules = ruleIndex.toRules().filter(preambleInfo::inPreamble)
 
-                val principalState = sessionState(frontState).filterValues { ruleMatcher ->
-                    preambleInfo.inPreamble(ruleMatcher.rule())
-                }
-
-                val rules = ruleIndex.toRules().filter(preambleInfo::inPreamble)
-
-                logicalState.reset()
-                tkn = SessionTokenImpl(histView, rules, principalState, logicalState, ruleIndex)
-            }
-            return tkn!!
+            SessionTokenImpl(histView, outputOccurrences, rules, emptyFrontState(), LogicalState(), ruleIndex)
         }
 
-        // get preamble and clear
+        private fun MatchJournal.View.filterOccurrences(without: OccurrenceStore = emptyStore()): OccurrenceStore {
+            val withoutSources = without.asSequence().map { it.cacheKey() }.toHashSet()
+            return chunks.asSequence()
+                .mapNotNull { (it as? MatchJournal.OccChunk)?.occ }
+                .filter { it.stored }
+                .filter { !withoutSources.contains(it.cacheKey()) }
+                .toList()
+        }
+
+        private fun ConstraintOccurrence.cacheKey() = sourceRule() to constraint().symbol()
+
         private fun getPreamble(session: SessionParts): MatchJournal.View = with(session) {
             val journalView = journal.view()
             val preambleView = journalView.getPreamble(preambleInfo)
+            // unnecessary work if dispatcher state isn't passed in SessionToken
+            // cleanNonPreamble(processing, journalView.chunks, preambleView)
+            preambleView
+        }
 
-            val cleaner = processing.getStateCleaner()
+        private fun cleanNonPreamble(processing: ConstraintsProcessing, chunks: Iterable<MatchJournal.Chunk>, preambleView: MatchJournal.View) {
             val preambleSet = preambleView.chunks.toSet()
+            val cleaner = processing.getStateCleaner()
 
-            journalView.chunks.asSequence().filterNot(preambleSet::contains).forEach {
+            chunks.asSequence().filterNot(preambleSet::contains).forEach {
                 it.activatedLog().forEach(cleaner::erase)
-
-                if (it is MatchJournal.MatchChunk)
+                if (it is MatchJournal.MatchChunk) {
                     cleaner.erase(it.match as RuleMatchEx)
+                }
             }
-
-            return preambleView
         }
     }
 
@@ -249,6 +264,8 @@ internal class EvaluationSessionImpl private constructor (
 
         var token: SessionToken? = null
 
+        var principalStore: Collection<Occurrence> = emptyList()
+
         override fun withTrace(computingTracer: EvaluationTrace): EvaluationSession.Config {
             this.evaluationTrace = computingTracer
             return this
@@ -258,6 +275,11 @@ internal class EvaluationSessionImpl private constructor (
             return this
         }
 
+        override fun withStore(store: Collection<ConstraintOccurrence>): EvaluationSession.Config {
+            this.principalStore = store as Collection<Occurrence>
+            return this
+        }
+
         override fun withSessionToken(token: SessionToken?): EvaluationSession.Config {
             this.token = token
             return this
@@ -286,7 +308,7 @@ internal class EvaluationSessionImpl private constructor (
             Backend.ourBackend.ourSession.set(session)
             try {
                 val main = parameters[ParameterKey.of("main", Constraint::class.java)] as Constraint
-                return session.launch(token, main)
+                return session.launch(token, principalStore, main)
             }
             finally {
                 try {

reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/OccurrenceContractObserver.kt

@@ -26,6 +26,13 @@ import jetbrains.mps.unification.Term
 import kotlin.collections.HashSet
 
 
+/**
+ * Ensures important contract of incremental algorithm.
+ *
+ * Contract states that arguments of principal [Occurrence]s
+ * must be immutable, that is, logicals can't be unified
+ * with either ground or free other logicals.
+ */
 internal class OccurrenceContractObserver(private val observable: LogicalStateObservable, override val ispec: IncrementalSpec): IncrSpecHolder {
     private val observers: HashMap<Int, UnmodifiableLogicalObserver> = hashMapOf()
 

reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStrategy.kt

@@ -231,6 +231,77 @@ internal class PreambleProcessing(
 }
 
 
+/**
+ * Strategy that works with occurrence store.
+ * Doesn't directly work with [MatchJournal] except for default logging.
+ * Strategy simply activates [Occurrence]s passed as input store
+ *
+ * Caller is responsible for handling new computed principal [Occurrences]:
+ * it can get them from [MatchJournal] and output for putting into cache.
+ *
+ * It requires that program in question adheres to incremental contracts.
+ * Importantly, one ensured by [OccurrenceContractObserver].
+ */
+internal class CachedOccurrencesProcessing(
+    override val ispec: IncrementalSpec,
+    private val occurrences: OccurrenceStore
+): ProcessingStrategy, IncrSpecHolder {
+
+    private var inPreamble = true
+
+    private val postponedMatches: MutableList<Pair<Occurrence, List<RuleMatchEx>>> = mutableListOf()
+
+
+    override fun invalidatedFeedback(): FeedbackKeySet = emptySet()
+
+    override fun invalidatedRules(): List<Any> = emptyList()
+
+    override fun processMatch(match: RuleMatchEx) = Unit
+
+    override fun processOccurrenceMatches(active: Occurrence, matches: List<RuleMatchEx>) =
+        if (inPreamble && matches.isNotEmpty()) {
+            postponedMatches.add(active to matches)
+            emptyList<RuleMatchEx>()
+        } else matches
+
+
+    override fun run(processing: ConstraintsProcessing, controller: Controller, main: Constraint): FeedbackStatus {
+        var status: FeedbackStatus = FeedbackStatus.NORMAL()
+
+        // NB: assume evaluation order doesn't matter for these occurrences
+
+        // first activate cached occurrences,
+        // but don't process their matches right away
+        this.inPreamble = true
+        for (occ in occurrences) {
+            if (occ.constraint().symbol() == main.symbol())
+                continue
+            status = processing.evaluate(controller, occ, status)
+            if (!status.operational)
+                return status
+        }
+        this.inPreamble = false
+
+        // then continue matches caused by cached occurrences
+        status = continueMatches(processing, controller, status)
+
+        // then proceed with normal execution
+        if (status.operational)
+            status = controller.activate(main)
+
+        return status
+    }
+
+    private fun continueMatches(processing: ConstraintsProcessing, controller: Controller, inStatus: FeedbackStatus): FeedbackStatus {
+        val parentChunk = processing.initialChunk() // fixme: get activation chunk of active occ?
+        val status = postponedMatches.fold(inStatus) { status, (active, matches) ->
+            processing.processMatches(controller, active, matches, parentChunk, status)
+        }
+        return status
+    }
+}
+
+
 private fun ContinueOccurrencesStage.runContinued(processing: ConstraintsProcessing, controller: Controller, chunkReader: ChunkReader): FeedbackStatus {
     var status: FeedbackStatus = FeedbackStatus.NORMAL()
     val parentChunk = processing.parentChunk()

reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ReteRuleMatcherImpl.kt

@@ -46,7 +46,7 @@ typealias SignatureIndex = TIntObjectHashMap<List<Signature>>
 
 fun signatureIndexOf() = TIntObjectHashMap<MutableList<Signature>>()
 
-internal class ReteRuleMatcherImpl(private var ruleLookup: RuleLookup,
+internal class ReteRuleMatcherImpl(private var ruleLookup: RuleLookup?,
                                    private val tag: Any) : RuleMatcher
 {
 
@@ -54,12 +54,14 @@ internal class ReteRuleMatcherImpl(private var ruleLookup: RuleLookup,
 
     val propagation = lookupRule().headReplaced().count() == 0
 
-    fun lookupRule(): Rule = ruleLookup.lookupRuleByTag(tag) ?: throw IllegalStateException("can't lookup rule by tag: '${tag}'")
+    fun lookupRule(): Rule = ruleLookup?.lookupRuleByTag(tag) ?: throw IllegalStateException("can't lookup rule by tag: '${tag}'")
 
     override fun rule() = lookupRule()
 
     override fun setRuleLookup(ruleLookup: RuleLookup) { this.ruleLookup = ruleLookup }
 
+    override fun resetRuleLookup() { this.ruleLookup = null }
+
     override fun newProbe(): RuleMatchingProbe = ReteNetwork(head.size).also { probe = it }
 
     override fun probe(): RuleMatchingProbe = probe ?: newProbe()

reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/RuleMatcherImpl.kt

@@ -33,7 +33,7 @@ import kotlin.collections.ArrayList
  *
  * @author Fedor Isakov
  */
-internal class RuleMatcherImpl(private var ruleLookup: RuleLookup,
+internal class RuleMatcherImpl(private var ruleLookup: RuleLookup?,
                                private val tag: Any) : RuleMatcher
 {
 
@@ -41,12 +41,14 @@ internal class RuleMatcherImpl(private var ruleLookup: RuleLookup,
 
     val propagation = lookupRule().headReplaced().count() == 0
 
-    fun lookupRule(): Rule = ruleLookup.lookupRuleByTag(tag) ?: throw IllegalStateException("can't lookup rule by tag: '${tag}'")
+    fun lookupRule(): Rule = ruleLookup?.lookupRuleByTag(tag) ?: throw IllegalStateException("can't lookup rule by tag: '${tag}'")
 
     override fun rule() = lookupRule()
 
     override fun setRuleLookup(ruleLookup: RuleLookup) { this.ruleLookup = ruleLookup }
 
+    override fun resetRuleLookup() { this.ruleLookup = null }
+
     override fun newProbe(): RuleMatchingProbe =
         RuleMatchFront(emptyList(),
                         emptyList(),

reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/SessionTokenImpl.kt

@@ -17,6 +17,7 @@
 package jetbrains.mps.logic.reactor.core.internal
 
 import jetbrains.mps.logic.reactor.core.DispatchingFrontState
+import jetbrains.mps.logic.reactor.core.Occurrence
 import jetbrains.mps.logic.reactor.core.RuleIndex
 import jetbrains.mps.logic.reactor.evaluation.MatchJournalView
 import jetbrains.mps.logic.reactor.evaluation.SessionToken
@@ -24,6 +25,7 @@ import jetbrains.mps.logic.reactor.program.Rule
 
 data class SessionTokenImpl(
                             private val journalView: MatchJournal.View,
+                            private var store: Collection<Occurrence>,
                             private val rules: Iterable<Rule>,
                             private val frontState: DispatchingFrontState,
                             val logicalState: LogicalState,
@@ -31,5 +33,8 @@ data class SessionTokenImpl(
 {
     override fun getJournalView(): MatchJournalView = journalView
     override fun getRules(): Iterable<Rule> = rules
+    override fun getPrincipalStore(): Collection<Occurrence> = store
+
     fun getFrontState(): DispatchingFrontState = frontState
+    fun setStore(newStore: Collection<Occurrence>) { this.store = newStore }
 }

reactor/Core/src/jetbrains/mps/logic/reactor/evaluation/EvaluationSession.java

@@ -19,6 +19,7 @@ package jetbrains.mps.logic.reactor.evaluation;
 
 import jetbrains.mps.logic.reactor.program.IncrementalSpec;
 import jetbrains.mps.logic.reactor.program.Program;
+import java.util.Collection;
 
 /**
  * The starting point to evaluate a program.
@@ -83,6 +84,8 @@ public abstract class EvaluationSession {
         @Deprecated
         public abstract Config withStoreView(StoreView storeView);
 
+        public Config withStore(Collection<ConstraintOccurrence> store) { return this; }
+
         public Config withSessionToken(SessionToken token) { return this; }
 
         public Config withIncrSpec(IncrementalSpec ispec) { return this; }

reactor/Core/src/jetbrains/mps/logic/reactor/evaluation/SessionToken.java

@@ -18,10 +18,13 @@ package jetbrains.mps.logic.reactor.evaluation;
 
 import jetbrains.mps.logic.reactor.program.Rule;
 import org.jetbrains.annotations.NotNull;
+import java.util.Collection;
 
 public interface SessionToken {
     @NotNull()
     MatchJournalView getJournalView();
     @NotNull()
     Iterable<Rule> getRules();
+    @NotNull
+    Collection<ConstraintOccurrence> getPrincipalStore();
 }