1 files changed, 311 insertions, 0 deletions

diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/eval/OutputCachingEvaluatorNode.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/eval/OutputCachingEvaluatorNode.java
new file mode 100644
index 00000000..40a20c4e
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/eval/OutputCachingEvaluatorNode.java
@@ -0,0 +1,311 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2013, Bergmann Gabor, Istvan Rath and Daniel Varro
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v. 2.0 which is available at
+ * http://www.eclipse.org/legal/epl-v20.html.
+ * 
+ * SPDX-License-Identifier: EPL-2.0
+ *******************************************************************************/
+package tools.refinery.viatra.runtime.rete.eval;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import tools.refinery.viatra.runtime.matchers.tuple.Tuple;
+import tools.refinery.viatra.runtime.matchers.tuple.Tuples;
+import tools.refinery.viatra.runtime.matchers.util.Clearable;
+import tools.refinery.viatra.runtime.matchers.util.CollectionsFactory;
+import tools.refinery.viatra.runtime.matchers.util.Direction;
+import tools.refinery.viatra.runtime.matchers.util.Signed;
+import tools.refinery.viatra.runtime.matchers.util.TimelyMemory;
+import tools.refinery.viatra.runtime.matchers.util.timeline.Diff;
+import tools.refinery.viatra.runtime.matchers.util.timeline.Timeline;
+import tools.refinery.viatra.runtime.rete.matcher.TimelyConfiguration.TimelineRepresentation;
+import tools.refinery.viatra.runtime.rete.network.ReteContainer;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
+import tools.refinery.viatra.runtime.rete.network.communication.timely.ResumableNode;
+
+/**
+ * An evaluator node that caches the evaluation result. This node is also capable of caching the timestamps associated
+ * with the result tuples if it is used in recursive differential dataflow evaluation.
+ * 
+ * @author Bergmann Gabor
+ * @author Tamas Szabo
+ */
+public class OutputCachingEvaluatorNode extends AbstractEvaluatorNode implements Clearable, ResumableNode {
+
+    /**
+     * @since 2.3
+     */
+    protected NetworkStructureChangeSensitiveLogic logic;
+
+    /**
+     * @since 2.4
+     */
+    protected Map<Tuple, Iterable<Tuple>> outputCache;
+
+    /**
+     * Maps input tuples to timestamps. It is wrong to map evaluation result to timestamps because the different input
+     * tuples may yield the same evaluation result. This field is null as long as this node is in a non-recursive group.
+     * 
+     * @since 2.4
+     */
+    protected TimelyMemory<Timestamp> memory;
+
+    /**
+     * @since 2.4
+     */
+    protected CommunicationGroup group;
+
+    /**
+     * @since 1.5
+     */
+    public OutputCachingEvaluatorNode(final ReteContainer reteContainer, final EvaluatorCore core) {
+        super(reteContainer, core);
+        reteContainer.registerClearable(this);
+        this.outputCache = CollectionsFactory.createMap();
+        this.logic = createLogic();
+    }
+
+    @Override
+    public CommunicationGroup getCurrentGroup() {
+        return this.group;
+    }
+
+    @Override
+    public void setCurrentGroup(final CommunicationGroup group) {
+        this.group = group;
+    }
+
+    @Override
+    public void networkStructureChanged() {
+        super.networkStructureChanged();
+        this.logic = createLogic();
+    }
+
+    @Override
+    public void clear() {
+        this.outputCache.clear();
+        if (this.memory != null) {
+            this.memory.clear();
+        }
+    }
+
+    /**
+     * @since 2.3
+     */
+    protected NetworkStructureChangeSensitiveLogic createLogic() {
+        if (this.reteContainer.isTimelyEvaluation()
+                && this.reteContainer.getCommunicationTracker().isInRecursiveGroup(this)) {
+            if (this.memory == null) {
+                this.memory = new TimelyMemory<Timestamp>(reteContainer.isTimelyEvaluation() && reteContainer
+                        .getTimelyConfiguration().getTimelineRepresentation() == TimelineRepresentation.FAITHFUL);
+            }
+            return TIMELY;
+        } else {
+            return TIMELESS;
+        }
+    }
+
+    @Override
+    public void pullInto(final Collection<Tuple> collector, final boolean flush) {
+        this.logic.pullInto(collector, flush);
+    }
+
+    @Override
+    public void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush) {
+        this.logic.pullIntoWithTimeline(collector, flush);
+    }
+
+    @Override
+    public void update(final Direction direction, final Tuple input, final Timestamp timestamp) {
+        this.logic.update(direction, input, timestamp);
+    }
+
+    /**
+     * @since 2.4
+     */
+    @Override
+    public Timestamp getResumableTimestamp() {
+        if (this.memory == null) {
+            return null;
+        } else {
+            return this.memory.getResumableTimestamp();
+        }
+    }
+
+    /**
+     * @since 2.4
+     */
+    @Override
+    public void resumeAt(final Timestamp timestamp) {
+        this.logic.resumeAt(timestamp);
+    }
+
+    /**
+     * @since 2.3
+     */
+    protected static abstract class NetworkStructureChangeSensitiveLogic {
+
+        /**
+         * @since 2.4
+         */
+        public abstract void update(final Direction direction, final Tuple input, final Timestamp timestamp);
+
+        public abstract void pullInto(final Collection<Tuple> collector, final boolean flush);
+
+        /**
+         * @since 2.4
+         */
+        public abstract void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush);
+
+        /**
+         * @since 2.4
+         */
+        public abstract void resumeAt(final Timestamp timestamp);
+
+    }
+
+    private final NetworkStructureChangeSensitiveLogic TIMELESS = new NetworkStructureChangeSensitiveLogic() {
+
+        @Override
+        public void resumeAt(final Timestamp timestamp) {
+            // there is nothing to resume in the timeless case because we do not even care about timestamps
+        }
+
+        @Override
+        public void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public void pullInto(final Collection<Tuple> collector, final boolean flush) {
+            for (final Iterable<Tuple> output : outputCache.values()) {
+                if (output != NORESULT) {
+                    final Iterator<Tuple> itr = output.iterator();
+                    while (itr.hasNext()) {
+                        collector.add(itr.next());
+                    }
+                }
+            }
+        }
+
+        @Override
+        public void update(final Direction direction, final Tuple input, final Timestamp timestamp) {
+            if (direction == Direction.INSERT) {
+                final Iterable<Tuple> output = core.performEvaluation(input);
+                if (output != null) {
+                    final Iterable<Tuple> previous = outputCache.put(input, output);
+                    if (previous != null) {
+                        throw new IllegalStateException(
+                                String.format("Duplicate insertion of tuple %s into node %s", input, this));
+                    }
+                    propagateIterableUpdate(direction, output, timestamp);
+                }
+            } else {
+                final Iterable<Tuple> output = outputCache.remove(input);
+                if (output != null) {
+                    // may be null if no result was yielded
+                    propagateIterableUpdate(direction, output, timestamp);
+                }
+            }
+        }
+    };
+
+    private final NetworkStructureChangeSensitiveLogic TIMELY = new NetworkStructureChangeSensitiveLogic() {
+
+        @Override
+        public void resumeAt(final Timestamp timestamp) {
+            final Map<Tuple, Diff<Timestamp>> diffMap = memory.resumeAt(timestamp);
+
+            for (final Entry<Tuple, Diff<Timestamp>> entry : diffMap.entrySet()) {
+                final Tuple input = entry.getKey();
+                final Iterable<Tuple> output = outputCache.get(input);
+                if (output != NORESULT) {
+                    for (final Signed<Timestamp> signed : entry.getValue()) {
+                        propagateIterableUpdate(signed.getDirection(), output, signed.getPayload());
+                    }
+                }
+
+                if (memory.get(input) == null) {
+                    outputCache.remove(input);
+                }
+            }
+
+            final Timestamp nextTimestamp = memory.getResumableTimestamp();
+            if (nextTimestamp != null) {
+                group.notifyHasMessage(mailbox, nextTimestamp);
+            }
+        }
+
+        @Override
+        public void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush) {
+            for (final Entry<Tuple, Timeline<Timestamp>> entry : memory.asMap().entrySet()) {
+                final Tuple input = entry.getKey();
+                final Iterable<Tuple> output = outputCache.get(input);
+                if (output != NORESULT) {
+                    final Timeline<Timestamp> timestamp = entry.getValue();
+                    final Iterator<Tuple> itr = output.iterator();
+                    while (itr.hasNext()) {
+                        collector.put(itr.next(), timestamp);
+                    }
+                }
+            }
+        }
+
+        @Override
+        public void pullInto(final Collection<Tuple> collector, final boolean flush) {
+            TIMELESS.pullInto(collector, flush);
+        }
+
+        @Override
+        public void update(final Direction direction, final Tuple input, final Timestamp timestamp) {
+            if (direction == Direction.INSERT) {
+                Iterable<Tuple> output = outputCache.get(input);
+                if (output == null) {
+                    output = core.performEvaluation(input);
+                    if (output == null) {
+                        // the evaluation result is really null
+                        output = NORESULT;
+                    }
+                    outputCache.put(input, output);
+                }
+                final Diff<Timestamp> diff = memory.put(input, timestamp);
+                if (output != NORESULT) {
+                    for (final Signed<Timestamp> signed : diff) {
+                        propagateIterableUpdate(signed.getDirection(), output, signed.getPayload());
+                    }
+                }
+            } else {
+                final Iterable<Tuple> output = outputCache.get(input);
+                final Diff<Timestamp> diff = memory.remove(input, timestamp);
+                if (memory.get(input) == null) {
+                    outputCache.remove(input);
+                }
+                if (output != NORESULT) {
+                    for (final Signed<Timestamp> signed : diff) {
+                        propagateIterableUpdate(signed.getDirection(), output, signed.getPayload());
+                    }
+                }
+            }
+        }
+    };
+
+    /**
+     * This field is used to represent the "null" evaluation result. This is an optimization used in the timely case
+     * where the same tuple may be inserted multiple times with different timestamps. This way, we can also cache if
+     * something evaluated to null (instead of just forgetting about the previously computed result), thus avoiding the
+     * need to re-run a potentially expensive evaluation.
+     */
+    private static final Iterable<Tuple> NORESULT = Collections
+            .singleton(Tuples.staticArityFlatTupleOf(NoResult.INSTANCE));
+
+    private enum NoResult {
+        INSTANCE
+    }
+
+}