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
|
/*******************************************************************************
* 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.single;
import java.util.Collection;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
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.TimelyMemory;
import tools.refinery.viatra.runtime.matchers.util.timeline.Diff;
import tools.refinery.viatra.runtime.matchers.util.timeline.Timeline;
import tools.refinery.viatra.runtime.rete.index.ProjectionIndexer;
import tools.refinery.viatra.runtime.rete.index.timely.TimelyMemoryIdentityIndexer;
import tools.refinery.viatra.runtime.rete.index.timely.TimelyMemoryNullIndexer;
import tools.refinery.viatra.runtime.rete.matcher.TimelyConfiguration.TimelineRepresentation;
import tools.refinery.viatra.runtime.rete.network.ReteContainer;
import tools.refinery.viatra.runtime.rete.network.communication.CommunicationGroup;
import tools.refinery.viatra.runtime.rete.network.communication.Timestamp;
import tools.refinery.viatra.runtime.rete.network.communication.timely.ResumableNode;
import tools.refinery.viatra.runtime.rete.network.mailbox.Mailbox;
import tools.refinery.viatra.runtime.rete.network.mailbox.timely.TimelyMailbox;
/**
* Timely uniqueness enforcer node implementation.
*
* @author Tamas Szabo
* @noinstantiate This class is not intended to be instantiated by clients.
* @noextend This class is not intended to be subclassed by clients.
* @since 2.4
*/
public class TimelyUniquenessEnforcerNode extends AbstractUniquenessEnforcerNode implements ResumableNode {
protected final TimelyMemory<Timestamp> memory;
/**
* @since 2.4
*/
protected CommunicationGroup group;
public TimelyUniquenessEnforcerNode(final ReteContainer container, final int tupleWidth) {
super(container, tupleWidth);
this.memory = new TimelyMemory<Timestamp>(
container.getTimelyConfiguration().getTimelineRepresentation() == TimelineRepresentation.FAITHFUL);
container.registerClearable(this.memory);
this.mailbox = instantiateMailbox();
container.registerClearable(this.mailbox);
}
protected Mailbox instantiateMailbox() {
return new TimelyMailbox(this, this.reteContainer);
}
@Override
public void pullInto(final Collection<Tuple> collector, final boolean flush) {
for (final Tuple tuple : this.memory.getTuplesAtInfinity()) {
collector.add(tuple);
}
}
@Override
public CommunicationGroup getCurrentGroup() {
return this.group;
}
@Override
public void setCurrentGroup(final CommunicationGroup group) {
this.group = group;
}
@Override
public Set<Tuple> getTuples() {
return this.memory.getTuplesAtInfinity();
}
/**
* @since 2.4
*/
@Override
public Timestamp getResumableTimestamp() {
return this.memory.getResumableTimestamp();
}
/**
* @since 2.4
*/
@Override
public void resumeAt(final Timestamp timestamp) {
final Map<Tuple, Diff<Timestamp>> diffMap = this.memory.resumeAt(timestamp);
for (final Entry<Tuple, Diff<Timestamp>> entry : diffMap.entrySet()) {
for (final Signed<Timestamp> signed : entry.getValue()) {
propagate(signed.getDirection(), entry.getKey(), signed.getPayload());
}
}
final Timestamp nextTimestamp = this.memory.getResumableTimestamp();
if (nextTimestamp != null) {
this.group.notifyHasMessage(this.mailbox, nextTimestamp);
}
}
@Override
public void update(final Direction direction, final Tuple update, final Timestamp timestamp) {
Diff<Timestamp> resultDiff = null;
if (direction == Direction.INSERT) {
resultDiff = this.memory.put(update, timestamp);
} else {
try {
resultDiff = this.memory.remove(update, timestamp);
} catch (final IllegalStateException e) {
issueError("[INTERNAL ERROR] Duplicate deletion of " + update + " was detected in "
+ this.getClass().getName() + " " + this + " for pattern(s) "
+ getTraceInfoPatternsEnumerated(), e);
// diff will remain unset in case of the exception, it is time to return
return;
}
}
for (final Signed<Timestamp> signed : resultDiff) {
propagate(signed.getDirection(), update, signed.getPayload());
}
}
@Override
public void pullIntoWithTimeline(final Map<Tuple, Timeline<Timestamp>> collector, final boolean flush) {
collector.putAll(this.memory.asMap());
}
@Override
public ProjectionIndexer getNullIndexer() {
if (this.memoryNullIndexer == null) {
this.memoryNullIndexer = new TimelyMemoryNullIndexer(this.reteContainer, this.tupleWidth, this.memory, this,
this, this.specializedListeners);
this.getCommunicationTracker().registerDependency(this, this.memoryNullIndexer);
}
return this.memoryNullIndexer;
}
@Override
public ProjectionIndexer getIdentityIndexer() {
if (this.memoryIdentityIndexer == null) {
this.memoryIdentityIndexer = new TimelyMemoryIdentityIndexer(this.reteContainer, this.tupleWidth,
this.memory, this, this, this.specializedListeners);
this.getCommunicationTracker().registerDependency(this, this.memoryIdentityIndexer);
}
return this.memoryIdentityIndexer;
}
@Override
public void networkStructureChanged() {
super.networkStructureChanged();
}
}
|
