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.FutureCallback;
11 import com.google.common.util.concurrent.Futures;
12 import com.google.common.util.concurrent.ListenableFuture;
13 import com.google.common.util.concurrent.MoreExecutors;
14 import io.netty.util.HashedWheelTimer;
15 import io.netty.util.internal.ConcurrentSet;
16 import java.util.Optional;
18 import java.util.concurrent.ConcurrentHashMap;
19 import java.util.concurrent.ConcurrentMap;
20 import java.util.concurrent.ScheduledThreadPoolExecutor;
21 import java.util.concurrent.TimeUnit;
22 import javax.annotation.Nonnull;
23 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
24 import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
25 import org.opendaylight.controller.md.sal.binding.api.WriteTransaction;
26 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
27 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
28 import org.opendaylight.openflowplugin.api.openflow.OFPContext;
29 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
30 import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
31 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
32 import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo;
33 import org.opendaylight.openflowplugin.api.openflow.device.DeviceManager;
34 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
35 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
36 import org.opendaylight.openflowplugin.extension.api.ExtensionConverterProviderKeeper;
37 import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
38 import org.opendaylight.openflowplugin.impl.connection.OutboundQueueProviderImpl;
39 import org.opendaylight.openflowplugin.impl.device.initialization.DeviceInitializerProvider;
40 import org.opendaylight.openflowplugin.impl.device.listener.OpenflowProtocolListenerFullImpl;
41 import org.opendaylight.openflowplugin.impl.services.sal.SalRoleServiceImpl;
42 import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtil;
43 import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor;
44 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId;
45 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRef;
46 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeRemovedBuilder;
47 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeUpdatedBuilder;
48 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
49 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
50 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflow.provider.config.rev160510.OpenflowProviderConfig;
51 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
52 import org.slf4j.Logger;
53 import org.slf4j.LoggerFactory;
58 public class DeviceManagerImpl implements DeviceManager, ExtensionConverterProviderKeeper {
60 private static final Logger LOG = LoggerFactory.getLogger(DeviceManagerImpl.class);
61 private static final int SPY_RATE = 10;
63 private final OpenflowProviderConfig config;
64 private final DataBroker dataBroker;
65 private final DeviceInitializerProvider deviceInitializerProvider;
66 private final ConvertorExecutor convertorExecutor;
67 private final ConcurrentMap<DeviceInfo, DeviceContext> deviceContexts = new ConcurrentHashMap<>();
68 private final Set<KeyedInstanceIdentifier<Node, NodeKey>> notificationCreateNodeSend = new ConcurrentSet<>();
69 private final NotificationPublishService notificationPublishService;
70 private final MessageSpy messageSpy;
71 private final HashedWheelTimer hashedWheelTimer;
72 private TranslatorLibrary translatorLibrary;
73 private ExtensionConverterProvider extensionConverterProvider;
74 private ScheduledThreadPoolExecutor spyPool;
76 public DeviceManagerImpl(@Nonnull final OpenflowProviderConfig config,
77 @Nonnull final DataBroker dataBroker,
78 @Nonnull final MessageSpy messageSpy,
79 @Nonnull final NotificationPublishService notificationPublishService,
80 @Nonnull final HashedWheelTimer hashedWheelTimer,
81 @Nonnull final ConvertorExecutor convertorExecutor,
82 @Nonnull final DeviceInitializerProvider deviceInitializerProvider) {
84 this.dataBroker = dataBroker;
85 this.deviceInitializerProvider = deviceInitializerProvider;
86 this.convertorExecutor = convertorExecutor;
87 this.hashedWheelTimer = hashedWheelTimer;
88 this.spyPool = new ScheduledThreadPoolExecutor(1);
89 this.notificationPublishService = notificationPublishService;
90 this.messageSpy = messageSpy;
91 DeviceInitializationUtil.makeEmptyNodes(dataBroker);
95 public TranslatorLibrary oook() {
96 return translatorLibrary;
100 public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
101 this.translatorLibrary = translatorLibrary;
105 public void close() {
106 deviceContexts.values().forEach(OFPContext::close);
107 deviceContexts.clear();
108 Optional.ofNullable(spyPool).ifPresent(ScheduledThreadPoolExecutor::shutdownNow);
114 public void initialize() {
115 spyPool.scheduleAtFixedRate(messageSpy, SPY_RATE, SPY_RATE, TimeUnit.SECONDS);
119 public void setExtensionConverterProvider(final ExtensionConverterProvider extensionConverterProvider) {
120 this.extensionConverterProvider = extensionConverterProvider;
124 public ExtensionConverterProvider getExtensionConverterProvider() {
125 return extensionConverterProvider;
129 public ListenableFuture<Void> removeDeviceFromOperationalDS(
130 @Nonnull final KeyedInstanceIdentifier<Node, NodeKey> ii) {
132 final WriteTransaction delWtx = dataBroker.newWriteOnlyTransaction();
133 delWtx.delete(LogicalDatastoreType.OPERATIONAL, ii);
134 return delWtx.submit();
138 public DeviceContext createContext(@Nonnull final ConnectionContext connectionContext) {
140 LOG.info("ConnectionEvent: Device connected to controller, Device:{}, NodeId:{}",
141 connectionContext.getConnectionAdapter().getRemoteAddress(),
142 connectionContext.getDeviceInfo().getNodeId());
144 connectionContext.getConnectionAdapter().setPacketInFiltering(true);
146 final OutboundQueueProvider outboundQueueProvider
147 = new OutboundQueueProviderImpl(connectionContext.getDeviceInfo().getVersion());
149 connectionContext.setOutboundQueueProvider(outboundQueueProvider);
150 final OutboundQueueHandlerRegistration<OutboundQueueProvider> outboundQueueHandlerRegistration =
151 connectionContext.getConnectionAdapter().registerOutboundQueueHandler(
152 outboundQueueProvider,
153 config.getBarrierCountLimit().getValue(),
154 TimeUnit.MILLISECONDS.toNanos(config.getBarrierIntervalTimeoutLimit().getValue()));
155 connectionContext.setOutboundQueueHandleRegistration(outboundQueueHandlerRegistration);
158 final DeviceContext deviceContext = new DeviceContextImpl(
164 config.isSkipTableFeatures(),
166 config.isUseSingleLayerSerialization(),
167 deviceInitializerProvider,
168 config.isEnableFlowRemovedNotification(),
169 config.isSwitchFeaturesMandatory());
171 deviceContext.setSalRoleService(new SalRoleServiceImpl(deviceContext, deviceContext));
172 ((ExtensionConverterProviderKeeper) deviceContext).setExtensionConverterProvider(extensionConverterProvider);
173 deviceContext.setNotificationPublishService(notificationPublishService);
175 deviceContexts.put(connectionContext.getDeviceInfo(), deviceContext);
176 updatePacketInRateLimiters();
178 final OpenflowProtocolListenerFullImpl messageListener = new OpenflowProtocolListenerFullImpl(
179 connectionContext.getConnectionAdapter(), deviceContext);
181 connectionContext.getConnectionAdapter().setMessageListener(messageListener);
182 connectionContext.getConnectionAdapter().setAlienMessageListener(messageListener);
184 return deviceContext;
187 private void updatePacketInRateLimiters() {
188 synchronized (deviceContexts) {
189 final int deviceContextsSize = deviceContexts.size();
190 if (deviceContextsSize > 0) {
191 long freshNotificationLimit = config.getGlobalNotificationQuota() / deviceContextsSize;
192 if (freshNotificationLimit < 100) {
193 freshNotificationLimit = 100;
195 if (LOG.isDebugEnabled()) {
196 LOG.debug("fresh notification limit = {}", freshNotificationLimit);
198 for (final DeviceContext deviceContext : deviceContexts.values()) {
199 deviceContext.updatePacketInRateLimit(freshNotificationLimit);
206 public void onDeviceRemoved(final DeviceInfo deviceInfo) {
207 deviceContexts.remove(deviceInfo);
208 if (LOG.isDebugEnabled()) {
209 LOG.debug("Device context removed for node {}", deviceInfo.getLOGValue());
211 if (deviceContexts.size() > 0) {
212 this.updatePacketInRateLimiters();
217 public void sendNodeRemovedNotification(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
218 if (notificationCreateNodeSend.remove(instanceIdentifier)) {
219 NodeRemovedBuilder builder = new NodeRemovedBuilder();
220 builder.setNodeRef(new NodeRef(instanceIdentifier));
221 LOG.info("Publishing node removed notification for {}", instanceIdentifier.firstKeyOf(Node.class).getId());
222 notificationPublishService.offerNotification(builder.build());
227 public void sendNodeAddedNotification(@Nonnull final KeyedInstanceIdentifier<Node, NodeKey> instanceIdentifier) {
228 if (!notificationCreateNodeSend.contains(instanceIdentifier)) {
229 notificationCreateNodeSend.add(instanceIdentifier);
230 final NodeId id = instanceIdentifier.firstKeyOf(Node.class).getId();
231 NodeUpdatedBuilder builder = new NodeUpdatedBuilder();
233 builder.setNodeRef(new NodeRef(instanceIdentifier));
234 LOG.info("Publishing node added notification for {}", id);
235 notificationPublishService.offerNotification(builder.build());