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.Optional;
  13 import com.google.common.base.Preconditions;
  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.FutureFallback;
  18 import com.google.common.util.concurrent.Futures;
  19 import com.google.common.util.concurrent.ListenableFuture;
  20 import io.netty.util.HashedWheelTimer;
  21 import io.netty.util.Timeout;
  22 import java.math.BigInteger;
  23 import java.util.Collection;
  24 import java.util.Collections;
  25 import java.util.HashMap;
  26 import java.util.HashSet;
  27 import java.util.LinkedList;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.concurrent.ConcurrentHashMap;
  31 import javax.annotation.CheckForNull;
  32 import javax.annotation.Nonnull;
  33 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  34 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
  35 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
  36 import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
  37 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  38 import org.opendaylight.controller.md.sal.common.api.data.ReadFailedException;
  39 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  40 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
  41 import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
  42 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
  43 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
  44 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
  45 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
  46 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
  47 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
  48 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
  49 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  50 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceContextClosedHandler;
  51 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  52 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
  53 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
  54 import org.opendaylight.openflowplugin.api.openflow.registry.ItemLifeCycleRegistry;
  55 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
  56 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowDescriptor;
  57 import org.opendaylight.openflowplugin.api.openflow.registry.flow.FlowRegistryKey;
  58 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
  59 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
  60 import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleKeeper;
  61 import org.opendaylight.openflowplugin.api.openflow.rpc.RpcContext;
  62 import org.opendaylight.openflowplugin.api.openflow.rpc.listener.ItemLifecycleListener;
  63 import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext;
  64 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
  65 import org.opendaylight.openflowplugin.extension.api.ConvertorMessageFromOFJava;
  66 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
  67 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
  68 import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
  69 import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
  70 import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
  71 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
  72 import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
  73 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
  74 import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
  75 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
  76 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
  77 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
  78 import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
  79 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
  80 import org.opendaylight.yang.gen.v1.urn.opendaylight.experimenter.message.service.rev151020.ExperimenterMessageFromDevBuilder;
  81 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNode;
  82 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
  83 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.Table;
  84 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.Flow;
  85 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.tables.table.FlowKey;
  86 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
  87 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef;
  88 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
  89 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
  90 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
  91 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
  92 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  93 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
  94 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
  95 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  96 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
  97 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  98 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  99 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
 100 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
 101 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
 102 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
 103 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.experimenter.core.ExperimenterDataOfChoice;
 104 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflowplugin.experimenter.types.rev151020.experimenter.core.message.ExperimenterMessageOfChoice;
 105 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
 106 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
 107 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
 108 import org.opendaylight.yang.gen.v1.urn.opendaylight.role.service.rev150727.OfpRole;
 109 import org.opendaylight.yangtools.yang.binding.DataObject;
 110 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
 111 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
 112 import org.slf4j.Logger;
 113 import org.slf4j.LoggerFactory;
 114
 115 /**
 116  *
 117  */
 118 public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper {
 119
 120     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
 121
 122     // TODO: drain factor should be parametrized
 123     public static final float REJECTED_DRAIN_FACTOR = 0.25f;
 124     // TODO: low water mark factor should be parametrized
 125     private static final float LOW_WATERMARK_FACTOR = 0.75f;
 126     // TODO: high water mark factor should be parametrized
 127     private static final float HIGH_WATERMARK_FACTOR = 0.95f;
 128
 129     private final ConnectionContext primaryConnectionContext;
 130     private final DeviceState deviceState;
 131     private final DataBroker dataBroker;
 132     private final HashedWheelTimer hashedWheelTimer;
 133     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
 134     private final TransactionChainManager transactionChainManager;
 135     private final DeviceFlowRegistry deviceFlowRegistry;
 136     private final DeviceGroupRegistry deviceGroupRegistry;
 137     private final DeviceMeterRegistry deviceMeterRegistry;
 138     private final Collection<DeviceContextClosedHandler> closeHandlers = new HashSet<>();
 139     private final PacketInRateLimiter packetInLimiter;
 140     private final MessageSpy messageSpy;
 141     private final ItemLifeCycleKeeper flowLifeCycleKeeper;
 142     private NotificationPublishService notificationPublishService;
 143     private NotificationService notificationService;
 144     private final OutboundQueue outboundQueueProvider;
 145     private Timeout barrierTaskTimeout;
 146     private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
 147     private final MessageTranslator<PacketInMessage, PacketReceived> packetInTranslator;
 148     private final MessageTranslator<FlowRemoved, org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved> flowRemovedTranslator;
 149     private final TranslatorLibrary translatorLibrary;
 150     private final Map<Long, NodeConnectorRef> nodeConnectorCache;
 151     private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
 152     private RpcContext rpcContext;
 153     private ExtensionConverterProvider extensionConverterProvider;
 154
 155     private final boolean switchFeaturesMandatory;
 156     private StatisticsContext statisticsContext;
 157
 158
 159     @VisibleForTesting
 160     DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
 161                       @Nonnull final DeviceState deviceState,
 162                       @Nonnull final DataBroker dataBroker,
 163                       @Nonnull final HashedWheelTimer hashedWheelTimer,
 164                       @Nonnull final MessageSpy _messageSpy,
 165                       @Nonnull final OutboundQueueProvider outboundQueueProvider,
 166                       @Nonnull final TranslatorLibrary translatorLibrary,
 167                       final boolean switchFeaturesMandatory) {
 168         this.switchFeaturesMandatory = switchFeaturesMandatory;
 169         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
 170         this.deviceState = Preconditions.checkNotNull(deviceState);
 171         this.dataBroker = Preconditions.checkNotNull(dataBroker);
 172         this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
 173         this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
 174         primaryConnectionContext.setDeviceDisconnectedHandler(DeviceContextImpl.this);
 175         this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState);
 176         auxiliaryConnectionContexts = new HashMap<>();
 177         deviceFlowRegistry = new DeviceFlowRegistryImpl();
 178         deviceGroupRegistry = new DeviceGroupRegistryImpl();
 179         deviceMeterRegistry = new DeviceMeterRegistryImpl();
 180         messageSpy = _messageSpy;
 181
 182         packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
 183                 /*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
 184
 185         this.translatorLibrary = translatorLibrary;
 186         portStatusTranslator = translatorLibrary.lookupTranslator(
 187                 new TranslatorKey(deviceState.getVersion(), PortGrouping.class.getName()));
 188         packetInTranslator = translatorLibrary.lookupTranslator(
 189                 new TranslatorKey(deviceState.getVersion(), PacketIn.class.getName()));
 190         flowRemovedTranslator = translatorLibrary.lookupTranslator(
 191                 new TranslatorKey(deviceState.getVersion(), FlowRemoved.class.getName()));
 192
 193
 194         nodeConnectorCache = new ConcurrentHashMap<>();
 195
 196         itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
 197         flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
 198         itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
 199     }
 200
 201     /**
 202      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
 203      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
 204      */
 205     void initialSubmitTransaction() {
 206         transactionChainManager.initialSubmitWriteTransaction();
 207     }
 208
 209     @Override
 210     public Long reservedXidForDeviceMessage() {
 211         return outboundQueueProvider.reserveEntry();
 212     }
 213
 214     @Override
 215     public void addAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 216         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 217         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
 218     }
 219
 220     private static SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
 221         return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
 222     }
 223
 224     @Override
 225     public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
 226         final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 227         if (null != connectionDistinguisher) {
 228             auxiliaryConnectionContexts.remove(connectionDistinguisher);
 229         }
 230     }
 231
 232     @Override
 233     public DeviceState getDeviceState() {
 234         return deviceState;
 235     }
 236
 237     @Override
 238     public ReadOnlyTransaction getReadTransaction() {
 239         return dataBroker.newReadOnlyTransaction();
 240     }
 241
 242     @Override
 243     public ListenableFuture<Void> onClusterRoleChange(final OfpRole oldRole, @CheckForNull final OfpRole role) {
 244         LOG.trace("onClusterRoleChange {} for node:", role, deviceState.getNodeId());
 245         Preconditions.checkArgument(role != null);
 246         if (role.equals(oldRole)) {
 247             LOG.debug("Demanded role change for device {} is not change OldRole: {}, NewRole {}", deviceState.getNodeId(), oldRole, role);
 248             return Futures.immediateFuture(null);
 249         }
 250         if (OfpRole.BECOMEMASTER.equals(role)) {
 251             if (!deviceState.deviceSynchronized()) {
 252                 //TODO: no necessary code for yet - it needs for initialization phase only
 253                 LOG.debug("Setup Empty TxManager {} for initialization phase", getDeviceState().getNodeId());
 254                 transactionChainManager.activateTransactionManager();
 255                 return Futures.immediateCheckedFuture(null);
 256             }
 257             return asyncClusterRoleChange();
 258
 259         } else if (OfpRole.BECOMESLAVE.equals(role)) {
 260             if (null != rpcContext) {
 261                 MdSalRegistrationUtils.registerSlaveServices(rpcContext, role);
 262             }
 263             return transactionChainManager.deactivateTransactionManager();
 264         } else {
 265             LOG.warn("Unknown OFCluster Role {} for Node {}", role, deviceState.getNodeId());
 266             if (null != rpcContext) {
 267                 MdSalRegistrationUtils.unregisterServices(rpcContext);
 268             }
 269             return transactionChainManager.deactivateTransactionManager();
 270         }
 271     }
 272
 273     /*
 274      * we don't have active TxManager so anything will not be stored to DS yet, but we have
 275      * check all NodeInformation for statistics otherwise statistics will not contains
 276      * all possible MultipartTypes for polling in StatTypeList
 277      */
 278     private ListenableFuture<Void> asyncClusterRoleChange() {
 279         if (statisticsContext == null) {
 280             final String errMsg = String.format("DeviceCtx %s is up but we are missing StatisticsContext", deviceState.getNodeId());
 281             LOG.warn(errMsg);
 282             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 283         }
 284         if (rpcContext == null) {
 285             final String errMsg = String.format("DeviceCtx %s is up but we are missing RpcContext", deviceState.getNodeId());
 286             LOG.warn(errMsg);
 287             return Futures.immediateFailedFuture(new IllegalStateException(errMsg));
 288         }
 289
 290         final InstanceIdentifier<FlowCapableNode> ofNodeII = deviceState.getNodeInstanceIdentifier()
 291                 .augmentation(FlowCapableNode.class);
 292         final ReadOnlyTransaction readTx = getReadTransaction();
 293         final CheckedFuture<Optional<FlowCapableNode>, ReadFailedException> readOfNodeFuture = readTx.read(
 294                 LogicalDatastoreType.OPERATIONAL, ofNodeII);
 295
 296         final ListenableFuture<Void> nodeInitInfoFuture = Futures.transform(readOfNodeFuture,
 297                 new AsyncFunction<Optional<FlowCapableNode>, Void>() {
 298                     @Override
 299                     public ListenableFuture<Void> apply(final Optional<FlowCapableNode> input) throws Exception {
 300                         if (!input.isPresent() || input.get().getTable() == null || input.get().getTable().isEmpty()) {
 301                             /* Last master close fail scenario so we would like to activate TxManager */
 302                             LOG.debug("Operational DS for Device {} has to be replaced", deviceState.getNodeId());
 303                             getDeviceState().setDeviceSynchronized(false);
 304                             transactionChainManager.activateTransactionManager();
 305                         }
 306                         return DeviceInitializationUtils.initializeNodeInformation(DeviceContextImpl.this, switchFeaturesMandatory);
 307                     }
 308                 });
 309
 310         final ListenableFuture<Boolean> statPollFuture = Futures.transform(nodeInitInfoFuture,
 311                 new AsyncFunction<Void, Boolean>() {
 312
 313                     @Override
 314                     public ListenableFuture<Boolean> apply(final Void input) throws Exception {
 315                         getStatisticsContext().statListForCollectingInitialization();
 316                         if (getDeviceState().deviceSynchronized()) {
 317                             return Futures.immediateFuture(Boolean.TRUE);
 318                         }
 319                         return getStatisticsContext().gatherDynamicData();
 320                     }
 321                 });
 322
 323         return Futures.transform(statPollFuture, new Function<Boolean, Void>() {
 324
 325             @Override
 326             public Void apply(final Boolean input) {
 327                 if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
 328                     final String errMsg = String.format("We lost connection for Device %s, context has to be closed.",
 329                             getDeviceState().getNodeId());
 330                     LOG.warn(errMsg);
 331                     transactionChainManager.clearUnsubmittedTransaction();
 332                     throw new IllegalStateException(errMsg);
 333                 }
 334                 if (!input.booleanValue()) {
 335                     final String errMsg = String.format("Get Initial Device %s information fails",
 336                             getDeviceState().getNodeId());
 337                     LOG.warn(errMsg);
 338                     transactionChainManager.clearUnsubmittedTransaction();
 339                     throw new IllegalStateException(errMsg);
 340                 }
 341                 LOG.debug("Get Initial Device {} information is successful", getDeviceState().getNodeId());
 342                 getDeviceState().setDeviceSynchronized(true);
 343                 transactionChainManager.activateTransactionManager();
 344                 MdSalRegistrationUtils.registerMasterServices(getRpcContext(), DeviceContextImpl.this, OfpRole.BECOMEMASTER);
 345                 getRpcContext().registerStatCompatibilityServices();
 346                 initialSubmitTransaction();
 347                 getDeviceState().setStatisticsPollingEnabledProp(true);
 348                 return null;
 349             }
 350         });
 351     }
 352
 353     @Override
 354     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 355                                                           final InstanceIdentifier<T> path, final T data) {
 356         transactionChainManager.writeToTransaction(store, path, data);
 357     }
 358
 359     @Override
 360     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
 361         transactionChainManager.addDeleteOperationTotTxChain(store, path);
 362     }
 363
 364     @Override
 365     public boolean submitTransaction() {
 366         return transactionChainManager.submitWriteTransaction();
 367     }
 368
 369     @Override
 370     public ConnectionContext getPrimaryConnectionContext() {
 371         return primaryConnectionContext;
 372     }
 373
 374     @Override
 375     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
 376         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
 377     }
 378
 379     @Override
 380     public DeviceFlowRegistry getDeviceFlowRegistry() {
 381         return deviceFlowRegistry;
 382     }
 383
 384     @Override
 385     public DeviceGroupRegistry getDeviceGroupRegistry() {
 386         return deviceGroupRegistry;
 387     }
 388
 389     @Override
 390     public DeviceMeterRegistry getDeviceMeterRegistry() {
 391         return deviceMeterRegistry;
 392     }
 393
 394     @Override
 395     public void processReply(final OfHeader ofHeader) {
 396         if (ofHeader instanceof Error) {
 397             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 398         } else {
 399             messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 400         }
 401     }
 402
 403     @Override
 404     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
 405         for (final MultipartReply multipartReply : ofHeaderList) {
 406             messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
 407         }
 408     }
 409
 410     @Override
 411     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
 412         final ItemLifecycleListener itemLifecycleListener = flowLifeCycleKeeper.getItemLifecycleListener();
 413         if (itemLifecycleListener != null) {
 414             //1. translate to general flow (table, priority, match, cookie)
 415             final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
 416                     flowRemovedTranslator.translate(flowRemoved, this, null);
 417             //2. create registry key
 418             final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
 419             //3. lookup flowId
 420             final FlowDescriptor flowDescriptor = deviceFlowRegistry.retrieveIdForFlow(flowRegKey);
 421             //4. if flowId present:
 422             if (flowDescriptor != null) {
 423                 // a) construct flow path
 424                 final KeyedInstanceIdentifier<Flow, FlowKey> flowPath = getDeviceState().getNodeInstanceIdentifier()
 425                         .augmentation(FlowCapableNode.class)
 426                         .child(Table.class, flowDescriptor.getTableKey())
 427                         .child(Flow.class, new FlowKey(flowDescriptor.getFlowId()));
 428                 // b) notify listener
 429                 itemLifecycleListener.onRemoved(flowPath);
 430             } else {
 431                 LOG.debug("flow id not found: nodeId={} tableId={}, priority={}",
 432                         getDeviceState().getNodeId(), flowRegKey.getTableId(), flowRemovedNotification.getPriority());
 433             }
 434         }
 435     }
 436
 437     @Override
 438     public void processPortStatusMessage(final PortStatusMessage portStatus) {
 439         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 440         final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this, null);
 441
 442         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
 443         if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
 444             // because of ADD status node connector has to be created
 445             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
 446             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
 447             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
 448             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
 449         } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
 450             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
 451         }
 452         submitTransaction();
 453     }
 454
 455     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
 456         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
 457         final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
 458         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
 459         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
 460     }
 461
 462     @Override
 463     public void processPacketInMessage(final PacketInMessage packetInMessage) {
 464         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
 465         final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
 466         final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this, null);
 467
 468         if (packetReceived == null) {
 469             LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
 470             messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
 471             return;
 472         } else {
 473             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
 474         }
 475
 476         if (!packetInLimiter.acquirePermit()) {
 477             LOG.debug("Packet limited");
 478             // TODO: save packet into emergency slot if possible
 479             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PACKET_IN_LIMIT_REACHED_AND_DROPPED);
 480             return;
 481         }
 482
 483         final ListenableFuture<? extends Object> offerNotification = notificationPublishService.offerNotification(packetReceived);
 484         if (NotificationPublishService.REJECTED.equals(offerNotification)) {
 485             LOG.debug("notification offer rejected");
 486             messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 487             packetInLimiter.drainLowWaterMark();
 488             packetInLimiter.releasePermit();
 489             return;
 490         }
 491
 492         Futures.addCallback(offerNotification, new FutureCallback<Object>() {
 493             @Override
 494             public void onSuccess(final Object result) {
 495                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
 496                 packetInLimiter.releasePermit();
 497             }
 498
 499             @Override
 500             public void onFailure(final Throwable t) {
 501                 messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
 502                 LOG.debug("notification offer failed: {}", t.getMessage());
 503                 LOG.trace("notification offer failed..", t);
 504                 packetInLimiter.releasePermit();
 505             }
 506         });
 507     }
 508
 509     @Override
 510     public void processExperimenterMessage(final ExperimenterMessage notification) {
 511         // lookup converter
 512         final ExperimenterDataOfChoice vendorData = notification.getExperimenterDataOfChoice();
 513         final MessageTypeKey<? extends ExperimenterDataOfChoice> key = new MessageTypeKey<>(
 514                 deviceState.getVersion(),
 515                 (Class<? extends ExperimenterDataOfChoice>) vendorData.getImplementedInterface());
 516         final ConvertorMessageFromOFJava<ExperimenterDataOfChoice, MessagePath> messageConverter = extensionConverterProvider.getMessageConverter(key);
 517         if (messageConverter == null) {
 518             LOG.warn("custom converter for {}[OF:{}] not found",
 519                     notification.getExperimenterDataOfChoice().getImplementedInterface(),
 520                     deviceState.getVersion());
 521             return;
 522         }
 523         // build notification
 524         final ExperimenterMessageOfChoice messageOfChoice;
 525         try {
 526             messageOfChoice = messageConverter.convert(vendorData, MessagePath.MESSAGE_NOTIFICATION);
 527             final ExperimenterMessageFromDevBuilder experimenterMessageFromDevBld = new ExperimenterMessageFromDevBuilder()
 528                 .setNode(new NodeRef(deviceState.getNodeInstanceIdentifier()))
 529                     .setExperimenterMessageOfChoice(messageOfChoice);
 530             // publish
 531             notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
 532         } catch (final ConversionException e) {
 533             LOG.warn("Conversion of experimenter notification failed", e);
 534         }
 535     }
 536
 537     @Override
 538     public TranslatorLibrary oook() {
 539         return translatorLibrary;
 540     }
 541
 542     @Override
 543     public HashedWheelTimer getTimer() {
 544         return hashedWheelTimer;
 545     }
 546
 547     @Override
 548     public synchronized void close() {
 549         LOG.debug("closing deviceContext: {}, nodeId:{}",
 550                 getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress(),
 551                 getDeviceState().getNodeId());
 552
 553         if (deviceState.isValid()) {
 554             primaryConnectionContext.closeConnection(false);
 555             tearDown();
 556         }
 557     }
 558
 559     private synchronized void tearDown() {
 560         LOG.trace("tearDown method for node {}", deviceState.getNodeId());
 561         if (deviceState.isValid()) {
 562             deviceState.setValid(false);
 563
 564             for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
 565                 connectionContext.closeConnection(false);
 566             }
 567
 568             deviceGroupRegistry.close();
 569             deviceFlowRegistry.close();
 570             deviceMeterRegistry.close();
 571
 572             final ListenableFuture<Void> future = transactionChainManager.shuttingDown();
 573             Futures.addCallback(future, new FutureCallback<Void>() {
 574
 575                 @Override
 576                 public void onSuccess(final Void result) {
 577                     LOG.info("TxChain {} was shutdown successful.", getDeviceState().getNodeId());
 578                     tearDownClean();
 579                 }
 580
 581                 @Override
 582                 public void onFailure(final Throwable t) {
 583                     LOG.warn("Shutdown TxChain {} fail.", getDeviceState().getNodeId(), t);
 584                     tearDownClean();
 585                 }
 586             });
 587         }
 588     }
 589
 590     private void tearDownClean() {
 591         LOG.info("Closing transaction chain manager without cleaning inventory operational");
 592         transactionChainManager.close();
 593
 594         final LinkedList<DeviceContextClosedHandler> reversedCloseHandlers = new LinkedList<>(closeHandlers);
 595         Collections.reverse(reversedCloseHandlers);
 596         for (final DeviceContextClosedHandler deviceContextClosedHandler : reversedCloseHandlers) {
 597             deviceContextClosedHandler.onDeviceContextClosed(this);
 598         }
 599         closeHandlers.clear();
 600     }
 601
 602     @Override
 603     public void onDeviceDisconnected(final ConnectionContext connectionContext) {
 604         if (getPrimaryConnectionContext().equals(connectionContext)) {
 605             try {
 606                 tearDown();
 607             } catch (final Exception e) {
 608                 LOG.trace("Error closing device context.");
 609             }
 610         } else {
 611             LOG.debug("auxiliary connection dropped: {}, nodeId:{}",
 612                     connectionContext.getConnectionAdapter().getRemoteAddress(),
 613                     getDeviceState().getNodeId());
 614             final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
 615             auxiliaryConnectionContexts.remove(connectionDistinguisher);
 616         }
 617     }
 618
 619     @Override
 620     public void setCurrentBarrierTimeout(final Timeout timeout) {
 621         barrierTaskTimeout = timeout;
 622     }
 623
 624     @Override
 625     public Timeout getBarrierTaskTimeout() {
 626         return barrierTaskTimeout;
 627     }
 628
 629     @Override
 630     public void setNotificationService(final NotificationService notificationService) {
 631         this.notificationService = notificationService;
 632     }
 633
 634     @Override
 635     public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
 636         this.notificationPublishService = notificationPublishService;
 637     }
 638
 639     @Override
 640     public MessageSpy getMessageSpy() {
 641         return messageSpy;
 642     }
 643
 644     @Override
 645     public void addDeviceContextClosedHandler(final DeviceContextClosedHandler deviceContextClosedHandler) {
 646         closeHandlers.add(deviceContextClosedHandler);
 647     }
 648
 649     @Override
 650     public void onPublished() {
 651         primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 652         for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
 653             switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
 654         }
 655     }
 656
 657     @Override
 658     public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
 659         return new MultiMsgCollectorImpl(this, requestContext);
 660     }
 661
 662     @Override
 663     public NodeConnectorRef lookupNodeConnectorRef(final Long portNumber) {
 664         return nodeConnectorCache.get(portNumber);
 665     }
 666
 667     @Override
 668     public void storeNodeConnectorRef(final Long portNumber, final NodeConnectorRef nodeConnectorRef) {
 669         nodeConnectorCache.put(
 670                 Preconditions.checkNotNull(portNumber),
 671                 Preconditions.checkNotNull(nodeConnectorRef));
 672     }
 673
 674     @Override
 675     public void updatePacketInRateLimit(final long upperBound) {
 676         packetInLimiter.changeWaterMarks((int) (LOW_WATERMARK_FACTOR * upperBound), (int) (HIGH_WATERMARK_FACTOR * upperBound));
 677     }
 678
 679     @Override
 680     public ItemLifeCycleRegistry getItemLifeCycleSourceRegistry() {
 681         return itemLifeCycleSourceRegistry;
 682     }
 683
 684     @Override
 685     public void setRpcContext(final RpcContext rpcContext) {
 686         this.rpcContext = rpcContext;
 687     }
 688
 689     @Override
 690     public RpcContext getRpcContext() {
 691         return rpcContext;
 692     }
 693
 694     @Override
 695     public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
 696         this.extensionConverterProvider = extensionConverterProvider;
 697     }
 698
 699     @Override
 700     public ExtensionConverterProvider getExtensionConverterProvider() {
 701         return extensionConverterProvider;
 702     }
 703
 704     @Override
 705     public void setStatisticsContext(final StatisticsContext statisticsContext) {
 706         this.statisticsContext = statisticsContext;
 707     }
 708
 709     @Override
 710     public StatisticsContext getStatisticsContext() {
 711         return statisticsContext;
 712     }
 713 }