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
|
/*******************************************************************************
* 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.tuple.TupleMask;
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;
/**
* @author Gabor Bergmann
*
*/
public class JoinNode extends DualInputNode {
public JoinNode(final ReteContainer reteContainer, final TupleMask complementerSecondaryMask) {
super(reteContainer, complementerSecondaryMask);
this.logic = createLogic();
}
@Override
public Tuple calibrate(final Tuple primary, final Tuple secondary) {
return unify(primary, secondary);
}
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) {
for (final Tuple primary : primaries) {
for (final Tuple opposite : opposites) {
collector.add(unify(primary, opposite));
}
}
}
}
}
@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);
final Collection<Tuple> opposites = retrieveOpposites(side, signature);
if (!coincidence) {
if (opposites != null) {
for (final Tuple opposite : opposites) {
propagateUpdate(direction, unify(side, updateElement, opposite), timestamp);
}
}
} else {
// compensate for coincidence of slots - this is the case when an Indexer is joined with itself
if (opposites != null) {
for (final Tuple opposite : opposites) {
if (opposite.equals(updateElement)) {
// handle self-joins of a single tuple separately
continue;
}
propagateUpdate(direction, unify(opposite, updateElement), timestamp);
propagateUpdate(direction, unify(updateElement, opposite), timestamp);
}
}
// handle self-joins here
propagateUpdate(direction, unify(updateElement, updateElement), timestamp);
}
}
};
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);
final Map<Tuple, Timeline<Timestamp>> opposites = getTimeline(signature, secondarySlot);
if (opposites != null) {
for (final Tuple primary : primaries.keySet()) {
for (final Tuple opposite : opposites.keySet()) {
final Timeline<Timestamp> primaryTimeline = primaries.get(primary);
final Timeline<Timestamp> oppositeTimeline = opposites.get(opposite);
final Timeline<Timestamp> mergedTimeline = primaryTimeline
.mergeMultiplicative(oppositeTimeline);
if (!mergedTimeline.isEmpty()) {
collector.put(unify(primary, opposite), mergedTimeline);
}
}
}
}
}
}
@Override
public void pullInto(final Collection<Tuple> collector, final boolean flush) {
JoinNode.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) {
final Indexer oppositeIndexer = getSlot(side.opposite());
final Map<Tuple, Timeline<Timestamp>> opposites = getTimeline(signature, oppositeIndexer);
if (!coincidence) {
if (opposites != null) {
for (final Tuple opposite : opposites.keySet()) {
final Tuple unifiedTuple = unify(side, updateElement, opposite);
for (final Signed<Timestamp> signed : opposites.get(opposite).asChangeSequence()) {
// TODO only consider signed timestamps that are greater or equal to timestamp
// plus compact the previous timestamps into at most one update
propagateUpdate(signed.getDirection().multiply(direction), unifiedTuple,
timestamp.max(signed.getPayload()));
}
}
}
} else {
// compensate for coincidence of slots - this is the case when an Indexer is joined with itself
if (opposites != null) {
for (final Tuple opposite : opposites.keySet()) {
if (opposite.equals(updateElement)) {
// handle self-joins of a single tuple separately
continue;
}
final Tuple u1 = unify(opposite, updateElement);
final Tuple u2 = unify(updateElement, opposite);
for (final Signed<Timestamp> oppositeSigned : opposites.get(opposite).asChangeSequence()) {
final Direction updateDirection = direction.multiply(oppositeSigned.getDirection());
final Timestamp updateTimestamp = timestamp.max(oppositeSigned.getPayload());
propagateUpdate(updateDirection, u1, updateTimestamp);
propagateUpdate(updateDirection, u2, updateTimestamp);
}
}
}
// handle self-join here
propagateUpdate(direction, unify(updateElement, updateElement), timestamp);
}
}
};
@Override
protected NetworkStructureChangeSensitiveLogic createTimelessLogic() {
return this.TIMELESS;
}
@Override
protected NetworkStructureChangeSensitiveLogic createTimelyLogic() {
return this.TIMELY;
}
}
|
