ONOS-4604 Fixed flow objective installation
Removed context from objective toString methods.
Removed duplicate flow objective delegate notifications in the store for next objectives.
Synchronized queueing of forwarding objectives for pending next objectives to avoid notifications race.
Changed logging for better readability.
Change-Id: Ic2bd411a891ea035a2c5513b24dea5fbd48f187d
diff --git a/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultFilteringObjective.java b/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultFilteringObjective.java
index 3cc646f..4cde989 100644
--- a/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultFilteringObjective.java
+++ b/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultFilteringObjective.java
@@ -160,7 +160,6 @@
.add("appId", appId())
.add("permanent", permanent())
.add("timeout", timeout())
- .add("context", context())
.toString();
}
diff --git a/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultForwardingObjective.java b/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultForwardingObjective.java
index f83d5e1..f076fda 100644
--- a/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultForwardingObjective.java
+++ b/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultForwardingObjective.java
@@ -169,7 +169,6 @@
.add("appId", appId())
.add("permanent", permanent())
.add("timeout", timeout())
- .add("context", context())
.toString();
}
diff --git a/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultNextObjective.java b/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultNextObjective.java
index dbb16a5..8279424 100644
--- a/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultNextObjective.java
+++ b/core/api/src/main/java/org/onosproject/net/flowobjective/DefaultNextObjective.java
@@ -138,7 +138,6 @@
.add("appId", appId())
.add("permanent", permanent())
.add("timeout", timeout())
- .add("context", context())
.toString();
}
diff --git a/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java b/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java
index d8a0dec..97c40d2 100644
--- a/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java
+++ b/core/net/src/main/java/org/onosproject/net/flowobjective/impl/FlowObjectiveManager.java
@@ -16,6 +16,7 @@
package org.onosproject.net.flowobjective.impl;
import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
@@ -55,21 +56,17 @@
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import static com.google.common.base.Preconditions.checkNotNull;
import static java.util.concurrent.Executors.newFixedThreadPool;
import static org.onlab.util.Tools.groupedThreads;
import static org.onosproject.security.AppGuard.checkPermission;
-import static org.onosproject.security.AppPermission.Type.*;
-
-
+import static org.onosproject.security.AppPermission.Type.FLOWRULE_WRITE;
/**
* Provides implementation of the flow objective programming service.
@@ -124,7 +121,7 @@
protected ServiceDirectory serviceDirectory = new DefaultServiceDirectory();
- private Map<Integer, Set<PendingNext>> pendingForwards = Maps.newConcurrentMap();
+ private final Map<Integer, Set<PendingNext>> pendingForwards = Maps.newConcurrentMap();
// local store to track which nextObjectives were sent to which device
// for debugging purposes
@@ -237,21 +234,33 @@
public void initPolicy(String policy) {}
private boolean queueObjective(DeviceId deviceId, ForwardingObjective fwd) {
- if (fwd.nextId() != null &&
- flowObjectiveStore.getNextGroup(fwd.nextId()) == null) {
- log.debug("Queuing forwarding objective {} for nextId {} meant for device {}",
- fwd.id(), fwd.nextId(), deviceId);
- // TODO: change to computeIfAbsent
- Set<PendingNext> newset = Collections.newSetFromMap(
- new ConcurrentHashMap<PendingNext, Boolean>());
- newset.add(new PendingNext(deviceId, fwd));
- Set<PendingNext> pnext = pendingForwards.putIfAbsent(fwd.nextId(), newset);
- if (pnext != null) {
- pnext.add(new PendingNext(deviceId, fwd));
- }
- return true;
+ if (fwd.nextId() == null ||
+ flowObjectiveStore.getNextGroup(fwd.nextId()) != null) {
+ // fast path
+ return false;
}
- return false;
+ boolean queued = false;
+ synchronized (pendingForwards) {
+ // double check the flow objective store, because this block could run
+ // after a notification arrives
+ if (flowObjectiveStore.getNextGroup(fwd.nextId()) == null) {
+ pendingForwards.compute(fwd.nextId(), (id, pending) -> {
+ PendingNext next = new PendingNext(deviceId, fwd);
+ if (pending == null) {
+ return Sets.newHashSet(next);
+ } else {
+ pending.add(next);
+ return pending;
+ }
+ });
+ queued = true;
+ }
+ }
+ if (queued) {
+ log.debug("Queued forwarding objective {} for nextId {} meant for device {}",
+ fwd.id(), fwd.nextId(), deviceId);
+ }
+ return queued;
}
// Retrieves the device pipeline behaviour from the cache.
@@ -396,7 +405,11 @@
public void notify(ObjectiveEvent event) {
if (event.type() == Type.ADD) {
log.debug("Received notification of obj event {}", event);
- Set<PendingNext> pending = pendingForwards.remove(event.subject());
+ Set<PendingNext> pending;
+ synchronized (pendingForwards) {
+ // needs to be synchronized for queueObjective lookup
+ pending = pendingForwards.remove(event.subject());
+ }
if (pending == null) {
log.debug("Nothing pending for this obj event {}", event);
diff --git a/core/net/src/main/java/org/onosproject/net/intent/impl/IntentInstaller.java b/core/net/src/main/java/org/onosproject/net/intent/impl/IntentInstaller.java
index fc5cbd9..68fc06e 100644
--- a/core/net/src/main/java/org/onosproject/net/intent/impl/IntentInstaller.java
+++ b/core/net/src/main/java/org/onosproject/net/intent/impl/IntentInstaller.java
@@ -120,7 +120,7 @@
// if toInstall was cause of error, then recompile (manage/increment counter, when exceeded -> CORRUPT)
if (toInstall.isPresent()) {
IntentData installData = toInstall.get();
- log.warn("Failed installation: {} {} on {}",
+ log.warn("Failed installation: {} {} due to {}",
installData.key(), installData.intent(), ctx.error());
installData.setState(CORRUPT);
installData.incrementErrorCount();
@@ -129,7 +129,7 @@
// if toUninstall was cause of error, then CORRUPT (another job will clean this up)
if (toUninstall.isPresent()) {
IntentData uninstallData = toUninstall.get();
- log.warn("Failed withdrawal: {} {} on {}",
+ log.warn("Failed withdrawal: {} {} due to {}",
uninstallData.key(), uninstallData.intent(), ctx.error());
uninstallData.setState(CORRUPT);
uninstallData.incrementErrorCount();
@@ -355,6 +355,7 @@
private class FlowObjectiveInstallationContext implements ObjectiveContext {
Objective objective;
DeviceId deviceId;
+ ObjectiveError error;
void setObjective(Objective objective, DeviceId deviceId) {
this.objective = objective;
@@ -368,6 +369,7 @@
@Override
public void onError(Objective objective, ObjectiveError error) {
+ this.error = error;
errorContexts.add(this);
finish();
}
@@ -387,7 +389,7 @@
@Override
public String toString() {
- return String.format("(%s, %s)", deviceId, objective);
+ return String.format("(%s on %s for %s)", error, deviceId, objective);
}
}
}
diff --git a/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java b/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java
index b5a57cd..263e93a 100644
--- a/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java
+++ b/core/store/dist/src/main/java/org/onosproject/store/flowobjective/impl/DistributedFlowObjectiveStore.java
@@ -97,7 +97,6 @@
@Override
public void putNextGroup(Integer nextId, NextGroup group) {
nextGroups.put(nextId, group.data());
- notifyDelegate(new ObjectiveEvent(ObjectiveEvent.Type.ADD, nextId));
}
@Override
@@ -113,7 +112,6 @@
public NextGroup removeNextGroup(Integer nextId) {
Versioned<byte[]> versionGroup = nextGroups.remove(nextId);
if (versionGroup != null) {
- notifyDelegate(new ObjectiveEvent(ObjectiveEvent.Type.REMOVE, nextId));
return new DefaultNextGroup(versionGroup.value());
}
return null;