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 javax.annotation.PostConstruct;
13 import javax.annotation.PreDestroy;
14 import javax.inject.Inject;
15 import javax.inject.Singleton;
16 import org.opendaylight.mdsal.binding.api.DataBroker;
17 import org.opendaylight.openflowplugin.common.txchain.TransactionChainManager;
18 import org.slf4j.Logger;
19 import org.slf4j.LoggerFactory;
22 public final class OperationProcessor implements AutoCloseable, Runnable {
23 private static final Logger LOG = LoggerFactory.getLogger(OperationProcessor.class);
24 private static final int MAX_TRANSACTION_OPERATIONS = 100;
25 private static final int OPERATION_QUEUE_DEPTH = 500;
26 private static final String TOPOLOGY_MANAGER = "ofp-topo-processor";
28 private final BlockingQueue<TopologyOperation> queue = new LinkedBlockingQueue<>(OPERATION_QUEUE_DEPTH);
29 private final Thread thread;
30 private final TransactionChainManager transactionChainManager;
31 private volatile boolean finishing = false;
34 public OperationProcessor(final DataBroker dataBroker) {
35 transactionChainManager = new TransactionChainManager(dataBroker, TOPOLOGY_MANAGER);
36 transactionChainManager.activateTransactionManager();
37 transactionChainManager.initialSubmitWriteTransaction();
39 thread = new Thread(this);
40 thread.setDaemon(true);
41 thread.setName("ofp-topo-expo-" + FlowCapableTopologyProvider.TOPOLOGY_ID);
44 void enqueueOperation(final TopologyOperation task) {
47 } catch (InterruptedException e) {
48 LOG.warn("Interrupted while submitting task {}", task, e);
61 TopologyOperation op = queue.take();
63 LOG.debug("New {} operation available, starting transaction", op);
67 op.applyOperation(transactionChainManager);
70 if (ops < MAX_TRANSACTION_OPERATIONS) {
76 LOG.debug("Next operation {}", op);
79 LOG.debug("Processed {} operations, submitting transaction", ops);
80 if (!transactionChainManager.submitTransaction()) {
81 cleanDataStoreOperQueue();
83 } catch (final InterruptedException e) {
84 // This should mean we're shutting down.
85 LOG.debug("Stat Manager DS Operation thread interrupted!", e);
89 // Drain all events, making sure any blocked threads are unblocked
90 cleanDataStoreOperQueue();
93 private void cleanDataStoreOperQueue() {
94 while (!queue.isEmpty()) {
101 public void close() {
105 } catch (InterruptedException e) {
106 LOG.debug("Join of thread {} was interrupted", thread.getName(), e);
109 transactionChainManager.close();
111 LOG.debug("OperationProcessor closed");