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