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