openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/TransactionChainManager.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
   9 package org.opendaylight.openflowplugin.impl.device;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.util.concurrent.CheckedFuture;
  14 import com.google.common.util.concurrent.FutureCallback;
  15 import com.google.common.util.concurrent.Futures;
  16 import com.google.common.util.concurrent.ListenableFuture;
  17 import javax.annotation.Nonnull;
  18 import javax.annotation.Nullable;
  19 import javax.annotation.concurrent.GuardedBy;
  20 import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
  21 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  22 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  23 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  24 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  25 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  26 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  27 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  28 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  29 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
  30 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
  33 import org.opendaylight.yangtools.yang.binding.DataObject;
  34 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  35 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38
  39 /**
  40  * openflowplugin-impl
  41  * org.opendaylight.openflowplugin.impl.device
  42  * <p/>
  43  * Package protected class for controlling {@link WriteTransaction} life cycle. It is
  44  * a {@link TransactionChainListener} and provide package protected methods for writeToTransaction
  45  * method (wrapped {@link WriteTransaction#put(LogicalDatastoreType, InstanceIdentifier, DataObject)})
  46  * and submitTransaction method (wrapped {@link WriteTransaction#submit()})
  47  *
  48  * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
  49  *         </p>
  50  *         Created: Apr 2, 2015
  51  */
  52 class TransactionChainManager implements TransactionChainListener, AutoCloseable {
  53
  54     private static final Logger LOG = LoggerFactory.getLogger(TransactionChainManager.class);
  55
  56     private final Object txLock = new Object();
  57     private final KeyedInstanceIdentifier<Node, NodeKey> nodeII;
  58     private final DeviceInfo deviceInfo;
  59     private final DataBroker dataBroker;
  60     private final LifecycleConductor conductor;
  61
  62     @GuardedBy("txLock")
  63     private WriteTransaction wTx;
  64     @GuardedBy("txLock")
  65     private BindingTransactionChain txChainFactory;
  66     @GuardedBy("txLock")
  67     private boolean submitIsEnabled;
  68     @GuardedBy("txLock")
  69     private ListenableFuture<Void> lastSubmittedFuture;
  70
  71     private boolean initCommit;
  72
  73     public TransactionChainManagerStatus getTransactionChainManagerStatus() {
  74         return transactionChainManagerStatus;
  75     }
  76
  77     @GuardedBy("txLock")
  78     private TransactionChainManagerStatus transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
  79
  80     TransactionChainManager(@Nonnull final DataBroker dataBroker,
  81                             @Nonnull final DeviceInfo deviceInfo,
  82                             @Nonnull final LifecycleConductor conductor) {
  83         this.dataBroker = Preconditions.checkNotNull(dataBroker);
  84         this.conductor = Preconditions.checkNotNull(conductor);
  85         this.deviceInfo = deviceInfo;
  86         this.nodeII = deviceInfo.getNodeInstanceIdentifier();
  87         this.transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
  88         lastSubmittedFuture = Futures.immediateFuture(null);
  89         LOG.debug("created txChainManager for {}", this.nodeII);
  90     }
  91
  92     private NodeId nodeId() {
  93         return nodeII.getKey().getId();
  94     }
  95
  96     @GuardedBy("txLock")
  97     private void createTxChain() {
  98         if (txChainFactory != null) {
  99             txChainFactory.close();
 100         }
 101         txChainFactory = dataBroker.createTransactionChain(TransactionChainManager.this);
 102     }
 103
 104     void initialSubmitWriteTransaction() {
 105         enableSubmit();
 106         submitWriteTransaction();
 107     }
 108
 109     /**
 110      * Method change status for TxChainManager to {@link TransactionChainManagerStatus#WORKING} and it has to make
 111      * registration for this class instance as {@link TransactionChainListener} to provide possibility a make DS
 112      * transactions. Call this method for MASTER role only.
 113      */
 114     public void activateTransactionManager() {
 115         LOG.trace("activateTransactionManager for node {} transaction submit is set to {}", nodeId(), submitIsEnabled);
 116         synchronized (txLock) {
 117             if (TransactionChainManagerStatus.SLEEPING.equals(transactionChainManagerStatus)) {
 118                 LOG.debug("Transaction Factory create {}", nodeId());
 119                 Preconditions.checkState(txChainFactory == null, "TxChainFactory survive last close.");
 120                 Preconditions.checkState(wTx == null, "We have some unexpected WriteTransaction.");
 121                 this.transactionChainManagerStatus = TransactionChainManagerStatus.WORKING;
 122                 this.submitIsEnabled = false;
 123                 this.initCommit = true;
 124                 createTxChain();
 125             } else {
 126                 LOG.debug("Transaction is active {}", nodeId());
 127             }
 128         }
 129     }
 130
 131     /**
 132      * Method change status for TxChainManger to {@link TransactionChainManagerStatus#SLEEPING} and it unregisters
 133      * this class instance as {@link TransactionChainListener} so it broke a possibility to write something to DS.
 134      * Call this method for SLAVE only.
 135      * @return Future
 136      */
 137     public ListenableFuture<Void> deactivateTransactionManager() {
 138         final ListenableFuture<Void> future;
 139         synchronized (txLock) {
 140             if (TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus)) {
 141                 LOG.debug("Submitting all transactions if we were in status WORKING for Node {}", nodeId());
 142                 transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
 143                 future = txChainShuttingDown();
 144                 Preconditions.checkState(wTx == null, "We have some unexpected WriteTransaction.");
 145                 LOG.debug("Transaction Factory deactivate for Node {}", nodeId());
 146                 Futures.addCallback(future, new FutureCallback<Void>() {
 147                     @Override
 148                     public void onSuccess(final Void result) {
 149                         txChainFactory.close();
 150                         txChainFactory = null;
 151                     }
 152
 153                     @Override
 154                     public void onFailure(final Throwable t) {
 155                         txChainFactory.close();
 156                         txChainFactory = null;
 157                     }
 158                 });
 159             } else {
 160                 // TODO : ignoring redundant deactivate invocation
 161                 future = Futures.immediateCheckedFuture(null);
 162             }
 163         }
 164         return future;
 165     }
 166
 167     boolean submitWriteTransaction() {
 168         synchronized (txLock) {
 169             if (!submitIsEnabled) {
 170                 LOG.trace("transaction not committed - submit block issued");
 171                 return false;
 172             }
 173             if (wTx == null) {
 174                 LOG.trace("nothing to commit - submit returns true");
 175                 return true;
 176             }
 177             Preconditions.checkState(TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus),
 178                     "we have here Uncompleted Transaction for node {} and we are not MASTER", nodeII);
 179             final CheckedFuture<Void, TransactionCommitFailedException> submitFuture = wTx.submit();
 180             Futures.addCallback(submitFuture, new FutureCallback<Void>() {
 181                 @Override
 182                 public void onSuccess(final Void result) {
 183                     if (initCommit) {
 184                         initCommit = false;
 185                     }
 186                 }
 187
 188                 @Override
 189                 public void onFailure(final Throwable t) {
 190                     if (t instanceof TransactionCommitFailedException) {
 191                         LOG.error("Transaction commit failed. {}", t);
 192                     } else {
 193                         LOG.error("Exception during transaction submitting. {}", t);
 194                     }
 195                     if (initCommit) {
 196                         LOG.error("Initial commit failed. {}", t);
 197                         conductor.closeConnection(deviceInfo);
 198                     }
 199                 }
 200             });
 201             lastSubmittedFuture = submitFuture;
 202             wTx = null;
 203         }
 204         return true;
 205     }
 206
 207     <T extends DataObject> void addDeleteOperationTotTxChain(final LogicalDatastoreType store,
 208                                                              final InstanceIdentifier<T> path) throws Exception {
 209         final WriteTransaction writeTx = getTransactionSafely();
 210         if (writeTx != null) {
 211             LOG.trace("addDeleteOperation called with path {} ", path);
 212             writeTx.delete(store, path);
 213         } else {
 214             LOG.debug("WriteTx is null for node {}. Delete {} was not realized.", nodeII, path);
 215             throw new Exception("Cannot write into transaction.");
 216         }
 217     }
 218
 219     <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 220                                                    final InstanceIdentifier<T> path,
 221                                                    final T data,
 222                                                    final boolean createParents) throws Exception {
 223         final WriteTransaction writeTx = getTransactionSafely();
 224         if (writeTx != null) {
 225             LOG.trace("writeToTransaction called with path {} ", path);
 226             writeTx.put(store, path, data, createParents);
 227         } else {
 228             LOG.debug("WriteTx is null for node {}. Write data for {} was not realized.", nodeII, path);
 229             throw new Exception("Cannot write into transaction.");
 230         }
 231     }
 232
 233     @Override
 234     public void onTransactionChainFailed(final TransactionChain<?, ?> chain,
 235                                          final AsyncTransaction<?, ?> transaction, final Throwable cause) {
 236         if (transactionChainManagerStatus.equals(TransactionChainManagerStatus.WORKING)) {
 237             LOG.warn("txChain failed -> recreating due to {}", cause);
 238             recreateTxChain();
 239         }
 240     }
 241
 242     @Override
 243     public void onTransactionChainSuccessful(final TransactionChain<?, ?> chain) {
 244         // NOOP
 245     }
 246
 247     private void recreateTxChain() {
 248         synchronized (txLock) {
 249             createTxChain();
 250             wTx = null;
 251         }
 252     }
 253
 254     @Nullable
 255     private WriteTransaction getTransactionSafely() {
 256         if (wTx == null && TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus)) {
 257             synchronized (txLock) {
 258                 if (wTx == null && TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus)) {
 259                     if (wTx == null && txChainFactory != null) {
 260                         wTx = txChainFactory.newWriteOnlyTransaction();
 261                     }
 262                 }
 263             }
 264         }
 265         return wTx;
 266     }
 267
 268     @VisibleForTesting
 269     void enableSubmit() {
 270         synchronized (txLock) {
 271             /* !!!IMPORTANT: never set true without txChainFactory */
 272             submitIsEnabled = txChainFactory != null;
 273         }
 274     }
 275
 276     ListenableFuture<Void> shuttingDown() {
 277         LOG.debug("TxManager is going SHUTTING_DOWN for node {}", nodeII);
 278         ListenableFuture<Void> future;
 279         synchronized (txLock) {
 280             this.transactionChainManagerStatus = TransactionChainManagerStatus.SHUTTING_DOWN;
 281             future = txChainShuttingDown();
 282         }
 283         return future;
 284     }
 285
 286     @GuardedBy("txLock")
 287     private ListenableFuture<Void> txChainShuttingDown() {
 288         submitIsEnabled = false;
 289         ListenableFuture<Void> future;
 290         if (txChainFactory == null) {
 291             // stay with actual thread
 292             future = Futures.immediateCheckedFuture(null);
 293         } else if (wTx == null) {
 294             // hijack md-sal thread
 295             future = lastSubmittedFuture;
 296         } else {
 297             // hijack md-sal thread
 298             future = wTx.submit();
 299             wTx = null;
 300         }
 301         return future;
 302     }
 303
 304     @Override
 305     public void close() {
 306         LOG.debug("Setting transactionChainManagerStatus to SHUTTING_DOWN for {}, will wait for ownershipservice to notify"
 307                 , nodeII);
 308         Preconditions.checkState(TransactionChainManagerStatus.SHUTTING_DOWN.equals(transactionChainManagerStatus));
 309         Preconditions.checkState(wTx == null);
 310         synchronized (txLock) {
 311             if (txChainFactory != null) {
 312                 txChainFactory.close();
 313                 txChainFactory = null;
 314             }
 315         }
 316         Preconditions.checkState(txChainFactory == null);
 317     }
 318
 319     private enum TransactionChainManagerStatus {
 320         /** txChainManager is sleeping - is not active (SLAVE or default init value) */
 321         WORKING,
 322         /** txChainManager is working - is active (MASTER) */
 323         SLEEPING,
 324         /** txChainManager is trying to be closed - device disconnecting */
 325         SHUTTING_DOWN;
 326     }
 327 }