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