X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fforwardingrulesmanager%2Fimplementation%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fforwardingrulesmanager%2Finternal%2FForwardingRulesManager.java;h=58d23655cac5a17353989c5cfd36d71a7a14ea74;hp=be139317dc9eb3aed2fa1066985cdc4eb152b060;hb=e383b2e0b9072b7b5b3e3903b8a42ab3917b7f63;hpb=6721c99eaa6c9d83d7469f243ec383046de467e6 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 be139317dc..58d23655ca 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 @@ -22,9 +22,12 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Callable; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; @@ -198,6 +201,59 @@ public class ForwardingRulesManager implements private ConcurrentMap workMonitor = new ConcurrentHashMap(); + /* + * Create an executor pool to create the distributionOrder, this is a stop + * gap solution caused by an issue with non-transactional caches in the + * implementation we use, being currently worked on. It has been noticed in + * fact that when non-transactional caches are being used sometime the key + * are no distributed to all the nodes properly. To workaround the issue + * transactional caches are being used, but there was a reason for using + * non-transactional caches to start with, in fact we needed to be able in + * the context of a northbound transaction to program the FRM entries + * irrespective of the fact that transaction would commit or no else we + * would not be able to achieve the entry programming and implement the + * scheme for recovery from network element failures. Bottom line, now in + * order to make sure an update on a transactional cache goes out while in a + * transaction that need to be initiated by a different thread. + */ + private ExecutorService executor; + + class DistributeOrderCallable implements Callable> { + private FlowEntryInstall e; + private FlowEntryInstall u; + private UpdateType t; + DistributeOrderCallable(FlowEntryInstall e, FlowEntryInstall u, UpdateType t) { + this.e = e; + this.u = u; + this.t = t; + } + + @Override + public Future call() throws Exception { + if (e == null || t == null) { + logsync.error("Unexpected null Entry up update type"); + return null; + } + // Create the work order and distribute it + FlowEntryDistributionOrder fe = + new FlowEntryDistributionOrder(e, t, clusterContainerService.getMyAddress()); + // First create the monitor job + FlowEntryDistributionOrderFutureTask ret = new FlowEntryDistributionOrderFutureTask(fe); + logsync.trace("Node {} not local so sending fe {}", e.getNode(), fe); + workMonitor.put(fe, ret); + if (t.equals(UpdateType.CHANGED)) { + // Then distribute the work + workOrder.put(fe, u); + } else { + // Then distribute the work + workOrder.put(fe, e); + } + logsync.trace("WorkOrder requested"); + // Now create an Handle to monitor the execution of the operation + return ret; + } + } + /** * @param e * Entry being installed/updated/removed @@ -218,27 +274,24 @@ public class ForwardingRulesManager implements Node n = e.getNode(); if (!connectionManager.isLocal(n)) { - // Create the work order and distribute it - FlowEntryDistributionOrder fe = - new FlowEntryDistributionOrder(e, t, clusterContainerService.getMyAddress()); - // First create the monitor job - FlowEntryDistributionOrderFutureTask ret = new FlowEntryDistributionOrderFutureTask(fe); - logsync.trace("Node {} not local so sending fe {}", n, fe); - workMonitor.put(fe, ret); - if (t.equals(UpdateType.CHANGED)) { - // Then distribute the work - workOrder.put(fe, u); - } else { - // Then distribute the work - workOrder.put(fe, e); + Callable> worker = new DistributeOrderCallable(e, u, t); + if (worker != null) { + Future> workerRes = this.executor.submit(worker); + try { + return workerRes.get(); + } catch (InterruptedException e1) { + // we where interrupted, not a big deal. + return null; + } catch (ExecutionException e1) { + logsync.error( + "We got an execution exception {} we cannot much, so returning we don't have nothing to wait for", + e); + return null; + } } - logsync.trace("WorkOrder requested"); - // Now create an Handle to monitor the execution of the operation - return ret; } logsync.trace("LOCAL Node {} so processing Entry:{} UpdateType:{}", n, e, t); - return null; } @@ -1353,10 +1406,10 @@ public class ForwardingRulesManager implements EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL)); clusterContainerService.createCache(WORKSTATUSCACHE, - EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL)); + EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL)); clusterContainerService.createCache(WORKORDERCACHE, - EnumSet.of(IClusterServices.cacheMode.NON_TRANSACTIONAL)); + EnumSet.of(IClusterServices.cacheMode.TRANSACTIONAL)); } catch (CacheConfigException cce) { log.error("CacheConfigException"); @@ -2043,51 +2096,78 @@ public class ForwardingRulesManager implements addStaticFlowInternal(allowARP, true); // skip validation on internal static flow name } + /** + * (non-Javadoc) + * + * @see org.opendaylight.controller.switchmanager.ISwitchManagerAware#modeChangeNotify(org.opendaylight.controller.sal.core.Node, + * boolean) + * + * This method can be called from within the OSGi framework context, + * given the programming operation can take sometime, it not good + * pratice to have in it's context operations that can take time, + * hence moving off to a different thread for async processing. + */ @Override - public void modeChangeNotify(Node node, boolean proactive) { - List defaultConfigs = new ArrayList(); - - List puntAction = new ArrayList(); - puntAction.add(ActionType.CONTROLLER.toString()); - - FlowConfig allowARP = new FlowConfig(); - allowARP.setInstallInHw(true); - allowARP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt ARP" + FlowConfig.INTERNALSTATICFLOWEND); - allowARP.setPriority("1"); - allowARP.setNode(node); - allowARP.setEtherType("0x" + Integer.toHexString(EtherTypes.ARP.intValue()).toUpperCase()); - allowARP.setActions(puntAction); - defaultConfigs.add(allowARP); - - FlowConfig allowLLDP = new FlowConfig(); - allowLLDP.setInstallInHw(true); - allowLLDP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt LLDP" + FlowConfig.INTERNALSTATICFLOWEND); - allowLLDP.setPriority("1"); - allowLLDP.setNode(node); - allowLLDP.setEtherType("0x" + Integer.toHexString(EtherTypes.LLDP.intValue()).toUpperCase()); - allowLLDP.setActions(puntAction); - defaultConfigs.add(allowLLDP); - - List dropAction = new ArrayList(); - dropAction.add(ActionType.DROP.toString()); - - FlowConfig dropAllConfig = new FlowConfig(); - dropAllConfig.setInstallInHw(true); - dropAllConfig.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Catch-All Drop" + FlowConfig.INTERNALSTATICFLOWEND); - dropAllConfig.setPriority("0"); - dropAllConfig.setNode(node); - dropAllConfig.setActions(dropAction); - defaultConfigs.add(dropAllConfig); - - log.info("Forwarding mode for node {} set to {}", node, (proactive ? "proactive" : "reactive")); - for (FlowConfig fc : defaultConfigs) { - Status status = (proactive) ? addStaticFlowInternal(fc, false) : removeStaticFlow(fc); - if (status.isSuccess()) { - log.info("{} Proactive Static flow: {}", (proactive ? "Installed" : "Removed"), fc.getName()); - } else { - log.warn("Failed to {} Proactive Static flow: {}", (proactive ? "install" : "remove"), fc.getName()); + public void modeChangeNotify(final Node node, final boolean proactive) { + Callable modeChangeCallable = new Callable() { + @Override + public Status call() throws Exception { + List defaultConfigs = new ArrayList(); + + List puntAction = new ArrayList(); + puntAction.add(ActionType.CONTROLLER.toString()); + + FlowConfig allowARP = new FlowConfig(); + allowARP.setInstallInHw(true); + allowARP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt ARP" + FlowConfig.INTERNALSTATICFLOWEND); + allowARP.setPriority("1"); + allowARP.setNode(node); + allowARP.setEtherType("0x" + Integer.toHexString(EtherTypes.ARP.intValue()) + .toUpperCase()); + allowARP.setActions(puntAction); + defaultConfigs.add(allowARP); + + FlowConfig allowLLDP = new FlowConfig(); + allowLLDP.setInstallInHw(true); + allowLLDP.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Punt LLDP" + FlowConfig.INTERNALSTATICFLOWEND); + allowLLDP.setPriority("1"); + allowLLDP.setNode(node); + allowLLDP.setEtherType("0x" + Integer.toHexString(EtherTypes.LLDP.intValue()) + .toUpperCase()); + allowLLDP.setActions(puntAction); + defaultConfigs.add(allowLLDP); + + List dropAction = new ArrayList(); + dropAction.add(ActionType.DROP.toString()); + + FlowConfig dropAllConfig = new FlowConfig(); + dropAllConfig.setInstallInHw(true); + dropAllConfig.setName(FlowConfig.INTERNALSTATICFLOWBEGIN + "Catch-All Drop" + + FlowConfig.INTERNALSTATICFLOWEND); + dropAllConfig.setPriority("0"); + dropAllConfig.setNode(node); + dropAllConfig.setActions(dropAction); + defaultConfigs.add(dropAllConfig); + + log.info("Forwarding mode for node {} set to {}", node, (proactive ? "proactive" : "reactive")); + for (FlowConfig fc : defaultConfigs) { + Status status = (proactive) ? addStaticFlowInternal(fc, false) : removeStaticFlow(fc); + if (status.isSuccess()) { + log.info("{} Proactive Static flow: {}", (proactive ? "Installed" : "Removed"), fc.getName()); + } else { + log.warn("Failed to {} Proactive Static flow: {}", (proactive ? "install" : "remove"), + fc.getName()); + } + } + return new Status(StatusCode.SUCCESS); } - } + }; + + /* + * Execute the work outside the caller context, this could be an + * expensive operation and we don't want to block the caller for it. + */ + this.executor.submit(modeChangeCallable); } /** @@ -2464,6 +2544,9 @@ public class ForwardingRulesManager implements // Initialize graceful stop flag stopping = false; + // Allocate the executor service + this.executor = Executors.newSingleThreadExecutor(); + // Start event handler thread frmEventHandler.start(); @@ -2484,6 +2567,8 @@ public class ForwardingRulesManager implements void stop() { stopping = true; uninstallAllFlowEntries(false); + // Shutdown executor + this.executor.shutdownNow(); } public void setFlowProgrammerService(IFlowProgrammerService service) {