You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kie.apache.org by pi...@apache.org on 2025/10/29 20:04:51 UTC
(incubator-kie-drools) branch main updated: [incubator-kie-issues #203] Cleanup Phreak code - Make RuleNetworkEvaluator per-session (#6499)
This is an automated email from the ASF dual-hosted git repository.
pibizza pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-kie-drools.git
The following commit(s) were added to refs/heads/main by this push:
new 8fe266298b [incubator-kie-issues #203] Cleanup Phreak code - Make RuleNetworkEvaluator per-session (#6499)
8fe266298b is described below
commit 8fe266298b17599f620a7680de7d7c8d663bd7d7
Author: Paolo Bizzarri <pi...@gmail.com>
AuthorDate: Wed Oct 29 21:04:46 2025 +0100
[incubator-kie-issues #203] Cleanup Phreak code - Make RuleNetworkEvaluator per-session (#6499)
* Making RNE per session
* Moved ForcedFlushLeftTuple inside RNE
* Added PhreakNodeOperations
* Moving stuff from SegmentPropagator and other utils class
* Minor fix
* Used static access for various methods
* Consolidated code into RuleNetworkEvaluator
* Minor cleanups
* Fixed licence files
---
.../java/org/drools/core/common/ReteEvaluator.java | 3 +
.../java/org/drools/core/common/TupleSets.java | 3 +
.../java/org/drools/core/common/TupleSetsImpl.java | 18 +
.../org/drools/core/phreak/LazyPhreakBuilder.java | 13 +-
.../drools/core/phreak/PhreakAccumulateNode.java | 126 ++-
.../drools/core/phreak/PhreakAsyncReceiveNode.java | 14 +-
.../drools/core/phreak/PhreakAsyncSendNode.java | 29 +-
.../org/drools/core/phreak/PhreakBranchNode.java | 24 +-
.../org/drools/core/phreak/PhreakEvalNode.java | 37 +-
.../org/drools/core/phreak/PhreakExistsNode.java | 52 +-
.../org/drools/core/phreak/PhreakFromNode.java | 29 +-
.../org/drools/core/phreak/PhreakGroupByNode.java | 13 +-
.../org/drools/core/phreak/PhreakJoinNode.java | 45 +-
.../core/phreak/PhreakNetworkNodeFactory.java | 31 +-
.../core/phreak/PhreakNetworkNodeFactoryImpl.java | 62 +-
.../drools/core/phreak/PhreakNodeOperations.java | 265 +++++
.../java/org/drools/core/phreak/PhreakNotNode.java | 55 +-
.../org/drools/core/phreak/PhreakQueryNode.java | 22 +-
.../core/phreak/PhreakQueryTerminalNode.java | 27 +-
.../drools/core/phreak/PhreakReactiveFromNode.java | 12 +-
.../drools/core/phreak/PhreakRuleTerminalNode.java | 9 +-
.../core/phreak/PhreakSubnetworkNotExistsNode.java | 20 +-
.../org/drools/core/phreak/PhreakTimerNode.java | 68 +-
.../org/drools/core/phreak/ReactiveObjectUtil.java | 3 +-
.../java/org/drools/core/phreak/RuleExecutor.java | 2 +-
.../drools/core/phreak/RuleNetworkEvaluator.java | 1022 +-------------------
...valuator.java => RuleNetworkEvaluatorImpl.java} | 599 ++++++------
.../org/drools/core/phreak/SegmentPropagator.java | 161 ---
.../drools/core/phreak/TupleEvaluationUtil.java | 144 ---
.../drools/core/reteoo/LeftInputAdapterNode.java | 22 +-
.../main/java/org/drools/core/reteoo/NotRight.java | 9 +-
.../drools/core/reteoo/RightInputAdapterNode.java | 16 +-
.../StatefulKnowledgeSessionForRHS.java | 6 +
.../session/StatefulKnowledgeSessionImpl.java | 10 +
.../phreak/MetricPhreakNetworkNodeFactoryImpl.java | 89 +-
.../metric/phreak/PhreakAccumulateNodeMetric.java | 10 +-
.../phreak/PhreakAsyncReceiveNodeMetric.java | 10 +-
.../metric/phreak/PhreakAsyncSendNodeMetric.java | 10 +-
.../metric/phreak/PhreakBranchNodeMetric.java | 6 +-
.../drools/metric/phreak/PhreakEvalNodeMetric.java | 10 +-
.../metric/phreak/PhreakExistsNodeMetric.java | 9 +-
.../drools/metric/phreak/PhreakFromNodeMetric.java | 9 +-
.../metric/phreak/PhreakGroupByNodeMetric.java | 10 +-
.../drools/metric/phreak/PhreakJoinNodeMetric.java | 9 +-
.../drools/metric/phreak/PhreakNotNodeMetric.java | 9 +-
.../metric/phreak/PhreakQueryNodeMetric.java | 8 +-
.../phreak/PhreakQueryTerminalNodeMetric.java | 12 +-
.../phreak/PhreakReactiveFromNodeMetric.java | 10 +-
.../metric/phreak/PhreakTimerNodeMetric.java | 5 +
.../impl/sessions/RuleUnitExecutorImpl.java | 12 +-
.../mvel/integrationtests/phreak/Scenario.java | 15 +-
51 files changed, 1171 insertions(+), 2043 deletions(-)
diff --git a/drools-core/src/main/java/org/drools/core/common/ReteEvaluator.java b/drools-core/src/main/java/org/drools/core/common/ReteEvaluator.java
index cbd066613e..30a5803ae9 100644
--- a/drools-core/src/main/java/org/drools/core/common/ReteEvaluator.java
+++ b/drools-core/src/main/java/org/drools/core/common/ReteEvaluator.java
@@ -32,6 +32,7 @@ import org.drools.core.event.RuleEventListenerSupport;
import org.drools.core.event.RuleRuntimeEventSupport;
import org.drools.core.impl.InternalRuleBase;
import org.drools.core.phreak.PropagationEntry;
+import org.drools.core.phreak.RuleNetworkEvaluator;
import org.drools.core.reteoo.ObjectTypeConf;
import org.drools.core.reteoo.RuntimeComponentFactory;
import org.drools.core.rule.accessor.FactHandleFactory;
@@ -51,6 +52,8 @@ public interface ReteEvaluator extends ValueResolver {
long getIdentifier();
+ RuleNetworkEvaluator getRuleNetworkEvaluator();
+
ActivationsManager getActivationsManager();
InternalRuleBase getKnowledgeBase();
diff --git a/drools-core/src/main/java/org/drools/core/common/TupleSets.java b/drools-core/src/main/java/org/drools/core/common/TupleSets.java
index fd3e25a13b..2634f3a522 100644
--- a/drools-core/src/main/java/org/drools/core/common/TupleSets.java
+++ b/drools-core/src/main/java/org/drools/core/common/TupleSets.java
@@ -18,6 +18,7 @@
*/
package org.drools.core.common;
+import org.drools.core.reteoo.Tuple;
import org.drools.core.reteoo.TupleImpl;
public interface TupleSets {
@@ -55,4 +56,6 @@ public interface TupleSets {
TupleImpl getNormalizedDeleteFirst();
boolean addNormalizedDelete(TupleImpl leftTuple);
+
+
}
diff --git a/drools-core/src/main/java/org/drools/core/common/TupleSetsImpl.java b/drools-core/src/main/java/org/drools/core/common/TupleSetsImpl.java
index f625f02652..21c8352352 100644
--- a/drools-core/src/main/java/org/drools/core/common/TupleSetsImpl.java
+++ b/drools-core/src/main/java/org/drools/core/common/TupleSetsImpl.java
@@ -23,6 +23,24 @@ import org.drools.core.reteoo.Tuple;
public class TupleSetsImpl implements TupleSets {
+ public static TupleSets createLeftTupleTupleSets(TupleImpl leftTuple, short stagedType) {
+ TupleSets leftTupleSets = new TupleSetsImpl();
+ if (leftTuple != null) {
+ switch (stagedType) {
+ case Tuple.INSERT:
+ leftTupleSets.addInsert(leftTuple);
+ break;
+ case Tuple.DELETE:
+ leftTupleSets.addDelete(leftTuple);
+ break;
+ case Tuple.UPDATE:
+ leftTupleSets.addUpdate(leftTuple);
+ break;
+ }
+ }
+ return leftTupleSets;
+ }
+
private TupleImpl insertFirst;
private TupleImpl deleteFirst;
private TupleImpl updateFirst;
diff --git a/drools-core/src/main/java/org/drools/core/phreak/LazyPhreakBuilder.java b/drools-core/src/main/java/org/drools/core/phreak/LazyPhreakBuilder.java
index 1c2098dbad..0aab6e6516 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/LazyPhreakBuilder.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/LazyPhreakBuilder.java
@@ -81,7 +81,6 @@ import org.slf4j.LoggerFactory;
import static org.drools.core.phreak.BuildtimeSegmentUtilities.isRootNode;
import static org.drools.core.phreak.EagerPhreakBuilder.deleteLeftTuple;
import static org.drools.core.phreak.EagerPhreakBuilder.Add.attachAdapterAndPropagate;
-import static org.drools.core.phreak.TupleEvaluationUtil.forceFlushLeftTuple;
class LazyPhreakBuilder implements PhreakBuilder {
@@ -614,7 +613,7 @@ class LazyPhreakBuilder implements PhreakBuilder {
List<LeftTupleNode> splits) {
// first flush the subject rule, then flush any staging lists that are part of a merge
if (pmem.isInitialized()) {
- RuleNetworkEvaluator.INSTANCE.evaluateNetwork(pmem.getRuleAgendaItem().getRuleExecutor(), pmem);
+ wm.getRuleNetworkEvaluator().evaluateNetwork(pmem.getRuleAgendaItem().getRuleExecutor(), pmem);
}
// With the removing rules being flushed, we need to check any splits that will be merged, to see if they need flushing
@@ -634,8 +633,8 @@ class LazyPhreakBuilder implements PhreakBuilder {
if (!childSmem.getStagedLeftTuples().isEmpty()) {
PathMemory childPmem = childSmem.getPathMemories().get(0);
flushed.add(new Flushed(childPmem, childSmem));
- forceFlushLeftTuple(wm, childPmem, childSmem, childSmem.getStagedLeftTuples()
- .takeAll());
+ wm.getRuleNetworkEvaluator().forceFlushLeftTuple(childPmem, childSmem, childSmem.getStagedLeftTuples()
+ .takeAll());
}
}
}
@@ -649,8 +648,8 @@ class LazyPhreakBuilder implements PhreakBuilder {
for (Flushed path : flushed) {
if (!path.segmentMemory.getStagedLeftTuples().isEmpty()) {
flushCount++;
- forceFlushLeftTuple(wm, pmem, path.segmentMemory, path.segmentMemory.getStagedLeftTuples()
- .takeAll());
+ wm.getRuleNetworkEvaluator().forceFlushLeftTuple(pmem, path.segmentMemory, path.segmentMemory.getStagedLeftTuples()
+ .takeAll());
}
}
}
@@ -682,7 +681,7 @@ class LazyPhreakBuilder implements PhreakBuilder {
if (smemIndex < length) {
// it only found a SM that needed flushing, if smemIndex < length
- forceFlushLeftTuple(wm, pmem, sm, sm.getStagedLeftTuples().takeAll());
+ wm.getRuleNetworkEvaluator().forceFlushLeftTuple(pmem, sm, sm.getStagedLeftTuples().takeAll());
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakAccumulateNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakAccumulateNode.java
index b1b58e8f5d..0defc2d5e1 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakAccumulateNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakAccumulateNode.java
@@ -44,12 +44,20 @@ import org.drools.core.util.AbstractHashTable;
import org.drools.core.util.FastIterator;
import org.kie.api.runtime.rule.FactHandle;
-import static org.drools.core.phreak.RuleNetworkEvaluator.normalizeStagedTuples;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderLeftMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderRightMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.normalizeStagedTuples;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
public class PhreakAccumulateNode {
- public void doNode(ReteEvaluator reteEvaluator,
- AccumulateNode accNode,
+ protected ReteEvaluator reteEvaluator;
+
+ public PhreakAccumulateNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(AccumulateNode accNode,
LeftTupleSink sink,
AccumulateMemory am,
TupleSets srcLeftTuples,
@@ -63,7 +71,7 @@ public class PhreakAccumulateNode {
// order of left and right operations is to minimise wasted of innefficient joins.
if (srcLeftTuples.getDeleteFirst() != null) {
// use the real target here, as dealing direct with left tuples
- doLeftDeletes(accNode, am, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftDeletes(accNode, am, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
// We need to collect which leftTuple where updated, so that we can
@@ -71,38 +79,38 @@ public class PhreakAccumulateNode {
TupleSets tempLeftTuples = new TupleSetsImpl();
if (srcRightTuples.getDeleteFirst() != null) {
- doRightDeletes(accNode, am, reteEvaluator, srcRightTuples, tempLeftTuples);
+ doRightDeletes(accNode, am, srcRightTuples, tempLeftTuples);
}
if (srcRightTuples.getUpdateFirst() != null) {
- RuleNetworkEvaluator.doUpdatesReorderRightMemory(bm, srcRightTuples);
- doRightUpdates(accNode, am, reteEvaluator, srcRightTuples, tempLeftTuples);
+ doUpdatesReorderRightMemory(bm, srcRightTuples);
+ doRightUpdates(accNode, am, srcRightTuples, tempLeftTuples);
}
if (srcLeftTuples.getUpdateFirst() != null ) {
- RuleNetworkEvaluator.doUpdatesReorderLeftMemory(bm, srcLeftTuples);
- doLeftUpdates(accNode, am, reteEvaluator, srcLeftTuples, tempLeftTuples);
+ doUpdatesReorderLeftMemory(bm, srcLeftTuples);
+ doLeftUpdates(accNode, am, srcLeftTuples, tempLeftTuples);
}
if (!accNode.getRightInput().inputIsTupleToObjectNode()) {
// Non subnetworks ore process right then left. This because it's typically faster to ensure all RightTuples
// are in place then you can iterate with the left evaluation cached.
if (srcRightTuples.getInsertFirst() != null) {
- doRightInserts(accNode, am, reteEvaluator, srcRightTuples, tempLeftTuples);
+ doRightInserts(accNode, am, srcRightTuples, tempLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(accNode, am, reteEvaluator, srcLeftTuples, tempLeftTuples);
+ doLeftInserts(accNode, am, srcLeftTuples, tempLeftTuples);
}
} else {
// subnetworks process left then right. It ensures the LTM is not empty and the acctx is initialised.
// It then returns and all the matching can safely be done by the RightTuple
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(accNode, am, reteEvaluator, srcLeftTuples, tempLeftTuples);
+ doLeftInserts(accNode, am, srcLeftTuples, tempLeftTuples);
}
if (srcRightTuples.getInsertFirst() != null) {
- doRightInserts(accNode, am, reteEvaluator, srcRightTuples, tempLeftTuples);
+ doRightInserts(accNode, am, srcRightTuples, tempLeftTuples);
}
}
@@ -112,8 +120,8 @@ public class PhreakAccumulateNode {
for (TupleImpl leftTuple = tempLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
evaluateResultConstraints(accNode, sink, accumulate, leftTuple, leftTuple.getPropagationContext(),
- reteEvaluator, am, (BaseAccumulation) leftTuple.getContextObject(),
- trgLeftTuples, stagedLeftTuples);
+ am, (BaseAccumulation) leftTuple.getContextObject(), trgLeftTuples,
+ stagedLeftTuples);
leftTuple.clearStaged();
leftTuple = next;
}
@@ -121,8 +129,8 @@ public class PhreakAccumulateNode {
for (TupleImpl leftTuple = tempLeftTuples.getUpdateFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
evaluateResultConstraints( accNode, sink, accumulate, leftTuple, leftTuple.getPropagationContext(),
- reteEvaluator, am, (BaseAccumulation) leftTuple.getContextObject(),
- trgLeftTuples, stagedLeftTuples );
+ am, (BaseAccumulation) leftTuple.getContextObject(), trgLeftTuples,
+ stagedLeftTuples );
leftTuple.clearStaged();
leftTuple = next;
}
@@ -134,7 +142,6 @@ public class PhreakAccumulateNode {
private void doLeftInserts(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
@@ -150,13 +157,13 @@ public class PhreakAccumulateNode {
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
- boolean useLeftMemory = leftTupleMemoryEnabled || RuleNetworkEvaluator.useLeftMemory(accNode, leftTuple);
+ boolean useLeftMemory = leftTupleMemoryEnabled || useLeftMemory(accNode, leftTuple);
if (useLeftMemory) {
ltm.add(leftTuple);
}
- BaseAccumulation accresult = initAccumulationContext( am, reteEvaluator, accumulate, leftTuple );
+ BaseAccumulation accresult = initAccumulationContext( am, accumulate, leftTuple );
if (accNode.getRightInput().inputIsTupleToObjectNode()) {
// This is a subnetwork, do not process further. As all matches will processed
// by the right insert. This is to avoid double iteration (first right side iteration
@@ -182,8 +189,8 @@ public class PhreakAccumulateNode {
rightTuple.getFactHandleForEvaluation())) {
// add a match
addMatch(accNode, accumulate, leftTuple, rightTuple,
- null, null, reteEvaluator, am,
- accresult, useLeftMemory, true);
+ null, null, am, accresult,
+ useLeftMemory, true);
}
rightTuple = nextRightTuple;
@@ -199,7 +206,7 @@ public class PhreakAccumulateNode {
constraints.resetTuple( contextEntry );
}
- BaseAccumulation initAccumulationContext(AccumulateMemory am, ReteEvaluator reteEvaluator, Accumulate accumulate, TupleImpl leftTuple) {
+ BaseAccumulation initAccumulationContext(AccumulateMemory am, Accumulate accumulate, TupleImpl leftTuple) {
AccumulateContext accContext = new AccumulateContext();
leftTuple.setContextObject(accContext);
@@ -216,7 +223,6 @@ public class PhreakAccumulateNode {
private void doRightInserts(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples) {
Accumulate accumulate = accNode.getAccumulate();
@@ -235,7 +241,7 @@ public class PhreakAccumulateNode {
for (TupleImpl rightTuple = srcRightTuples.getInsertFirst(); rightTuple != null; ) {
TupleImpl next = rightTuple.getStagedNext();
- boolean useTupleMemory = tupleMemoryEnabled || RuleNetworkEvaluator.useLeftMemory(accNode, rightTuple);
+ boolean useTupleMemory = tupleMemoryEnabled || useLeftMemory(accNode, rightTuple);
if (useTupleMemory || !accNode.getRightInput().inputIsTupleToObjectNode()) {
// If tuple memory is off, it will still be when it is not a subnetwork.
@@ -253,8 +259,8 @@ public class PhreakAccumulateNode {
if ( constraints.isAllowedCachedRight(leftTuple, contextEntry) ) {
final BaseAccumulation accctx = (BaseAccumulation) leftTuple.getContextObject();
addMatch( accNode, accumulate, leftTuple, rightTuple,
- null, null, reteEvaluator, am,
- accctx, true, false );
+ null, null, am, accctx,
+ true, false );
// right inserts and updates are done first
// so any existing leftTuples we know are updates, but only add if not already added
@@ -272,7 +278,6 @@ public class PhreakAccumulateNode {
private void doLeftUpdates(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
BetaMemory bm = am.getBetaMemory();
@@ -309,7 +314,7 @@ public class PhreakAccumulateNode {
// if rightTuple is null, we assume there was a bucket change and that bucket is empty
if (childLeftTuple != null && rtm.isIndexed() && !rightIt.isFullIterator() && (rightTuple == null || (rightTuple.getMemory() != childLeftTuple.getRightParent().getMemory()))) {
// our index has changed, so delete all the previous matchings
- removePreviousMatchesForLeftTuple(accumulate, leftTuple, reteEvaluator, am, accctx, true);
+ removePreviousMatchesForLeftTuple(accumulate, leftTuple, am, accctx, true);
childLeftTuple = null; // null so the next check will attempt matches for new bucket
}
@@ -318,7 +323,6 @@ public class PhreakAccumulateNode {
if (rightTuple != null) {
doLeftUpdatesProcessChildren(accNode,
am,
- reteEvaluator,
bm,
accumulate,
constraints,
@@ -339,7 +343,6 @@ public class PhreakAccumulateNode {
private void doLeftUpdatesProcessChildren(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
BetaMemory bm,
Accumulate accumulate,
BetaConstraints constraints,
@@ -356,8 +359,8 @@ public class PhreakAccumulateNode {
rightTuple.getFactHandleForEvaluation())) {
// add a new match
addMatch(accNode, accumulate, leftTuple, rightTuple,
- null, null, reteEvaluator, am,
- accctx, true, true);
+ null, null, am, accctx,
+ true, true);
}
}
} else {
@@ -369,8 +372,8 @@ public class PhreakAccumulateNode {
if (match == null || match.getRightParent() != rightTuple) {
// add a new match
addMatch(accNode, accumulate, leftTuple, rightTuple,
- match, null, reteEvaluator, am,
- accctx, true, true);
+ match, null, am, accctx,
+ true, true);
} else {
// we must re-add this to ensure deterministic iteration
TupleImpl temp = match.getHandleNext();
@@ -382,7 +385,7 @@ public class PhreakAccumulateNode {
TupleImpl temp = match.getHandleNext();
// remove the match
boolean reversed = removeMatch(accNode, accumulate, rightTuple, match,
- reteEvaluator, am, accctx, false);
+ am, accctx, false);
match = temp;
// the next line means that when a match is removed from the current leftTuple
// and the accumulate does not support the reverse operation, then the whole
@@ -398,7 +401,6 @@ public class PhreakAccumulateNode {
leftTuple,
null,
null,
- reteEvaluator,
am,
accctx,
true);
@@ -408,7 +410,6 @@ public class PhreakAccumulateNode {
private void doRightUpdates(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples) {
BetaMemory bm = am.getBetaMemory();
@@ -437,7 +438,6 @@ public class PhreakAccumulateNode {
removePreviousMatchesForRightTuple( accNode,
accumulate,
rightTuple,
- reteEvaluator,
am,
childLeftTuple,
trgLeftTuples );
@@ -448,7 +448,6 @@ public class PhreakAccumulateNode {
if ( leftTuple != null ) {
doRightUpdatesProcessChildren( accNode,
am,
- reteEvaluator,
bm,
constraints,
accumulate,
@@ -467,7 +466,6 @@ public class PhreakAccumulateNode {
private void doRightUpdatesProcessChildren(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
BetaMemory bm,
BetaConstraints constraints,
Accumulate accumulate,
@@ -488,8 +486,8 @@ public class PhreakAccumulateNode {
final BaseAccumulation accctx = (BaseAccumulation) leftTuple.getContextObject();
// add a new match
addMatch(accNode, accumulate, leftTuple, rightTuple,
- null, null, reteEvaluator, am,
- accctx, true, false);
+ null, null, am, accctx,
+ true, false);
}
}
} else {
@@ -510,7 +508,6 @@ public class PhreakAccumulateNode {
accumulate,
rightTuple,
childLeftTuple,
- reteEvaluator,
am,
accctx,
true);
@@ -518,8 +515,8 @@ public class PhreakAccumulateNode {
}
// add a new match
addMatch(accNode, accumulate, leftTuple, rightTuple,
- null, childLeftTuple, reteEvaluator, am,
- accctx, true, false);
+ null, childLeftTuple, am, accctx,
+ true, false);
} else if (childLeftTuple != null && childLeftTuple.getLeftParent() == leftTuple) {
if (leftTuple.getStagedType() == LeftTuple.NONE) {
trgLeftTuples.addUpdate(leftTuple);
@@ -533,7 +530,6 @@ public class PhreakAccumulateNode {
accumulate,
rightTuple,
childLeftTuple,
- reteEvaluator,
am,
accctx,
true);
@@ -548,7 +544,6 @@ public class PhreakAccumulateNode {
private void doLeftDeletes(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -565,7 +560,7 @@ public class PhreakAccumulateNode {
BaseAccumulation accctx = (BaseAccumulation) leftTuple.getContextObject();
leftTuple.setContextObject( null );
- removePreviousMatchesForLeftTuple(accumulate, leftTuple, reteEvaluator, am, accctx, false);
+ removePreviousMatchesForLeftTuple(accumulate, leftTuple, am, accctx, false);
propagateDelete( trgLeftTuples, stagedLeftTuples, accctx );
}
@@ -578,14 +573,13 @@ public class PhreakAccumulateNode {
protected void propagateDelete( TupleSets trgLeftTuples, TupleSets stagedLeftTuples, Object accPropCtx ) {
AccumulateContextEntry entry = (AccumulateContextEntry) accPropCtx;
if ( entry.isPropagated() ) {
- normalizeStagedTuples( stagedLeftTuples, (TupleImpl) entry.getResultLeftTuple() );
+ normalizeStagedTuples(stagedLeftTuples, (TupleImpl) entry.getResultLeftTuple());
trgLeftTuples.addDelete( (TupleImpl) entry.getResultLeftTuple() );
}
}
private void doRightDeletes(AccumulateNode accNode,
AccumulateMemory am,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples) {
TupleMemory rtm = am.getBetaMemory().getRightTupleMemory();
@@ -606,7 +600,7 @@ public class PhreakAccumulateNode {
TupleImpl leftTuple = match.getLeftParent();
final BaseAccumulation accctx = (BaseAccumulation) leftTuple.getContextObject();
// FIXME This will be really slow, if it re-accumulates on the same LeftTuple (MDP)
- removeMatch(accNode, accumulate, rightTuple, match, reteEvaluator, am, accctx, true);
+ removeMatch(accNode, accumulate, rightTuple, match, am, accctx, true);
if (leftTuple.getStagedType() == LeftTuple.NONE) {
trgLeftTuples.addUpdate(leftTuple);
@@ -626,7 +620,6 @@ public class PhreakAccumulateNode {
final Accumulate accumulate,
final TupleImpl leftTuple,
final PropagationContext context,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory memory,
final BaseAccumulation accctx,
final TupleSets trgLeftTuples,
@@ -636,14 +629,14 @@ public class PhreakAccumulateNode {
accctx.setPropagationContext( null );
Object result = accumulate.getResult(memory.workingMemoryContext, accctx, leftTuple, reteEvaluator);
- propagateResult( accNode, sink, leftTuple, context, reteEvaluator, memory, trgLeftTuples, stagedLeftTuples,
- null, result, (AccumulateContextEntry) accctx, propagationContext, reteEvaluator.getRuleSessionConfiguration().isAccumulateNullPropagation());
+ propagateResult( accNode, sink, leftTuple, context, memory, trgLeftTuples, stagedLeftTuples, null,
+ result, (AccumulateContextEntry) accctx, propagationContext, reteEvaluator.getRuleSessionConfiguration().isAccumulateNullPropagation());
}
protected final void propagateResult(AccumulateNode accNode, LeftTupleSink sink, TupleImpl leftTuple, PropagationContext context,
- ReteEvaluator reteEvaluator, AccumulateMemory memory, TupleSets trgLeftTuples,
- TupleSets stagedLeftTuples, Object key, Object result,
- AccumulateContextEntry accPropCtx, PropagationContext propagationContext, boolean allowNullPropagation) {
+ AccumulateMemory memory, TupleSets trgLeftTuples, TupleSets stagedLeftTuples,
+ Object key, Object result, AccumulateContextEntry accPropCtx,
+ PropagationContext propagationContext, boolean allowNullPropagation) {
if ( !allowNullPropagation && result == null) {
if ( accPropCtx.isPropagated()) {
// retract
@@ -684,7 +677,7 @@ public class PhreakAccumulateNode {
childLeftTuple.setPropagationContext( propagationContext != null ? propagationContext : leftTuple.getPropagationContext() );
if ( accPropCtx.isPropagated()) {
- normalizeStagedTuples( stagedLeftTuples, childLeftTuple );
+ normalizeStagedTuples(stagedLeftTuples, childLeftTuple);
if (isAllowed) {
// modify
@@ -711,7 +704,6 @@ public class PhreakAccumulateNode {
final TupleImpl rightTuple,
final TupleImpl currentLeftChild,
final TupleImpl currentRightChild,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory am,
final BaseAccumulation accctx,
final boolean useLeftMemory,
@@ -759,7 +751,6 @@ public class PhreakAccumulateNode {
final Accumulate accumulate,
final TupleImpl rightTuple,
final TupleImpl match,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory am,
final BaseAccumulation accctx,
final boolean reaccumulate) {
@@ -793,7 +784,6 @@ public class PhreakAccumulateNode {
leftParent,
rightParent,
match,
- reteEvaluator,
am,
accctx,
reaccumulate);
@@ -807,12 +797,11 @@ public class PhreakAccumulateNode {
final TupleImpl leftParent,
final TupleImpl unused1,
final TupleImpl unused2,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory am,
final BaseAccumulation accctx,
final boolean reaccumulate) {
if (reaccumulate) {
- reinit(accumulate, leftParent, reteEvaluator, am, accctx);
+ reinit(accumulate, leftParent, am, accctx);
for (TupleImpl childMatch = leftParent.getFirstChild(); childMatch != null; childMatch = childMatch.getHandleNext()) {
TupleImpl rightTuple = childMatch.getRightParent();
@@ -834,7 +823,6 @@ public class PhreakAccumulateNode {
private void removePreviousMatchesForRightTuple(final AccumulateNode accNode,
final Accumulate accumulate,
final TupleImpl rightTuple,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory memory,
final TupleImpl firstChild,
final TupleSets trgLeftTuples) {
@@ -847,7 +835,6 @@ public class PhreakAccumulateNode {
accumulate,
rightTuple,
match,
- reteEvaluator,
memory,
accctx,
true);
@@ -860,9 +847,8 @@ public class PhreakAccumulateNode {
}
}
- private static void removePreviousMatchesForLeftTuple(final Accumulate accumulate,
+ private void removePreviousMatchesForLeftTuple(final Accumulate accumulate,
final TupleImpl leftTuple,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory memory,
final BaseAccumulation accctx,
boolean reInit) {
@@ -874,11 +860,11 @@ public class PhreakAccumulateNode {
}
if (reInit) {
- reinit(accumulate, leftTuple, reteEvaluator, memory, accctx);
+ reinit(accumulate, leftTuple, memory, accctx);
}
}
- private static void reinit(Accumulate accumulate, TupleImpl leftTuple, ReteEvaluator reteEvaluator, AccumulateMemory memory, BaseAccumulation accctx) {
+ private void reinit(Accumulate accumulate, TupleImpl leftTuple, AccumulateMemory memory, BaseAccumulation accctx) {
// Create the function context, but allow init to override it.
Object funcContext = ((AccumulateContextEntry) accctx).getFunctionContext();
funcContext = accumulate.init(memory.workingMemoryContext, accctx, funcContext, leftTuple, reteEvaluator);
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncReceiveNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncReceiveNode.java
index 8ee2e02c19..427db05f19 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncReceiveNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncReceiveNode.java
@@ -36,8 +36,13 @@ import static org.drools.core.phreak.PhreakAsyncSendNode.isAllowed;
public class PhreakAsyncReceiveNode {
private static final Logger log = LoggerFactory.getLogger( PhreakAsyncReceiveNode.class );
- public void doNode(ReteEvaluator reteEvaluator,
- AsyncReceiveNode node,
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakAsyncReceiveNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(AsyncReceiveNode node,
AsyncReceiveMemory memory,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -47,7 +52,7 @@ public class PhreakAsyncReceiveNode {
doLeftInserts( memory, srcLeftTuples );
}
- doPropagateChildLeftTuples( node, memory, reteEvaluator, sink, trgLeftTuples );
+ doPropagateChildLeftTuples( node, memory, sink, trgLeftTuples );
srcLeftTuples.resetAll();
}
@@ -63,9 +68,8 @@ public class PhreakAsyncReceiveNode {
}
}
- private static void doPropagateChildLeftTuples(AsyncReceiveNode node,
+ private void doPropagateChildLeftTuples(AsyncReceiveNode node,
AsyncReceiveMemory memory,
- ReteEvaluator reteEvaluator,
LeftTupleSink sink,
TupleSets trgLeftTuples) {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncSendNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncSendNode.java
index 533ac228a4..d79696033a 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncSendNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakAsyncSendNode.java
@@ -36,19 +36,26 @@ import org.drools.core.reteoo.BetaMemory;
import org.drools.core.reteoo.TupleImpl;
import org.kie.internal.concurrent.ExecutorProviderFactory;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
+
public class PhreakAsyncSendNode {
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakAsyncSendNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
private Executor executor() {
return ExecutorProviderFactory.getExecutorProvider().getExecutor();
}
- public void doNode(ReteEvaluator reteEvaluator,
- AsyncSendNode node,
+ public void doNode(AsyncSendNode node,
AsyncSendMemory memory,
TupleSets srcLeftTuples) {
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(node, memory, reteEvaluator, srcLeftTuples);
+ doLeftInserts(node, memory, srcLeftTuples);
}
srcLeftTuples.resetAll();
@@ -56,7 +63,6 @@ public class PhreakAsyncSendNode {
public void doLeftInserts(AsyncSendNode node,
AsyncSendMemory memory,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples) {
BetaMemory bm = memory.getBetaMemory();
@@ -71,7 +77,7 @@ public class PhreakAsyncSendNode {
PropagationContext propagationContext = leftTuple.getPropagationContext();
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory(node, leftTuple);
+ boolean useLeftMemory = useLeftMemory(node, leftTuple);
if (useLeftMemory) {
memory.getBetaMemory().getLeftTupleMemory().add(leftTuple);
@@ -84,8 +90,8 @@ public class PhreakAsyncSendNode {
executor().execute( () -> {
// TODO context is not thread safe, it needs to be cloned
- fetchAndSendResults( node, memory, reteEvaluator, context, betaConstraints, alphaConstraints, dataProvider,
- resultClass, finalLeftTuple, propagationContext );
+ fetchAndSendResults( node, memory, context, betaConstraints, alphaConstraints, dataProvider, resultClass,
+ finalLeftTuple, propagationContext );
} );
leftTuple.clearStaged();
@@ -94,9 +100,9 @@ public class PhreakAsyncSendNode {
betaConstraints.resetTuple(context);
}
- private void fetchAndSendResults(AsyncSendNode node, AsyncSendMemory memory, ReteEvaluator reteEvaluator,
- Object context, BetaConstraints betaConstraints, AlphaNodeFieldConstraint[] alphaConstraints,
- DataProvider dataProvider, Class<?> resultClass, TupleImpl leftTuple, PropagationContext propagationContext ) {
+ private void fetchAndSendResults(AsyncSendNode node, AsyncSendMemory memory, Object context,
+ BetaConstraints betaConstraints, AlphaNodeFieldConstraint[] alphaConstraints, DataProvider dataProvider,
+ Class<?> resultClass, TupleImpl leftTuple, PropagationContext propagationContext ) {
for (final java.util.Iterator<?> it = dataProvider.getResults(leftTuple,
reteEvaluator,
memory.providerContext); it.hasNext(); ) {
@@ -108,7 +114,7 @@ public class PhreakAsyncSendNode {
InternalFactHandle factHandle = node.createFactHandle(leftTuple, propagationContext, reteEvaluator, object);
if ( isAllowed( factHandle, alphaConstraints, reteEvaluator ) ) {
- propagate( node, reteEvaluator, factHandle, betaConstraints, context );
+ propagate( node, factHandle, betaConstraints, context );
}
}
}
@@ -127,7 +133,6 @@ public class PhreakAsyncSendNode {
}
public void propagate( AsyncSendNode node,
- ReteEvaluator reteEvaluator,
InternalFactHandle factHandle,
BetaConstraints betaConstraints,
Object context ) {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakBranchNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakBranchNode.java
index dca64b46b5..1d75fa1cdc 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakBranchNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakBranchNode.java
@@ -19,21 +19,33 @@
package org.drools.core.phreak;
import org.drools.core.common.ActivationsManager;
+import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.reteoo.ConditionalBranchEvaluator;
import org.drools.core.reteoo.ConditionalBranchEvaluator.ConditionalExecution;
import org.drools.core.reteoo.ConditionalBranchNode;
import org.drools.core.reteoo.ConditionalBranchNode.ConditionalBranchMemory;
+
+import static org.drools.core.phreak.PhreakNodeOperations.deleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.normalizeStagedTuples;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
+
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.core.reteoo.RuleTerminalNode;
import org.drools.core.reteoo.RuleTerminalNodeLeftTuple;
import org.drools.core.reteoo.TupleFactory;
import org.drools.core.reteoo.TupleImpl;
-import static org.drools.core.phreak.RuleNetworkEvaluator.normalizeStagedTuples;
public class PhreakBranchNode {
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakBranchNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
public void doNode(ActivationsManager activationsManager,
ConditionalBranchNode branchNode,
ConditionalBranchMemory cbm,
@@ -71,9 +83,9 @@ public class PhreakBranchNode {
TupleImpl next = leftTuple.getStagedNext();
boolean breaking = false;
- ConditionalExecution conditionalExecution = branchEvaluator.evaluate(leftTuple, activationsManager.getReteEvaluator(), cbm.context);
+ ConditionalExecution conditionalExecution = branchEvaluator.evaluate(leftTuple, reteEvaluator, cbm.context);
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory(branchNode, leftTuple);
+ boolean useLeftMemory = useLeftMemory(branchNode, leftTuple);
if (conditionalExecution != null) {
RuleTerminalNode rtn = (RuleTerminalNode) conditionalExecution.getSink().getFirstLeftTupleSink();
@@ -117,7 +129,7 @@ public class PhreakBranchNode {
oldRtn = (RuleTerminalNode) branchTuples.rtnLeftTuple.getSink();
}
- ConditionalExecution conditionalExecution = branchEvaluator.evaluate(leftTuple, activationsManager.getReteEvaluator(), cbm.context);
+ ConditionalExecution conditionalExecution = branchEvaluator.evaluate(leftTuple, reteEvaluator, cbm.context);
RuleTerminalNode newRtn = null;
boolean breaking = false;
@@ -163,7 +175,7 @@ public class PhreakBranchNode {
// Handle main branch
if (branchTuples.mainLeftTuple != null) {
- normalizeStagedTuples( stagedLeftTuples, branchTuples.mainLeftTuple );
+ normalizeStagedTuples(stagedLeftTuples, branchTuples.mainLeftTuple);
if (!breaking) {
// default consequence will also be executed
@@ -200,7 +212,7 @@ public class PhreakBranchNode {
}
if (branchTuples.mainLeftTuple != null) {
- RuleNetworkEvaluator.deleteChildLeftTuple(branchTuples.mainLeftTuple, trgLeftTuples, stagedLeftTuples);
+ deleteChildLeftTuple(branchTuples.mainLeftTuple, trgLeftTuples, stagedLeftTuples);
}
leftTuple.clearStaged();
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakEvalNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakEvalNode.java
index 3a88e6b5d7..edc3e4b5e2 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakEvalNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakEvalNode.java
@@ -23,11 +23,16 @@ import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.reteoo.EvalConditionNode;
import org.drools.core.reteoo.EvalConditionNode.EvalMemory;
+
+import static org.drools.core.phreak.PhreakNodeOperations.deleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.normalizeStagedTuples;
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
+import static org.drools.core.reteoo.TupleFactory.createLeftTuple;
+
import org.drools.core.reteoo.LeftTupleSink;
-import org.drools.core.reteoo.TupleFactory;
import org.drools.core.reteoo.TupleImpl;
-import static org.drools.core.phreak.RuleNetworkEvaluator.normalizeStagedTuples;
/**
* Created with IntelliJ IDEA.
@@ -39,9 +44,13 @@ import static org.drools.core.phreak.RuleNetworkEvaluator.normalizeStagedTuples;
public class PhreakEvalNode {
private static final String EVAL_LEFT_TUPLE_DELETED = "EVAL_LEFT_TUPLE_DELETED";
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakEvalNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
- public void doNode(ReteEvaluator reteEvaluator,
- EvalConditionNode evalNode,
+ public void doNode(EvalConditionNode evalNode,
EvalMemory em,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -53,11 +62,11 @@ public class PhreakEvalNode {
}
if (srcLeftTuples.getUpdateFirst() != null) {
- doLeftUpdates(evalNode, em, sink, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftUpdates(evalNode, em, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(evalNode, em, sink, reteEvaluator, srcLeftTuples, trgLeftTuples);
+ doLeftInserts(evalNode, em, sink, srcLeftTuples, trgLeftTuples);
}
srcLeftTuples.resetAll();
@@ -66,7 +75,6 @@ public class PhreakEvalNode {
public void doLeftInserts(EvalConditionNode evalNode,
EvalMemory em,
LeftTupleSink sink,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
EvalCondition condition = evalNode.getCondition();
@@ -76,11 +84,9 @@ public class PhreakEvalNode {
final boolean allowed = condition.isAllowed(leftTuple, reteEvaluator, em.context);
if (allowed) {
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory(evalNode, leftTuple);
+ boolean useLeftMemory = useLeftMemory(evalNode, leftTuple);
- trgLeftTuples.addInsert(TupleFactory.createLeftTuple(leftTuple,
- sink,
- leftTuple.getPropagationContext(), useLeftMemory));
+ trgLeftTuples.addInsert(createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), useLeftMemory));
}
leftTuple.clearStaged();
@@ -91,7 +97,6 @@ public class PhreakEvalNode {
public void doLeftUpdates(EvalConditionNode evalNode,
EvalMemory em,
LeftTupleSink sink,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -109,11 +114,11 @@ public class PhreakEvalNode {
// update
TupleImpl childLeftTuple = leftTuple.getFirstChild();
childLeftTuple.setPropagationContext( leftTuple.getPropagationContext());
- normalizeStagedTuples( stagedLeftTuples, childLeftTuple );
+ normalizeStagedTuples(stagedLeftTuples, childLeftTuple);
trgLeftTuples.addUpdate(childLeftTuple);
} else {
// assert
- trgLeftTuples.addInsert(TupleFactory.createLeftTuple(leftTuple,
+ trgLeftTuples.addInsert(createLeftTuple(leftTuple,
sink,
leftTuple.getPropagationContext(), true));
}
@@ -124,7 +129,7 @@ public class PhreakEvalNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
childLeftTuple.setPropagationContext( leftTuple.getPropagationContext());
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
// else do nothing
}
@@ -144,7 +149,7 @@ public class PhreakEvalNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
if (childLeftTuple != null) {
childLeftTuple.setPropagationContext( leftTuple.getPropagationContext());
- RuleNetworkEvaluator.deleteChildLeftTuple( childLeftTuple, trgLeftTuples, stagedLeftTuples );
+ deleteChildLeftTuple(childLeftTuple, trgLeftTuples, stagedLeftTuples);
}
leftTuple.clearStaged();
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakExistsNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakExistsNode.java
index 080fe2f76c..9a04d17567 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakExistsNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakExistsNode.java
@@ -33,17 +33,28 @@ import org.drools.core.reteoo.TupleMemory;
import org.drools.core.util.FastIterator;
import static org.drools.core.phreak.PhreakJoinNode.updateChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesExistentialReorderRightMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderLeftMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.findLeftTupleBlocker;
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
public class PhreakExistsNode {
- public void doNode(ReteEvaluator reteEvaluator,
- ExistsNode existsNode,
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakExistsNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
if (!existsNode.getRightInput().inputIsTupleToObjectNode()) {
- doNormalNode(existsNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doNormalNode(existsNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} else {
PhreakSubnetworkNotExistsNode.doSubNetworkNode(existsNode, sink, bm,
srcLeftTuples, trgLeftTuples, stagedLeftTuples);
@@ -53,7 +64,6 @@ public class PhreakExistsNode {
public void doNormalNode(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -65,39 +75,36 @@ public class PhreakExistsNode {
}
if (srcLeftTuples.getUpdateFirst() != null ) {
- RuleNetworkEvaluator.doUpdatesExistentialReorderLeftMemory(bm,
- srcLeftTuples);
+ doUpdatesReorderLeftMemory(bm, srcLeftTuples);
}
if ( srcRightTuples.getUpdateFirst() != null ) {
- RuleNetworkEvaluator.doUpdatesExistentialReorderRightMemory(bm,
- existsNode,
- srcRightTuples); // this also preserves the next rightTuple
+ doUpdatesExistentialReorderRightMemory(bm, existsNode, srcRightTuples); // this also preserves the next rightTuple
}
if (srcRightTuples.getInsertFirst() != null) {
// left deletes must come before right deletes. Otherwise right deletes could
// stage a deletion, that is later deleted in the rightDelete, causing potential problems
- doRightInserts(existsNode, sink, bm, reteEvaluator, srcRightTuples, trgLeftTuples);
+ doRightInserts(existsNode, sink, bm, srcRightTuples, trgLeftTuples);
}
if (srcRightTuples.getUpdateFirst() != null) {
// must come after rightInserts and before rightDeletes, to avoid staging clash
- doRightUpdates(existsNode, sink, bm, reteEvaluator, srcRightTuples, trgLeftTuples, stagedLeftTuples);
+ doRightUpdates(existsNode, sink, bm, srcRightTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcRightTuples.getDeleteFirst() != null) {
// must come after rightUpdetes, to avoid staging clash
- doRightDeletes(existsNode, bm, reteEvaluator, srcRightTuples, trgLeftTuples, stagedLeftTuples);
+ doRightDeletes(existsNode, bm, srcRightTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getUpdateFirst() != null) {
- doLeftUpdates(existsNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftUpdates(existsNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(existsNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples);
+ doLeftInserts(existsNode, sink, bm, srcLeftTuples, trgLeftTuples);
}
srcRightTuples.resetAll();
@@ -107,7 +114,6 @@ public class PhreakExistsNode {
public void doLeftInserts(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
TupleMemory ltm = bm.getLeftTupleMemory();
@@ -118,14 +124,14 @@ public class PhreakExistsNode {
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory(existsNode, leftTuple);
+ boolean useLeftMemory = useLeftMemory(existsNode, leftTuple);
constraints.updateFromTuple( contextEntry,
reteEvaluator,
leftTuple );
// This method will also remove rightTuples that are from subnetwork where no leftmemory use used
- RuleNetworkEvaluator.findLeftTupleBlocker( existsNode, rtm, contextEntry, constraints, leftTuple, useLeftMemory );
+ findLeftTupleBlocker(existsNode, rtm, contextEntry, constraints, leftTuple, useLeftMemory);
if (leftTuple.getBlocker() != null) {
// tuple is not blocked to propagate
@@ -143,7 +149,6 @@ public class PhreakExistsNode {
public void doRightInserts(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples) {
TupleMemory ltm = bm.getLeftTupleMemory();
@@ -195,7 +200,6 @@ public class PhreakExistsNode {
public void doLeftUpdates(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -266,7 +270,7 @@ public class PhreakExistsNode {
if (leftTuple.getFirstChild() != null) {
// no need to update pctx, as no right available, and pctx will exist on a parent LeftTuple anyway
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, leftTuple.getFirstChild() );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, leftTuple.getFirstChild());
}
// with no previous children. do nothing.
} else if (leftTuple.getFirstChild() == null) {
@@ -294,7 +298,6 @@ public class PhreakExistsNode {
public void doRightUpdates(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -401,7 +404,7 @@ public class PhreakExistsNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
if ( childLeftTuple != null ) {
childLeftTuple.setPropagationContext( rightTuple.getPropagationContext() );
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
}
@@ -431,7 +434,7 @@ public class PhreakExistsNode {
} else {
if (leftTuple.getFirstChild() != null) {
// no need to update pctx, as no right available, and pctx will exist on a parent LeftTuple anyway
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, leftTuple.getFirstChild() );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, leftTuple.getFirstChild());
}
blocker.removeBlocked(leftTuple);
}
@@ -443,7 +446,6 @@ public class PhreakExistsNode {
public void doRightDeletes(ExistsNode existsNode,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -495,7 +497,7 @@ public class PhreakExistsNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
if (childLeftTuple != null) {
childLeftTuple.setPropagationContext(rightTuple.getPropagationContext());
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakFromNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakFromNode.java
index c9858a8617..a1df534ad8 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakFromNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakFromNode.java
@@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;
+import org.drools.base.base.ValueResolver;
import org.drools.base.rule.accessor.DataProvider;
import org.drools.base.rule.constraint.AlphaNodeFieldConstraint;
import org.drools.core.common.BetaConstraints;
@@ -41,10 +42,18 @@ import org.drools.core.util.LinkedList;
import org.kie.api.runtime.rule.FactHandle;
import static org.drools.core.phreak.PhreakJoinNode.updateChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
public class PhreakFromNode {
- public void doNode(ReteEvaluator reteEvaluator,
- FromNode fromNode,
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakFromNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(FromNode fromNode,
FromMemory fm,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -56,11 +65,11 @@ public class PhreakFromNode {
}
if (srcLeftTuples.getUpdateFirst() != null) {
- doLeftUpdates(fromNode, fm, sink, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftUpdates(fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(fromNode, fm, sink, reteEvaluator, srcLeftTuples, trgLeftTuples);
+ doLeftInserts(fromNode, fm, sink, srcLeftTuples, trgLeftTuples);
}
srcLeftTuples.resetAll();
@@ -69,7 +78,6 @@ public class PhreakFromNode {
public void doLeftInserts(FromNode fromNode,
FromMemory fm,
LeftTupleSink sink,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
@@ -86,7 +94,7 @@ public class PhreakFromNode {
PropagationContext propagationContext = leftTuple.getPropagationContext();
Map<Object, RightTuple> matches = null;
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory(fromNode, leftTuple);
+ boolean useLeftMemory = useLeftMemory(fromNode, leftTuple);
if (useLeftMemory) {
fm.getBetaMemory().getLeftTupleMemory().add(leftTuple);
@@ -124,7 +132,6 @@ public class PhreakFromNode {
public void doLeftUpdates(FromNode fromNode,
FromMemory fm,
LeftTupleSink sink,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -205,7 +212,7 @@ public class PhreakFromNode {
while (childLeftTuple != null) {
childLeftTuple.setPropagationContext( leftTuple.getPropagationContext());
TupleImpl nextChild = childLeftTuple.getHandleNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
}
@@ -217,11 +224,11 @@ public class PhreakFromNode {
public static boolean isAllowed( FactHandle factHandle,
AlphaNodeFieldConstraint[] alphaConstraints,
- ReteEvaluator reteEvaluator,
+ ValueResolver valueResolver,
FromMemory fm ) {
if (alphaConstraints != null) {
for (int i = 0, length = alphaConstraints.length; i < length; i++) {
- if (!alphaConstraints[i].isAllowed(factHandle, reteEvaluator)) {
+ if (!alphaConstraints[i].isAllowed(factHandle, valueResolver)) {
return false;
}
}
@@ -266,7 +273,7 @@ public class PhreakFromNode {
TupleImpl childLeftTuple) {
if (childLeftTuple != null) {
childLeftTuple.setPropagationContext( propagationContext );
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakGroupByNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakGroupByNode.java
index dfb1603c0d..10c6040a0c 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakGroupByNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakGroupByNode.java
@@ -33,8 +33,12 @@ import org.kie.api.runtime.rule.FactHandle;
public class PhreakGroupByNode extends PhreakAccumulateNode {
+ public PhreakGroupByNode(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- AccumulateNode.BaseAccumulation initAccumulationContext(AccumulateMemory am, ReteEvaluator reteEvaluator, Accumulate accumulate, TupleImpl leftTuple) {
+ AccumulateNode.BaseAccumulation initAccumulationContext(AccumulateMemory am, Accumulate accumulate, TupleImpl leftTuple) {
GroupByContext accContext = new GroupByContext();
leftTuple.setContextObject( accContext );
// A lot less is done here, compared to super, as it needs to be done on demand during the Group creation.
@@ -59,7 +63,6 @@ public class PhreakGroupByNode extends PhreakAccumulateNode {
final Accumulate accumulate,
final TupleImpl leftTuple,
final PropagationContext context,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory memory,
final AccumulateNode.BaseAccumulation accctx,
final TupleSets trgLeftTuples,
@@ -75,8 +78,8 @@ public class PhreakGroupByNode extends PhreakAccumulateNode {
Object result = accumulate.getResult(memory.workingMemoryContext, contextEntry, leftTuple, reteEvaluator);
- propagateResult( accNode, sink, leftTuple, context, reteEvaluator, memory, trgLeftTuples, stagedLeftTuples,
- contextEntry.getKey(), result, contextEntry, propagationContext, false ); // don't want to propagate null
+ propagateResult( accNode, sink, leftTuple, context, memory, trgLeftTuples, stagedLeftTuples, contextEntry.getKey(),
+ result, contextEntry, propagationContext, false ); // don't want to propagate null
contextEntry.setToPropagate(false);
}
@@ -88,7 +91,6 @@ public class PhreakGroupByNode extends PhreakAccumulateNode {
final TupleImpl leftTuple,
final TupleImpl rightParent,
final TupleImpl match,
- final ReteEvaluator reteEvaluator,
final AccumulateMemory am,
final AccumulateNode.BaseAccumulation accctx,
final boolean reaccumulate) {
@@ -128,7 +130,6 @@ public class PhreakGroupByNode extends PhreakAccumulateNode {
leftTuple,
null,
null,
- reteEvaluator,
am,
accctx,
true);
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakJoinNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakJoinNode.java
index 2f55e33847..47e8d94086 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakJoinNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakJoinNode.java
@@ -31,9 +31,20 @@ import org.drools.core.reteoo.TupleMemory;
import org.drools.core.util.AbstractHashTable;
import org.drools.core.util.FastIterator;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderLeftMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderRightMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
+
public class PhreakJoinNode {
- public void doNode(ReteEvaluator reteEvaluator,
- JoinNode joinNode,
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakJoinNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(JoinNode joinNode,
LeftTupleSink sink,
BetaMemory bm,
TupleSets srcLeftTuples,
@@ -51,27 +62,27 @@ public class PhreakJoinNode {
}
if (srcRightTuples.getUpdateFirst() != null) {
- RuleNetworkEvaluator.doUpdatesReorderRightMemory(bm, srcRightTuples);
+ doUpdatesReorderRightMemory(bm, srcRightTuples);
}
if (srcLeftTuples.getUpdateFirst() != null ) {
- RuleNetworkEvaluator.doUpdatesReorderLeftMemory(bm, srcLeftTuples);
+ doUpdatesReorderLeftMemory(bm, srcLeftTuples);
}
if (srcRightTuples.getUpdateFirst() != null) {
- doRightUpdates(joinNode, sink, bm, reteEvaluator, srcRightTuples, trgLeftTuples, stagedLeftTuples);
+ doRightUpdates(joinNode, sink, bm, srcRightTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getUpdateFirst() != null ) {
- doLeftUpdates(joinNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftUpdates(joinNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcRightTuples.getInsertFirst() != null) {
- doRightInserts(joinNode, sink, bm, reteEvaluator, srcRightTuples, trgLeftTuples);
+ doRightInserts(joinNode, sink, bm, srcRightTuples, trgLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(joinNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples);
+ doLeftInserts(joinNode, sink, bm, srcLeftTuples, trgLeftTuples);
}
srcRightTuples.resetAll();
@@ -81,7 +92,6 @@ public class PhreakJoinNode {
public void doLeftInserts(JoinNode joinNode,
LeftTupleSink sink,
BetaMemory<?> bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
TupleMemory ltm = bm.getLeftTupleMemory();
@@ -92,7 +102,7 @@ public class PhreakJoinNode {
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory( joinNode, leftTuple );
+ boolean useLeftMemory = useLeftMemory(joinNode, leftTuple);
if (useLeftMemory) {
ltm.add(leftTuple);
@@ -125,7 +135,6 @@ public class PhreakJoinNode {
public void doRightInserts(JoinNode joinNode,
LeftTupleSink sink,
BetaMemory<?> bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples) {
TupleMemory ltm = bm.getLeftTupleMemory();
@@ -175,7 +184,6 @@ public class PhreakJoinNode {
public void doLeftUpdates(JoinNode joinNode,
LeftTupleSink sink,
BetaMemory<?> bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -200,7 +208,7 @@ public class PhreakJoinNode {
for (TupleImpl childLeftTuple = leftTuple.getFirstChild(); childLeftTuple != null; ) {
TupleImpl nextChild = childLeftTuple.getHandleNext();
if (rightTuple == null || rightTuple.getMemory() != childLeftTuple.getRightParent().getMemory()) {
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
childLeftTuple = nextChild;
}
@@ -266,7 +274,7 @@ public class PhreakJoinNode {
} else if (childLeftTuple != null && childLeftTuple.getRightParent() == rightTuple) {
// delete, childLeftTuple is updated
TupleImpl nextChild = childLeftTuple.getHandleNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
}
@@ -278,7 +286,6 @@ public class PhreakJoinNode {
public void doRightUpdates(JoinNode joinNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -305,7 +312,7 @@ public class PhreakJoinNode {
while ( childLeftTuple != null ) {
childLeftTuple.setPropagationContext( rightTuple.getPropagationContext() );
TupleImpl nextChild = childLeftTuple.getRightParentNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
// childLeftTuple is now null, so the next check will attempt matches for new bucket
@@ -382,7 +389,7 @@ public class PhreakJoinNode {
// delete, childLeftTuple is updated
childLeftTuple.setPropagationContext(rightTuple.getPropagationContext());
TupleImpl nextChild = childLeftTuple.getRightParentNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
}
@@ -409,7 +416,7 @@ public class PhreakJoinNode {
while (childLeftTuple != null) {
TupleImpl nextChild = childLeftTuple.getHandleNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
}
@@ -436,7 +443,7 @@ public class PhreakJoinNode {
childLeftTuple.setPropagationContext(rightTuple.getPropagationContext());
while (childLeftTuple != null) {
TupleImpl nextChild = childLeftTuple.getRightParentNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactory.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactory.java
index f287dac4ad..4441387119 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactory.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactory.java
@@ -18,39 +18,40 @@
*/
package org.drools.core.phreak;
+import org.drools.core.common.ReteEvaluator;
import org.kie.api.internal.utils.KieService;
public interface PhreakNetworkNodeFactory extends KieService {
- PhreakJoinNode createPhreakJoinNode();
+ PhreakJoinNode createPhreakJoinNode(ReteEvaluator reteEvaluator);
- PhreakEvalNode createPhreakEvalNode();
+ PhreakEvalNode createPhreakEvalNode(ReteEvaluator reteEvaluator);
- PhreakFromNode createPhreakFromNode();
+ PhreakFromNode createPhreakFromNode(ReteEvaluator reteEvaluator);
- PhreakReactiveFromNode createPhreakReactiveFromNode();
+ PhreakReactiveFromNode createPhreakReactiveFromNode(ReteEvaluator reteEvaluator);
- PhreakNotNode createPhreakNotNode();
+ PhreakNotNode createPhreakNotNode(ReteEvaluator reteEvaluator);
- PhreakExistsNode createPhreakExistsNode();
+ PhreakExistsNode createPhreakExistsNode(ReteEvaluator reteEvaluator);
- PhreakAccumulateNode createPhreakAccumulateNode();
+ PhreakAccumulateNode createPhreakAccumulateNode(ReteEvaluator reteEvaluator);
- PhreakGroupByNode createPhreakGroupByNode();
+ PhreakGroupByNode createPhreakGroupByNode(ReteEvaluator reteEvaluator);
- PhreakBranchNode createPhreakBranchNode();
+ PhreakBranchNode createPhreakBranchNode(ReteEvaluator reteEvaluator);
- PhreakQueryNode createPhreakQueryNode();
+ PhreakQueryNode createPhreakQueryNode(ReteEvaluator reteEvaluator);
- PhreakTimerNode createPhreakTimerNode();
+ PhreakTimerNode createPhreakTimerNode(ReteEvaluator reteEvaluator);
- PhreakAsyncSendNode createPhreakAsyncSendNode();
+ PhreakAsyncSendNode createPhreakAsyncSendNode(ReteEvaluator reteEvaluator);
- PhreakAsyncReceiveNode createPhreakAsyncReceiveNode();
+ PhreakAsyncReceiveNode createPhreakAsyncReceiveNode(ReteEvaluator reteEvaluator);
- PhreakRuleTerminalNode createPhreakRuleTerminalNode();
+ PhreakRuleTerminalNode createPhreakRuleTerminalNode(ReteEvaluator reteEvaluator);
- PhreakQueryTerminalNode createPhreakQueryTerminalNode();
+ PhreakQueryTerminalNode createPhreakQueryTerminalNode(ReteEvaluator reteEvaluator);
class Factory {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactoryImpl.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactoryImpl.java
index f7b2db1ca5..0c5f117db5 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactoryImpl.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakNetworkNodeFactoryImpl.java
@@ -18,81 +18,83 @@
*/
package org.drools.core.phreak;
+import org.drools.core.common.ReteEvaluator;
+
public class PhreakNetworkNodeFactoryImpl implements PhreakNetworkNodeFactory {
@Override
- public PhreakJoinNode createPhreakJoinNode() {
- return new PhreakJoinNode();
+ public PhreakJoinNode createPhreakJoinNode(ReteEvaluator reteEvaluator) {
+ return new PhreakJoinNode(reteEvaluator);
}
@Override
- public PhreakEvalNode createPhreakEvalNode() {
- return new PhreakEvalNode();
+ public PhreakEvalNode createPhreakEvalNode(ReteEvaluator reteEvaluator) {
+ return new PhreakEvalNode(reteEvaluator);
}
@Override
- public PhreakFromNode createPhreakFromNode() {
- return new PhreakFromNode();
+ public PhreakFromNode createPhreakFromNode(ReteEvaluator reteEvaluator) {
+ return new PhreakFromNode(reteEvaluator);
}
@Override
- public PhreakReactiveFromNode createPhreakReactiveFromNode() {
- return new PhreakReactiveFromNode();
+ public PhreakReactiveFromNode createPhreakReactiveFromNode(ReteEvaluator reteEvaluator) {
+ return new PhreakReactiveFromNode(reteEvaluator);
}
@Override
- public PhreakNotNode createPhreakNotNode() {
- return new PhreakNotNode();
+ public PhreakNotNode createPhreakNotNode(ReteEvaluator reteEvaluator) {
+ return new PhreakNotNode(reteEvaluator);
}
@Override
- public PhreakExistsNode createPhreakExistsNode() {
- return new PhreakExistsNode();
+ public PhreakExistsNode createPhreakExistsNode(ReteEvaluator reteEvaluator) {
+ return new PhreakExistsNode(reteEvaluator);
}
@Override
- public PhreakAccumulateNode createPhreakAccumulateNode() {
- return new PhreakAccumulateNode();
+ public PhreakAccumulateNode createPhreakAccumulateNode(ReteEvaluator reteEvaluator) {
+ return new PhreakAccumulateNode(reteEvaluator);
}
@Override
- public PhreakGroupByNode createPhreakGroupByNode() {
- return new PhreakGroupByNode();
+ public PhreakGroupByNode createPhreakGroupByNode(ReteEvaluator reteEvaluator) {
+ return new PhreakGroupByNode(reteEvaluator);
}
@Override
- public PhreakBranchNode createPhreakBranchNode() {
- return new PhreakBranchNode();
+ public PhreakBranchNode createPhreakBranchNode(ReteEvaluator reteEvaluator) {
+ return new PhreakBranchNode(reteEvaluator);
}
@Override
- public PhreakQueryNode createPhreakQueryNode() {
- return new PhreakQueryNode();
+ public PhreakQueryNode createPhreakQueryNode(ReteEvaluator reteEvaluator) {
+ return new PhreakQueryNode(reteEvaluator);
}
@Override
- public PhreakTimerNode createPhreakTimerNode() {
- return new PhreakTimerNode();
+ public PhreakTimerNode createPhreakTimerNode(ReteEvaluator reteEvaluator) {
+ return new PhreakTimerNode(reteEvaluator);
}
@Override
- public PhreakAsyncSendNode createPhreakAsyncSendNode() {
- return new PhreakAsyncSendNode();
+ public PhreakAsyncSendNode createPhreakAsyncSendNode(ReteEvaluator reteEvaluator) {
+ return new PhreakAsyncSendNode(reteEvaluator);
}
@Override
- public PhreakAsyncReceiveNode createPhreakAsyncReceiveNode() {
+ public PhreakAsyncReceiveNode createPhreakAsyncReceiveNode(ReteEvaluator reteEvaluator) {
- return new PhreakAsyncReceiveNode();
+ return new PhreakAsyncReceiveNode(reteEvaluator);
}
@Override
- public PhreakRuleTerminalNode createPhreakRuleTerminalNode() {
- return new PhreakRuleTerminalNode();
+ public PhreakRuleTerminalNode createPhreakRuleTerminalNode(ReteEvaluator reteEvaluator) {
+ return new PhreakRuleTerminalNode(reteEvaluator);
}
@Override
- public PhreakQueryTerminalNode createPhreakQueryTerminalNode() {
- return new PhreakQueryTerminalNode();
+ public PhreakQueryTerminalNode createPhreakQueryTerminalNode(ReteEvaluator reteEvaluator) {
+ return new PhreakQueryTerminalNode(reteEvaluator);
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakNodeOperations.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakNodeOperations.java
new file mode 100644
index 0000000000..cc059077f2
--- /dev/null
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakNodeOperations.java
@@ -0,0 +1,265 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.drools.core.phreak;
+
+import org.drools.core.base.DroolsQueryImpl;
+import org.drools.core.common.BetaConstraints;
+import org.drools.core.common.ReteEvaluator;
+import org.drools.core.common.TupleSets;
+import org.drools.core.reteoo.BetaMemory;
+import org.drools.core.reteoo.BetaNode;
+import org.drools.core.reteoo.LeftTuple;
+import org.drools.core.reteoo.LeftTupleSource;
+import org.drools.core.reteoo.NotNode;
+import org.drools.core.reteoo.RightTuple;
+import org.drools.core.reteoo.SubnetworkTuple;
+import org.drools.core.reteoo.Tuple;
+import org.drools.core.reteoo.TupleImpl;
+import org.drools.core.reteoo.TupleMemory;
+import org.drools.core.util.FastIterator;
+
+import static org.drools.core.phreak.PhreakNotNode.updateBlockersAndPropagate;
+import static org.drools.core.reteoo.BetaNode.getBetaMemory;
+
+public class PhreakNodeOperations {
+
+ public static boolean useLeftMemory(LeftTupleSource tupleSource,
+ TupleImpl tuple) {
+ boolean useLeftMemory = true;
+ if (!tupleSource.isLeftTupleMemoryEnabled()) {
+ // This is a hack, to not add closed DroolsQuery objects
+ Object object = tuple.getRootTuple().getFactHandle().getObject();
+ if (!(object instanceof DroolsQueryImpl) || !((DroolsQueryImpl) object).isOpen()) {
+ useLeftMemory = false;
+ }
+ }
+ return useLeftMemory;
+ }
+
+ public static void normalizeStagedTuples(TupleSets stagedLeftTuples,
+ TupleImpl childLeftTuple) {
+ if (!childLeftTuple.isStagedOnRight()) {
+ switch (childLeftTuple.getStagedType()) {
+ // handle clash with already staged entries
+ case LeftTuple.INSERT:
+ stagedLeftTuples.removeInsert(childLeftTuple);
+ break;
+ case LeftTuple.UPDATE:
+ stagedLeftTuples.removeUpdate(childLeftTuple);
+ break;
+ }
+ }
+ }
+
+ public static void findLeftTupleBlocker(BetaNode betaNode,
+ TupleMemory rtm,
+ Object contextEntry,
+ BetaConstraints constraints,
+ TupleImpl leftTuple,
+ boolean useLeftMemory) {
+ // This method will also remove rightTuples that are from subnetwork where no leftmemory use used
+ FastIterator it = betaNode.getRightIterator(rtm);
+ for (RightTuple rightTuple = betaNode.getFirstRightTuple(leftTuple, rtm, it); rightTuple != null;) {
+ RightTuple nextRight = (RightTuple) it.next(rightTuple);
+ if (constraints.isAllowedCachedLeft(contextEntry,
+ rightTuple.getFactHandleForEvaluation())) {
+ leftTuple.setBlocker(rightTuple);
+
+ if (useLeftMemory) {
+ rightTuple.addBlocked((LeftTuple) leftTuple);
+ break;
+ } else if (betaNode.getRightInput().inputIsTupleToObjectNode()) {
+ // If we aren't using leftMemory and the right input is a TupleToObjectNode, then we must iterate and find all subetwork right tuples and remove them
+ // so we don't break
+ rtm.remove(rightTuple);
+ } else {
+ break;
+ }
+ }
+ rightTuple = nextRight;
+ }
+ }
+
+ public static void unlinkAndDeleteChildLeftTuple(TupleSets trgLeftTuples,
+ TupleSets stagedLeftTuples,
+ TupleImpl childLeftTuple) {
+ childLeftTuple.unlinkFromRightParent();
+ childLeftTuple.unlinkFromLeftParent();
+ deleteChildLeftTuple(childLeftTuple, trgLeftTuples, stagedLeftTuples);
+ }
+
+
+ public static void deleteChildLeftTuple(TupleImpl childLeftTuple, TupleSets trgLeftTuples, TupleSets stagedLeftTuples) {
+ if (childLeftTuple.isStagedOnRight()) {
+ ((SubnetworkTuple) childLeftTuple).moveStagingFromRightToLeft();
+ } else {
+ switch (childLeftTuple.getStagedType()) {
+ // handle clash with already staged entries
+ case Tuple.INSERT:
+ stagedLeftTuples.removeInsert(childLeftTuple);
+ trgLeftTuples.addNormalizedDelete(childLeftTuple);
+ return;
+ case Tuple.UPDATE:
+ stagedLeftTuples.removeUpdate(childLeftTuple);
+ break;
+ }
+ }
+ trgLeftTuples.addDelete(childLeftTuple);
+ }
+
+
+ public static void doAddExistentialRightMemoryForReorder(TupleMemory rtm,
+ RightTuple rightTuple,
+ boolean resumeFromCurrent) {
+ rtm.add(rightTuple);
+
+ if (resumeFromCurrent) {
+ if (rightTuple.getBlocked() != null && rightTuple.getTempNextRightTuple() == null) {
+ // the next RightTuple was null, but current RightTuple was added back into the same bucket, so reset as root blocker to re-match can be attempted
+ rightTuple.setTempNextRightTuple(rightTuple);
+ }
+ }
+
+ doUpdatesReorderChildLeftTuple(rightTuple);
+ }
+
+ public static void doRemoveExistentialRightMemoryForReorder(TupleMemory rtm,
+ RightTuple rightTuple,
+ boolean resumeFromCurrent) {
+ if (rightTuple.getMemory() != null) {
+
+ if (resumeFromCurrent) {
+ if (rightTuple.getBlocked() != null) {
+ // look for a non-staged right tuple first forward ...
+ RightTuple tempRightTuple = (RightTuple) rightTuple.getNext();
+ while (tempRightTuple != null && tempRightTuple.getStagedType() != LeftTuple.NONE) {
+ // next cannot be an updated or deleted rightTuple
+ tempRightTuple = (RightTuple) tempRightTuple.getNext();
+ }
+
+ // ... and if cannot find one try backward
+ if (tempRightTuple == null) {
+ tempRightTuple = (RightTuple) rightTuple.getPrevious();
+ while (tempRightTuple != null && tempRightTuple.getStagedType() != LeftTuple.NONE) {
+ // next cannot be an updated or deleted rightTuple
+ tempRightTuple = (RightTuple) tempRightTuple.getPrevious();
+ }
+ }
+
+ rightTuple.setTempNextRightTuple(tempRightTuple);
+ }
+ }
+
+ rightTuple.setTempBlocked(rightTuple.getBlocked());
+ rightTuple.setBlocked(null);
+ rtm.remove(rightTuple);
+ }
+ }
+
+
+
+ public static void doUpdatesReorderRightMemory(BetaMemory bm,
+ TupleSets srcRightTuples) {
+ TupleMemory rtm = bm.getRightTupleMemory();
+
+ for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
+ if (rightTuple.getMemory() != null) {
+ rtm.removeAdd(rightTuple);
+ doUpdatesReorderChildLeftTuple(rightTuple);
+ }
+ }
+ }
+
+ public static void doUpdatesReorderChildLeftTuple(TupleImpl rightTuple) {
+ for (TupleImpl childLeftTuple = rightTuple.getFirstChild(); childLeftTuple != null;) {
+ TupleImpl childNext = childLeftTuple.getRightParentNext();
+ childLeftTuple.reAddLeft();
+ childLeftTuple = childNext;
+ }
+ }
+
+
+ public static void doExistentialUpdatesReorderChildLeftTuple(ReteEvaluator reteEvaluator,
+ NotNode notNode,
+ RightTuple rightTuple) {
+ BetaMemory bm = getBetaMemory(notNode, reteEvaluator);
+ TupleMemory rtm = bm.getRightTupleMemory();
+
+ boolean resumeFromCurrent = !(notNode.isIndexedUnificationJoin() || rtm.getIndexType().isComparison());
+ doRemoveExistentialRightMemoryForReorder(rtm, rightTuple, resumeFromCurrent);
+ doAddExistentialRightMemoryForReorder(rtm, rightTuple, resumeFromCurrent);
+
+ updateBlockersAndPropagate(notNode, rightTuple, reteEvaluator, rtm, bm.getContext(), notNode.getRawConstraints(), !resumeFromCurrent, null, null, null);
+ }
+
+
+ public static void doUpdatesReorderLeftMemory(BetaMemory bm,
+ TupleSets srcLeftTuples) {
+ TupleMemory ltm = bm.getLeftTupleMemory();
+
+ // sides must first be re-ordered, to ensure iteration integrity
+ for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
+ if (leftTuple.getMemory() != null) {
+ ltm.remove(leftTuple);
+ }
+ }
+
+ for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
+ ltm.add(leftTuple);
+ for (TupleImpl childLeftTuple = leftTuple.getFirstChild(); childLeftTuple != null;) {
+ TupleImpl childNext = childLeftTuple.getHandleNext();
+ childLeftTuple.reAddRight();
+ childLeftTuple = childNext;
+ }
+ }
+ }
+
+ public static void doUpdatesExistentialReorderRightMemory(BetaMemory bm,
+ BetaNode betaNode,
+ TupleSets srcRightTuples) {
+ TupleMemory rtm = bm.getRightTupleMemory();
+
+ boolean resumeFromCurrent = !(betaNode.isIndexedUnificationJoin() || rtm.getIndexType().isComparison());
+
+ // remove all the staged rightTuples from the memory before to readd them all
+ // this is to avoid split bucket when an updated rightTuple hasn't been moved yet
+ // and so it is the first entry in the wrong bucket
+
+ if (rtm.getIndexType() != TupleMemory.IndexType.NONE) {
+ for (TupleImpl rightTuple = srcRightTuples.getDeleteFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
+ rtm.remove(rightTuple);
+ }
+ }
+
+ for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
+ doRemoveExistentialRightMemoryForReorder(rtm, (RightTuple) rightTuple, resumeFromCurrent);
+ }
+
+ for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
+ doAddExistentialRightMemoryForReorder(rtm, (RightTuple) rightTuple, resumeFromCurrent);
+ }
+
+ if (rtm.getIndexType() != TupleMemory.IndexType.NONE) {
+ for (TupleImpl rightTuple = srcRightTuples.getDeleteFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
+ rtm.add(rightTuple);
+ }
+ }
+ }
+
+}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakNotNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakNotNode.java
index 8feb1b1200..ae3da6f756 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakNotNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakNotNode.java
@@ -18,6 +18,7 @@
*/
package org.drools.core.phreak;
+import org.drools.base.base.ValueResolver;
import org.drools.core.common.BetaConstraints;
import org.drools.core.common.PropagationContext;
import org.drools.core.common.ReteEvaluator;
@@ -33,10 +34,21 @@ import org.drools.core.reteoo.TupleMemory;
import org.drools.core.util.FastIterator;
import static org.drools.core.phreak.PhreakJoinNode.updateChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesExistentialReorderRightMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderLeftMemory;
+import static org.drools.core.phreak.PhreakNodeOperations.findLeftTupleBlocker;
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
public class PhreakNotNode {
- public void doNode(ReteEvaluator reteEvaluator,
- NotNode notNode,
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakNotNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
TupleSets srcLeftTuples,
@@ -44,7 +56,7 @@ public class PhreakNotNode {
TupleSets stagedLeftTuples) {
if (!notNode.getRightInput().inputIsTupleToObjectNode()) {
- doNormalNode(notNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doNormalNode(notNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} else {
PhreakSubnetworkNotExistsNode.doSubNetworkNode(notNode, sink, bm,
srcLeftTuples, trgLeftTuples, stagedLeftTuples);
@@ -54,7 +66,6 @@ public class PhreakNotNode {
public void doNormalNode(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -69,39 +80,36 @@ public class PhreakNotNode {
if (srcLeftTuples.getUpdateFirst() != null) {
// must happen before right inserts, so it can find left tuples to block.
- RuleNetworkEvaluator.doUpdatesExistentialReorderLeftMemory(bm,
- srcLeftTuples);
+ doUpdatesReorderLeftMemory(bm, srcLeftTuples);
}
if ( srcRightTuples.getUpdateFirst() != null) {
- RuleNetworkEvaluator.doUpdatesExistentialReorderRightMemory(bm,
- notNode,
- srcRightTuples); // this also preserves the next rightTuple
+ doUpdatesExistentialReorderRightMemory(bm, notNode, srcRightTuples); // this also preserves the next rightTuple
}
if (srcRightTuples.getInsertFirst() != null) {
// must come before right updates and inserts, as they might cause insert propagation, while this causes delete propagations, resulting in staging clash.
- doRightInserts(notNode, bm, reteEvaluator, srcRightTuples, trgLeftTuples, stagedLeftTuples);
+ doRightInserts(notNode, bm, srcRightTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcRightTuples.getUpdateFirst() != null) {
// must come after rightInserts and before rightDeletes, to avoid staging clash
- doRightUpdates(notNode, sink, bm, reteEvaluator, srcRightTuples, trgLeftTuples, stagedLeftTuples);
+ doRightUpdates(notNode, sink, bm, srcRightTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcRightTuples.getDeleteFirst() != null) {
// must come after rightUpdates, to avoid staging clash
- doRightDeletes(notNode, sink, bm, reteEvaluator, srcRightTuples, trgLeftTuples);
+ doRightDeletes(notNode, sink, bm, srcRightTuples, trgLeftTuples);
}
if (srcLeftTuples.getUpdateFirst() != null) {
- doLeftUpdates(notNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftUpdates(notNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(notNode, sink, bm, reteEvaluator, srcLeftTuples, trgLeftTuples);
+ doLeftInserts(notNode, sink, bm, srcLeftTuples, trgLeftTuples);
}
srcRightTuples.resetAll();
@@ -111,7 +119,6 @@ public class PhreakNotNode {
public void doLeftInserts(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
TupleMemory ltm = bm.getLeftTupleMemory();
@@ -122,14 +129,14 @@ public class PhreakNotNode {
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
- boolean useLeftMemory = RuleNetworkEvaluator.useLeftMemory(notNode, leftTuple);
+ boolean useLeftMemory = useLeftMemory(notNode, leftTuple);
constraints.updateFromTuple( contextEntry,
reteEvaluator,
leftTuple );
// This method will also remove rightTuples that are from subnetwork where no leftmemory use used
- RuleNetworkEvaluator.findLeftTupleBlocker( notNode, rtm, contextEntry, constraints, leftTuple, useLeftMemory );
+ findLeftTupleBlocker(notNode, rtm, contextEntry, constraints, leftTuple, useLeftMemory);
if (leftTuple.getBlocker() == null) {
insertChildLeftTuple( sink, trgLeftTuples, ltm, leftTuple, leftTuple.getPropagationContext(), useLeftMemory );
@@ -142,7 +149,6 @@ public class PhreakNotNode {
public void doRightInserts(NotNode notNode,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples ) {
@@ -191,7 +197,7 @@ public class PhreakNotNode {
if ( childLeftTuple != null ) { // NotNode only has one child
childLeftTuple.setPropagationContext( rightTuple.getPropagationContext() );
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
}
@@ -216,7 +222,6 @@ public class PhreakNotNode {
public void doLeftUpdates(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -286,7 +291,7 @@ public class PhreakNotNode {
// no need to remove, as we removed at the start
// to be matched against, as it's now blocked
childLeftTuple.setPropagationContext(leftTuple.getBlocker().getPropagationContext()); // we have the righttuple, so use it for the pctx
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
} // else: it's blocked now and no children so blocked before, thus do nothing
} else if (childLeftTuple == null) {
// not blocked, with no children, must have been previously blocked so assert
@@ -309,7 +314,6 @@ public class PhreakNotNode {
public void doRightUpdates(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -355,7 +359,7 @@ public class PhreakNotNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
if ( childLeftTuple != null ) {
childLeftTuple.setPropagationContext( rightTuple.getPropagationContext() );
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
}
@@ -372,7 +376,7 @@ public class PhreakNotNode {
constraints.resetTuple(contextEntry);
}
- public static boolean updateBlockersAndPropagate(NotNode notNode, RightTuple rightTuple, ReteEvaluator reteEvaluator, TupleMemory rtm, Object contextEntry,
+ public static boolean updateBlockersAndPropagate(NotNode notNode, RightTuple rightTuple, ValueResolver reteEvaluator, TupleMemory rtm, Object contextEntry,
BetaConstraints constraints, boolean iterateFromStart, LeftTupleSink sink, TupleSets trgLeftTuples, TupleMemory ltm) {
LeftTuple firstBlocked = rightTuple.getTempBlocked();
if ( firstBlocked != null ) {
@@ -448,7 +452,7 @@ public class PhreakNotNode {
if (childLeftTuple != null) { // NotNode only has one child
childLeftTuple.setPropagationContext(leftTuple.getPropagationContext());
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple ); // no need to update pctx, as no right available, and pctx will exist on a parent LeftTuple anyway
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple); // no need to update pctx, as no right available, and pctx will exist on a parent LeftTuple anyway
}
} else {
blocker.removeBlocked(leftTuple);
@@ -461,7 +465,6 @@ public class PhreakNotNode {
public void doRightDeletes(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
- ReteEvaluator reteEvaluator,
TupleSets srcRightTuples,
TupleSets trgLeftTuples) {
TupleMemory ltm = bm.getLeftTupleMemory();
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryNode.java
index 6d2497d91a..b56b4af96f 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryNode.java
@@ -27,28 +27,37 @@ import org.drools.core.reteoo.LeftInputAdapterNode;
import org.drools.core.reteoo.LeftInputAdapterNode.LiaNodeMemory;
import org.drools.core.reteoo.QueryElementNode;
import org.drools.core.reteoo.QueryElementNode.QueryElementNodeMemory;
+
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+
import org.drools.core.reteoo.SegmentMemory;
import org.drools.core.reteoo.TupleImpl;
public class PhreakQueryNode {
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakQueryNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
public void doNode(QueryElementNode queryNode,
QueryElementNodeMemory qmem,
StackEntry stackEntry,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
if (srcLeftTuples.getDeleteFirst() != null) {
- doLeftDeletes(qmem, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ doLeftDeletes(qmem, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
}
if (srcLeftTuples.getUpdateFirst() != null) {
- doLeftUpdates(queryNode, qmem, reteEvaluator, srcLeftTuples);
+ doLeftUpdates(queryNode, qmem, srcLeftTuples);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(queryNode, qmem, stackEntry, reteEvaluator, srcLeftTuples);
+ doLeftInserts(queryNode, qmem, stackEntry, srcLeftTuples);
}
srcLeftTuples.resetAll();
@@ -57,7 +66,6 @@ public class PhreakQueryNode {
public void doLeftInserts(QueryElementNode queryNode,
QueryElementNodeMemory qmem,
StackEntry stackEntry,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples) {
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
@@ -84,7 +92,6 @@ public class PhreakQueryNode {
public void doLeftUpdates(QueryElementNode queryNode,
QueryElementNodeMemory qmem,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples) {
for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
@@ -113,7 +120,6 @@ public class PhreakQueryNode {
}
public void doLeftDeletes(QueryElementNodeMemory qmem,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -131,7 +137,7 @@ public class PhreakQueryNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
while (childLeftTuple != null) {
TupleImpl nextChild = childLeftTuple.getHandleNext();
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple( trgLeftTuples, stagedLeftTuples, childLeftTuple );
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
childLeftTuple = nextChild;
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryTerminalNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryTerminalNode.java
index b553d644b0..9898a21251 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryTerminalNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakQueryTerminalNode.java
@@ -21,7 +21,7 @@ package org.drools.core.phreak;
import java.util.List;
import org.drools.core.base.DroolsQueryImpl;
-import org.drools.core.common.ActivationsManager;
+import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.core.reteoo.PathMemory;
@@ -39,27 +39,32 @@ import org.drools.core.util.LinkedList;
* To change this template use File | Settings | File Templates.
*/
public class PhreakQueryTerminalNode {
- public void doNode(ActivationsManager activationsManager,
- LinkedList<StackEntry> stack,
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakQueryTerminalNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
+ public void doNode(LinkedList<StackEntry> stack,
QueryTerminalNode qtnNode,
TupleSets srcLeftTuples) {
if (srcLeftTuples.getDeleteFirst() != null) {
- doLeftDeletes(qtnNode, activationsManager, srcLeftTuples, stack);
+ doLeftDeletes(qtnNode, srcLeftTuples, stack);
}
if (srcLeftTuples.getUpdateFirst() != null) {
- doLeftUpdates(qtnNode, activationsManager, srcLeftTuples, stack);
+ doLeftUpdates(qtnNode, srcLeftTuples, stack);
}
if (srcLeftTuples.getInsertFirst() != null) {
- doLeftInserts(qtnNode, activationsManager, srcLeftTuples, stack);
+ doLeftInserts(qtnNode, srcLeftTuples, stack);
}
srcLeftTuples.resetAll();
}
public void doLeftInserts(QueryTerminalNode qtnNode,
- ActivationsManager activationsManager,
TupleSets srcLeftTuples,
LinkedList<StackEntry> stack) {
@@ -79,7 +84,7 @@ public class PhreakQueryTerminalNode {
// Add results to the adapter
dquery.getQueryResultCollector().rowAdded(qtnNode.getQuery(),
leftTuple,
- activationsManager.getReteEvaluator());
+ reteEvaluator);
leftTuple.clearStaged();
leftTuple = next;
@@ -87,7 +92,6 @@ public class PhreakQueryTerminalNode {
}
public void doLeftUpdates(QueryTerminalNode qtnNode,
- ActivationsManager activationsManager,
TupleSets srcLeftTuples,
LinkedList<StackEntry> stack) {
@@ -107,7 +111,7 @@ public class PhreakQueryTerminalNode {
// Add results to the adapter
dquery.getQueryResultCollector().rowUpdated(qtnNode.getQuery(),
leftTuple,
- activationsManager.getReteEvaluator());
+ reteEvaluator);
leftTuple.clearStaged();
leftTuple = next;
@@ -115,7 +119,6 @@ public class PhreakQueryTerminalNode {
}
public void doLeftDeletes(QueryTerminalNode qtnNode,
- ActivationsManager activationsManager,
TupleSets srcLeftTuples,
LinkedList<StackEntry> stack) {
@@ -137,7 +140,7 @@ public class PhreakQueryTerminalNode {
// Add results to the adapter
dquery.getQueryResultCollector().rowRemoved(qtnNode.getQuery(),
leftTuple,
- activationsManager.getReteEvaluator());
+ reteEvaluator);
leftTuple.clearStaged();
leftTuple = next;
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakReactiveFromNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakReactiveFromNode.java
index 9e697ccc19..b76fe05a2b 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakReactiveFromNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakReactiveFromNode.java
@@ -25,15 +25,21 @@ import org.drools.core.reteoo.ReactiveFromNode;
import org.drools.core.reteoo.ReactiveFromNode.ReactiveFromMemory;
public class PhreakReactiveFromNode extends PhreakFromNode {
- public void doNode(ReteEvaluator reteEvaluator,
- ReactiveFromNode fromNode,
+
+
+
+ public PhreakReactiveFromNode(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
+ public void doNode(ReactiveFromNode fromNode,
ReactiveFromMemory fm,
LeftTupleSink sink,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
- super.doNode(reteEvaluator, fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
trgLeftTuples.addAll(fm.getStagedLeftTuples().takeAll());
}
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakRuleTerminalNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakRuleTerminalNode.java
index 911791e818..fc8500c708 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakRuleTerminalNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakRuleTerminalNode.java
@@ -44,6 +44,13 @@ import org.kie.api.event.rule.MatchCancelledCause;
* To change this template use File | Settings | File Templates.
*/
public class PhreakRuleTerminalNode {
+
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakRuleTerminalNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
public void doNode(ActivationsManager activationsManager,
RuleExecutor executor,
TerminalNode rtnNode,
@@ -111,7 +118,7 @@ public class PhreakRuleTerminalNode {
activationsManager.createAgendaItem( leftTuple, salienceInt, pctx, ruleAgendaItem, ruleAgendaItem.getAgendaGroup() );
- activationsManager.getAgendaEventSupport().fireActivationCreated(leftTuple, activationsManager.getReteEvaluator());
+ activationsManager.getAgendaEventSupport().fireActivationCreated(leftTuple, reteEvaluator);
if ( rtnNode.getRule().isLockOnActive() &&
leftTuple.getPropagationContext().getType() != PropagationContext.Type.RULE_ADDITION ) {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakSubnetworkNotExistsNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakSubnetworkNotExistsNode.java
index 43b6ddbcea..f96fce139a 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakSubnetworkNotExistsNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakSubnetworkNotExistsNode.java
@@ -25,12 +25,14 @@ import org.drools.core.reteoo.BetaNode;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.core.reteoo.SubnetworkTuple;
import org.drools.core.reteoo.Tuple;
-import org.drools.core.reteoo.TupleFactory;
import org.drools.core.reteoo.TupleImpl;
import org.drools.core.reteoo.TupleMemory;
import org.drools.core.util.index.TupleList;
import static org.drools.core.phreak.PhreakJoinNode.updateChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.unlinkAndDeleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
+import static org.drools.core.reteoo.TupleFactory.createLeftTuple;
public class PhreakSubnetworkNotExistsNode {
public static void doSubNetworkNode(BetaNode node,
@@ -96,9 +98,9 @@ public class PhreakSubnetworkNotExistsNode {
if (node.getType() == NodeTypeEnums.ExistsNode) {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
childLeftTuple.setPropagationContext(rightTuple.getPropagationContext());
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
} else if (!leftTuple.isExpired()) { // else !exists
- trgLeftTuples.addInsert(TupleFactory.createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), true));
+ trgLeftTuples.addInsert(createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), true));
}
}
rightTuple.clearStaged();
@@ -122,7 +124,7 @@ public class PhreakSubnetworkNotExistsNode {
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
- boolean useTupleMemory = tupleMemoryEnabled || RuleNetworkEvaluator.useLeftMemory(node, leftTuple);
+ boolean useTupleMemory = tupleMemoryEnabled || useLeftMemory(node, leftTuple);
// Do not need to init tupleList for matches, as this is done on right inserts.
if (useTupleMemory) {
ltm.add(leftTuple);
@@ -133,7 +135,7 @@ public class PhreakSubnetworkNotExistsNode {
TupleList matches = (TupleList) leftTuple.getContextObject();
if (matches == null && !leftTuple.isExpired()) {
// By adding the child now, it avoid iterating again to find all leftTuples that have no matches
- trgLeftTuples.addInsert(TupleFactory.createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), useTupleMemory));
+ trgLeftTuples.addInsert(createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), useTupleMemory));
}
}
@@ -161,14 +163,14 @@ public class PhreakSubnetworkNotExistsNode {
// Not/Exists nodes only have one child
if (node.getType() == NodeTypeEnums.ExistsNode) {
if (!leftTuple.isExpired()) {
- boolean useTupleMemory = tupleMemoryEnabled || RuleNetworkEvaluator.useLeftMemory(node, rightTuple);
- trgLeftTuples.addInsert(TupleFactory.createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), useTupleMemory));
+ boolean useTupleMemory = tupleMemoryEnabled || useLeftMemory(node, rightTuple);
+ trgLeftTuples.addInsert(createLeftTuple(leftTuple, sink, leftTuple.getPropagationContext(), useTupleMemory));
}
} else { // else !exists
TupleImpl childLeftTuple = leftTuple.getFirstChild();
if (childLeftTuple != null) { // this can be null if the LT is not yet added
childLeftTuple.setPropagationContext(rightTuple.getPropagationContext());
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
}
}
@@ -189,7 +191,7 @@ public class PhreakSubnetworkNotExistsNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild();
if (childLeftTuple != null) {
childLeftTuple.setPropagationContext(leftTuple.getPropagationContext());
- RuleNetworkEvaluator.unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
+ unlinkAndDeleteChildLeftTuple(trgLeftTuples, stagedLeftTuples, childLeftTuple);
}
leftTuple.setContextObject(null); // this is now right delete knows the LT is also being removed
diff --git a/drools-core/src/main/java/org/drools/core/phreak/PhreakTimerNode.java b/drools-core/src/main/java/org/drools/core/phreak/PhreakTimerNode.java
index b0d503b900..0c46f07294 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/PhreakTimerNode.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/PhreakTimerNode.java
@@ -21,7 +21,6 @@ package org.drools.core.phreak;
import java.util.Date;
import java.util.List;
-import org.drools.base.common.NetworkNode;
import org.drools.base.time.JobHandle;
import org.drools.base.time.Trigger;
import org.drools.base.time.impl.Timer;
@@ -34,7 +33,6 @@ import org.drools.core.common.WorkingMemoryAction;
import org.drools.core.marshalling.TupleKey;
import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
-import org.drools.core.reteoo.LeftTupleSource;
import org.drools.core.reteoo.PathMemory;
import org.drools.core.reteoo.SegmentMemory;
import org.drools.core.reteoo.TimerNode;
@@ -52,12 +50,18 @@ import org.kie.api.runtime.conf.TimedRuleExecutionFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.drools.core.phreak.BuildtimeSegmentUtilities.nextNodePosMask;
-import static org.drools.core.phreak.RuleNetworkEvaluator.normalizeStagedTuples;
+import static org.drools.core.phreak.PhreakNodeOperations.deleteChildLeftTuple;
+import static org.drools.core.phreak.PhreakNodeOperations.normalizeStagedTuples;
public class PhreakTimerNode {
private static final Logger log = LoggerFactory.getLogger( PhreakTimerNode.class );
+ private final ReteEvaluator reteEvaluator;
+
+ public PhreakTimerNode(ReteEvaluator reteEvaluator) {
+ this.reteEvaluator = reteEvaluator;
+ }
+
public void doNode(TimerNode timerNode,
TimerNodeMemory tm,
PathMemory pmem,
@@ -69,7 +73,7 @@ public class PhreakTimerNode {
TupleSets stagedLeftTuples) {
if ( srcLeftTuples.getDeleteFirst() != null ) {
- doLeftDeletes( timerNode, tm, pmem, sink, activationsManager, srcLeftTuples, trgLeftTuples, stagedLeftTuples );
+ doLeftDeletes( timerNode, tm, pmem, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples );
}
if ( srcLeftTuples.getUpdateFirst() != null ) {
@@ -94,10 +98,10 @@ public class PhreakTimerNode {
TupleSets srcLeftTuples,
TupleSets trgLeftTuples) {
Timer timer = timerNode.getTimer();
- TimerService timerService = activationsManager.getReteEvaluator().getTimerService();
+ TimerService timerService = reteEvaluator.getTimerService();
long timestamp = timerService.getCurrentTime();
String[] calendarNames = timerNode.getCalendarNames();
- Calendars calendars = activationsManager.getReteEvaluator().getCalendars();
+ Calendars calendars = reteEvaluator.getCalendars();
for (TupleImpl leftTuple = srcLeftTuples.getInsertFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
@@ -121,10 +125,10 @@ public class PhreakTimerNode {
Timer timer = timerNode.getTimer();
// Variables may have changed for ExpressionIntervalTimer, so it must be rescheduled
- TimerService timerService = activationsManager.getReteEvaluator().getTimerService();
+ TimerService timerService = reteEvaluator.getTimerService();
long timestamp = timerService.getCurrentTime();
String[] calendarNames = timerNode.getCalendarNames();
- Calendars calendars = activationsManager.getReteEvaluator().getCalendars();
+ Calendars calendars = reteEvaluator.getCalendars();
for ( TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; ) {
TupleImpl next = leftTuple.getStagedNext();
@@ -145,11 +149,10 @@ public class PhreakTimerNode {
TimerNodeMemory tm,
PathMemory pmem,
LeftTupleSink sink,
- ActivationsManager activationsManager,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
- TimerService timerService = activationsManager.getReteEvaluator().getTimerService();
+ TimerService timerService = reteEvaluator.getTimerService();
TupleList leftTuples = tm.getInsertOrUpdateLeftTuples();
TupleList deletes = tm.getDeleteLeftTuples();
@@ -194,7 +197,7 @@ public class PhreakTimerNode {
TupleImpl childLeftTuple = leftTuple.getFirstChild(); // only has one child
if ( childLeftTuple != null ) {
childLeftTuple.setPropagationContext( leftTuple.getPropagationContext() );
- RuleNetworkEvaluator.deleteChildLeftTuple( childLeftTuple, trgLeftTuples, stagedLeftTuples );
+ deleteChildLeftTuple(childLeftTuple, trgLeftTuples, stagedLeftTuples);
if ( log.isTraceEnabled() ) {
log.trace( "Timer Delete {}", leftTuple );
}
@@ -220,23 +223,22 @@ public class PhreakTimerNode {
final TupleImpl leftTuple,
final TupleSets trgLeftTuples,
final TupleSets stagedLeftTuples) {
- ReteEvaluator reteEvaluator = activationsManager.getReteEvaluator();
if ( leftTuple.getPropagationContext().getReaderContext() == null ) {
- final Trigger trigger = createTrigger( timerNode, reteEvaluator, timer, timestamp, calendarNames, calendars, leftTuple );
+ final Trigger trigger = createTrigger( timerNode, timer, timestamp, calendarNames, calendars, leftTuple );
// regular propagation
- scheduleTimer( timerNode, tm, smem, sink, reteEvaluator, timerService, timestamp, leftTuple, trgLeftTuples, stagedLeftTuples, trigger );
+ scheduleTimer( timerNode, tm, smem, sink, timerService, timestamp, leftTuple, trgLeftTuples, stagedLeftTuples, trigger );
} else {
// de-serializing, so we need to correlate timers before scheduling them
Scheduler scheduler = new Scheduler() {
@Override
public void schedule( Trigger t ) {
- scheduleTimer( timerNode, tm, smem, sink, reteEvaluator, timerService, timestamp, leftTuple, trgLeftTuples, stagedLeftTuples, t );
- evaluate( pmem, activationsManager, sink, tm, trgLeftTuples );
+ scheduleTimer( timerNode, tm, smem, sink, timerService, timestamp, leftTuple, trgLeftTuples, stagedLeftTuples, t );
+ reteEvaluator.getRuleNetworkEvaluator().evaluate(pmem, activationsManager, sink, tm, trgLeftTuples);
}
@Override
public Trigger getTrigger() {
- return createTrigger( timerNode, reteEvaluator, timer, timestamp, calendarNames, calendars, leftTuple );
+ return createTrigger( timerNode, timer, timestamp, calendarNames, calendars, leftTuple );
}
};
TupleKey key = TupleKey.createTupleKey( leftTuple );
@@ -246,7 +248,6 @@ public class PhreakTimerNode {
}
private Trigger createTrigger(final TimerNode timerNode,
- final ReteEvaluator reteEvaluator,
final Timer timer,
final long timestamp,
final String[] calendarNames,
@@ -266,7 +267,6 @@ public class PhreakTimerNode {
TimerNodeMemory tm,
SegmentMemory smem,
LeftTupleSink sink,
- ReteEvaluator reteEvaluator,
TimerService timerService,
long timestamp,
TupleImpl leftTuple,
@@ -348,7 +348,7 @@ public class PhreakTimerNode {
// This childLeftTuple has been created in this doNode loop, just skip it
childLeftTuple.setContextObject( null );
} else {
- normalizeStagedTuples( stagedLeftTuples, childLeftTuple );
+ normalizeStagedTuples(stagedLeftTuples, childLeftTuple);
trgLeftTuples.addUpdate( childLeftTuple );
if ( log.isTraceEnabled() ) {
log.trace( "Timer Update {}", childLeftTuple );
@@ -431,32 +431,6 @@ public class PhreakTimerNode {
}
}
- private static void evaluate(PathMemory pmem,
- ActivationsManager activationsManager,
- LeftTupleSink sink,
- TimerNodeMemory tm,
- TupleSets trgLeftTuples) {
- SegmentMemory[] smems = pmem.getSegmentMemories();
- SegmentMemory sm = tm.getSegmentMemory();
- int smemIndex = 0;
- for (SegmentMemory smem : smems) {
- if (smem == sm) {
- break;
- }
- smemIndex++;
- }
-
- long bit = 1;
- for (NetworkNode node = sm.getRootNode(); node != sink; node = ((LeftTupleSource)node).getSinkPropagator().getFirstLeftTupleSink() ) {
- //update the bit to the correct node position.
- bit = nextNodePosMask(bit);
- }
-
- RuleNetworkEvaluator.INSTANCE.outerEval(activationsManager, pmem.getRuleAgendaItem().getRuleExecutor(), pmem, smems,
- smemIndex, bit, tm,
- sink, trgLeftTuples, true);
- }
-
public static class TimerNodeJobContext
implements
JobContext {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/ReactiveObjectUtil.java b/drools-core/src/main/java/org/drools/core/phreak/ReactiveObjectUtil.java
index dd10993c10..38ab793818 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/ReactiveObjectUtil.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/ReactiveObjectUtil.java
@@ -36,6 +36,7 @@ import org.drools.core.reteoo.TupleImpl;
import static org.drools.core.phreak.PhreakFromNode.deleteChildLeftTuple;
import static org.drools.core.phreak.PhreakFromNode.isAllowed;
import static org.drools.core.phreak.PhreakFromNode.propagate;
+import static org.drools.core.phreak.PhreakNodeOperations.useLeftMemory;
public class ReactiveObjectUtil {
@@ -101,7 +102,7 @@ public class ReactiveObjectUtil {
betaConstraints,
propagationContext,
context,
- RuleNetworkEvaluator.useLeftMemory( node, leftTuple ),
+ useLeftMemory(node, leftTuple),
mem.getStagedLeftTuples(),
null );
} else {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/RuleExecutor.java b/drools-core/src/main/java/org/drools/core/phreak/RuleExecutor.java
index 812084485d..26e3dea77a 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/RuleExecutor.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/RuleExecutor.java
@@ -227,7 +227,7 @@ public class RuleExecutor {
}
public void evaluateNetwork(ActivationsManager activationsManager) {
- RuleNetworkEvaluator.INSTANCE.evaluateNetwork( activationsManager, this, pmem );
+ activationsManager.getReteEvaluator().getRuleNetworkEvaluator().evaluateNetwork( activationsManager, this, pmem );
setDirty( false );
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java b/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java
index 7f27978925..5dcc47f39b 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java
@@ -21,1011 +21,45 @@ package org.drools.core.phreak;
import java.util.List;
import org.drools.base.common.NetworkNode;
-import org.drools.base.reteoo.NodeTypeEnums;
-import org.drools.core.base.DroolsQueryImpl;
import org.drools.core.common.ActivationsManager;
-import org.drools.core.common.BetaConstraints;
import org.drools.core.common.Memory;
-import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
-import org.drools.core.common.TupleSetsImpl;
-import org.drools.core.reteoo.AbstractTerminalNode;
-import org.drools.core.reteoo.AccumulateNode;
-import org.drools.core.reteoo.AccumulateNode.AccumulateMemory;
-import org.drools.core.reteoo.AsyncReceiveNode;
-import org.drools.core.reteoo.AsyncReceiveNode.AsyncReceiveMemory;
-import org.drools.core.reteoo.AsyncSendNode;
-import org.drools.core.reteoo.AsyncSendNode.AsyncSendMemory;
-import org.drools.core.reteoo.BetaMemory;
-import org.drools.core.reteoo.BetaNode;
-import org.drools.core.reteoo.ConditionalBranchNode;
-import org.drools.core.reteoo.ConditionalBranchNode.ConditionalBranchMemory;
-import org.drools.core.reteoo.EvalConditionNode;
-import org.drools.core.reteoo.EvalConditionNode.EvalMemory;
-import org.drools.core.reteoo.ExistsNode;
-import org.drools.core.reteoo.FromNode;
-import org.drools.core.reteoo.FromNode.FromMemory;
-import org.drools.core.reteoo.JoinNode;
-import org.drools.core.reteoo.LeftInputAdapterNode;
-import org.drools.core.reteoo.LeftTuple;
-import org.drools.core.reteoo.LeftTupleNode;
-import org.drools.core.reteoo.LeftTupleSink;
-import org.drools.core.reteoo.LeftTupleSinkNode;
-import org.drools.core.reteoo.LeftTupleSource;
-import org.drools.core.reteoo.NotNode;
-import org.drools.core.reteoo.ObjectSink;
import org.drools.core.reteoo.PathMemory;
-import org.drools.core.reteoo.QueryElementNode;
-import org.drools.core.reteoo.QueryElementNode.QueryElementNodeMemory;
-import org.drools.core.reteoo.QueryTerminalNode;
-import org.drools.core.reteoo.ReactiveFromNode;
-import org.drools.core.reteoo.RightInputAdapterNode;
-import org.drools.core.reteoo.RightTuple;
import org.drools.core.reteoo.SegmentMemory;
-import org.drools.core.reteoo.SubnetworkTuple;
-import org.drools.core.reteoo.TerminalNode;
-import org.drools.core.reteoo.TimerNode;
-import org.drools.core.reteoo.TimerNode.TimerNodeMemory;
-import org.drools.core.reteoo.Tuple;
-import org.drools.core.reteoo.TupleFactory;
import org.drools.core.reteoo.TupleImpl;
-import org.drools.core.reteoo.TupleMemory;
-import org.drools.core.reteoo.TupleToObjectNode;
-import org.drools.core.reteoo.TupleToObjectNode.SubnetworkPathMemory;
-import org.drools.core.util.FastIterator;
-import org.drools.core.util.LinkedList;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import static org.drools.core.phreak.BuildtimeSegmentUtilities.nextNodePosMask;
-import static org.drools.core.phreak.PhreakNotNode.updateBlockersAndPropagate;
-import static org.drools.core.reteoo.BetaNode.getBetaMemory;
+public interface RuleNetworkEvaluator {
-public class RuleNetworkEvaluator {
+ void evaluateNetwork(RuleExecutor executor,
+ PathMemory pmem);
- private static final Logger log = LoggerFactory.getLogger(RuleNetworkEvaluator.class);
+ void evaluateNetwork(ActivationsManager activationsManager,
+ RuleExecutor executor,
+ PathMemory pmem);
- public static final RuleNetworkEvaluator INSTANCE = new RuleNetworkEvaluator();
-
- private static final PhreakJoinNode pJoinNode = PhreakNetworkNodeFactory.Factory.get().createPhreakJoinNode();
- private static final PhreakEvalNode pEvalNode = PhreakNetworkNodeFactory.Factory.get().createPhreakEvalNode();
- private static final PhreakFromNode pFromNode = PhreakNetworkNodeFactory.Factory.get().createPhreakFromNode();
- private static final PhreakReactiveFromNode pReactiveFromNode = PhreakNetworkNodeFactory.Factory.get().createPhreakReactiveFromNode();
- private static final PhreakNotNode pNotNode = PhreakNetworkNodeFactory.Factory.get().createPhreakNotNode();
- private static final PhreakExistsNode pExistsNode = PhreakNetworkNodeFactory.Factory.get().createPhreakExistsNode();
- private static final PhreakAccumulateNode pAccNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAccumulateNode();
- private static final PhreakAccumulateNode pGroupByNode = PhreakNetworkNodeFactory.Factory.get().createPhreakGroupByNode();
- private static final PhreakBranchNode pBranchNode = PhreakNetworkNodeFactory.Factory.get().createPhreakBranchNode();
- private static final PhreakQueryNode pQueryNode = PhreakNetworkNodeFactory.Factory.get().createPhreakQueryNode();
- private static final PhreakTimerNode pTimerNode = PhreakNetworkNodeFactory.Factory.get().createPhreakTimerNode();
- private static final PhreakAsyncSendNode pSendNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAsyncSendNode();
- private static final PhreakAsyncReceiveNode pReceiveNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAsyncReceiveNode();
- private static final PhreakRuleTerminalNode pRtNode = PhreakNetworkNodeFactory.Factory.get().createPhreakRuleTerminalNode();
- private static final PhreakQueryTerminalNode pQtNode = PhreakNetworkNodeFactory.Factory.get().createPhreakQueryTerminalNode();
-
- private static int cycle = 0;
-
-
- private RuleNetworkEvaluator() { }
-
- public void evaluateNetwork(RuleExecutor executor,
- PathMemory pmem) {
- evaluateNetwork(pmem.getActualActivationsManager(), executor, pmem);
- }
-
- public void evaluateNetwork(ActivationsManager activationsManager,
- RuleExecutor executor,
- PathMemory pmem) {
- SegmentMemory[] smems = pmem.getSegmentMemories();
-
-
- SegmentMemory smem = smems[0];
- if (smem == null) {
- // if there's no first smem it's a pure alpha firing and then doesn't require any furthe evaluation
- return;
- }
-
- LeftInputAdapterNode liaNode = (LeftInputAdapterNode) smem.getRootNode();
-
- NetworkNode node;
- Memory nodeMem;
- boolean firstSegmentIsOnlyLia = liaNode == smem.getTipNode();
- if (firstSegmentIsOnlyLia) {
- // segment only has liaNode in it
- // nothing is staged in the liaNode, so skip to next segment
- smem = smems[1];
- node = smem.getRootNode();
- nodeMem = smem.getNodeMemories()[0];
- } else {
- // lia is in shared segment, so point to next node
- node = liaNode.getSinkPropagator().getFirstLeftTupleSink();
- nodeMem = smem.getNodeMemories()[1]; // skip the liaNode memory
- }
-
- TupleSets srcTuples = smem.getStagedLeftTuples();
- if (log.isTraceEnabled()) {
- log.trace("Rule[name={}] segments={} {}", ((TerminalNode)pmem.getPathEndNode()).getRule().getName(), smems.length, srcTuples.toStringSizes());
- }
- outerEval(activationsManager, executor, pmem, smems, firstSegmentIsOnlyLia ? 1 : 0, firstSegmentIsOnlyLia ? 1L : 2L, nodeMem, node, srcTuples, true);
- }
-
- public void outerEval(ActivationsManager activationsManager,
- RuleExecutor executor,
- PathMemory pmem,
- SegmentMemory[] smems,
- int smemIndex,
- long bit,
- Memory nodeMem,
- NetworkNode node,
- TupleSets trgTuples,
- boolean processSubnetwork) {
-
- LinkedList<StackEntry> stack = new LinkedList<>();
- innerEval(activationsManager, executor, stack, pmem, smems, smemIndex, bit, nodeMem, node, trgTuples, processSubnetwork);
- while (!stack.isEmpty()) {
- // eval
- StackEntry entry = stack.removeLast();
- evalStackEntry(activationsManager, executor, stack, entry);
- }
- }
-
- public void evalStackEntry(ActivationsManager activationsManager,
- RuleExecutor executor,
- LinkedList<StackEntry> stack,
- StackEntry entry) {
- NetworkNode node = entry.getNode();
- Memory nodeMem = entry.getNodeMem();
- TupleSets trgTuples = entry.getTrgTuples();
- if (node.getType() == NodeTypeEnums.QueryElementNode) {
- // copy across the results, if any from the query node memory
- QueryElementNodeMemory qmem = (QueryElementNodeMemory) nodeMem;
- qmem.setNodeCleanWithoutNotify();
- qmem.getResultLeftTuples().addTo(trgTuples);
- }
-
- LeftTupleSinkNode sink = entry.getSink();
- PathMemory pmem = entry.getRmem();
-
- SegmentMemory[] smems = entry.getSmems();
- int smemIndex = entry.getSmemIndex();
- boolean processSubnetwork = entry.isProcessSubnetwork();
-
- long bit = entry.getBit();
- if (entry.isResumeFromNextNode()) {
- SegmentMemory smem = smems[smemIndex];
- if (node != smem.getTipNode()) {
- // get next node and node memory in the segment
- LeftTupleSink nextSink = sink.getNextLeftTupleSinkNode();
- if (nextSink == null) {
- node = sink;
- } else {
- // there is a nested subnetwork, take out path
- node = nextSink;
- }
-
- nodeMem = nodeMem.getNext();
- bit = nextNodePosMask(bit); // update bit to new node
- } else {
- // Reached end of segment, start on new segment.
- SegmentPropagator.propagate(smem,
- trgTuples,
- activationsManager.getReteEvaluator());
- smem = smems[++smemIndex];
- trgTuples = smem.getStagedLeftTuples().takeAll();
- node = smem.getRootNode();
- nodeMem = smem.getNodeMemories()[0];
- bit = 1; // update bit to start of new segment
- }
- }
-
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} Resume {} {}", indent(offset), node.toString(), trgTuples.toStringSizes());
- }
- innerEval(activationsManager, executor, stack, pmem, smems, smemIndex, bit, nodeMem, node, trgTuples, processSubnetwork);
- }
-
- public void innerEval(ActivationsManager activationsManager,
- RuleExecutor executor,
- LinkedList<StackEntry> stack,
- PathMemory pmem,
- final SegmentMemory[] smems,
- int smemIndex,
- long bit,
- Memory nodeMem,
- NetworkNode node,
- TupleSets trgTuples,
- boolean processSubnetwork) {
- SegmentMemory smem = smems[smemIndex];
- TupleSets stagedLeftTuples = null;
- while (true) {
- TupleSets srcTuples = trgTuples; // previous target, is now the source
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} {} {} {}", indent(offset), ++cycle, node.toString(), srcTuples.toStringSizes());
- }
-
- if (!(NodeTypeEnums.isBetaNodeWithSubnetwork(node))) {
- // The engine cannot skip a TupleToObjectNode node, as the dirty might be several levels deep
- if (srcTuples.isEmpty() && smem.getDirtyNodeMask() == 0) {
- // empty sources and segment is not dirty, skip to non empty src tuples or dirty segment.
- boolean foundDirty = false;
- for (int i = ++smemIndex, length = smems.length; i < length; i++) {
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} Skip Segment {}", indent(offset), i-1);
- }
-
- // this is needed for subnetworks that feed into a parent network that has no right inputs,
- // and may not yet be initialized
- if (smem.isEmpty() && !NodeTypeEnums.isTerminalNode(smem.getTipNode())) {
- activationsManager.getReteEvaluator().getSegmentMemorySupport().createChildSegments(smem.getTipNode().getSinkPropagator(), smem);
- }
-
- smem = smems[i];
- bit = 1;
- srcTuples = smem.getStagedLeftTuples().takeAll();
- node = smem.getRootNode();
- nodeMem = smem.getNodeMemories()[0];
- if (!srcTuples.isEmpty() ||
- smem.getDirtyNodeMask() != 0 ||
- (NodeTypeEnums.isBetaNode(node) && ((BetaNode)node).getRightInput().inputIsTupleToObjectNode())) {
- // break if dirty or if we reach a subnetwork. It must break for subnetworks, so they can be searched.
- foundDirty = true;
- smemIndex = i;
- break;
- }
- }
- if (!foundDirty) {
- break;
- }
- }
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} Segment {}", indent(offset), smemIndex);
- log.trace("{} {} {} {}", indent(offset), cycle, node.toString(), srcTuples.toStringSizes());
- }
- }
-
- if (srcTuples.isEmpty()) {
- while ((smem.getDirtyNodeMask() & bit) == 0 && node != smem.getTipNode() && !NodeTypeEnums.isBetaNodeWithSubnetwork(node)) {
- //while ((dirtyMask & bit) == 0 && node != smem.getTipNode() && NodeTypeEnums.isBetaNodeWithoutSubnetwork(node)) {
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} Skip Node {}", indent(offset), node);
- }
- bit = nextNodePosMask(bit); // shift to check the next node
- node = ((LeftTupleSource) node).getSinkPropagator().getFirstLeftTupleSink();
- nodeMem = nodeMem.getNext();
- }
- }
-
- boolean terminalNode = true;
- switch (node.getType()) {
- case NodeTypeEnums.RuleTerminalNode:
- pRtNode.doNode(activationsManager, executor, (AbstractTerminalNode) node, srcTuples);
- break;
- case NodeTypeEnums.QueryTerminalNode:
- pQtNode.doNode(activationsManager, stack, (QueryTerminalNode) node, srcTuples);
- break;
- case NodeTypeEnums.TupleToObjectNode:
- doSubnetwork2(activationsManager.getReteEvaluator(), (TupleToObjectNode) node, srcTuples);
- break;
- default:
- terminalNode = false;
- }
- if (terminalNode) {
- break;
- }
-
- stagedLeftTuples = getTargetStagedLeftTuples(activationsManager.getReteEvaluator(), smem, node);
- LeftTupleSinkNode sink = ((LeftTupleSource) node).getSinkPropagator().getFirstLeftTupleSink();
-
- trgTuples = evalNode(activationsManager, executor, stack, pmem, smems, smem, smemIndex, bit, nodeMem, node, sink, srcTuples, stagedLeftTuples, processSubnetwork);
- if (trgTuples == null) {
- break; // Queries exists and has been placed StackEntry, and there are no current trgTuples to process
- }
-
- if (node != smem.getTipNode()) {
- // get next node and node memory in the segment
- node = sink;
- nodeMem = nodeMem.getNext();
- bit = nextNodePosMask(bit);
- } else {
- // Reached end of segment, start on new segment.
- smem.getFirst().getStagedLeftTuples().addAll(stagedLeftTuples); // must put back all the LTs
- // end of SegmentMemory, so we know that stagedLeftTuples is not null
- SegmentPropagator.propagate(smem, trgTuples, activationsManager.getReteEvaluator());
- bit = 1;
- smem = smems[++smemIndex];
- trgTuples = smem.getStagedLeftTuples().takeAll();
-
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} Segment {}", indent(offset), smemIndex);
- }
- node = smem.getRootNode();
- nodeMem = smem.getNodeMemories()[0];
- }
- processSubnetwork = true; // make sure it's reset, so ria nodes are processed
- }
-
- if (stagedLeftTuples != null && !stagedLeftTuples.isEmpty()) {
- smem.getFirst().getStagedLeftTuples().addAll(stagedLeftTuples); // must put back all the LTs
- }
- }
-
- public TupleSets evalNode(ActivationsManager activationsManager,
- RuleExecutor executor,
- LinkedList<StackEntry> stack,
- PathMemory pmem,
- SegmentMemory[] smems,
- SegmentMemory smem,
- int smemIndex,
- long bit,
- Memory nodeMem,
- NetworkNode node,
- LeftTupleSinkNode sink,
- TupleSets srcTuples,
- TupleSets stagedLeftTuples,
- boolean processSubnetwork) {
- TupleSets trgTuples = new TupleSetsImpl();
- if (NodeTypeEnums.isBetaNode(node)) {
- boolean exitInnerEval = evalBetaNode(activationsManager, executor, stack, pmem, smems, smemIndex, nodeMem, node, sink, trgTuples, srcTuples, stagedLeftTuples, processSubnetwork);
- if (exitInnerEval) {
- return null;
- }
- } else {
- boolean exitInnerEval = false;
- switch (node.getType()) {
- case NodeTypeEnums.EvalConditionNode: {
- pEvalNode.doNode(activationsManager.getReteEvaluator(), (EvalConditionNode) node, (EvalMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples);
- break;
-
- }
- case NodeTypeEnums.FromNode: {
- pFromNode.doNode(activationsManager.getReteEvaluator(), (FromNode) node, (FromMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.ReactiveFromNode: {
- pReactiveFromNode.doNode(activationsManager.getReteEvaluator(), (ReactiveFromNode) node, (ReactiveFromNode.ReactiveFromMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.QueryElementNode: {
- exitInnerEval = evalQueryNode(activationsManager.getReteEvaluator(), stack, pmem, smems, smemIndex, bit, nodeMem,
- node, sink, trgTuples, srcTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.TimerConditionNode: {
- pTimerNode.doNode((TimerNode) node, (TimerNodeMemory) nodeMem, pmem, smem, sink, activationsManager, srcTuples, trgTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.ConditionalBranchNode: {
- pBranchNode.doNode(activationsManager, (ConditionalBranchNode) node, (ConditionalBranchMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples, executor);
- break;
- }
- case NodeTypeEnums.AsyncSendNode: {
- pSendNode.doNode(activationsManager.getReteEvaluator(), (AsyncSendNode) node, (AsyncSendMemory) nodeMem, srcTuples);
- break;
- }
- case NodeTypeEnums.AsyncReceiveNode: {
- pReceiveNode.doNode(activationsManager.getReteEvaluator(), (AsyncReceiveNode) node, (AsyncReceiveMemory) nodeMem, sink, srcTuples, trgTuples);
- break;
- }
- }
- if (exitInnerEval && trgTuples.isEmpty()) {
- return null;
- }
- }
- return trgTuples;
- }
-
- private static TupleSets getTargetStagedLeftTuples(ReteEvaluator reteEvaluator,
- SegmentMemory smem,
- NetworkNode node) {
- if (node == smem.getTipNode()) {
- // we are about to process the segment tip, allow it to merge insert/update/delete clashes
- if (smem.isEmpty()) {
- reteEvaluator.getSegmentMemorySupport().createChildSegments(((LeftTupleSource) node).getSinkPropagator(), smem);
- }
- return smem.getFirst().getStagedLeftTuples().takeAll();
- } else {
- return null;
- }
- }
-
- private boolean evalQueryNode(ReteEvaluator reteEvaluator,
- LinkedList<StackEntry> stack,
- PathMemory pmem,
- SegmentMemory[] smems,
- int smemIndex,
- long bit,
- Memory nodeMem,
- NetworkNode node,
- LeftTupleSinkNode sink,
- TupleSets trgTuples,
- TupleSets srcTuples,
- TupleSets stagedLeftTuples) {
- QueryElementNodeMemory qmem = (QueryElementNodeMemory) nodeMem;
-
- if (srcTuples.isEmpty() && qmem.getResultLeftTuples().isEmpty()) {
- // no point in evaluating query element, and setting up stack, if there is nothing to process
- return false;
- }
-
- QueryElementNode qnode = (QueryElementNode) node;
-
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} query result tuples {}", indent(offset), qmem.getResultLeftTuples().toStringSizes());
- }
-
- // result tuples can happen when reactivity occurs inside of the query, prior to evaluation
- // we will need special behaviour to add the results again, when this query result resumes
- qmem.getResultLeftTuples().addTo(trgTuples);
- qmem.setNodeCleanWithoutNotify();
-
- if (srcTuples.isEmpty()) {
- return false;
- }
- // only process the Query Node if there are src tuples
- StackEntry stackEntry = new StackEntry(node, bit, sink, pmem, nodeMem, smems,
- smemIndex, trgTuples, true, true);
-
- stack.add(stackEntry);
-
- pQueryNode.doNode(qnode, (QueryElementNodeMemory) nodeMem, stackEntry,
- reteEvaluator, srcTuples, trgTuples, stagedLeftTuples);
-
- SegmentMemory qsmem = ((QueryElementNodeMemory) nodeMem).getQuerySegmentMemory();
- List<PathMemory> qpmems = qsmem.getPathMemories();
-
- // Build the evaluation information for each 'or' branch
- for (int i = 0; i < qpmems.size() ; i++) {
- PathMemory qpmem = qpmems.get(i);
-
- pmem = qpmem;
- smems = qpmem.getSegmentMemories();
- smemIndex = 0;
- SegmentMemory smem = smems[smemIndex]; // 0
-
- LeftTupleNode liaNode = qpmem.getPathEndNode().getPathNodes()[0];
-
- if (liaNode == smem.getTipNode()) {
- // segment only has liaNode in it
- // nothing is staged in the liaNode, so skip to next segment
- smem = smems[++smemIndex]; // 1
- node = smem.getRootNode();
- nodeMem = smem.getNodeMemories()[0];
- bit = 1;
- } else {
- // lia is in shared segment, so point to next node
- node = liaNode.getSinkPropagator().getFirstLeftTupleSink();
- nodeMem = smem.getNodeMemories()[1]; // skip the liaNode memory
- bit = 2;
- }
-
- trgTuples = smem.getStagedLeftTuples().takeAll();
- stackEntry = new StackEntry(node, bit, null, pmem,
- nodeMem, smems, smemIndex,
- trgTuples, false, true);
- if (log.isTraceEnabled()) {
- int offset = getOffset(stackEntry.getNode());
- log.trace("{} ORQueue branch={} {} {}", indent(offset), i, stackEntry.getNode().toString(), trgTuples.toStringSizes());
- }
- stack.add(stackEntry);
- }
- return true;
-
- }
-
- private boolean evalBetaNode(ActivationsManager activationsManager,
- RuleExecutor executor,
- LinkedList<StackEntry> stack,
- PathMemory pmem,
- SegmentMemory[] smems,
- int smemIndex,
- Memory nodeMem,
- NetworkNode node,
- LeftTupleSinkNode sink,
- TupleSets trgTuples,
- TupleSets srcTuples,
- TupleSets stagedLeftTuples,
- boolean processSubnetwork) {
- BetaNode betaNode = (BetaNode) node;
- BetaMemory bm;
- AccumulateMemory am = null;
- if (NodeTypeEnums.AccumulateNode == node.getType()) {
- am = (AccumulateMemory) nodeMem;
- bm = am.getBetaMemory();
- } else {
- bm = (BetaMemory) nodeMem;
- }
-
- if (processSubnetwork && betaNode.getRightInput().inputIsTupleToObjectNode()) {
- // if the subnetwork is nested in this segment, it will create srcTuples containing
- // peer LeftTuples, suitable for the node in the main path.
- doSubnetwork(activationsManager, executor, stack,
- pmem, smems, smemIndex, nodeMem, srcTuples, bm, betaNode, sink);
- return true; // return here, TupleToObjectNode queues the evaluation on the stack, which is necessary to handled nested query nodes
- }
-
- switchOnDoBetaNode(activationsManager.getReteEvaluator(), bm, am, node, sink, trgTuples, srcTuples, stagedLeftTuples);
-
- return false;
- }
-
- private void switchOnDoBetaNode(ReteEvaluator reteEvaluator,
- BetaMemory bm,
- AccumulateMemory am,
- NetworkNode node,
- LeftTupleSinkNode sink,
- TupleSets trgTuples,
- TupleSets srcTuples,
- TupleSets stagedLeftTuples) {
- if (log.isTraceEnabled()) {
- int offset = getOffset(node);
- log.trace("{} rightTuples {}", indent(offset), bm.getStagedRightTuples().toStringSizes());
- }
-
- switch (node.getType()) {
- case NodeTypeEnums.JoinNode: {
- pJoinNode.doNode(reteEvaluator, (JoinNode) node, sink,
- bm, srcTuples, trgTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.NotNode: {
- pNotNode.doNode(reteEvaluator, (NotNode) node, sink,
- bm, srcTuples, trgTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.ExistsNode: {
- pExistsNode.doNode(reteEvaluator, (ExistsNode) node, sink,
- bm, srcTuples, trgTuples, stagedLeftTuples);
- break;
- }
- case NodeTypeEnums.AccumulateNode: {
- AccumulateNode accumulateNode = (AccumulateNode) node;
- if (accumulateNode.getAccumulate().isGroupBy()) {
- pGroupByNode.doNode(reteEvaluator, accumulateNode, sink, am, srcTuples, trgTuples, stagedLeftTuples);
- } else {
- pAccNode.doNode(reteEvaluator, accumulateNode, sink, am, srcTuples, trgTuples, stagedLeftTuples);
- }
- break;
- }
- }
- }
-
- private void doSubnetwork(ActivationsManager activationsManager,
- RuleExecutor executor,
- LinkedList<StackEntry> stack,
- PathMemory pmem,
- SegmentMemory[] smems,
- int smemIndex,
- Memory nodeMem,
- TupleSets srcTuples,
- BetaMemory bm,
- BetaNode betaNode,
- LeftTupleSinkNode sink) {
- SubnetworkPathMemory pathMem = bm.getSubnetworkPathMemory();
- SegmentMemory[] subnetworkSmems = pathMem.getSegmentMemories();
- SegmentMemory subSmem = null;
- for (int i = 0; subSmem == null; i++) {
- // segment positions outside of the subnetwork, in the parent chain, are null
- // so we must iterate to find the first non null segment memory
- subSmem = subnetworkSmems[i];
- }
-
- // Resume the node after the TupleToObjectNode segment has been processed and the right input memory populated
- StackEntry stackEntry = new StackEntry(betaNode, bm.getNodePosMaskBit(), sink, pmem, nodeMem, smems,
- smemIndex, srcTuples, false, false);
- stack.add(stackEntry);
- if (log.isTraceEnabled()) {
- int offset = getOffset(betaNode);
- log.trace("{} SubnetworkQueue {} {}", indent(offset), betaNode.toString(), srcTuples.toStringSizes());
- }
-
-
- TupleSets subLts = subSmem.getStagedLeftTuples().takeAll();
- // node is first in the segment, so bit is 1
- innerEval(activationsManager, executor, stack,
- pathMem,
- subnetworkSmems, subSmem.getPos(),
- 1, subSmem.getNodeMemories()[0], subSmem.getRootNode(), subLts, true);
- }
-
- private void doSubnetwork2(ReteEvaluator reteEvaluator,
- TupleToObjectNode tton,
- TupleSets srcTuples) {
-
- ObjectSink[] sinks = tton.getObjectSinkPropagator().getSinks();
-
- BetaNode betaNode = ((RightInputAdapterNode)sinks[0]).getBetaNode();
- BetaMemory bm;
- Memory nodeMem = reteEvaluator.getNodeMemory(betaNode);
- if (NodeTypeEnums.AccumulateNode == betaNode.getType()) {
- bm = ((AccumulateMemory) nodeMem).getBetaMemory();
- } else {
- bm = (BetaMemory) nodeMem;
- }
- TupleSets rightTuples = bm.getStagedRightTuples();
-
- // Build up iteration array for other sinks
- BetaNode[] bns = null;
- BetaMemory[] bms = null;
- int length = sinks.length;
- if (length > 1) {
- bns = new BetaNode[sinks.length - 1];
- bms = new BetaMemory[sinks.length - 1];
- for (int i = 1; i < length; i++) {
- bns[i - 1] = ((RightInputAdapterNode)sinks[i]).getBetaNode();
- Memory nodeMem2 = reteEvaluator.getNodeMemory(bns[i - 1]);
- if (NodeTypeEnums.AccumulateNode == betaNode.getType()) {
- bms[i - 1] = ((AccumulateMemory) nodeMem2).getBetaMemory();
- } else {
- bms[i - 1] = (BetaMemory) nodeMem2;
- }
- }
- }
-
- length--; // subtract one, as first is not in the array;
- for (SubnetworkTuple subnetworkTuple = (SubnetworkTuple) srcTuples.getInsertFirst(); subnetworkTuple != null;) {
- SubnetworkTuple next = (SubnetworkTuple) subnetworkTuple.getStagedNext();
-
- if (bm.getStagedRightTuples().isEmpty()) {
- bm.setNodeDirtyWithoutNotify();
- }
-
- subnetworkTuple.prepareStagingOnRight();
- rightTuples.addInsert(subnetworkTuple);
-
- if (bns != null) {
- for (int i = 0; i < length; i++) {
- if (bms[i].getStagedRightTuples().isEmpty()) {
- bms[i].setNodeDirtyWithoutNotify();
- }
- subnetworkTuple = (SubnetworkTuple) TupleFactory.createPeer(tton,
- subnetworkTuple);
- bms[i].getStagedRightTuples().addInsert(subnetworkTuple);
- }
- }
- subnetworkTuple = next;
- }
-
- for (SubnetworkTuple subnetworkTuple = (SubnetworkTuple) srcTuples.getDeleteFirst(); subnetworkTuple != null;) {
- SubnetworkTuple next = (SubnetworkTuple) subnetworkTuple.getStagedNext();
-
- if (rightTuples.isEmpty()) {
- bm.setNodeDirtyWithoutNotify();
- }
-
- switch (subnetworkTuple.getStagedTypeOnRight()) {
- // handle clash with already staged entries
- case Tuple.INSERT:
- rightTuples.removeInsert(subnetworkTuple.moveStagingFromLeftToRight());
- break;
- case Tuple.UPDATE:
- rightTuples.removeUpdate(subnetworkTuple.moveStagingFromLeftToRight());
- break;
- }
-
- subnetworkTuple.prepareStagingOnRight();
- rightTuples.addDelete(subnetworkTuple);
-
- if (bns != null) {
- for (int i = 0; i < length; i++) {
- subnetworkTuple = (SubnetworkTuple) subnetworkTuple.getPeer();
- if (bms[i].getStagedRightTuples().isEmpty()) {
- bms[i].setNodeDirtyWithoutNotify();
- }
- bms[i].getStagedRightTuples().addDelete(subnetworkTuple);
- subnetworkTuple.setStagedOnRight();
- }
- }
- subnetworkTuple = next;
- }
-
- for (SubnetworkTuple subnetworkTuple = (SubnetworkTuple) srcTuples.getUpdateFirst(); subnetworkTuple != null;) {
- SubnetworkTuple next = (SubnetworkTuple) subnetworkTuple.getStagedNext();
-
- if (rightTuples.isEmpty()) {
- bm.setNodeDirtyWithoutNotify();
- }
-
- subnetworkTuple.prepareStagingOnRight();
- rightTuples.addUpdate(subnetworkTuple);
-
- if (bns != null) {
- for (int i = 0; i < length; i++) {
- subnetworkTuple = (SubnetworkTuple) subnetworkTuple.getPeer();
-
- if (bms[i].getStagedRightTuples().isEmpty()) {
- bms[i].setNodeDirtyWithoutNotify();
- }
- bms[i].getStagedRightTuples().addUpdate(subnetworkTuple);
- subnetworkTuple.setStagedOnRight();
- }
- }
- subnetworkTuple = next;
- }
- srcTuples.resetAll();
- }
-
- public static void findLeftTupleBlocker(BetaNode betaNode,
- TupleMemory rtm,
- Object contextEntry,
- BetaConstraints constraints,
- TupleImpl leftTuple,
- boolean useLeftMemory) {
- // This method will also remove rightTuples that are from subnetwork where no leftmemory use used
- FastIterator it = betaNode.getRightIterator(rtm);
- for (RightTuple rightTuple = betaNode.getFirstRightTuple(leftTuple, rtm, it); rightTuple != null;) {
- RightTuple nextRight = (RightTuple) it.next(rightTuple);
- if (constraints.isAllowedCachedLeft(contextEntry,
- rightTuple.getFactHandleForEvaluation())) {
- leftTuple.setBlocker(rightTuple);
-
- if (useLeftMemory) {
- rightTuple.addBlocked((LeftTuple) leftTuple);
- break;
- } else if (betaNode.getRightInput().inputIsTupleToObjectNode()) {
- // If we aren't using leftMemory and the right input is a TupleToObjectNode, then we must iterate and find all subetwork right tuples and remove them
- // so we don't break
- rtm.remove(rightTuple);
- } else {
- break;
- }
- }
- rightTuple = nextRight;
- }
- }
-
-
- public static void unlinkAndDeleteChildLeftTuple(TupleSets trgLeftTuples,
- TupleSets stagedLeftTuples,
- TupleImpl childLeftTuple) {
- childLeftTuple.unlinkFromRightParent();
- childLeftTuple.unlinkFromLeftParent();
- deleteChildLeftTuple(childLeftTuple, trgLeftTuples, stagedLeftTuples);
- }
-
- public static void deleteChildLeftTuple(TupleImpl childLeftTuple, TupleSets trgLeftTuples, TupleSets stagedLeftTuples) {
- if (childLeftTuple.isStagedOnRight()) {
- ((SubnetworkTuple) childLeftTuple).moveStagingFromRightToLeft();
- } else {
- switch (childLeftTuple.getStagedType()) {
- // handle clash with already staged entries
- case Tuple.INSERT:
- stagedLeftTuples.removeInsert(childLeftTuple);
- trgLeftTuples.addNormalizedDelete(childLeftTuple);
- return;
- case Tuple.UPDATE:
- stagedLeftTuples.removeUpdate(childLeftTuple);
- break;
- }
- }
- trgLeftTuples.addDelete(childLeftTuple);
- }
-
- public static void doUpdatesReorderLeftMemory(BetaMemory bm,
- TupleSets srcLeftTuples) {
- TupleMemory ltm = bm.getLeftTupleMemory();
-
- // sides must first be re-ordered, to ensure iteration integrity
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- if (leftTuple.getMemory() != null) {
- ltm.remove(leftTuple);
- }
- }
-
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- ltm.add(leftTuple);
- for (TupleImpl childLeftTuple = leftTuple.getFirstChild(); childLeftTuple != null;) {
- TupleImpl childNext = childLeftTuple.getHandleNext();
- childLeftTuple.reAddRight();
- childLeftTuple = childNext;
- }
- }
- }
-
- public static void doUpdatesExistentialReorderLeftMemory(BetaMemory bm,
- TupleSets srcLeftTuples) {
- TupleMemory ltm = bm.getLeftTupleMemory();
-
- // sides must first be re-ordered, to ensure iteration integrity
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- if (leftTuple.getMemory() != null) {
- ltm.remove(leftTuple);
- }
- }
-
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- RightTuple blocker = leftTuple.getBlocker();
- if (blocker == null) {
- ltm.add(leftTuple);
- for (TupleImpl childLeftTuple = leftTuple.getFirstChild(); childLeftTuple != null;) {
- TupleImpl childNext = childLeftTuple.getHandleNext();
- childLeftTuple.reAddRight();
- childLeftTuple = childNext;
- }
- } else if (blocker.getStagedType() != LeftTuple.NONE) {
- // it's blocker is also being updated, so remove to force it to start from the beginning
- blocker.removeBlocked((LeftTuple) leftTuple);
- }
- }
- }
-
- public static void doUpdatesReorderRightMemory(BetaMemory bm,
- TupleSets srcRightTuples) {
- TupleMemory rtm = bm.getRightTupleMemory();
-
- for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- if (rightTuple.getMemory() != null) {
- rtm.removeAdd(rightTuple);
- doUpdatesReorderChildLeftTuple(rightTuple);
- }
- }
- }
-
- public static void doUpdatesReorderChildLeftTuple(TupleImpl rightTuple) {
- for (TupleImpl childLeftTuple = rightTuple.getFirstChild(); childLeftTuple != null;) {
- TupleImpl childNext = childLeftTuple.getRightParentNext();
- childLeftTuple.reAddLeft();
- childLeftTuple = childNext;
- }
- }
-
- public static void doUpdatesExistentialReorderRightMemory(BetaMemory bm,
- BetaNode betaNode,
- TupleSets srcRightTuples) {
- TupleMemory rtm = bm.getRightTupleMemory();
-
- boolean resumeFromCurrent = !(betaNode.isIndexedUnificationJoin() || rtm.getIndexType().isComparison());
-
- // remove all the staged rightTuples from the memory before to readd them all
- // this is to avoid split bucket when an updated rightTuple hasn't been moved yet
- // and so it is the first entry in the wrong bucket
-
- if (rtm.getIndexType() != TupleMemory.IndexType.NONE) {
- for (TupleImpl rightTuple = srcRightTuples.getDeleteFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- rtm.remove(rightTuple);
- }
- }
-
- for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- doRemoveExistentialRightMemoryForReorder(rtm, (RightTuple) rightTuple, resumeFromCurrent);
- }
-
- for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- doAddExistentialRightMemoryForReorder(rtm, (RightTuple) rightTuple, resumeFromCurrent);
- }
-
- if (rtm.getIndexType() != TupleMemory.IndexType.NONE) {
- for (TupleImpl rightTuple = srcRightTuples.getDeleteFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- rtm.add(rightTuple);
- }
- }
- }
-
- public static void doExistentialUpdatesReorderChildLeftTuple(ReteEvaluator reteEvaluator,
- NotNode notNode,
- RightTuple rightTuple) {
- BetaMemory bm = getBetaMemory(notNode, reteEvaluator);
- TupleMemory rtm = bm.getRightTupleMemory();
-
- boolean resumeFromCurrent = !(notNode.isIndexedUnificationJoin() || rtm.getIndexType().isComparison());
- doRemoveExistentialRightMemoryForReorder(rtm, rightTuple, resumeFromCurrent);
- doAddExistentialRightMemoryForReorder(rtm, rightTuple, resumeFromCurrent);
-
- updateBlockersAndPropagate(notNode, rightTuple, reteEvaluator, rtm, bm.getContext(), notNode.getRawConstraints(), !resumeFromCurrent, null, null, null);
- }
-
- private static void doAddExistentialRightMemoryForReorder(TupleMemory rtm,
- RightTuple rightTuple,
- boolean resumeFromCurrent) {
- rtm.add(rightTuple);
-
- if (resumeFromCurrent) {
- if (rightTuple.getBlocked() != null && rightTuple.getTempNextRightTuple() == null) {
- // the next RightTuple was null, but current RightTuple was added back into the same bucket, so reset as root blocker to re-match can be attempted
- rightTuple.setTempNextRightTuple(rightTuple);
- }
- }
-
- doUpdatesReorderChildLeftTuple(rightTuple);
- }
-
- private static void doRemoveExistentialRightMemoryForReorder(TupleMemory rtm,
- RightTuple rightTuple,
- boolean resumeFromCurrent) {
- if (rightTuple.getMemory() != null) {
-
- if (resumeFromCurrent) {
- if (rightTuple.getBlocked() != null) {
- // look for a non-staged right tuple first forward ...
- RightTuple tempRightTuple = (RightTuple) rightTuple.getNext();
- while (tempRightTuple != null && tempRightTuple.getStagedType() != LeftTuple.NONE) {
- // next cannot be an updated or deleted rightTuple
- tempRightTuple = (RightTuple) tempRightTuple.getNext();
- }
-
- // ... and if cannot find one try backward
- if (tempRightTuple == null) {
- tempRightTuple = (RightTuple) rightTuple.getPrevious();
- while (tempRightTuple != null && tempRightTuple.getStagedType() != LeftTuple.NONE) {
- // next cannot be an updated or deleted rightTuple
- tempRightTuple = (RightTuple) tempRightTuple.getPrevious();
- }
- }
-
- rightTuple.setTempNextRightTuple(tempRightTuple);
- }
- }
-
- rightTuple.setTempBlocked(rightTuple.getBlocked());
- rightTuple.setBlocked(null);
- rtm.remove(rightTuple);
- }
- }
-
- public static boolean useLeftMemory(LeftTupleSource tupleSource,
- TupleImpl tuple) {
- boolean useLeftMemory = true;
- if (!tupleSource.isLeftTupleMemoryEnabled()) {
- // This is a hack, to not add closed DroolsQuery objects
- Object object = tuple.getRootTuple().getFactHandle().getObject();
- if (!(object instanceof DroolsQueryImpl) || !((DroolsQueryImpl) object).isOpen()) {
- useLeftMemory = false;
- }
- }
- return useLeftMemory;
- }
-
- public static void normalizeStagedTuples(TupleSets stagedLeftTuples,
- TupleImpl childLeftTuple) {
- if (!childLeftTuple.isStagedOnRight()) {
- switch (childLeftTuple.getStagedType()) {
- // handle clash with already staged entries
- case LeftTuple.INSERT:
- stagedLeftTuples.removeInsert(childLeftTuple);
- break;
- case LeftTuple.UPDATE:
- stagedLeftTuples.removeUpdate(childLeftTuple);
- break;
- }
- }
- }
+ void evaluate(PathMemory pmem,
+ ActivationsManager activationsManager,
+ NetworkNode sink,
+ Memory tm,
+ TupleSets trgLeftTuples);
+ void forceFlushPath(PathMemory outPmem);
- private static String indent(int size) {
- StringBuilder sbuilder = new StringBuilder();
- for (int i = 0; i < size; i++) {
- sbuilder.append(" ");
- }
-
- return sbuilder.toString();
- }
-
- private static int getOffset(NetworkNode node) {
- LeftTupleSource lt;
- int offset = 1;
- if (NodeTypeEnums.isTerminalNode(node)) {
- lt = ((TerminalNode) node).getLeftTupleSource();
- offset++;
- } else if (node.getType() == NodeTypeEnums.TupleToObjectNode) {
- lt = ((TupleToObjectNode) node).getLeftTupleSource();
- } else {
- lt = (LeftTupleSource) node;
- }
- while (!NodeTypeEnums.isLeftInputAdapterNode(lt)) {
- offset++;
- lt = lt.getLeftTupleSource();
- }
+ void forceFlushLeftTuple(PathMemory pmem,
+ SegmentMemory sm,
+ TupleSets leftTupleSets);
+
+ void forceFlushWhenSubnetwork(PathMemory pmem);
+
+ public boolean flushLeftTupleIfNecessary(SegmentMemory sm, boolean streamMode);
+
+ public boolean flushLeftTupleIfNecessary(SegmentMemory sm,
+ TupleImpl leftTuple,
+ boolean streamMode,
+ short stagedType);
+
+ List<PathMemory> findPathsToFlushFromSubnetwork(PathMemory pmem);
+
+ void propagate(SegmentMemory sourceSegment, TupleSets leftTuples);
- return offset;
- }
}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java b/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluatorImpl.java
similarity index 63%
copy from drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java
copy to drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluatorImpl.java
index 7f27978925..4917b97de4 100644
--- a/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluator.java
+++ b/drools-core/src/main/java/org/drools/core/phreak/RuleNetworkEvaluatorImpl.java
@@ -18,15 +18,18 @@
*/
package org.drools.core.phreak;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
import java.util.List;
import org.drools.base.common.NetworkNode;
import org.drools.base.reteoo.NodeTypeEnums;
-import org.drools.core.base.DroolsQueryImpl;
import org.drools.core.common.ActivationsManager;
-import org.drools.core.common.BetaConstraints;
import org.drools.core.common.Memory;
+import org.drools.core.common.NodeMemories;
import org.drools.core.common.ReteEvaluator;
+import org.drools.core.common.SegmentMemorySupport;
import org.drools.core.common.TupleSets;
import org.drools.core.common.TupleSetsImpl;
import org.drools.core.reteoo.AbstractTerminalNode;
@@ -54,13 +57,13 @@ import org.drools.core.reteoo.LeftTupleSinkNode;
import org.drools.core.reteoo.LeftTupleSource;
import org.drools.core.reteoo.NotNode;
import org.drools.core.reteoo.ObjectSink;
+import org.drools.core.reteoo.PathEndNode;
import org.drools.core.reteoo.PathMemory;
import org.drools.core.reteoo.QueryElementNode;
import org.drools.core.reteoo.QueryElementNode.QueryElementNodeMemory;
import org.drools.core.reteoo.QueryTerminalNode;
import org.drools.core.reteoo.ReactiveFromNode;
import org.drools.core.reteoo.RightInputAdapterNode;
-import org.drools.core.reteoo.RightTuple;
import org.drools.core.reteoo.SegmentMemory;
import org.drools.core.reteoo.SubnetworkTuple;
import org.drools.core.reteoo.TerminalNode;
@@ -69,50 +72,69 @@ import org.drools.core.reteoo.TimerNode.TimerNodeMemory;
import org.drools.core.reteoo.Tuple;
import org.drools.core.reteoo.TupleFactory;
import org.drools.core.reteoo.TupleImpl;
-import org.drools.core.reteoo.TupleMemory;
import org.drools.core.reteoo.TupleToObjectNode;
import org.drools.core.reteoo.TupleToObjectNode.SubnetworkPathMemory;
-import org.drools.core.util.FastIterator;
import org.drools.core.util.LinkedList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import static org.drools.base.reteoo.NodeTypeEnums.AccumulateNode;
+import static org.drools.core.common.TupleSetsImpl.createLeftTupleTupleSets;
import static org.drools.core.phreak.BuildtimeSegmentUtilities.nextNodePosMask;
-import static org.drools.core.phreak.PhreakNotNode.updateBlockersAndPropagate;
-import static org.drools.core.reteoo.BetaNode.getBetaMemory;
-
-public class RuleNetworkEvaluator {
-
- private static final Logger log = LoggerFactory.getLogger(RuleNetworkEvaluator.class);
-
- public static final RuleNetworkEvaluator INSTANCE = new RuleNetworkEvaluator();
-
- private static final PhreakJoinNode pJoinNode = PhreakNetworkNodeFactory.Factory.get().createPhreakJoinNode();
- private static final PhreakEvalNode pEvalNode = PhreakNetworkNodeFactory.Factory.get().createPhreakEvalNode();
- private static final PhreakFromNode pFromNode = PhreakNetworkNodeFactory.Factory.get().createPhreakFromNode();
- private static final PhreakReactiveFromNode pReactiveFromNode = PhreakNetworkNodeFactory.Factory.get().createPhreakReactiveFromNode();
- private static final PhreakNotNode pNotNode = PhreakNetworkNodeFactory.Factory.get().createPhreakNotNode();
- private static final PhreakExistsNode pExistsNode = PhreakNetworkNodeFactory.Factory.get().createPhreakExistsNode();
- private static final PhreakAccumulateNode pAccNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAccumulateNode();
- private static final PhreakAccumulateNode pGroupByNode = PhreakNetworkNodeFactory.Factory.get().createPhreakGroupByNode();
- private static final PhreakBranchNode pBranchNode = PhreakNetworkNodeFactory.Factory.get().createPhreakBranchNode();
- private static final PhreakQueryNode pQueryNode = PhreakNetworkNodeFactory.Factory.get().createPhreakQueryNode();
- private static final PhreakTimerNode pTimerNode = PhreakNetworkNodeFactory.Factory.get().createPhreakTimerNode();
- private static final PhreakAsyncSendNode pSendNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAsyncSendNode();
- private static final PhreakAsyncReceiveNode pReceiveNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAsyncReceiveNode();
- private static final PhreakRuleTerminalNode pRtNode = PhreakNetworkNodeFactory.Factory.get().createPhreakRuleTerminalNode();
- private static final PhreakQueryTerminalNode pQtNode = PhreakNetworkNodeFactory.Factory.get().createPhreakQueryTerminalNode();
- private static int cycle = 0;
+public class RuleNetworkEvaluatorImpl implements RuleNetworkEvaluator {
+
+ private static final Logger log = LoggerFactory.getLogger(RuleNetworkEvaluatorImpl.class);
+
+ private final PhreakJoinNode pJoinNode;
+ private final PhreakEvalNode pEvalNode;
+ private final PhreakFromNode pFromNode;
+ private final PhreakReactiveFromNode pReactiveFromNode;
+ private final PhreakNotNode pNotNode;
+ private final PhreakExistsNode pExistsNode;
+ private final PhreakAccumulateNode pAccNode;
+ private final PhreakAccumulateNode pGroupByNode;
+ private final PhreakBranchNode pBranchNode;
+ private final PhreakQueryNode pQueryNode;
+ private final PhreakTimerNode pTimerNode;
+ private final PhreakAsyncSendNode pSendNode;
+ private final PhreakAsyncReceiveNode pReceiveNode;
+ private final PhreakRuleTerminalNode pRtNode;
+ private final PhreakQueryTerminalNode pQtNode;
+ private static int cycle = 0;
- private RuleNetworkEvaluator() { }
+ private NodeMemories nodeMemories;
+
+ private SegmentMemorySupport segmentMemorySupport;
+
+ public RuleNetworkEvaluatorImpl(ReteEvaluator reteEvaluator, NodeMemories nodeMemories, SegmentMemorySupport segmentMemorySupport) {
+ this.nodeMemories = nodeMemories;
+ this.segmentMemorySupport = segmentMemorySupport;
+ pJoinNode = PhreakNetworkNodeFactory.Factory.get().createPhreakJoinNode(reteEvaluator);
+ pEvalNode = PhreakNetworkNodeFactory.Factory.get().createPhreakEvalNode(reteEvaluator);
+ pFromNode = PhreakNetworkNodeFactory.Factory.get().createPhreakFromNode(reteEvaluator);
+ pReactiveFromNode = PhreakNetworkNodeFactory.Factory.get().createPhreakReactiveFromNode(reteEvaluator);
+ pNotNode = PhreakNetworkNodeFactory.Factory.get().createPhreakNotNode(reteEvaluator);
+ pExistsNode = PhreakNetworkNodeFactory.Factory.get().createPhreakExistsNode(reteEvaluator);
+ pAccNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAccumulateNode(reteEvaluator);
+ pGroupByNode = PhreakNetworkNodeFactory.Factory.get().createPhreakGroupByNode(reteEvaluator);
+ pBranchNode = PhreakNetworkNodeFactory.Factory.get().createPhreakBranchNode(reteEvaluator);
+ pQueryNode = PhreakNetworkNodeFactory.Factory.get().createPhreakQueryNode(reteEvaluator);
+ pTimerNode = PhreakNetworkNodeFactory.Factory.get().createPhreakTimerNode(reteEvaluator);
+ pSendNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAsyncSendNode(reteEvaluator);
+ pReceiveNode = PhreakNetworkNodeFactory.Factory.get().createPhreakAsyncReceiveNode(reteEvaluator);
+ pRtNode = PhreakNetworkNodeFactory.Factory.get().createPhreakRuleTerminalNode(reteEvaluator);
+ pQtNode = PhreakNetworkNodeFactory.Factory.get().createPhreakQueryTerminalNode(reteEvaluator);
+ }
+ @Override
public void evaluateNetwork(RuleExecutor executor,
PathMemory pmem) {
evaluateNetwork(pmem.getActualActivationsManager(), executor, pmem);
}
+ @Override
public void evaluateNetwork(ActivationsManager activationsManager,
RuleExecutor executor,
PathMemory pmem) {
@@ -148,8 +170,129 @@ public class RuleNetworkEvaluator {
}
outerEval(activationsManager, executor, pmem, smems, firstSegmentIsOnlyLia ? 1 : 0, firstSegmentIsOnlyLia ? 1L : 2L, nodeMem, node, srcTuples, true);
}
+
+ @Override
+ public void evaluate(PathMemory pmem,
+ ActivationsManager activationsManager,
+ NetworkNode sink,
+ Memory tm,
+ TupleSets trgLeftTuples) {
+ SegmentMemory[] smems = pmem.getSegmentMemories();
+ SegmentMemory sm = tm.getSegmentMemory();
+ int smemIndex = 0;
+ for (SegmentMemory smem : smems) {
+ if (smem == sm) {
+ break;
+ }
+ smemIndex++;
+ }
+
+ long bit = 1;
+ for (NetworkNode node = sm.getRootNode(); node != sink; node = ((LeftTupleSource) node).getSinkPropagator()
+ .getFirstLeftTupleSink()) {
+ //update the bit to the correct node position.
+ bit = nextNodePosMask(bit);
+ }
+
+ outerEval(activationsManager, pmem.getRuleAgendaItem().getRuleExecutor(), pmem, smems,
+ smemIndex, bit, tm, sink, trgLeftTuples, true);
+ }
+
+
+ @Override
+ public void forceFlushWhenSubnetwork(PathMemory pmem) {
+ for (PathMemory outPmem : findPathsToFlushFromSubnetwork(pmem)) {
+ forceFlushPath(outPmem);
+ }
+ }
+
+ @Override
+ public List<PathMemory> findPathsToFlushFromSubnetwork(PathMemory pmem) {
+ List<PathMemory> paths = null;
+ if (pmem.isDataDriven() && pmem.getNodeType() == NodeTypeEnums.TupleToObjectNode) {
+ for (PathEndNode pnode : pmem.getPathEndNode().getPathEndNodes()) {
+ if (NodeTypeEnums.isTerminalNode(pnode)) {
+ PathMemory outPmem = nodeMemories.getNodeMemory(pnode);
+ if (outPmem.isDataDriven()) {
+ if (paths == null) {
+ paths = new ArrayList<>();
+ }
+ paths.add(outPmem);
+ }
+ }
+ }
+ }
+ return paths == null ? Collections.emptyList() : paths;
+ }
+
+
+ @Override
+ public void forceFlushPath(PathMemory outPmem) {
+ SegmentMemory outSmem = outPmem.getSegmentMemories()[0];
+ if (outSmem != null) {
+ forceFlushLeftTuple(outPmem, outSmem, new TupleSetsImpl());
+ }
+ }
+
+
+ @Override
+ public void forceFlushLeftTuple(PathMemory pmem,
+ SegmentMemory sm,
+ TupleSets leftTupleSets) {
+ SegmentMemory[] smems = pmem.getSegmentMemories();
- public void outerEval(ActivationsManager activationsManager,
+ LeftTupleNode node;
+ Memory mem;
+ long bit = 1;
+ if (NodeTypeEnums.isLeftInputAdapterNode(sm.getRootNode()) && !NodeTypeEnums.isLeftInputAdapterNode(sm
+ .getTipNode())) {
+ // The segment is the first and it has the lian shared with other nodes, the lian must be skipped, so adjust the bit and sink
+ node = sm.getRootNode().getSinkPropagator().getFirstLeftTupleSink();
+ mem = sm.getNodeMemories()[1];
+ bit = 2; // adjust bit to point to next node
+ } else {
+ node = sm.getRootNode();
+ mem = sm.getNodeMemories()[0];
+ }
+
+ PathMemory rtnPmem = NodeTypeEnums.isTerminalNode(pmem.getPathEndNode()) ? pmem : nodeMemories.getNodeMemory(
+ (AbstractTerminalNode) pmem.getPathEndNode().getPathEndNodes()[0]);
+
+ ActivationsManager activationsManager = pmem.getActualActivationsManager();
+ outerEval(activationsManager, rtnPmem.getOrCreateRuleAgendaItem().getRuleExecutor(), pmem, smems, sm.getPos(), bit, mem, node,
+ leftTupleSets,
+ true);
+ }
+
+
+ @Override
+ public boolean flushLeftTupleIfNecessary(SegmentMemory sm, boolean streamMode) {
+ return flushLeftTupleIfNecessary(sm, null, streamMode, Tuple.NONE);
+ }
+
+ @Override
+ public boolean flushLeftTupleIfNecessary(SegmentMemory sm,
+ TupleImpl leftTuple,
+ boolean streamMode,
+ short stagedType) {
+ PathMemory pmem = findPathToFlush(sm, leftTuple, streamMode);
+
+ if (pmem == null) {
+ return false;
+ }
+
+ forceFlushLeftTuple(pmem, sm, createLeftTupleTupleSets(leftTuple, stagedType));
+ forceFlushWhenSubnetwork(pmem);
+ return true;
+ }
+
+ public static PathMemory findPathToFlush(SegmentMemory sm, TupleImpl leftTuple, boolean streamMode) {
+ boolean forceFlush = streamMode || (leftTuple != null && leftTuple.getFactHandle() != null && leftTuple
+ .getFactHandle().isEvent());
+ return forceFlush ? sm.getPathMemories().get(0) : sm.getFirstDataDrivenPathMemory();
+ }
+
+ private void outerEval(ActivationsManager activationsManager,
RuleExecutor executor,
PathMemory pmem,
SegmentMemory[] smems,
@@ -169,7 +312,7 @@ public class RuleNetworkEvaluator {
}
}
- public void evalStackEntry(ActivationsManager activationsManager,
+ private void evalStackEntry(ActivationsManager activationsManager,
RuleExecutor executor,
LinkedList<StackEntry> stack,
StackEntry entry) {
@@ -207,9 +350,7 @@ public class RuleNetworkEvaluator {
bit = nextNodePosMask(bit); // update bit to new node
} else {
// Reached end of segment, start on new segment.
- SegmentPropagator.propagate(smem,
- trgTuples,
- activationsManager.getReteEvaluator());
+ propagate(smem, trgTuples);
smem = smems[++smemIndex];
trgTuples = smem.getStagedLeftTuples().takeAll();
node = smem.getRootNode();
@@ -225,7 +366,7 @@ public class RuleNetworkEvaluator {
innerEval(activationsManager, executor, stack, pmem, smems, smemIndex, bit, nodeMem, node, trgTuples, processSubnetwork);
}
- public void innerEval(ActivationsManager activationsManager,
+ private void innerEval(ActivationsManager activationsManager,
RuleExecutor executor,
LinkedList<StackEntry> stack,
PathMemory pmem,
@@ -259,7 +400,7 @@ public class RuleNetworkEvaluator {
// this is needed for subnetworks that feed into a parent network that has no right inputs,
// and may not yet be initialized
if (smem.isEmpty() && !NodeTypeEnums.isTerminalNode(smem.getTipNode())) {
- activationsManager.getReteEvaluator().getSegmentMemorySupport().createChildSegments(smem.getTipNode().getSinkPropagator(), smem);
+ segmentMemorySupport.createChildSegments(smem.getTipNode().getSinkPropagator(), smem);
}
smem = smems[i];
@@ -306,10 +447,10 @@ public class RuleNetworkEvaluator {
pRtNode.doNode(activationsManager, executor, (AbstractTerminalNode) node, srcTuples);
break;
case NodeTypeEnums.QueryTerminalNode:
- pQtNode.doNode(activationsManager, stack, (QueryTerminalNode) node, srcTuples);
+ pQtNode.doNode(stack, (QueryTerminalNode) node, srcTuples);
break;
case NodeTypeEnums.TupleToObjectNode:
- doSubnetwork2(activationsManager.getReteEvaluator(), (TupleToObjectNode) node, srcTuples);
+ doSubnetwork2((TupleToObjectNode) node, srcTuples);
break;
default:
terminalNode = false;
@@ -318,7 +459,7 @@ public class RuleNetworkEvaluator {
break;
}
- stagedLeftTuples = getTargetStagedLeftTuples(activationsManager.getReteEvaluator(), smem, node);
+ stagedLeftTuples = getTargetStagedLeftTuples(smem, node);
LeftTupleSinkNode sink = ((LeftTupleSource) node).getSinkPropagator().getFirstLeftTupleSink();
trgTuples = evalNode(activationsManager, executor, stack, pmem, smems, smem, smemIndex, bit, nodeMem, node, sink, srcTuples, stagedLeftTuples, processSubnetwork);
@@ -335,7 +476,7 @@ public class RuleNetworkEvaluator {
// Reached end of segment, start on new segment.
smem.getFirst().getStagedLeftTuples().addAll(stagedLeftTuples); // must put back all the LTs
// end of SegmentMemory, so we know that stagedLeftTuples is not null
- SegmentPropagator.propagate(smem, trgTuples, activationsManager.getReteEvaluator());
+ propagate(smem, trgTuples);
bit = 1;
smem = smems[++smemIndex];
trgTuples = smem.getStagedLeftTuples().takeAll();
@@ -355,7 +496,7 @@ public class RuleNetworkEvaluator {
}
}
- public TupleSets evalNode(ActivationsManager activationsManager,
+ private TupleSets evalNode(ActivationsManager activationsManager,
RuleExecutor executor,
LinkedList<StackEntry> stack,
PathMemory pmem,
@@ -379,24 +520,24 @@ public class RuleNetworkEvaluator {
boolean exitInnerEval = false;
switch (node.getType()) {
case NodeTypeEnums.EvalConditionNode: {
- pEvalNode.doNode(activationsManager.getReteEvaluator(), (EvalConditionNode) node, (EvalMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples);
+ pEvalNode.doNode((EvalConditionNode) node, (EvalMemory) nodeMem, sink,
+ srcTuples, trgTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.FromNode: {
- pFromNode.doNode(activationsManager.getReteEvaluator(), (FromNode) node, (FromMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples);
+ pFromNode.doNode((FromNode) node, (FromMemory) nodeMem, sink,
+ srcTuples, trgTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.ReactiveFromNode: {
- pReactiveFromNode.doNode(activationsManager.getReteEvaluator(), (ReactiveFromNode) node, (ReactiveFromNode.ReactiveFromMemory) nodeMem,
- sink, srcTuples, trgTuples, stagedLeftTuples);
+ pReactiveFromNode.doNode((ReactiveFromNode) node, (ReactiveFromNode.ReactiveFromMemory) nodeMem, sink,
+ srcTuples, trgTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.QueryElementNode: {
- exitInnerEval = evalQueryNode(activationsManager.getReteEvaluator(), stack, pmem, smems, smemIndex, bit, nodeMem,
- node, sink, trgTuples, srcTuples, stagedLeftTuples);
+ exitInnerEval = evalQueryNode(stack, pmem, smems, smemIndex, bit, nodeMem, node,
+ sink, trgTuples, srcTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.TimerConditionNode: {
@@ -409,11 +550,11 @@ public class RuleNetworkEvaluator {
break;
}
case NodeTypeEnums.AsyncSendNode: {
- pSendNode.doNode(activationsManager.getReteEvaluator(), (AsyncSendNode) node, (AsyncSendMemory) nodeMem, srcTuples);
+ pSendNode.doNode((AsyncSendNode) node, (AsyncSendMemory) nodeMem, srcTuples);
break;
}
case NodeTypeEnums.AsyncReceiveNode: {
- pReceiveNode.doNode(activationsManager.getReteEvaluator(), (AsyncReceiveNode) node, (AsyncReceiveMemory) nodeMem, sink, srcTuples, trgTuples);
+ pReceiveNode.doNode((AsyncReceiveNode) node, (AsyncReceiveMemory) nodeMem, sink, srcTuples, trgTuples);
break;
}
}
@@ -424,13 +565,12 @@ public class RuleNetworkEvaluator {
return trgTuples;
}
- private static TupleSets getTargetStagedLeftTuples(ReteEvaluator reteEvaluator,
- SegmentMemory smem,
+ private TupleSets getTargetStagedLeftTuples(SegmentMemory smem,
NetworkNode node) {
if (node == smem.getTipNode()) {
// we are about to process the segment tip, allow it to merge insert/update/delete clashes
if (smem.isEmpty()) {
- reteEvaluator.getSegmentMemorySupport().createChildSegments(((LeftTupleSource) node).getSinkPropagator(), smem);
+ segmentMemorySupport.createChildSegments(((LeftTupleSource) node).getSinkPropagator(), smem);
}
return smem.getFirst().getStagedLeftTuples().takeAll();
} else {
@@ -438,8 +578,7 @@ public class RuleNetworkEvaluator {
}
}
- private boolean evalQueryNode(ReteEvaluator reteEvaluator,
- LinkedList<StackEntry> stack,
+ private boolean evalQueryNode(LinkedList<StackEntry> stack,
PathMemory pmem,
SegmentMemory[] smems,
int smemIndex,
@@ -479,7 +618,7 @@ public class RuleNetworkEvaluator {
stack.add(stackEntry);
pQueryNode.doNode(qnode, (QueryElementNodeMemory) nodeMem, stackEntry,
- reteEvaluator, srcTuples, trgTuples, stagedLeftTuples);
+ srcTuples, trgTuples, stagedLeftTuples);
SegmentMemory qsmem = ((QueryElementNodeMemory) nodeMem).getQuerySegmentMemory();
List<PathMemory> qpmems = qsmem.getPathMemories();
@@ -554,13 +693,12 @@ public class RuleNetworkEvaluator {
return true; // return here, TupleToObjectNode queues the evaluation on the stack, which is necessary to handled nested query nodes
}
- switchOnDoBetaNode(activationsManager.getReteEvaluator(), bm, am, node, sink, trgTuples, srcTuples, stagedLeftTuples);
+ switchOnDoBetaNode(bm, am, node, sink, trgTuples, srcTuples, stagedLeftTuples);
return false;
}
- private void switchOnDoBetaNode(ReteEvaluator reteEvaluator,
- BetaMemory bm,
+ private void switchOnDoBetaNode(BetaMemory bm,
AccumulateMemory am,
NetworkNode node,
LeftTupleSinkNode sink,
@@ -574,26 +712,26 @@ public class RuleNetworkEvaluator {
switch (node.getType()) {
case NodeTypeEnums.JoinNode: {
- pJoinNode.doNode(reteEvaluator, (JoinNode) node, sink,
- bm, srcTuples, trgTuples, stagedLeftTuples);
+ pJoinNode.doNode((JoinNode) node, sink, bm,
+ srcTuples, trgTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.NotNode: {
- pNotNode.doNode(reteEvaluator, (NotNode) node, sink,
- bm, srcTuples, trgTuples, stagedLeftTuples);
+ pNotNode.doNode((NotNode) node, sink, bm,
+ srcTuples, trgTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.ExistsNode: {
- pExistsNode.doNode(reteEvaluator, (ExistsNode) node, sink,
- bm, srcTuples, trgTuples, stagedLeftTuples);
+ pExistsNode.doNode((ExistsNode) node, sink, bm,
+ srcTuples, trgTuples, stagedLeftTuples);
break;
}
case NodeTypeEnums.AccumulateNode: {
AccumulateNode accumulateNode = (AccumulateNode) node;
if (accumulateNode.getAccumulate().isGroupBy()) {
- pGroupByNode.doNode(reteEvaluator, accumulateNode, sink, am, srcTuples, trgTuples, stagedLeftTuples);
+ pGroupByNode.doNode(accumulateNode, sink, am, srcTuples, trgTuples, stagedLeftTuples);
} else {
- pAccNode.doNode(reteEvaluator, accumulateNode, sink, am, srcTuples, trgTuples, stagedLeftTuples);
+ pAccNode.doNode(accumulateNode, sink, am, srcTuples, trgTuples, stagedLeftTuples);
}
break;
}
@@ -638,15 +776,14 @@ public class RuleNetworkEvaluator {
1, subSmem.getNodeMemories()[0], subSmem.getRootNode(), subLts, true);
}
- private void doSubnetwork2(ReteEvaluator reteEvaluator,
- TupleToObjectNode tton,
+ private void doSubnetwork2(TupleToObjectNode tton,
TupleSets srcTuples) {
ObjectSink[] sinks = tton.getObjectSinkPropagator().getSinks();
BetaNode betaNode = ((RightInputAdapterNode)sinks[0]).getBetaNode();
BetaMemory bm;
- Memory nodeMem = reteEvaluator.getNodeMemory(betaNode);
+ Memory nodeMem = nodeMemories.getNodeMemory(betaNode);
if (NodeTypeEnums.AccumulateNode == betaNode.getType()) {
bm = ((AccumulateMemory) nodeMem).getBetaMemory();
} else {
@@ -663,7 +800,7 @@ public class RuleNetworkEvaluator {
bms = new BetaMemory[sinks.length - 1];
for (int i = 1; i < length; i++) {
bns[i - 1] = ((RightInputAdapterNode)sinks[i]).getBetaNode();
- Memory nodeMem2 = reteEvaluator.getNodeMemory(bns[i - 1]);
+ Memory nodeMem2 = nodeMemories.getNodeMemory(bns[i - 1]);
if (NodeTypeEnums.AccumulateNode == betaNode.getType()) {
bms[i - 1] = ((AccumulateMemory) nodeMem2).getBetaMemory();
} else {
@@ -754,251 +891,133 @@ public class RuleNetworkEvaluator {
}
srcTuples.resetAll();
}
-
- public static void findLeftTupleBlocker(BetaNode betaNode,
- TupleMemory rtm,
- Object contextEntry,
- BetaConstraints constraints,
- TupleImpl leftTuple,
- boolean useLeftMemory) {
- // This method will also remove rightTuples that are from subnetwork where no leftmemory use used
- FastIterator it = betaNode.getRightIterator(rtm);
- for (RightTuple rightTuple = betaNode.getFirstRightTuple(leftTuple, rtm, it); rightTuple != null;) {
- RightTuple nextRight = (RightTuple) it.next(rightTuple);
- if (constraints.isAllowedCachedLeft(contextEntry,
- rightTuple.getFactHandleForEvaluation())) {
- leftTuple.setBlocker(rightTuple);
-
- if (useLeftMemory) {
- rightTuple.addBlocked((LeftTuple) leftTuple);
- break;
- } else if (betaNode.getRightInput().inputIsTupleToObjectNode()) {
- // If we aren't using leftMemory and the right input is a TupleToObjectNode, then we must iterate and find all subetwork right tuples and remove them
- // so we don't break
- rtm.remove(rightTuple);
- } else {
- break;
- }
- }
- rightTuple = nextRight;
- }
- }
-
-
- public static void unlinkAndDeleteChildLeftTuple(TupleSets trgLeftTuples,
- TupleSets stagedLeftTuples,
- TupleImpl childLeftTuple) {
- childLeftTuple.unlinkFromRightParent();
- childLeftTuple.unlinkFromLeftParent();
- deleteChildLeftTuple(childLeftTuple, trgLeftTuples, stagedLeftTuples);
- }
-
- public static void deleteChildLeftTuple(TupleImpl childLeftTuple, TupleSets trgLeftTuples, TupleSets stagedLeftTuples) {
- if (childLeftTuple.isStagedOnRight()) {
- ((SubnetworkTuple) childLeftTuple).moveStagingFromRightToLeft();
- } else {
- switch (childLeftTuple.getStagedType()) {
- // handle clash with already staged entries
- case Tuple.INSERT:
- stagedLeftTuples.removeInsert(childLeftTuple);
- trgLeftTuples.addNormalizedDelete(childLeftTuple);
- return;
- case Tuple.UPDATE:
- stagedLeftTuples.removeUpdate(childLeftTuple);
- break;
- }
- }
- trgLeftTuples.addDelete(childLeftTuple);
- }
-
- public static void doUpdatesReorderLeftMemory(BetaMemory bm,
- TupleSets srcLeftTuples) {
- TupleMemory ltm = bm.getLeftTupleMemory();
-
- // sides must first be re-ordered, to ensure iteration integrity
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- if (leftTuple.getMemory() != null) {
- ltm.remove(leftTuple);
- }
- }
-
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- ltm.add(leftTuple);
- for (TupleImpl childLeftTuple = leftTuple.getFirstChild(); childLeftTuple != null;) {
- TupleImpl childNext = childLeftTuple.getHandleNext();
- childLeftTuple.reAddRight();
- childLeftTuple = childNext;
- }
+
+ public void propagate(SegmentMemory sourceSegment, TupleSets leftTuples) {
+ if (leftTuples.isEmpty()) {
+ return;
}
- }
- public static void doUpdatesExistentialReorderLeftMemory(BetaMemory bm,
- TupleSets srcLeftTuples) {
- TupleMemory ltm = bm.getLeftTupleMemory();
-
- // sides must first be re-ordered, to ensure iteration integrity
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- if (leftTuple.getMemory() != null) {
- ltm.remove(leftTuple);
- }
+ LeftTupleSource source = ( LeftTupleSource ) sourceSegment.getTipNode();
+
+ if ( sourceSegment.isEmpty() ) {
+ segmentMemorySupport.createChildSegments(source.getSinkPropagator(), sourceSegment);
}
-
- for (TupleImpl leftTuple = srcLeftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- RightTuple blocker = leftTuple.getBlocker();
- if (blocker == null) {
- ltm.add(leftTuple);
- for (TupleImpl childLeftTuple = leftTuple.getFirstChild(); childLeftTuple != null;) {
- TupleImpl childNext = childLeftTuple.getHandleNext();
- childLeftTuple.reAddRight();
- childLeftTuple = childNext;
+
+ processPeers(sourceSegment, leftTuples);
+
+ Iterator<SegmentMemory> peersIterator = sourceSegment.getPeersWithDataDrivenPathMemoriesIterator();
+ while (peersIterator.hasNext()) {
+ SegmentMemory smem = peersIterator.next();
+ for (PathMemory dataDrivenPmem : smem.getDataDrivenPathMemories()) {
+ if (smem.getStagedLeftTuples().getDeleteFirst() == null &&
+ smem.getStagedLeftTuples().getUpdateFirst() == null &&
+ !dataDrivenPmem.isRuleLinked()) {
+ // skip flushing segments that have only inserts staged and the path is not linked
+ continue;
}
- } else if (blocker.getStagedType() != LeftTuple.NONE) {
- // it's blocker is also being updated, so remove to force it to start from the beginning
- blocker.removeBlocked((LeftTuple) leftTuple);
- }
- }
- }
-
- public static void doUpdatesReorderRightMemory(BetaMemory bm,
- TupleSets srcRightTuples) {
- TupleMemory rtm = bm.getRightTupleMemory();
-
- for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- if (rightTuple.getMemory() != null) {
- rtm.removeAdd(rightTuple);
- doUpdatesReorderChildLeftTuple(rightTuple);
- }
- }
- }
-
- public static void doUpdatesReorderChildLeftTuple(TupleImpl rightTuple) {
- for (TupleImpl childLeftTuple = rightTuple.getFirstChild(); childLeftTuple != null;) {
- TupleImpl childNext = childLeftTuple.getRightParentNext();
- childLeftTuple.reAddLeft();
- childLeftTuple = childNext;
- }
- }
-
- public static void doUpdatesExistentialReorderRightMemory(BetaMemory bm,
- BetaNode betaNode,
- TupleSets srcRightTuples) {
- TupleMemory rtm = bm.getRightTupleMemory();
-
- boolean resumeFromCurrent = !(betaNode.isIndexedUnificationJoin() || rtm.getIndexType().isComparison());
-
- // remove all the staged rightTuples from the memory before to readd them all
- // this is to avoid split bucket when an updated rightTuple hasn't been moved yet
- // and so it is the first entry in the wrong bucket
-
- if (rtm.getIndexType() != TupleMemory.IndexType.NONE) {
- for (TupleImpl rightTuple = srcRightTuples.getDeleteFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- rtm.remove(rightTuple);
- }
- }
-
- for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- doRemoveExistentialRightMemoryForReorder(rtm, (RightTuple) rightTuple, resumeFromCurrent);
- }
-
- for (TupleImpl rightTuple = srcRightTuples.getUpdateFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- doAddExistentialRightMemoryForReorder(rtm, (RightTuple) rightTuple, resumeFromCurrent);
- }
-
- if (rtm.getIndexType() != TupleMemory.IndexType.NONE) {
- for (TupleImpl rightTuple = srcRightTuples.getDeleteFirst(); rightTuple != null; rightTuple = rightTuple.getStagedNext()) {
- rtm.add(rightTuple);
+ forceFlushLeftTuple(dataDrivenPmem, smem, smem.getStagedLeftTuples());
+ forceFlushWhenSubnetwork(dataDrivenPmem);
}
}
}
- public static void doExistentialUpdatesReorderChildLeftTuple(ReteEvaluator reteEvaluator,
- NotNode notNode,
- RightTuple rightTuple) {
- BetaMemory bm = getBetaMemory(notNode, reteEvaluator);
- TupleMemory rtm = bm.getRightTupleMemory();
+ private static void processPeers(SegmentMemory sourceSegment, TupleSets leftTuples) {
+ SegmentMemory firstSmem = sourceSegment.getFirst();
- boolean resumeFromCurrent = !(notNode.isIndexedUnificationJoin() || rtm.getIndexType().isComparison());
- doRemoveExistentialRightMemoryForReorder(rtm, rightTuple, resumeFromCurrent);
- doAddExistentialRightMemoryForReorder(rtm, rightTuple, resumeFromCurrent);
+ processPeerDeletes( leftTuples.getDeleteFirst(), firstSmem );
+ processPeerDeletes( leftTuples.getNormalizedDeleteFirst(), firstSmem );
+ processPeerUpdates( leftTuples, firstSmem );
+ processPeerInserts( leftTuples, firstSmem );
- updateBlockersAndPropagate(notNode, rightTuple, reteEvaluator, rtm, bm.getContext(), notNode.getRawConstraints(), !resumeFromCurrent, null, null, null);
+ firstSmem.getStagedLeftTuples().addAll( leftTuples );
+ leftTuples.resetAll();
}
- private static void doAddExistentialRightMemoryForReorder(TupleMemory rtm,
- RightTuple rightTuple,
- boolean resumeFromCurrent) {
- rtm.add(rightTuple);
-
- if (resumeFromCurrent) {
- if (rightTuple.getBlocked() != null && rightTuple.getTempNextRightTuple() == null) {
- // the next RightTuple was null, but current RightTuple was added back into the same bucket, so reset as root blocker to re-match can be attempted
- rightTuple.setTempNextRightTuple(rightTuple);
+ private static void processPeerInserts(TupleSets leftTuples, SegmentMemory firstSmem) {
+ for (TupleImpl leftTuple = leftTuples.getInsertFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
+ SegmentMemory smem = firstSmem.getNext();
+ if ( smem != null ) {
+ // It's possible for a deleted tuple and set of peers, to be cached on a delete (such as with accumulates).
+ // So we should check if the instances already exist and use them if they do.
+ if (leftTuple.getPeer() == null) {
+ TupleImpl peer = leftTuple;
+ // peers do not exist, so create and add them.
+ for (; smem != null; smem = smem.getNext()) {
+ LeftTupleSink sink = smem.getSinkFactory();
+ peer = TupleFactory.createPeer(sink, peer); // pctx is set during peer cloning
+ smem.getStagedLeftTuples().addInsert( peer );
+ }
+ } else {
+ TupleImpl peer = leftTuple.getPeer();
+ // peers exist, so update them as an insert, which also handles staged clashing.
+ for (; smem != null; smem = smem.getNext()) {
+ peer.setPropagationContext( leftTuple.getPropagationContext() );
+ // ... and update the staged LeftTupleSets according to its current staged state
+ updateChildLeftTupleDuringInsert(peer, smem.getStagedLeftTuples(), smem.getStagedLeftTuples());
+ peer = peer.getPeer();
+ }
+ }
}
}
-
- doUpdatesReorderChildLeftTuple(rightTuple);
}
- private static void doRemoveExistentialRightMemoryForReorder(TupleMemory rtm,
- RightTuple rightTuple,
- boolean resumeFromCurrent) {
- if (rightTuple.getMemory() != null) {
-
- if (resumeFromCurrent) {
- if (rightTuple.getBlocked() != null) {
- // look for a non-staged right tuple first forward ...
- RightTuple tempRightTuple = (RightTuple) rightTuple.getNext();
- while (tempRightTuple != null && tempRightTuple.getStagedType() != LeftTuple.NONE) {
- // next cannot be an updated or deleted rightTuple
- tempRightTuple = (RightTuple) tempRightTuple.getNext();
+ private static void processPeerUpdates(TupleSets leftTuples, SegmentMemory firstSmem) {
+ for (TupleImpl leftTuple = leftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
+ SegmentMemory smem = firstSmem.getNext();
+ if ( smem != null ) {
+ for ( TupleImpl peer = leftTuple.getPeer(); peer != null; peer = peer.getPeer() ) {
+ // only stage, if not already staged, if insert, leave as insert
+ if ( peer.getStagedType() == LeftTuple.NONE ) {
+ peer.setPropagationContext( leftTuple.getPropagationContext() );
+ smem.getStagedLeftTuples().addUpdate( peer );
}
- // ... and if cannot find one try backward
- if (tempRightTuple == null) {
- tempRightTuple = (RightTuple) rightTuple.getPrevious();
- while (tempRightTuple != null && tempRightTuple.getStagedType() != LeftTuple.NONE) {
- // next cannot be an updated or deleted rightTuple
- tempRightTuple = (RightTuple) tempRightTuple.getPrevious();
- }
- }
-
- rightTuple.setTempNextRightTuple(tempRightTuple);
+ smem = smem.getNext();
}
}
-
- rightTuple.setTempBlocked(rightTuple.getBlocked());
- rightTuple.setBlocked(null);
- rtm.remove(rightTuple);
}
}
- public static boolean useLeftMemory(LeftTupleSource tupleSource,
- TupleImpl tuple) {
- boolean useLeftMemory = true;
- if (!tupleSource.isLeftTupleMemoryEnabled()) {
- // This is a hack, to not add closed DroolsQuery objects
- Object object = tuple.getRootTuple().getFactHandle().getObject();
- if (!(object instanceof DroolsQueryImpl) || !((DroolsQueryImpl) object).isOpen()) {
- useLeftMemory = false;
- }
+ private static void updateChildLeftTupleDuringInsert(TupleImpl childLeftTuple, TupleSets stagedLeftTuples, TupleSets trgLeftTuples) {
+ switch ( childLeftTuple.getStagedType() ) {
+ // handle clash with already staged entries
+ case LeftTuple.INSERT:
+ // Was insert before, should continue as insert
+ stagedLeftTuples.removeInsert( childLeftTuple );
+ trgLeftTuples.addInsert( childLeftTuple );
+ break;
+ case LeftTuple.UPDATE:
+ stagedLeftTuples.removeUpdate( childLeftTuple );
+ trgLeftTuples.addUpdate( childLeftTuple );
+ break;
+ default:
+ // no clash, so just add
+ if (childLeftTuple.getSink().getType() == AccumulateNode ) {
+ trgLeftTuples.addInsert(childLeftTuple);
+ } else {
+ trgLeftTuples.addUpdate(childLeftTuple);
+ }
}
- return useLeftMemory;
}
- public static void normalizeStagedTuples(TupleSets stagedLeftTuples,
- TupleImpl childLeftTuple) {
- if (!childLeftTuple.isStagedOnRight()) {
- switch (childLeftTuple.getStagedType()) {
- // handle clash with already staged entries
- case LeftTuple.INSERT:
- stagedLeftTuples.removeInsert(childLeftTuple);
- break;
- case LeftTuple.UPDATE:
- stagedLeftTuples.removeUpdate(childLeftTuple);
- break;
+ private static void processPeerDeletes(TupleImpl leftTuple, SegmentMemory firstSmem) {
+ for (; leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
+ SegmentMemory smem = firstSmem.getNext();
+ if ( smem != null ) {
+ for ( TupleImpl peer = leftTuple.getPeer(); peer != null; peer = peer.getPeer() ) {
+ peer.setPropagationContext( leftTuple.getPropagationContext() );
+ TupleSets stagedLeftTuples = smem.getStagedLeftTuples();
+ // if the peer is already staged as insert or update the LeftTupleSets will reconcile it internally
+ stagedLeftTuples.addDelete( peer );
+ smem = smem.getNext();
+ }
}
}
}
+
+
+
+
private static String indent(int size) {
diff --git a/drools-core/src/main/java/org/drools/core/phreak/SegmentPropagator.java b/drools-core/src/main/java/org/drools/core/phreak/SegmentPropagator.java
deleted file mode 100644
index 5c116cb0cb..0000000000
--- a/drools-core/src/main/java/org/drools/core/phreak/SegmentPropagator.java
+++ /dev/null
@@ -1,161 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.drools.core.phreak;
-
-import java.util.Iterator;
-
-import org.drools.core.common.ReteEvaluator;
-import org.drools.core.common.TupleSets;
-import org.drools.core.reteoo.LeftTuple;
-import org.drools.core.reteoo.LeftTupleSink;
-import org.drools.core.reteoo.LeftTupleSource;
-import org.drools.core.reteoo.PathMemory;
-import org.drools.core.reteoo.SegmentMemory;
-import org.drools.core.reteoo.TupleFactory;
-import org.drools.core.reteoo.TupleImpl;
-
-import static org.drools.base.reteoo.NodeTypeEnums.AccumulateNode;
-import static org.drools.core.phreak.TupleEvaluationUtil.forceFlushLeftTuple;
-import static org.drools.core.phreak.TupleEvaluationUtil.forceFlushWhenSubnetwork;
-
-public class SegmentPropagator {
-
- public static void propagate(SegmentMemory sourceSegment, TupleSets leftTuples, ReteEvaluator reteEvaluator) {
- if (leftTuples.isEmpty()) {
- return;
- }
-
- LeftTupleSource source = ( LeftTupleSource ) sourceSegment.getTipNode();
-
- if ( sourceSegment.isEmpty() ) {
- reteEvaluator.getSegmentMemorySupport().createChildSegments(source.getSinkPropagator(), sourceSegment);
- }
-
- processPeers(sourceSegment, leftTuples);
-
- Iterator<SegmentMemory> peersIterator = sourceSegment.getPeersWithDataDrivenPathMemoriesIterator();
- while (peersIterator.hasNext()) {
- SegmentMemory smem = peersIterator.next();
- for (PathMemory dataDrivenPmem : smem.getDataDrivenPathMemories()) {
- if (smem.getStagedLeftTuples().getDeleteFirst() == null &&
- smem.getStagedLeftTuples().getUpdateFirst() == null &&
- !dataDrivenPmem.isRuleLinked()) {
- // skip flushing segments that have only inserts staged and the path is not linked
- continue;
- }
- forceFlushLeftTuple(reteEvaluator, dataDrivenPmem, smem, smem.getStagedLeftTuples());
- forceFlushWhenSubnetwork(reteEvaluator, dataDrivenPmem);
- }
- }
- }
-
- private static void processPeers(SegmentMemory sourceSegment, TupleSets leftTuples) {
- SegmentMemory firstSmem = sourceSegment.getFirst();
-
- processPeerDeletes( leftTuples.getDeleteFirst(), firstSmem );
- processPeerDeletes( leftTuples.getNormalizedDeleteFirst(), firstSmem );
- processPeerUpdates( leftTuples, firstSmem );
- processPeerInserts( leftTuples, firstSmem );
-
- firstSmem.getStagedLeftTuples().addAll( leftTuples );
- leftTuples.resetAll();
- }
-
- private static void processPeerInserts(TupleSets leftTuples, SegmentMemory firstSmem) {
- for (TupleImpl leftTuple = leftTuples.getInsertFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- SegmentMemory smem = firstSmem.getNext();
- if ( smem != null ) {
- // It's possible for a deleted tuple and set of peers, to be cached on a delete (such as with accumulates).
- // So we should check if the instances already exist and use them if they do.
- if (leftTuple.getPeer() == null) {
- TupleImpl peer = leftTuple;
- // peers do not exist, so create and add them.
- for (; smem != null; smem = smem.getNext()) {
- LeftTupleSink sink = smem.getSinkFactory();
- peer = TupleFactory.createPeer(sink, peer); // pctx is set during peer cloning
- smem.getStagedLeftTuples().addInsert( peer );
- }
- } else {
- TupleImpl peer = leftTuple.getPeer();
- // peers exist, so update them as an insert, which also handles staged clashing.
- for (; smem != null; smem = smem.getNext()) {
- peer.setPropagationContext( leftTuple.getPropagationContext() );
- // ... and update the staged LeftTupleSets according to its current staged state
- updateChildLeftTupleDuringInsert(peer, smem.getStagedLeftTuples(), smem.getStagedLeftTuples());
- peer = peer.getPeer();
- }
- }
- }
- }
- }
-
- private static void processPeerUpdates(TupleSets leftTuples, SegmentMemory firstSmem) {
- for (TupleImpl leftTuple = leftTuples.getUpdateFirst(); leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- SegmentMemory smem = firstSmem.getNext();
- if ( smem != null ) {
- for ( TupleImpl peer = leftTuple.getPeer(); peer != null; peer = peer.getPeer() ) {
- // only stage, if not already staged, if insert, leave as insert
- if ( peer.getStagedType() == LeftTuple.NONE ) {
- peer.setPropagationContext( leftTuple.getPropagationContext() );
- smem.getStagedLeftTuples().addUpdate( peer );
- }
-
- smem = smem.getNext();
- }
- }
- }
- }
-
- private static void updateChildLeftTupleDuringInsert(TupleImpl childLeftTuple, TupleSets stagedLeftTuples, TupleSets trgLeftTuples) {
- switch ( childLeftTuple.getStagedType() ) {
- // handle clash with already staged entries
- case LeftTuple.INSERT:
- // Was insert before, should continue as insert
- stagedLeftTuples.removeInsert( childLeftTuple );
- trgLeftTuples.addInsert( childLeftTuple );
- break;
- case LeftTuple.UPDATE:
- stagedLeftTuples.removeUpdate( childLeftTuple );
- trgLeftTuples.addUpdate( childLeftTuple );
- break;
- default:
- // no clash, so just add
- if (childLeftTuple.getSink().getType() == AccumulateNode ) {
- trgLeftTuples.addInsert(childLeftTuple);
- } else {
- trgLeftTuples.addUpdate(childLeftTuple);
- }
- }
- }
-
- private static void processPeerDeletes(TupleImpl leftTuple, SegmentMemory firstSmem) {
- for (; leftTuple != null; leftTuple = leftTuple.getStagedNext()) {
- SegmentMemory smem = firstSmem.getNext();
- if ( smem != null ) {
- for ( TupleImpl peer = leftTuple.getPeer(); peer != null; peer = peer.getPeer() ) {
- peer.setPropagationContext( leftTuple.getPropagationContext() );
- TupleSets stagedLeftTuples = smem.getStagedLeftTuples();
- // if the peer is already staged as insert or update the LeftTupleSets will reconcile it internally
- stagedLeftTuples.addDelete( peer );
- smem = smem.getNext();
- }
- }
- }
- }
-}
diff --git a/drools-core/src/main/java/org/drools/core/phreak/TupleEvaluationUtil.java b/drools-core/src/main/java/org/drools/core/phreak/TupleEvaluationUtil.java
deleted file mode 100644
index 67de54b239..0000000000
--- a/drools-core/src/main/java/org/drools/core/phreak/TupleEvaluationUtil.java
+++ /dev/null
@@ -1,144 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you 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 org.drools.core.phreak;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-
-import org.drools.base.reteoo.NodeTypeEnums;
-import org.drools.core.common.ActivationsManager;
-import org.drools.core.common.Memory;
-import org.drools.core.common.ReteEvaluator;
-import org.drools.core.common.TupleSets;
-import org.drools.core.common.TupleSetsImpl;
-import org.drools.core.reteoo.AbstractTerminalNode;
-import org.drools.core.reteoo.TupleImpl;
-import org.drools.core.reteoo.LeftTupleNode;
-import org.drools.core.reteoo.PathEndNode;
-import org.drools.core.reteoo.PathMemory;
-import org.drools.core.reteoo.SegmentMemory;
-import org.drools.core.reteoo.Tuple;
-
-public class TupleEvaluationUtil {
-
- public static boolean flushLeftTupleIfNecessary(ReteEvaluator reteEvaluator, SegmentMemory sm, boolean streamMode) {
- return flushLeftTupleIfNecessary(reteEvaluator, sm, null, streamMode, Tuple.NONE);
- }
-
- public static boolean flushLeftTupleIfNecessary(ReteEvaluator reteEvaluator,
- SegmentMemory sm,
- TupleImpl leftTuple,
- boolean streamMode,
- short stagedType) {
- PathMemory pmem = findPathToFlush(sm, leftTuple, streamMode);
-
- if (pmem == null) {
- return false;
- }
-
- forceFlushLeftTuple(reteEvaluator, pmem, sm, createLeftTupleTupleSets(leftTuple, stagedType));
- forceFlushWhenSubnetwork(reteEvaluator, pmem);
- return true;
- }
-
- public static PathMemory findPathToFlush(SegmentMemory sm, TupleImpl leftTuple, boolean streamMode) {
- boolean forceFlush = streamMode || (leftTuple != null && leftTuple.getFactHandle() != null && leftTuple
- .getFactHandle().isEvent());
- return forceFlush ? sm.getPathMemories().get(0) : sm.getFirstDataDrivenPathMemory();
- }
-
- public static TupleSets createLeftTupleTupleSets(TupleImpl leftTuple, short stagedType) {
- TupleSets leftTupleSets = new TupleSetsImpl();
- if (leftTuple != null) {
- switch (stagedType) {
- case Tuple.INSERT:
- leftTupleSets.addInsert(leftTuple);
- break;
- case Tuple.DELETE:
- leftTupleSets.addDelete(leftTuple);
- break;
- case Tuple.UPDATE:
- leftTupleSets.addUpdate(leftTuple);
- break;
- }
- }
- return leftTupleSets;
- }
-
- public static void forceFlushWhenSubnetwork(ReteEvaluator reteEvaluator, PathMemory pmem) {
- for (PathMemory outPmem : findPathsToFlushFromSubnetwork(reteEvaluator, pmem)) {
- forceFlushPath(reteEvaluator, outPmem);
- }
- }
-
- public static List<PathMemory> findPathsToFlushFromSubnetwork(ReteEvaluator reteEvaluator, PathMemory pmem) {
- List<PathMemory> paths = null;
- if (pmem.isDataDriven() && pmem.getNodeType() == NodeTypeEnums.TupleToObjectNode) {
- for (PathEndNode pnode : pmem.getPathEndNode().getPathEndNodes()) {
- if (NodeTypeEnums.isTerminalNode(pnode)) {
- PathMemory outPmem = reteEvaluator.getNodeMemory(pnode);
- if (outPmem.isDataDriven()) {
- if (paths == null) {
- paths = new ArrayList<>();
- }
- paths.add(outPmem);
- }
- }
- }
- }
- return paths == null ? Collections.emptyList() : paths;
- }
-
- public static void forceFlushPath(ReteEvaluator reteEvaluator, PathMemory outPmem) {
- SegmentMemory outSmem = outPmem.getSegmentMemories()[0];
- if (outSmem != null) {
- forceFlushLeftTuple(reteEvaluator, outPmem, outSmem, new TupleSetsImpl());
- }
- }
-
- public static void forceFlushLeftTuple(ReteEvaluator reteEvaluator,
- PathMemory pmem,
- SegmentMemory sm,
- TupleSets leftTupleSets) {
- SegmentMemory[] smems = pmem.getSegmentMemories();
-
- LeftTupleNode node;
- Memory mem;
- long bit = 1;
- if (NodeTypeEnums.isLeftInputAdapterNode(sm.getRootNode()) && !NodeTypeEnums.isLeftInputAdapterNode(sm
- .getTipNode())) {
- // The segment is the first and it has the lian shared with other nodes, the lian must be skipped, so adjust the bit and sink
- node = sm.getRootNode().getSinkPropagator().getFirstLeftTupleSink();
- mem = sm.getNodeMemories()[1];
- bit = 2; // adjust bit to point to next node
- } else {
- node = sm.getRootNode();
- mem = sm.getNodeMemories()[0];
- }
-
- PathMemory rtnPmem = NodeTypeEnums.isTerminalNode(pmem.getPathEndNode()) ? pmem : reteEvaluator.getNodeMemory(
- (AbstractTerminalNode) pmem.getPathEndNode().getPathEndNodes()[0]);
-
- ActivationsManager activationsManager = pmem.getActualActivationsManager();
- RuleNetworkEvaluator.INSTANCE.outerEval(activationsManager, rtnPmem.getOrCreateRuleAgendaItem().getRuleExecutor(), pmem, smems, sm.getPos(), bit, mem, node,
- leftTupleSets,
- true);
- }
-}
diff --git a/drools-core/src/main/java/org/drools/core/reteoo/LeftInputAdapterNode.java b/drools-core/src/main/java/org/drools/core/reteoo/LeftInputAdapterNode.java
index a5368ff34e..7cf58f76cf 100644
--- a/drools-core/src/main/java/org/drools/core/reteoo/LeftInputAdapterNode.java
+++ b/drools-core/src/main/java/org/drools/core/reteoo/LeftInputAdapterNode.java
@@ -40,8 +40,10 @@ import org.drools.core.common.PropagationContext;
import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.SuperCacheFixer;
import org.drools.core.common.TupleSets;
+import org.drools.core.common.TupleSetsImpl;
import org.drools.core.common.UpdateContext;
import org.drools.core.phreak.DetachedTuple;
+import org.drools.core.phreak.RuleNetworkEvaluatorImpl;
import org.drools.core.reteoo.builder.BuildContext;
import org.drools.core.rule.consequence.InternalMatch;
import org.drools.core.util.AbstractLinkedListNode;
@@ -52,12 +54,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static org.drools.base.reteoo.PropertySpecificUtil.isPropertyReactive;
-import static org.drools.core.phreak.TupleEvaluationUtil.createLeftTupleTupleSets;
-import static org.drools.core.phreak.TupleEvaluationUtil.findPathToFlush;
-import static org.drools.core.phreak.TupleEvaluationUtil.findPathsToFlushFromSubnetwork;
-import static org.drools.core.phreak.TupleEvaluationUtil.flushLeftTupleIfNecessary;
-import static org.drools.core.phreak.TupleEvaluationUtil.forceFlushLeftTuple;
-import static org.drools.core.phreak.TupleEvaluationUtil.forceFlushPath;
/**
* All asserting Facts must propagated into the right <code>ObjectSink</code> side of a BetaNode, if this is the first Pattern
@@ -228,25 +224,25 @@ public class LeftInputAdapterNode extends LeftTupleSource
}
for (PathMemory outPmem : pathsToFlush) {
- forceFlushPath(reteEvaluator, outPmem);
+ reteEvaluator.getRuleNetworkEvaluator().forceFlushPath(outPmem);
}
}
}
public static void doInsertSegmentMemoryWithFlush(ReteEvaluator reteEvaluator, boolean notifySegment, LiaNodeMemory lm, SegmentMemory sm, TupleImpl leftTuple, boolean streamMode) {
for (PathMemory outPmem : doInsertSegmentMemory(reteEvaluator, notifySegment, lm, sm, leftTuple, streamMode )) {
- forceFlushPath(reteEvaluator, outPmem);
+ reteEvaluator.getRuleNetworkEvaluator().forceFlushPath(outPmem);
}
}
public static List<PathMemory> doInsertSegmentMemory(ReteEvaluator reteEvaluator, boolean linkOrNotify, LiaNodeMemory lm, SegmentMemory sm, TupleImpl leftTuple, boolean streamMode) {
- PathMemory pmem = findPathToFlush(sm, leftTuple, streamMode);
+ PathMemory pmem = RuleNetworkEvaluatorImpl.findPathToFlush(sm, leftTuple, streamMode);
if ( pmem != null ) {
- forceFlushLeftTuple( reteEvaluator, pmem, sm, createLeftTupleTupleSets(leftTuple, Tuple.INSERT) );
+ reteEvaluator.getRuleNetworkEvaluator().forceFlushLeftTuple(pmem, sm, TupleSetsImpl.createLeftTupleTupleSets(leftTuple, Tuple.INSERT));
if ( linkOrNotify ) {
lm.setNodeDirty( );
}
- return findPathsToFlushFromSubnetwork(reteEvaluator, pmem);
+ return reteEvaluator.getRuleNetworkEvaluator().findPathsToFlushFromSubnetwork(pmem);
}
// mask check is necessary if insert is a result of a modify
@@ -301,7 +297,7 @@ public class LeftInputAdapterNode extends LeftTupleSource
private static void doDeleteSegmentMemory(TupleImpl leftTuple, PropagationContext pctx, final LiaNodeMemory lm,
SegmentMemory sm, ReteEvaluator reteEvaluator, boolean linkOrNotify, boolean streamMode) {
leftTuple.setPropagationContext( pctx );
- if ( flushLeftTupleIfNecessary( reteEvaluator, sm, leftTuple, streamMode, Tuple.DELETE ) ) {
+ if ( reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(sm, leftTuple, streamMode, Tuple.DELETE) ) {
if ( linkOrNotify ) {
lm.setNodeDirty( );
}
@@ -351,7 +347,7 @@ public class LeftInputAdapterNode extends LeftTupleSource
TupleSets leftTuples = sm.getStagedLeftTuples();
if ( leftTuple.getStagedType() == LeftTuple.NONE ) {
- if ( flushLeftTupleIfNecessary( reteEvaluator, sm, leftTuple, streamMode, Tuple.UPDATE ) ) {
+ if ( reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(sm, leftTuple, streamMode, Tuple.UPDATE) ) {
if ( linkOrNotify ) {
lm.setNodeDirty( );
}
diff --git a/drools-core/src/main/java/org/drools/core/reteoo/NotRight.java b/drools-core/src/main/java/org/drools/core/reteoo/NotRight.java
index b3ddda7069..67e9353567 100644
--- a/drools-core/src/main/java/org/drools/core/reteoo/NotRight.java
+++ b/drools-core/src/main/java/org/drools/core/reteoo/NotRight.java
@@ -25,8 +25,7 @@ import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.reteoo.builder.BuildContext;
-import static org.drools.core.phreak.RuleNetworkEvaluator.doExistentialUpdatesReorderChildLeftTuple;
-import static org.drools.core.phreak.TupleEvaluationUtil.flushLeftTupleIfNecessary;
+import static org.drools.core.phreak.PhreakNodeOperations.doExistentialUpdatesReorderChildLeftTuple;
public class NotRight extends RightInputAdapterNode<NotNode> {
@@ -36,7 +35,7 @@ public class NotRight extends RightInputAdapterNode<NotNode> {
@Override
protected void reorderRightTuple(ReteEvaluator reteEvaluator, TupleImpl rightTuple) {
- doExistentialUpdatesReorderChildLeftTuple(reteEvaluator, (NotNode) betaNode, (RightTuple) rightTuple);
+ doExistentialUpdatesReorderChildLeftTuple(reteEvaluator, betaNode, (RightTuple) rightTuple);
}
@Override
@@ -69,7 +68,7 @@ public class NotRight extends RightInputAdapterNode<NotNode> {
memory.setNodeDirty(betaNode, reteEvaluator);
}
- flushLeftTupleIfNecessary( reteEvaluator, memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode() );
+ reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode());
}
@Override
@@ -104,7 +103,7 @@ public class NotRight extends RightInputAdapterNode<NotNode> {
memory.setNodeDirty( betaNode, reteEvaluator );
}
- flushLeftTupleIfNecessary( reteEvaluator, memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode() );
+ reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode());
}
@Override
diff --git a/drools-core/src/main/java/org/drools/core/reteoo/RightInputAdapterNode.java b/drools-core/src/main/java/org/drools/core/reteoo/RightInputAdapterNode.java
index 267ae40acb..79c0ac3704 100644
--- a/drools-core/src/main/java/org/drools/core/reteoo/RightInputAdapterNode.java
+++ b/drools-core/src/main/java/org/drools/core/reteoo/RightInputAdapterNode.java
@@ -42,8 +42,8 @@ import java.util.List;
import java.util.Objects;
import static org.drools.base.reteoo.PropertySpecificUtil.isPropertyReactive;
-import static org.drools.core.phreak.RuleNetworkEvaluator.doUpdatesReorderChildLeftTuple;
-import static org.drools.core.phreak.TupleEvaluationUtil.flushLeftTupleIfNecessary;
+import static org.drools.core.phreak.PhreakNodeOperations.doUpdatesReorderChildLeftTuple;
+import static org.drools.core.reteoo.BetaNode.getBetaMemory;
public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNode
implements ObjectSinkNode,
@@ -120,7 +120,7 @@ public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNod
}
if (shouldFlush) {
- flushLeftTupleIfNecessary( reteEvaluator, memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode() );
+ reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode());
}
}
@@ -133,7 +133,7 @@ public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNod
// we skipped this node, due to alpha hashing, so retract now
rightTuple.setPropagationContext( context );
- BetaMemory bm = BetaNode.getBetaMemory(((RightInputAdapterNode)rightTuple.getSink()).getBetaNode(), reteEvaluator);
+ BetaMemory bm = getBetaMemory(((RightInputAdapterNode)rightTuple.getSink()).getBetaNode(), reteEvaluator);
((RightInputAdapterNode) rightTuple.getSink()).doDeleteRightTuple(rightTuple, reteEvaluator, bm);
rightTuple = modifyPreviousTuples.peekRightTuple(partitionId);
}
@@ -145,7 +145,7 @@ public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNod
// RightTuple previously existed, so continue as modify
rightTuple.setPropagationContext( context ); // only update, if the mask intersects
- BetaMemory bm = BetaNode.getBetaMemory(betaNode, reteEvaluator);
+ BetaMemory bm = getBetaMemory(betaNode, reteEvaluator);
rightTuple.setPropagationContext( context );
doUpdateRightTuple(rightTuple, reteEvaluator, bm);
} else if (rightTuple.getMemory() != null) {
@@ -160,7 +160,7 @@ public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNod
}
protected void reorderRightTuple(ReteEvaluator reteEvaluator, TupleImpl rightTuple) {
- BetaNode.getBetaMemory(betaNode, reteEvaluator).getRightTupleMemory().removeAdd(rightTuple);
+ getBetaMemory(betaNode, reteEvaluator).getRightTupleMemory().removeAdd(rightTuple);
doUpdatesReorderChildLeftTuple(rightTuple);
}
@@ -183,7 +183,7 @@ public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNod
}
if (shouldFlush) {
- flushLeftTupleIfNecessary( reteEvaluator, memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode() );
+ reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode());
}
}
@@ -200,7 +200,7 @@ public abstract class RightInputAdapterNode<T extends BetaNode> extends BaseNod
}
if (shouldFlush) {
- flushLeftTupleIfNecessary( reteEvaluator, memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode() );
+ reteEvaluator.getRuleNetworkEvaluator().flushLeftTupleIfNecessary(memory.getOrCreateSegmentMemory( betaNode, reteEvaluator ), isStreamMode());
}
}
diff --git a/drools-kiesession/src/main/java/org/drools/kiesession/consequence/StatefulKnowledgeSessionForRHS.java b/drools-kiesession/src/main/java/org/drools/kiesession/consequence/StatefulKnowledgeSessionForRHS.java
index f4ddcb1a55..b077395b8b 100644
--- a/drools-kiesession/src/main/java/org/drools/kiesession/consequence/StatefulKnowledgeSessionForRHS.java
+++ b/drools-kiesession/src/main/java/org/drools/kiesession/consequence/StatefulKnowledgeSessionForRHS.java
@@ -46,6 +46,7 @@ import org.drools.core.event.AgendaEventSupport;
import org.drools.core.event.RuleEventListenerSupport;
import org.drools.core.event.RuleRuntimeEventSupport;
import org.drools.core.phreak.PropagationEntry;
+import org.drools.core.phreak.RuleNetworkEvaluator;
import org.drools.core.reteoo.EntryPointNode;
import org.drools.core.reteoo.TerminalNode;
import org.drools.core.rule.accessor.FactHandleFactory;
@@ -501,6 +502,11 @@ public class StatefulKnowledgeSessionForRHS
public GlobalResolver getGlobalResolver() {
return delegate.getGlobalResolver();
}
+
+ @Override
+ public RuleNetworkEvaluator getRuleNetworkEvaluator() {
+ return delegate.getRuleNetworkEvaluator();
+ }
@Override
public ActivationsManager getActivationsManager() {
diff --git a/drools-kiesession/src/main/java/org/drools/kiesession/session/StatefulKnowledgeSessionImpl.java b/drools-kiesession/src/main/java/org/drools/kiesession/session/StatefulKnowledgeSessionImpl.java
index 11dc3d32bd..14479ab79c 100644
--- a/drools-kiesession/src/main/java/org/drools/kiesession/session/StatefulKnowledgeSessionImpl.java
+++ b/drools-kiesession/src/main/java/org/drools/kiesession/session/StatefulKnowledgeSessionImpl.java
@@ -66,6 +66,8 @@ import org.drools.core.management.DroolsManagementAgent;
import org.drools.core.marshalling.MarshallerReaderContext;
import org.drools.core.phreak.PropagationEntry;
import org.drools.core.phreak.RuleAgendaItem;
+import org.drools.core.phreak.RuleNetworkEvaluator;
+import org.drools.core.phreak.RuleNetworkEvaluatorImpl;
import org.drools.core.phreak.SegmentMemorySupportImpl;
import org.drools.core.reteoo.AsyncReceiveNode;
import org.drools.core.reteoo.EntryPointNode;
@@ -173,6 +175,8 @@ public class StatefulKnowledgeSessionImpl extends AbstractRuntime
public byte[] bytes;
protected Long id;
+ private RuleNetworkEvaluator ruleNetworkEvaluator;
+
/** The actual memory for the <code>JoinNode</code>s. */
private NodeMemories nodeMemories;
@@ -354,6 +358,8 @@ public class StatefulKnowledgeSessionImpl extends AbstractRuntime
this.sequential = conf.isSequential();
this.globalResolver = RuntimeComponentFactory.get().createGlobalResolver(this, this.environment);
+
+ this.ruleNetworkEvaluator = new RuleNetworkEvaluatorImpl(this, nodeMemories, segmentMemorySupport);
if (initInitFactHandle) {
this.initialFactHandle = initInitialFact(null);
@@ -398,6 +404,10 @@ public class StatefulKnowledgeSessionImpl extends AbstractRuntime
return runtimeFactory.get(cls);
}
+ public RuleNetworkEvaluator getRuleNetworkEvaluator() {
+ return ruleNetworkEvaluator;
+ }
+
public WorkingMemoryEntryPoint getEntryPoint(String name) {
return this.entryPointsManager.getEntryPoint(name);
}
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/MetricPhreakNetworkNodeFactoryImpl.java b/drools-metric/src/main/java/org/drools/metric/phreak/MetricPhreakNetworkNodeFactoryImpl.java
index e3218923ea..67475c9920 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/MetricPhreakNetworkNodeFactoryImpl.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/MetricPhreakNetworkNodeFactoryImpl.java
@@ -18,6 +18,7 @@
*/
package org.drools.metric.phreak;
+import org.drools.core.common.ReteEvaluator;
import org.drools.core.phreak.PhreakAccumulateNode;
import org.drools.core.phreak.PhreakAsyncReceiveNode;
import org.drools.core.phreak.PhreakAsyncSendNode;
@@ -39,133 +40,133 @@ import org.drools.metric.util.MetricLogUtils;
public class MetricPhreakNetworkNodeFactoryImpl implements PhreakNetworkNodeFactory {
@Override
- public PhreakJoinNode createPhreakJoinNode() {
+ public PhreakJoinNode createPhreakJoinNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakJoinNodeMetric();
+ return new PhreakJoinNodeMetric(reteEvaluator);
} else {
- return new PhreakJoinNode();
+ return new PhreakJoinNode(reteEvaluator);
}
}
@Override
- public PhreakEvalNode createPhreakEvalNode() {
+ public PhreakEvalNode createPhreakEvalNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakEvalNodeMetric();
+ return new PhreakEvalNodeMetric(reteEvaluator);
} else {
- return new PhreakEvalNode();
+ return new PhreakEvalNode(reteEvaluator);
}
}
@Override
- public PhreakFromNode createPhreakFromNode() {
+ public PhreakFromNode createPhreakFromNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakFromNodeMetric();
+ return new PhreakFromNodeMetric(reteEvaluator);
} else {
- return new PhreakFromNode();
+ return new PhreakFromNode(reteEvaluator);
}
}
@Override
- public PhreakReactiveFromNode createPhreakReactiveFromNode() {
+ public PhreakReactiveFromNode createPhreakReactiveFromNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakReactiveFromNodeMetric();
+ return new PhreakReactiveFromNodeMetric(reteEvaluator);
} else {
- return new PhreakReactiveFromNode();
+ return new PhreakReactiveFromNode(reteEvaluator);
}
}
@Override
- public PhreakNotNode createPhreakNotNode() {
+ public PhreakNotNode createPhreakNotNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakNotNodeMetric();
+ return new PhreakNotNodeMetric(reteEvaluator);
} else {
- return new PhreakNotNode();
+ return new PhreakNotNode(reteEvaluator);
}
}
@Override
- public PhreakExistsNode createPhreakExistsNode() {
+ public PhreakExistsNode createPhreakExistsNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakExistsNodeMetric();
+ return new PhreakExistsNodeMetric(reteEvaluator);
} else {
- return new PhreakExistsNode();
+ return new PhreakExistsNode(reteEvaluator);
}
}
@Override
- public PhreakAccumulateNode createPhreakAccumulateNode() {
+ public PhreakAccumulateNode createPhreakAccumulateNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakAccumulateNodeMetric();
+ return new PhreakAccumulateNodeMetric(reteEvaluator);
} else {
- return new PhreakAccumulateNode();
+ return new PhreakAccumulateNode(reteEvaluator);
}
}
@Override
- public PhreakGroupByNode createPhreakGroupByNode() {
+ public PhreakGroupByNode createPhreakGroupByNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakGroupByNodeMetric();
+ return new PhreakGroupByNodeMetric(reteEvaluator);
} else {
- return new PhreakGroupByNode();
+ return new PhreakGroupByNode(reteEvaluator);
}
}
@Override
- public PhreakBranchNode createPhreakBranchNode() {
+ public PhreakBranchNode createPhreakBranchNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakBranchNodeMetric();
+ return new PhreakBranchNodeMetric(reteEvaluator);
} else {
- return new PhreakBranchNode();
+ return new PhreakBranchNode(reteEvaluator);
}
}
@Override
- public PhreakQueryNode createPhreakQueryNode() {
+ public PhreakQueryNode createPhreakQueryNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakQueryNodeMetric();
+ return new PhreakQueryNodeMetric(reteEvaluator);
} else {
- return new PhreakQueryNode();
+ return new PhreakQueryNode(reteEvaluator);
}
}
@Override
- public PhreakTimerNode createPhreakTimerNode() {
+ public PhreakTimerNode createPhreakTimerNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakTimerNodeMetric();
+ return new PhreakTimerNodeMetric(reteEvaluator);
} else {
- return new PhreakTimerNode();
+ return new PhreakTimerNode(reteEvaluator);
}
}
@Override
- public PhreakAsyncSendNode createPhreakAsyncSendNode() {
+ public PhreakAsyncSendNode createPhreakAsyncSendNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakAsyncSendNodeMetric();
+ return new PhreakAsyncSendNodeMetric(reteEvaluator);
} else {
- return new PhreakAsyncSendNode();
+ return new PhreakAsyncSendNode(reteEvaluator);
}
}
@Override
- public PhreakAsyncReceiveNode createPhreakAsyncReceiveNode() {
+ public PhreakAsyncReceiveNode createPhreakAsyncReceiveNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakAsyncReceiveNodeMetric();
+ return new PhreakAsyncReceiveNodeMetric(reteEvaluator);
} else {
- return new PhreakAsyncReceiveNode();
+ return new PhreakAsyncReceiveNode(reteEvaluator);
}
}
@Override
- public PhreakRuleTerminalNode createPhreakRuleTerminalNode() {
- return new PhreakRuleTerminalNode(); // TerminalNode is not BaseNode so cannot use PerfLogUtils
+ public PhreakRuleTerminalNode createPhreakRuleTerminalNode(ReteEvaluator reteEvaluator) {
+ return new PhreakRuleTerminalNode(reteEvaluator); // TerminalNode is not BaseNode so cannot use PerfLogUtils
}
@Override
- public PhreakQueryTerminalNode createPhreakQueryTerminalNode() {
+ public PhreakQueryTerminalNode createPhreakQueryTerminalNode(ReteEvaluator reteEvaluator) {
if (MetricLogUtils.getInstance().isEnabled()) {
- return new PhreakQueryTerminalNodeMetric();
+ return new PhreakQueryTerminalNodeMetric(reteEvaluator);
} else {
- return new PhreakQueryTerminalNode();
+ return new PhreakQueryTerminalNode(reteEvaluator);
}
}
}
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAccumulateNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAccumulateNodeMetric.java
index c54e133a6c..e23aa8690e 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAccumulateNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAccumulateNodeMetric.java
@@ -23,15 +23,17 @@ import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakAccumulateNode;
import org.drools.core.reteoo.AccumulateNode;
import org.drools.core.reteoo.AccumulateNode.AccumulateMemory;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.metric.util.MetricLogUtils;
public class PhreakAccumulateNodeMetric extends PhreakAccumulateNode {
+ public PhreakAccumulateNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- AccumulateNode accNode,
+ public void doNode(AccumulateNode accNode,
LeftTupleSink sink,
AccumulateMemory am,
TupleSets srcLeftTuples,
@@ -41,7 +43,7 @@ public class PhreakAccumulateNodeMetric extends PhreakAccumulateNode {
try {
MetricLogUtils.getInstance().startMetrics(accNode);
- super.doNode(reteEvaluator, accNode, sink, am, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(accNode, sink, am, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncReceiveNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncReceiveNodeMetric.java
index 8a3cb23bb3..bc7954befd 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncReceiveNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncReceiveNodeMetric.java
@@ -23,15 +23,17 @@ import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakAsyncReceiveNode;
import org.drools.core.reteoo.AsyncReceiveNode;
import org.drools.core.reteoo.AsyncReceiveNode.AsyncReceiveMemory;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.metric.util.MetricLogUtils;
public class PhreakAsyncReceiveNodeMetric extends PhreakAsyncReceiveNode {
+ public PhreakAsyncReceiveNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- AsyncReceiveNode node,
+ public void doNode(AsyncReceiveNode node,
AsyncReceiveMemory memory,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -40,7 +42,7 @@ public class PhreakAsyncReceiveNodeMetric extends PhreakAsyncReceiveNode {
try {
MetricLogUtils.getInstance().startMetrics(node);
- super.doNode(reteEvaluator, node, memory, sink, srcLeftTuples, trgLeftTuples);
+ super.doNode(node, memory, sink, srcLeftTuples, trgLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncSendNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncSendNodeMetric.java
index dd32849ec2..2d61b6a76e 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncSendNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakAsyncSendNodeMetric.java
@@ -23,21 +23,23 @@ import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakAsyncSendNode;
import org.drools.core.reteoo.AsyncSendNode;
import org.drools.core.reteoo.AsyncSendNode.AsyncSendMemory;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.metric.util.MetricLogUtils;
public class PhreakAsyncSendNodeMetric extends PhreakAsyncSendNode {
+ public PhreakAsyncSendNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- AsyncSendNode node,
+ public void doNode(AsyncSendNode node,
AsyncSendMemory memory,
TupleSets srcLeftTuples) {
try {
MetricLogUtils.getInstance().startMetrics(node);
- super.doNode(reteEvaluator, node, memory, srcLeftTuples);
+ super.doNode(node, memory, srcLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakBranchNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakBranchNodeMetric.java
index 50d0817409..12db5d3f16 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakBranchNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakBranchNodeMetric.java
@@ -19,17 +19,21 @@
package org.drools.metric.phreak;
import org.drools.core.common.ActivationsManager;
+import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakBranchNode;
import org.drools.core.phreak.RuleExecutor;
import org.drools.core.reteoo.ConditionalBranchNode;
import org.drools.core.reteoo.ConditionalBranchNode.ConditionalBranchMemory;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.metric.util.MetricLogUtils;
public class PhreakBranchNodeMetric extends PhreakBranchNode {
+ public PhreakBranchNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
public void doNode(ActivationsManager activationsManager,
ConditionalBranchNode branchNode,
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakEvalNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakEvalNodeMetric.java
index f867a474f9..732b43a36a 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakEvalNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakEvalNodeMetric.java
@@ -23,15 +23,17 @@ import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakEvalNode;
import org.drools.core.reteoo.EvalConditionNode;
import org.drools.core.reteoo.EvalConditionNode.EvalMemory;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.metric.util.MetricLogUtils;
public class PhreakEvalNodeMetric extends PhreakEvalNode {
+ public PhreakEvalNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- EvalConditionNode evalNode,
+ public void doNode(EvalConditionNode evalNode,
EvalMemory em,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -41,7 +43,7 @@ public class PhreakEvalNodeMetric extends PhreakEvalNode {
try {
MetricLogUtils.getInstance().startMetrics(evalNode);
- super.doNode(reteEvaluator, evalNode, em, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(evalNode, em, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakExistsNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakExistsNodeMetric.java
index d559776b66..a4dd925f2e 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakExistsNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakExistsNodeMetric.java
@@ -28,9 +28,12 @@ import org.drools.metric.util.MetricLogUtils;
public class PhreakExistsNodeMetric extends PhreakExistsNode {
+ public PhreakExistsNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- ExistsNode existsNode,
+ public void doNode(ExistsNode existsNode,
LeftTupleSink sink,
BetaMemory bm,
TupleSets srcLeftTuples,
@@ -40,7 +43,7 @@ public class PhreakExistsNodeMetric extends PhreakExistsNode {
try {
MetricLogUtils.getInstance().startMetrics(existsNode);
- super.doNode(reteEvaluator, existsNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(existsNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakFromNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakFromNodeMetric.java
index bfa84666e1..e5895305fc 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakFromNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakFromNodeMetric.java
@@ -29,9 +29,12 @@ import org.drools.metric.util.MetricLogUtils;
public class PhreakFromNodeMetric extends PhreakFromNode {
+ public PhreakFromNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- FromNode fromNode,
+ public void doNode(FromNode fromNode,
FromMemory fm,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -41,7 +44,7 @@ public class PhreakFromNodeMetric extends PhreakFromNode {
try {
MetricLogUtils.getInstance().startMetrics(fromNode);
- super.doNode(reteEvaluator, fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakGroupByNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakGroupByNodeMetric.java
index 34fc58bd15..f3b937aeb2 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakGroupByNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakGroupByNodeMetric.java
@@ -22,15 +22,17 @@ import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakGroupByNode;
import org.drools.core.reteoo.AccumulateNode;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.metric.util.MetricLogUtils;
public class PhreakGroupByNodeMetric extends PhreakGroupByNode {
+ public PhreakGroupByNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode( ReteEvaluator reteEvaluator,
- AccumulateNode accNode,
+ public void doNode( AccumulateNode accNode,
LeftTupleSink sink,
AccumulateNode.AccumulateMemory am,
TupleSets srcLeftTuples,
@@ -40,7 +42,7 @@ public class PhreakGroupByNodeMetric extends PhreakGroupByNode {
try {
MetricLogUtils.getInstance().startMetrics(accNode);
- super.doNode(reteEvaluator, accNode, sink, am, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(accNode, sink, am, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakJoinNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakJoinNodeMetric.java
index 826197cef8..6dad21dd1f 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakJoinNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakJoinNodeMetric.java
@@ -28,9 +28,12 @@ import org.drools.metric.util.MetricLogUtils;
public class PhreakJoinNodeMetric extends PhreakJoinNode {
+ public PhreakJoinNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- JoinNode joinNode,
+ public void doNode(JoinNode joinNode,
LeftTupleSink sink,
BetaMemory bm,
TupleSets srcLeftTuples,
@@ -39,7 +42,7 @@ public class PhreakJoinNodeMetric extends PhreakJoinNode {
try {
MetricLogUtils.getInstance().startMetrics(joinNode);
- super.doNode(reteEvaluator, joinNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(joinNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakNotNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakNotNodeMetric.java
index 9c9d2e9da2..0a4594f3a7 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakNotNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakNotNodeMetric.java
@@ -28,9 +28,12 @@ import org.drools.metric.util.MetricLogUtils;
public class PhreakNotNodeMetric extends PhreakNotNode {
+ public PhreakNotNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- NotNode notNode,
+ public void doNode(NotNode notNode,
LeftTupleSink sink,
BetaMemory bm,
TupleSets srcLeftTuples,
@@ -40,7 +43,7 @@ public class PhreakNotNodeMetric extends PhreakNotNode {
try {
MetricLogUtils.getInstance().startMetrics(notNode);
- super.doNode(reteEvaluator, notNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(notNode, sink, bm, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryNodeMetric.java
index d5733255bd..578672627d 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryNodeMetric.java
@@ -22,18 +22,20 @@ import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakQueryNode;
import org.drools.core.phreak.StackEntry;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.QueryElementNode;
import org.drools.core.reteoo.QueryElementNode.QueryElementNodeMemory;
import org.drools.metric.util.MetricLogUtils;
public class PhreakQueryNodeMetric extends PhreakQueryNode {
+ public PhreakQueryNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
public void doNode(QueryElementNode queryNode,
QueryElementNodeMemory qmem,
StackEntry stackEntry,
- ReteEvaluator reteEvaluator,
TupleSets srcLeftTuples,
TupleSets trgLeftTuples,
TupleSets stagedLeftTuples) {
@@ -41,7 +43,7 @@ public class PhreakQueryNodeMetric extends PhreakQueryNode {
try {
MetricLogUtils.getInstance().startMetrics(queryNode);
- super.doNode(queryNode, qmem, stackEntry, reteEvaluator, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(queryNode, qmem, stackEntry, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryTerminalNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryTerminalNodeMetric.java
index c7decec097..8c6a54900e 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryTerminalNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakQueryTerminalNodeMetric.java
@@ -18,27 +18,29 @@
*/
package org.drools.metric.phreak;
-import org.drools.core.common.ActivationsManager;
+import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakQueryTerminalNode;
import org.drools.core.phreak.StackEntry;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.QueryTerminalNode;
import org.drools.core.util.LinkedList;
import org.drools.metric.util.MetricLogUtils;
public class PhreakQueryTerminalNodeMetric extends PhreakQueryTerminalNode {
+ public PhreakQueryTerminalNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ActivationsManager activationsManager,
- LinkedList<StackEntry> stack,
+ public void doNode(LinkedList<StackEntry> stack,
QueryTerminalNode qtnNode,
TupleSets srcLeftTuples) {
try {
MetricLogUtils.getInstance().startMetrics(qtnNode);
- super.doNode(activationsManager, stack, qtnNode, srcLeftTuples);
+ super.doNode(stack, qtnNode, srcLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakReactiveFromNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakReactiveFromNodeMetric.java
index fc55833a50..76d1d62704 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakReactiveFromNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakReactiveFromNodeMetric.java
@@ -21,7 +21,6 @@ package org.drools.metric.phreak;
import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakReactiveFromNode;
-import org.drools.core.reteoo.LeftTuple;
import org.drools.core.reteoo.LeftTupleSink;
import org.drools.core.reteoo.ReactiveFromNode;
import org.drools.core.reteoo.ReactiveFromNode.ReactiveFromMemory;
@@ -29,9 +28,12 @@ import org.drools.metric.util.MetricLogUtils;
public class PhreakReactiveFromNodeMetric extends PhreakReactiveFromNode {
+ public PhreakReactiveFromNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
- public void doNode(ReteEvaluator reteEvaluator,
- ReactiveFromNode fromNode,
+ public void doNode(ReactiveFromNode fromNode,
ReactiveFromMemory fm,
LeftTupleSink sink,
TupleSets srcLeftTuples,
@@ -41,7 +43,7 @@ public class PhreakReactiveFromNodeMetric extends PhreakReactiveFromNode {
try {
MetricLogUtils.getInstance().startMetrics(fromNode);
- super.doNode(reteEvaluator, fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
+ super.doNode(fromNode, fm, sink, srcLeftTuples, trgLeftTuples, stagedLeftTuples);
} finally {
MetricLogUtils.getInstance().logAndEndMetrics();
diff --git a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakTimerNodeMetric.java b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakTimerNodeMetric.java
index d1cf00a9ed..be0d215885 100644
--- a/drools-metric/src/main/java/org/drools/metric/phreak/PhreakTimerNodeMetric.java
+++ b/drools-metric/src/main/java/org/drools/metric/phreak/PhreakTimerNodeMetric.java
@@ -19,6 +19,7 @@
package org.drools.metric.phreak;
import org.drools.core.common.ActivationsManager;
+import org.drools.core.common.ReteEvaluator;
import org.drools.core.common.TupleSets;
import org.drools.core.phreak.PhreakTimerNode;
import org.drools.core.reteoo.LeftTuple;
@@ -31,6 +32,10 @@ import org.drools.metric.util.MetricLogUtils;
public class PhreakTimerNodeMetric extends PhreakTimerNode {
+ public PhreakTimerNodeMetric(ReteEvaluator reteEvaluator) {
+ super(reteEvaluator);
+ }
+
@Override
public void doNode(TimerNode timerNode,
TimerNodeMemory tm,
diff --git a/drools-ruleunits/drools-ruleunits-impl/src/main/java/org/drools/ruleunits/impl/sessions/RuleUnitExecutorImpl.java b/drools-ruleunits/drools-ruleunits-impl/src/main/java/org/drools/ruleunits/impl/sessions/RuleUnitExecutorImpl.java
index b190b6b07a..968acd5e18 100644
--- a/drools-ruleunits/drools-ruleunits-impl/src/main/java/org/drools/ruleunits/impl/sessions/RuleUnitExecutorImpl.java
+++ b/drools-ruleunits/drools-ruleunits-impl/src/main/java/org/drools/ruleunits/impl/sessions/RuleUnitExecutorImpl.java
@@ -54,6 +54,8 @@ import org.drools.core.event.RuleRuntimeEventSupport;
import org.drools.core.impl.ActivationsManagerImpl;
import org.drools.core.impl.InternalRuleBase;
import org.drools.core.phreak.PropagationEntry;
+import org.drools.core.phreak.RuleNetworkEvaluator;
+import org.drools.core.phreak.RuleNetworkEvaluatorImpl;
import org.drools.core.phreak.SegmentMemorySupportImpl;
import org.drools.core.reteoo.ObjectTypeNode;
import org.drools.core.reteoo.RuntimeComponentFactory;
@@ -114,7 +116,9 @@ public class RuleUnitExecutorImpl implements ReteEvaluator {
private final GlobalResolver globalResolver = new MapGlobalResolver();
private final TimerService timerService;
-
+
+ private final RuleNetworkEvaluator ruleNetworkEvaluator;
+
private Calendars calendars;
private RuleUnits ruleUnits;
@@ -138,6 +142,7 @@ public class RuleUnitExecutorImpl implements ReteEvaluator {
this.segmentMemorySupport = new SegmentMemorySupportImpl(nodeMemories, ruleBase.getSegmentPrototypeRegistry(), entryPointsManager.getDefaultEntryPoint());
this.timerService = sessionConfiguration.createTimerService();
+ this.ruleNetworkEvaluator = new RuleNetworkEvaluatorImpl(this, nodeMemories, segmentMemorySupport);
initInitialFact();
}
@@ -160,6 +165,11 @@ public class RuleUnitExecutorImpl implements ReteEvaluator {
return this.identifier;
}
+ @Override
+ public RuleNetworkEvaluator getRuleNetworkEvaluator() {
+ return ruleNetworkEvaluator;
+ }
+
@Override
public ActivationsManager getActivationsManager() {
return activationsManager;
diff --git a/drools-test-coverage/test-compiler-integration/src/test/java/org/drools/mvel/integrationtests/phreak/Scenario.java b/drools-test-coverage/test-compiler-integration/src/test/java/org/drools/mvel/integrationtests/phreak/Scenario.java
index eaef488031..582fe8b9df 100644
--- a/drools-test-coverage/test-compiler-integration/src/test/java/org/drools/mvel/integrationtests/phreak/Scenario.java
+++ b/drools-test-coverage/test-compiler-integration/src/test/java/org/drools/mvel/integrationtests/phreak/Scenario.java
@@ -24,7 +24,6 @@ import org.drools.core.common.TupleSetsImpl;
import org.drools.core.phreak.PhreakExistsNode;
import org.drools.core.phreak.PhreakJoinNode;
import org.drools.core.phreak.PhreakNotNode;
-import org.drools.core.phreak.SegmentPropagator;
import org.drools.core.reteoo.BetaMemory;
import org.drools.core.reteoo.BetaNode;
import org.drools.core.reteoo.ExistsNode;
@@ -177,15 +176,15 @@ public class Scenario {
actualResultLeftTuples = new TupleSetsImpl();
if ( phreakNode == PhreakJoinNode.class ) {
- new PhreakJoinNode().doNode( wm, (JoinNode) betaNode,
- sinkNode, bm, leftTuples, actualResultLeftTuples, previousResultTuples );
+ new PhreakJoinNode(wm).doNode( (JoinNode) betaNode, sinkNode,
+ bm, leftTuples, actualResultLeftTuples, previousResultTuples );
} else if ( phreakNode == PhreakNotNode.class ) {
- new PhreakNotNode().doNode( wm, (NotNode) betaNode,
- sinkNode, bm, leftTuples, actualResultLeftTuples, previousResultTuples );
+ new PhreakNotNode(wm).doNode( (NotNode) betaNode, sinkNode,
+ bm, leftTuples, actualResultLeftTuples, previousResultTuples );
} else if ( phreakNode == PhreakExistsNode.class ) {
- new PhreakExistsNode().doNode( wm, (ExistsNode) betaNode,
- sinkNode, bm, leftTuples, actualResultLeftTuples, previousResultTuples );
+ new PhreakExistsNode(wm).doNode( (ExistsNode) betaNode, sinkNode,
+ bm, leftTuples, actualResultLeftTuples, previousResultTuples );
}
if ( expectedResultBuilder != null ) {
@@ -203,7 +202,7 @@ public class Scenario {
}
SegmentMemory smem = bm.getSegmentMemory();
- SegmentPropagator.propagate(smem, actualResultLeftTuples, wm);
+ wm.getRuleNetworkEvaluator().propagate(smem, actualResultLeftTuples);
if ( testLeftMemory ) {
equalsLeftMemory( leftMemory );
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@kie.apache.org
For additional commands, e-mail: commits-help@kie.apache.org