openflowplugin-common/src/main/java/org/opendaylight/openflowplugin/common/txchain/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 package org.opendaylight.openflowplugin.common.txchain;
   9
  10 import com.google.common.base.Preconditions;
  11 import com.google.common.util.concurrent.FluentFuture;
  12 import com.google.common.util.concurrent.FutureCallback;
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.MoreExecutors;
  15 import java.util.Optional;
  16 import java.util.concurrent.CancellationException;
  17 import java.util.concurrent.ExecutionException;
  18 import java.util.concurrent.TimeUnit;
  19 import java.util.concurrent.TimeoutException;
  20 import java.util.concurrent.locks.ReadWriteLock;
  21 import java.util.concurrent.locks.ReentrantReadWriteLock;
  22 import org.checkerframework.checker.lock.qual.GuardedBy;
  23 import org.checkerframework.checker.lock.qual.Holding;
  24 import org.eclipse.jdt.annotation.NonNull;
  25 import org.opendaylight.mdsal.binding.api.DataBroker;
  26 import org.opendaylight.mdsal.binding.api.ReadWriteTransaction;
  27 import org.opendaylight.mdsal.binding.api.Transaction;
  28 import org.opendaylight.mdsal.binding.api.TransactionChain;
  29 import org.opendaylight.mdsal.binding.api.TransactionChainClosedException;
  30 import org.opendaylight.mdsal.binding.api.TransactionChainListener;
  31 import org.opendaylight.mdsal.binding.api.WriteTransaction;
  32 import org.opendaylight.mdsal.common.api.CommitInfo;
  33 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
  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  * The openflowplugin-impl.org.opendaylight.openflowplugin.impl.device
  41  * package protected class for controlling {@link WriteTransaction} life cycle. It is
  42  * a {@link TransactionChainListener} and provide package protected methods for writeToTransaction
  43  * method (wrapped {@link WriteTransaction#put(LogicalDatastoreType, InstanceIdentifier, DataObject)})
  44  * and submitTransaction method (wrapped {@link WriteTransaction#commit()}).
  45  */
  46 public class TransactionChainManager implements TransactionChainListener, AutoCloseable {
  47
  48     private static final Logger LOG = LoggerFactory.getLogger(TransactionChainManager.class);
  49     private static final String CANNOT_WRITE_INTO_TRANSACTION = "Cannot write into transaction.";
  50
  51     private final Object txLock = new Object();
  52     private final DataBroker dataBroker;
  53     private final String nodeId;
  54
  55     @GuardedBy("txLock")
  56     private ReadWriteTransaction writeTx;
  57     @GuardedBy("txLock")
  58     private TransactionChain transactionChain;
  59     @GuardedBy("txLock")
  60     private boolean submitIsEnabled;
  61     @GuardedBy("txLock")
  62     private FluentFuture<? extends CommitInfo> lastSubmittedFuture;
  63
  64     private volatile boolean initCommit;
  65
  66     @GuardedBy("txLock")
  67     private TransactionChainManagerStatus transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
  68     private ReadWriteLock readWriteTransactionLock = new ReentrantReadWriteLock();
  69
  70     public TransactionChainManager(@NonNull final DataBroker dataBroker,
  71                                    @NonNull final String deviceIdentifier) {
  72         this.dataBroker = dataBroker;
  73         this.nodeId = deviceIdentifier;
  74         this.lastSubmittedFuture = CommitInfo.emptyFluentFuture();
  75     }
  76
  77     @Holding("txLock")
  78     private void createTxChain() {
  79         TransactionChain txChainFactoryTemp = transactionChain;
  80         transactionChain = dataBroker.createTransactionChain(TransactionChainManager.this);
  81         if (txChainFactoryTemp != null) {
  82             txChainFactoryTemp.close();
  83         }
  84     }
  85
  86     public boolean initialSubmitWriteTransaction() {
  87         enableSubmit();
  88         return submitTransaction();
  89     }
  90
  91     /**
  92      * Method change status for TxChainManager to 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     public 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(transactionChain == null,
 104                         "TxChainFactory survive last close.");
 105                 Preconditions.checkState(writeTx == 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 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     public FluentFuture<?> deactivateTransactionManager() {
 122         if (LOG.isDebugEnabled()) {
 123             LOG.debug("deactivateTransactionManager for node {}", this.nodeId);
 124         }
 125         final FluentFuture<? extends CommitInfo> future;
 126         synchronized (txLock) {
 127             if (TransactionChainManagerStatus.WORKING == transactionChainManagerStatus) {
 128                 transactionChainManagerStatus = TransactionChainManagerStatus.SLEEPING;
 129                 future =  txChainShuttingDown();
 130                 Preconditions.checkState(writeTx == null,
 131                         "We have some unexpected WriteTransaction.");
 132                 future.addCallback(new FutureCallback<CommitInfo>() {
 133                     @Override
 134                     public void onSuccess(final CommitInfo result) {
 135                         closeTransactionChain();
 136                     }
 137
 138                     @Override
 139                     public void onFailure(@NonNull final Throwable throwable) {
 140                         closeTransactionChain();
 141                     }
 142                 }, MoreExecutors.directExecutor());
 143             } else {
 144                 // ignoring redundant deactivate invocation
 145                 future = CommitInfo.emptyFluentFuture();
 146             }
 147         }
 148         return future;
 149     }
 150
 151     private void closeTransactionChain() {
 152         if (writeTx != null) {
 153             writeTx.cancel();
 154             writeTx = null;
 155         }
 156         if (transactionChain != null) {
 157             transactionChain.close();
 158             transactionChain = null;
 159         }
 160     }
 161
 162     @GuardedBy("txLock")
 163     public boolean submitTransaction() {
 164         return submitTransaction(false);
 165     }
 166
 167     @GuardedBy("txLock")
 168     public boolean submitTransaction(boolean doSync) {
 169         synchronized (txLock) {
 170             if (!submitIsEnabled) {
 171                 LOG.trace("transaction not committed - submit block issued");
 172                 return false;
 173             }
 174             if (writeTx == null) {
 175                 LOG.trace("nothing to commit - submit returns true");
 176                 return true;
 177             }
 178             Preconditions.checkState(TransactionChainManagerStatus.WORKING == transactionChainManagerStatus,
 179                     "we have here Uncompleted Transaction for node {} and we are not MASTER",
 180                     this.nodeId);
 181             final FluentFuture<? extends CommitInfo> submitFuture = writeTx.commit();
 182             lastSubmittedFuture = submitFuture;
 183             writeTx = null;
 184
 185             if (initCommit || doSync) {
 186                 try {
 187                     submitFuture.get(5L, TimeUnit.SECONDS);
 188                 } catch (InterruptedException | ExecutionException | TimeoutException ex) {
 189                     LOG.error("Exception during INITIAL({}) || doSync({}) transaction submitting. ",
 190                             initCommit, doSync, ex);
 191                     return false;
 192                 }
 193                 initCommit = false;
 194                 return true;
 195             }
 196
 197             submitFuture.addCallback(new FutureCallback<CommitInfo>() {
 198                 @Override
 199                 public void onSuccess(final CommitInfo result) {
 200                     //NOOP
 201                 }
 202
 203                 @Override
 204                 public void onFailure(final Throwable throwable) {
 205                     if (throwable instanceof InterruptedException || throwable instanceof ExecutionException) {
 206                         LOG.error("Transaction commit failed. ", throwable);
 207                     } else {
 208                         if (throwable instanceof CancellationException) {
 209                             LOG.warn("Submit task was canceled");
 210                             LOG.trace("Submit exception: ", throwable);
 211                         } else {
 212                             LOG.error("Exception during transaction submitting. ", throwable);
 213                         }
 214                     }
 215                 }
 216             }, MoreExecutors.directExecutor());
 217         }
 218         return true;
 219     }
 220
 221     public <T extends DataObject> void addDeleteOperationToTxChain(final LogicalDatastoreType store,
 222                                                                     final InstanceIdentifier<T> path) {
 223         synchronized (txLock) {
 224             ensureTransaction();
 225             if (writeTx == null) {
 226                 LOG.debug("WriteTx is null for node {}. Delete {} was not realized.", this.nodeId, path);
 227                 throw new TransactionChainClosedException(CANNOT_WRITE_INTO_TRANSACTION);
 228             }
 229
 230             writeTx.delete(store, path);
 231         }
 232     }
 233
 234     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
 235                                                           final InstanceIdentifier<T> path,
 236                                                           final T data,
 237                                                           final boolean createParents) {
 238         synchronized (txLock) {
 239             ensureTransaction();
 240             if (writeTx == null) {
 241                 LOG.debug("WriteTx is null for node {}. Write data for {} was not realized.", this.nodeId, path);
 242                 throw new TransactionChainClosedException(CANNOT_WRITE_INTO_TRANSACTION);
 243             }
 244
 245             writeTx.put(store, path, data, createParents);
 246         }
 247     }
 248
 249     public <T extends DataObject> void mergeToTransaction(final LogicalDatastoreType store,
 250                                                           final InstanceIdentifier<T> path,
 251                                                           final T data,
 252                                                           final boolean createParents) {
 253         synchronized (txLock) {
 254             ensureTransaction();
 255             if (writeTx == null) {
 256                 LOG.debug("WriteTx is null for node {}. Merge data for {} was not realized.", this.nodeId, path);
 257                 throw new TransactionChainClosedException(CANNOT_WRITE_INTO_TRANSACTION);
 258             }
 259
 260             writeTx.merge(store, path, data, createParents);
 261         }
 262     }
 263
 264     public <T extends DataObject> ListenableFuture<Optional<T>>
 265         readFromTransaction(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
 266         synchronized (txLock) {
 267             ensureTransaction();
 268             if (writeTx == null) {
 269                 LOG.debug("WriteTx is null for node {}. Read data for {} was not realized.", this.nodeId, path);
 270                 throw new TransactionChainClosedException(CANNOT_WRITE_INTO_TRANSACTION);
 271             }
 272
 273             return writeTx.read(store, path);
 274         }
 275     }
 276
 277     @Override
 278     public void onTransactionChainFailed(final TransactionChain chain,
 279                                          final Transaction transaction, final Throwable cause) {
 280         synchronized (txLock) {
 281             if (TransactionChainManagerStatus.WORKING == transactionChainManagerStatus
 282                     && chain.equals(this.transactionChain)) {
 283                 LOG.warn("Transaction chain failed, recreating chain due to ", cause);
 284                 closeTransactionChain();
 285                 createTxChain();
 286                 writeTx = null;
 287             }
 288         }
 289     }
 290
 291     @Override
 292     public void onTransactionChainSuccessful(final TransactionChain chain) {
 293         // NOOP
 294     }
 295
 296     @Holding("txLock")
 297     private void ensureTransaction() {
 298         if (writeTx == null && TransactionChainManagerStatus.WORKING == transactionChainManagerStatus
 299                 && transactionChain != null) {
 300             writeTx = transactionChain.newReadWriteTransaction();
 301         }
 302     }
 303
 304     private void enableSubmit() {
 305         synchronized (txLock) {
 306             /* !!!IMPORTANT: never set true without transactionChain */
 307             submitIsEnabled = transactionChain != null;
 308         }
 309     }
 310
 311     public FluentFuture<?> shuttingDown() {
 312         if (LOG.isDebugEnabled()) {
 313             LOG.debug("TxManager is going SHUTTING_DOWN for node {}", this.nodeId);
 314         }
 315         synchronized (txLock) {
 316             this.transactionChainManagerStatus = TransactionChainManagerStatus.SHUTTING_DOWN;
 317             return txChainShuttingDown();
 318         }
 319     }
 320
 321     @GuardedBy("txLock")
 322     private FluentFuture<? extends CommitInfo> txChainShuttingDown() {
 323         boolean wasSubmitEnabled = submitIsEnabled;
 324         submitIsEnabled = false;
 325         FluentFuture<? extends CommitInfo> future;
 326
 327         if (!wasSubmitEnabled || transactionChain == null) {
 328             // stay with actual thread
 329             future = CommitInfo.emptyFluentFuture();
 330
 331             if (writeTx != null) {
 332                 writeTx.cancel();
 333                 writeTx = null;
 334             }
 335         } else if (writeTx == null) {
 336             // hijack md-sal thread
 337             future = lastSubmittedFuture;
 338         } else {
 339             if (LOG.isDebugEnabled()) {
 340                 LOG.debug("Submitting all transactions for Node {}", this.nodeId);
 341             }
 342             // hijack md-sal thread
 343             future = writeTx.commit();
 344             writeTx = null;
 345         }
 346
 347         return future;
 348     }
 349
 350     @Override
 351     public void close() {
 352         if (LOG.isDebugEnabled()) {
 353             LOG.debug("Setting transactionChainManagerStatus to SHUTTING_DOWN for {}", this.nodeId);
 354         }
 355         synchronized (txLock) {
 356             closeTransactionChain();
 357         }
 358     }
 359
 360     private enum TransactionChainManagerStatus {
 361         /**
 362          * txChainManager is working - is active (MASTER).
 363          */
 364         WORKING,
 365         /**
 366          * txChainManager is sleeping - is not active (SLAVE or default init value).
 367          */
 368         SLEEPING,
 369         /**
 370          * txChainManager is trying to be closed - device disconnecting.
 371          */
 372         SHUTTING_DOWN
 373     }
 374
 375     public void acquireWriteTransactionLock() {
 376         readWriteTransactionLock.writeLock().lock();
 377     }
 378
 379     public void releaseWriteTransactionLock() {
 380         readWriteTransactionLock.writeLock().unlock();
 381     }
 382
 383 }