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 java.util.Objects;
  18 import java.util.Optional;
  19 import java.util.concurrent.CancellationException;
  20 import javax.annotation.Nonnull;
  21 import javax.annotation.Nullable;
  22 import javax.annotation.concurrent.GuardedBy;
  23 import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
  24 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  25 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
  26 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
  28 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  29 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
  30 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  31 import org.opendaylight.controller.md.sal.common.api.data.TransactionCommitFailedException;
  32 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
  33 import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
  34 import org.opendaylight.yangtools.yang.binding.DataObject;
  35 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
  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     private static final String CANNOT_WRITE_INTO_TRANSACTION = "Cannot write into transaction.";
  52
  53     private final Object txLock = new Object();
  54     private final DataBroker dataBroker;
  55     private final String nodeId;
  56     private LifecycleService lifecycleService;
  57
  58     @GuardedBy("txLock")
  59     private WriteTransaction wTx;
  60     @GuardedBy("txLock")
  61     private BindingTransactionChain txChainFactory;
  62     @GuardedBy("txLock")
  63     private boolean submitIsEnabled;
  64     @GuardedBy("txLock")
  65     private ListenableFuture<Void> lastSubmittedFuture;
  66
  67     private boolean initCommit;
  68
  69     @GuardedBy("txLock")
  70     private TransactionChainManagerStatus transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
  71
  72     TransactionChainManager(@Nonnull final DataBroker dataBroker,
  73                             @Nonnull final DeviceInfo deviceInfo) {
  74         this.dataBroker = dataBroker;
  75         this.nodeId = deviceInfo.getNodeInstanceIdentifier().getKey().getId().getValue();
  76         this.transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
  77         this.lastSubmittedFuture = Futures.immediateFuture(null);
  78     }
  79
  80     @GuardedBy("txLock")
  81     private void createTxChain() {
  82         Optional.ofNullable(txChainFactory).ifPresent(TransactionChain::close);
  83         txChainFactory = dataBroker.createTransactionChain(TransactionChainManager.this);
  84     }
  85
  86     public void setLifecycleService(final LifecycleService lifecycleService) {
  87         this.lifecycleService = lifecycleService;
  88     }
  89
  90     void initialSubmitWriteTransaction() {
  91         enableSubmit();
  92         submitWriteTransaction();
  93     }
  94
  95     /**
  96      * Method change status for TxChainManager to {@link TransactionChainManagerStatus#WORKING} and it has to make
  97      * registration for this class instance as {@link TransactionChainListener} to provide possibility a make DS
  98      * transactions. Call this method for MASTER role only.
  99      */
 100     void activateTransactionManager() {
 101         if (LOG.isDebugEnabled()) {
 102             LOG.debug("activateTransactionManager for node {} transaction submit is set to {}", this.nodeId, submitIsEnabled);
 103         }
 104         synchronized (txLock) {
 105             if (TransactionChainManagerStatus.SLEEPING.equals(transactionChainManagerStatus)) {
 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             }
 113         }
 114     }
 115
 116     /**
 117      * Method change status for TxChainManger to {@link TransactionChainManagerStatus#SLEEPING} and it unregisters
 118      * this class instance as {@link TransactionChainListener} so it broke a possibility to write something to DS.
 119      * Call this method for SLAVE only.
 120      * @return Future
 121      */
 122     ListenableFuture<Void> deactivateTransactionManager() {
 123         if (LOG.isDebugEnabled()) {
 124             LOG.debug("deactivateTransactionManager for node {}", this.nodeId);
 125         }
 126         final ListenableFuture<Void> future;
 127         synchronized (txLock) {
 128             if (TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus)) {
 129                 transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
 130                 future = txChainShuttingDown();
 131                 Preconditions.checkState(wTx == null, "We have some unexpected WriteTransaction.");
 132                 Futures.addCallback(future, new FutureCallback<Void>() {
 133                     @Override
 134                     public void onSuccess(final Void result) {
 135                         removeTxChainFactory();
 136                     }
 137
 138                     @Override
 139                     public void onFailure(final Throwable t) {
 140                         removeTxChainFactory();
 141                     }
 142                 });
 143             } else {
 144                 // TODO : ignoring redundant deactivate invocation
 145                 future = Futures.immediateCheckedFuture(null);
 146             }
 147         }
 148         return future;
 149     }
 150
 151     private void removeTxChainFactory() {
 152         Optional.ofNullable(txChainFactory).ifPresent(TransactionChain::close);
 153         txChainFactory = null;
 154     }
 155
 156     boolean submitWriteTransaction() {
 157         synchronized (txLock) {
 158             if (!submitIsEnabled) {
 159                 if (LOG.isTraceEnabled()) {
 160                     LOG.trace("transaction not committed - submit block issued");
 161                 }
 162                 return false;
 163             }
 164             if (Objects.isNull(wTx)) {
 165                 if (LOG.isTraceEnabled()) {
 166                     LOG.trace("nothing to commit - submit returns true");
 167                 }
 168                 return true;
 169             }
 170             Preconditions.checkState(TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus),
 171                     "we have here Uncompleted Transaction for node {} and we are not MASTER", this.nodeId);
 172             final CheckedFuture<Void, TransactionCommitFailedException> submitFuture = wTx.submit();
 173             Futures.addCallback(submitFuture, new FutureCallback<Void>() {
 174                 @Override
 175                 public void onSuccess(final Void result) {
 176                     if (initCommit) {
 177                         initCommit = false;
 178                     }
 179                 }
 180
 181                 @Override
 182                 public void onFailure(final Throwable t) {
 183                     if (t instanceof TransactionCommitFailedException) {
 184                         LOG.error("Transaction commit failed. ", t);
 185                     } else {
 186                         if (t instanceof CancellationException) {
 187                             LOG.warn("Submit task was canceled");
 188                             LOG.trace("Submit exception: ", t);
 189                         } else {
 190                             LOG.error("Exception during transaction submitting. ", t);
 191                         }
 192                     }
 193                     if (initCommit) {
 194                         wTx = null;
 195                         Optional.ofNullable(lifecycleService).ifPresent(LifecycleService::closeConnection);
 196                     }
 197                 }
 198             });
 199             lastSubmittedFuture = submitFuture;
 200             wTx = null;
 201         }
 202         return true;
 203     }
 204
 205     <T extends DataObject> void addDeleteOperationTotTxChain(final LogicalDatastoreType store,
 206                                                              final InstanceIdentifier<T> path){
 207         final WriteTransaction writeTx = getTransactionSafely();
 208         if (Objects.nonNull(writeTx)) {
 209             writeTx.delete(store, path);
 210         } else {
 211             if (LOG.isDebugEnabled()) {
 212                 LOG.debug("WriteTx is null for node {}. Delete {} was not realized.", this.nodeId, path);
 213             }
 214             throw new TransactionChainClosedException(CANNOT_WRITE_INTO_TRANSACTION);
 215         }
 216     }
 217
 218     <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 219                                                    final InstanceIdentifier<T> path,
 220                                                    final T data,
 221                                                    final boolean createParents){
 222         final WriteTransaction writeTx = getTransactionSafely();
 223         if (Objects.nonNull(writeTx)) {
 224             writeTx.put(store, path, data, createParents);
 225         } else {
 226             if (LOG.isDebugEnabled()) {
 227                 LOG.debug("WriteTx is null for node {}. Write data for {} was not realized.", this.nodeId, path);
 228             }
 229             throw new TransactionChainClosedException(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             synchronized (txLock) {
 257                 if (wTx == null && TransactionChainManagerStatus.WORKING.equals(transactionChainManagerStatus)) {
 258                     Optional.ofNullable(txChainFactory).ifPresent(bindingTransactionChain -> wTx = txChainFactory.newWriteOnlyTransaction());
 259                 }
 260             }
 261         return wTx;
 262     }
 263
 264     @VisibleForTesting
 265     void enableSubmit() {
 266         synchronized (txLock) {
 267             /* !!!IMPORTANT: never set true without txChainFactory */
 268             submitIsEnabled = txChainFactory != null;
 269         }
 270     }
 271
 272     ListenableFuture<Void> shuttingDown() {
 273         if (LOG.isDebugEnabled()) {
 274             LOG.debug("TxManager is going SHUTTING_DOWN for node {}", this.nodeId);
 275         }
 276         ListenableFuture<Void> future;
 277         synchronized (txLock) {
 278             this.transactionChainManagerStatus = TransactionChainManagerStatus.SHUTTING_DOWN;
 279             future = txChainShuttingDown();
 280         }
 281         return future;
 282     }
 283
 284     @GuardedBy("txLock")
 285     private ListenableFuture<Void> txChainShuttingDown() {
 286         submitIsEnabled = false;
 287         ListenableFuture<Void> future;
 288         if (txChainFactory == null) {
 289             // stay with actual thread
 290             future = Futures.immediateCheckedFuture(null);
 291         } else if (wTx == null) {
 292             // hijack md-sal thread
 293             future = lastSubmittedFuture;
 294         } else {
 295             if (LOG.isDebugEnabled()) {
 296                 LOG.debug("Submitting all transactions for Node {}", this.nodeId);
 297             }
 298             // hijack md-sal thread
 299             future = wTx.submit();
 300             wTx = null;
 301         }
 302         return future;
 303     }
 304
 305     @Override
 306     public void close() {
 307         if (LOG.isDebugEnabled()) {
 308             LOG.debug("Setting transactionChainManagerStatus to SHUTTING_DOWN for {}", this.nodeId);
 309         }
 310         synchronized (txLock) {
 311             removeTxChainFactory();
 312         }
 313     }
 314
 315     private enum TransactionChainManagerStatus {
 316         /** txChainManager is sleeping - is not active (SLAVE or default init value) */
 317         WORKING,
 318         /** txChainManager is working - is active (MASTER) */
 319         SLEEPING,
 320         /** txChainManager is trying to be closed - device disconnecting */
 321         SHUTTING_DOWN;
 322     }
 323 }