2 * Copyright (c) 2015 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.impl.device;
10 import com.google.common.base.Preconditions;
11 import com.google.common.base.Verify;
12 import com.google.common.collect.Iterators;
13 import io.netty.util.HashedWheelTimer;
14 import java.util.Collections;
15 import java.util.Iterator;
16 import java.util.Map.Entry;
17 import java.util.concurrent.ConcurrentHashMap;
18 import java.util.concurrent.ConcurrentMap;
19 import java.util.concurrent.ExecutionException;
20 import java.util.concurrent.ScheduledThreadPoolExecutor;
21 import java.util.concurrent.TimeUnit;
22 import javax.annotation.CheckForNull;
23 import javax.annotation.Nonnull;
24 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
25 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
26 import org.opendaylight.controller.md.sal.binding.api.NotificationService;
27 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
28 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
29 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
30 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
31 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
32 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
34 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
35 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
36 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
37 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler;
38 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageIntelligenceAgency;
39 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
40 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
41 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
42 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.Nodes;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodesBuilder;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
47 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
48 import org.slf4j.Logger;
49 import org.slf4j.LoggerFactory;
54 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
56 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
58 private static final long TICK_DURATION = 10; // 0.5 sec.
59 private final long globalNotificationQuota;
60 private final boolean switchFeaturesMandatory;
62 private ScheduledThreadPoolExecutor spyPool;
63 private final int spyRate = 10;
65 private final DataBroker dataBroker;
66 private final HashedWheelTimer hashedWheelTimer;
67 private TranslatorLibrary translatorLibrary;
68 private DeviceInitializationPhaseHandler deviceInitPhaseHandler;
69 private NotificationService notificationService;
70 private NotificationPublishService notificationPublishService;
72 private final ConcurrentMap<NodeId, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
73 private final MessageIntelligenceAgency messageIntelligenceAgency;
75 private final long barrierIntervalNanos;
76 private final int barrierCountLimit;
77 private ExtensionConverterProvider extensionConverterProvider;
79 public DeviceManagerImpl(@Nonnull final DataBroker dataBroker,
80 @Nonnull final MessageIntelligenceAgency messageIntelligenceAgency,
81 final long globalNotificationQuota, final boolean switchFeaturesMandatory,
82 final long barrierInterval, final int barrierCountLimit) {
83 this.switchFeaturesMandatory = switchFeaturesMandatory;
84 this.globalNotificationQuota = globalNotificationQuota;
85 this.dataBroker = Preconditions.checkNotNull(dataBroker);
86 hashedWheelTimer = new HashedWheelTimer(TICK_DURATION, TimeUnit.MILLISECONDS, 500);
87 /* merge empty nodes to oper DS to predict any problems with missing parent for Node */
88 final WriteTransaction tx = dataBroker.newWriteOnlyTransaction();
90 final NodesBuilder nodesBuilder = new NodesBuilder();
91 nodesBuilder.setNode(Collections.<Node>emptyList());
92 tx.merge(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(Nodes.class), nodesBuilder.build());
95 } catch (ExecutionException | InterruptedException e) {
96 LOG.error("Creation of node failed.", e);
97 throw new IllegalStateException(e);
100 this.messageIntelligenceAgency = messageIntelligenceAgency;
101 this.barrierIntervalNanos = TimeUnit.MILLISECONDS.toNanos(barrierInterval);
102 this.barrierCountLimit = barrierCountLimit;
107 public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) {
108 deviceInitPhaseHandler = handler;
112 public void onDeviceContextLevelUp(final DeviceContext deviceContext) throws Exception {
113 // final phase - we have to add new Device to MD-SAL DataStore
114 LOG.debug("Final phase of DeviceContextLevelUp for Node: {} ", deviceContext.getDeviceState().getNodeId());
115 Preconditions.checkNotNull(deviceContext);
116 ((DeviceContextImpl) deviceContext).initialSubmitTransaction();
117 deviceContext.onPublished();
121 public void deviceConnected(@CheckForNull final ConnectionContext connectionContext) throws Exception {
122 Preconditions.checkArgument(connectionContext != null);
123 Preconditions.checkState(!deviceContexts.containsKey(connectionContext.getNodeId()),
124 "Rejecting connection from node which is already connected and there exist deviceContext for it: {}",
125 connectionContext.getNodeId()
127 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
128 connectionContext.getConnectionAdapter().getRemoteAddress(), connectionContext.getNodeId());
130 // Cache this for clarity
131 final ConnectionAdapter connectionAdapter = connectionContext.getConnectionAdapter();
133 //FIXME: as soon as auxiliary connection are fully supported then this is needed only before device context published
134 connectionAdapter.setPacketInFiltering(true);
136 final Short version = connectionContext.getFeatures().getVersion();
137 final OutboundQueueProvider outboundQueueProvider = new OutboundQueueProviderImpl(version);
139 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
140 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
141 connectionAdapter.registerOutboundQueueHandler(outboundQueueProvider, barrierCountLimit, barrierIntervalNanos);
142 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
144 final DeviceState deviceState = createDeviceState(connectionContext);
145 final DeviceContext deviceContext = new DeviceContextImpl(connectionContext, deviceState, dataBroker,
146 hashedWheelTimer, messageIntelligenceAgency, outboundQueueProvider, translatorLibrary, switchFeaturesMandatory);
148 Verify.verify(deviceContexts.putIfAbsent(connectionContext.getNodeId(), deviceContext) == null, "DeviceCtx still not closed.");
149 deviceContext.addDeviceContextClosedHandler(this);
151 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
152 deviceContext.setNotificationService(notificationService);
153 deviceContext.setNotificationPublishService(notificationPublishService);
155 updatePacketInRateLimiters();
157 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
158 connectionAdapter, deviceContext);
159 connectionAdapter.setMessageListener(messageListener);
161 deviceCtxLevelUp(deviceContext);
164 private static DeviceStateImpl createDeviceState(final @Nonnull ConnectionContext connectionContext) {
165 return new DeviceStateImpl(connectionContext.getFeatures(), connectionContext.getNodeId());
168 private void updatePacketInRateLimiters() {
169 synchronized (deviceContexts) {
170 final int deviceContextsSize = deviceContexts.size();
171 if (deviceContextsSize > 0) {
172 long freshNotificationLimit = globalNotificationQuota / deviceContextsSize;
173 if (freshNotificationLimit < 100) {
174 freshNotificationLimit = 100;
176 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
177 for (final DeviceContext deviceContext : deviceContexts.values()) {
178 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
184 private void deviceCtxLevelUp(final DeviceContext deviceContext) throws Exception {
185 deviceContext.getDeviceState().setValid(true);
186 LOG.trace("Device context level up called.");
187 deviceInitPhaseHandler.onDeviceContextLevelUp(deviceContext);
191 public TranslatorLibrary oook() {
192 return translatorLibrary;
196 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
197 this.translatorLibrary = translatorLibrary;
201 public void setNotificationService(final NotificationService notificationServiceParam) {
202 notificationService = notificationServiceParam;
206 public void setNotificationPublishService(final NotificationPublishService notificationService) {
207 notificationPublishService = notificationService;
211 public void close() {
212 for (final Iterator<Entry<NodeId, DeviceContext>> iterator = Iterators
213 .consumingIterator(deviceContexts.entrySet().iterator()); iterator.hasNext();) {
214 iterator.next().getValue().close();
217 if (spyPool != null) {
218 spyPool.shutdownNow();
224 public void onDeviceContextClosed(final DeviceContext deviceContext) {
225 LOG.trace("onDeviceContextClosed for Node {}", deviceContext.getDeviceState().getNodeId());
226 deviceContexts.remove(deviceContext.getPrimaryConnectionContext().getNodeId());
227 updatePacketInRateLimiters();
231 public void initialize() {
232 spyPool = new ScheduledThreadPoolExecutor(1);
233 spyPool.scheduleAtFixedRate(messageIntelligenceAgency, spyRate, spyRate, TimeUnit.SECONDS);
237 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
238 this.extensionConverterProvider = extensionConverterProvider;
242 public ExtensionConverterProvider getExtensionConverterProvider() {
243 return extensionConverterProvider;