From 2dba8890f3842e563ec327f54cb00a7dd4187d58 Mon Sep 17 00:00:00 2001 From: Grigorii Kirgizov Date: Wed, 10 Jul 2019 13:36:54 +0300 Subject: [PATCH] Refactor ProcessingStateImpl: move logic related to execution queue to its own class --- .../reactor/core/internal/ExecutionQueue.kt | 121 ++++++++++++++++++ .../reactor/core/internal/MatchJournal.kt | 5 + .../reactor/core/internal/MatchJournalImpl.kt | 2 + .../core/internal/ProcessingStateImpl.kt | 114 ++++------------- 4 files changed, 150 insertions(+), 92 deletions(-) create mode 100644 reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ExecutionQueue.kt diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ExecutionQueue.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ExecutionQueue.kt new file mode 100644 index 00000000..873cd799 --- /dev/null +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ExecutionQueue.kt @@ -0,0 +1,121 @@ +/* + * Copyright 2014-2019 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.Controller +import jetbrains.mps.logic.reactor.core.Occurrence +import jetbrains.mps.logic.reactor.core.RuleMatchEx +import jetbrains.mps.logic.reactor.util.Id +import java.util.* + +internal class ExecutionQueue( + private val journalIndex: MatchJournal.Index, + private val ruleOrdering: RuleOrdering +) { + + private data class ExecPos(val pos: MatchJournal.Pos, val activeOcc: Occurrence) + + // It is a position in Journal from previous session, + // from which incremental execution continues. + // Needed for pos comparison in postponeFutureMatches. + private lateinit var lastIncrementalRootPos: MatchJournal.Pos + + private val postponedMatches: MutableMap, List> = HashMap() + + private val execQueue: Queue = + PriorityQueue(1 + journalIndex.size / 2) { // just an estimate + lhs, rhs -> journalIndex.compare(lhs.pos, rhs.pos) + } + + + fun run(controller: Controller, state: ProcessingStateImpl): FeedbackStatus.NORMAL { + if (execQueue.isNotEmpty()) { + state.resetStore() + + var prevPos: MatchJournal.Pos? = null + do { + val execPos = execQueue.poll() + + // Handles the case when several matches are added to the same position. + // Then shouldn't replay, because currentPos is valid and more recent (!) than execPos. + if (execPos.pos != prevPos) { + state.replay(controller, execPos.pos) + lastIncrementalRootPos = execPos.pos + } + prevPos = execPos.pos + + state.reactivate(controller, execPos.activeOcc) + } while (execQueue.isNotEmpty()) + } + // Also replay to the end after queue is fully executed + state.replay(controller, state.last().toPos()) + // fixme: get FeedbackStatus out of reactivate() + return FeedbackStatus.NORMAL() + } + + fun withPostponedMatches(active: Occurrence, matches: List): List = + postponedMatches.remove(Id(active))?.let { postponed -> + // Sort matches according to rule priorities + (matches + postponed).sortedBy { ruleOrdering.orderOf(it.rule()) } + } ?: matches + + /** + * Determines, filters out and enqueues future matches. + * Returns only current matches. + */ + fun postponeFutureMatches(matches: List): List { + val currentMatches = mutableListOf() + for (m in matches) { + + // Returns null for matches with occurrences only from this session + // because journalIndex indexes only previous session. + val pos = journalIndex.activationPos(m) + + // if it is a future match + if (pos != null && journalIndex.compare(lastIncrementalRootPos, pos) < 0) { + val idOcc = Id(pos.occ) + postponedMatches[idOcc] = (postponedMatches[idOcc] ?: emptyList()) + listOf(m) + offer(pos) + } else { + currentMatches.add(m) + } + } + return currentMatches + } + + + // todo: do need to reactivate only the main, matching~activating match? + // (i.e. don't reactivate additional, inactive heads that only completed the match?) + fun offerAll(occs: Iterable): Boolean = + execQueue.addAll(occs.mapNotNull { occ -> + journalIndex.activatingChunkOf(Id(occ))?.let { chunk -> + ExecPos(chunk.toPos(), occ) + } + }) + + fun offer(posInJournal: MatchJournal.Pos, activeOcc: Occurrence): Boolean = + execQueue.offer(ExecPos(posInJournal, activeOcc)) + + // special case when position in trace corresponds to position of activated occurrence + fun offer(posInJournal: MatchJournal.Pos): Boolean = + execQueue.offer(ExecPos(posInJournal, posInJournal.occ)) + + fun isEmpty(): Boolean = execQueue.isEmpty() + + fun isNotEmpty(): Boolean = execQueue.isNotEmpty() + +} \ No newline at end of file diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournal.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournal.kt index 1b7cc0ea..f06f7f7f 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournal.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournal.kt @@ -108,6 +108,11 @@ interface MatchJournal : MutableIterable { match.signature().mapNotNull { occSig -> occSig?.let { activatingChunkOf(it)?.toPos() } }.maxWith(this) // compare positions: find latest + + /** + * Length of the indexed [MatchJournal] + */ + val size: Int } /** diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournalImpl.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournalImpl.kt index 3cb34ed6..f97a788e 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournalImpl.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/MatchJournalImpl.kt @@ -181,6 +181,8 @@ internal open class MatchJournalImpl( } } + override val size: Int = chunks.count() + override fun activatingChunkOf(occId: Id) = occChunks[occId] // todo: throw for invalid positions? diff --git a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStateImpl.kt b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStateImpl.kt index eba90569..cfdb125c 100644 --- a/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStateImpl.kt +++ b/reactor/Core/src/jetbrains/mps/logic/reactor/core/internal/ProcessingStateImpl.kt @@ -54,23 +54,24 @@ internal class ProcessingStateImpl(private var dispatchingFront: Dispatcher.Disp private val journalIndex: MatchJournal.Index = journal.index() - // It is a position in Journal from previous session, - // from which incremental execution continues. - // Needed for pos comparison in postponeFutureMatches. - private lateinit var lastIncrementalRootPos: MatchJournal.Pos + private val execQueue: ExecutionQueue = ExecutionQueue(journalIndex, ruleOrdering) - private val postponedMatches: MutableMap, List> = HashMap() - - private val execQueue: Queue = - PriorityQueue(1 + this.count() / 2) { // just an estimate - lhs, rhs -> journalIndex.compare(lhs.pos, rhs.pos) - } - - private data class ExecPos(val pos: MatchJournal.Pos, val activeOcc: Occurrence) private data class MatchCandidate(val rule: Rule, val occChunk: MatchJournal.OccChunk) + fun reactivate(controller: Controller, activeOcc: Occurrence) { + // If the occurrence is still in the store after replay (i.e. if it's valid to activate it) + if (activeOcc.stored) { + // Forget that occ was seen. + // Incremental reactivation isn't like the usual reactivation, + // it should proceed more like usual activation. + this.dispatchingFront = dispatchingFront.forgetSeen(activeOcc) + trace.reactivateIncremental(activeOcc) + controller.reactivate(activeOcc) + } + } + fun invalidateDependentRules(ruleIds: Set) { val it = this.iterator() while (it.hasNext()) { @@ -129,18 +130,12 @@ internal class ProcessingStateImpl(private var dispatchingFront: Dispatcher.Disp // We removed the match, so need to reactivate all still valid occurrences from the head // by definition of Chunk and principal rule, all occurrences from the head are principal val matchedOccs = chunk.match.allHeads() as Iterable - val (invalidatedOccs, validOccs) = matchedOccs.partition { occ -> - occ.justifications().intersects(justificationRoots) + val validOccs = matchedOccs.filter { occ -> + !occ.justifications().intersects(justificationRoots) } assert(matchedOccs.all { it.isPrincipal() }) - // todo: do need to reactivate only the main, matching~activating match? - // (i.e. don't reactivate additional, inactive heads that only completed the match?) - execQueue.addAll(validOccs.mapNotNull { occ -> - journalIndex.activatingChunkOf(Id(occ))?.let { chunk -> - ExecPos(chunk.toPos(), occ) - } - }) + execQueue.offerAll(validOccs) } } @@ -196,7 +191,7 @@ internal class ProcessingStateImpl(private var dispatchingFront: Dispatcher.Disp else if (!it.hasNext()) chunk.toPos() else continue - execQueue.offer(ExecPos(pos, occChunk.occ)) + execQueue.offer(pos, occChunk.occ) trace.potentialMatch(occChunk.occ, candRule) // Drop the candidate if appropriate activation place is found. aIt.remove() @@ -208,38 +203,8 @@ internal class ProcessingStateImpl(private var dispatchingFront: Dispatcher.Disp } } - fun launchQueue(controller: Controller): FeedbackStatus.NORMAL { - if (execQueue.isNotEmpty()) { - resetStore() - - var prevPos: MatchJournal.Pos? = null - do { - val execPos = execQueue.poll() - - // Handles the case when several matches are added to the same position. - // Then shouldn't replay, because currentPos is valid and more recent (!) than execPos. - if (execPos.pos != prevPos) { - replay(controller, execPos.pos) - lastIncrementalRootPos = execPos.pos - } - prevPos = execPos.pos - - // If the occurrence is still in the store after replay (i.e. if it's valid to activate it) - if (execPos.activeOcc.stored) { - // Forget that occ was seen. - // Incremental reactivation isn't like the usual reactivation, - // it should proceed more like usual activation. - this.dispatchingFront = dispatchingFront.forgetSeen(execPos.activeOcc) - trace.reactivateIncremental(execPos.activeOcc) - controller.reactivate(execPos.activeOcc) - } - } while (execQueue.isNotEmpty()) - } - // Also replay to the end after queue is fully executed - replay(controller, this.last().toPos()) - // fixme: get FeedbackStatus out of reactivate() - return FeedbackStatus.NORMAL() - } + fun launchQueue(controller: Controller): FeedbackStatus.NORMAL = + execQueue.run(controller, this) fun snapshot(): SessionToken = SessionTokenImpl(view(), ruleOrdering.ruleTags, dispatchingFront.state()) @@ -274,9 +239,10 @@ internal class ProcessingStateImpl(private var dispatchingFront: Dispatcher.Disp if (isFront() || !active.isPrincipal()) { matches } else { - postponeFutureMatches(matches) + assert( matches.all { ispec.isPrincipal(it.rule()) } ) + execQueue.postponeFutureMatches(matches) } - val currentMatches = withPostponedMatches(active, newCurrentMatches) + val currentMatches = execQueue.withPostponedMatches(active, newCurrentMatches) val outStatus = currentMatches.fold(inStatus) { status, match -> // TODO: paranoid check. should be isAlive() instead @@ -296,42 +262,6 @@ internal class ProcessingStateImpl(private var dispatchingFront: Dispatcher.Disp } } - private fun withPostponedMatches(active: Occurrence, matches: List): List = - postponedMatches.remove(Id(active))?.let { postponed -> - // Sort matches according to rule priorities - (matches + postponed).sortedBy { ruleOrdering.orderOf(it.rule()) } - } ?: matches - - /** - * Determines, filters out and enqueues (to execution queue) future matches. - * Returns only current matches. - */ - private fun postponeFutureMatches(matches: List): List { - assert( - matches.all { ispec.isPrincipal(it.rule()) }, - { "non-principal ctrs in head of principal rule: ${ matches.filter { !ispec.isPrincipal(it.rule()) } }" } - ) - - val currentMatches = mutableListOf() - for (m in matches) { - - // Returns null for matches with occurrences only from this session - // because journalIndex indexes only previous session. - val pos = journalIndex.activationPos(m) - - // if it is a future match - if (pos != null && journalIndex.compare(lastIncrementalRootPos, pos) < 0) { - val idOcc = Id(pos.occ) - postponedMatches[idOcc] = (postponedMatches[idOcc] ?: emptyList()) + listOf(m) - execQueue.offer(ExecPos(pos, pos.occ)) - } else { - currentMatches.add(m) - } - } - return currentMatches - } - - private inline fun FeedbackStatus.then(action: (FeedbackStatus) -> FeedbackStatus) : FeedbackStatus = if (operational) action(this) else this