FlowEntry must not be modified outside store.
- Remove set method from FlowEntry
- Storing last seen timestamp for FlowEntry eviction locally on FlowManager.
FlowEntry eviction based on packet counter will take longer time to timeout
after master Node change.
Change-Id: I7134d698dd5b9bf7cca379c5ba7c4fbcc2e3d5f3
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/DefaultFlowEntry.java b/core/api/src/main/java/org/onlab/onos/net/flow/DefaultFlowEntry.java
index 7513894..905469f 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/DefaultFlowEntry.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/DefaultFlowEntry.java
@@ -6,7 +6,8 @@
import org.onlab.onos.net.DeviceId;
import org.slf4j.Logger;
-public class DefaultFlowEntry extends DefaultFlowRule implements FlowEntry {
+public class DefaultFlowEntry extends DefaultFlowRule
+ implements FlowEntry, StoredFlowEntry {
private static final Logger log = getLogger(DefaultFlowEntry.class);
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/FlowEntry.java b/core/api/src/main/java/org/onlab/onos/net/flow/FlowEntry.java
index 882c9df..cdccaa9 100644
--- a/core/api/src/main/java/org/onlab/onos/net/flow/FlowEntry.java
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/FlowEntry.java
@@ -65,6 +65,7 @@
*/
long bytes();
+ // TODO: consider removing this attribute
/**
* When this flow entry was last deemed active.
* @return epoch time of last activity
@@ -72,35 +73,6 @@
long lastSeen();
/**
- * Sets the last active epoch time.
- */
- void setLastSeen();
-
- /**
- * Sets the new state for this entry.
- * @param newState new flow entry state.
- */
- void setState(FlowEntryState newState);
-
- /**
- * Sets how long this entry has been entered in the system.
- * @param life epoch time
- */
- void setLife(long life);
-
- /**
- * Number of packets seen by this entry.
- * @param packets a long value
- */
- void setPackets(long packets);
-
- /**
- * Number of bytes seen by this rule.
- * @param bytes a long value
- */
- void setBytes(long bytes);
-
- /**
* Indicates the error type.
* @return an integer value of the error
*/
diff --git a/core/api/src/main/java/org/onlab/onos/net/flow/StoredFlowEntry.java b/core/api/src/main/java/org/onlab/onos/net/flow/StoredFlowEntry.java
new file mode 100644
index 0000000..e68ed68
--- /dev/null
+++ b/core/api/src/main/java/org/onlab/onos/net/flow/StoredFlowEntry.java
@@ -0,0 +1,35 @@
+package org.onlab.onos.net.flow;
+
+
+public interface StoredFlowEntry extends FlowEntry {
+
+ /**
+ * Sets the last active epoch time.
+ */
+ void setLastSeen();
+
+ /**
+ * Sets the new state for this entry.
+ * @param newState new flow entry state.
+ */
+ void setState(FlowEntryState newState);
+
+ /**
+ * Sets how long this entry has been entered in the system.
+ * @param life epoch time
+ */
+ void setLife(long life);
+
+ /**
+ * Number of packets seen by this entry.
+ * @param packets a long value
+ */
+ void setPackets(long packets);
+
+ /**
+ * Number of bytes seen by this rule.
+ * @param bytes a long value
+ */
+ void setBytes(long bytes);
+
+}
diff --git a/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java b/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
index ad48ae7..9ea99c3 100644
--- a/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
+++ b/core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java
@@ -3,8 +3,8 @@
import static com.google.common.base.Preconditions.checkNotNull;
import static org.slf4j.LoggerFactory.getLogger;
-import java.util.Iterator;
import java.util.List;
+import java.util.Map;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
@@ -45,6 +45,7 @@
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
import com.google.common.collect.Multimap;
/**
@@ -197,6 +198,8 @@
extends AbstractProviderService<FlowRuleProvider>
implements FlowRuleProviderService {
+ final Map<FlowEntry, Long> lastSeen = Maps.newConcurrentMap();
+
protected InternalFlowRuleProviderService(FlowRuleProvider provider) {
super(provider);
}
@@ -205,6 +208,7 @@
public void flowRemoved(FlowEntry flowEntry) {
checkNotNull(flowEntry, FLOW_RULE_NULL);
checkValidity();
+ lastSeen.remove(flowEntry);
FlowEntry stored = store.getFlowEntry(flowEntry);
if (stored == null) {
log.info("Rule already evicted from store: {}", flowEntry);
@@ -292,14 +296,25 @@
if (storedRule == null) {
return false;
}
- long timeout = storedRule.timeout() * 1000;
- Long currentTime = System.currentTimeMillis();
+ final long timeout = storedRule.timeout() * 1000;
+ final long currentTime = System.currentTimeMillis();
if (storedRule.packets() != swRule.packets()) {
- storedRule.setLastSeen();
+ lastSeen.put(storedRule, currentTime);
return true;
}
+ if (!lastSeen.containsKey(storedRule)) {
+ // checking for the first time
+ lastSeen.put(storedRule, storedRule.lastSeen());
+ // Use following if lastSeen attr. was removed.
+ //lastSeen.put(storedRule, currentTime);
+ }
+ Long last = lastSeen.get(storedRule);
+ if (last == null) {
+ // concurrently removed? let the liveness check fail
+ return false;
+ }
- if ((currentTime - storedRule.lastSeen()) <= timeout) {
+ if ((currentTime - last) <= timeout) {
return true;
}
return false;
@@ -316,10 +331,7 @@
public void pushFlowMetrics(DeviceId deviceId, Iterable<FlowEntry> flowEntries) {
List<FlowEntry> storedRules = Lists.newLinkedList(store.getFlowEntries(deviceId));
- Iterator<FlowEntry> switchRulesIterator = flowEntries.iterator();
-
- while (switchRulesIterator.hasNext()) {
- FlowEntry rule = switchRulesIterator.next();
+ for (FlowEntry rule : flowEntries) {
if (storedRules.remove(rule)) {
// we both have the rule, let's update some info then.
flowAdded(rule);
diff --git a/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java b/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
index 472416a..dff740a 100644
--- a/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
+++ b/core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java
@@ -46,6 +46,7 @@
import org.onlab.onos.net.flow.FlowRuleProviderRegistry;
import org.onlab.onos.net.flow.FlowRuleProviderService;
import org.onlab.onos.net.flow.FlowRuleService;
+import org.onlab.onos.net.flow.StoredFlowEntry;
import org.onlab.onos.net.flow.TrafficSelector;
import org.onlab.onos.net.flow.TrafficTreatment;
import org.onlab.onos.net.flow.criteria.Criterion;
@@ -232,7 +233,7 @@
public void flowRemoved() {
FlowRule f1 = addFlowRule(1);
FlowRule f2 = addFlowRule(2);
- FlowEntry fe1 = new DefaultFlowEntry(f1);
+ StoredFlowEntry fe1 = new DefaultFlowEntry(f1);
FlowEntry fe2 = new DefaultFlowEntry(f2);
providerService.pushFlowMetrics(DID, ImmutableList.of(fe1, fe2));
service.removeFlowRules(f1);
diff --git a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
index 9c67f69..dbe6734 100644
--- a/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
+++ b/core/store/dist/src/main/java/org/onlab/onos/store/flow/impl/DistributedFlowRuleStore.java
@@ -26,6 +26,7 @@
import org.onlab.onos.net.flow.FlowRuleEvent.Type;
import org.onlab.onos.net.flow.FlowRuleStore;
import org.onlab.onos.net.flow.FlowRuleStoreDelegate;
+import org.onlab.onos.net.flow.StoredFlowEntry;
import org.onlab.onos.store.AbstractStore;
import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
import org.onlab.onos.store.cluster.messaging.ClusterMessage;
@@ -53,8 +54,8 @@
private final Logger log = getLogger(getClass());
// store entries as a pile of rules, no info about device tables
- private final Multimap<DeviceId, FlowEntry> flowEntries =
- ArrayListMultimap.<DeviceId, FlowEntry>create();
+ private final Multimap<DeviceId, StoredFlowEntry> flowEntries =
+ ArrayListMultimap.<DeviceId, StoredFlowEntry>create();
private final Multimap<Short, FlowRule> flowEntriesById =
ArrayListMultimap.<Short, FlowRule>create();
@@ -99,7 +100,11 @@
@Override
public synchronized FlowEntry getFlowEntry(FlowRule rule) {
- for (FlowEntry f : flowEntries.get(rule.deviceId())) {
+ return getFlowEntryInternal(rule);
+ }
+
+ private synchronized StoredFlowEntry getFlowEntryInternal(FlowRule rule) {
+ for (StoredFlowEntry f : flowEntries.get(rule.deviceId())) {
if (f.equals(rule)) {
return f;
}
@@ -109,7 +114,7 @@
@Override
public synchronized Iterable<FlowEntry> getFlowEntries(DeviceId deviceId) {
- Collection<FlowEntry> rules = flowEntries.get(deviceId);
+ Collection<? extends FlowEntry> rules = flowEntries.get(deviceId);
if (rules == null) {
return Collections.emptyList();
}
@@ -148,7 +153,7 @@
}
private synchronized void storeFlowEntryInternal(FlowRule flowRule) {
- FlowEntry flowEntry = new DefaultFlowEntry(flowRule);
+ StoredFlowEntry flowEntry = new DefaultFlowEntry(flowRule);
DeviceId deviceId = flowRule.deviceId();
// write to local copy.
if (!flowEntries.containsEntry(deviceId, flowEntry)) {
@@ -182,7 +187,7 @@
}
private synchronized void deleteFlowRuleInternal(FlowRule flowRule) {
- FlowEntry entry = getFlowEntry(flowRule);
+ StoredFlowEntry entry = getFlowEntryInternal(flowRule);
if (entry == null) {
return;
}
@@ -215,7 +220,7 @@
DeviceId did = rule.deviceId();
// check if this new rule is an update to an existing entry
- FlowEntry stored = getFlowEntry(rule);
+ StoredFlowEntry stored = getFlowEntryInternal(rule);
if (stored != null) {
stored.setBytes(rule.bytes());
stored.setLife(rule.life());
@@ -227,7 +232,8 @@
return new FlowRuleEvent(Type.RULE_UPDATED, rule);
}
- flowEntries.put(did, rule);
+ // TODO: Confirm if this behavior is correct. See SimpleFlowRuleStore
+ flowEntries.put(did, new DefaultFlowEntry(rule));
return null;
// TODO: also update backup.
diff --git a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
index b6ebda4..8ca2ca0 100644
--- a/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
+++ b/core/store/trivial/src/main/java/org/onlab/onos/store/trivial/impl/SimpleFlowRuleStore.java
@@ -5,6 +5,7 @@
import static org.apache.commons.lang3.concurrent.ConcurrentUtils.createIfAbsentUnchecked;
import static java.util.Collections.unmodifiableCollection;
+import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -25,6 +26,7 @@
import org.onlab.onos.net.flow.FlowRuleEvent.Type;
import org.onlab.onos.net.flow.FlowRuleStore;
import org.onlab.onos.net.flow.FlowRuleStoreDelegate;
+import org.onlab.onos.net.flow.StoredFlowEntry;
import org.onlab.onos.store.AbstractStore;
import org.onlab.util.NewConcurrentHashMap;
import org.slf4j.Logger;
@@ -43,7 +45,7 @@
// inner Map is Device flow table
// Assumption: FlowId cannot have synonyms
- private final ConcurrentMap<DeviceId, ConcurrentMap<FlowId, FlowEntry>>
+ private final ConcurrentMap<DeviceId, ConcurrentMap<FlowId, StoredFlowEntry>>
flowEntries = new ConcurrentHashMap<>();
@Activate
@@ -61,14 +63,14 @@
@Override
public int getFlowRuleCount() {
int sum = 0;
- for (ConcurrentMap<FlowId, FlowEntry> ft : flowEntries.values()) {
+ for (ConcurrentMap<FlowId, StoredFlowEntry> ft : flowEntries.values()) {
sum += ft.size();
}
return sum;
}
- private static NewConcurrentHashMap<FlowId, FlowEntry> lazyEmptyFlowTable() {
- return NewConcurrentHashMap.<FlowId, FlowEntry>ifNeeded();
+ private static NewConcurrentHashMap<FlowId, StoredFlowEntry> lazyEmptyFlowTable() {
+ return NewConcurrentHashMap.<FlowId, StoredFlowEntry>ifNeeded();
}
/**
@@ -77,12 +79,12 @@
* @param deviceId identifier of the device
* @return Map representing Flow Table of given device.
*/
- private ConcurrentMap<FlowId, FlowEntry> getFlowTable(DeviceId deviceId) {
+ private ConcurrentMap<FlowId, StoredFlowEntry> getFlowTable(DeviceId deviceId) {
return createIfAbsentUnchecked(flowEntries,
deviceId, lazyEmptyFlowTable());
}
- private FlowEntry getFlowEntry(DeviceId deviceId, FlowId flowId) {
+ private StoredFlowEntry getFlowEntry(DeviceId deviceId, FlowId flowId) {
return getFlowTable(deviceId).get(flowId);
}
@@ -93,7 +95,8 @@
@Override
public Iterable<FlowEntry> getFlowEntries(DeviceId deviceId) {
- return unmodifiableCollection(getFlowTable(deviceId).values());
+ return unmodifiableCollection((Collection<? extends FlowEntry>)
+ getFlowTable(deviceId).values());
}
@Override
@@ -101,7 +104,7 @@
Set<FlowRule> rules = new HashSet<>();
for (DeviceId did : flowEntries.keySet()) {
- ConcurrentMap<FlowId, FlowEntry> ft = getFlowTable(did);
+ ConcurrentMap<FlowId, StoredFlowEntry> ft = getFlowTable(did);
for (FlowEntry fe : ft.values()) {
if (fe.appId() == appId.id()) {
rules.add(fe);
@@ -117,7 +120,7 @@
}
private boolean storeFlowRuleInternal(FlowRule rule) {
- FlowEntry f = new DefaultFlowEntry(rule);
+ StoredFlowEntry f = new DefaultFlowEntry(rule);
final DeviceId did = f.deviceId();
final FlowId fid = f.id();
FlowEntry existing = getFlowTable(did).putIfAbsent(fid, f);
@@ -133,7 +136,7 @@
@Override
public void deleteFlowRule(FlowRule rule) {
- FlowEntry entry = getFlowEntry(rule.deviceId(), rule.id());
+ StoredFlowEntry entry = getFlowEntry(rule.deviceId(), rule.id());
if (entry == null) {
//log.warn("Cannot find rule {}", rule);
return;
@@ -146,7 +149,7 @@
@Override
public FlowRuleEvent addOrUpdateFlowRule(FlowEntry rule) {
// check if this new rule is an update to an existing entry
- FlowEntry stored = getFlowEntry(rule.deviceId(), rule.id());
+ StoredFlowEntry stored = getFlowEntry(rule.deviceId(), rule.id());
if (stored != null) {
synchronized (stored) {
stored.setBytes(rule.bytes());
@@ -174,7 +177,7 @@
// This is where one could mark a rule as removed and still keep it in the store.
final DeviceId did = rule.deviceId();
- ConcurrentMap<FlowId, FlowEntry> ft = getFlowTable(did);
+ ConcurrentMap<FlowId, StoredFlowEntry> ft = getFlowTable(did);
if (ft.remove(rule.id(), rule)) {
return new FlowRuleEvent(RULE_REMOVED, rule);
} else {