resilient flows and application id

Change-Id: Ic9f192d4451ae962737ab2b45c644372535e7bdb

apps/fwd/src/main/java/org/onlab/onos/fwd/ReactiveForwarding.java

@@ -9,6 +9,7 @@ import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.Host;
 import org.onlab.onos.net.HostId;
 import org.onlab.onos.net.Path;
@@ -53,14 +54,18 @@ public class ReactiveForwarding {
 
     private ReactivePacketProcessor processor = new ReactivePacketProcessor();
 
+    private ApplicationId appId;
+
     @Activate
     public void activate() {
+        appId = ApplicationId.getAppId();
         packetService.addProcessor(processor, PacketProcessor.ADVISOR_MAX + 1);
-        log.info("Started");
+        log.info("Started with Application ID {}", appId.id());
     }
 
     @Deactivate
     public void deactivate() {
+        flowRuleService.removeFlowRulesById(appId);
         packetService.removeProcessor(processor);
         processor = null;
         log.info("Stopped");
@@ -169,7 +174,7 @@ public class ReactiveForwarding {
         treat.add(Instructions.createOutput(portNumber));
 
         FlowRule f = new DefaultFlowRule(context.inPacket().receivedFrom().deviceId(),
-                builder.build(), treat.build(), 0);
+                builder.build(), treat.build(), 0, appId);
 
         flowRuleService.applyFlowRules(f);
     }

core/api/src/main/java/org/onlab/onos/net/flow/DefaultFlowRule.java

@@ -5,6 +5,7 @@ import static org.slf4j.LoggerFactory.getLogger;
 
 import java.util.Objects;
 
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
 import org.slf4j.Logger;
 
@@ -24,6 +25,8 @@ public class DefaultFlowRule implements FlowRule {
 
     private final FlowId id;
 
+    private final ApplicationId appId;
+
     public DefaultFlowRule(DeviceId deviceId, TrafficSelector selector,
             TrafficTreatment treatment, int priority, FlowRuleState state,
             long life, long packets, long bytes, long flowId) {
@@ -32,7 +35,7 @@ public class DefaultFlowRule implements FlowRule {
         this.selector = selector;
         this.treatment = treatment;
         this.state = state;
-
+        this.appId = ApplicationId.valueOf((int) (flowId >> 32));
         this.id = FlowId.valueOf(flowId);
 
         this.life = life;
@@ -42,18 +45,18 @@ public class DefaultFlowRule implements FlowRule {
     }
 
     public DefaultFlowRule(DeviceId deviceId, TrafficSelector selector,
-            TrafficTreatment treatement, int priority) {
+            TrafficTreatment treatement, int priority, ApplicationId appId) {
-        this(deviceId, selector, treatement, priority, FlowRuleState.CREATED);
+        this(deviceId, selector, treatement, priority, FlowRuleState.CREATED, appId);
     }
 
     public DefaultFlowRule(FlowRule rule, FlowRuleState state) {
         this(rule.deviceId(), rule.selector(), rule.treatment(),
-                rule.priority(), state, rule.id());
+                rule.priority(), state, rule.id(), rule.appId());
     }
 
     private DefaultFlowRule(DeviceId deviceId,
             TrafficSelector selector, TrafficTreatment treatment,
-            int priority, FlowRuleState state) {
+            int priority, FlowRuleState state, ApplicationId appId) {
         this.deviceId = deviceId;
         this.priority = priority;
         this.selector = selector;
@@ -62,13 +65,15 @@ public class DefaultFlowRule implements FlowRule {
         this.life = 0;
         this.packets = 0;
         this.bytes = 0;
-        this.id = FlowId.valueOf(this.hashCode());
+        this.appId = appId;
+
+        this.id = FlowId.valueOf((((long) appId().id()) << 32) | (this.hash() & 0xffffffffL));
         this.created = System.currentTimeMillis();
     }
 
     private DefaultFlowRule(DeviceId deviceId,
             TrafficSelector selector, TrafficTreatment treatment,
-            int priority, FlowRuleState state, FlowId flowId) {
+            int priority, FlowRuleState state, FlowId flowId, ApplicationId appId) {
         this.deviceId = deviceId;
         this.priority = priority;
         this.selector = selector;
@@ -77,6 +82,7 @@ public class DefaultFlowRule implements FlowRule {
         this.life = 0;
         this.packets = 0;
         this.bytes = 0;
+        this.appId = appId;
         this.id = flowId;
         this.created = System.currentTimeMillis();
     }
@@ -87,6 +93,11 @@ public class DefaultFlowRule implements FlowRule {
         return id;
     }
 
+    @Override
+    public ApplicationId appId() {
+        return appId;
+    }
+
     @Override
     public int priority() {
         return priority;
@@ -136,7 +147,11 @@ public class DefaultFlowRule implements FlowRule {
      * @see java.lang.Object#equals(java.lang.Object)
      */
     public int hashCode() {
-        return Objects.hash(deviceId, selector, treatment);
+        return Objects.hash(deviceId, id);
+    }
+
+    public int hash() {
+        return Objects.hash(deviceId, selector, id);
     }
 
     @Override

core/api/src/main/java/org/onlab/onos/net/flow/FlowRule.java

@@ -1,5 +1,6 @@
 package org.onlab.onos.net.flow;
 
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
 
 /**
@@ -52,6 +53,13 @@ public interface FlowRule {
      */
     FlowId id();
 
+    /**
+     * Returns the application id of this flow.
+     *
+     * @return an applicationId
+     */
+    ApplicationId appId();
+
     /**
      * Returns the flow rule priority given in natural order; higher numbers
      * mean higher priorities.

core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleProvider.java

@@ -1,5 +1,6 @@
 package org.onlab.onos.net.flow;
 
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.provider.Provider;
 
 /**
@@ -25,4 +26,10 @@ public interface FlowRuleProvider extends Provider {
      */
     void removeFlowRule(FlowRule... flowRules);
 
+    /**
+     * Removes rules by their id.
+     * @param id the id to remove
+     */
+    void removeRulesById(ApplicationId id, FlowRule... flowRules);
+
 }

core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleService.java

@@ -1,5 +1,6 @@
 package org.onlab.onos.net.flow;
 
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
 
 /**
@@ -43,6 +44,20 @@ public interface FlowRuleService {
      */
     void removeFlowRules(FlowRule... flowRules);
 
+    /**
+     * Removes all rules by id.
+     *
+     * @param appId id to remove
+     */
+    void removeFlowRulesById(ApplicationId appId);
+
+    /**
+     * Returns a list of rules with this application id.
+     *
+     * @param id the id to look up
+     * @return collection of flow rules
+     */
+    Iterable<FlowRule> getFlowRulesById(ApplicationId id);
 
     /**
      * Adds the specified flow rule listener.
@@ -58,4 +73,6 @@ public interface FlowRuleService {
      */
     void removeListener(FlowRuleListener listener);
 
+
+
 }

core/api/src/main/java/org/onlab/onos/net/flow/FlowRuleStore.java

@@ -1,5 +1,6 @@
 package org.onlab.onos.net.flow;
 
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
 
 /**
@@ -7,6 +8,13 @@ import org.onlab.onos.net.DeviceId;
  */
 public interface FlowRuleStore {
 
+    /**
+     * Returns the stored flow.
+     * @param rule the rule to look for
+     * @return a flow rule
+     */
+    FlowRule getFlowRule(FlowRule rule);
+
     /**
      * Returns the flow entries associated with a device.
      *
@@ -15,6 +23,14 @@ public interface FlowRuleStore {
      */
     Iterable<FlowRule> getFlowEntries(DeviceId deviceId);
 
+    /**
+     * Returns the flow entries associated with an application.
+     *
+     * @param appId the application id
+     * @return the flow entries
+     */
+    Iterable<FlowRule> getFlowEntriesByAppId(ApplicationId appId);
+
     /**
      * Stores a new flow rule without generating events.
      *

core/net/src/main/java/org/onlab/onos/net/flow/impl/FlowRuleManager.java

@@ -12,14 +12,13 @@ import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.event.AbstractListenerRegistry;
 import org.onlab.onos.event.EventDeliveryService;
 import org.onlab.onos.net.Device;
 import org.onlab.onos.net.DeviceId;
 import org.onlab.onos.net.device.DeviceService;
-import org.onlab.onos.net.flow.DefaultFlowRule;
 import org.onlab.onos.net.flow.FlowRule;
-import org.onlab.onos.net.flow.FlowRule.FlowRuleState;
 import org.onlab.onos.net.flow.FlowRuleEvent;
 import org.onlab.onos.net.flow.FlowRuleListener;
 import org.onlab.onos.net.flow.FlowRuleProvider;
@@ -77,7 +76,7 @@ implements FlowRuleService, FlowRuleProviderRegistry {
     @Override
     public void applyFlowRules(FlowRule... flowRules) {
         for (int i = 0; i < flowRules.length; i++) {
-            FlowRule f = new DefaultFlowRule(flowRules[i], FlowRuleState.PENDING_ADD);
+            FlowRule f = flowRules[i];
             final Device device = deviceService.getDevice(f.deviceId());
             final FlowRuleProvider frp = getProvider(device.providerId());
             store.storeFlowRule(f);
@@ -88,14 +87,33 @@ implements FlowRuleService, FlowRuleProviderRegistry {
     @Override
     public void removeFlowRules(FlowRule... flowRules) {
         FlowRule f;
+        FlowRuleProvider frp;
+        Device device;
         for (int i = 0; i < flowRules.length; i++) {
-            f = new DefaultFlowRule(flowRules[i], FlowRuleState.PENDING_REMOVE);
+            f = flowRules[i];
-            final Device device = deviceService.getDevice(f.deviceId());
+            device = deviceService.getDevice(f.deviceId());
-            final FlowRuleProvider frp = getProvider(device.providerId());
+            frp = getProvider(device.providerId());
             store.deleteFlowRule(f);
             frp.removeFlowRule(f);
         }
+    }
 
+    @Override
+    public void removeFlowRulesById(ApplicationId id) {
+        Iterable<FlowRule> rules =  getFlowRulesById(id);
+        FlowRuleProvider frp;
+        Device device;
+        for (FlowRule f : rules) {
+            store.deleteFlowRule(f);
+            device = deviceService.getDevice(f.deviceId());
+            frp = getProvider(device.providerId());
+            frp.removeRulesById(id, f);
+        }
+    }
+
+    @Override
+    public Iterable<FlowRule> getFlowRulesById(ApplicationId id) {
+        return store.getFlowEntriesByAppId(id);
     }
 
     @Override
@@ -126,8 +144,27 @@ implements FlowRuleService, FlowRuleProviderRegistry {
         public void flowRemoved(FlowRule flowRule) {
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkValidity();
-            FlowRuleEvent event = store.removeFlowRule(flowRule);
+            FlowRule stored = store.getFlowRule(flowRule);
+            if (stored == null) {
+                log.debug("Rule already evicted from store: {}", flowRule);
+                return;
+            }
+            Device device = deviceService.getDevice(flowRule.deviceId());
+            FlowRuleProvider frp = getProvider(device.providerId());
+            FlowRuleEvent event = null;
+            switch (stored.state()) {
+            case ADDED:
+            case PENDING_ADD:
+                frp.applyFlowRule(flowRule);
+                break;
+            case PENDING_REMOVE:
+            case REMOVED:
+                event = store.removeFlowRule(flowRule);
+                break;
+            default:
+                break;
 
+            }
             if (event != null) {
                 log.debug("Flow {} removed", flowRule);
                 post(event);
@@ -138,7 +175,22 @@ implements FlowRuleService, FlowRuleProviderRegistry {
         public void flowMissing(FlowRule flowRule) {
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkValidity();
-            log.debug("Flow {} has not been installed.", flowRule);
+            Device device = deviceService.getDevice(flowRule.deviceId());
+            FlowRuleProvider frp = getProvider(device.providerId());
+            switch (flowRule.state()) {
+            case PENDING_REMOVE:
+            case REMOVED:
+                store.removeFlowRule(flowRule);
+                frp.removeFlowRule(flowRule);
+                break;
+            case ADDED:
+            case PENDING_ADD:
+                frp.applyFlowRule(flowRule);
+                break;
+            default:
+                log.debug("Flow {} has not been installed.", flowRule);
+            }
+
 
         }
 
@@ -146,6 +198,7 @@ implements FlowRuleService, FlowRuleProviderRegistry {
         public void extraneousFlow(FlowRule flowRule) {
             checkNotNull(flowRule, FLOW_RULE_NULL);
             checkValidity();
+            removeFlowRules(flowRule);
             log.debug("Flow {} is on switch but not in store.", flowRule);
         }
 

core/net/src/test/java/org/onlab/onos/net/flow/impl/FlowRuleManagerTest.java

@@ -14,6 +14,7 @@ import java.util.List;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.event.impl.TestEventDispatcher;
 import org.onlab.onos.net.DefaultDevice;
 import org.onlab.onos.net.Device;
@@ -61,6 +62,7 @@ public class FlowRuleManagerTest {
     protected FlowRuleProviderService providerService;
     protected TestProvider provider;
     protected TestListener listener = new TestListener();
+    private ApplicationId appId;
 
     @Before
     public void setUp() {
@@ -75,6 +77,7 @@ public class FlowRuleManagerTest {
         mgr.addListener(listener);
         provider = new TestProvider(PID);
         providerService = registry.register(provider);
+        appId = ApplicationId.getAppId();
         assertTrue("provider should be registered",
                 registry.getProviders().contains(provider.id()));
     }
@@ -93,7 +96,7 @@ public class FlowRuleManagerTest {
     private FlowRule flowRule(int tsval, int trval) {
         TestSelector ts = new TestSelector(tsval);
         TestTreatment tr = new TestTreatment(trval);
-        return new DefaultFlowRule(DID, ts, tr, 0);
+        return new DefaultFlowRule(DID, ts, tr, 0, appId);
     }
 
     private FlowRule flowRule(FlowRule rule, FlowRuleState state) {
@@ -159,8 +162,8 @@ public class FlowRuleManagerTest {
     public void applyFlowRules() {
 
         FlowRule r1 = flowRule(1, 1);
-        FlowRule r2 = flowRule(1, 2);
+        FlowRule r2 = flowRule(2, 2);
-        FlowRule r3 = flowRule(1, 3);
+        FlowRule r3 = flowRule(3, 3);
 
         assertTrue("store should be empty",
                 Sets.newHashSet(service.getFlowEntries(DID)).isEmpty());
@@ -196,6 +199,7 @@ public class FlowRuleManagerTest {
     @Test
     public void flowRemoved() {
         FlowRule f1 = addFlowRule(1);
+        service.removeFlowRules(f1);
         addFlowRule(2);
         FlowRule rem1 = flowRule(f1, FlowRuleState.REMOVED);
         providerService.flowRemoved(rem1);
@@ -293,6 +297,11 @@ public class FlowRuleManagerTest {
         public void removeFlowRule(FlowRule... flowRules) {
         }
 
+        @Override
+        public void removeRulesById(ApplicationId id, FlowRule... flowRules) {
+        }
+
+
     }
 
     private class TestSelector implements TrafficSelector {

core/trivial/src/main/java/org/onlab/onos/net/trivial/impl/SimpleFlowRuleStore.java

@@ -4,12 +4,18 @@ import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_ADDED;
 import static org.onlab.onos.net.flow.FlowRuleEvent.Type.RULE_REMOVED;
 import static org.slf4j.LoggerFactory.getLogger;
 
+import java.util.Collection;
+import java.util.Collections;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Service;
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
+import org.onlab.onos.net.flow.DefaultFlowRule;
 import org.onlab.onos.net.flow.FlowRule;
+import org.onlab.onos.net.flow.FlowRule.FlowRuleState;
 import org.onlab.onos.net.flow.FlowRuleEvent;
 import org.onlab.onos.net.flow.FlowRuleEvent.Type;
 import org.onlab.onos.net.flow.FlowRuleStore;
@@ -29,7 +35,11 @@ public class SimpleFlowRuleStore implements FlowRuleStore {
     private final Logger log = getLogger(getClass());
 
     // store entries as a pile of rules, no info about device tables
-    private final Multimap<DeviceId, FlowRule> flowEntries = ArrayListMultimap.create();
+    private final Multimap<DeviceId, FlowRule> flowEntries =
+            ArrayListMultimap.<DeviceId, FlowRule>create();
+
+    private final Multimap<ApplicationId, FlowRule> flowEntriesById =
+            ArrayListMultimap.<ApplicationId, FlowRule>create();
 
     @Activate
     public void activate() {
@@ -41,48 +51,76 @@ public class SimpleFlowRuleStore implements FlowRuleStore {
         log.info("Stopped");
     }
 
+
     @Override
-    public Iterable<FlowRule> getFlowEntries(DeviceId deviceId) {
+    public synchronized FlowRule getFlowRule(FlowRule rule) {
-        return ImmutableSet.copyOf(flowEntries.get(deviceId));
+        for (FlowRule f : flowEntries.get(rule.deviceId())) {
+            if (f.equals(rule)) {
+                return f;
+            }
+        }
+        return null;
     }
 
     @Override
-    public void storeFlowRule(FlowRule rule) {
+    public synchronized Iterable<FlowRule> getFlowEntries(DeviceId deviceId) {
-        DeviceId did = rule.deviceId();
+        Collection<FlowRule> rules = flowEntries.get(deviceId);
-        flowEntries.put(did, rule);
+        if (rules == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableSet.copyOf(rules);
     }
 
     @Override
-    public void deleteFlowRule(FlowRule rule) {
+    public synchronized Iterable<FlowRule> getFlowEntriesByAppId(ApplicationId appId) {
-        DeviceId did = rule.deviceId();
+        Collection<FlowRule> rules = flowEntriesById.get(appId);
+        if (rules == null) {
+            return Collections.emptyList();
+        }
+        return ImmutableSet.copyOf(rules);
+    }
+
+    @Override
+    public synchronized void storeFlowRule(FlowRule rule) {
+        FlowRule f = new DefaultFlowRule(rule, FlowRuleState.PENDING_ADD);
+        DeviceId did = f.deviceId();
+        if (!flowEntries.containsEntry(did, f)) {
+            flowEntries.put(did, f);
+            flowEntriesById.put(rule.appId(), f);
+        }
+    }
+
+    @Override
+    public synchronized void deleteFlowRule(FlowRule rule) {
+        FlowRule f = new DefaultFlowRule(rule, FlowRuleState.PENDING_REMOVE);
+        DeviceId did = f.deviceId();
 
         /*
          *  find the rule and mark it for deletion.
          *  Ultimately a flow removed will come remove it.
          */
 
-        if (flowEntries.containsEntry(did, rule)) {
+        if (flowEntries.containsEntry(did, f)) {
-            synchronized (flowEntries) {
+            //synchronized (flowEntries) {
-
+            flowEntries.remove(did, f);
-                flowEntries.remove(did, rule);
+            flowEntries.put(did, f);
-                flowEntries.put(did, rule);
+            flowEntriesById.remove(rule.appId(), rule);
-            }
+            //}
         }
     }
 
     @Override
-    public FlowRuleEvent addOrUpdateFlowRule(FlowRule rule) {
+    public synchronized FlowRuleEvent addOrUpdateFlowRule(FlowRule rule) {
         DeviceId did = rule.deviceId();
 
         // check if this new rule is an update to an existing entry
         if (flowEntries.containsEntry(did, rule)) {
-            synchronized (flowEntries) {
+            //synchronized (flowEntries) {
-                // Multimaps support duplicates so we have to remove our rule
+            // Multimaps support duplicates so we have to remove our rule
-                // and replace it with the current version.
+            // and replace it with the current version.
-
+            flowEntries.remove(did, rule);
-                flowEntries.remove(did, rule);
+            flowEntries.put(did, rule);
-                flowEntries.put(did, rule);
+            //}
-            }
             return new FlowRuleEvent(Type.RULE_UPDATED, rule);
         }
 
@@ -91,16 +129,20 @@ public class SimpleFlowRuleStore implements FlowRuleStore {
     }
 
     @Override
-    public FlowRuleEvent removeFlowRule(FlowRule rule) {
+    public synchronized FlowRuleEvent removeFlowRule(FlowRule rule) {
-        synchronized (this) {
+        //synchronized (this) {
-            if (flowEntries.remove(rule.deviceId(), rule)) {
+        if (flowEntries.remove(rule.deviceId(), rule)) {
-                return new FlowRuleEvent(RULE_REMOVED, rule);
+            return new FlowRuleEvent(RULE_REMOVED, rule);
-            } else {
+        } else {
-                return null;
+            return null;
-            }
         }
+        //}
     }
 
 
 
+
+
+
+
 }

providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/FlowRuleBuilder.java

@@ -183,7 +183,7 @@ public class FlowRuleBuilder {
                 break;
             case ETH_DST:
                 MacAddress dMac = MacAddress.valueOf(match.get(MatchField.ETH_DST).getLong());
-                builder.add(Criteria.matchEthSrc(dMac));
+                builder.add(Criteria.matchEthDst(dMac));
                 break;
             case ETH_TYPE:
                 int ethType = match.get(MatchField.ETH_TYPE).getValue();

providers/openflow/flow/src/main/java/org/onlab/onos/provider/of/flow/impl/OpenFlowRuleProvider.java

@@ -9,6 +9,7 @@ import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
+import org.onlab.onos.ApplicationId;
 import org.onlab.onos.net.DeviceId;
 import org.onlab.onos.net.flow.FlowRule;
 import org.onlab.onos.net.flow.FlowRuleProvider;
@@ -102,12 +103,17 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
 
     }
 
-
     private void removeRule(FlowRule flowRule) {
         OpenFlowSwitch sw = controller.getSwitch(Dpid.dpid(flowRule.deviceId().uri()));
         sw.sendMsg(new FlowModBuilder(flowRule, sw.factory()).buildFlowDel());
     }
 
+    @Override
+    public void removeRulesById(ApplicationId id, FlowRule... flowRules) {
+        // TODO: optimize using the ApplicationId
+        removeFlowRule(flowRules);
+    }
+
 
     //TODO: InternalFlowRuleProvider listening to stats and error and flowremoved.
     // possibly barriers as well. May not be internal at all...
@@ -179,4 +185,6 @@ public class OpenFlowRuleProvider extends AbstractProvider implements FlowRulePr
     }
 
 
+
+
 }