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