X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fservices%2FSalFlowServiceImpl.java;h=19895116001e2a5a6f206509ea3208f3fec5ab24;hb=6a97ad9e38ec2ad669c7b51acb86c9b3a5cefad1;hp=8d11b8a60ef64f2b1c0c02a89c245b1e8e2e2f19;hpb=6ae68a28f2f853f5ba7837e181edd44d254c3c1d;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java index 8d11b8a60e..1989511600 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/services/SalFlowServiceImpl.java @@ -1,6 +1,6 @@ /** * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved. - * + *

* This program and the accompanying materials are made available under the * terms of the Eclipse Public License v1.0 which accompanies this distribution, * and is available at http://www.eclipse.org/legal/epl-v10.html @@ -13,21 +13,22 @@ import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.JdkFutureAdapters; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.SettableFuture; -import java.math.BigInteger; import java.util.ArrayList; import java.util.Collection; +import java.util.Iterator; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import org.opendaylight.openflowplugin.api.OFConstants; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; -import org.opendaylight.openflowplugin.api.openflow.device.RequestContext; import org.opendaylight.openflowplugin.api.openflow.device.RequestContextStack; import org.opendaylight.openflowplugin.api.openflow.device.Xid; -import org.opendaylight.openflowplugin.api.openflow.device.exception.DeviceDataException; import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor; import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowHash; +import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy; import org.opendaylight.openflowplugin.impl.registry.flow.FlowDescriptorFactory; import org.opendaylight.openflowplugin.impl.registry.flow.FlowHashFactory; +import org.opendaylight.openflowplugin.impl.util.FlowUtil; import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.FlowConvertor; import org.opendaylight.openflowplugin.openflow.md.util.FlowCreatorUtil; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowId; @@ -44,6 +45,7 @@ import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.Upda import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.UpdateFlowOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.OriginalFlow; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.flow.update.UpdatedFlow; +import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowModInputBuilder; import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcError.ErrorType; @@ -59,47 +61,37 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService super(requestContextStack, deviceContext); } - ListenableFuture> handleServiceCall(final BigInteger connectionID, - final FlowModInputBuilder flowModInputBuilder, final Function, ListenableFuture>> function) { - LOG.debug("Calling the FlowMod RPC method on MessageDispatchService"); - - final RequestContext requestContext = requestContextStack.createRequestContext(); - final SettableFuture> result = requestContextStack.storeOrFail(requestContext); - final DataCrate dataCrate = DataCrateBuilder.builder().setiDConnection(connectionID) - .setRequestContext(requestContext).setFlowModInputBuilder(flowModInputBuilder).build(); - - if (!result.isDone()) { - final ListenableFuture> resultFromOFLib = function.apply(dataCrate); - - final OFJResult2RequestCtxFuture OFJResult2RequestCtxFuture = new OFJResult2RequestCtxFuture<>(requestContext, deviceContext); - OFJResult2RequestCtxFuture.processResultFromOfJava(resultFromOFLib); - - } else { - RequestContextUtil.closeRequstContext(requestContext); - } - return result; - } - @Override public Future> addFlow(final AddFlowInput input) { + final FlowId flowId; + if (null != input.getFlowRef()) { + flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); + } else { + flowId = FlowUtil.createAlienFlowId(input.getTableId()); + } - final FlowId flowId = input.getFlowRef().getValue().firstKeyOf(Flow.class, FlowKey.class).getId(); final FlowHash flowHash = FlowHashFactory.create(input); final FlowDescriptor flowDescriptor = FlowDescriptorFactory.create(input.getTableId(), flowId); deviceContext.getDeviceFlowRegistry().store(flowHash, flowDescriptor); final List ofFlowModInputs = FlowConvertor.toFlowModInputs(input, version, datapathId); - final ListenableFuture future = processFlowModInputBuilders(ofFlowModInputs); + final ListenableFuture> future = processFlowModInputBuilders(ofFlowModInputs); - Futures.addCallback(future, new FutureCallback() { + Futures.addCallback(future, new FutureCallback>() { @Override - public void onSuccess(final Object o) { - LOG.debug("flow add finished without error, id={}", flowId.getValue()); + public void onSuccess(final RpcResult rpcResult) { + messageSpy.spyMessage(input, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMITTED_SUCCESS); + if (rpcResult.isSuccessful()) { + LOG.debug("flow add finished without error, id={}", flowId.getValue()); + } else { + LOG.debug("flow add failed with error, id={}", flowId.getValue()); + } } @Override public void onFailure(final Throwable throwable) { + messageSpy.spyMessage(input, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMITTED_FAILURE); deviceContext.getDeviceFlowRegistry().markToBeremoved(flowHash); LOG.trace("Service call for adding flows failed, id={}.", flowId.getValue(), throwable); } @@ -121,12 +113,14 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService Futures.addCallback(future, new FutureCallback() { @Override public void onSuccess(final Object o) { + messageSpy.spyMessage(input, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMITTED_SUCCESS); FlowHash flowHash = FlowHashFactory.create(input); deviceContext.getDeviceFlowRegistry().markToBeremoved(flowHash); } @Override public void onFailure(final Throwable throwable) { + messageSpy.spyMessage(input, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMITTED_FAILURE); StringBuffer errors = new StringBuffer(); try { RpcResult result = future.get(); @@ -176,6 +170,7 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService Futures.addCallback(future, new FutureCallback() { @Override public void onSuccess(final Object o) { + messageSpy.spyMessage(input, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMITTED_SUCCESS); FlowHash flowHash = FlowHashFactory.create(original); deviceContext.getDeviceFlowRegistry().markToBeremoved(flowHash); @@ -188,7 +183,7 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService @Override public void onFailure(final Throwable throwable) { - + messageSpy.spyMessage(input, MessageSpy.STATISTIC_GROUP.TO_SWITCH_SUBMITTED_FAILURE); } }); return future; @@ -197,45 +192,93 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService private ListenableFuture> processFlowModInputBuilders( final List ofFlowModInputs) { final List>> partialFutures = new ArrayList<>(); + for (FlowModInputBuilder flowModInputBuilder : ofFlowModInputs) { - ListenableFuture> partialFuture = handleServiceCall(PRIMARY_CONNECTION, flowModInputBuilder, + DataCrateBuilder dataCrateBuilder = DataCrateBuilder.builder().setFlowModInputBuilder(flowModInputBuilder); + ListenableFuture> partialFuture = handleServiceCall( + PRIMARY_CONNECTION, new Function, ListenableFuture>>() { @Override public ListenableFuture> apply(final DataCrate data) { return createResultForFlowMod(data); } - }); + }, + dataCrateBuilder + ); partialFutures.add(partialFuture); } - final ListenableFuture>> allFutures = Futures.allAsList(partialFutures); + // processing of final (optionally composite future) + final ListenableFuture>> allFutures = Futures.successfulAsList(partialFutures); final SettableFuture> finalFuture = SettableFuture.create(); Futures.addCallback(allFutures, new FutureCallback>>() { @Override - public void onSuccess(List> result) { - LOG.warn("Positive confirmation of flow push is not supported by OF-spec"); - for (FlowModInputBuilder ofFlowModInput : ofFlowModInputs) { - LOG.warn("flow future result was successful [{}] = this should have never happen", - ofFlowModInput.getXid()); - } - finalFuture.setException(new DeviceDataException("positive confirmation of flow occurred")); - } - - @Override - public void onFailure(Throwable t) { - LOG.trace("Flow mods chained future failed."); + public void onSuccess(List> results) { + List rpcErrorLot = new ArrayList<>(); RpcResultBuilder resultBuilder; - if (allFutures.isCancelled()) { - if (LOG.isTraceEnabled()) { - for (FlowModInputBuilder ofFlowModInput : ofFlowModInputs) { - LOG.trace("flow future result was cancelled [{}] = barrier passed it without error", - ofFlowModInput.getXid()); + + Iterator flowModInputBldIterator = ofFlowModInputs.iterator(); + Iterator> resultIterator = results.iterator(); + + for (ListenableFuture> partFutureFromRqCtx : partialFutures) { + FlowModInputBuilder flowModInputBld = flowModInputBldIterator.next(); + RpcResult result = resultIterator.next(); + Long xid = flowModInputBld.getXid(); + + + LOG.trace("flowMod future processing [{}], result={}", xid, result); + if (partFutureFromRqCtx.isCancelled()) { // one and only positive case + if (LOG.isTraceEnabled()) { + LOG.trace("flow future result was cancelled [{}] = barrier passed it without error", xid); + } + } else { // all negative cases + if (result == null) { // there is exception or null value set + try { + partFutureFromRqCtx.get(); + } catch (Exception e) { + rpcErrorLot.add(RpcResultBuilder.newError(ErrorType.APPLICATION, "", + "flow future result [" + xid + "] failed with exception", + OFConstants.APPLICATION_TAG, e.getMessage(), e)); + + // xid might be not available in case requestContext not even stored + if (xid != null) { + deviceContext.unhookRequestCtx(new Xid(xid)); + } + } + } else { + if (result.isSuccessful()) { // positive confirmation - never happens + LOG.warn("Positive confirmation of flow push is not supported by OF-spec"); + LOG.warn("flow future result was successful [{}] = this should have never happen", + xid); + rpcErrorLot.add(RpcResultBuilder.newError(ErrorType.APPLICATION, "", + "flow future result was successful [" + xid + "] = this should have never happen")); + } else { // standard error occurred + LOG.trace("passing original rpcErrors [{}]", xid); + if (LOG.isTraceEnabled()) { + for (RpcError rpcError : result.getErrors()) { + LOG.trace("passed rpcError [{}]: {}", xid, rpcError); + } + } + rpcErrorLot.addAll(result.getErrors()); + } } } + } + + if (rpcErrorLot.isEmpty()) { resultBuilder = RpcResultBuilder.success(); } else { - resultBuilder = RpcResultBuilder.failed().withError(ErrorType.APPLICATION, "", t.getMessage()); + resultBuilder = RpcResultBuilder.failed().withRpcErrors(rpcErrorLot); } + + finalFuture.set(resultBuilder.build()); + } + + @Override + public void onFailure(Throwable t) { + LOG.trace("Flow mods chained future failed."); + RpcResultBuilder resultBuilder = RpcResultBuilder.failed() + .withError(ErrorType.APPLICATION, "", t.getMessage()); finalFuture.set(resultBuilder.build()); } }); @@ -247,13 +290,15 @@ public class SalFlowServiceImpl extends CommonService implements SalFlowService return createResultForFlowMod(data, data.getFlowModInputBuilder()); } - protected ListenableFuture> createResultForFlowMod(final DataCrate data, final FlowModInputBuilder flowModInput) { - final Xid xId = deviceContext.getNextXid(); - flowModInput.setXid(xId.getValue()); - data.getRequestContext().setXid(xId); + protected ListenableFuture> createResultForFlowMod(final DataCrate data, final FlowModInputBuilder flowModInputBuilder) { + final Xid xid = data.getRequestContext().getXid(); + flowModInputBuilder.setXid(xid.getValue()); + final FlowModInput flowModInput = flowModInputBuilder.build(); Future> flowModResult = provideConnectionAdapter(data.getiDConnection()).flowMod( - flowModInput.build()); - return JdkFutureAdapters.listenInPoolThread(flowModResult); + flowModInput); + + final ListenableFuture> result = JdkFutureAdapters.listenInPoolThread(flowModResult); + return result; } }