1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
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;
}
};
}
};
}
|
