Introduce IncrementalStage-s (MPSCR-71)
Incremental algorithm is stricter now in relation to how it reads MatchJournal and operates on it. Previously each incremental stage required full traversal. Now all stages run on each chunk in a single traversal. Separate journal traversal from incremental logic required for each Chunk.
This commit is contained in:
Grigorii Kirgizov
parent
19fa16c731
commit
1bcf80a9b2
|
|
@ -0,0 +1,25 @@
|
|||
/*
|
||||
* 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
|
||||
|
||||
interface ComparatorExt<T>: Comparator<T> {
|
||||
infix fun T.before(other: T): Boolean = compare(this, other) < 0
|
||||
infix fun T.beforeOrEq(other: T): Boolean = compare(this, other) <= 0
|
||||
|
||||
infix fun T.after(other: T): Boolean = compare(this, other) > 0
|
||||
infix fun T.afterOrEq(other: T): Boolean = compare(this, other) >= 0
|
||||
}
|
||||
|
|
@ -26,7 +26,6 @@ import jetbrains.mps.logic.reactor.program.Rule
|
|||
import jetbrains.mps.logic.reactor.util.Profiler
|
||||
import jetbrains.mps.logic.reactor.util.profile
|
||||
import kotlin.collections.ArrayList
|
||||
import kotlin.collections.HashSet
|
||||
|
||||
|
||||
/**
|
||||
|
|
@ -39,26 +38,73 @@ import kotlin.collections.HashSet
|
|||
*
|
||||
* @author Fedor Isakov
|
||||
*/
|
||||
internal class ConstraintsProcessing(
|
||||
private var dispatchingFront: Dispatcher.DispatchingFront,
|
||||
journal: MatchJournalImpl,
|
||||
private val ruleIndex: RuleIndex,
|
||||
val logicalState: LogicalState,
|
||||
override val ispec: IncrementalSpec = IncrementalSpec.DefaultSpec,
|
||||
val trace: EvaluationTrace = EvaluationTrace.NULL,
|
||||
val profiler: Profiler? = null
|
||||
|
||||
) : StoreAwareJournalImpl(journal, logicalState), IncrSpecHolder {
|
||||
|
||||
internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.DispatchingFront,
|
||||
journal: MatchJournalImpl,
|
||||
private val ruleIndex: RuleIndex,
|
||||
val logicalState: LogicalState,
|
||||
override val ispec: IncrementalSpec = IncrementalSpec.DefaultSpec,
|
||||
val trace: EvaluationTrace = EvaluationTrace.NULL,
|
||||
val profiler: Profiler? = null)
|
||||
: StoreAwareJournalImpl(journal, logicalState), IncrSpecHolder
|
||||
{
|
||||
private val journalIndex: MatchJournal.Index = journal.index()
|
||||
|
||||
private val activationQueue: ContinuedActivationQueue = ContinuedActivationQueue(journalIndex, RuleOrdering(ruleIndex))
|
||||
|
||||
private val occurrenceContractObserver: OccurrenceContractObserver? =
|
||||
if (ispec.assertLevel().assertContracts()) OccurrenceContractObserver(logicalState, ispec) else null
|
||||
|
||||
private val invalidFeedbackKeys: MutableSet<Any> = mutableSetOf<Any>()
|
||||
|
||||
private data class MatchCandidate(val rule: Rule, val occChunk: MatchJournal.OccChunk)
|
||||
inner class ProgramStateCleaner{
|
||||
fun erase(occurrence: Occurrence) {
|
||||
dispatchingFront = dispatchingFront.forget(occurrence)
|
||||
occurrence.clearLogicalState(logicalState)
|
||||
}
|
||||
|
||||
fun erase(match: RuleMatchEx) {
|
||||
dispatchingFront = dispatchingFront.forget(match)
|
||||
}
|
||||
}
|
||||
val stateCleaner = ProgramStateCleaner()
|
||||
|
||||
|
||||
//TODO: move to MatchJournal
|
||||
/**
|
||||
* Preserves last valid [Chunk] while iterating on Chunks and removing them through [nextOrRemove].
|
||||
*/
|
||||
private class ChunkRemover(private val it: MutableIterator<MatchJournal.Chunk>): MatchJournal.ChunkReader {
|
||||
private var _current_: MatchJournal.Chunk
|
||||
private var _previous_: MatchJournal.Chunk
|
||||
|
||||
init {
|
||||
assert(it.hasNext()) {"initial chunk must be always present"}
|
||||
_current_ = it.next()
|
||||
_previous_ = _current_
|
||||
}
|
||||
|
||||
/**
|
||||
* Removes next [Chunk] if [action] returns [true], else just advances.
|
||||
* @return [true] if iterator advanced (i.e. next [Chunk] wasn't removed), [false] otherwise.
|
||||
*/
|
||||
inline fun nextOrRemove(action: (MatchJournal.ChunkReader) -> Boolean): Boolean {
|
||||
val lastPrev = _previous_
|
||||
_previous_ = _current_
|
||||
_current_ = it.next()
|
||||
|
||||
val toRemove = action(this)
|
||||
if (toRemove) {
|
||||
it.remove()
|
||||
// don't advance
|
||||
_current_ = _previous_
|
||||
_previous_ = lastPrev
|
||||
}
|
||||
return !toRemove
|
||||
}
|
||||
|
||||
override val current: MatchJournal.Chunk get() = _current_
|
||||
override val previous: MatchJournal.Chunk get() = _previous_
|
||||
override val reachedEnd: Boolean get() = !it.hasNext()
|
||||
}
|
||||
|
||||
|
||||
fun activateContinue(controller: Controller, activeOcc: Occurrence, parent: MatchJournal.MatchChunk): FeedbackStatus {
|
||||
|
|
@ -73,145 +119,73 @@ internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.Di
|
|||
return processActivated(controller, activeOcc, parent, FeedbackStatus.NORMAL())
|
||||
}
|
||||
|
||||
/**
|
||||
* Invalidation includes several activities:
|
||||
* - removing chunks (i.e. principal matches) corresponding to
|
||||
* removed rules and chunks depending on them from journal
|
||||
* - reactivating occurrences that led to invalidated matches
|
||||
* - pruning invalidated occurrences and matches from Dispatcher's state
|
||||
*/
|
||||
fun invalidateRuleMatches(ruleIds: Set<Any>) {
|
||||
// Track only root invalidated entities for invalidation,
|
||||
// relying on transitivity of Justified.justifiedBy relation.
|
||||
val justificationRoots = mutableListOf<Justified>()
|
||||
private lateinit var invalidator: InvalidationStage
|
||||
private lateinit var adder: AdditionStage
|
||||
private lateinit var postponer: PostponeMatchesStage
|
||||
private lateinit var continuator: ContinueOccurrencesStage
|
||||
|
||||
val it = this.iterator()
|
||||
var lastValidChunk = it.next() // skip initial chunk
|
||||
fun invalidateAndAddRules(controller: Controller, rulesDiff: RulesDiff): FeedbackStatus {
|
||||
val ruleOrdering = RuleOrdering(ruleIndex)
|
||||
|
||||
while (it.hasNext()) {
|
||||
val chunk = it.next()
|
||||
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)
|
||||
|
||||
if (chunk is MatchJournal.MatchChunk && chunk.dependsOnAny(ruleIds)) {
|
||||
justificationRoots.add(chunk)
|
||||
}
|
||||
val chunkReader = ChunkRemover(this.iterator())
|
||||
|
||||
// Invalidating dependent chunks
|
||||
if (chunk.justifiedByAny(justificationRoots)) {
|
||||
var status: FeedbackStatus = FeedbackStatus.NORMAL()
|
||||
while (!chunkReader.reachedEnd) {
|
||||
val advanced = chunkReader.nextOrRemove(invalidator::onNext)
|
||||
if (!advanced) continue
|
||||
|
||||
// Remove chunk from the journal
|
||||
it.remove()
|
||||
adder.onNext(chunkReader)
|
||||
|
||||
val validOccs = invalidateChunk(chunk, justificationRoots)
|
||||
postponer.onNext(chunkReader)
|
||||
|
||||
activationQueue.offerAll(lastValidChunk.toPos(), validOccs)
|
||||
|
||||
} else {
|
||||
lastValidChunk = chunk
|
||||
}
|
||||
status = continuator.run(controller, chunkReader)
|
||||
if (!status.operational) break
|
||||
}
|
||||
return status
|
||||
}
|
||||
|
||||
private fun invalidateChunk(chunk: MatchJournal.Chunk, invalidJustifications: Collection<Justified>): Iterable<Occurrence> {
|
||||
// 'Undo' all activated in this chunk occurrences: clear Dispatcher & LogicalState
|
||||
chunk.activatedLog().forEach {
|
||||
dispatchingFront = dispatchingFront.forget(it)
|
||||
it.clearLogicalState(logicalState)
|
||||
private fun ContinueOccurrencesStage.run(controller: Controller, chunkReader: MatchJournal.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
|
||||
}
|
||||
}
|
||||
|
||||
val validOccs: Sequence<Occurrence>
|
||||
if (chunk is MatchJournal.MatchChunk) {
|
||||
trace.invalidate(chunk.match)
|
||||
invalidFeedbackKeys.add(chunk.match.feedbackKey)
|
||||
dispatchingFront = dispatchingFront.forget(chunk.match as RuleMatchEx)
|
||||
|
||||
// Valid head occurrences could match more rules
|
||||
// without this match, so need to reactivate them.
|
||||
// E.g. occurrences discarded in this match on
|
||||
// previous run but revived here can match more rules.
|
||||
validOccs = chunk.match.allHeads().filter { occ ->
|
||||
!occ.justifiedByAny(invalidJustifications)
|
||||
}
|
||||
// By definition of Chunk and principal rule,
|
||||
// all occurrences from the head are principal.
|
||||
assert(chunk.match.allHeads().all { it.isPrincipal })
|
||||
|
||||
} else validOccs = emptySequence()
|
||||
return validOccs.asIterable()
|
||||
return status
|
||||
}
|
||||
|
||||
fun addRuleMatches(rules: Iterable<Rule>) {
|
||||
|
||||
val activationCandidates = mutableListOf<MatchCandidate>()
|
||||
val it = this.iterator()
|
||||
var chunk = it.next() // initial chunk
|
||||
var prevChunk = chunk
|
||||
|
||||
while (true) {
|
||||
if (chunk is MatchJournal.OccChunk) {
|
||||
// filters out rules using occurrence's arguments
|
||||
val allRuleCandidates = ruleIndex.forOccurrence(chunk.occ).map { it.uniqueTag() }.toHashSet()
|
||||
|
||||
for (rule in rules) {
|
||||
if (allRuleCandidates.contains(rule.uniqueTag()) && rule.canMatch(chunk.occ.constraint)) {
|
||||
// Can this rule be matched by principal occurrence?
|
||||
// Then we will need to find the place among existing child chunks
|
||||
// (i.e. among some number of following ones)
|
||||
// to activate this occurrence, to (possibly) match this rule.
|
||||
// Also remember the parent justification of this rule candidate
|
||||
// to drop it from monitoring when child chunks end.
|
||||
activationCandidates.add(MatchCandidate(rule, chunk))
|
||||
// todo: also use the rule to help Dispatcher in future?
|
||||
// i.e. try matching only on the candidate rule
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
val aIt = activationCandidates.listIterator()
|
||||
while (aIt.hasNext()) {
|
||||
val (candRule, occChunk) = aIt.next()
|
||||
|
||||
val pos =
|
||||
if (activationQueue.canBeInserted(candRule, occChunk, chunk))
|
||||
// We need the previous chunk as pos here (i.e. adding after it).
|
||||
prevChunk.toPos()
|
||||
else if (!it.hasNext())
|
||||
// Case when adding at the very end
|
||||
chunk.toPos()
|
||||
else
|
||||
continue
|
||||
|
||||
activationQueue.offer(pos, occChunk)
|
||||
trace.potentialMatch(occChunk.occ, candRule)
|
||||
// Drop the candidate if appropriate activation place is found.
|
||||
aIt.remove()
|
||||
}
|
||||
|
||||
if (!it.hasNext()) break
|
||||
prevChunk = chunk
|
||||
chunk = it.next()
|
||||
}
|
||||
}
|
||||
|
||||
fun launchQueue(controller: Controller): FeedbackStatus =
|
||||
activationQueue.run(controller, this)
|
||||
|
||||
/**
|
||||
* Clears state unneeded between incremental sessions
|
||||
* and returns [SessionToken] with session results.
|
||||
*/
|
||||
fun endSession(): SessionToken {
|
||||
invalidFeedbackKeys.clear()
|
||||
val histView = view()
|
||||
resetStore() // clear observers
|
||||
// todo: need clearing occurrenceContractObservers? (MPSCR-66)
|
||||
val rules = ArrayList<Rule>().apply { ruleIndex.forEach { add(it) } }
|
||||
// preserve only relevant and non-empty RuleMatchers
|
||||
val principalState = dispatchingFront.state().filterValues { ruleMatcher ->
|
||||
ruleMatcher.rule().isPrincipal || ruleMatcher.probe().hasOccurrences()
|
||||
}
|
||||
val principalState = dispatchingFront.sessionState()
|
||||
return SessionTokenImpl(histView, rules, principalState, logicalState.clear(), ruleIndex)
|
||||
}
|
||||
|
||||
fun invalidatedFeedback(): FeedbackKeySet = HashSet<Any>(invalidFeedbackKeys)
|
||||
/**
|
||||
* Preserve data needed between sessions:
|
||||
* preserve only relevant and non-empty RuleMatchers
|
||||
*/
|
||||
private fun Dispatcher.DispatchingFront.sessionState() =
|
||||
this.state().filterValues { ruleMatcher ->
|
||||
ruleMatcher.rule().isPrincipal || ruleMatcher.probe().hasOccurrences()
|
||||
}
|
||||
|
||||
fun invalidatedFeedback(): FeedbackKeySet = invalidator.invalidatedFeedback()
|
||||
|
||||
/**
|
||||
* Called to update the state with the currently active constraint occurrence.
|
||||
|
|
@ -237,12 +211,7 @@ internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.Di
|
|||
}
|
||||
|
||||
val matches = dispatchingFront.matches().toList()
|
||||
val currentMatches =
|
||||
if (isFront() || !active.isPrincipal) {
|
||||
matches
|
||||
} else {
|
||||
activationQueue.postponeFutureMatches(matches)
|
||||
}
|
||||
val currentMatches = postponeFutureMatches(active, matches)
|
||||
|
||||
val outStatus = currentMatches.fold(inStatus) { status, match ->
|
||||
// TODO: paranoid check. should be isAlive() instead
|
||||
|
|
@ -261,6 +230,7 @@ internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.Di
|
|||
return outStatus
|
||||
}
|
||||
|
||||
|
||||
@Deprecated("obsolete machinery, superseded by MPSCR-65")
|
||||
private fun dropDiscardingMatchesFor(ancestor: MatchJournal.OccChunk, droppedRuleTags: MutableSet<Any>) =
|
||||
this.dropDescendantsWhile(ancestor) { chunk ->
|
||||
|
|
@ -296,21 +266,34 @@ internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.Di
|
|||
.also { trace.finish(match) }
|
||||
|
||||
private fun continueReplacedHeads(match: RuleMatchEx, parent: MatchJournal.MatchChunk) {
|
||||
if (!isFront() && match.isPrincipal) {
|
||||
profiler.profile("continueReplaced") {
|
||||
|
||||
val invalidJustifications = match.matchHeadReplaced().filter { it.isPrincipal }
|
||||
if (invalidJustifications.isNotEmpty()) {
|
||||
|
||||
dropDescendants(invalidJustifications) { lastValidChunk, currentChunk ->
|
||||
val validOccs = invalidateChunk(currentChunk, invalidJustifications)
|
||||
activationQueue.offerAll(lastValidChunk.toPos(), validOccs)
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
if (requiresIncrementalProcessing(match)) {
|
||||
val invalidJustifications = match.matchHeadReplaced().filter { it.isPrincipal }
|
||||
invalidator.receive(invalidJustifications)
|
||||
}
|
||||
}
|
||||
// private fun continueReplacedHeads(match: RuleMatchEx, parent: MatchJournal.MatchChunk) {
|
||||
// if (!isFront() && match.isPrincipal) {
|
||||
// profiler.profile("continueReplaced") {
|
||||
//
|
||||
// val invalidJustifications = match.matchHeadReplaced().filter { it.isPrincipal }
|
||||
// if (invalidJustifications.isNotEmpty()) {
|
||||
//
|
||||
// dropDescendants(invalidJustifications) { lastValidChunk, currentChunk ->
|
||||
// val validOccs = invalidator.invalidateChunk(currentChunk, invalidJustifications)
|
||||
// activationQueue.offerAll(lastValidChunk.toPos(), validOccs)
|
||||
// }
|
||||
//
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
// }
|
||||
|
||||
private fun postponeFutureMatches(active: Occurrence, matches: List<RuleMatchEx>) =
|
||||
if (requiresIncrementalProcessing(active)) {
|
||||
postponer.process(active, matches)
|
||||
// activationQueue.postponeFutureMatches(matches)
|
||||
} else matches
|
||||
|
||||
|
||||
private fun accept(controller: Controller, match: RuleMatchEx) {
|
||||
profiler.profile("logMatch") {
|
||||
|
|
@ -348,7 +331,7 @@ internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.Di
|
|||
}
|
||||
|
||||
private fun Occurrence.clearLogicalState(observable: LogicalStateObservable) {
|
||||
terminate(logicalState)
|
||||
terminate(observable)
|
||||
if (occurrenceContractObserver != null && this.isPrincipal) {
|
||||
occurrenceContractObserver.onDiscarded(this)
|
||||
}
|
||||
|
|
@ -386,6 +369,10 @@ internal class ConstraintsProcessing(private var dispatchingFront: Dispatcher.Di
|
|||
}
|
||||
}
|
||||
|
||||
private fun requiresIncrementalProcessing(match: RuleMatchEx) = ispec.ability().allowed() && match.isPrincipal
|
||||
|
||||
private fun requiresIncrementalProcessing(occ: Occurrence) = ispec.ability().allowed() && occ.isPrincipal
|
||||
|
||||
private fun MatchJournal.MatchChunk.dependsOnAny(utags: Iterable<Any>): Boolean =
|
||||
utags.contains(this.ruleUniqueTag) || utags.any { utag -> dependsOnRule(utag) }
|
||||
|
||||
|
|
|
|||
|
|
@ -17,13 +17,18 @@
|
|||
package jetbrains.mps.logic.reactor.core.internal
|
||||
|
||||
import jetbrains.mps.logic.reactor.core.*
|
||||
import jetbrains.mps.logic.reactor.program.Rule
|
||||
import java.util.*
|
||||
|
||||
internal interface ContinuedActivationSink {
|
||||
|
||||
fun offerAll(continueFromPos: MatchJournal.Pos, occs: Iterable<Occurrence>)
|
||||
|
||||
fun offer(continueFromPos: MatchJournal.Pos, ancestor: MatchJournal.OccChunk): Boolean
|
||||
}
|
||||
|
||||
internal class ContinuedActivationQueue(
|
||||
private val journalIndex: MatchJournal.Index,
|
||||
private val ruleOrdering: RuleOrdering
|
||||
) {
|
||||
private val journalIndex: MatchJournal.Index
|
||||
): ContinuedActivationSink {
|
||||
|
||||
/**
|
||||
* Specifies position in [MatchJournal] for continuing program evaluation.
|
||||
|
|
@ -96,8 +101,7 @@ internal class ContinuedActivationQueue(
|
|||
val occChunk = journalIndex.activationPos(m)
|
||||
val pos = occChunk?.toPos()
|
||||
|
||||
// if it is a future match
|
||||
if (pos != null && journalIndex.compare(lastIncrementalRootPos, pos) < 0) {
|
||||
if (pos != null && isFuturePos(pos)) {
|
||||
offer(pos, occChunk)
|
||||
} else {
|
||||
currentMatches.add(m)
|
||||
|
|
@ -106,35 +110,17 @@ internal class ContinuedActivationQueue(
|
|||
return currentMatches
|
||||
}
|
||||
|
||||
/**
|
||||
* Checks whether [candidateRule] can be inserted in journal as a child
|
||||
* of [parentChunk] before one of its child chunks, [beforeChunk].
|
||||
* It is assumed that [candidateRule] can be matched by [Occurrence] from [parentChunk].
|
||||
*/
|
||||
fun canBeInserted(candidateRule: Rule, parentChunk: MatchJournal.OccChunk, beforeChunk: MatchJournal.Chunk): Boolean {
|
||||
// Place to try activating candidate rule is:
|
||||
// either according to the ordering between rules
|
||||
// or as the last one, after all existing activations
|
||||
assert(candidateRule.canMatch(parentChunk.occ.constraint))
|
||||
private fun isFuturePos(pos: MatchJournal.Pos): Boolean =
|
||||
with(journalIndex) { pos after lastIncrementalRootPos }
|
||||
|
||||
val placeToInsertFound = beforeChunk is MatchJournal.MatchChunk
|
||||
&& ruleOrdering.isEarlierThan(candidateRule, beforeChunk.match.rule())
|
||||
|
||||
val isDescendant = beforeChunk.justifiedBy(parentChunk)
|
||||
val isSibling = beforeChunk.evidence == parentChunk.evidence && beforeChunk != parentChunk
|
||||
val childChunksEnded = !isDescendant || isSibling
|
||||
|
||||
return (childChunksEnded || placeToInsertFound)
|
||||
}
|
||||
|
||||
fun offerAll(continueFromPos: MatchJournal.Pos, occs: Iterable<Occurrence>) =
|
||||
override fun offerAll(continueFromPos: MatchJournal.Pos, occs: Iterable<Occurrence>) =
|
||||
occs.forEach {
|
||||
journalIndex.activatingChunkOf(it)?.let { occChunk ->
|
||||
offer(continueFromPos, occChunk)
|
||||
}
|
||||
}
|
||||
|
||||
fun offer(continueFromPos: MatchJournal.Pos, ancestor: MatchJournal.OccChunk): Boolean =
|
||||
override fun offer(continueFromPos: MatchJournal.Pos, ancestor: MatchJournal.OccChunk) =
|
||||
ExecPos(continueFromPos, ancestor).let {
|
||||
it.assertValid()
|
||||
if (seen.add(it)) execQueue.offer(it) else false
|
||||
|
|
|
|||
|
|
@ -56,23 +56,10 @@ internal class ControllerImpl (
|
|||
|
||||
fun incrLaunch(constraint: Constraint, rulesDiff: RulesDiff): Pair<FeedbackStatus, FeedbackKeySet> {
|
||||
|
||||
if (rulesDiff.removed.isNotEmpty()) {
|
||||
profiler.profile("invalidation") {
|
||||
processing.invalidateRuleMatches(rulesDiff.removed)
|
||||
}
|
||||
val status = profiler.profile<FeedbackStatus>("journalTraverse") {
|
||||
processing.invalidateAndAddRules(this, rulesDiff)
|
||||
}
|
||||
|
||||
if (rulesDiff.added.count() > 0) {
|
||||
profiler.profile("adding_matches") {
|
||||
processing.addRuleMatches(rulesDiff.added)
|
||||
}
|
||||
}
|
||||
|
||||
val status =
|
||||
profiler.profile<FeedbackStatus>("reexecution") {
|
||||
processing.launchQueue(this)
|
||||
}
|
||||
|
||||
return status to processing.invalidatedFeedback()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,303 @@
|
|||
/*
|
||||
* 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.core.feedbackKey
|
||||
import jetbrains.mps.logic.reactor.evaluation.EvaluationTrace
|
||||
import jetbrains.mps.logic.reactor.program.IncrementalSpec
|
||||
import jetbrains.mps.logic.reactor.program.Rule
|
||||
|
||||
|
||||
internal interface IncrementalStage: IncrSpecHolder {
|
||||
|
||||
// fun onNext(reader: MatchJournal.ChunkReader)
|
||||
|
||||
}
|
||||
|
||||
typealias RuleTagSet = Set<Any>
|
||||
//typealias ContinuedActivations = Collection<Occurrence>
|
||||
|
||||
|
||||
/**
|
||||
* Invalidation includes several activities:
|
||||
* - removing chunks (i.e. principal matches) corresponding to
|
||||
* removed rules and chunks depending on them from journal
|
||||
* - reactivating occurrences that led to invalidated matches
|
||||
* - pruning invalidated occurrences and matches from Dispatcher's state
|
||||
*/
|
||||
internal class InvalidationStage(
|
||||
override val ispec: IncrementalSpec,
|
||||
// private val invalidRules: RuleTagSet,
|
||||
private val invalidRuleIds: Set<Any>,
|
||||
private val activationSink: ContinuedActivationSink,
|
||||
private val stateCleaner: ConstraintsProcessing.ProgramStateCleaner,
|
||||
private val trace: EvaluationTrace
|
||||
): IncrementalStage {
|
||||
|
||||
private val invalidJustifications = mutableListOf<Justified>()
|
||||
|
||||
private val invalidFeedbackKeys: MutableSet<Any> = mutableSetOf<Any>()
|
||||
|
||||
fun invalidatedFeedback(): FeedbackKeySet = HashSet<Any>(invalidFeedbackKeys)
|
||||
|
||||
|
||||
fun receive(invalid: Iterable<Justified>) { invalidJustifications.addAll(invalid) }
|
||||
|
||||
fun onNext(reader: MatchJournal.ChunkReader): Boolean {
|
||||
val chunk = reader.current
|
||||
if (chunk is MatchJournal.MatchChunk && chunk.dependsOnAny(invalidRuleIds)) {
|
||||
invalidJustifications.add(chunk)
|
||||
}
|
||||
|
||||
// Invalidating dependent chunks
|
||||
if (chunk.justifiedByAny(invalidJustifications)) {
|
||||
|
||||
val validOccs = invalidateChunk(chunk, invalidJustifications)
|
||||
activationSink.offerAll(reader.previous.toPos(), validOccs)
|
||||
|
||||
// Remove chunk from the journal
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// todo: make private
|
||||
fun invalidateChunk(chunk: MatchJournal.Chunk, invalidJustifications: Collection<Justified>): Iterable<Occurrence> {
|
||||
// fun invalidateChunk(chunk: MatchJournal.Chunk): Iterable<Occurrence> {
|
||||
// 'Undo' all activated in this chunk occurrences: clear Dispatcher & LogicalState
|
||||
chunk.activatedLog().forEach(stateCleaner::erase)
|
||||
|
||||
val validOccs: Sequence<Occurrence>
|
||||
if (chunk is MatchJournal.MatchChunk) {
|
||||
trace.invalidate(chunk.match)
|
||||
|
||||
invalidFeedbackKeys.add(chunk.match.feedbackKey)
|
||||
stateCleaner.erase(chunk.match as RuleMatchEx)
|
||||
|
||||
// Valid head occurrences could match more rules
|
||||
// without this match, so need to reactivate them.
|
||||
// E.g. occurrences discarded in this match on
|
||||
// previous run but revived here can match more rules.
|
||||
validOccs = chunk.match.allHeads().filter { occ ->
|
||||
!occ.justifiedByAny(invalidJustifications)
|
||||
}
|
||||
// By definition of Chunk and principal rule,
|
||||
// all occurrences from the head are principal.
|
||||
assert(chunk.match.allHeads().all { it.isPrincipal })
|
||||
|
||||
} else validOccs = emptySequence()
|
||||
return validOccs.asIterable()
|
||||
}
|
||||
|
||||
private fun MatchJournal.MatchChunk.dependsOnAny(utags: Iterable<Any>): Boolean =
|
||||
utags.contains(this.ruleUniqueTag) || utags.any { utag -> dependsOnRule(utag) }
|
||||
|
||||
}
|
||||
|
||||
|
||||
internal class AdditionStage(
|
||||
override val ispec: IncrementalSpec,
|
||||
private val addedRules: Iterable<Rule>,
|
||||
private val activationSink: ContinuedActivationSink,
|
||||
private val ruleOrdering: RuleOrdering,
|
||||
private val ruleIndex: RuleIndex,
|
||||
private val trace: EvaluationTrace
|
||||
): IncrementalStage {
|
||||
|
||||
private data class MatchCandidate(val rule: Rule, val occChunk: MatchJournal.OccChunk)
|
||||
|
||||
private val activationCandidates = mutableListOf<MatchCandidate>()
|
||||
|
||||
private val isEmpty = !addedRules.iterator().hasNext()
|
||||
|
||||
|
||||
fun onNext(reader: MatchJournal.ChunkReader) {
|
||||
if (isEmpty) return
|
||||
|
||||
val chunk = reader.current
|
||||
if (chunk is MatchJournal.OccChunk) {
|
||||
addRuleCandidates(chunk)
|
||||
}
|
||||
offerCandidates(reader)
|
||||
}
|
||||
|
||||
private fun addRuleCandidates(chunk: MatchJournal.OccChunk) {
|
||||
// filters out rules using occurrence's arguments
|
||||
val allRuleCandidates = ruleIndex.forOccurrence(chunk.occ).map { it.uniqueTag() }.toHashSet()
|
||||
|
||||
for (rule in addedRules) {
|
||||
if (allRuleCandidates.contains(rule.uniqueTag()) && rule.canMatch(chunk.occ.constraint)) {
|
||||
// Can this rule be matched by principal occurrence?
|
||||
// Then we will need to find the place among existing child chunks
|
||||
// (i.e. among some number of following ones)
|
||||
// to activate this occurrence, to (possibly) match this rule.
|
||||
// Also remember the parent justification of this rule candidate
|
||||
// to drop it from monitoring when child chunks end.
|
||||
activationCandidates.add(MatchCandidate(rule, chunk))
|
||||
// todo: also use the rule to help Dispatcher in future?
|
||||
// i.e. try matching only on the candidate rule
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private fun offerCandidates(reader: MatchJournal.ChunkReader) {
|
||||
val aIt = activationCandidates.listIterator()
|
||||
while (aIt.hasNext()) {
|
||||
val (candRule, occChunk) = aIt.next()
|
||||
|
||||
val pos =
|
||||
if (ruleOrdering.canBeInserted(candRule, occChunk, reader.current))
|
||||
// We need the previous chunk as pos here (i.e. adding after it).
|
||||
reader.previous.toPos()
|
||||
else if (reader.reachedEnd)
|
||||
// Case when adding at the very end
|
||||
reader.current.toPos()
|
||||
else
|
||||
continue
|
||||
|
||||
activationSink.offer(pos, occChunk)
|
||||
trace.potentialMatch(occChunk.occ, candRule)
|
||||
// Drop the candidate if appropriate activation place is found.
|
||||
aIt.remove()
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
internal class PostponeMatchesStage(
|
||||
override val ispec: IncrementalSpec,
|
||||
journal: MatchJournal, // todo: remove this dependency, needed only for initial chunk
|
||||
private val journalIndex: MatchJournal.Index,
|
||||
private val ruleOrdering: RuleOrdering
|
||||
): IncrementalStage {
|
||||
|
||||
private var lastIncrementalRootPos: MatchJournal.Pos = journal.initialChunk().toPos()
|
||||
|
||||
private val postponedMatches: MutableMap<Int, MutableCollection<RuleMatchEx>> = hashMapOf()
|
||||
|
||||
|
||||
fun onNext(reader: MatchJournal.ChunkReader) =
|
||||
updateBasisPos(reader.current)
|
||||
|
||||
fun process(active: Occurrence, matches: List<RuleMatchEx>): List<RuleMatchEx> =
|
||||
postponeFutureMatches(matches).withPostponedMatches(active)
|
||||
|
||||
|
||||
private fun updateBasisPos(current: MatchJournal.Chunk) {
|
||||
assert(journalIndex.isKnown(current)) {
|
||||
"expected known position (from previous incremental session)"
|
||||
}
|
||||
lastIncrementalRootPos = current.toPos()
|
||||
}
|
||||
|
||||
/**
|
||||
* Adds previously postponed matches on [active] [Occurrence] if any, or returns original list.
|
||||
*/
|
||||
private fun List<RuleMatchEx>.withPostponedMatches(active: Occurrence): List<RuleMatchEx> =
|
||||
postponedMatches.remove(active.identity)?.let { postponed ->
|
||||
// Sort matches according to rule priorities
|
||||
(this + postponed).sortedWith(ruleOrdering.matchComparator)
|
||||
} ?: this
|
||||
|
||||
/**
|
||||
* Determines, filters out, and postpones future matches. Returns only current matches.
|
||||
*/
|
||||
private fun postponeFutureMatches(matches: List<RuleMatchEx>): List<RuleMatchEx> {
|
||||
val currentMatches = mutableListOf<RuleMatchEx>()
|
||||
for (m in matches) {
|
||||
|
||||
// Returns null for matches with occurrences only from this session
|
||||
// because journalIndex indexes only previous session.
|
||||
val occChunk = journalIndex.activationPos(m)
|
||||
val pos = occChunk?.toPos()
|
||||
|
||||
if (pos != null && isFuturePos(pos)) {
|
||||
postponedMatches.putIfAbsent(occChunk.identity, mutableListOf())!!.add(m)
|
||||
} else {
|
||||
currentMatches.add(m)
|
||||
}
|
||||
}
|
||||
return currentMatches
|
||||
}
|
||||
|
||||
private fun isFuturePos(pos: MatchJournal.Pos): Boolean =
|
||||
with(journalIndex) { pos after lastIncrementalRootPos }
|
||||
|
||||
private val MatchJournal.OccChunk.identity get() = this.occ.identity
|
||||
}
|
||||
|
||||
|
||||
internal class ContinueOccurrencesStage(
|
||||
override val ispec: IncrementalSpec,
|
||||
private val journalIndex: MatchJournal.Index
|
||||
): IncrementalStage, ContinuedActivationSink {
|
||||
|
||||
/**
|
||||
* Specifies position in [MatchJournal] for continuing program evaluation.
|
||||
* Position of [reactivated] in journal must precede execution position [continueFrom].
|
||||
* (see contract in [assertValid]).
|
||||
*/
|
||||
private data class ExecPos(val continueFrom: MatchJournal.Pos, val reactivated: MatchJournal.OccChunk) {
|
||||
val reactivatedOcc: Occurrence get() = reactivated.occ
|
||||
}
|
||||
|
||||
private infix fun MatchJournal.ChunkReader.isAt(execPos: ExecPos) =
|
||||
this.current == execPos.continueFrom.chunk
|
||||
|
||||
private fun ExecPos.assertValid() {
|
||||
val isAncestor = continueFrom.chunk.justifiedBy(reactivated)
|
||||
val isPredecessor = journalIndex.compare(continueFrom, reactivated.toPos()) >= 0
|
||||
// todo: ensure it must be so
|
||||
val fromPreviousSession = journalIndex.isKnown(reactivated)
|
||||
assert(fromPreviousSession && (isAncestor || isPredecessor))
|
||||
}
|
||||
|
||||
|
||||
private val queue: MutableList<ExecPos> = ArrayList<ExecPos>(1 + journalIndex.size / 8) // rough estimate
|
||||
|
||||
private val seen: MutableSet<ExecPos> = HashSet<ExecPos>()
|
||||
|
||||
|
||||
fun onNext(reader: MatchJournal.ChunkReader): Collection<Occurrence> {
|
||||
val continued = mutableListOf<Occurrence>()
|
||||
if (queue.isEmpty()) return continued
|
||||
|
||||
while (reader isAt queue.top()) {
|
||||
continued.add(queue.pop().reactivatedOcc)
|
||||
}
|
||||
return continued
|
||||
}
|
||||
|
||||
// todo: check sorting is correct
|
||||
override fun offerAll(continueFromPos: MatchJournal.Pos, occs: Iterable<Occurrence>) =
|
||||
occs.mapNotNull(journalIndex::activatingChunkOf)
|
||||
.sortedWith(journalIndex.chunkComparator)
|
||||
.forEach { offer(continueFromPos, it) }
|
||||
|
||||
override fun offer(continueFromPos: MatchJournal.Pos, ancestor: MatchJournal.OccChunk): Boolean =
|
||||
ExecPos(continueFromPos, ancestor).let {
|
||||
it.assertValid()
|
||||
if (seen.add(it)) queue.push(it) else false
|
||||
}
|
||||
|
||||
|
||||
private fun <E> MutableList<E>.top(): E = this.last()
|
||||
private fun <E> MutableList<E>.pop(): E = this.removeAt(this.size - 1)
|
||||
private fun <E> MutableList<E>.push(element: E): Boolean = this.add(element)
|
||||
}
|
||||
|
|
@ -21,6 +21,7 @@ import jetbrains.mps.logic.reactor.evaluation.*
|
|||
import jetbrains.mps.logic.reactor.program.*
|
||||
import jetbrains.mps.logic.reactor.util.Id
|
||||
import java.util.*
|
||||
import kotlin.Comparator
|
||||
|
||||
|
||||
interface MatchJournal : MutableIterable<MatchJournal.Chunk>, EvidenceSource {
|
||||
|
|
@ -118,11 +119,22 @@ interface MatchJournal : MutableIterable<MatchJournal.Chunk>, EvidenceSource {
|
|||
fun index(): Index
|
||||
|
||||
|
||||
interface ChunkReader {
|
||||
val current: Chunk get
|
||||
val previous: Chunk get
|
||||
val reachedEnd: Boolean get
|
||||
}
|
||||
|
||||
/**
|
||||
* Simplifies some search operations on [MatchJournal].
|
||||
* Also serves as a comparator for positions: later in journal means greater.
|
||||
*/
|
||||
interface Index : Comparator<Pos> {
|
||||
interface Index : ComparatorExt<Pos> {
|
||||
|
||||
val chunkComparator: Comparator<Chunk> get
|
||||
|
||||
fun isKnown(chunk: Chunk): Boolean
|
||||
|
||||
/**
|
||||
* Returns [Chunk] where provided principal occurrence was activated.
|
||||
* Returns null for non-principal occurrences.
|
||||
|
|
@ -141,10 +153,6 @@ interface MatchJournal : MutableIterable<MatchJournal.Chunk>, EvidenceSource {
|
|||
*/
|
||||
val size: Int
|
||||
|
||||
// some context functions
|
||||
infix fun Pos.before(other: Pos): Boolean = compare(this, other) <= 0
|
||||
|
||||
infix fun Pos.after(other: Pos): Boolean = compare(this, other) >= 0
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
|
|
@ -415,6 +415,8 @@ internal open class MatchJournalImpl(
|
|||
|
||||
override val size: Int = chunks.count()
|
||||
|
||||
override fun isKnown(chunk: Chunk): Boolean = chunkOrder.containsKey(Id(chunk))
|
||||
|
||||
override fun activatingChunkOf(occ: Occurrence) = occChunks[occ.identity]
|
||||
|
||||
override fun activationPos(match: RuleMatchEx): OccChunk? =
|
||||
|
|
@ -424,13 +426,19 @@ internal open class MatchJournalImpl(
|
|||
occSig?.let { activatingChunkOf(it.wrapped) }
|
||||
}.maxBy { orderOf(it)!! } // compare positions: find latest
|
||||
|
||||
// todo: throw for invalid positions?
|
||||
override val chunkComparator: Comparator<Chunk> get() = compareBy<Chunk>(this::orderOfThrow)
|
||||
|
||||
override fun compare(lhs: Pos, rhs: Pos): Int =
|
||||
compareBy<Pos>{ orderOf(it.chunk) }
|
||||
compareBy<Pos>{ orderOfThrow(it.chunk) }
|
||||
.thenComparingInt { it.entriesCount }
|
||||
.compare(lhs, rhs)
|
||||
|
||||
private fun orderOf(chunk: Chunk): Int? = chunkOrder[Id(chunk)]
|
||||
|
||||
private fun orderOfThrow(chunk: Chunk): Int = when (val ord = orderOf(chunk)) {
|
||||
null -> throw IllegalStateException("Chunk (${chunk.evidence}) must be known by journal index!")
|
||||
else -> ord
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -16,10 +16,13 @@
|
|||
|
||||
package jetbrains.mps.logic.reactor.core.internal
|
||||
|
||||
import jetbrains.mps.logic.reactor.core.Occurrence
|
||||
import jetbrains.mps.logic.reactor.core.canMatch
|
||||
import jetbrains.mps.logic.reactor.evaluation.RuleMatch
|
||||
import jetbrains.mps.logic.reactor.program.Rule
|
||||
|
||||
|
||||
internal class RuleOrdering(order: Iterable<Rule>): Comparator<Rule> {
|
||||
internal class RuleOrdering(order: Iterable<Rule>): ComparatorExt<Rule> {
|
||||
|
||||
private val ruleOrder: Map<Any, Int> = HashMap<Any, Int>().apply {
|
||||
put(MatchJournalImpl.InitRuleMatch.rule().uniqueTag(), -1) // less than anything
|
||||
|
|
@ -28,20 +31,37 @@ internal class RuleOrdering(order: Iterable<Rule>): Comparator<Rule> {
|
|||
|
||||
val ruleTags: Set<Any> = order.map { it.uniqueTag() }.toHashSet() // NB: without initial rule
|
||||
|
||||
fun orderOf(rule: Rule): Int? = ruleOrder[rule.uniqueTag()]
|
||||
override fun compare(lhs: Rule, rhs: Rule): Int = compareBy<Rule>(this::orderOfThrow).compare(lhs, rhs)
|
||||
|
||||
override fun compare(lhs: Rule, rhs: Rule): Int {
|
||||
val lhsRuleOrder = orderOf(lhs)
|
||||
val rhsRuleOrder = orderOf(rhs)
|
||||
val matchComparator: Comparator<RuleMatch> get() = compareBy<RuleMatch>{ orderOfThrow(it.rule()) }
|
||||
|
||||
if (lhsRuleOrder == null || rhsRuleOrder == null) {
|
||||
throw(IllegalStateException("Rules compared must be present in the rule index!"))
|
||||
}
|
||||
return lhsRuleOrder.compareTo(rhsRuleOrder)
|
||||
|
||||
private fun orderOf(rule: Rule): Int? = ruleOrder[rule.uniqueTag()]
|
||||
|
||||
private fun orderOfThrow(rule: Rule): Int = when (val res = orderOf(rule)) {
|
||||
null -> throw IllegalStateException("Compared rule ($rule) must be present in rule index!")
|
||||
else -> res
|
||||
}
|
||||
}
|
||||
|
||||
fun isEarlierThan(lhs: Rule, rhs: Rule): Boolean = compare(lhs, rhs) < 0
|
||||
|
||||
fun isLaterThan(lhs: Rule, rhs: Rule): Boolean = compare(lhs, rhs) > 0
|
||||
/**
|
||||
* Checks whether [candidateRule] can be inserted in journal as a child
|
||||
* of [parentChunk] before one of its child chunks, [beforeChunk].
|
||||
* It is assumed that [candidateRule] can be matched by [Occurrence] from [parentChunk].
|
||||
*/
|
||||
internal fun RuleOrdering.canBeInserted(candidateRule: Rule, parentChunk: MatchJournal.OccChunk, beforeChunk: MatchJournal.Chunk): Boolean {
|
||||
// Place to try activating candidate rule is:
|
||||
// either according to the ordering between rules
|
||||
// or as the last one, after all existing activations
|
||||
assert(candidateRule.canMatch(parentChunk.occ.constraint))
|
||||
|
||||
}
|
||||
val placeToInsertFound = beforeChunk is MatchJournal.MatchChunk
|
||||
&& candidateRule before beforeChunk.match.rule()
|
||||
|
||||
val isDescendant = beforeChunk.justifiedBy(parentChunk)
|
||||
val isSibling = beforeChunk.evidence == parentChunk.evidence && beforeChunk != parentChunk
|
||||
val childChunksEnded = !isDescendant || isSibling
|
||||
|
||||
return (childChunksEnded || placeToInsertFound)
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue