applications/topology-manager/src/main/java/org/opendaylight/openflowplugin/applications/topology/manager/OperationProcessor.java

   1 /*
   2  * Copyright (c) 2014 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.applications.topology.manager;
   9
  10 import java.util.concurrent.BlockingQueue;
  11 import java.util.concurrent.LinkedBlockingQueue;
  12 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
  13 import org.opendaylight.openflowplugin.common.txchain.TransactionChainManager;
  14 import org.slf4j.Logger;
  15 import org.slf4j.LoggerFactory;
  16
  17 public final class OperationProcessor implements AutoCloseable, Runnable {
  18     private static final Logger LOG = LoggerFactory.getLogger(OperationProcessor.class);
  19     private static final int MAX_TRANSACTION_OPERATIONS = 100;
  20     private static final int OPERATION_QUEUE_DEPTH = 500;
  21     private static final String TOPOLOGY_MANAGER = "topology-manager";
  22
  23     private final BlockingQueue<TopologyOperation> queue = new LinkedBlockingQueue<>(OPERATION_QUEUE_DEPTH);
  24     private final Thread thread;
  25     private TransactionChainManager transactionChainManager;
  26     private volatile boolean finishing = false;
  27
  28     public OperationProcessor(final DataBroker dataBroker) {
  29         transactionChainManager = new TransactionChainManager(dataBroker, TOPOLOGY_MANAGER);
  30         transactionChainManager.activateTransactionManager();
  31         transactionChainManager.initialSubmitWriteTransaction();
  32
  33         thread = new Thread(this);
  34         thread.setDaemon(true);
  35         thread.setName("FlowCapableTopologyExporter-" + FlowCapableTopologyProvider.TOPOLOGY_ID);
  36     }
  37
  38     void enqueueOperation(final TopologyOperation task) {
  39         try {
  40             queue.put(task);
  41         } catch (InterruptedException e) {
  42             LOG.warn("Interrupted while submitting task {}", task, e);
  43         }
  44     }
  45
  46     public void start() {
  47         thread.start();
  48     }
  49
  50     @Override
  51     public void run() {
  52             while (!finishing) {
  53                 try {
  54                     TopologyOperation op = queue.take();
  55
  56                     LOG.debug("New {} operation available, starting transaction", op);
  57
  58                     int ops = 0;
  59                     do {
  60                         op.applyOperation(transactionChainManager);
  61
  62                         ops++;
  63                         if (ops < MAX_TRANSACTION_OPERATIONS) {
  64                             op = queue.poll();
  65                         } else {
  66                             op = null;
  67                         }
  68
  69                         LOG.debug("Next operation {}", op);
  70                     } while (op != null);
  71
  72                     LOG.debug("Processed {} operations, submitting transaction", ops);
  73                     if (!transactionChainManager.submitTransaction()) {
  74                         cleanDataStoreOperQueue();
  75                     }
  76                 } catch (final InterruptedException e) {
  77                     // This should mean we're shutting down.
  78                     LOG.debug("Stat Manager DS Operation thread interrupted!", e);
  79                     finishing = true;
  80                 }
  81             }
  82         // Drain all events, making sure any blocked threads are unblocked
  83         cleanDataStoreOperQueue();
  84     }
  85
  86     private void cleanDataStoreOperQueue() {
  87         while (!queue.isEmpty()) {
  88             queue.poll();
  89         }
  90     }
  91
  92     @Override
  93     public void close() {
  94         thread.interrupt();
  95         try {
  96             thread.join();
  97         } catch(InterruptedException e) {
  98             LOG.debug("Join of thread {} was interrupted", thread.getName(), e);
  99         }
 100
 101         transactionChainManager.close();
 102
 103         LOG.debug("OperationProcessor closed");
 104     }
 105 }