15 files changed, 1832 insertions, 0 deletions
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/CommunicationGroup.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/CommunicationGroup.java
new file mode 100644
index 00000000..8cedeb11
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/CommunicationGroup.java
@@ -0,0 +1,103 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2017, Tamas Szabo, 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.network.communication;
+import java.util.Collection;
+import java.util.Map;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+/**
+ * A communication group represents a set of nodes in the communication graph that form a strongly connected component.
+ * 
+ * @author Tamas Szabo
+ * @since 1.6
+ */
+public abstract class CommunicationGroup implements Comparable<CommunicationGroup> {
+    public static final String UNSUPPORTED_MESSAGE_KIND = "Unsupported message kind ";
+    /**
+     * Marker for the {@link CommunicationTracker}
+     */
+    public boolean isEnqueued = false;
+    protected final Node representative;
+    /**
+     * May be changed during bumping in {@link CommunicationTracker.registerDependency}
+     */
+    protected int identifier;
+    /**
+     * @since 1.7
+     */
+    protected final CommunicationTracker tracker;
+    /**
+     * @since 1.7
+     */
+    public CommunicationGroup(final CommunicationTracker tracker, final Node representative, final int identifier) {
+        this.tracker = tracker;
+        this.representative = representative;
+        this.identifier = identifier;
+    }
+    public abstract void deliverMessages();
+    public Node getRepresentative() {
+        return representative;
+    }
+    public abstract boolean isEmpty();
+    /**
+     * @since 2.0
+     */
+    public abstract void notifyLostAllMessages(final Mailbox mailbox, final MessageSelector kind);
+    /**
+     * @since 2.0
+     */
+    public abstract void notifyHasMessage(final Mailbox mailbox, final MessageSelector kind);
+    public abstract Map<MessageSelector, Collection<Mailbox>> getMailboxes();
+    public abstract boolean isRecursive();
+    @Override
+    public int hashCode() {
+        return this.identifier;
+    }
+    @Override
+    public String toString() {
+        return this.getClass().getSimpleName() + " " + this.identifier + " - representative: " + this.representative
+                + " - isEmpty: " + isEmpty();
+    }
+    @Override
+    public boolean equals(final Object obj) {
+        if (obj == null || this.getClass() != obj.getClass()) {
+            return false;
+        } else if (this == obj) {
+            return true;
+        } else {
+            final CommunicationGroup that = (CommunicationGroup) obj;
+            return this.identifier == that.identifier;
+        }
+    }
+    @Override
+    public int compareTo(final CommunicationGroup that) {
+        return this.identifier - that.identifier;
+    }
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/CommunicationTracker.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/CommunicationTracker.java
new file mode 100644
index 00000000..d244e644
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/CommunicationTracker.java
@@ -0,0 +1,467 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2017, Tamas Szabo, 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.network.communication;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import tools.refinery.viatra.runtime.rete.itc.alg.incscc.IncSCCAlg;
+import tools.refinery.viatra.runtime.rete.itc.alg.misc.topsort.TopologicalSorting;
+import tools.refinery.viatra.runtime.rete.itc.graphimpl.Graph;
+import tools.refinery.viatra.runtime.matchers.tuple.TupleMask;
+import tools.refinery.viatra.runtime.rete.aggregation.IAggregatorNode;
+import tools.refinery.viatra.runtime.rete.boundary.ExternalInputEnumeratorNode;
+import tools.refinery.viatra.runtime.rete.eval.RelationEvaluatorNode;
+import tools.refinery.viatra.runtime.rete.index.DualInputNode;
+import tools.refinery.viatra.runtime.rete.index.ExistenceNode;
+import tools.refinery.viatra.runtime.rete.index.Indexer;
+import tools.refinery.viatra.runtime.rete.index.IndexerListener;
+import tools.refinery.viatra.runtime.rete.index.IterableIndexer;
+import tools.refinery.viatra.runtime.rete.index.SpecializedProjectionIndexer;
+import tools.refinery.viatra.runtime.rete.network.IGroupable;
+import tools.refinery.viatra.runtime.rete.network.NetworkStructureChangeSensitiveNode;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.ProductionNode;
+import tools.refinery.viatra.runtime.rete.network.Receiver;
+import tools.refinery.viatra.runtime.rete.network.ReteContainer;
+import tools.refinery.viatra.runtime.rete.network.communication.timely.TimelyIndexerListenerProxy;
+import tools.refinery.viatra.runtime.rete.network.communication.timely.TimelyMailboxProxy;
+import tools.refinery.viatra.runtime.rete.network.mailbox.FallThroughCapableMailbox;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+import tools.refinery.viatra.runtime.rete.network.mailbox.timeless.BehaviorChangingMailbox;
+import tools.refinery.viatra.runtime.rete.single.TransitiveClosureNode;
+import tools.refinery.viatra.runtime.rete.single.TrimmerNode;
+/**
+ * An instance of this class is associated with every {@link ReteContainer}. The tracker serves two purposes: <br>
+ * (1) It allows RETE nodes to register their communication dependencies on-the-fly. These dependencies can be
+ * registered or unregistered when nodes are disposed of. <br>
+ * (2) It allows RETE nodes to register their mailboxes as dirty, that is, they can tell the tracker that they have
+ * something to send to other nodes in the network. The tracker is then responsible for ordering these messages (more
+ * precisely, the mailboxes that contain the messages) for the associated {@link ReteContainer}. The ordering is
+ * governed by the strongly connected components in the dependency network and follows a topological sorting scheme;
+ * those mailboxes will be emptied first whose owner nodes do not depend on other undelivered messages.
+ *
+ * @author Tamas Szabo
+ * @since 1.6
+ *
+ */
+public abstract class CommunicationTracker {
+    /**
+     * The minimum group id assigned so far
+     */
+    protected int minGroupId;
+    /**
+     * The maximum group id assigned so far
+     */
+    protected int maxGroupId;
+    /**
+     * The dependency graph of the communications in the RETE network
+     */
+    protected final Graph<Node> dependencyGraph;
+    /**
+     * Incremental SCC information about the dependency graph
+     */
+    protected final IncSCCAlg<Node> sccInformationProvider;
+    /**
+     * Precomputed node -> communication group map
+     */
+    protected final Map<Node, CommunicationGroup> groupMap;
+    /**
+     * Priority queue of active communication groups
+     */
+    protected final Queue<CommunicationGroup> groupQueue;
+    // groups should have a simple integer flag which represents its position in a priority queue
+    // priority queue only contains the ACTIVE groups
+    public CommunicationTracker() {
+        this.dependencyGraph = new Graph<Node>();
+        this.sccInformationProvider = new IncSCCAlg<Node>(this.dependencyGraph);
+        this.groupQueue = new PriorityQueue<CommunicationGroup>();
+        this.groupMap = new HashMap<Node, CommunicationGroup>();
+    }
+    public Graph<Node> getDependencyGraph() {
+        return dependencyGraph;
+    }
+    public CommunicationGroup getGroup(final Node node) {
+        return this.groupMap.get(node);
+    }
+    private void precomputeGroups() {
+        groupMap.clear();
+        // reconstruct group map from dependency graph
+        final Graph<Node> reducedGraph = sccInformationProvider.getReducedGraph();
+        final List<Node> representatives = TopologicalSorting.compute(reducedGraph);
+        for (int i = 0; i < representatives.size(); i++) { // groups for SCC representatives
+            final Node representative = representatives.get(i);
+            createAndStoreGroup(representative, i);
+        }
+        minGroupId = 0;
+        maxGroupId = representatives.size() - 1;
+        for (final Node node : dependencyGraph.getAllNodes()) { // extend group map to the rest of nodes
+            final Node representative = sccInformationProvider.getRepresentative(node);
+            final CommunicationGroup group = groupMap.get(representative);
+            if (representative != node) {
+                addToGroup(node, group);
+            }
+        }
+        for (final Node node : dependencyGraph.getAllNodes()) {
+            // set fall-through flags of default mailboxes
+            precomputeFallThroughFlag(node);
+            // perform further tracker-specific post-processing
+            postProcessNode(node);
+        }
+        // reconstruct new queue contents based on new group map
+        if (!groupQueue.isEmpty()) {
+            final Set<CommunicationGroup> oldActiveGroups = new HashSet<CommunicationGroup>(groupQueue);
+            groupQueue.clear();
+            reconstructQueueContents(oldActiveGroups);
+        }
+        // post process the groups
+        for (final CommunicationGroup group : groupMap.values()) {
+            postProcessGroup(group);
+        }
+    }
+    /**
+     * This method is responsible for reconstructing the active queue contents after the network structure has changed.
+     * It it defined as abstract because the reconstruction logic is specific to each {@link CommunicationTracker}.
+     * @since 2.4
+     */
+    protected abstract void reconstructQueueContents(final Set<CommunicationGroup> oldActiveGroups);
+    private void addToGroup(final Node node, final CommunicationGroup group) {
+        groupMap.put(node, group);
+        if (node instanceof Receiver) {
+            ((Receiver) node).getMailbox().setCurrentGroup(group);
+            if (node instanceof IGroupable) {
+                ((IGroupable) node).setCurrentGroup(group);
+            }
+        }
+    }
+    /**
+     * Depends on the groups, as well as the parent nodes of the argument, so recomputation is needed if these change
+     */
+    private void precomputeFallThroughFlag(final Node node) {
+        CommunicationGroup group = groupMap.get(node);
+        if (node instanceof Receiver) {
+            IGroupable mailbox = ((Receiver) node).getMailbox();
+            if (mailbox instanceof FallThroughCapableMailbox) {
+                Set<Node> directParents = dependencyGraph.getSourceNodes(node).distinctValues();
+                // decide between using quick&cheap fall-through, or allowing for update cancellation
+                boolean fallThrough =
+                        // disallow fallthrough: updates at production nodes should cancel, if they can be trimmed or
+                        // disjunctive
+                        (!(node instanceof ProductionNode && ( // it is a production node...
+                        // with more than one parent
+                        directParents.size() > 0 ||
+                        // or true trimming in its sole parent
+                                directParents.size() == 1 && trueTrimming(directParents.iterator().next())))) &&
+                        // disallow fallthrough: external updates should be stored (if updates are delayed)
+                                (!(node instanceof ExternalInputEnumeratorNode)) &&
+                        // disallow fallthrough: RelationEvaluatorNode needs to be notified in batch-style, and the batching is done by the mailbox
+                        // however, it is not the RelationEvaluatorNode itself that is interesting here, as that indirectly uses the BatchingReceiver
+                        // so we need to disable fall-through for the BatchingReceiver
+                                (!(node instanceof RelationEvaluatorNode.BatchingReceiver));
+                // do additional checks
+                if (fallThrough) {
+                    // recursive parent groups generate excess updates that should be cancelled after delete&rederive
+                    // phases
+                    // aggregator and transitive closure parent nodes also generate excess updates that should be
+                    // cancelled
+                    directParentLoop: for (Node directParent : directParents) {
+                        Set<Node> parentsToCheck = new HashSet<>();
+                        // check the case where a direct parent is the reason for mailbox usage
+                        parentsToCheck.add(directParent);
+                        // check the case where an indirect parent (join slot) is the reason for mailbox usage
+                        if (directParent instanceof DualInputNode) {
+                            // in case of existence join (typically antijoin), a mailbox should allow
+                            // an insertion and deletion (at the secondary slot) to cancel each other out
+                            if (directParent instanceof ExistenceNode) {
+                                fallThrough = false;
+                                break directParentLoop;
+                            }
+                            // in beta nodes, indexer slots (or their active nodes) are considered indirect parents
+                            DualInputNode dualInput = (DualInputNode) directParent;
+                            IterableIndexer primarySlot = dualInput.getPrimarySlot();
+                            if (primarySlot != null)
+                                parentsToCheck.add(primarySlot.getActiveNode());
+                            Indexer secondarySlot = dualInput.getSecondarySlot();
+                            if (secondarySlot != null)
+                                parentsToCheck.add(secondarySlot.getActiveNode());
+                        }
+                        for (Node parent : parentsToCheck) {
+                            CommunicationGroup parentGroup = groupMap.get(parent);
+                            if ( // parent is in a different, recursive group
+                            (group != parentGroup && parentGroup.isRecursive()) ||
+                            // node and parent within the same recursive group, and...
+                                    (group == parentGroup && group.isRecursive() && (
+                                    // parent is a transitive closure or aggregator node, or a trimmer
+                                    // allow trimmed or disjunctive tuple updates to cancel each other
+                                    (parent instanceof TransitiveClosureNode) || (parent instanceof IAggregatorNode)
+                                            || trueTrimming(parent)))) {
+                                fallThrough = false;
+                                break directParentLoop;
+                            }
+                        }
+                    }
+                }
+                // overwrite fallthrough flag with newly computed value
+                ((FallThroughCapableMailbox) mailbox).setFallThrough(fallThrough);
+            }
+        }
+    }
+    /**
+     * A trimmer node that actually eliminates some columns (not just reorders)
+     */
+    private boolean trueTrimming(Node node) {
+        if (node instanceof TrimmerNode) {
+            TupleMask mask = ((TrimmerNode) node).getMask();
+            return (mask.indices.length != mask.sourceWidth);
+        }
+        return false;
+    }
+    public void activateUnenqueued(final CommunicationGroup group) {
+        groupQueue.add(group);
+        group.isEnqueued = true;
+    }
+    public void deactivate(final CommunicationGroup group) {
+        groupQueue.remove(group);
+        group.isEnqueued = false;
+    }
+    public CommunicationGroup getAndRemoveFirstGroup() {
+        final CommunicationGroup group = groupQueue.poll();
+        group.isEnqueued = false;
+        return group;
+    }
+    public boolean isEmpty() {
+        return groupQueue.isEmpty();
+    }
+    protected abstract CommunicationGroup createGroup(final Node representative, final int index);
+    protected CommunicationGroup createAndStoreGroup(final Node representative, final int index) {
+        final CommunicationGroup group = createGroup(representative, index);
+        addToGroup(representative, group);
+        return group;
+    }
+    /**
+     * Registers the dependency that the target {@link Node} depends on the source {@link Node}. In other words, source
+     * may send messages to target in the RETE network. If the dependency edge is already present, this method call is a
+     * noop.
+     *
+     * @param source
+     *            the source node
+     * @param target
+     *            the target node
+     */
+    public void registerDependency(final Node source, final Node target) {
+        // nodes can be immediately inserted, if they already exist in the graph, this is a noop
+        dependencyGraph.insertNode(source);
+        dependencyGraph.insertNode(target);
+        if (!this.dependencyGraph.getTargetNodes(source).containsNonZero(target)) {
+            // query all these information before the actual edge insertion
+            // because SCCs may be unified during the process
+            final Node sourceRepresentative = sccInformationProvider.getRepresentative(source);
+            final Node targetRepresentative = sccInformationProvider.getRepresentative(target);
+            final boolean targetHadOutgoingEdges = sccInformationProvider.hasOutgoingEdges(targetRepresentative);
+            // insert the edge
+            dependencyGraph.insertEdge(source, target);
+            // create groups if they do not yet exist
+            CommunicationGroup sourceGroup = groupMap.get(sourceRepresentative);
+            if (sourceGroup == null) {
+                // create on-demand with the next smaller group id
+                sourceGroup = createAndStoreGroup(sourceRepresentative, --minGroupId);
+            }
+            final int sourceIndex = sourceGroup.identifier;
+            CommunicationGroup targetGroup = groupMap.get(targetRepresentative);
+            if (targetGroup == null) {
+                // create on-demand with the next larger group id
+                targetGroup = createAndStoreGroup(targetRepresentative, ++maxGroupId);
+            }
+            final int targetIndex = targetGroup.identifier;
+            if (sourceIndex <= targetIndex) {
+                // indices obey current topological ordering
+                refreshFallThroughFlag(target);
+                postProcessNode(source);
+                postProcessNode(target);
+                postProcessGroup(sourceGroup);
+                if (sourceGroup != targetGroup) {
+                    postProcessGroup(targetGroup);
+                }
+            } else if (sourceIndex > targetIndex && !targetHadOutgoingEdges) {
+                // indices violate current topological ordering, but we can simply bump the target index
+                final boolean wasEnqueued = targetGroup.isEnqueued;
+                if (wasEnqueued) {
+                    groupQueue.remove(targetGroup);
+                }
+                targetGroup.identifier = ++maxGroupId;
+                if (wasEnqueued) {
+                    groupQueue.add(targetGroup);
+                }
+                refreshFallThroughFlag(target);
+                postProcessNode(source);
+                postProcessNode(target);
+                postProcessGroup(sourceGroup);
+                postProcessGroup(targetGroup);
+            } else {
+                // needs a full re-computation because of more complex change
+                precomputeGroups();
+            }
+        }
+    }
+    /**
+     * Returns true if the given {@link Node} is in a recursive {@link CommunicationGroup}, false otherwise.
+     */
+    public boolean isInRecursiveGroup(final Node node) {
+        final CommunicationGroup group = this.getGroup(node);
+        if (group == null) {
+            return false;
+        } else {
+            return group.isRecursive();
+        }
+    }
+    /**
+     * Returns true if the given two {@link Node}s are in the same {@link CommunicationGroup}.
+     */
+    public boolean areInSameGroup(final Node left, final Node right) {
+        final CommunicationGroup leftGroup = this.getGroup(left);
+        final CommunicationGroup rightGroup = this.getGroup(right);
+        return leftGroup != null && leftGroup == rightGroup;
+    }
+    /**
+     * Unregisters a dependency between source and target.
+     *
+     * @param source
+     *            the source node
+     * @param target
+     *            the target node
+     */
+    public void unregisterDependency(final Node source, final Node target) {
+        // delete the edge first, and then query the SCC info provider
+        this.dependencyGraph.deleteEdgeIfExists(source, target);
+        final Node sourceRepresentative = sccInformationProvider.getRepresentative(source);
+        final Node targetRepresentative = sccInformationProvider.getRepresentative(target);
+        // if they are still in the same SCC,
+        // then this deletion did not affect the SCCs,
+        // and it is sufficient to recompute affected fall-through flags;
+        // otherwise, we need a new pre-computation for the groupMap and groupQueue
+        if (sourceRepresentative.equals(targetRepresentative)) {
+            // this deletion could not have affected the split flags
+            refreshFallThroughFlag(target);
+            postProcessNode(source);
+            postProcessNode(target);
+        } else {
+            // preComputeGroups takes care of the split flag maintenance
+            precomputeGroups();
+        }
+    }
+    /**
+     * Refresh fall-through flags if dependencies change for given target, but no SCC change
+     */
+    private void refreshFallThroughFlag(final Node target) {
+        precomputeFallThroughFlag(target);
+        if (target instanceof DualInputNode) {
+            for (final Node indirectTarget : dependencyGraph.getTargetNodes(target).distinctValues()) {
+                precomputeFallThroughFlag(indirectTarget);
+            }
+        }
+    }
+    /**
+     * Returns true if the given source-target edge in the communication network acts as a recursion cut point.
+     * The current implementation considers edges leading into {@link ProductionNode}s as cut point iff
+     * both source and target belong to the same group.
+     *
+     * @param source the source node
+     * @param target the target node
+     * @return true if the edge is a cut point, false otherwise
+     * @since 2.4
+     */
+    protected boolean isRecursionCutPoint(final Node source, final Node target) {
+        final Node effectiveSource = source instanceof SpecializedProjectionIndexer
+                ? ((SpecializedProjectionIndexer) source).getActiveNode()
+                : source;
+        final CommunicationGroup sourceGroup = this.getGroup(effectiveSource);
+        final CommunicationGroup targetGroup = this.getGroup(target);
+        return sourceGroup != null && sourceGroup == targetGroup && target instanceof ProductionNode;
+    }
+    /**
+     * This hook allows concrete tracker implementations to perform tracker-specific post processing on nodes (cf.
+     * {@link NetworkStructureChangeSensitiveNode} and {@link BehaviorChangingMailbox}). At the time of the invocation,
+     * the network topology has already been updated.
+     */
+    protected abstract void postProcessNode(final Node node);
+    /**
+     * This hook allows concrete tracker implementations to perform tracker-specific post processing on groups. At the
+     * time of the invocation, the network topology has already been updated.
+     * @since 2.4
+     */
+    protected abstract void postProcessGroup(final CommunicationGroup group);
+    /**
+     * Creates a proxy for the given {@link Mailbox} for the given requester {@link Node}. The proxy creation is
+     * {@link CommunicationTracker}-specific and depends on the identity of the requester. This method is primarily used
+     * to create {@link TimelyMailboxProxy}s depending on the network topology. There is no guarantee that the same
+     * proxy instance is returned when this method is called multiple times with the same arguments.
+     */
+    public abstract Mailbox proxifyMailbox(final Node requester, final Mailbox original);
+    /**
+     * Creates a proxy for the given {@link IndexerListener} for the given requester {@link Node}. The proxy creation is
+     * {@link CommunicationTracker}-specific and depends on the identity of the requester. This method is primarily used
+     * to create {@link TimelyIndexerListenerProxy}s depending on the network topology. There is no guarantee that the
+     * same proxy instance is returned when this method is called multiple times with the same arguments.
+     */
+    public abstract IndexerListener proxifyIndexerListener(final Node requester, final IndexerListener original);
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/MessageSelector.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/MessageSelector.java
new file mode 100644
index 00000000..e1a61693
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/MessageSelector.java
@@ -0,0 +1,19 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2018, Tamas Szabo, 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.network.communication;
+/**
+ * Subclasses of this interface represent meta data of update messages in Rete. 
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public interface MessageSelector {
+    
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/NodeComparator.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/NodeComparator.java
new file mode 100644
index 00000000..27779352
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/NodeComparator.java
@@ -0,0 +1,32 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication;
+import java.util.Comparator;
+import java.util.Map;
+import tools.refinery.viatra.runtime.rete.network.Node;
+/**
+ * @since 2.4
+ */
+public class NodeComparator implements Comparator<Node> {
+    protected final Map<Node, Integer> nodeMap;
+    public NodeComparator(final Map<Node, Integer> nodeMap) {
+        this.nodeMap = nodeMap;
+    }
+    @Override
+    public int compare(final Node left, final Node right) {
+        return this.nodeMap.get(left) - this.nodeMap.get(right);
+    }
+    
+}
+\ No newline at end of file
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/PhasedSelector.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/PhasedSelector.java
new file mode 100644
index 00000000..41cd8cd3
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/PhasedSelector.java
@@ -0,0 +1,34 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2017, Tamas Szabo, 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.network.communication;
+/**
+ * A default message selector that can be used to associate phases to messages. 
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public enum PhasedSelector implements MessageSelector {
+    
+    /**
+     * No special distinguishing feature
+     */
+    DEFAULT,
+    
+    /**
+     * Inserts and delete-insert monotone change pairs
+     */
+    MONOTONE,
+    
+    /**
+     * Deletes
+     */
+    ANTI_MONOTONE
+    
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/Timestamp.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/Timestamp.java
new file mode 100644
index 00000000..a50a63a8
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/Timestamp.java
@@ -0,0 +1,124 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2018, Tamas Szabo, 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.network.communication;
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import tools.refinery.viatra.runtime.matchers.util.timeline.Timeline;
+import tools.refinery.viatra.runtime.matchers.util.timeline.Timelines;
+/**
+ * A timestamp associated with update messages in timely evaluation.
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public class Timestamp implements Comparable<Timestamp>, MessageSelector {
+    protected final int value;
+    public static final Timestamp ZERO = new Timestamp(0);
+    /**
+     * @since 2.4
+     */
+    public static final Timeline<Timestamp> INSERT_AT_ZERO_TIMELINE = Timelines.createFrom(Timestamp.ZERO);
+    public Timestamp(final int value) {
+        this.value = value;
+    }
+    public int getValue() {
+        return value;
+    }
+    public Timestamp max(final Timestamp that) {
+        if (this.value >= that.value) {
+            return this;
+        } else {
+            return that;
+        }
+    }
+    
+    /**
+     * @since 2.4
+     */
+    public Timestamp min(final Timestamp that) {
+        if (this.value <= that.value) {
+            return this;
+        } else {
+            return that;
+        }
+    }
+    @Override
+    public int compareTo(final Timestamp that) {
+        return this.value - that.value;
+    }
+    @Override
+    public boolean equals(final Object obj) {
+        if (obj == null || !(obj instanceof Timestamp)) {
+            return false;
+        } else {
+            return this.value == ((Timestamp) obj).value;
+        }
+    }
+    @Override
+    public int hashCode() {
+        return this.value;
+    }
+    @Override
+    public String toString() {
+        return Integer.toString(this.value);
+    }
+    
+    /**
+     * A {@link Map} implementation that associates the zero timestamp with every key. There is no suppor for
+     * {@link Map#entrySet()} due to performance reasons.
+     * 
+     * @author Tamas Szabo
+     */
+    public static final class AllZeroMap<T> extends AbstractMap<T, Timeline<Timestamp>> {
+        private final Collection<T> wrapped;
+        public AllZeroMap(Set<T> wrapped) {
+            this.wrapped = wrapped;
+        }
+        @Override
+        public Set<Entry<T, Timeline<Timestamp>>> entrySet() {
+            throw new UnsupportedOperationException("Use the combination of keySet() and get()!");
+        }
+        /**
+         * @since 2.4
+         */
+        @Override
+        public Timeline<Timestamp> get(final Object key) {
+            return INSERT_AT_ZERO_TIMELINE;
+        }
+        @Override
+        public Set<T> keySet() {
+            return (Set<T>) this.wrapped;
+        }
+        @Override
+        public String toString() {
+            return this.getClass().getSimpleName() + ": " + this.keySet().toString();
+        }
+    }
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/RecursiveCommunicationGroup.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/RecursiveCommunicationGroup.java
new file mode 100644
index 00000000..d8260384
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/RecursiveCommunicationGroup.java
@@ -0,0 +1,164 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timeless;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.EnumMap;
+import java.util.LinkedHashSet;
+import java.util.Map;
+import java.util.Set;
+import tools.refinery.viatra.runtime.matchers.util.CollectionsFactory;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.RederivableNode;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationTracker;
+import tools.refinery.viatra.runtime.rete.network.communication.MessageSelector;
+import tools.refinery.viatra.runtime.rete.network.communication.PhasedSelector;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+/**
+ * A communication group representing either a single node where the
+ * node is a monotonicity aware one or a set of nodes that form an SCC. 
+ * 
+ * @author Tamas Szabo
+ * @since 2.4
+ */
+public class RecursiveCommunicationGroup extends CommunicationGroup {
+    private final Set<Mailbox> antiMonotoneMailboxes;
+    private final Set<Mailbox> monotoneMailboxes;
+    private final Set<Mailbox> defaultMailboxes;
+    private final Set<RederivableNode> rederivables;
+    private boolean currentlyDelivering;
+    /**
+     * @since 1.7
+     */
+    public RecursiveCommunicationGroup(final CommunicationTracker tracker, final Node representative, final int identifier) {
+        super(tracker, representative, identifier);
+        this.antiMonotoneMailboxes = CollectionsFactory.createSet();
+        this.monotoneMailboxes = CollectionsFactory.createSet();
+        this.defaultMailboxes = CollectionsFactory.createSet();
+        this.rederivables = new LinkedHashSet<RederivableNode>();
+        this.currentlyDelivering = false;
+    }
+    @Override
+    public void deliverMessages() {
+        this.currentlyDelivering = true;
+        
+        // ANTI-MONOTONE PHASE
+        while (!this.antiMonotoneMailboxes.isEmpty() || !this.defaultMailboxes.isEmpty()) {
+            while (!this.antiMonotoneMailboxes.isEmpty()) {
+                final Mailbox mailbox = this.antiMonotoneMailboxes.iterator().next();
+                this.antiMonotoneMailboxes.remove(mailbox);
+                mailbox.deliverAll(PhasedSelector.ANTI_MONOTONE);
+            }
+            while (!this.defaultMailboxes.isEmpty()) {
+                final Mailbox mailbox = this.defaultMailboxes.iterator().next();
+                this.defaultMailboxes.remove(mailbox);
+                mailbox.deliverAll(PhasedSelector.DEFAULT);
+            }
+        }
+        // REDERIVE PHASE
+        while (!this.rederivables.isEmpty()) {
+            // re-derivable nodes take care of their unregistration!!
+            final RederivableNode node = this.rederivables.iterator().next();
+            node.rederiveOne();
+        }
+        // MONOTONE PHASE
+        while (!this.monotoneMailboxes.isEmpty() || !this.defaultMailboxes.isEmpty()) {
+            while (!this.monotoneMailboxes.isEmpty()) {
+                final Mailbox mailbox = this.monotoneMailboxes.iterator().next();
+                this.monotoneMailboxes.remove(mailbox);
+                mailbox.deliverAll(PhasedSelector.MONOTONE);
+            }
+            while (!this.defaultMailboxes.isEmpty()) {
+                final Mailbox mailbox = this.defaultMailboxes.iterator().next();
+                this.defaultMailboxes.remove(mailbox);
+                mailbox.deliverAll(PhasedSelector.DEFAULT);
+            }
+        }
+        
+        this.currentlyDelivering = false;
+    }
+    @Override
+    public boolean isEmpty() {
+        return this.rederivables.isEmpty() && this.antiMonotoneMailboxes.isEmpty()
+                && this.monotoneMailboxes.isEmpty() && this.defaultMailboxes.isEmpty();
+    }
+    @Override
+    public void notifyHasMessage(final Mailbox mailbox, final MessageSelector kind) {
+        final Collection<Mailbox> mailboxes = getMailboxContainer(kind);
+        mailboxes.add(mailbox);
+        if (!this.isEnqueued && !this.currentlyDelivering) {
+            this.tracker.activateUnenqueued(this);
+        }
+    }
+    @Override
+    public void notifyLostAllMessages(final Mailbox mailbox, final MessageSelector kind) {
+        final Collection<Mailbox> mailboxes = getMailboxContainer(kind);
+        mailboxes.remove(mailbox);
+        if (isEmpty()) {
+            this.tracker.deactivate(this);
+        }
+    }
+    private Collection<Mailbox> getMailboxContainer(final MessageSelector kind) {
+        if (kind == PhasedSelector.ANTI_MONOTONE) {
+            return this.antiMonotoneMailboxes;
+        } else if (kind == PhasedSelector.MONOTONE) {
+            return this.monotoneMailboxes;
+        } else if (kind == PhasedSelector.DEFAULT) {
+            return this.defaultMailboxes;
+        } else {
+            throw new IllegalArgumentException(UNSUPPORTED_MESSAGE_KIND + kind);
+        }
+    }
+    public void addRederivable(final RederivableNode node) {
+        this.rederivables.add(node);
+        if (!this.isEnqueued) {
+            this.tracker.activateUnenqueued(this);
+        }
+    }
+    public void removeRederivable(final RederivableNode node) {
+        this.rederivables.remove(node);
+        if (isEmpty()) {
+            this.tracker.deactivate(this);
+        }
+    }
+    public Collection<RederivableNode> getRederivables() {
+        return this.rederivables;
+    }
+    @Override
+    public Map<MessageSelector, Collection<Mailbox>> getMailboxes() {
+        Map<PhasedSelector, Collection<Mailbox>> map = new EnumMap<>(PhasedSelector.class);
+        map.put(PhasedSelector.ANTI_MONOTONE, antiMonotoneMailboxes);
+        map.put(PhasedSelector.MONOTONE, monotoneMailboxes);
+        map.put(PhasedSelector.DEFAULT, defaultMailboxes);
+        return Collections.unmodifiableMap(map);
+    }
+    
+    @Override
+    public boolean isRecursive() {
+        return true;
+    }
+    
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/SingletonCommunicationGroup.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/SingletonCommunicationGroup.java
new file mode 100644
index 00000000..c51c7dbf
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/SingletonCommunicationGroup.java
@@ -0,0 +1,86 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timeless;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationTracker;
+import tools.refinery.viatra.runtime.rete.network.communication.MessageSelector;
+import tools.refinery.viatra.runtime.rete.network.communication.PhasedSelector;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+/**
+ * A communication group containing only a single node with a single default
+ * mailbox.
+ * 
+ * @author Tamas Szabo
+ * @since 1.6
+ */
+public class SingletonCommunicationGroup extends CommunicationGroup {
+    
+    private Mailbox mailbox;
+    /**
+     * @since 1.7
+     */
+    public SingletonCommunicationGroup(final CommunicationTracker tracker, final Node representative, final int identifier) {
+        super(tracker, representative, identifier);
+    }
+    @Override
+    public void deliverMessages() {
+        this.mailbox.deliverAll(PhasedSelector.DEFAULT);
+    }
+    @Override
+    public boolean isEmpty() {
+        return this.mailbox == null;
+    }
+    @Override
+    public void notifyHasMessage(final Mailbox mailbox, final MessageSelector kind) {
+        if (kind == PhasedSelector.DEFAULT) {
+            this.mailbox = mailbox;
+            if (!this.isEnqueued) {
+                this.tracker.activateUnenqueued(this);
+            }
+        } else {
+            throw new IllegalArgumentException(UNSUPPORTED_MESSAGE_KIND + kind);
+        }
+    }
+    @Override
+    public void notifyLostAllMessages(final Mailbox mailbox, final MessageSelector kind) {
+        if (kind == PhasedSelector.DEFAULT) {
+            this.mailbox = null;
+            this.tracker.deactivate(this);
+        } else {
+            throw new IllegalArgumentException(UNSUPPORTED_MESSAGE_KIND + kind);
+        }
+    }
+    @Override
+    public Map<MessageSelector, Collection<Mailbox>> getMailboxes() {
+        if (mailbox != null) {
+            return Collections.singletonMap(PhasedSelector.DEFAULT, Collections.singleton(mailbox));
+        } else {
+            return Collections.emptyMap();
+        }
+    }
+    @Override
+    public boolean isRecursive() {
+        return false;
+    }
+    
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/TimelessCommunicationTracker.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/TimelessCommunicationTracker.java
new file mode 100644
index 00000000..1c18c1cd
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timeless/TimelessCommunicationTracker.java
@@ -0,0 +1,149 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timeless;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.Map.Entry;
+import tools.refinery.viatra.runtime.rete.index.DualInputNode;
+import tools.refinery.viatra.runtime.rete.index.Indexer;
+import tools.refinery.viatra.runtime.rete.index.IndexerListener;
+import tools.refinery.viatra.runtime.rete.index.IterableIndexer;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.Receiver;
+import tools.refinery.viatra.runtime.rete.network.RederivableNode;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationTracker;
+import tools.refinery.viatra.runtime.rete.network.communication.MessageSelector;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+import tools.refinery.viatra.runtime.rete.network.mailbox.timeless.BehaviorChangingMailbox;
+/**
+ * Timeless implementation of the communication tracker.
+ * 
+ * @author Tamas Szabo
+ * @since 2.2
+ */
+public class TimelessCommunicationTracker extends CommunicationTracker {
+    @Override
+    protected CommunicationGroup createGroup(Node representative, int index) {
+        final boolean isSingleton = this.sccInformationProvider.sccs.getPartition(representative).size() == 1;
+        final boolean isReceiver = representative instanceof Receiver;
+        final boolean isPosetIndifferent = isReceiver
+                && ((Receiver) representative).getMailbox() instanceof BehaviorChangingMailbox;
+        final boolean isSingletonInDRedMode = isSingleton && (representative instanceof RederivableNode)
+                && ((RederivableNode) representative).isInDRedMode();
+        CommunicationGroup group = null;
+        // we can only use a singleton group iff
+        // (1) the SCC has one node AND
+        // (2) either we have a poset-indifferent mailbox OR the node is not even a receiver AND
+        // (3) the node does not run in DRed mode in a singleton group
+        if (isSingleton && (isPosetIndifferent || !isReceiver) && !isSingletonInDRedMode) {
+            group = new SingletonCommunicationGroup(this, representative, index);
+        } else {
+            group = new RecursiveCommunicationGroup(this, representative, index);
+        }
+        return group;
+    }
+    @Override
+    protected void reconstructQueueContents(final Set<CommunicationGroup> oldActiveGroups) {
+        for (final CommunicationGroup oldGroup : oldActiveGroups) {
+            for (final Entry<MessageSelector, Collection<Mailbox>> entry : oldGroup.getMailboxes().entrySet()) {
+                for (final Mailbox mailbox : entry.getValue()) {
+                    final CommunicationGroup newGroup = this.groupMap.get(mailbox.getReceiver());
+                    newGroup.notifyHasMessage(mailbox, entry.getKey());
+                }
+            }
+            if (oldGroup instanceof RecursiveCommunicationGroup) {
+                for (final RederivableNode node : ((RecursiveCommunicationGroup) oldGroup).getRederivables()) {
+                    final CommunicationGroup newGroup = this.groupMap.get(node);
+                    if (!(newGroup instanceof RecursiveCommunicationGroup)) {
+                        throw new IllegalStateException("The new group must also be recursive! " + newGroup);
+                    }
+                    ((RecursiveCommunicationGroup) newGroup).addRederivable(node);
+                }
+            }
+        }
+    }
+    @Override
+    public Mailbox proxifyMailbox(final Node requester, final Mailbox original) {
+        return original;
+    }
+    @Override
+    public IndexerListener proxifyIndexerListener(final Node requester, final IndexerListener original) {
+        return original;
+    }
+    @Override
+    protected void postProcessNode(final Node node) {
+        if (node instanceof Receiver) {
+            final Mailbox mailbox = ((Receiver) node).getMailbox();
+            if (mailbox instanceof BehaviorChangingMailbox) {
+                final CommunicationGroup group = this.groupMap.get(node);
+                final Set<Node> sccNodes = this.sccInformationProvider.sccs.getPartition(node);
+                // a default mailbox must split its messages iff
+                // (1) its receiver is in a recursive group and
+                final boolean c1 = group.isRecursive();
+                // (2) its receiver is at the SCC boundary of that group
+                final boolean c2 = isAtSCCBoundary(node);
+                // (3) its group consists of more than one node
+                final boolean c3 = sccNodes.size() > 1;
+                ((BehaviorChangingMailbox) mailbox).setSplitFlag(c1 && c2 && c3);
+            }
+        }
+    }
+    
+    @Override
+    protected void postProcessGroup(final CommunicationGroup group) {
+        
+    }
+    /**
+     * @since 2.0
+     */
+    private boolean isAtSCCBoundary(final Node node) {
+        final CommunicationGroup ownGroup = this.groupMap.get(node);
+        assert ownGroup != null;
+        for (final Node source : this.dependencyGraph.getSourceNodes(node).distinctValues()) {
+            final Set<Node> sourcesToCheck = new HashSet<Node>();
+            sourcesToCheck.add(source);
+            // DualInputNodes must be checked additionally because they do not use a mailbox directly.
+            // It can happen that their indexers actually belong to other SCCs.
+            if (source instanceof DualInputNode) {
+                final DualInputNode dualInput = (DualInputNode) source;
+                final IterableIndexer primarySlot = dualInput.getPrimarySlot();
+                if (primarySlot != null) {
+                    sourcesToCheck.add(primarySlot.getActiveNode());
+                }
+                final Indexer secondarySlot = dualInput.getSecondarySlot();
+                if (secondarySlot != null) {
+                    sourcesToCheck.add(secondarySlot.getActiveNode());
+                }
+            }
+            for (final Node current : sourcesToCheck) {
+                final CommunicationGroup otherGroup = this.groupMap.get(current);
+                assert otherGroup != null;
+                if (!ownGroup.equals(otherGroup)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/ResumableNode.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/ResumableNode.java
new file mode 100644
index 00000000..8097bd91
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/ResumableNode.java
@@ -0,0 +1,36 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2016, Tamas Szabo, 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.network.communication.timely;
+import tools.refinery.viatra.runtime.rete.network.IGroupable;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
+/**
+ * {@link Node}s that implement this interface can resume folding of their states when instructed during timely evaluation. 
+ * 
+ * @since 2.3
+ * @author Tamas Szabo
+ */
+public interface ResumableNode extends Node, IGroupable {
+    
+    /**
+     * When called, the folding of the state shall be resumed at the given timestamp. The resumable is expected to
+     * do a folding step at the given timestamp only. Afterwards, folding shall be interrupted, even if there is more
+     * folding to do towards higher timestamps.
+     */
+    public void resumeAt(final Timestamp timestamp);
+    /**
+     * Returns the smallest timestamp where lazy folding shall be resumed, or null if there is no more folding to do in this
+     * resumable.
+     */
+    public Timestamp getResumableTimestamp();
+    
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyCommunicationGroup.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyCommunicationGroup.java
new file mode 100644
index 00000000..0394d92c
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyCommunicationGroup.java
@@ -0,0 +1,171 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timely;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.TreeSet;
+import org.apache.log4j.Logger;
+import tools.refinery.viatra.runtime.matchers.util.CollectionsFactory;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.MessageSelector;
+import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+import tools.refinery.viatra.runtime.rete.network.mailbox.timely.TimelyMailbox;
+import tools.refinery.viatra.runtime.rete.util.Options;
+/**
+ * A timely communication group implementation. {@link TimelyMailbox}es and {@link LazyFoldingNode}s are ordered in the
+ * increasing order of timestamps.
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public class TimelyCommunicationGroup extends CommunicationGroup {
+    private final boolean isSingleton;
+    private final TreeMap<Timestamp, Set<Mailbox>> mailboxQueue;
+    // may be null - only used in the scattered case where we need to take care of mailboxes and resumables too
+    private Comparator<Node> nodeComparator;
+    private boolean currentlyDelivering;
+    private Timestamp currentlyDeliveredTimestamp;
+    public TimelyCommunicationGroup(final TimelyCommunicationTracker tracker, final Node representative,
+            final int identifier, final boolean isSingleton) {
+        super(tracker, representative, identifier);
+        this.isSingleton = isSingleton;
+        this.mailboxQueue = CollectionsFactory.createTreeMap();
+        this.currentlyDelivering = false;
+    }
+    /**
+     * Sets the {@link Comparator} to be used to order the {@link Mailbox}es at a given {@link Timestamp} in the mailbox
+     * queue. Additionally, reorders already queued {@link Mailbox}es to reflect the new comparator. The comparator may
+     * be null, in this case, no set ordering will be enforced among the {@link Mailbox}es.
+     */
+    public void setComparatorAndReorderMailboxes(final Comparator<Node> nodeComparator) {
+        this.nodeComparator = nodeComparator;
+        if (!this.mailboxQueue.isEmpty()) {
+            final HashMap<Timestamp, Set<Mailbox>> queueCopy = new HashMap<Timestamp, Set<Mailbox>>(this.mailboxQueue);
+            this.mailboxQueue.clear();
+            for (final Entry<Timestamp, Set<Mailbox>> entry : queueCopy.entrySet()) {
+                for (final Mailbox mailbox : entry.getValue()) {
+                    this.notifyHasMessage(mailbox, entry.getKey());
+                }
+            }
+        }
+    }
+    @Override
+    public void deliverMessages() {
+        this.currentlyDelivering = true;
+        while (!this.mailboxQueue.isEmpty()) {
+            // care must be taken here how we iterate over the mailboxes
+            // it is not okay to loop over the mailboxes at once because a mailbox may disappear from the collection as
+            // a result of delivering messages from another mailboxes under the same timestamp
+            // because of this, it is crucial that we pick the mailboxes one by one
+            final Entry<Timestamp, Set<Mailbox>> entry = this.mailboxQueue.firstEntry();
+            final Timestamp timestamp = entry.getKey();
+            final Set<Mailbox> mailboxes = entry.getValue();
+            final Mailbox mailbox = mailboxes.iterator().next();
+            mailboxes.remove(mailbox);
+            if (mailboxes.isEmpty()) {
+                this.mailboxQueue.pollFirstEntry();
+            }
+            assert mailbox instanceof TimelyMailbox;
+            /* debug */ this.currentlyDeliveredTimestamp = timestamp;
+            mailbox.deliverAll(timestamp);
+            /* debug */ this.currentlyDeliveredTimestamp = null;
+        }
+        this.currentlyDelivering = false;
+    }
+    @Override
+    public boolean isEmpty() {
+        return this.mailboxQueue.isEmpty();
+    }
+    @Override
+    public void notifyHasMessage(final Mailbox mailbox, MessageSelector kind) {
+        if (kind instanceof Timestamp) {
+            final Timestamp timestamp = (Timestamp) kind;
+            if (Options.MONITOR_VIOLATION_OF_DIFFERENTIAL_DATAFLOW_TIMESTAMPS) {
+                if (timestamp.compareTo(this.currentlyDeliveredTimestamp) < 0) {
+                    final Logger logger = this.representative.getContainer().getNetwork().getEngine().getLogger();
+                    logger.error(
+                            "[INTERNAL ERROR] Violation of differential dataflow communication schema! The communication component with representative "
+                                    + this.representative + " observed decreasing timestamp during message delivery!");
+                }
+            }
+            final Set<Mailbox> mailboxes = this.mailboxQueue.computeIfAbsent(timestamp, k -> {
+                if (this.nodeComparator == null) {
+                    return CollectionsFactory.createSet();
+                } else {
+                    return new TreeSet<Mailbox>(new Comparator<Mailbox>() {
+                        @Override
+                        public int compare(final Mailbox left, final Mailbox right) {
+                            return nodeComparator.compare(left.getReceiver(), right.getReceiver());
+                        }
+                    });
+                }
+            });
+            mailboxes.add(mailbox);
+            if (!this.isEnqueued && !this.currentlyDelivering) {
+                this.tracker.activateUnenqueued(this);
+            }
+        } else {
+            throw new IllegalArgumentException(UNSUPPORTED_MESSAGE_KIND + kind);
+        }
+    }
+    @Override
+    public void notifyLostAllMessages(final Mailbox mailbox, final MessageSelector kind) {
+        if (kind instanceof Timestamp) {
+            final Timestamp timestamp = (Timestamp) kind;
+            this.mailboxQueue.compute(timestamp, (k, v) -> {
+                if (v == null) {
+                    throw new IllegalStateException("No mailboxes registered at timestamp " + timestamp + "!");
+                }
+                if (!v.remove(mailbox)) {
+                    throw new IllegalStateException(
+                            "The mailbox " + mailbox + " was not registered at timestamp " + timestamp + "!");
+                }
+                if (v.isEmpty()) {
+                    return null;
+                } else {
+                    return v;
+                }
+            });
+            if (this.mailboxQueue.isEmpty()) {
+                this.tracker.deactivate(this);
+            }
+        } else {
+            throw new IllegalArgumentException(UNSUPPORTED_MESSAGE_KIND + kind);
+        }
+    }
+    @Override
+    public Map<MessageSelector, Collection<Mailbox>> getMailboxes() {
+        return Collections.unmodifiableMap(this.mailboxQueue);
+    }
+    @Override
+    public boolean isRecursive() {
+        return !this.isSingleton;
+    }
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyCommunicationTracker.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyCommunicationTracker.java
new file mode 100644
index 00000000..79179880
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyCommunicationTracker.java
@@ -0,0 +1,216 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timely;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.function.Function;
+import tools.refinery.viatra.runtime.rete.itc.alg.misc.topsort.TopologicalSorting;
+import tools.refinery.viatra.runtime.rete.itc.graphimpl.Graph;
+import tools.refinery.viatra.runtime.matchers.util.CollectionsFactory;
+import tools.refinery.viatra.runtime.rete.index.IndexerListener;
+import tools.refinery.viatra.runtime.rete.index.SpecializedProjectionIndexer;
+import tools.refinery.viatra.runtime.rete.index.SpecializedProjectionIndexer.ListenerSubscription;
+import tools.refinery.viatra.runtime.rete.index.StandardIndexer;
+import tools.refinery.viatra.runtime.rete.matcher.TimelyConfiguration;
+import tools.refinery.viatra.runtime.rete.matcher.TimelyConfiguration.TimelineRepresentation;
+import tools.refinery.viatra.runtime.rete.network.NetworkStructureChangeSensitiveNode;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.ProductionNode;
+import tools.refinery.viatra.runtime.rete.network.StandardNode;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationTracker;
+import tools.refinery.viatra.runtime.rete.network.communication.MessageSelector;
+import tools.refinery.viatra.runtime.rete.network.communication.NodeComparator;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+import tools.refinery.viatra.runtime.rete.single.DiscriminatorDispatcherNode;
+/**
+ * Timely (DDF) implementation of the {@link CommunicationTracker}.
+ *
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public class TimelyCommunicationTracker extends CommunicationTracker {
+    protected final TimelyConfiguration configuration;
+    public TimelyCommunicationTracker(final TimelyConfiguration configuration) {
+        this.configuration = configuration;
+    }
+    @Override
+    protected CommunicationGroup createGroup(final Node representative, final int index) {
+        final boolean isSingleton = this.sccInformationProvider.sccs.getPartition(representative).size() == 1;
+        return new TimelyCommunicationGroup(this, representative, index, isSingleton);
+    }
+    @Override
+    protected void reconstructQueueContents(final Set<CommunicationGroup> oldActiveGroups) {
+        for (final CommunicationGroup oldGroup : oldActiveGroups) {
+            for (final Entry<MessageSelector, Collection<Mailbox>> entry : oldGroup.getMailboxes().entrySet()) {
+                for (final Mailbox mailbox : entry.getValue()) {
+                    final CommunicationGroup newGroup = this.groupMap.get(mailbox.getReceiver());
+                    newGroup.notifyHasMessage(mailbox, entry.getKey());
+                }
+            }
+        }
+    }
+    @Override
+    public Mailbox proxifyMailbox(final Node requester, final Mailbox original) {
+        final Mailbox mailboxToProxify = (original instanceof TimelyMailboxProxy)
+                ? ((TimelyMailboxProxy) original).getWrappedMailbox()
+                : original;
+        final TimestampTransformation preprocessor = getPreprocessor(requester, mailboxToProxify.getReceiver());
+        if (preprocessor == null) {
+            return mailboxToProxify;
+        } else {
+            return new TimelyMailboxProxy(mailboxToProxify, preprocessor);
+        }
+    }
+    @Override
+    public IndexerListener proxifyIndexerListener(final Node requester, final IndexerListener original) {
+        final IndexerListener listenerToProxify = (original instanceof TimelyIndexerListenerProxy)
+                ? ((TimelyIndexerListenerProxy) original).getWrappedIndexerListener()
+                : original;
+        final TimestampTransformation preprocessor = getPreprocessor(requester, listenerToProxify.getOwner());
+        if (preprocessor == null) {
+            return listenerToProxify;
+        } else {
+            return new TimelyIndexerListenerProxy(listenerToProxify, preprocessor);
+        }
+    }
+    protected TimestampTransformation getPreprocessor(final Node source, final Node target) {
+        final Node effectiveSource = source instanceof SpecializedProjectionIndexer
+                ? ((SpecializedProjectionIndexer) source).getActiveNode()
+                : source;
+        final CommunicationGroup sourceGroup = this.getGroup(effectiveSource);
+        final CommunicationGroup targetGroup = this.getGroup(target);
+        if (sourceGroup != null && targetGroup != null) {
+            // during RETE construction, the groups may be still null
+            if (sourceGroup != targetGroup && sourceGroup.isRecursive()) {
+                // targetGroup is a successor SCC of sourceGroup
+                // and sourceGroup is a recursive SCC
+                // then we need to zero out the timestamps
+                return TimestampTransformation.RESET;
+            }
+            if (sourceGroup == targetGroup && target instanceof ProductionNode) {
+                // if requester and receiver are in the same SCC
+                // and receiver is a production node
+                // then we need to increment the timestamps
+                return TimestampTransformation.INCREMENT;
+            }
+        }
+        return null;
+    }
+    @Override
+    protected void postProcessNode(final Node node) {
+        if (node instanceof NetworkStructureChangeSensitiveNode) {
+            ((NetworkStructureChangeSensitiveNode) node).networkStructureChanged();
+        }
+    }
+    @Override
+    protected void postProcessGroup(final CommunicationGroup group) {
+        if (this.configuration.getTimelineRepresentation() == TimelineRepresentation.FAITHFUL) {
+            final Node representative = group.getRepresentative();
+            final Set<Node> groupMembers = this.sccInformationProvider.sccs.getPartition(representative);
+            if (groupMembers.size() > 1) {
+                final Graph<Node> graph = new Graph<Node>();
+                for (final Node node : groupMembers) {
+                    graph.insertNode(node);
+                }
+                for (final Node source : groupMembers) {
+                    for (final Node target : this.dependencyGraph.getTargetNodes(source)) {
+                        // (1) the edge is not a recursion cut point
+                        // (2) the edge is within this group
+                        if (!this.isRecursionCutPoint(source, target) && groupMembers.contains(target)) {
+                            graph.insertEdge(source, target);
+                        }
+                    }
+                }
+                final List<Node> orderedNodes = TopologicalSorting.compute(graph);
+                final Map<Node, Integer> nodeMap = CollectionsFactory.createMap();
+                int identifier = 0;
+                for (final Node orderedNode : orderedNodes) {
+                    nodeMap.put(orderedNode, identifier++);
+                }
+                ((TimelyCommunicationGroup) group).setComparatorAndReorderMailboxes(new NodeComparator(nodeMap));
+            }
+        }
+    }
+    /**
+     * This static field is used for debug purposes in the DotGenerator.
+     */
+    public static final Function<Node, Function<Node, String>> EDGE_LABEL_FUNCTION = new Function<Node, Function<Node, String>>() {
+        @Override
+        public Function<Node, String> apply(final Node source) {
+            return new Function<Node, String>() {
+                @Override
+                public String apply(final Node target) {
+                    if (source instanceof SpecializedProjectionIndexer) {
+                        final Collection<ListenerSubscription> subscriptions = ((SpecializedProjectionIndexer) source)
+                                .getSubscriptions();
+                        for (final ListenerSubscription subscription : subscriptions) {
+                            if (subscription.getListener().getOwner() == target
+                                    && subscription.getListener() instanceof TimelyIndexerListenerProxy) {
+                                return ((TimelyIndexerListenerProxy) subscription.getListener()).preprocessor
+                                        .toString();
+                            }
+                        }
+                    }
+                    if (source instanceof StandardIndexer) {
+                        final Collection<IndexerListener> listeners = ((StandardIndexer) source).getListeners();
+                        for (final IndexerListener listener : listeners) {
+                            if (listener.getOwner() == target && listener instanceof TimelyIndexerListenerProxy) {
+                                return ((TimelyIndexerListenerProxy) listener).preprocessor.toString();
+                            }
+                        }
+                    }
+                    if (source instanceof StandardNode) {
+                        final Collection<Mailbox> mailboxes = ((StandardNode) source).getChildMailboxes();
+                        for (final Mailbox mailbox : mailboxes) {
+                            if (mailbox.getReceiver() == target && mailbox instanceof TimelyMailboxProxy) {
+                                return ((TimelyMailboxProxy) mailbox).preprocessor.toString();
+                            }
+                        }
+                    }
+                    if (source instanceof DiscriminatorDispatcherNode) {
+                        final Collection<Mailbox> mailboxes = ((DiscriminatorDispatcherNode) source)
+                                .getBucketMailboxes().values();
+                        for (final Mailbox mailbox : mailboxes) {
+                            if (mailbox.getReceiver() == target && mailbox instanceof TimelyMailboxProxy) {
+                                return ((TimelyMailboxProxy) mailbox).preprocessor.toString();
+                            }
+                        }
+                    }
+                    return null;
+                }
+            };
+        }
+    };
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyIndexerListenerProxy.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyIndexerListenerProxy.java
new file mode 100644
index 00000000..e8fbf84e
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyIndexerListenerProxy.java
@@ -0,0 +1,81 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timely;
+import tools.refinery.viatra.runtime.matchers.tuple.Tuple;
+import tools.refinery.viatra.runtime.matchers.util.Direction;
+import tools.refinery.viatra.runtime.matchers.util.Preconditions;
+import tools.refinery.viatra.runtime.rete.index.IndexerListener;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.ProductionNode;
+import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
+/**
+ * A timely proxy for another {@link IndexerListener}, which performs some preprocessing 
+ * on the differential timestamps before passing it on to the real recipient. 
+ * <p>
+ * These proxies are used on edges leading into {@link ProductionNode}s. Because {@link ProductionNode}s 
+ * never ask back the indexer for its contents, there is no need to also apply the proxy on that direction. 
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public class TimelyIndexerListenerProxy implements IndexerListener {
+    protected final TimestampTransformation preprocessor;
+    protected final IndexerListener wrapped;
+    public TimelyIndexerListenerProxy(final IndexerListener wrapped,
+            final TimestampTransformation preprocessor) {
+        Preconditions.checkArgument(!(wrapped instanceof TimelyIndexerListenerProxy), "Proxy in a proxy is not allowed!");
+        this.wrapped = wrapped;
+        this.preprocessor = preprocessor;
+    }
+    
+    public IndexerListener getWrappedIndexerListener() {
+        return wrapped;
+    }
+    @Override
+    public Node getOwner() {
+        return this.wrapped.getOwner();
+    }
+    @Override
+    public void notifyIndexerUpdate(final Direction direction, final Tuple updateElement, final Tuple signature,
+            final boolean change, final Timestamp timestamp) {
+        this.wrapped.notifyIndexerUpdate(direction, updateElement, signature, change, preprocessor.process(timestamp));
+    }
+    @Override
+    public String toString() {
+        return this.preprocessor.toString() + "_PROXY -> " + this.wrapped.toString();
+    }
+    @Override
+    public boolean equals(final Object obj) {
+        if (obj == null || obj.getClass() != this.getClass()) {
+            return false;
+        } else if (obj == this) {
+            return true;
+        } else {
+            final TimelyIndexerListenerProxy that = (TimelyIndexerListenerProxy) obj;
+            return this.wrapped.equals(that.wrapped) && this.preprocessor == that.preprocessor;
+        }
+    }
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = hash * 17 + this.wrapped.hashCode();
+        hash = hash * 31 + this.preprocessor.hashCode();
+        return hash;
+    }
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyMailboxProxy.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyMailboxProxy.java
new file mode 100644
index 00000000..550bfbeb
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimelyMailboxProxy.java
@@ -0,0 +1,102 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timely;
+import tools.refinery.viatra.runtime.matchers.tuple.Tuple;
+import tools.refinery.viatra.runtime.matchers.util.Direction;
+import tools.refinery.viatra.runtime.matchers.util.Preconditions;
+import tools.refinery.viatra.runtime.rete.network.Receiver;
+import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
+import tools.refinery.viatra.runtime.rete.network.communication.MessageSelector;
+import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
+import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
+/**
+ * A timely proxy for another {@link Mailbox}, which performs some preprocessing 
+ * on the differential timestamps before passing it on to the real recipient. 
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public class TimelyMailboxProxy implements Mailbox {
+    protected final TimestampTransformation preprocessor;
+    protected final Mailbox wrapped;
+    public TimelyMailboxProxy(final Mailbox wrapped, final TimestampTransformation preprocessor) {
+        Preconditions.checkArgument(!(wrapped instanceof TimelyMailboxProxy), "Proxy in a proxy is not allowed!");
+        this.wrapped = wrapped;
+        this.preprocessor = preprocessor;
+    }
+    
+    public Mailbox getWrappedMailbox() {
+        return wrapped;
+    }
+    @Override
+    public void postMessage(final Direction direction, final Tuple update, final Timestamp timestamp) {
+        this.wrapped.postMessage(direction, update, preprocessor.process(timestamp));
+    }
+    @Override
+    public String toString() {
+        return this.preprocessor.toString() + "_PROXY -> " + this.wrapped.toString();
+    }
+    @Override
+    public void clear() {
+        this.wrapped.clear();
+    }
+    @Override
+    public void deliverAll(final MessageSelector selector) {
+        this.wrapped.deliverAll(selector);
+    }
+    @Override
+    public CommunicationGroup getCurrentGroup() {
+        return this.wrapped.getCurrentGroup();
+    }
+    @Override
+    public void setCurrentGroup(final CommunicationGroup group) {
+        this.wrapped.setCurrentGroup(group);
+    }
+    @Override
+    public Receiver getReceiver() {
+        return this.wrapped.getReceiver();
+    }
+    @Override
+    public boolean isEmpty() {
+        return this.wrapped.isEmpty();
+    }
+    @Override
+    public boolean equals(final Object obj) {
+        if (obj == null || obj.getClass() != this.getClass()) {
+            return false;
+        } else if (obj == this) {
+            return true;
+        } else {
+            final TimelyMailboxProxy that = (TimelyMailboxProxy) obj;
+            return this.wrapped.equals(that.wrapped) && this.preprocessor == that.preprocessor;
+        }
+    }
+    @Override
+    public int hashCode() {
+        int hash = 1;
+        hash = hash * 17 + this.wrapped.hashCode();
+        hash = hash * 31 + this.preprocessor.hashCode();
+        return hash;
+    }
+}
diff --git a/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimestampTransformation.java b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimestampTransformation.java
new file mode 100644
index 00000000..8929eb5c
--- /dev/null
+++ b/subprojects/viatra-runtime-rete/src/main/java/tools/refinery/viatra/runtime/rete/network/communication/timely/TimestampTransformation.java
@@ -0,0 +1,48 @@
+/*******************************************************************************
+ * Copyright (c) 2010-2019, Tamas Szabo, 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.network.communication.timely;
+import tools.refinery.viatra.runtime.rete.network.Node;
+import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
+/**
+ * Values of this enum perform different kind of preprocessing on {@link Timestamp}s. 
+ * This is used on edges leading in and out from {@link Node}s in recursive {@link TimelyCommunicationGroup}s. 
+ * 
+ * @author Tamas Szabo
+ * @since 2.3
+ */
+public enum TimestampTransformation {
+    INCREMENT {
+        @Override
+        public Timestamp process(final Timestamp timestamp) {
+            return new Timestamp(timestamp.getValue() + 1);
+        }
+        
+        @Override
+        public String toString() {
+            return "INCREMENT";
+        }
+    },
+    RESET {
+        @Override
+        public Timestamp process(final Timestamp timestamp) {
+            return Timestamp.ZERO;
+        }
+        
+        @Override
+        public String toString() {
+            return "RESET";
+        }
+    };
+    public abstract Timestamp process(final Timestamp timestamp);
+}