2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
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
8 package org.opendaylight.openflowplugin.applications.topology.manager;
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;
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";
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;
28 public OperationProcessor(final DataBroker dataBroker) {
29 transactionChainManager = new TransactionChainManager(dataBroker, TOPOLOGY_MANAGER);
30 transactionChainManager.activateTransactionManager();
31 transactionChainManager.initialSubmitWriteTransaction();
33 thread = new Thread(this);
34 thread.setDaemon(true);
35 thread.setName("FlowCapableTopologyExporter-" + FlowCapableTopologyProvider.TOPOLOGY_ID);
38 void enqueueOperation(final TopologyOperation task) {
41 } catch (InterruptedException e) {
42 LOG.warn("Interrupted while submitting task {}", task, e);
54 TopologyOperation op = queue.take();
56 LOG.debug("New {} operation available, starting transaction", op);
60 op.applyOperation(transactionChainManager);
63 if (ops < MAX_TRANSACTION_OPERATIONS) {
69 LOG.debug("Next operation {}", op);
72 LOG.debug("Processed {} operations, submitting transaction", ops);
73 if (!transactionChainManager.submitTransaction()) {
74 cleanDataStoreOperQueue();
76 } catch (final InterruptedException e) {
77 // This should mean we're shutting down.
78 LOG.debug("Stat Manager DS Operation thread interrupted!", e);
82 // Drain all events, making sure any blocked threads are unblocked
83 cleanDataStoreOperQueue();
86 private void cleanDataStoreOperQueue() {
87 while (!queue.isEmpty()) {
97 } catch(InterruptedException e) {
98 LOG.debug("Join of thread {} was interrupted", thread.getName(), e);
101 transactionChainManager.close();
103 LOG.debug("OperationProcessor closed");