openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/DeviceContextImpl.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 com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.base.Function;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.base.Verify;
  14 import com.google.common.util.concurrent.AsyncFunction;
  15 import com.google.common.util.concurrent.CheckedFuture;
  16 import com.google.common.util.concurrent.FutureCallback;
  17 import com.google.common.util.concurrent.Futures;
  18 import com.google.common.util.concurrent.JdkFutureAdapters;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import io.netty.util.HashedWheelTimer;
  21 import io.netty.util.Timeout;
  22 import io.netty.util.TimerTask;
  23 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  24 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  25 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  27 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  28 import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
  29 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  30 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  31 import org.opendaylight.openflowplugin.api.OFConstants;
  32 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
  36 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  37 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  38 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  39 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  40 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  41 import org.opendaylight.openflowplugin.api.openflow.device.handlers.ClusterInitializationPhaseHandler;
  42 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  43 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  44 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  45 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  46 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  47 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  48 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  49 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  50 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  51 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  52 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  53 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  54 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  55 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  56 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  57 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  58 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  59 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  60 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  61 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  62 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  63 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  64 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  65 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  66 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  67 import org.opendaylight.openflowplugin.impl.rpc.AbstractRequestContext;
  68 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
  69 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
  70 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  71 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  72 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  73 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  74 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  75 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  76 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  77 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  78 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  79 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SalRoleService;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInput;
 103 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleInputBuilder;
 104 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.SetRoleOutput;
 105 import org.opendaylight.yangtools.yang.binding.DataObject;
 106 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 107 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 108 import org.opendaylight.yangtools.yang.common.RpcResult;
 109 import org.slf4j.Logger;
 110 import org.slf4j.LoggerFactory;
 111
 112 import javax.annotation.Nonnull;
 113 import javax.annotation.Nullable;
 114 import java.math.BigInteger;
 115 import java.util.HashMap;
 116 import java.util.List;
 117 import java.util.Map;
 118 import java.util.concurrent.ExecutionException;
 119 import java.util.concurrent.Future;
 120 import java.util.concurrent.TimeUnit;
 121
 122 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
 123
 124     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 125
 126     // TODO: drain factor should be parametrized
 127     private static final float REJECTED_DRAIN_FACTOR = 0.25f;
 128     // TODO: low water mark factor should be parametrized
 129     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 130     // TODO: high water mark factor should be parametrized
 131     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 132
 133     // Timeout in seconds after what we will give up on propagating role
 134     private static final int SET_ROLE_TIMEOUT = 10;
 135
 136     private boolean initialized;
 137     private static final Long RETRY_DELAY = 100L;
 138     private static final int RETRY_COUNT = 3;
 139
 140     private SalRoleService salRoleService = null;
 141     private final HashedWheelTimer hashedWheelTimer;
 142     private ConnectionContext primaryConnectionContext;
 143     private final DeviceState deviceState;
 144     private final DataBroker dataBroker;
 145     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 146     private TransactionChainManager transactionChainManager;
 147     private DeviceFlowRegistry deviceFlowRegistry;
 148     private DeviceGroupRegistry deviceGroupRegistry;
 149     private DeviceMeterRegistry deviceMeterRegistry;
 150     private final PacketInRateLimiter packetInLimiter;
 151     private final MessageSpy messageSpy;
 152     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 153     private NotificationPublishService notificationPublishService;
 154     private Timeout barrierTaskTimeout;
 155     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 156     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 157     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 158     private final TranslatorLibrary translatorLibrary;
 159     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 160     private ExtensionConverterProvider extensionConverterProvider;
 161     private final DeviceManager deviceManager;
 162     private boolean skipTableFeatures;
 163     private boolean switchFeaturesMandatory;
 164     private final DeviceInfo deviceInfo;
 165     private final ConvertorExecutor convertorExecutor;
 166     private volatile CONTEXT_STATE state;
 167     private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
 168     private final DeviceManager myManager;
 169     private Boolean isAddNotificationSent = false;
 170
 171     DeviceContextImpl(
 172             @Nonnull final ConnectionContext primaryConnectionContext,
 173             @Nonnull final DataBroker dataBroker,
 174             @Nonnull final MessageSpy messageSpy,
 175             @Nonnull final TranslatorLibrary translatorLibrary,
 176             @Nonnull final DeviceManager manager,
 177             final ConvertorExecutor convertorExecutor,
 178             final boolean skipTableFeatures,
 179             final HashedWheelTimer hashedWheelTimer,
 180             final DeviceManager myManager) {
 181         this.primaryConnectionContext = primaryConnectionContext;
 182         this.deviceInfo = primaryConnectionContext.getDeviceInfo();
 183         this.hashedWheelTimer = hashedWheelTimer;
 184         this.myManager = myManager;
 185         this.deviceState = new DeviceStateImpl();
 186         this.dataBroker = dataBroker;
 187         this.auxiliaryConnectionContexts = new HashMap<>();
 188         this.messageSpy = Preconditions.checkNotNull(messageSpy);
 189         this.deviceManager = manager;
 190
 191         this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 192                 /*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR);
 193
 194         this.translatorLibrary = translatorLibrary;
 195         this.portStatusTranslator = translatorLibrary.lookupTranslator(
 196                 new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName()));
 197         this.packetInTranslator = translatorLibrary.lookupTranslator(
 198                 new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
 199         this.flowRemovedTranslator = translatorLibrary.lookupTranslator(
 200                 new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
 201
 202         this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 203         this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 204         this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 205         this.state = CONTEXT_STATE.INITIALIZATION;
 206         this.convertorExecutor = convertorExecutor;
 207         this.skipTableFeatures = skipTableFeatures;
 208         this.initialized = false;
 209     }
 210
 211     @Override
 212     public void initialSubmitTransaction() {
 213         if (initialized) {
 214             transactionChainManager.initialSubmitWriteTransaction();
 215         }
 216     }
 217
 218     @Override
 219     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 220         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 221         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 222     }
 223
 224     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 225         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 226     }
 227
 228     @Override
 229     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 230         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 231         LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
 232                 .getRemoteAddress(), getDeviceInfo().getLOGValue());
 233         auxiliaryConnectionContexts.remove(connectionDistinguisher);
 234     }
 235
 236     @Override
 237     public DeviceState getDeviceState() {
 238         return deviceState;
 239     }
 240
 241     @Override
 242     public ReadOnlyTransaction getReadTransaction() {
 243         return dataBroker.newReadOnlyTransaction();
 244     }
 245
 246     @Override
 247     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 248                                                           final InstanceIdentifier<T> path,
 249                                                           final T data){
 250         if (initialized) {
 251             transactionChainManager.writeToTransaction(store, path, data, false);
 252         }
 253     }
 254
 255     @Override
 256     public <T extends DataObject> void writeToTransactionWithParentsSlow(final LogicalDatastoreType store,
 257                                                                          final InstanceIdentifier<T> path,
 258                                                                          final T data){
 259         if (initialized) {
 260             transactionChainManager.writeToTransaction(store, path, data, true);
 261         }
 262     }
 263
 264     @Override
 265     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws TransactionChainClosedException {
 266         if (initialized) {
 267             transactionChainManager.addDeleteOperationTotTxChain(store, path);
 268         }
 269     }
 270
 271     @Override
 272     public boolean submitTransaction() {
 273         return initialized && transactionChainManager.submitWriteTransaction();
 274     }
 275
 276     @Override
 277     public ConnectionContext getPrimaryConnectionContext() {
 278         return primaryConnectionContext;
 279     }
 280
 281     @Override
 282     public ConnectionContext getAuxiliaryConnectionContexts(final BigInteger cookie) {
 283         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 284     }
 285
 286     @Override
 287     public DeviceFlowRegistry getDeviceFlowRegistry() {
 288         return deviceFlowRegistry;
 289     }
 290
 291     @Override
 292     public DeviceGroupRegistry getDeviceGroupRegistry() {
 293         return deviceGroupRegistry;
 294     }
 295
 296     @Override
 297     public DeviceMeterRegistry getDeviceMeterRegistry() {
 298         return deviceMeterRegistry;
 299     }
 300
 301     @Override
 302     public void processReply(final OfHeader ofHeader) {
 303         if (ofHeader instanceof Error) {
 304             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 305         } else {
 306             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 307         }
 308     }
 309
 310     @Override
 311     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 312         for (final MultipartReply multipartReply : ofHeaderList) {
 313             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 314         }
 315     }
 316
 317     @Override
 318     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 319         //1. translate to general flow (table, priority, match, cookie)
 320         final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 321                 flowRemovedTranslator.translate(flowRemoved, deviceInfo, null);
 322
 323         if(deviceManager.isFlowRemovedNotificationOn()) {
 324             // Trigger off a notification
 325             notificationPublishService.offerNotification(flowRemovedNotification);
 326         } else if(LOG.isDebugEnabled()) {
 327             LOG.debug("For nodeId={} isFlowRemovedNotificationOn={}", getDeviceInfo().getLOGValue(), deviceManager.isFlowRemovedNotificationOn());
 328         }
 329
 330         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 331         if (itemLifecycleListener != null) {
 332             //2. create registry key
 333             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(getDeviceInfo().getVersion(), flowRemovedNotification);
 334             //3. lookup flowId
 335             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 336             //4. if flowId present:
 337             if (flowDescriptor != null) {
 338                 // a) construct flow path
 339                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceInfo().getNodeInstanceIdentifier()
 340                         .augmentation(FlowCapableNode.class)
 341                         .child(Table.class, flowDescriptor.getTableKey())
 342                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 343                 // b) notify listener
 344                 itemLifecycleListener.onRemoved(flowPath);
 345             } else {
 346                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 347                         getDeviceInfo().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 348             }
 349         }
 350     }
 351
 352     @Override
 353     public void sendNodeAddedNotification() {
 354         if (!isAddNotificationSent) {
 355             isAddNotificationSent = true;
 356             NodeUpdatedBuilder builder = new NodeUpdatedBuilder();
 357             builder.setId(getDeviceInfo().getNodeId());
 358             builder.setNodeRef(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()));
 359             LOG.debug("Publishing node added notification for {}", builder.build());
 360             notificationPublishService.offerNotification(builder.build());
 361         }
 362     }
 363
 364     @Override
 365     public void sendNodeRemovedNotification() {
 366         NodeRemovedBuilder builder = new NodeRemovedBuilder();
 367         builder.setNodeRef(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()));
 368         LOG.debug("Publishing node removed notification for {}", builder.build());
 369         notificationPublishService.offerNotification(builder.build());
 370     }
 371
 372     @Override
 373     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 374         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 375         try {
 376             updatePortStatusMessage(portStatus);
 377         } catch (final Exception e) {
 378             LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo().toString(),
 379                     getDeviceInfo().getNodeId().toString(), e);
 380             retryProcessPortStatusMessage(portStatus,RETRY_COUNT);
 381         }
 382     }
 383
 384     private void retryProcessPortStatusMessage(final PortStatusMessage portStatus, final int retryCount){
 385         Thread thread = new Thread(new PortStatusRetryRunnable(portStatus, retryCount));
 386         thread.start();
 387     }
 388
 389     private class PortStatusRetryRunnable implements Runnable {
 390         private final PortStatusMessage portStatusMessage;
 391         private final int retryCount;
 392         private int count ;
 393
 394         PortStatusRetryRunnable (final PortStatusMessage portStatus,final int retry){
 395             portStatusMessage = portStatus;
 396             retryCount = retry;
 397             count = 0;
 398         }
 399         @Override
 400         public void run() {
 401             while(count < retryCount) {
 402                 try {
 403                     TimeUnit.MILLISECONDS.sleep(RETRY_DELAY);
 404                 } catch (InterruptedException e) {
 405                     LOG.warn("Caught interrupted exception while sleeping during a port retry task");
 406                 }
 407
 408                 try {
 409                     updatePortStatusMessage(portStatusMessage);
 410                     break;
 411                 } catch (final Exception e) {
 412                     LOG.warn("Error processing port status message for {} on port {} on device {} : {}",count,
 413                             portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString(), e);
 414                     count ++ ;
 415                 }
 416             }
 417             LOG.warn("Failed to update port status for port {} on device {} even after 3 retries",
 418                     portStatusMessage.getPortNo().toString(),getDeviceInfo().getNodeId().toString());
 419         }
 420     }
 421
 422     void updatePortStatusMessage(final PortStatusMessage portStatusMessage){
 423         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatusMessage,
 424                 getDeviceInfo(), null);
 425
 426         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector =
 427                 provideIIToNodeConnector(portStatusMessage.getPortNo(), portStatusMessage.getVersion());
 428
 429         if (portStatusMessage.getReason().equals(PortReason.OFPPRADD) || portStatusMessage.getReason().
 430                 equals(PortReason.OFPPRMODIFY)) {
 431             // because of ADD status node connector has to be created
 432             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 433             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class,
 434                     new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 435             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 436             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 437         } else if (portStatusMessage.getReason().equals(PortReason.OFPPRDELETE)) {
 438             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 439         }
 440         submitTransaction();
 441     }
 442
 443     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 444         final InstanceIdentifier<Node> iiToNodes = getDeviceInfo().getNodeInstanceIdentifier();
 445         final BigInteger dataPathId = getDeviceInfo().getDatapathId();
 446         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 447         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 448     }
 449
 450     @Override
 451     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 452         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 453         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 454         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, getDeviceInfo(), null);
 455
 456         if (packetReceived == null) {
 457             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 458             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 459             return;
 460         } else {
 461             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 462         }
 463
 464         if (!packetInLimiter.acquirePermit()) {
 465             LOG.debug("Packet limited");
 466             // TODO: save packet into emergency slot if possible
 467             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 468             return;
 469         }
 470
 471         final ListenableFuture<?> offerNotification = notificationPublishService.offerNotification(packetReceived);
 472         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 473             LOG.debug("notification offer rejected");
 474             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 475             packetInLimiter.drainLowWaterMark();
 476             packetInLimiter.releasePermit();
 477             return;
 478         }
 479
 480         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 481             @Override
 482             public void onSuccess(final Object result) {
 483                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 484                 packetInLimiter.releasePermit();
 485             }
 486
 487             @Override
 488             public void onFailure(final Throwable t) {
 489                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 490                 LOG.debug("notification offer failed: {}", t.getMessage());
 491                 LOG.trace("notification offer failed..", t);
 492                 packetInLimiter.releasePermit();
 493             }
 494         });
 495     }
 496
 497     @Override
 498     public void processExperimenterMessage(final ExperimenterMessage notification) {
 499         // lookup converter
 500         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 501         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 502                 getDeviceInfo().getVersion(),
 503                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 504         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 505         if (messageConverter == null) {
 506             LOG.warn("custom converter for {}[OF:{}] not found",
 507                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 508                     getDeviceInfo().getVersion());
 509             return;
 510         }
 511         // build notification
 512         final ExperimenterMessageOfChoice messageOfChoice;
 513         try {
 514             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 515             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 516                     .setNode(new NodeRef(getDeviceInfo().getNodeInstanceIdentifier()))
 517                     .setExperimenterMessageOfChoice(messageOfChoice);
 518             // publish
 519             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 520         } catch (final ConversionException e) {
 521             LOG.error("Conversion of experimenter notification failed", e);
 522         }
 523     }
 524
 525     @Override
 526     public TranslatorLibrary oook() {
 527         return translatorLibrary;
 528     }
 529
 530     @Override
 531     public void setCurrentBarrierTimeout(final Timeout timeout) {
 532         barrierTaskTimeout = timeout;
 533     }
 534
 535     @Override
 536     public Timeout getBarrierTaskTimeout() {
 537         return barrierTaskTimeout;
 538     }
 539
 540     @Override
 541     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 542         this.notificationPublishService = notificationPublishService;
 543     }
 544
 545     @Override
 546     public MessageSpy getMessageSpy() {
 547         return messageSpy;
 548     }
 549
 550     @Override
 551     public void onPublished() {
 552         Verify.verify(CONTEXT_STATE.INITIALIZATION.equals(getState()));
 553         this.state = CONTEXT_STATE.WORKING;
 554         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 555         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 556             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 557         }
 558     }
 559
 560     @Override
 561     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 562         return new MultiMsgCollectorImpl(this, requestContext);
 563     }
 564
 565     @Override
 566     public void updatePacketInRateLimit(final long upperBound) {
 567         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 568     }
 569
 570     @Override
 571     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 572         return itemLifeCycleSourceRegistry;
 573     }
 574
 575     @Override
 576     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 577         this.extensionConverterProvider = extensionConverterProvider;
 578     }
 579
 580     @Override
 581     public ExtensionConverterProvider getExtensionConverterProvider() {
 582         return extensionConverterProvider;
 583     }
 584
 585     @Override
 586     public synchronized void shutdownConnection() {
 587         if (LOG.isDebugEnabled()) {
 588             LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue());
 589         }
 590         if (CONTEXT_STATE.TERMINATION.equals(getState())) {
 591             LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue());
 592             return;
 593         }
 594
 595         if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 596             LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getLOGValue());
 597             return;
 598         }
 599
 600         // Terminate Auxiliary Connection
 601         for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 602             LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
 603             connectionContext.closeConnection(false);
 604         }
 605
 606         // Terminate Primary Connection
 607         getPrimaryConnectionContext().closeConnection(true);
 608
 609         // Close all datastore registries
 610         if (initialized) {
 611             deviceGroupRegistry.close();
 612             deviceFlowRegistry.close();
 613             deviceMeterRegistry.close();
 614         }
 615     }
 616
 617     @Override
 618     public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
 619         return initialized
 620                 ? this.transactionChainManager.shuttingDown()
 621                 : Futures.immediateFuture(null);
 622     }
 623
 624     @VisibleForTesting
 625     TransactionChainManager getTransactionChainManager() {
 626         return this.transactionChainManager;
 627     }
 628
 629     @Override
 630     public void setSwitchFeaturesMandatory(boolean switchFeaturesMandatory) {
 631         this.switchFeaturesMandatory = switchFeaturesMandatory;
 632     }
 633
 634     @Override
 635     public CONTEXT_STATE getState() {
 636         return this.state;
 637     }
 638
 639     @Override
 640     public ListenableFuture<Void> stopClusterServices() {
 641         final ListenableFuture<Void> deactivateTxManagerFuture = initialized
 642                 ? transactionChainManager.deactivateTransactionManager()
 643                 : Futures.immediateFuture(null);
 644
 645         final boolean connectionInterrupted =
 646                 this.getPrimaryConnectionContext()
 647                         .getConnectionState()
 648                         .equals(ConnectionContext.CONNECTION_STATE.RIP);
 649         if (!connectionInterrupted) {
 650             LOG.info("This controller instance is now acting as a non-owner for node {}", deviceInfo.getLOGValue());
 651         }
 652
 653         return deactivateTxManagerFuture;
 654     }
 655
 656     @Override
 657     public void cleanupDeviceData() {
 658         myManager.removeDeviceFromOperationalDS(deviceInfo);
 659     }
 660
 661     @Override
 662     public ServiceGroupIdentifier getServiceIdentifier() {
 663         return this.deviceInfo.getServiceIdentifier();
 664     }
 665
 666     @Override
 667     public DeviceInfo getDeviceInfo() {
 668         return this.deviceInfo;
 669     }
 670
 671     @Override
 672     public void close() {
 673         if (CONTEXT_STATE.TERMINATION.equals(getState())){
 674             if (LOG.isDebugEnabled()) {
 675                 LOG.debug("DeviceContext for node {} is already in TERMINATION state.", getDeviceInfo().getLOGValue());
 676             }
 677         } else {
 678             this.state = CONTEXT_STATE.TERMINATION;
 679         }
 680         sendNodeRemovedNotification();
 681     }
 682
 683     @Override
 684     public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){
 685         if (initialized) {
 686             this.transactionChainManager.setLifecycleService(lifecycleService);
 687         }
 688     }
 689
 690     @Override
 691     public void replaceConnectionContext(final ConnectionContext connectionContext){
 692         // Act like we are initializing the context
 693         this.state = CONTEXT_STATE.INITIALIZATION;
 694         this.primaryConnectionContext = connectionContext;
 695         this.onPublished();
 696     }
 697
 698     @Override
 699     public boolean isSkipTableFeatures() {
 700         return this.skipTableFeatures;
 701     }
 702
 703     @Override
 704     public void setSalRoleService(@Nonnull SalRoleService salRoleService) {
 705         this.salRoleService = salRoleService;
 706     }
 707
 708     @Override
 709     public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
 710         this.clusterInitializationPhaseHandler = handler;
 711     }
 712
 713     @Override
 714     public boolean onContextInstantiateService(final ConnectionContext connectionContext) {
 715
 716         if (getPrimaryConnectionContext().getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) {
 717             LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue());
 718             return false;
 719         }
 720
 721         LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue());
 722
 723         lazyTransactionManagerInitialization();
 724
 725         this.transactionChainManager.activateTransactionManager();
 726
 727         try {
 728             DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor);
 729         } catch (ExecutionException | InterruptedException e) {
 730             LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), e);
 731             return false;
 732         }
 733
 734         Futures.addCallback(sendRoleChangeToDevice(OfpRole.BECOMEMASTER), new RpcResultFutureCallback());
 735
 736         return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext());
 737     }
 738
 739     @VisibleForTesting
 740     void lazyTransactionManagerInitialization() {
 741         if (!this.initialized) {
 742             if (LOG.isDebugEnabled()) {
 743                 LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue());
 744             }
 745             this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo);
 746             this.deviceFlowRegistry = new DeviceFlowRegistryImpl(deviceInfo.getVersion(), dataBroker, deviceInfo.getNodeInstanceIdentifier());
 747             this.deviceGroupRegistry = new DeviceGroupRegistryImpl();
 748             this.deviceMeterRegistry = new DeviceMeterRegistryImpl();
 749             this.initialized = true;
 750         }
 751     }
 752
 753     @Nullable
 754     @Override
 755     public <T> RequestContext<T> createRequestContext() {
 756         return new AbstractRequestContext<T>(deviceInfo.reserveXidForDeviceMessage()) {
 757             @Override
 758             public void close() {
 759             }
 760         };
 761
 762     }
 763
 764     ListenableFuture<RpcResult<SetRoleOutput>> sendRoleChangeToDevice(final OfpRole newRole) {
 765         if (LOG.isDebugEnabled()) {
 766             LOG.debug("Sending new role {} to device {}", newRole, deviceInfo.getNodeId());
 767         }
 768
 769         final Future<RpcResult<SetRoleOutput>> setRoleOutputFuture;
 770
 771         if (deviceInfo.getVersion() >= OFConstants.OFP_VERSION_1_3) {
 772             final SetRoleInput setRoleInput = (new SetRoleInputBuilder()).setControllerRole(newRole)
 773                     .setNode(new NodeRef(deviceInfo.getNodeInstanceIdentifier())).build();
 774
 775             setRoleOutputFuture = this.salRoleService.setRole(setRoleInput);
 776
 777             final TimerTask timerTask = timeout -> {
 778                 if (!setRoleOutputFuture.isDone()) {
 779                     LOG.warn("New role {} was not propagated to device {} during {} sec", newRole, deviceInfo.getLOGValue(), SET_ROLE_TIMEOUT);
 780                     setRoleOutputFuture.cancel(true);
 781                 }
 782             };
 783
 784             hashedWheelTimer.newTimeout(timerTask, SET_ROLE_TIMEOUT, TimeUnit.SECONDS);
 785         } else {
 786             LOG.info("Device: {} with version: {} does not support role", deviceInfo.getLOGValue(), deviceInfo.getVersion());
 787             return Futures.immediateFuture(null);
 788         }
 789
 790         return JdkFutureAdapters.listenInPoolThread(setRoleOutputFuture);
 791     }
 792
 793     @Override
 794     public ListenableFuture<RpcResult<SetRoleOutput>> makeDeviceSlave() {
 795         return sendRoleChangeToDevice(OfpRole.BECOMESLAVE);
 796     }
 797
 798     private class RpcResultFutureCallback implements FutureCallback<RpcResult<SetRoleOutput>> {
 799         @Override
 800         public void onSuccess(@Nullable RpcResult<SetRoleOutput> setRoleOutputRpcResult) {
 801             if (LOG.isDebugEnabled()) {
 802                 LOG.debug("Role MASTER was successfully set on device, node {}", deviceInfo.getLOGValue());
 803             }
 804             sendNodeAddedNotification();
 805         }
 806
 807         @Override
 808         public void onFailure(final Throwable throwable) {
 809             LOG.warn("Was not able to set MASTER role on device, node {}", deviceInfo.getLOGValue());
 810             shutdownConnection();
 811         }
 812     }
 813 }