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