-/**
+/*
* Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
*
* This program and the accompanying materials are made available under the
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
+import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
import io.netty.util.HashedWheelTimer;
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.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
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;
-import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
+import org.opendaylight.mdsal.binding.api.DataBroker;
+import org.opendaylight.mdsal.binding.api.NotificationPublishService;
+import org.opendaylight.mdsal.binding.api.ReadTransaction;
+import org.opendaylight.mdsal.common.api.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.openflow.device.TranslatorLibrary;
import org.opendaylight.openflowplugin.api.openflow.device.Xid;
import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChain;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipState;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainMastershipWatcher;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainState;
private final HashedWheelTimer hashedWheelTimer;
private final DeviceState deviceState;
private final DataBroker dataBroker;
- private final Collection<RequestContext<?>> requestContexts = new HashSet<>();
+ private final Collection<RequestContext<?>> requestContexts = ConcurrentHashMap.newKeySet();
private final MessageSpy messageSpy;
private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
private final AtomicBoolean initialized = new AtomicBoolean(false);
private final AtomicBoolean hasState = new AtomicBoolean(false);
private final AtomicBoolean isInitialTransactionSubmitted = new AtomicBoolean(false);
+ private final ContextChainHolder contextChainHolder;
private NotificationPublishService notificationPublishService;
private TransactionChainManager transactionChainManager;
private DeviceFlowRegistry deviceFlowRegistry;
final boolean useSingleLayerSerialization,
final DeviceInitializerProvider deviceInitializerProvider,
final boolean isFlowRemovedNotificationOn,
- final boolean switchFeaturesMandatory) {
+ final boolean switchFeaturesMandatory,
+ final ContextChainHolder contextChainHolder) {
this.primaryConnectionContext = primaryConnectionContext;
this.deviceInfo = primaryConnectionContext.getDeviceInfo();
this.deviceState = new DeviceStateImpl();
this.dataBroker = dataBroker;
this.messageSpy = messageSpy;
+ this.contextChainHolder = contextChainHolder;
this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
/*initial*/ LOW_WATERMARK, /*initial*/HIGH_WATERMARK, this.messageSpy, REJECTED_DRAIN_FACTOR);
}
@Override
- public ReadOnlyTransaction getReadTransaction() {
+ public ReadTransaction getReadTransaction() {
return dataBroker.newReadOnlyTransaction();
}
return initialized.get() && transactionChainManager.submitTransaction();
}
+ @Override
+ public boolean syncSubmitTransaction() {
+ return initialized.get() && transactionChainManager.submitTransaction(true);
+ }
+
@Override
public ConnectionContext getPrimaryConnectionContext() {
return primaryConnectionContext;
@Override
public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
- //1. translate to general flow (table, priority, match, cookie)
- final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
- flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
-
- if (isFlowRemovedNotificationOn) {
- // Trigger off a notification
- notificationPublishService.offerNotification(flowRemovedNotification);
+ if (isMasterOfDevice()) {
+ //1. translate to general flow (table, priority, match, cookie)
+ final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819
+ .FlowRemoved flowRemovedNotification = flowRemovedTranslator
+ .translate(flowRemoved, deviceInfo, null);
+
+ if (isFlowRemovedNotificationOn) {
+ // Trigger off a notification
+ notificationPublishService.offerNotification(flowRemovedNotification);
+ }
+ } else {
+ LOG.debug("Controller is not owner of the device {}, skipping Flow Removed message",
+ deviceInfo.getLOGValue());
}
}
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);
OpenflowVersion.get(deviceInfo.getVersion()))));
writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, new NodeConnectorBuilder()
- .setKey(iiToNodeConnector.getKey())
+ .withKey(iiToNodeConnector.getKey())
.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new
FlowCapableNodeConnectorStatisticsDataBuilder().build())
.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector)
.build());
- submitTransaction();
+ syncSubmitTransaction();
if (PortReason.OFPPRDELETE.equals(portStatusMessage.getReason())) {
addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
- submitTransaction();
+ syncSubmitTransaction();
}
}
@Override
public void processPacketInMessage(final PacketInMessage packetInMessage) {
- final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null);
- handlePacketInMessage(packetReceived, packetInMessage.getImplementedInterface(), packetReceived.getMatch());
+ if (isMasterOfDevice()) {
+ final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null);
+ handlePacketInMessage(packetReceived, packetInMessage.getImplementedInterface(), packetReceived.getMatch());
+ } else {
+ LOG.debug("Controller is not owner of the device {}, skipping packet_in message", deviceInfo.getLOGValue());
+ }
+ }
+
+ private Boolean isMasterOfDevice() {
+ final ContextChain contextChain = contextChainHolder.getContextChain(deviceInfo);
+ boolean result = false;
+ if (contextChain != null) {
+ result = contextChain.isMastered(ContextChainMastershipState.CHECK, false);
+ }
+ return result;
}
private void handlePacketInMessage(final PacketIn packetIn,
@Override
public void processExperimenterMessage(final ExperimenterMessage notification) {
- // lookup converter
- final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
- final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
- getDeviceInfo().getVersion(),
- (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
- final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter =
- extensionConverterProvider.getMessageConverter(key);
- if (messageConverter == null) {
- LOG.warn("custom converter for {}[OF:{}] not found",
- notification.getExperimenterDataOfChoice().getImplementedInterface(),
- getDeviceInfo().getVersion());
- return;
- }
- // build notification
- final ExperimenterMessageOfChoice messageOfChoice;
- try {
- messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
- final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new
- ExperimenterMessageFromDevBuilder()
- .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()))
- .setExperimenterMessageOfChoice(messageOfChoice);
- // publish
- notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
- } catch (final ConversionException e) {
- LOG.error("Conversion of experimenter notification failed", e);
+ if (isMasterOfDevice()) {
+ // lookup converter
+ final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
+ final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
+ getDeviceInfo().getVersion(),
+ (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
+ final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter =
+ extensionConverterProvider.getMessageConverter(key);
+ if (messageConverter == null) {
+ LOG.warn("custom converter for {}[OF:{}] not found",
+ notification.getExperimenterDataOfChoice().getImplementedInterface(),
+ getDeviceInfo().getVersion());
+ return;
+ }
+ // build notification
+ final ExperimenterMessageOfChoice messageOfChoice;
+ try {
+ messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
+ final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new
+ ExperimenterMessageFromDevBuilder()
+ .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()))
+ .setExperimenterMessageOfChoice(messageOfChoice);
+ // publish
+ notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
+ } catch (final ConversionException e) {
+ LOG.error("Conversion of experimenter notification failed", e);
+ }
+ } else {
+ LOG.debug("Controller is not owner of the device {}, skipping experimenter message",
+ deviceInfo.getLOGValue());
}
}
@Override
+ // The cast to PacketInMessage is safe as the implemented interface is verified before the cas tbut FB doesn't
+ // recognize it.
+ @SuppressFBWarnings("BC_UNCONFIRMED_CAST")
public boolean processAlienMessage(final OfHeader message) {
final Class<? extends DataContainer> implementedInterface = message.getImplementedInterface();
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);
+ .PacketInMessage packetInMessage = (org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service
+ .rev130709.PacketInMessage) message;
handlePacketInMessage(packetInMessage, implementedInterface, packetInMessage.getMatch());
return true;
}
@Override
- public ListenableFuture<Void> closeServiceInstance() {
- final ListenableFuture<Void> listenableFuture = initialized.get()
+ public ListenableFuture<?> closeServiceInstance() {
+ final ListenableFuture<?> listenableFuture = initialized.get()
? transactionChainManager.deactivateTransactionManager()
: Futures.immediateFuture(null);
deviceFlowRegistry.close();
deviceMeterRegistry.close();
- final ListenableFuture<Void> txChainShuttingDown = transactionChainManager.shuttingDown();
+ final ListenableFuture<?> txChainShuttingDown = transactionChainManager.shuttingDown();
- Futures.addCallback(txChainShuttingDown, new FutureCallback<Void>() {
+ Futures.addCallback(txChainShuttingDown, new FutureCallback<Object>() {
@Override
- public void onSuccess(@Nullable final Void result) {
+ public void onSuccess(final Object result) {
transactionChainManager.close();
transactionChainManager = null;
}
deviceInfo.toString()));
}
- final ListenableFuture<List<com.google.common.base.Optional<FlowCapableNode>>> deviceFlowRegistryFill =
+ final ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill =
getDeviceFlowRegistry().fill();
Futures.addCallback(deviceFlowRegistryFill,
new DeviceFlowRegistryCallback(deviceFlowRegistryFill, contextChainMastershipWatcher),
hasState.set(true);
}
- private class DeviceFlowRegistryCallback implements FutureCallback<List<com.google.common.base
- .Optional<FlowCapableNode>>> {
- private final ListenableFuture<List<com.google.common.base.Optional<FlowCapableNode>>> deviceFlowRegistryFill;
+ private class DeviceFlowRegistryCallback implements FutureCallback<List<Optional<FlowCapableNode>>> {
+ private final ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill;
private final ContextChainMastershipWatcher contextChainMastershipWatcher;
DeviceFlowRegistryCallback(
- ListenableFuture<List<com.google.common.base.Optional<FlowCapableNode>>> deviceFlowRegistryFill,
+ ListenableFuture<List<Optional<FlowCapableNode>>> deviceFlowRegistryFill,
ContextChainMastershipWatcher contextChainMastershipWatcher) {
this.deviceFlowRegistryFill = deviceFlowRegistryFill;
this.contextChainMastershipWatcher = contextChainMastershipWatcher;
}
@Override
- public void onSuccess(@Nullable List<com.google.common.base.Optional<FlowCapableNode>> result) {
+ public void onSuccess(@Nullable List<Optional<FlowCapableNode>> result) {
if (LOG.isDebugEnabled()) {
// Count all flows we read from datastore for debugging purposes.
// This number do not always represent how many flows were actually added
.stream()
.flatMap(Collection::stream)
.filter(Objects::nonNull)
- .flatMap(flowCapableNodeOptional -> flowCapableNodeOptional.asSet().stream())
+ .flatMap(flowCapableNodeOptional
+ -> com.google.common.base.Optional.fromJavaUtil(flowCapableNodeOptional).asSet().stream())
.filter(Objects::nonNull)
.filter(flowCapableNode -> Objects.nonNull(flowCapableNode.getTable()))
.flatMap(flowCapableNode -> flowCapableNode.getTable().stream())
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);
+ LOG.warn("Failed filling flow registry with flows for node: {}", deviceInfo, throwable);
}
contextChainMastershipWatcher.onNotAbleToStartMastership(
deviceInfo,
Code Review / openflowplugin.git / blobdiff