X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=opendaylight%2Fforwardingrulesmanager%2Fimplementation%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fforwardingrulesmanager%2Finternal%2FForwardingRulesManager.java;h=3c936de769d6a22f426b01bab868593eefb899e2;hb=223620094a190aa5f747078d7d84d05a064e4eb9;hp=c839d15f10c81308909a03f804e5587ec6ef612c;hpb=19a34ba9057d07bfbcfb491786ffeb91f5b2c318;p=controller.git diff --git a/opendaylight/forwardingrulesmanager/implementation/src/main/java/org/opendaylight/controller/forwardingrulesmanager/internal/ForwardingRulesManager.java b/opendaylight/forwardingrulesmanager/implementation/src/main/java/org/opendaylight/controller/forwardingrulesmanager/internal/ForwardingRulesManager.java index c839d15f10..3c936de769 100644 --- a/opendaylight/forwardingrulesmanager/implementation/src/main/java/org/opendaylight/controller/forwardingrulesmanager/internal/ForwardingRulesManager.java +++ b/opendaylight/forwardingrulesmanager/implementation/src/main/java/org/opendaylight/controller/forwardingrulesmanager/internal/ForwardingRulesManager.java @@ -39,7 +39,7 @@ import org.opendaylight.controller.clustering.services.ICacheUpdateAware; import org.opendaylight.controller.clustering.services.IClusterContainerServices; import org.opendaylight.controller.clustering.services.IClusterServices; import org.opendaylight.controller.configuration.IConfigurationContainerAware; -import org.opendaylight.controller.connectionmanager.ConnectionLocality; +import org.opendaylight.controller.sal.connection.ConnectionLocality; import org.opendaylight.controller.connectionmanager.IConnectionManager; import org.opendaylight.controller.forwardingrulesmanager.FlowConfig; import org.opendaylight.controller.forwardingrulesmanager.FlowEntry; @@ -57,6 +57,7 @@ import org.opendaylight.controller.sal.action.Controller; import org.opendaylight.controller.sal.action.Flood; import org.opendaylight.controller.sal.action.Output; import org.opendaylight.controller.sal.action.PopVlan; +import org.opendaylight.controller.sal.core.Config; import org.opendaylight.controller.sal.core.ContainerFlow; import org.opendaylight.controller.sal.core.IContainer; import org.opendaylight.controller.sal.core.IContainerListener; @@ -102,7 +103,7 @@ public class ForwardingRulesManager implements IConfigurationContainerAware, IInventoryListener, IObjectReader, - ICacheUpdateAware, + ICacheUpdateAware, CommandProvider, IFlowProgrammerListener { private static final String NODEDOWN = "Node is Down"; @@ -213,7 +214,8 @@ public class ForwardingRulesManager implements * @return a Future object for monitoring the progress of the result, or * null in case the processing should take place locally */ - private Future distributeWorkOrder(FlowEntryInstall e, FlowEntryInstall u, UpdateType t) { + private FlowEntryDistributionOrderFutureTask distributeWorkOrder(FlowEntryInstall e, FlowEntryInstall u, + UpdateType t) { // A null entry it's an unexpected condition, anyway it's safe to keep // the handling local if (e == null) { @@ -543,11 +545,17 @@ public class ForwardingRulesManager implements * contain the unique id assigned to this request */ private Status modifyEntryInternal(FlowEntryInstall currentEntries, FlowEntryInstall newEntries, boolean async) { - Future futureStatus = distributeWorkOrder(currentEntries, newEntries, UpdateType.CHANGED); + FlowEntryDistributionOrderFutureTask futureStatus = + distributeWorkOrder(currentEntries, newEntries, UpdateType.CHANGED); if (futureStatus != null) { Status retStatus = new Status(StatusCode.UNDEFINED); try { retStatus = futureStatus.get(); + if (retStatus.getCode() + .equals(StatusCode.TIMEOUT)) { + // A timeout happened, lets cleanup the workMonitor + workMonitor.remove(futureStatus.getOrder()); + } } catch (InterruptedException e) { log.error("", e); } catch (ExecutionException e) { @@ -655,11 +663,16 @@ public class ForwardingRulesManager implements * contain the unique id assigned to this request */ private Status removeEntryInternal(FlowEntryInstall entry, boolean async) { - Future futureStatus = distributeWorkOrder(entry, null, UpdateType.REMOVED); + FlowEntryDistributionOrderFutureTask futureStatus = distributeWorkOrder(entry, null, UpdateType.REMOVED); if (futureStatus != null) { Status retStatus = new Status(StatusCode.UNDEFINED); try { retStatus = futureStatus.get(); + if (retStatus.getCode() + .equals(StatusCode.TIMEOUT)) { + // A timeout happened, lets cleanup the workMonitor + workMonitor.remove(futureStatus.getOrder()); + } } catch (InterruptedException e) { log.error("", e); } catch (ExecutionException e) { @@ -703,11 +716,16 @@ public class ForwardingRulesManager implements * contain the unique id assigned to this request */ private Status addEntriesInternal(FlowEntryInstall entry, boolean async) { - Future futureStatus = distributeWorkOrder(entry, null, UpdateType.ADDED); + FlowEntryDistributionOrderFutureTask futureStatus = distributeWorkOrder(entry, null, UpdateType.ADDED); if (futureStatus != null) { Status retStatus = new Status(StatusCode.UNDEFINED); try { retStatus = futureStatus.get(); + if (retStatus.getCode() + .equals(StatusCode.TIMEOUT)) { + // A timeout happened, lets cleanup the workMonitor + workMonitor.remove(futureStatus.getOrder()); + } } catch (InterruptedException e) { log.error("", e); } catch (ExecutionException e) { @@ -1312,7 +1330,6 @@ public class ForwardingRulesManager implements retrieveCaches(); } - @SuppressWarnings("deprecation") private void allocateCaches() { if (this.clusterContainerService == null) { log.warn("Un-initialized clusterContainerService, can't create cache"); @@ -1368,7 +1385,7 @@ public class ForwardingRulesManager implements } } - @SuppressWarnings({ "unchecked", "deprecation" }) + @SuppressWarnings({ "unchecked" }) private void retrieveCaches() { ConcurrentMap map; @@ -2174,7 +2191,99 @@ public class ForwardingRulesManager implements @Override public void notifyNodeConnector(NodeConnector nodeConnector, UpdateType type, Map propMap) { + boolean updateStaticFlowCluster = false; + switch (type) { + case ADDED: + break; + case CHANGED: + Config config = (propMap == null) ? null : (Config) propMap.get(Config.ConfigPropName); + if (config != null) { + switch (config.getValue()) { + case Config.ADMIN_DOWN: + log.trace("Port {} is administratively down: uninstalling interested flows", nodeConnector); + updateStaticFlowCluster = removeFlowsOnNodeConnectorDown(nodeConnector); + break; + case Config.ADMIN_UP: + log.trace("Port {} is administratively up: installing interested flows", nodeConnector); + updateStaticFlowCluster = installFlowsOnNodeConnectorUp(nodeConnector); + break; + case Config.ADMIN_UNDEF: + break; + default: + } + } + break; + case REMOVED: + // This is the case where a switch port is removed from the SDN agent space + log.trace("Port {} was removed from our control: uninstalling interested flows", nodeConnector); + updateStaticFlowCluster = removeFlowsOnNodeConnectorDown(nodeConnector); + break; + default: + + } + + if (updateStaticFlowCluster) { + refreshClusterStaticFlowsStatus(nodeConnector.getNode()); + } + } + + /* + * It goes through the static flows configuration, it identifies the ones + * which have the specified node connector as input or output port and + * install them on the network node if they are marked to be installed in + * hardware and their status shows they were not installed yet + */ + private boolean installFlowsOnNodeConnectorUp(NodeConnector nodeConnector) { + boolean updated = false; + List flowConfigForNode = getStaticFlows(nodeConnector.getNode()); + for (FlowConfig flowConfig : flowConfigForNode) { + if (doesFlowContainNodeConnector(flowConfig.getFlow(), nodeConnector)) { + if (flowConfig.installInHw() && !flowConfig.getStatus().equals(SUCCESS)) { + Status status = this.installFlowEntry(flowConfig.getFlowEntry()); + if (!status.isSuccess()) { + flowConfig.setStatus(status.getDescription()); + } else { + flowConfig.setStatus(SUCCESS); + } + updated = true; + } + } + } + return updated; + } + + /* + * Remove from the network node all the flows which have the specified node + * connector as input or output port. If any of the flow entry is a static + * flow, it updates the correspondent configuration. + */ + private boolean removeFlowsOnNodeConnectorDown(NodeConnector nodeConnector) { + boolean updated = false; + List nodeFlowEntries = nodeFlows.get(nodeConnector.getNode()); + if (nodeFlowEntries == null) { + return updated; + } + for (FlowEntryInstall fei : new ArrayList(nodeFlowEntries)) { + if (doesFlowContainNodeConnector(fei.getInstall().getFlow(), nodeConnector)) { + Status status = this.removeEntryInternal(fei, true); + if (!status.isSuccess()) { + continue; + } + /* + * If the flow entry is a static flow, then update its + * configuration + */ + if (fei.getGroupName().equals(FlowConfig.STATICFLOWGROUP)) { + FlowConfig flowConfig = getStaticFlow(fei.getFlowName(), fei.getNode()); + if (flowConfig != null) { + flowConfig.setStatus(PORTREMOVED); + updated = true; + } + } + } + } + return updated; } private FlowConfig getDerivedFlowConfig(FlowConfig original, String configName, Short port) { @@ -2462,6 +2571,14 @@ public class ForwardingRulesManager implements } else { log.warn("Not expected null WorkStatus", work); } + } else if (event instanceof ContainerFlowChangeEvent) { + /* + * Whether it is an addition or removal, we have to + * recompute the merged flows entries taking into + * account all the current container flows because + * flow merging is not an injective function + */ + updateFlowsContainerFlow(); } else { log.warn("Dequeued unknown event {}", event.getClass() .getSimpleName()); @@ -2560,12 +2677,8 @@ public class ForwardingRulesManager implements } log.trace("Container {}: Updating installed flows because of container flow change: {} {}", container.getName(), t, current); - /* - * Whether it is an addition or removal, we have to recompute the merged - * flows entries taking into account all the current container flows - * because flow merging is not an injective function - */ - updateFlowsContainerFlow(); + ContainerFlowChangeEvent ev = new ContainerFlowChangeEvent(previous, current, t); + pendingEvents.offer(ev); } @Override @@ -2578,57 +2691,22 @@ public class ForwardingRulesManager implements switch (t) { case REMOVED: - - List nodeFlowEntries = nodeFlows.get(nc.getNode()); - if (nodeFlowEntries == null) { - return; - } - for (FlowEntryInstall fei : new ArrayList(nodeFlowEntries)) { - if (doesFlowContainNodeConnector(fei.getInstall().getFlow(), nc)) { - Status status = this.removeEntryInternal(fei, true); - if (!status.isSuccess()) { - continue; - } - /* - * If the flow entry is a static flow, then update its - * configuration - */ - if (fei.getGroupName().equals(FlowConfig.STATICFLOWGROUP)) { - FlowConfig flowConfig = getStaticFlow(fei.getFlowName(), fei.getNode()); - if (flowConfig != null) { - flowConfig.setStatus(PORTREMOVED); - updateStaticFlowCluster = true; - } - } - } - } - if (updateStaticFlowCluster) { - refreshClusterStaticFlowsStatus(nc.getNode()); - } + log.trace("Port {} was removed from container: uninstalling interested flows", nc); + updateStaticFlowCluster = removeFlowsOnNodeConnectorDown(nc); break; case ADDED: - List flowConfigForNode = getStaticFlows(nc.getNode()); - for (FlowConfig flowConfig : flowConfigForNode) { - if (doesFlowContainNodeConnector(flowConfig.getFlow(), nc)) { - if (flowConfig.installInHw()) { - Status status = this.installFlowEntry(flowConfig.getFlowEntry()); - if (!status.isSuccess()) { - flowConfig.setStatus(status.getDescription()); - } else { - flowConfig.setStatus(SUCCESS); - } - updateStaticFlowCluster = true; - } - } - } - if (updateStaticFlowCluster) { - refreshClusterStaticFlowsStatus(nc.getNode()); - } + log.trace("Port {} was added to container: reinstall interested flows", nc); + updateStaticFlowCluster = installFlowsOnNodeConnectorUp(nc); + break; case CHANGED: break; default: } + + if (updateStaticFlowCluster) { + refreshClusterStaticFlowsStatus(nc.getNode()); + } } @Override @@ -2733,6 +2811,30 @@ public class ForwardingRulesManager implements return newEntry; } } + private class ContainerFlowChangeEvent extends FRMEvent { + private final ContainerFlow previous; + private final ContainerFlow current; + private final UpdateType type; + + public ContainerFlowChangeEvent(ContainerFlow previous, ContainerFlow current, UpdateType type) { + this.previous = previous; + this.current = current; + this.type = type; + } + + public ContainerFlow getPrevious() { + return this.previous; + } + + public ContainerFlow getCurrent() { + return this.current; + } + + public UpdateType getType() { + return this.type; + } + } + private class WorkStatusCleanup extends FRMEvent { private FlowEntryDistributionOrder fe; @@ -2982,11 +3084,18 @@ public class ForwardingRulesManager implements * mapping will have to be added in future */ FlowEntryInstall target = null; - for (FlowEntryInstall index : nodeFlows.get(node)) { - FlowEntryInstall entry = installedSwView.get(index); - if (entry.getRequestId() == rid) { - target = entry; - break; + List flowEntryInstallList = nodeFlows.get(node); + // flowEntryInstallList could be null. + // so check for it. + if(flowEntryInstallList != null) { + for (FlowEntryInstall index : flowEntryInstallList) { + FlowEntryInstall entry = installedSwView.get(index); + if(entry != null) { + if (entry.getRequestId() == rid) { + target = entry; + break; + } + } } } if (target != null) { @@ -3079,7 +3188,7 @@ public class ForwardingRulesManager implements */ if (fe.getRequestorController() .equals(clusterContainerService.getMyAddress())) { - FlowEntryDistributionOrderFutureTask fet = workMonitor.get(fe); + FlowEntryDistributionOrderFutureTask fet = workMonitor.remove(fe); if (fet != null) { logsync.trace("workStatus response is for us {}", fe); // Signal we got the status