X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fdevice%2FDeviceContextImpl.java;h=872a0df1fa5de427253b3f6303068f655ff17639;hb=9688b3b10ee4113753705f0080b29cd30e61a85d;hp=59eb2b943a602be6f07b608fad6b8941eb396d00;hpb=3cfeb72ceab5ff49efcc5446191b0e37280e8f38;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java index 59eb2b943a..872a0df1fa 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.java @@ -8,17 +8,24 @@ package org.opendaylight.openflowplugin.impl.device; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Function; -import com.google.common.base.Preconditions; -import com.google.common.base.Verify; -import com.google.common.util.concurrent.AsyncFunction; import com.google.common.util.concurrent.FutureCallback; 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.MoreExecutors; import io.netty.util.HashedWheelTimer; -import io.netty.util.Timeout; -import io.netty.util.TimerTask; +import java.util.Collection; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Objects; +import java.util.Optional; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; import org.opendaylight.controller.md.sal.binding.api.DataBroker; import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService; import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction; @@ -26,102 +33,76 @@ import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType; import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier; import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter; import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey; -import org.opendaylight.openflowplugin.api.ConnectionException; import org.opendaylight.openflowplugin.api.OFConstants; import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo; -import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager; import org.opendaylight.openflowplugin.api.openflow.device.DeviceState; import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator; import org.opendaylight.openflowplugin.api.openflow.device.RequestContext; import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary; import org.opendaylight.openflowplugin.api.openflow.device.Xid; -import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler; import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector; -import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService; -import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipWatcher; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainState; import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey; import org.opendaylight.openflowplugin.api.openflow.md.util.OpenflowVersion; -import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry; import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry; -import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor; -import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey; import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry; import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry; -import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper; -import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener; import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy; +import org.opendaylight.openflowplugin.common.txchain.TransactionChainManager; import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava; import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper; import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider; import org.opendaylight.openflowplugin.extension.api.exception.ConversionException; import org.opendaylight.openflowplugin.extension.api.path.MessagePath; -import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl; import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProvider; import org.opendaylight.openflowplugin.impl.datastore.MultipartWriterProviderFactory; import org.opendaylight.openflowplugin.impl.device.initialization.AbstractDeviceInitializer; import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider; import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl; import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl; -import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory; import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl; import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl; import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext; +import org.opendaylight.openflowplugin.impl.services.util.RequestContextUtil; +import org.opendaylight.openflowplugin.impl.util.MatchUtil; import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor; -import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl; import org.opendaylight.openflowplugin.openflow.md.util.InventoryDataServiceUtil; import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode; import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector; -import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table; -import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow; -import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId; +import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node; +import org.opendaylight.yang.gen.v1.urn.opendaylight.model.match.types.rev131026.Match; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader; -import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping; +import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatus; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice; import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice; +import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketIn; import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived; +import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceivedBuilder; import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData; import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole; -import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SalRoleService; -import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInput; -import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInputBuilder; -import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleOutput; +import org.opendaylight.yangtools.yang.binding.DataContainer; import org.opendaylight.yangtools.yang.binding.DataObject; import org.opendaylight.yangtools.yang.binding.InstanceIdentifier; import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier; -import org.opendaylight.yangtools.yang.common.RpcResult; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.Nonnull; -import javax.annotation.Nullable; -import java.math.BigInteger; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; -import java.util.concurrent.TimeUnit; - public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper { private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class); @@ -133,112 +114,96 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi // TODO: high water mark factor should be parametrized private static final float HIGH_WATERMARK_FACTOR = 0.95f; - // Timeout in seconds after what we will give up on propagating role - private static final int SET_ROLE_TIMEOUT = 10; + // Timeout in milliseconds after what we will give up on initializing device + private static final int DEVICE_INIT_TIMEOUT = 9000; - private boolean initialized; + // Timeout in milliseconds after what we will give up on closing transaction chain + private static final int TX_CHAIN_CLOSE_TIMEOUT = 10000; - private SalRoleService salRoleService = null; + private static final int LOW_WATERMARK = 1000; + private static final int HIGH_WATERMARK = 2000; + + private final MultipartWriterProvider writerProvider; private final HashedWheelTimer hashedWheelTimer; - private ConnectionContext primaryConnectionContext; private final DeviceState deviceState; private final DataBroker dataBroker; - private final Map auxiliaryConnectionContexts; - private TransactionChainManager transactionChainManager; - private DeviceFlowRegistry deviceFlowRegistry; - private DeviceGroupRegistry deviceGroupRegistry; - private DeviceMeterRegistry deviceMeterRegistry; - private final PacketInRateLimiter packetInLimiter; + private final Collection> requestContexts = new HashSet<>(); private final MessageSpy messageSpy; - private final ItemLifeCycleKeeper flowLifeCycleKeeper; - private NotificationPublishService notificationPublishService; - private Timeout barrierTaskTimeout; private final MessageTranslator portStatusTranslator; private final MessageTranslator packetInTranslator; - private final MessageTranslator flowRemovedTranslator; + private final MessageTranslator flowRemovedTranslator; private final TranslatorLibrary translatorLibrary; - private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry; - private ExtensionConverterProvider extensionConverterProvider; - private final DeviceManager deviceManager; - private boolean skipTableFeatures; - private boolean switchFeaturesMandatory; - private final DeviceInfo deviceInfo; private final ConvertorExecutor convertorExecutor; - private volatile CONTEXT_STATE state; - private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler; - private final DeviceManager myManager; private final DeviceInitializerProvider deviceInitializerProvider; + private final PacketInRateLimiter packetInLimiter; + private final DeviceInfo deviceInfo; + private final ConnectionContext primaryConnectionContext; + private final boolean skipTableFeatures; + private final boolean switchFeaturesMandatory; + private final boolean isFlowRemovedNotificationOn; private final boolean useSingleLayerSerialization; - private Boolean isAddNotificationSent = false; - - DeviceContextImpl( - @Nonnull final ConnectionContext primaryConnectionContext, - @Nonnull final DataBroker dataBroker, - @Nonnull final MessageSpy messageSpy, - @Nonnull final TranslatorLibrary translatorLibrary, - @Nonnull final DeviceManager manager, - final ConvertorExecutor convertorExecutor, - final boolean skipTableFeatures, - final HashedWheelTimer hashedWheelTimer, - final DeviceManager myManager, - final boolean useSingleLayerSerialization, - final DeviceInitializerProvider deviceInitializerProvider) { + private final AtomicBoolean initialized = new AtomicBoolean(false); + private final AtomicBoolean hasState = new AtomicBoolean(false); + private final AtomicBoolean isInitialTransactionSubmitted = new AtomicBoolean(false); + private NotificationPublishService notificationPublishService; + private TransactionChainManager transactionChainManager; + private DeviceFlowRegistry deviceFlowRegistry; + private DeviceGroupRegistry deviceGroupRegistry; + private DeviceMeterRegistry deviceMeterRegistry; + private ExtensionConverterProvider extensionConverterProvider; + private ContextChainMastershipWatcher contextChainMastershipWatcher; + + DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext, + @Nonnull final DataBroker dataBroker, + @Nonnull final MessageSpy messageSpy, + @Nonnull final TranslatorLibrary translatorLibrary, + final ConvertorExecutor convertorExecutor, + final boolean skipTableFeatures, + final HashedWheelTimer hashedWheelTimer, + final boolean useSingleLayerSerialization, + final DeviceInitializerProvider deviceInitializerProvider, + final boolean isFlowRemovedNotificationOn, + final boolean switchFeaturesMandatory) { this.primaryConnectionContext = primaryConnectionContext; this.deviceInfo = primaryConnectionContext.getDeviceInfo(); this.hashedWheelTimer = hashedWheelTimer; - this.myManager = myManager; this.deviceInitializerProvider = deviceInitializerProvider; + this.isFlowRemovedNotificationOn = isFlowRemovedNotificationOn; + this.switchFeaturesMandatory = switchFeaturesMandatory; this.deviceState = new DeviceStateImpl(); this.dataBroker = dataBroker; - this.auxiliaryConnectionContexts = new HashMap<>(); - this.messageSpy = Preconditions.checkNotNull(messageSpy); - this.deviceManager = manager; + this.messageSpy = messageSpy; this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(), - /*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR); + /*initial*/ LOW_WATERMARK, /*initial*/HIGH_WATERMARK, this.messageSpy, REJECTED_DRAIN_FACTOR); this.translatorLibrary = translatorLibrary; this.portStatusTranslator = translatorLibrary.lookupTranslator( new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName())); this.packetInTranslator = translatorLibrary.lookupTranslator( - new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName())); + new TranslatorKey(deviceInfo.getVersion(), org.opendaylight.yang.gen.v1.urn.opendaylight.openflow + .protocol.rev130731 + .PacketIn.class.getName())); this.flowRemovedTranslator = translatorLibrary.lookupTranslator( new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName())); - this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl(); - this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl(); - this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper); - this.state = CONTEXT_STATE.INITIALIZATION; this.convertorExecutor = convertorExecutor; this.skipTableFeatures = skipTableFeatures; this.useSingleLayerSerialization = useSingleLayerSerialization; - this.initialized = false; + writerProvider = MultipartWriterProviderFactory.createDefaultProvider(this); } @Override - public void initialSubmitTransaction() { - if (initialized) { - transactionChainManager.initialSubmitWriteTransaction(); + public boolean initialSubmitTransaction() { + if (!initialized.get()) { + return false; } - } - @Override - public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) { - final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext); - auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext); - } - - private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) { - return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId()); - } - - @Override - public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) { - final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext); - LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter() - .getRemoteAddress(), getDeviceInfo().getLOGValue()); - auxiliaryConnectionContexts.remove(connectionDistinguisher); + final boolean initialSubmit = transactionChainManager.initialSubmitWriteTransaction(); + isInitialTransactionSubmitted.set(initialSubmit); + return initialSubmit; } @Override @@ -251,11 +216,16 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return dataBroker.newReadOnlyTransaction(); } + @Override + public boolean isTransactionsEnabled() { + return isInitialTransactionSubmitted.get(); + } + @Override public void writeToTransaction(final LogicalDatastoreType store, final InstanceIdentifier path, - final T data){ - if (initialized) { + final T data) { + if (initialized.get()) { transactionChainManager.writeToTransaction(store, path, data, false); } } @@ -263,22 +233,23 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void writeToTransactionWithParentsSlow(final LogicalDatastoreType store, final InstanceIdentifier path, - final T data){ - if (initialized) { + final T data) { + if (initialized.get()) { transactionChainManager.writeToTransaction(store, path, data, true); } } @Override - public void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier path) { - if (initialized) { - transactionChainManager.addDeleteOperationTotTxChain(store, path); + public void addDeleteToTxChain(final LogicalDatastoreType store, + final InstanceIdentifier path) { + if (initialized.get()) { + transactionChainManager.addDeleteOperationToTxChain(store, path); } } @Override public boolean submitTransaction() { - return initialized && transactionChainManager.submitWriteTransaction(); + return initialized.get() && transactionChainManager.submitTransaction(); } @Override @@ -286,11 +257,6 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return primaryConnectionContext; } - @Override - public ConnectionContext getAuxiliaryConnectionContexts(final BigInteger cookie) { - return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue())); - } - @Override public DeviceFlowRegistry getDeviceFlowRegistry() { return deviceFlowRegistry; @@ -309,19 +275,19 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void processReply(final OfHeader ofHeader) { messageSpy.spyMessage( - ofHeader.getImplementedInterface(), - (ofHeader instanceof Error) - ? MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE - : MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS); + ofHeader.getImplementedInterface(), + ofHeader instanceof Error + ? MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_FAILURE + : MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS); } @Override public void processReply(final Xid xid, final List ofHeaderList) { ofHeaderList.forEach(header -> messageSpy.spyMessage( - header.getImplementedInterface(), - (header instanceof Error) - ? MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE - : MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS)); + header.getImplementedInterface(), + header instanceof Error + ? MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_FAILURE + : MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS)); } @Override @@ -330,110 +296,113 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification = flowRemovedTranslator.translate(flowRemoved, deviceInfo, null); - if(deviceManager.isFlowRemovedNotificationOn()) { + if (isFlowRemovedNotificationOn) { // Trigger off a notification notificationPublishService.offerNotification(flowRemovedNotification); - } else if(LOG.isDebugEnabled()) { - LOG.debug("For nodeId={} isFlowRemovedNotificationOn={}", getDeviceInfo().getLOGValue(), deviceManager.isFlowRemovedNotificationOn()); } - - final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener(); - if (itemLifecycleListener != null) { - //2. create registry key - final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(getDeviceInfo().getVersion(), flowRemovedNotification); - //3. lookup flowId - final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey); - //4. if flowId present: - if (flowDescriptor != null) { - // a) construct flow path - final KeyedInstanceIdentifier flowPath = getDeviceInfo().getNodeInstanceIdentifier() - .augmentation(FlowCapableNode.class) - .child(Table.class, flowDescriptor.getTableKey()) - .child(Flow.class, new FlowKey(flowDescriptor.getFlowId())); - // b) notify listener - itemLifecycleListener.onRemoved(flowPath); - } else { - LOG.debug("flow id not found: nodeId={} tableId={}, priority={}", - getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority()); - } - } - } - - @Override - public void sendNodeAddedNotification() { - if (!isAddNotificationSent) { - isAddNotificationSent = true; - NodeUpdatedBuilder builder = new NodeUpdatedBuilder(); - builder.setId(getDeviceInfo().getNodeId()); - builder.setNodeRef(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())); - LOG.debug("Publishing node added notification for {}", builder.build()); - notificationPublishService.offerNotification(builder.build()); - } - } - - @Override - public void sendNodeRemovedNotification() { - NodeRemovedBuilder builder = new NodeRemovedBuilder(); - builder.setNodeRef(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())); - LOG.debug("Publishing node removed notification for {}", builder.build()); - notificationPublishService.offerNotification(builder.build()); } @Override + @SuppressWarnings("checkstyle:IllegalCatch") public void processPortStatusMessage(final PortStatusMessage portStatus) { - messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS); - final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, getDeviceInfo(), null); - - final KeyedInstanceIdentifier iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion()); - try { - if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) { - // because of ADD status node connector has to be created - final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey()); - nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build()); - nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector); - writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build()); - } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) { - addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector); + messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.StatisticsGroup + .FROM_SWITCH_PUBLISHED_SUCCESS); + + if (initialized.get()) { + try { + writePortStatusMessage(portStatus); + submitTransaction(); + } catch (final Exception e) { + LOG.warn("Error processing port status message for port {} on device {}", + portStatus.getPortNo(), getDeviceInfo(), e); } - submitTransaction(); - } catch (final Exception e) { - LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo(), - getDeviceInfo().getNodeId().toString(), e); + } else if (!hasState.get()) { + primaryConnectionContext.handlePortStatusMessage(portStatus); } } - private KeyedInstanceIdentifier provideIIToNodeConnector(final long portNo, final short version) { - final InstanceIdentifier iiToNodes = getDeviceInfo().getNodeInstanceIdentifier(); - final BigInteger dataPathId = getDeviceInfo().getDatapathId(); - final NodeConnectorId nodeConnectorId = InventoryDataServiceUtil.nodeConnectorIdfromDatapathPortNo(dataPathId, portNo, OpenflowVersion.get(version)); - return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId)); + private void writePortStatusMessage(final PortStatus portStatusMessage) { + final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator + .translate(portStatusMessage, getDeviceInfo(), null); + + final KeyedInstanceIdentifier iiToNodeConnector = getDeviceInfo() + .getNodeInstanceIdentifier() + .child(NodeConnector.class, new NodeConnectorKey(InventoryDataServiceUtil + .nodeConnectorIdfromDatapathPortNo( + deviceInfo.getDatapathId(), + portStatusMessage.getPortNo(), + OpenflowVersion.get(deviceInfo.getVersion())))); + + if (PortReason.OFPPRADD.equals(portStatusMessage.getReason()) + || PortReason.OFPPRMODIFY.equals(portStatusMessage.getReason())) { + // because of ADD status node connector has to be created + writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, new NodeConnectorBuilder() + .setKey(iiToNodeConnector.getKey()) + .addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new + FlowCapableNodeConnectorStatisticsDataBuilder().build()) + .addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector) + .build()); + } else if (PortReason.OFPPRDELETE.equals(portStatusMessage.getReason())) { + addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector); + } } @Override public void processPacketInMessage(final PacketInMessage packetInMessage) { - messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH); - final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter(); final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null); + handlePacketInMessage(packetReceived, packetInMessage.getImplementedInterface(), packetReceived.getMatch()); + } - if (packetReceived == null) { + private void handlePacketInMessage(final PacketIn packetIn, + final Class implementedInterface, + final Match match) { + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup.FROM_SWITCH); + final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter(); + + if (packetIn == null) { LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress()); - messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE); + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup.FROM_SWITCH_TRANSLATE_SRC_FAILURE); return; - } else { - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS); } + final OpenflowVersion openflowVersion = OpenflowVersion.get(deviceInfo.getVersion()); + + // Try to get ingress from match + final NodeConnectorRef nodeConnectorRef = Objects.nonNull(packetIn.getIngress()) + ? packetIn.getIngress() : Optional.ofNullable(match) + .map(Match::getInPort) + .map(nodeConnectorId -> InventoryDataServiceUtil + .portNumberfromNodeConnectorId( + openflowVersion, + nodeConnectorId)) + .map(portNumber -> InventoryDataServiceUtil + .nodeConnectorRefFromDatapathIdPortno( + deviceInfo.getDatapathId(), + portNumber, + openflowVersion)) + .orElse(null); + + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup.FROM_SWITCH_TRANSLATE_OUT_SUCCESS); + if (!packetInLimiter.acquirePermit()) { LOG.debug("Packet limited"); // TODO: save packet into emergency slot if possible - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED); + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup + .FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED); return; } - final ListenableFuture offerNotification = notificationPublishService.offerNotification(packetReceived); + final ListenableFuture offerNotification = notificationPublishService + .offerNotification(new PacketReceivedBuilder(packetIn) + .setIngress(nodeConnectorRef) + .setMatch(MatchUtil.transformMatch(match, + org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.packet.received + .Match.class)) + .build()); + if (NotificationPublishService.REJECTED.equals(offerNotification)) { LOG.debug("notification offer rejected"); - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED); + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup.FROM_SWITCH_NOTIFICATION_REJECTED); packetInLimiter.drainLowWaterMark(); packetInLimiter.releasePermit(); return; @@ -442,18 +411,19 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi Futures.addCallback(offerNotification, new FutureCallback() { @Override public void onSuccess(final Object result) { - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS); + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup.FROM_SWITCH_PUBLISHED_SUCCESS); packetInLimiter.releasePermit(); } @Override - public void onFailure(final Throwable t) { - messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED); - LOG.debug("notification offer failed: {}", t.getMessage()); - LOG.trace("notification offer failed..", t); + public void onFailure(final Throwable throwable) { + messageSpy.spyMessage(implementedInterface, MessageSpy.StatisticsGroup + .FROM_SWITCH_NOTIFICATION_REJECTED); + LOG.debug("notification offer failed: {}", throwable.getMessage()); + LOG.trace("notification offer failed..", throwable); packetInLimiter.releasePermit(); } - }); + }, MoreExecutors.directExecutor()); } @Override @@ -463,7 +433,8 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi final MessageTypeKey key = new MessageTypeKey<>( getDeviceInfo().getVersion(), (Class) vendorData.getImplementedInterface()); - final ConvertorMessageFromOFJava messageConverter = extensionConverterProvider.getMessageConverter(key); + final ConvertorMessageFromOFJava messageConverter = + extensionConverterProvider.getMessageConverter(key); if (messageConverter == null) { LOG.warn("custom converter for {}[OF:{}] not found", notification.getExperimenterDataOfChoice().getImplementedInterface(), @@ -474,8 +445,9 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi final ExperimenterMessageOfChoice messageOfChoice; try { messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION); - final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder() - .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())) + final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new + ExperimenterMessageFromDevBuilder() + .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier())) .setExperimenterMessageOfChoice(messageOfChoice); // publish notificationPublishService.offerNotification(experimenterMessageFromDevBld.build()); @@ -485,18 +457,25 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi } @Override - public TranslatorLibrary oook() { - return translatorLibrary; - } + public boolean processAlienMessage(final OfHeader message) { + final Class implementedInterface = message.getImplementedInterface(); - @Override - public void setCurrentBarrierTimeout(final Timeout timeout) { - barrierTaskTimeout = timeout; + if (Objects.nonNull(implementedInterface) && org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service + .rev130709.PacketInMessage.class.equals(implementedInterface)) { + final org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709 + .PacketInMessage packetInMessage = org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service + .rev130709.PacketInMessage.class.cast(message); + + handlePacketInMessage(packetInMessage, implementedInterface, packetInMessage.getMatch()); + return true; + } + + return false; } @Override - public Timeout getBarrierTaskTimeout() { - return barrierTaskTimeout; + public TranslatorLibrary oook() { + return translatorLibrary; } @Override @@ -511,27 +490,19 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi @Override public void onPublished() { - Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState())); - this.state = CONTEXT_STATE.WORKING; primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false); - for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) { - switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false); - } } @Override - public MultiMsgCollector getMultiMsgCollector(final RequestContext> requestContext) { + public MultiMsgCollector getMultiMsgCollector(final RequestContext> + requestContext) { return new MultiMsgCollectorImpl<>(this, requestContext); } @Override public void updatePacketInRateLimit(final long upperBound) { - packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound)); - } - - @Override - public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() { - return itemLifeCycleSourceRegistry; + packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), + (int) (HIGH_WATERMARK_FACTOR * upperBound)); } @Override @@ -544,144 +515,70 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return extensionConverterProvider; } - @Override - public synchronized void shutdownConnection() { - if (LOG.isDebugEnabled()) { - LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue()); - } - if (CONTEXT_STATE.TERMINATION.equals(getState())) { - LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue()); - return; - } - - if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) { - LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getLOGValue()); - return; - } - - // Terminate Auxiliary Connection - for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) { - LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId()); - connectionContext.closeConnection(false); - } - - // Terminate Primary Connection - getPrimaryConnectionContext().closeConnection(true); - - // Close all datastore registries - if (initialized) { - deviceGroupRegistry.close(); - deviceFlowRegistry.close(); - deviceMeterRegistry.close(); - } + @VisibleForTesting + TransactionChainManager getTransactionChainManager() { + return this.transactionChainManager; } @Override - public ListenableFuture shuttingDownDataStoreTransactions() { - return initialized - ? this.transactionChainManager.shuttingDown() + public ListenableFuture closeServiceInstance() { + final ListenableFuture listenableFuture = initialized.get() + ? transactionChainManager.deactivateTransactionManager() : Futures.immediateFuture(null); + + hashedWheelTimer.newTimeout((timerTask) -> { + if (!listenableFuture.isDone() && !listenableFuture.isCancelled()) { + listenableFuture.cancel(true); + } + }, TX_CHAIN_CLOSE_TIMEOUT, TimeUnit.MILLISECONDS); + + return listenableFuture; } - @VisibleForTesting - TransactionChainManager getTransactionChainManager() { - return this.transactionChainManager; + @Override + public DeviceInfo getDeviceInfo() { + return this.deviceInfo; } @Override - public void setSwitchFeaturesMandatory(boolean switchFeaturesMandatory) { - this.switchFeaturesMandatory = switchFeaturesMandatory; + public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher newWatcher) { + this.contextChainMastershipWatcher = newWatcher; } + @Nonnull @Override - public CONTEXT_STATE getState() { - return this.state; + public ServiceGroupIdentifier getIdentifier() { + return deviceInfo.getServiceIdentifier(); } @Override - public ListenableFuture stopClusterServices(boolean connectionInterrupted) { - final ListenableFuture deactivateTxManagerFuture = initialized - ? transactionChainManager.deactivateTransactionManager() - : Futures.immediateFuture(null); + public void close() { + // Close all datastore registries and transactions + if (initialized.getAndSet(false)) { + deviceGroupRegistry.close(); + deviceFlowRegistry.close(); + deviceMeterRegistry.close(); - if (!connectionInterrupted) { - final ListenableFuture makeSlaveFuture = Futures.transform(makeDeviceSlave(), new Function, Void>() { - @Nullable - @Override - public Void apply(@Nullable RpcResult setRoleOutputRpcResult) { - return null; - } - }); + final ListenableFuture txChainShuttingDown = transactionChainManager.shuttingDown(); - Futures.addCallback(makeSlaveFuture, new FutureCallback() { + Futures.addCallback(txChainShuttingDown, new FutureCallback() { @Override - public void onSuccess(@Nullable Void aVoid) { - if (LOG.isDebugEnabled()) { - LOG.debug("Role SLAVE was successfully propagated on device, node {}", deviceInfo.getLOGValue()); - } - sendNodeAddedNotification(); + public void onSuccess(@Nullable final Void result) { + transactionChainManager.close(); + transactionChainManager = null; } @Override public void onFailure(final Throwable throwable) { - LOG.warn("Was not able to set role SLAVE to device on node {} ", deviceInfo.getLOGValue()); - LOG.trace("Error occurred on device role setting, probably connection loss: ", throwable); - } - }); - - return Futures.transform(deactivateTxManagerFuture, new AsyncFunction() { - @Override - public ListenableFuture apply(Void aVoid) throws Exception { - // Add fallback to remove device from operational DS if setting slave fails - return Futures.withFallback(makeSlaveFuture, t -> - myManager.removeDeviceFromOperationalDS(deviceInfo)); - } - }); - } else { - return Futures.transform(deactivateTxManagerFuture, new AsyncFunction() { - @Override - public ListenableFuture apply(Void aVoid) throws Exception { - return myManager.removeDeviceFromOperationalDS(deviceInfo); + transactionChainManager.close(); + transactionChainManager = null; } - }); + }, MoreExecutors.directExecutor()); } - } - - @Override - public ServiceGroupIdentifier getServiceIdentifier() { - return this.deviceInfo.getServiceIdentifier(); - } - @Override - public DeviceInfo getDeviceInfo() { - return this.deviceInfo; - } - - @Override - public void close() { - if (CONTEXT_STATE.TERMINATION.equals(getState())){ - if (LOG.isDebugEnabled()) { - LOG.debug("DeviceContext for node {} is already in TERMINATION state.", getDeviceInfo().getLOGValue()); - } - } else { - this.state = CONTEXT_STATE.TERMINATION; - } - sendNodeRemovedNotification(); - } - - @Override - public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){ - if (initialized) { - this.transactionChainManager.setLifecycleService(lifecycleService); - } - } - - @Override - public void replaceConnectionContext(final ConnectionContext connectionContext){ - // Act like we are initializing the context - this.state = CONTEXT_STATE.INITIALIZATION; - this.primaryConnectionContext = connectionContext; - this.onPublished(); + requestContexts.forEach(requestContext -> RequestContextUtil + .closeRequestContextWithRpcError(requestContext, "Connection closed.")); + requestContexts.clear(); } @Override @@ -689,126 +586,147 @@ public class DeviceContextImpl implements DeviceContext, ExtensionConverterProvi return useSingleLayerSerialization && getDeviceInfo().getVersion() >= OFConstants.OFP_VERSION_1_3; } + // TODO: exception handling should be fixed by using custom checked exception, never RuntimeExceptions @Override - public boolean isSkipTableFeatures() { - return this.skipTableFeatures; - } - - @Override - public void setSalRoleService(@Nonnull SalRoleService salRoleService) { - this.salRoleService = salRoleService; - } - - @Override - public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) { - this.clusterInitializationPhaseHandler = handler; - } + @SuppressWarnings({"checkstyle:IllegalCatch"}) + public void instantiateServiceInstance() { + lazyTransactionManagerInitialization(); - @Override - public boolean onContextInstantiateService(final ConnectionContext connectionContext) { + try { + final List portStatusMessages = primaryConnectionContext + .retrieveAndClearPortStatusMessages(); - if (getPrimaryConnectionContext().getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) { - LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue()); - return false; + portStatusMessages.forEach(this::writePortStatusMessage); + submitTransaction(); + } catch (final Exception ex) { + throw new RuntimeException(String.format("Error processing port status messages from device %s: %s", + deviceInfo.toString(), ex.toString()), ex); } - LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue()); - - lazyTransactionManagerInitialization(); - - this.transactionChainManager.activateTransactionManager(); - - try { - final Optional initializer = deviceInitializerProvider + final Optional initializer = deviceInitializerProvider .lookup(deviceInfo.getVersion()); - if (initializer.isPresent()) { - final MultipartWriterProvider writerProvider = MultipartWriterProviderFactory.createDefaultProvider(this); - initializer.get().initialize(this, switchFeaturesMandatory, writerProvider, convertorExecutor); - } else { - throw new ExecutionException(new ConnectionException("Unsupported version " + deviceInfo.getVersion())); + if (initializer.isPresent()) { + final Future initialize = initializer + .get() + .initialize(this, switchFeaturesMandatory, skipTableFeatures, writerProvider, convertorExecutor); + + try { + initialize.get(DEVICE_INIT_TIMEOUT, TimeUnit.MILLISECONDS); + } catch (TimeoutException ex) { + initialize.cancel(true); + throw new RuntimeException(String.format("Failed to initialize device %s in %ss: %s", + deviceInfo.toString(), String.valueOf(DEVICE_INIT_TIMEOUT / 1000), ex.toString()), ex); + } catch (ExecutionException | InterruptedException ex) { + throw new RuntimeException( + String.format("Device %s cannot be initialized: %s", deviceInfo.toString(), ex.toString()), ex); } - } catch (ExecutionException | InterruptedException e) { - LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), e); - return false; + } else { + throw new RuntimeException(String.format("Unsupported version %s for device %s", + deviceInfo.getVersion(), + deviceInfo.toString())); } - Futures.addCallback(sendRoleChangeToDevice(OfpRole.BECOMEMASTER), new RpcResultFutureCallback()); - return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext()); + final ListenableFuture>> deviceFlowRegistryFill = + getDeviceFlowRegistry().fill(); + Futures.addCallback(deviceFlowRegistryFill, + new DeviceFlowRegistryCallback(deviceFlowRegistryFill, contextChainMastershipWatcher), + MoreExecutors.directExecutor()); } @VisibleForTesting void lazyTransactionManagerInitialization() { - if (!this.initialized) { + if (!this.initialized.get()) { if (LOG.isDebugEnabled()) { - LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue()); + LOG.debug("Transaction chain manager for node {} created", deviceInfo); } - this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo); - this.deviceFlowRegistry = new DeviceFlowRegistryImpl(deviceInfo.getVersion(), dataBroker, deviceInfo.getNodeInstanceIdentifier()); + this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo.getNodeId().getValue()); + this.deviceFlowRegistry = new DeviceFlowRegistryImpl(deviceInfo.getVersion(), dataBroker, + deviceInfo.getNodeInstanceIdentifier()); this.deviceGroupRegistry = new DeviceGroupRegistryImpl(); this.deviceMeterRegistry = new DeviceMeterRegistryImpl(); - this.initialized = true; } + + transactionChainManager.activateTransactionManager(); + initialized.set(true); } @Nullable @Override public RequestContext createRequestContext() { - return new AbstractRequestContext(deviceInfo.reserveXidForDeviceMessage()) { + final Long xid = deviceInfo.reserveXidForDeviceMessage(); + + final AbstractRequestContext abstractRequestContext = new AbstractRequestContext(xid) { @Override public void close() { + requestContexts.remove(this); } }; + requestContexts.add(abstractRequestContext); + return abstractRequestContext; } - ListenableFuture> sendRoleChangeToDevice(final OfpRole newRole) { - if (LOG.isDebugEnabled()) { - LOG.debug("Sending new role {} to device {}", newRole, deviceInfo.getNodeId()); - } - - final Future> setRoleOutputFuture; - - if (deviceInfo.getVersion() >= OFConstants.OFP_VERSION_1_3) { - final SetRoleInput setRoleInput = (new SetRoleInputBuilder()).setControllerRole(newRole) - .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier())).build(); + @Override + public void onStateAcquired(final ContextChainState state) { + hasState.set(true); + } - setRoleOutputFuture = this.salRoleService.setRole(setRoleInput); + private class DeviceFlowRegistryCallback implements FutureCallback>> { + private final ListenableFuture>> deviceFlowRegistryFill; + private final ContextChainMastershipWatcher contextChainMastershipWatcher; - final TimerTask timerTask = timeout -> { - if (!setRoleOutputFuture.isDone()) { - LOG.warn("New role {} was not propagated to device {} during {} sec", newRole, deviceInfo.getLOGValue(), SET_ROLE_TIMEOUT); - setRoleOutputFuture.cancel(true); - } - }; - - hashedWheelTimer.newTimeout(timerTask, SET_ROLE_TIMEOUT, TimeUnit.SECONDS); - } else { - LOG.info("Device: {} with version: {} does not support role", deviceInfo.getLOGValue(), deviceInfo.getVersion()); - return Futures.immediateFuture(null); + DeviceFlowRegistryCallback( + ListenableFuture>> deviceFlowRegistryFill, + ContextChainMastershipWatcher contextChainMastershipWatcher) { + this.deviceFlowRegistryFill = deviceFlowRegistryFill; + this.contextChainMastershipWatcher = contextChainMastershipWatcher; } - return JdkFutureAdapters.listenInPoolThread(setRoleOutputFuture); - } - - @Override - public ListenableFuture> makeDeviceSlave() { - return sendRoleChangeToDevice(OfpRole.BECOMESLAVE); - } - - private class RpcResultFutureCallback implements FutureCallback> { @Override - public void onSuccess(@Nullable RpcResult setRoleOutputRpcResult) { + public void onSuccess(@Nullable List> result) { if (LOG.isDebugEnabled()) { - LOG.debug("Role MASTER was successfully set on device, node {}", deviceInfo.getLOGValue()); + // Count all flows we read from datastore for debugging purposes. + // This number do not always represent how many flows were actually added + // to DeviceFlowRegistry, because of possible duplicates. + long flowCount = Optional.ofNullable(result) + .map(Collections::singleton) + .orElse(Collections.emptySet()) + .stream() + .flatMap(Collection::stream) + .filter(Objects::nonNull) + .flatMap(flowCapableNodeOptional -> flowCapableNodeOptional.asSet().stream()) + .filter(Objects::nonNull) + .filter(flowCapableNode -> Objects.nonNull(flowCapableNode.getTable())) + .flatMap(flowCapableNode -> flowCapableNode.getTable().stream()) + .filter(Objects::nonNull) + .filter(table -> Objects.nonNull(table.getFlow())) + .flatMap(table -> table.getFlow().stream()) + .filter(Objects::nonNull) + .count(); + + LOG.debug("Finished filling flow registry with {} flows for node: {}", flowCount, deviceInfo); } - sendNodeAddedNotification(); + this.contextChainMastershipWatcher.onMasterRoleAcquired(deviceInfo, ContextChainMastershipState + .INITIAL_FLOW_REGISTRY_FILL); } @Override - public void onFailure(final Throwable throwable) { - LOG.warn("Was not able to set MASTER role on device, node {}", deviceInfo.getLOGValue()); - shutdownConnection(); + public void onFailure(Throwable throwable) { + if (deviceFlowRegistryFill.isCancelled()) { + if (LOG.isDebugEnabled()) { + LOG.debug("Cancelled filling flow registry with flows for node: {}", deviceInfo); + } + } else { + LOG.warn("Failed filling flow registry with flows for node: {} with exception: {}", deviceInfo, + throwable); + } + contextChainMastershipWatcher.onNotAbleToStartMastership( + deviceInfo, + "Was not able to fill flow registry on device", + false); } } + }