openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceManagerImpl.java

   1 /**
   2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.openflowplugin.impl.device;
   9
  10 import javax.annotation.CheckForNull;
  11 import javax.annotation.Nonnull;
  12 import java.util.Collections;
  13 import java.util.Set;
  14 import java.util.concurrent.ExecutionException;
  15 import java.util.concurrent.ScheduledThreadPoolExecutor;
  16 import java.util.concurrent.TimeUnit;
  17
  18 import com.google.common.base.Preconditions;
  19 import com.google.common.collect.Sets;
  20 import io.netty.util.HashedWheelTimer;
  21 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  22 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  23 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  24 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  25 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  26 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  27 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
  28 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  29 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  30 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  31 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  32 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  33 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  34 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
  35 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
  36 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  37 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  38 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
  39 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
  45 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  46 import org.slf4j.Logger;
  47 import org.slf4j.LoggerFactory;
  48
  49 /**
  50  *
  51  */
  52 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper, AutoCloseable {
  53
  54     private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
  55
  56     private static final long TICK_DURATION = 10; // 0.5 sec.
  57     private final long globalNotificationQuota;
  58     private ScheduledThreadPoolExecutor spyPool;
  59     private final int spyRate = 10;
  60
  61     private final DataBroker dataBroker;
  62     private final HashedWheelTimer hashedWheelTimer;
  63     private TranslatorLibrary translatorLibrary;
  64     private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
  65     private NotificationService notificationService;
  66     private NotificationPublishService notificationPublishService;
  67
  68     private final Set<DeviceContext> deviceContexts = Sets.newConcurrentHashSet();
  69     private final MessageIntelligenceAgency messageIntelligenceAgency;
  70
  71     private final long barrierNanos = TimeUnit.MILLISECONDS.toNanos(500);
  72     private final int maxQueueDepth = 25600;
  73     private final boolean switchFeaturesMandatory;
  74     private final DeviceTransactionChainManagerProvider deviceTransactionChainManagerProvider;
  75     private ExtensionConverterProvider extensionConverterProvider;
  76
  77     public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
  78                              @Nonnull final MessageIntelligenceAgency messageIntelligenceAgency,
  79                              final boolean switchFeaturesMandatory,
  80                              final long globalNotificationQuota) {
  81         this.globalNotificationQuota = globalNotificationQuota;
  82         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  83         hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, 500);
  84         /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
  85         final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
  86
  87         final NodesBuilder nodesBuilder = new NodesBuilder();
  88         nodesBuilder.setNode(Collections.<Node>emptyList());
  89         tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
  90         try {
  91             tx.submit().get();
  92         } catch (ExecutionException | InterruptedException e) {
  93             LOG.error("Creation of node failed.", e);
  94             throw new IllegalStateException(e);
  95         }
  96
  97         this.messageIntelligenceAgency = messageIntelligenceAgency;
  98         this.switchFeaturesMandatory = switchFeaturesMandatory;
  99         deviceTransactionChainManagerProvider = new DeviceTransactionChainManagerProvider(dataBroker);
 100     }
 101
 102
 103     @Override
 104     public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
 105         deviceInitPhaseHandler = handler;
 106     }
 107
 108     @Override
 109     public void onDeviceContextLevelUp(final DeviceContext deviceContext) {
 110         // final phase - we have to add new Device to MD-SAL DataStore
 111         Preconditions.checkNotNull(deviceContext);
 112         try {
 113
 114             if (deviceContext.getDeviceState().getRole() != OfpRole.BECOMESLAVE) {
 115                 ((DeviceContextImpl) deviceContext).initialSubmitTransaction();
 116                 deviceContext.onPublished();
 117
 118             } else {
 119                 //if role = slave
 120                 try {
 121                     ((DeviceContextImpl) deviceContext).cancelTransaction();
 122                 } catch (final Exception e) {
 123                     //TODO: how can we avoid it. pingpong does not have cancel
 124                     LOG.debug("Expected Exception: Cancel Txn exception thrown for slaves", e);
 125                 }
 126
 127             }
 128
 129         } catch (final Exception e) {
 130             LOG.warn("Node {} can not be add to OPERATIONAL DataStore yet because {} ", deviceContext.getDeviceState().getNodeId(), e.getMessage());
 131             LOG.trace("Problem with add node {} to OPERATIONAL DataStore", deviceContext.getDeviceState().getNodeId(), e);
 132             try {
 133                 deviceContext.close();
 134             } catch (final Exception e1) {
 135                 LOG.warn("Device context close FAIL - " + deviceContext.getDeviceState().getNodeId());
 136             }
 137         }
 138     }
 139
 140     @Override
 141     public void deviceConnected(@CheckForNull final ConnectionContext connectionContext) {
 142         Preconditions.checkArgument(connectionContext != null);
 143         try {
 144             initializeDeviceContext(connectionContext);
 145         } catch (Exception e) {
 146             LOG.warn("Exception during initialization phase.", e);
 147         }
 148     }
 149
 150     private void initializeDeviceContext(final ConnectionContext connectionContext) throws Exception{
 151         LOG.info("Initializing New Connection DeviceContext for node:{}",  connectionContext.getNodeId());
 152         // Cache this for clarity
 153         final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
 154
 155         //FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
 156         connectionAdapter.setPacketInFiltering(true);
 157
 158         final Short version = connectionContext.getFeatures().getVersion();
 159         final OutboundQueueProvider outboundQueueProvider = new OutboundQueueProviderImpl(version);
 160
 161         connectionContext.setOutboundQueueProvider(outboundQueueProvider);
 162         final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
 163                 connectionAdapter.registerOutboundQueueHandler(outboundQueueProvider, maxQueueDepth, barrierNanos);
 164         connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
 165
 166         final NodeId nodeId = connectionContext.getNodeId();
 167         final DeviceState deviceState = new DeviceStateImpl(connectionContext.getFeatures(), nodeId);
 168
 169         final DeviceContext deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker,
 170                 hashedWheelTimer, messageIntelligenceAgency, outboundQueueProvider, translatorLibrary);
 171
 172         deviceContext.addDeviceContextClosedHandler(this);
 173         // We would like to crete/register TxChainManager after
 174         final DeviceTransactionChainManagerProvider.TransactionChainManagerRegistration txChainManagerReg = deviceTransactionChainManagerProvider
 175                 .provideTransactionChainManager(connectionContext);
 176         if (txChainManagerReg.ownedByInvokingConnectionContext()) {
 177             //this actually is new registration for currently processed connection context
 178             ((DeviceContextImpl) deviceContext).setTransactionChainManager(txChainManagerReg.getTransactionChainManager());
 179         } else {
 180             LOG.info("In deviceConnected {}, ownedByInvokingConnectionContext is false", connectionContext.getNodeId());
 181             deviceContext.close();
 182             return;
 183         }
 184         ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
 185         deviceContext.setNotificationService(notificationService);
 186         deviceContext.setNotificationPublishService(notificationPublishService);
 187
 188         deviceContexts.add(deviceContext);
 189
 190         updatePacketInRateLimiters();
 191
 192         final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
 193                 connectionAdapter, deviceContext);
 194         connectionAdapter.setMessageListener(messageListener);
 195
 196         deviceCtxLevelUp(deviceContext);
 197     }
 198
 199     private void updatePacketInRateLimiters() {
 200         synchronized (deviceContexts) {
 201             final int deviceContextsSize = deviceContexts.size();
 202             if (deviceContextsSize > 0) {
 203                 long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
 204                 if (freshNotificationLimit < 100) {
 205                     freshNotificationLimit = 100;
 206                 }
 207                 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
 208                 for (DeviceContext deviceContext : deviceContexts) {
 209                     deviceContext.updatePacketInRateLimit(freshNotificationLimit);
 210                 }
 211             }
 212         }
 213     }
 214
 215     void deviceCtxLevelUp(final DeviceContext deviceContext) {
 216         deviceContext.getDeviceState().setValid(true);
 217         LOG.trace("Device context level up called.");
 218         deviceInitPhaseHandler.onDeviceContextLevelUp(deviceContext);
 219     }
 220
 221     @Override
 222     public TranslatorLibrary oook() {
 223         return translatorLibrary;
 224     }
 225
 226     @Override
 227     public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
 228         this.translatorLibrary = translatorLibrary;
 229     }
 230
 231     @Override
 232     public void setNotificationService(final NotificationService notificationServiceParam) {
 233         notificationService = notificationServiceParam;
 234     }
 235
 236     @Override
 237     public void setNotificationPublishService(final NotificationPublishService notificationService) {
 238         notificationPublishService = notificationService;
 239     }
 240
 241     @Override
 242     public void close() throws Exception {
 243         for (final DeviceContext deviceContext : deviceContexts) {
 244             deviceContext.close();
 245         }
 246     }
 247
 248     @Override
 249     public void onDeviceContextClosed(final DeviceContext deviceContext) {
 250         deviceContexts.remove(deviceContext);
 251         updatePacketInRateLimiters();
 252     }
 253
 254     @Override
 255     public void initialize() {
 256         spyPool = new ScheduledThreadPoolExecutor(1);
 257         spyPool.scheduleAtFixedRate(messageIntelligenceAgency, spyRate, spyRate, TimeUnit.SECONDS);
 258     }
 259
 260     @Override
 261     public void setExtensionConverterProvider(ExtensionConverterProvider extensionConverterProvider) {
 262         this.extensionConverterProvider = extensionConverterProvider;
 263     }
 264
 265     @Override
 266     public ExtensionConverterProvider getExtensionConverterProvider() {
 267         return extensionConverterProvider;
 268     }
 269 }