southbound/southbound-impl/src/main/java/org/opendaylight/ovsdb/southbound/transactions/md/TransactionInvokerImpl.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.ovsdb.southbound.transactions.md;
   9
  10 import com.google.common.annotations.VisibleForTesting;
  11 import com.google.common.util.concurrent.FutureCallback;
  12 import com.google.common.util.concurrent.Futures;
  13 import com.google.common.util.concurrent.MoreExecutors;
  14 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  15 import java.util.ArrayList;
  16 import java.util.Collections;
  17 import java.util.HashMap;
  18 import java.util.List;
  19 import java.util.Map;
  20 import java.util.concurrent.BlockingQueue;
  21 import java.util.concurrent.ExecutorService;
  22 import java.util.concurrent.Executors;
  23 import java.util.concurrent.LinkedBlockingQueue;
  24 import java.util.concurrent.ThreadFactory;
  25 import java.util.concurrent.TimeUnit;
  26 import java.util.concurrent.atomic.AtomicBoolean;
  27 import org.checkerframework.checker.lock.qual.GuardedBy;
  28 import org.opendaylight.controller.md.sal.binding.api.BindingTransactionChain;
  29 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  30 import org.opendaylight.controller.md.sal.binding.api.ReadWriteTransaction;
  31 import org.opendaylight.controller.md.sal.common.api.data.AsyncTransaction;
  32 import org.opendaylight.controller.md.sal.common.api.data.TransactionChain;
  33 import org.opendaylight.controller.md.sal.common.api.data.TransactionChainListener;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36
  37 public class TransactionInvokerImpl implements TransactionInvoker,TransactionChainListener, Runnable, AutoCloseable {
  38     private static final Logger LOG = LoggerFactory.getLogger(TransactionInvokerImpl.class);
  39     private static final int QUEUE_SIZE = 10000;
  40
  41     private final DataBroker db;
  42     private final BlockingQueue<TransactionCommand> inputQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
  43     private final BlockingQueue<AsyncTransaction<?, ?>> failedTransactionQueue = new LinkedBlockingQueue<>(QUEUE_SIZE);
  44     private final ExecutorService executor;
  45
  46     private final AtomicBoolean runTask = new AtomicBoolean(true);
  47
  48     @GuardedBy("this")
  49     private final Map<ReadWriteTransaction, TransactionCommand> transactionToCommand = new HashMap<>();
  50     @GuardedBy("this")
  51     private final List<ReadWriteTransaction> pendingTransactions = new ArrayList<>();
  52
  53     private BindingTransactionChain chain;
  54
  55     public TransactionInvokerImpl(final DataBroker db) {
  56         this.db = db;
  57         this.chain = db.createTransactionChain(this);
  58         ThreadFactory threadFact = new ThreadFactoryBuilder().setNameFormat("transaction-invoker-impl-%d").build();
  59         executor = Executors.newSingleThreadExecutor(threadFact);
  60         executor.execute(this);
  61     }
  62
  63     @VisibleForTesting
  64     TransactionInvokerImpl(final DataBroker db, final ExecutorService executor) {
  65         this.db = db;
  66         this.chain = db.createTransactionChain(this);
  67         this.executor = executor;
  68     }
  69
  70     @VisibleForTesting
  71     TransactionInvokerImpl(final DataBroker db, final List<ReadWriteTransaction> pendingTransactions,
  72             final List<ReadWriteTransaction> failedTransactions,
  73             final Map<ReadWriteTransaction, TransactionCommand> transactionToCommand) {
  74         this(db, (ExecutorService) null);
  75
  76         // Initialize state
  77         this.pendingTransactions.addAll(pendingTransactions);
  78         this.failedTransactionQueue.addAll(failedTransactions);
  79         this.transactionToCommand.putAll(transactionToCommand);
  80     }
  81
  82     @VisibleForTesting
  83     TransactionInvokerImpl(final DataBroker db, final List<ReadWriteTransaction> pendingTransactions) {
  84         this(db, pendingTransactions, Collections.emptyList(), Collections.emptyMap());
  85     }
  86
  87     @Override
  88     public void invoke(final TransactionCommand command) {
  89         // TODO what do we do if queue is full?
  90         if (!inputQueue.offer(command)) {
  91             LOG.error("inputQueue is full (size: {}) - could not offer {}", inputQueue.size(), command);
  92         }
  93     }
  94
  95     @Override
  96     public void onTransactionChainFailed(final TransactionChain<?, ?> chainArg,
  97             final AsyncTransaction<?, ?> transaction, final Throwable cause) {
  98         LOG.error("Failed to write operational topology", cause);
  99         offerFailedTransaction(transaction);
 100     }
 101
 102     @Override
 103     public void onTransactionChainSuccessful(final TransactionChain<?, ?> chainArg) {
 104         // NO OP
 105     }
 106
 107     @Override
 108     public void run() {
 109         while (runTask.get()) {
 110             final List<TransactionCommand> commands;
 111             try {
 112                 commands = extractCommands();
 113             } catch (InterruptedException e) {
 114                 LOG.warn("Extracting commands was interrupted.", e);
 115                 continue;
 116             }
 117
 118             ReadWriteTransaction transactionInFlight = null;
 119             try {
 120                 for (TransactionCommand command: commands) {
 121                     synchronized (this) {
 122                         final ReadWriteTransaction transaction = chain.newReadWriteTransaction();
 123                         transactionInFlight = transaction;
 124                         recordPendingTransaction(command, transaction);
 125                         command.execute(transaction);
 126                         Futures.addCallback(transaction.submit(), new FutureCallback<Void>() {
 127                             @Override
 128                             public void onSuccess(final Void result) {
 129                                 forgetSuccessfulTransaction(transaction);
 130                                 command.onSuccess();
 131                             }
 132
 133                             @Override
 134                             public void onFailure(final Throwable throwable) {
 135                                 command.onFailure(throwable);
 136                                 // NOOP - handled by failure of transaction chain
 137                             }
 138                         }, MoreExecutors.directExecutor());
 139                     }
 140                 }
 141             } catch (IllegalStateException e) {
 142                 if (transactionInFlight != null) {
 143                     // TODO: This method should distinguish exceptions on which the command should be
 144                     // retried from exceptions on which the command should NOT be retried.
 145                     // Then it should retry only the commands which should be retried, otherwise
 146                     // this method will retry commands which will never be successful forever.
 147                     offerFailedTransaction(transactionInFlight);
 148                 }
 149                 LOG.warn("Failed to process an update notification from OVS.", e);
 150             }
 151         }
 152     }
 153
 154     private void offerFailedTransaction(final AsyncTransaction<?, ?> transaction) {
 155         if (!failedTransactionQueue.offer(transaction)) {
 156             LOG.warn("failedTransactionQueue is full (size: {})", failedTransactionQueue.size());
 157         }
 158     }
 159
 160     @VisibleForTesting
 161     synchronized List<TransactionCommand> extractResubmitCommands() {
 162         AsyncTransaction<?, ?> transaction = failedTransactionQueue.poll();
 163         List<TransactionCommand> commands = new ArrayList<>();
 164         if (transaction != null) {
 165             int index = pendingTransactions.lastIndexOf(transaction);
 166             List<ReadWriteTransaction> transactions =
 167                     pendingTransactions.subList(index, pendingTransactions.size() - 1);
 168             for (ReadWriteTransaction tx: transactions) {
 169                 commands.add(transactionToCommand.get(tx));
 170             }
 171             resetTransactionQueue();
 172         }
 173         return commands;
 174     }
 175
 176     @VisibleForTesting
 177     synchronized void resetTransactionQueue() {
 178         chain.close();
 179         chain = db.createTransactionChain(this);
 180         pendingTransactions.clear();
 181         transactionToCommand.clear();
 182         failedTransactionQueue.clear();
 183     }
 184
 185     synchronized void forgetSuccessfulTransaction(final ReadWriteTransaction transaction) {
 186         pendingTransactions.remove(transaction);
 187         transactionToCommand.remove(transaction);
 188     }
 189
 190     @VisibleForTesting
 191     synchronized void recordPendingTransaction(final TransactionCommand command,
 192             final ReadWriteTransaction transaction) {
 193         transactionToCommand.put(transaction, command);
 194         pendingTransactions.add(transaction);
 195     }
 196
 197     @VisibleForTesting
 198     List<TransactionCommand> extractCommands() throws InterruptedException {
 199         List<TransactionCommand> commands = extractResubmitCommands();
 200         commands.addAll(extractCommandsFromQueue());
 201         return commands;
 202     }
 203
 204     @VisibleForTesting
 205     List<TransactionCommand> extractCommandsFromQueue() throws InterruptedException {
 206         List<TransactionCommand> result = new ArrayList<>();
 207         TransactionCommand command = inputQueue.take();
 208         result.add(command);
 209         inputQueue.drainTo(result);
 210         return result;
 211     }
 212
 213     @Override
 214     public void close() throws InterruptedException {
 215         this.chain.close();
 216         this.executor.shutdown();
 217         if (!this.executor.awaitTermination(1, TimeUnit.SECONDS)) {
 218             runTask.set(false);
 219             this.executor.shutdownNow();
 220         }
 221     }
 222 }