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