2 * Copyright (c) 2015, 2017 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.util.concurrent.ListenableFuture;
11 import io.netty.util.HashedWheelTimer;
13 import java.util.concurrent.ConcurrentHashMap;
14 import java.util.concurrent.ConcurrentMap;
15 import java.util.concurrent.ExecutorService;
16 import java.util.concurrent.ScheduledThreadPoolExecutor;
17 import java.util.concurrent.TimeUnit;
18 import org.eclipse.jdt.annotation.NonNull;
19 import org.opendaylight.mdsal.binding.api.DataBroker;
20 import org.opendaylight.mdsal.binding.api.NotificationPublishService;
21 import org.opendaylight.mdsal.binding.api.WriteTransaction;
22 import org.opendaylight.mdsal.common.api.LogicalDatastoreType;
23 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
24 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
25 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
26 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
27 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
28 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
29 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
30 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
31 import org.opendaylight.openflowplugin.api.openflow.lifecycle.ContextChainHolder;
32 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
33 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
34 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
35 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
36 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
37 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
38 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtil;
39 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
47 import org.opendaylight.yangtools.util.concurrent.QueuedNotificationManager;
48 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
49 import org.slf4j.Logger;
50 import org.slf4j.LoggerFactory;
52 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
54 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
55 private static final int SPY_RATE = 10;
57 private final OpenflowProviderConfig config;
58 private final DataBroker dataBroker;
59 private final DeviceInitializerProvider deviceInitializerProvider;
60 private final ConvertorExecutor convertorExecutor;
61 private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
62 private final Set<KeyedInstanceIdentifier<Node, NodeKey>> notificationCreateNodeSend =
63 ConcurrentHashMap.newKeySet();
64 private final NotificationPublishService notificationPublishService;
65 private final MessageSpy messageSpy;
66 private final HashedWheelTimer hashedWheelTimer;
67 private final Object updatePacketInRateLimitersLock = new Object();
68 private TranslatorLibrary translatorLibrary;
69 private ExtensionConverterProvider extensionConverterProvider;
70 private ScheduledThreadPoolExecutor spyPool;
71 private ContextChainHolder contextChainHolder;
72 private final QueuedNotificationManager<String, Runnable> queuedNotificationManager;
74 public DeviceManagerImpl(@NonNull final OpenflowProviderConfig config,
75 @NonNull final DataBroker dataBroker,
76 @NonNull final MessageSpy messageSpy,
77 @NonNull final NotificationPublishService notificationPublishService,
78 @NonNull final HashedWheelTimer hashedWheelTimer,
79 @NonNull final ConvertorExecutor convertorExecutor,
80 @NonNull final DeviceInitializerProvider deviceInitializerProvider,
81 @NonNull final ExecutorService executorService) {
83 this.dataBroker = dataBroker;
84 this.deviceInitializerProvider = deviceInitializerProvider;
85 this.convertorExecutor = convertorExecutor;
86 this.hashedWheelTimer = hashedWheelTimer;
87 this.spyPool = new ScheduledThreadPoolExecutor(1);
88 this.notificationPublishService = notificationPublishService;
89 this.messageSpy = messageSpy;
90 DeviceInitializationUtil.makeEmptyNodes(dataBroker);
91 this.queuedNotificationManager = QueuedNotificationManager.create(executorService, (key, entries) -> {
92 entries.forEach(Runnable::run);
93 }, 2048, "port-status-queue");
97 public TranslatorLibrary oook() {
98 return translatorLibrary;
102 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
103 this.translatorLibrary = translatorLibrary;
107 public void close() {
108 deviceContexts.values().forEach(OFPContext::close);
109 deviceContexts.clear();
110 if (spyPool != null) {
111 spyPool.shutdownNow();
117 public void initialize() {
118 spyPool.scheduleAtFixedRate(messageSpy, SPY_RATE, SPY_RATE, TimeUnit.SECONDS);
122 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
123 this.extensionConverterProvider = extensionConverterProvider;
127 public ExtensionConverterProvider getExtensionConverterProvider() {
128 return extensionConverterProvider;
132 public ListenableFuture<?> removeDeviceFromOperationalDS(@NonNull final KeyedInstanceIdentifier<Node, NodeKey> ii) {
134 final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
135 delWtx.delete(LogicalDatastoreType.OPERATIONAL, ii);
136 return delWtx.commit();
141 public DeviceContext createContext(@NonNull final ConnectionContext connectionContext) {
143 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
144 connectionContext.getConnectionAdapter().getRemoteAddress(),
145 connectionContext.getDeviceInfo().getNodeId());
147 connectionContext.getConnectionAdapter().setPacketInFiltering(true);
149 final OutboundQueueProvider outboundQueueProvider
150 = new OutboundQueueProviderImpl(connectionContext.getDeviceInfo().getVersion());
152 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
153 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
154 connectionContext.getConnectionAdapter().registerOutboundQueueHandler(
155 outboundQueueProvider,
156 config.getBarrierCountLimit().getValue().toJava(),
157 TimeUnit.MILLISECONDS.toNanos(config.getBarrierIntervalTimeoutLimit().getValue().toJava()));
158 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
161 final DeviceContext deviceContext = new DeviceContextImpl(
167 config.getSkipTableFeatures(),
169 config.getUseSingleLayerSerialization(),
170 deviceInitializerProvider,
171 config.getEnableFlowRemovedNotification(),
172 config.getSwitchFeaturesMandatory(),
174 queuedNotificationManager,
175 config.getIsStatisticsPollingOn());
176 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
177 deviceContext.setNotificationPublishService(notificationPublishService);
179 deviceContexts.put(connectionContext.getDeviceInfo(), deviceContext);
180 updatePacketInRateLimiters();
182 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
183 connectionContext.getConnectionAdapter(), deviceContext);
185 connectionContext.getConnectionAdapter().setMessageListener(messageListener);
186 connectionContext.getConnectionAdapter().setAlienMessageListener(messageListener);
188 return deviceContext;
191 private void updatePacketInRateLimiters() {
192 synchronized (updatePacketInRateLimitersLock) {
193 final int deviceContextsSize = deviceContexts.size();
194 if (deviceContextsSize > 0) {
195 long freshNotificationLimit = config.getGlobalNotificationQuota().toJava() / deviceContextsSize;
196 if (freshNotificationLimit < 100) {
197 freshNotificationLimit = 100;
199 if (LOG.isDebugEnabled()) {
200 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
202 for (final DeviceContext deviceContext : deviceContexts.values()) {
203 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
210 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
211 deviceContexts.remove(deviceInfo);
212 if (LOG.isDebugEnabled()) {
213 LOG.debug("Device context removed for node {}", deviceInfo);
215 this.updatePacketInRateLimiters();
219 public void sendNodeRemovedNotification(@NonNull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
220 if (notificationCreateNodeSend.remove(instanceIdentifier)) {
221 NodeRemovedBuilder builder = new NodeRemovedBuilder();
222 builder.setNodeRef(new NodeRef(instanceIdentifier));
223 LOG.info("Publishing node removed notification for {}", instanceIdentifier.firstKeyOf(Node.class).getId());
224 notificationPublishService.offerNotification(builder.build());
229 public void setContextChainHolder(@NonNull final ContextChainHolder contextChainHolder) {
230 this.contextChainHolder = contextChainHolder;
234 public void sendNodeAddedNotification(@NonNull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
235 if (!notificationCreateNodeSend.contains(instanceIdentifier)) {
236 notificationCreateNodeSend.add(instanceIdentifier);
237 final NodeId id = instanceIdentifier.firstKeyOf(Node.class).getId();
238 NodeUpdatedBuilder builder = new NodeUpdatedBuilder();
240 builder.setNodeRef(new NodeRef(instanceIdentifier));
241 LOG.info("Publishing node added notification for {}", id);
242 notificationPublishService.offerNotification(builder.build());