openflowplugin/src/main/java/org/opendaylight/openflowplugin/openflow/md/core/HandshakeManagerImpl.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 package org.opendaylight.openflowplugin.openflow.md.core;
   9
  10 import com.google.common.util.concurrent.FutureCallback;
  11 import com.google.common.util.concurrent.Futures;
  12 import com.google.common.util.concurrent.JdkFutureAdapters;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.SettableFuture;
  15 import java.util.List;
  16 import java.util.Objects;
  17 import java.util.concurrent.Future;
  18 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
  19 import org.opendaylight.openflowplugin.api.OFConstants;
  20 import org.opendaylight.openflowplugin.api.openflow.md.core.ErrorHandler;
  21 import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeListener;
  22 import org.opendaylight.openflowplugin.api.openflow.md.core.HandshakeManager;
  23 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesInputBuilder;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.GetFeaturesOutput;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloInput;
  26 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.HelloMessage;
  27 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.hello.Elements;
  28 import org.opendaylight.yangtools.yang.common.RpcError;
  29 import org.opendaylight.yangtools.yang.common.RpcResult;
  30 import org.slf4j.Logger;
  31 import org.slf4j.LoggerFactory;
  32
  33 public class HandshakeManagerImpl implements HandshakeManager {
  34
  35     private static final long activeXID = 20L;
  36
  37     private static final Logger LOG = LoggerFactory
  38             .getLogger(HandshakeManagerImpl.class);
  39
  40     private Short lastProposedVersion;
  41     private Short lastReceivedVersion;
  42     private final List<Short> versionOrder;
  43
  44
  45     private final ConnectionAdapter connectionAdapter;
  46     private Short version;
  47     private ErrorHandler errorHandler;
  48
  49
  50
  51     private Short highestVersion;
  52
  53     private Long activeXid;
  54
  55     private HandshakeListener handshakeListener;
  56
  57     private boolean useVersionBitmap;
  58
  59     /**
  60      * @param connectionAdapter connection adaptor for switch
  61      * @param highestVersion highest openflow version
  62      * @param versionOrder list of version in order for connection protocol negotiation
  63      */
  64     public HandshakeManagerImpl(ConnectionAdapter connectionAdapter, Short highestVersion,
  65             List<Short> versionOrder) {
  66         this.highestVersion = highestVersion;
  67         this.versionOrder = versionOrder;
  68         this.connectionAdapter = connectionAdapter;
  69     }
  70
  71     @Override
  72     public void setHandshakeListener(HandshakeListener handshakeListener) {
  73         this.handshakeListener = handshakeListener;
  74     }
  75
  76     @Override
  77     public synchronized void shake(HelloMessage receivedHello) {
  78
  79         if (version != null) {
  80             // Some switches respond with a second HELLO acknowledging our HELLO
  81             // but we've already completed the handshake based on the negotiated
  82             // version and have registered this switch.
  83             LOG.debug("Hello recieved after handshake already settled ... ignoring.");
  84             return;
  85         }
  86
  87         LOG.trace("handshake STARTED");
  88         setActiveXid(activeXID);
  89
  90         try {
  91             if (receivedHello == null) {
  92                 // first Hello sending
  93                 sendHelloMessage(highestVersion, getNextXid());
  94                 lastProposedVersion = highestVersion;
  95                 LOG.trace("ret - firstHello+wait");
  96                 return;
  97             }
  98
  99             // process the 2. and later hellos
 100             Short remoteVersion = receivedHello.getVersion();
 101             List<Elements> elements = receivedHello.getElements();
 102             setActiveXid(receivedHello.getXid());
 103             List<Boolean> remoteVersionBitmap = MessageFactory.digVersions(elements);
 104             LOG.debug("Hello message: version={}, xid={}, bitmap={}", remoteVersion,
 105                     receivedHello.getXid(), remoteVersionBitmap);
 106
 107             if (useVersionBitmap && remoteVersionBitmap != null) {
 108                 // versionBitmap on both sides -> ONE STEP DECISION
 109                 handleVersionBitmapNegotiation(elements);
 110             } else {
 111                 // versionBitmap missing at least on one side -> STEP-BY-STEP NEGOTIATION applying
 112                 handleStepByStepVersionNegotiation(remoteVersion);
 113             }
 114         } catch (Exception ex) {
 115             errorHandler.handleException(ex);
 116             LOG.trace("ret - shake fail - closing");
 117             handshakeListener.onHandshakeFailure();
 118         }
 119     }
 120
 121     /**
 122      * @param remoteVersion remote version
 123      * @throws Exception exception
 124      */
 125     private void handleStepByStepVersionNegotiation(final Short remoteVersion) throws Exception {
 126         LOG.debug("remoteVersion:{} lastProposedVersion:{}, highestVersion:{}",
 127                 remoteVersion, lastProposedVersion, highestVersion);
 128
 129         if (lastProposedVersion == null) {
 130             // first hello has not been sent yet, send it and either wait for next remote
 131             // version or proceed
 132             lastProposedVersion = proposeNextVersion(remoteVersion);
 133             final Long nextHelloXid = getNextXid();
 134             ListenableFuture<Void> helloResult = sendHelloMessage(lastProposedVersion, nextHelloXid);
 135             Futures.addCallback(helloResult, new FutureCallback<Void>() {
 136                 @Override
 137                 public void onSuccess(Void result) {
 138                     try {
 139                         stepByStepVersionSubStep(remoteVersion, lastProposedVersion);
 140                     } catch (Exception e) {
 141                         errorHandler.handleException(e);
 142                         handshakeListener.onHandshakeFailure();
 143                     }
 144                 }
 145
 146                 @Override
 147                 public void onFailure(Throwable t) {
 148                     LOG.info("hello sending seriously failed [{}]", nextHelloXid);
 149                     LOG.trace("detail of hello send problem", t);
 150                 }
 151             });
 152         } else {
 153             stepByStepVersionSubStep(remoteVersion, lastProposedVersion);
 154         }
 155     }
 156
 157     private void stepByStepVersionSubStep(Short remoteVersion, Short lastProposedVersion) throws Exception {
 158         if (remoteVersion.equals(lastProposedVersion)) {
 159             postHandshake(lastProposedVersion, getNextXid());
 160             LOG.trace("ret - OK - switch answered with lastProposedVersion");
 161         } else {
 162             checkNegotiationStalling(remoteVersion);
 163
 164             if (remoteVersion > (lastProposedVersion == null ? highestVersion : this.lastProposedVersion)) {
 165                 // wait for next version
 166                 LOG.trace("ret - wait");
 167             } else {
 168                 //propose lower version
 169                 handleLowerVersionProposal(remoteVersion);
 170             }
 171         }
 172     }
 173
 174     /**
 175      * @param remoteVersion remote version
 176      * @throws Exception exception
 177      */
 178     private void handleLowerVersionProposal(Short remoteVersion) throws Exception {
 179         Short proposedVersion;
 180         // find the version from header version field
 181         proposedVersion = proposeNextVersion(remoteVersion);
 182         lastProposedVersion = proposedVersion;
 183         sendHelloMessage(proposedVersion, getNextXid());
 184
 185         if (! Objects.equals(proposedVersion, remoteVersion)) {
 186             LOG.trace("ret - sent+wait");
 187         } else {
 188             LOG.trace("ret - sent+OK");
 189             postHandshake(proposedVersion, getNextXid());
 190         }
 191     }
 192
 193     /**
 194      * @param elements version elements
 195      * @throws Exception exception
 196      */
 197     private void handleVersionBitmapNegotiation(List<Elements> elements) throws Exception {
 198         final Short proposedVersion = proposeCommonBitmapVersion(elements);
 199         if (lastProposedVersion == null) {
 200             // first hello has not been sent yet
 201             Long nexHelloXid = getNextXid();
 202             ListenableFuture<Void> helloDone = sendHelloMessage(proposedVersion, nexHelloXid);
 203             Futures.addCallback(helloDone, new FutureCallback<Void>() {
 204                 @Override
 205                 public void onSuccess(Void result) {
 206                     LOG.trace("ret - DONE - versionBitmap");
 207                     postHandshake(proposedVersion, getNextXid());
 208                 }
 209
 210                 @Override
 211                 public void onFailure(Throwable t) {
 212                     // NOOP
 213                 }
 214             });
 215             LOG.trace("next proposal [{}] with versionBitmap hooked ..", nexHelloXid);
 216         } else {
 217             LOG.trace("ret - DONE - versionBitmap");
 218             postHandshake(proposedVersion, getNextXid());
 219         }
 220     }
 221
 222     /**
 223      *
 224      * @return next tx id
 225      */
 226     private Long getNextXid() {
 227         activeXid += 1;
 228         return activeXid;
 229     }
 230
 231     /**
 232      * @param xid tx id
 233      */
 234     private void setActiveXid(Long xid) {
 235         this.activeXid = xid;
 236     }
 237
 238     /**
 239      * @param remoteVersion remove version
 240      */
 241     private void checkNegotiationStalling(Short remoteVersion) {
 242         if (lastReceivedVersion != null && lastReceivedVersion.equals(remoteVersion)) {
 243             throw new IllegalStateException("version negotiation stalled: version = "+remoteVersion);
 244         }
 245         lastReceivedVersion = remoteVersion;
 246     }
 247
 248     @Override
 249     public Short getVersion() {
 250         return version;
 251     }
 252
 253     /**
 254      * find common highest supported bitmap version
 255      * @param list bitmap list
 256      * @return proposed bitmap value
 257      */
 258     protected Short proposeCommonBitmapVersion(List<Elements> list) {
 259         Short supportedHighestVersion = null;
 260         if((null != list) && (0 != list.size())) {
 261             for(Elements element : list) {
 262                 List<Boolean> bitmap = element.getVersionBitmap();
 263                 // check for version bitmap
 264                 for(short bitPos : OFConstants.VERSION_ORDER) {
 265                     // with all the version it should work.
 266                     if(bitmap.get(bitPos % Integer.SIZE)) {
 267                         supportedHighestVersion = bitPos;
 268                         break;
 269                     }
 270                 }
 271             }
 272
 273             if(null == supportedHighestVersion) {
 274                 LOG.trace("versionBitmap: no common version found");
 275                 throw new IllegalArgumentException("no common version found in versionBitmap");
 276             }
 277         }
 278
 279         return supportedHighestVersion;
 280     }
 281
 282     /**
 283      * find supported version based on remoteVersion
 284      * @param remoteVersion openflow version supported by remote entity
 285      * @return openflow version
 286      */
 287     protected short proposeNextVersion(short remoteVersion) {
 288         Short proposal = null;
 289         for (short offer : versionOrder) {
 290             if (offer <= remoteVersion) {
 291                 proposal = offer;
 292                 break;
 293             }
 294         }
 295         if (proposal == null) {
 296             throw new IllegalArgumentException("no equal or lower version found, unsupported version: "
 297                     + remoteVersion);
 298         }
 299         return proposal;
 300     }
 301
 302     /**
 303      * send hello reply without versionBitmap
 304      * @param helloVersion initial hello version for openflow connection negotiation
 305      * @param helloXid transaction id
 306      * @throws Exception
 307      */
 308     private ListenableFuture<Void> sendHelloMessage(Short helloVersion, final Long helloXid) throws Exception {
 309
 310
 311         HelloInput helloInput = MessageFactory.createHelloInput(helloVersion, helloXid, versionOrder);
 312
 313         final SettableFuture<Void> resultFtr = SettableFuture.create();
 314
 315         LOG.debug("sending hello message: version{}, xid={}, version bitmap={}",
 316                 helloVersion, helloXid, MessageFactory.digVersions(helloInput.getElements()));
 317
 318         Future<RpcResult<Void>> helloResult = connectionAdapter.hello(helloInput);
 319
 320         ListenableFuture<RpcResult<Void>> rpcResultListenableFuture = JdkFutureAdapters.listenInPoolThread(helloResult);
 321         Futures.addCallback(rpcResultListenableFuture, new FutureCallback<RpcResult<Void>>() {
 322             @Override
 323             public void onSuccess(RpcResult<Void> result) {
 324                 if (result.isSuccessful()) {
 325                     LOG.debug("hello successfully sent, xid={}, addr={}", helloXid, connectionAdapter.getRemoteAddress());
 326                     resultFtr.set(null);
 327                 } else {
 328                     for (RpcError error : result.getErrors()) {
 329                         LOG.debug("hello sending failed [{}]: i:{} s:{} m:{}, addr:{}", helloXid,
 330                                 error.getInfo(), error.getSeverity(), error.getMessage(),
 331                                 connectionAdapter.getRemoteAddress());
 332                         if (error.getCause() != null) {
 333                             LOG.trace("DETAIL of sending hello failure", error.getCause());
 334                         }
 335                     }
 336                     resultFtr.cancel(false);
 337                     handshakeListener.onHandshakeFailure();
 338                 }
 339             }
 340
 341             @Override
 342             public void onFailure(Throwable t) {
 343                 LOG.warn("sending of hello failed seriously [{}, addr:{}]: {}", helloXid,
 344                         connectionAdapter.getRemoteAddress(), t.getMessage());
 345                 LOG.trace("DETAIL of sending of hello failure:", t);
 346                 resultFtr.cancel(false);
 347                 handshakeListener.onHandshakeFailure();
 348             }
 349         });
 350         LOG.trace("sending hello message [{}] - result hooked ..", helloXid);
 351         return resultFtr;
 352     }
 353
 354
 355     /**
 356      * after handshake set features, register to session
 357      * @param proposedVersion proposed openflow version
 358      * @param xid transaction id
 359      */
 360     protected void postHandshake(final Short proposedVersion, final Long xid) {
 361         // set version
 362         version = proposedVersion;
 363
 364         LOG.debug("version set: {}", proposedVersion);
 365         // request features
 366         GetFeaturesInputBuilder featuresBuilder = new GetFeaturesInputBuilder();
 367         featuresBuilder.setVersion(version).setXid(xid);
 368         LOG.debug("sending feature request for version={} and xid={}", version, xid);
 369         Future<RpcResult<GetFeaturesOutput>> featuresFuture = connectionAdapter
 370                 .getFeatures(featuresBuilder.build());
 371
 372         Futures.addCallback(JdkFutureAdapters.listenInPoolThread(featuresFuture),
 373                 new FutureCallback<RpcResult<GetFeaturesOutput>>() {
 374                     @Override
 375                     public void onSuccess(RpcResult<GetFeaturesOutput> rpcFeatures) {
 376                         LOG.trace("features are back");
 377                         if (rpcFeatures.isSuccessful()) {
 378                             GetFeaturesOutput featureOutput = rpcFeatures.getResult();
 379
 380                             LOG.debug("obtained features: datapathId={}",
 381                                     featureOutput.getDatapathId());
 382                             LOG.debug("obtained features: auxiliaryId={}",
 383                                     featureOutput.getAuxiliaryId());
 384                             LOG.trace("handshake SETTLED: version={}, datapathId={}, auxiliaryId={}",
 385                                     version, featureOutput.getDatapathId(), featureOutput.getAuxiliaryId());
 386                             handshakeListener.onHandshakeSuccessful(featureOutput, proposedVersion);
 387                         } else {
 388                             // handshake failed
 389                             LOG.warn("issuing disconnect during handshake [{}]", connectionAdapter.getRemoteAddress());
 390                             for (RpcError rpcError : rpcFeatures.getErrors()) {
 391                                 LOG.debug("handshake - features failure [{}]: i:{} | m:{} | s:{}", xid,
 392                                         rpcError.getInfo(), rpcError.getMessage(), rpcError.getSeverity(),
 393                                         rpcError.getCause()
 394                                 );
 395                             }
 396                             handshakeListener.onHandshakeFailure();
 397                         }
 398
 399                         LOG.debug("postHandshake DONE");
 400                     }
 401
 402                     @Override
 403                     public void onFailure(Throwable t) {
 404                         LOG.warn("getting feature failed seriously [{}, addr:{}]: {}", xid,
 405                                 connectionAdapter.getRemoteAddress(), t.getMessage());
 406                         LOG.trace("DETAIL of sending of hello failure:", t);
 407                     }
 408                 });
 409
 410         LOG.debug("future features [{}] hooked ..", xid);
 411
 412     }
 413
 414     @Override
 415     public void setUseVersionBitmap(boolean useVersionBitmap) {
 416         this.useVersionBitmap = useVersionBitmap;
 417     }
 418
 419     @Override
 420     public void setErrorHandler(ErrorHandler errorHandler) {
 421         this.errorHandler = errorHandler;
 422     }
 423 }