openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/core/ConnectionConductorImpl.java

   1 /**
   2  * Copyright (c) 2013 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
   9 package org.opendaylight.openflowplugin.openflow.md.core;
  10
  11 import java.math.BigInteger;
  12 import java.util.concurrent.ExecutorService;
  13 import java.util.concurrent.Executors;
  14 import java.util.concurrent.Future;
  15 import java.util.concurrent.TimeUnit;
  16
  17 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  18 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  19 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
  20 import org.opendaylight.openflowplugin.openflow.md.core.session.OFSessionUtil;
  21 import org.opendaylight.openflowplugin.openflow.md.core.session.SessionContext;
  22 import org.opendaylight.openflowplugin.openflow.md.core.session.SessionManager;
  23 import org.opendaylight.openflowplugin.openflow.md.queue.QueueKeeper;
  24 import org.opendaylight.openflowplugin.openflow.md.util.InventoryDataServiceUtil;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeUpdated;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeUpdatedBuilder;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdated;
  28 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
  29 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartRequestFlags;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoInputBuilder;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoOutput;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoReplyInputBuilder;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.EchoRequestMessage;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ErrorMessage;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.ExperimenterMessage;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemovedMessage;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloMessage;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReplyMessage;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartRequestInputBuilder;
  42 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
  43 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OpenflowProtocolListener;
  44 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
  45 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
  46 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.reply.multipart.reply.body.MultipartReplyDesc;
  47 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.multipart.request.multipart.request.body.MultipartRequestDescBuilder;
  48 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.DisconnectEvent;
  49 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SwitchIdleEvent;
  50 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.system.rev130927.SystemNotificationsListener;
  51 import org.opendaylight.yangtools.yang.binding.DataObject;
  52 import org.opendaylight.yangtools.yang.common.RpcError;
  53 import org.opendaylight.yangtools.yang.common.RpcResult;
  54 import org.slf4j.Logger;
  55 import org.slf4j.LoggerFactory;
  56
  57 import com.google.common.util.concurrent.Futures;
  58
  59 /**
  60  * @author mirehak
  61  */
  62 public class ConnectionConductorImpl implements OpenflowProtocolListener,
  63         SystemNotificationsListener, ConnectionConductor, ConnectionReadyListener, HandshakeListener {
  64
  65     protected static final Logger LOG = LoggerFactory
  66             .getLogger(ConnectionConductorImpl.class);
  67
  68     /* variable to make BitMap-based negotiation enabled / disabled.
  69      * it will help while testing and isolating issues related to processing of
  70      * BitMaps from switches.
  71      */
  72     protected boolean isBitmapNegotiationEnable = true;
  73     protected ErrorHandler errorHandler;
  74
  75     private final ConnectionAdapter connectionAdapter;
  76     private ConnectionConductor.CONDUCTOR_STATE conductorState;
  77     private Short version;
  78
  79     private SwitchConnectionDistinguisher auxiliaryKey;
  80
  81     private SessionContext sessionContext;
  82
  83     protected boolean isFirstHelloNegotiation = true;
  84     protected Short lastProposedVersion = null;
  85
  86     private QueueKeeper<OfHeader, DataObject> queueKeeper;
  87     private ExecutorService hsPool;
  88     private HandshakeManager handshakeManager;
  89
  90     /**
  91      * @param connectionAdapter
  92      */
  93     public ConnectionConductorImpl(ConnectionAdapter connectionAdapter) {
  94         this.connectionAdapter = connectionAdapter;
  95         conductorState = CONDUCTOR_STATE.HANDSHAKING;
  96         hsPool = Executors.newFixedThreadPool(1);
  97         handshakeManager = new HandshakeManagerImpl(connectionAdapter,
  98                 ConnectionConductor.versionOrder.get(0), ConnectionConductor.versionOrder);
  99         handshakeManager.setUseVersionBitmap(isBitmapNegotiationEnable);
 100         handshakeManager.setHandshakeListener(this);
 101     }
 102
 103     @Override
 104     public void init() {
 105         connectionAdapter.setMessageListener(this);
 106         connectionAdapter.setSystemListener(this);
 107         connectionAdapter.setConnectionReadyListener(this);
 108     }
 109
 110     @Override
 111     public void setQueueKeeper(QueueKeeper<OfHeader, DataObject> queueKeeper) {
 112         this.queueKeeper = queueKeeper;
 113     }
 114
 115     /**
 116      * @param errorHandler the errorHandler to set
 117      */
 118     @Override
 119     public void setErrorHandler(ErrorHandler errorHandler) {
 120         this.errorHandler = errorHandler;
 121         handshakeManager.setErrorHandler(errorHandler);
 122     }
 123
 124     @Override
 125     public void onEchoRequestMessage(final EchoRequestMessage echoRequestMessage) {
 126         new Thread(new Runnable() {
 127             @Override
 128             public void run() {
 129                 LOG.debug("echo request received: " + echoRequestMessage.getXid());
 130                 EchoReplyInputBuilder builder = new EchoReplyInputBuilder();
 131                 builder.setVersion(echoRequestMessage.getVersion());
 132                 builder.setXid(echoRequestMessage.getXid());
 133                 builder.setData(echoRequestMessage.getData());
 134
 135                 getConnectionAdapter().echoReply(builder.build());
 136             }
 137         }).start();
 138     }
 139
 140     @Override
 141     public void onErrorMessage(ErrorMessage errorMessage) {
 142         queueKeeper.push(errorMessage, this);
 143     }
 144
 145     @Override
 146     public void onExperimenterMessage(ExperimenterMessage experimenterMessage) {
 147         queueKeeper.push(experimenterMessage, this);
 148     }
 149
 150     @Override
 151     public void onFlowRemovedMessage(FlowRemovedMessage message) {
 152         queueKeeper.push(message, this);
 153     }
 154
 155
 156     /**
 157      * version negotiation happened as per following steps:
 158      * 1. If HelloMessage version field has same version, continue connection processing.
 159      *    If HelloMessage version is lower than supported versions, just disconnect.
 160      * 2. If HelloMessage contains bitmap and common version found in bitmap
 161      *    then continue connection processing. if no common version found, just disconnect.
 162      * 3. If HelloMessage version is not supported, send HelloMessage with lower supported version.
 163      * 4. If Hello message received again with not supported version, just disconnect.
 164      *
 165      *   TODO: Better to handle handshake into a maintainable innerclass which uses State-Pattern.
 166      */
 167     @Override
 168     public synchronized void onHelloMessage(final HelloMessage hello) {
 169         LOG.debug("processing HELLO.xid{}", hello.getXid());
 170         checkState(CONDUCTOR_STATE.HANDSHAKING);
 171         handshakeManager.setReceivedHello(hello);
 172         hsPool.execute(handshakeManager);
 173     }
 174
 175     /**
 176      * @return rpc-response timeout in [ms]
 177      */
 178     protected long getMaxTimeout() {
 179         // TODO:: get from configuration
 180         return 2000;
 181     }
 182
 183     /**
 184      * @return milliseconds
 185      */
 186     protected TimeUnit getMaxTimeoutUnit() {
 187         // TODO:: get from configuration
 188         return TimeUnit.MILLISECONDS;
 189     }
 190
 191     @Override
 192     public void onMultipartReplyMessage(MultipartReplyMessage message) {
 193         queueKeeper.push(message, this);
 194     }
 195
 196     @Override
 197     public void onPacketInMessage(PacketInMessage message) {
 198         queueKeeper.push(message, this);
 199     }
 200
 201     @Override
 202     public void onPortStatusMessage(PortStatusMessage message) {
 203         this.getSessionContext().processPortStatusMsg(message);
 204         queueKeeper.push(message, this);
 205     }
 206
 207     @Override
 208     public void onSwitchIdleEvent(SwitchIdleEvent notification) {
 209         new Thread(new Runnable() {
 210             @Override
 211             public void run() {
 212                 if (!CONDUCTOR_STATE.WORKING.equals(getConductorState())) {
 213                     // idle state in any other conductorState than WORKING means real
 214                     // problem and wont be handled by echoReply, but disconnection
 215                     disconnect();
 216                     OFSessionUtil.getSessionManager().invalidateOnDisconnect(ConnectionConductorImpl.this);
 217                 } else {
 218                     LOG.debug("first idle state occured");
 219                     EchoInputBuilder builder = new EchoInputBuilder();
 220                     builder.setVersion(getVersion());
 221                     builder.setXid(getSessionContext().getNextXid());
 222
 223                     Future<RpcResult<EchoOutput>> echoReplyFuture = getConnectionAdapter()
 224                             .echo(builder.build());
 225
 226                     try {
 227                         RpcResult<EchoOutput> echoReplyValue = echoReplyFuture.get(getMaxTimeout(),
 228                                 getMaxTimeoutUnit());
 229                         if (echoReplyValue.isSuccessful()) {
 230                             setConductorState(CONDUCTOR_STATE.WORKING);
 231                         } else {
 232                             for (RpcError replyError : echoReplyValue.getErrors()) {
 233                                 Throwable cause = replyError.getCause();
 234                                 LOG.error(
 235                                         "while receiving echoReply in TIMEOUTING state: "
 236                                                 + cause.getMessage(), cause);
 237                             }
 238                             //switch issue occurred
 239                             throw new Exception("switch issue occurred");
 240                         }
 241                     } catch (Exception e) {
 242                         LOG.error("while waiting for echoReply in TIMEOUTING state: "
 243                                 + e.getMessage(), e);
 244                         //switch is not responding
 245                         disconnect();
 246                         OFSessionUtil.getSessionManager().invalidateOnDisconnect(ConnectionConductorImpl.this);
 247                     }
 248                 }
 249             }
 250
 251         }).start();
 252     }
 253
 254     /**
 255      * @param conductorState
 256      *            the connectionState to set
 257      */
 258     @Override
 259     public void setConductorState(CONDUCTOR_STATE conductorState) {
 260         this.conductorState = conductorState;
 261     }
 262
 263     @Override
 264     public CONDUCTOR_STATE getConductorState() {
 265         return conductorState;
 266     }
 267
 268     /**
 269      * @param handshaking
 270      */
 271     protected void checkState(CONDUCTOR_STATE expectedState) {
 272         if (!conductorState.equals(expectedState)) {
 273             throw new IllegalStateException("Expected state: " + expectedState
 274                     + ", actual state:" + conductorState);
 275         }
 276     }
 277
 278     @Override
 279     public void onDisconnectEvent(DisconnectEvent arg0) {
 280         SessionManager sessionManager = OFSessionUtil.getSessionManager();
 281         sessionManager.invalidateOnDisconnect(this);
 282     }
 283
 284     @Override
 285     public Short getVersion() {
 286         return version;
 287     }
 288
 289     @Override
 290     public Future<Boolean> disconnect() {
 291         LOG.info("disconnecting: sessionCtx="+sessionContext+"|auxId="+auxiliaryKey);
 292
 293         Future<Boolean> result = null;
 294         if (connectionAdapter.isAlive()) {
 295             result = connectionAdapter.disconnect();
 296         } else {
 297             LOG.debug("connection already disconnected");
 298             result = Futures.immediateFuture(true);
 299         }
 300
 301         return result;
 302     }
 303
 304     @Override
 305     public void setConnectionCookie(SwitchConnectionDistinguisher auxiliaryKey) {
 306         this.auxiliaryKey = auxiliaryKey;
 307     }
 308
 309     @Override
 310     public void setSessionContext(SessionContext sessionContext) {
 311         this.sessionContext = sessionContext;
 312     }
 313
 314     @Override
 315     public SwitchConnectionDistinguisher getAuxiliaryKey() {
 316         return auxiliaryKey;
 317     }
 318
 319     @Override
 320     public SessionContext getSessionContext() {
 321         return sessionContext;
 322     }
 323
 324     @Override
 325     public ConnectionAdapter getConnectionAdapter() {
 326         return connectionAdapter;
 327     }
 328
 329     @Override
 330     public void onConnectionReady() {
 331         LOG.debug("connection is ready-to-use");
 332         hsPool.execute(handshakeManager);
 333     }
 334
 335     @Override
 336     public void onHandshakeSuccessfull(GetFeaturesOutput featureOutput,
 337             Short negotiatedVersion) {
 338         version = negotiatedVersion;
 339         conductorState = CONDUCTOR_STATE.WORKING;
 340         OFSessionUtil.registerSession(this, featureOutput, negotiatedVersion);
 341         requestDesc();
 342     }
 343
 344     /*
 345      *  Send an OFPMP_DESC request message to the switch
 346      */
 347
 348     private void requestDesc() {
 349         MultipartRequestInputBuilder builder = new MultipartRequestInputBuilder();
 350         builder.setType(MultipartType.OFPMPDESC);
 351         builder.setVersion(getVersion());
 352         builder.setFlags(new MultipartRequestFlags(false));
 353         builder.setMultipartRequestBody(new MultipartRequestDescBuilder().build());
 354         builder.setXid(getSessionContext().getNextXid());
 355         getConnectionAdapter().multipartRequest(builder.build());
 356     }
 357
 358     /**
 359      * @param isBitmapNegotiationEnable the isBitmapNegotiationEnable to set
 360      */
 361     public void setBitmapNegotiationEnable(
 362             boolean isBitmapNegotiationEnable) {
 363         this.isBitmapNegotiationEnable = isBitmapNegotiationEnable;
 364     }
 365
 366     protected void shutdownPool() {
 367         hsPool.shutdownNow();
 368         LOG.debug("pool is terminated: {}", hsPool.isTerminated());
 369     }
 370 }