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
|
/*******************************************************************************
* Copyright (c) 2004-2008 Gabor Bergmann 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.index;
import java.util.Collection;
import java.util.Map;
import tools.refinery.viatra.runtime.matchers.tuple.Tuple;
import tools.refinery.viatra.runtime.matchers.util.Direction;
import tools.refinery.viatra.runtime.matchers.util.Signed;
import tools.refinery.viatra.runtime.matchers.util.timeline.Timeline;
import tools.refinery.viatra.runtime.rete.network.ReteContainer;
import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
/**
* Propagates all substitutions arriving at the PRIMARY slot if and only if (a matching substitution on the SECONDARY is
* present) xor (NEGATIVE).
*
* The negative parameter specifies whether this node checks for existence or non-existence.
* <p>
* It is mandatory in differential dataflow evaluation that the secondary parent is in an upstream dependency component
* (so that every secondary tuple comes with zero timestamp).
*
* @author Gabor Bergmann
*/
public class ExistenceNode extends DualInputNode {
protected boolean negative;
/**
* @param reteContainer
* @param negative
* if false, act as existence checker, otherwise a nonexistence-checker
*/
public ExistenceNode(final ReteContainer reteContainer, final boolean negative) {
super(reteContainer, null);
this.negative = negative;
this.logic = createLogic();
}
@Override
public Tuple calibrate(final Tuple primary, final Tuple secondary) {
return primary;
}
@Override
public void networkStructureChanged() {
if (this.reteContainer.isTimelyEvaluation() && this.secondarySlot != null
&& this.reteContainer.getCommunicationTracker().areInSameGroup(this, this.secondarySlot)) {
throw new IllegalStateException("Secondary parent must be in an upstream dependency component!");
}
super.networkStructureChanged();
}
private final NetworkStructureChangeSensitiveLogic TIMELESS = new NetworkStructureChangeSensitiveLogic() {
@Override
public void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush) {
throw new UnsupportedOperationException();
}
@Override
public void pullInto(final Collection<Tuple> collector, final boolean flush) {
if (primarySlot == null || secondarySlot == null) {
return;
}
if (flush) {
reteContainer.flushUpdates();
}
for (final Tuple signature : primarySlot.getSignatures()) {
// primaries can not be null due to the contract of IterableIndex.getSignatures()
final Collection<Tuple> primaries = primarySlot.get(signature);
final Collection<Tuple> opposites = secondarySlot.get(signature);
if ((opposites != null) ^ negative) {
collector.addAll(primaries);
}
}
}
@Override
public void notifyUpdate(final Side side, final Direction direction, final Tuple updateElement,
final Tuple signature, final boolean change, final Timestamp timestamp) {
// in the default case, all timestamps must be zero
assert Timestamp.ZERO.equals(timestamp);
switch (side) {
case PRIMARY:
if ((retrieveOpposites(side, signature) != null) ^ negative) {
propagateUpdate(direction, updateElement, timestamp);
}
break;
case SECONDARY:
if (change) {
final Collection<Tuple> opposites = retrieveOpposites(side, signature);
if (opposites != null) {
for (final Tuple opposite : opposites) {
propagateUpdate((negative ? direction.opposite() : direction), opposite, timestamp);
}
}
}
break;
case BOTH:
// in case the slots coincide,
// negative --> always empty
// !positive --> identity
if (!negative) {
propagateUpdate(direction, updateElement, timestamp);
}
break;
}
}
};
private final NetworkStructureChangeSensitiveLogic TIMELY = new NetworkStructureChangeSensitiveLogic() {
@Override
public void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush) {
if (primarySlot == null || secondarySlot == null) {
return;
}
if (flush) {
reteContainer.flushUpdates();
}
for (final Tuple signature : primarySlot.getSignatures()) {
// primaries can not be null due to the contract of IterableIndex.getSignatures()
final Map<Tuple, Timeline<Timestamp>> primaries = getTimeline(signature, primarySlot);
// see contract: secondary must be in an upstream SCC
final Collection<Tuple> opposites = secondarySlot.get(signature);
if ((opposites != null) ^ negative) {
for (final Tuple primary : primaries.keySet()) {
collector.put(primary, primaries.get(primary));
}
}
}
}
@Override
public void pullInto(final Collection<Tuple> collector, final boolean flush) {
ExistenceNode.this.TIMELESS.pullInto(collector, flush);
}
@Override
public void notifyUpdate(final Side side, final Direction direction, final Tuple updateElement,
final Tuple signature, final boolean change, final Timestamp timestamp) {
switch (side) {
case PRIMARY: {
final Collection<Tuple> opposites = secondarySlot.get(signature);
if ((opposites != null) ^ negative) {
propagateUpdate(direction, updateElement, timestamp);
}
break;
}
case SECONDARY: {
final Map<Tuple, Timeline<Timestamp>> opposites = primarySlot.getTimeline(signature);
if (change) {
if (opposites != null) {
for (final Tuple opposite : opposites.keySet()) {
for (final Signed<Timestamp> oppositeSigned : opposites.get(opposite).asChangeSequence()) {
final Direction product = direction.multiply(oppositeSigned.getDirection());
propagateUpdate((negative ? product.opposite() : product), opposite,
oppositeSigned.getPayload());
}
}
}
}
break;
}
case BOTH:
// in case the slots coincide,
// negative --> always empty
// positive --> identity
if (!negative) {
propagateUpdate(direction, updateElement, timestamp);
}
break;
}
}
};
@Override
protected NetworkStructureChangeSensitiveLogic createTimelessLogic() {
return this.TIMELESS;
}
@Override
protected NetworkStructureChangeSensitiveLogic createTimelyLogic() {
return this.TIMELY;
}
}
|
