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