requests map in deviceContext is synchronized now
[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / device / DeviceContextImpl.java
1 /**
2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.openflowplugin.impl.device;
9
10 import com.google.common.annotations.VisibleForTesting;
11 import com.google.common.base.Preconditions;
12 import com.google.common.util.concurrent.SettableFuture;
13 import io.netty.util.HashedWheelTimer;
14 import io.netty.util.Timeout;
15 import java.math.BigInteger;
16 import java.util.ArrayList;
17 import java.util.Collections;
18 import java.util.HashMap;
19 import java.util.Iterator;
20 import java.util.List;
21 import java.util.Map;
22 import java.util.TreeMap;
23 import java.util.concurrent.atomic.AtomicLong;
24 import javax.annotation.Nonnull;
25 import org.opendaylight.controller.md.sal.binding.api.DataBroker;
26 import org.opendaylight.controller.md.sal.binding.api.ReadTransaction;
27 import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
28 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
29 import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
30 import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
31 import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
32 import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
33 import org.opendaylight.openflowplugin.api.openflow.device.RequestContext;
34 import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
35 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
36 import org.opendaylight.openflowplugin.api.openflow.device.exception.DeviceDataException;
37 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceContextClosedHandler;
38 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceDisconnectedHandler;
39 import org.opendaylight.openflowplugin.api.openflow.device.listener.OpenflowMessageListenerFacade;
40 import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
41 import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
42 import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
43 import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
44 import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
45 import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
46 import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
47 import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
48 import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
49 import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
50 import org.opendaylight.openflowplugin.impl.services.RequestContextUtil;
51 import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
52 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
53 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
54 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnector;
55 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorBuilder;
56 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.node.NodeConnectorKey;
57 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
58 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.PortReason;
59 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.Error;
60 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.FlowRemoved;
61 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
62 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
63 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketIn;
64 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PacketInMessage;
65 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortGrouping;
66 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.PortStatusMessage;
67 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
68 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
69 import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
70 import org.opendaylight.yangtools.yang.binding.ChildOf;
71 import org.opendaylight.yangtools.yang.binding.DataObject;
72 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
73 import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
74 import org.opendaylight.yangtools.yang.common.RpcError;
75 import org.opendaylight.yangtools.yang.common.RpcResult;
76 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
77 import org.slf4j.Logger;
78 import org.slf4j.LoggerFactory;
79
80 /**
81  *
82  */
83 public class DeviceContextImpl implements DeviceContext {
84
85     private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
86     public static final String DEVICE_DISCONNECTED = "Device disconnected.";
87
88     private final ConnectionContext primaryConnectionContext;
89     private final DeviceState deviceState;
90     private final DataBroker dataBroker;
91     private final XidGenerator xidGenerator;
92     private final HashedWheelTimer hashedWheelTimer;
93     private Map<Long, RequestContext> requests = Collections.synchronizedMap(new TreeMap());
94
95     private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
96     private final TransactionChainManager txChainManager;
97     private TranslatorLibrary translatorLibrary;
98     private OpenflowMessageListenerFacade openflowMessageListenerFacade;
99     private final DeviceFlowRegistry deviceFlowRegistry;
100     private final DeviceGroupRegistry deviceGroupRegistry;
101     private final DeviceMeterRegistry deviceMeterRegistry;
102     private Timeout barrierTaskTimeout;
103     private NotificationProviderService notificationService;
104     private final MessageSpy<Class> messageSpy;
105     private DeviceDisconnectedHandler deviceDisconnectedHandler;
106     private List<DeviceContextClosedHandler> closeHandlers = new ArrayList<>();
107
108
109     @VisibleForTesting
110     DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
111                       @Nonnull final DeviceState deviceState,
112                       @Nonnull final DataBroker dataBroker,
113                       @Nonnull final HashedWheelTimer hashedWheelTimer,
114                       @Nonnull final MessageSpy _messageSpy) {
115         this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
116         this.deviceState = Preconditions.checkNotNull(deviceState);
117         this.dataBroker = Preconditions.checkNotNull(dataBroker);
118         this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
119         xidGenerator = new XidGenerator();
120         txChainManager = new TransactionChainManager(dataBroker, hashedWheelTimer, 500L, 500L);
121         auxiliaryConnectionContexts = new HashMap<>();
122         requests = new HashMap<>();
123         deviceFlowRegistry = new DeviceFlowRegistryImpl();
124         deviceGroupRegistry = new DeviceGroupRegistryImpl();
125         deviceMeterRegistry = new DeviceMeterRegistryImpl();
126         messageSpy = _messageSpy;
127     }
128
129     /**
130      * This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
131      * and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
132      */
133     void submitTransaction() {
134         txChainManager.submitTransaction();
135         txChainManager.enableCounter();
136     }
137
138     @Override
139     public <M extends ChildOf<DataObject>> void onMessage(final M message, final RequestContext requestContext) {
140         // TODO Auto-generated method stub
141
142     }
143
144     @Override
145     public void addAuxiliaryConenctionContext(final ConnectionContext connectionContext) {
146         final SwitchConnectionDistinguisher connectionDistinguisher = new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
147         auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
148     }
149
150     @Override
151     public void removeAuxiliaryConenctionContext(final ConnectionContext connectionContext) {
152         // TODO Auto-generated method stub
153     }
154
155     @Override
156     public DeviceState getDeviceState() {
157         return deviceState;
158     }
159
160     @Override
161     public ReadTransaction getReadTransaction() {
162         return dataBroker.newReadOnlyTransaction();
163     }
164
165     @Override
166     public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
167                                                           final InstanceIdentifier<T> path, final T data) {
168         txChainManager.writeToTransaction(store, path, data);
169     }
170
171     @Override
172     public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
173         txChainManager.addDeleteOperationTotTxChain(store, path);
174     }
175
176     @Override
177     public ConnectionContext getPrimaryConnectionContext() {
178         return primaryConnectionContext;
179     }
180
181     @Override
182     public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
183         return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
184     }
185
186     @Override
187     public Xid getNextXid() {
188         return xidGenerator.generate();
189     }
190
191     @Override
192     public RequestContext lookupRequest(Xid xid) {
193         return requests.get(xid.getValue());
194     }
195
196     @Override
197     public int getNumberOfOutstandingRequests() {
198         return requests.size();
199     }
200
201     @Override
202     public void hookRequestCtx(final Xid xid, final RequestContext requestFutureContext) {
203         requests.put(xid.getValue(), requestFutureContext);
204     }
205
206     @Override
207     public RequestContext unhookRequestCtx(Xid xid) {
208         return requests.remove(xid.getValue());
209     }
210
211     @Override
212     public void attachOpenflowMessageListener(final OpenflowMessageListenerFacade openflowMessageListenerFacade) {
213         this.openflowMessageListenerFacade = openflowMessageListenerFacade;
214         primaryConnectionContext.getConnectionAdapter().setMessageListener(openflowMessageListenerFacade);
215     }
216
217     @Override
218     public OpenflowMessageListenerFacade getOpenflowMessageListenerFacade() {
219         return openflowMessageListenerFacade;
220     }
221
222     @Override
223     public DeviceFlowRegistry getDeviceFlowRegistry() {
224         return deviceFlowRegistry;
225     }
226
227     @Override
228     public DeviceGroupRegistry getDeviceGroupRegistry() {
229         return deviceGroupRegistry;
230     }
231
232     @Override
233     public DeviceMeterRegistry getDeviceMeterRegistry() {
234         return deviceMeterRegistry;
235     }
236
237     @Override
238     public void processReply(final OfHeader ofHeader) {
239         final RequestContext requestContext = requests.get(ofHeader.getXid());
240         if (null != requestContext) {
241             final SettableFuture replyFuture = requestContext.getFuture();
242             requests.remove(ofHeader.getXid());
243             RpcResult<OfHeader> rpcResult;
244             if (ofHeader instanceof Error) {
245                 //TODO : this is the point, where we can discover that add flow operation failed and where we should
246                 //TODO : remove this flow from deviceFlowRegistry
247                 final Error error = (Error) ofHeader;
248                 final String message = "Operation on device failed";
249                 rpcResult = RpcResultBuilder
250                         .<OfHeader>failed()
251                         .withError(RpcError.ErrorType.APPLICATION, message, new DeviceDataException(message, error))
252                         .build();
253                 messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
254             } else {
255                 rpcResult = RpcResultBuilder
256                         .<OfHeader>success()
257                         .withResult(ofHeader)
258                         .build();
259                 messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
260             }
261
262             replyFuture.set(rpcResult);
263             try {
264                 requestContext.close();
265             } catch (final Exception e) {
266                 LOG.error("Closing RequestContext failed: ", e);
267             }
268         } else {
269             LOG.error("Can't find request context registered for xid : {}. Type of reply: {}. From address: {}", ofHeader.getXid(), ofHeader.getClass().getName(),
270                     getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress());
271         }
272     }
273
274     @Override
275     public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
276         final RequestContext requestContext = requests.get(xid.getValue());
277         if (null != requestContext) {
278             final SettableFuture replyFuture = requestContext.getFuture();
279             requests.remove(xid.getValue());
280             final RpcResult<List<MultipartReply>> rpcResult = RpcResultBuilder
281                     .<List<MultipartReply>>success()
282                     .withResult(ofHeaderList)
283                     .build();
284             replyFuture.set(rpcResult);
285             for (MultipartReply multipartReply : ofHeaderList) {
286                 messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
287             }
288
289             try {
290                 requestContext.close();
291             } catch (final Exception e) {
292                 LOG.error("Closing RequestContext failed: ", e);
293             }
294         } else {
295             LOG.error("Can't find request context registered for xid : {}. Type of reply: MULTIPART. From address: {}", xid.getValue(),
296                     getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress());
297         }
298     }
299
300     @Override
301     public void processException(final Xid xid, final DeviceDataException deviceDataException) {
302
303         LOG.trace("Processing exception for xid : {}", xid.getValue());
304
305         final RequestContext requestContext = requests.get(xid.getValue());
306
307         if (null != requestContext) {
308             final SettableFuture replyFuture = requestContext.getFuture();
309             requests.remove(xid.getValue());
310             final RpcResult<List<OfHeader>> rpcResult = RpcResultBuilder
311                     .<List<OfHeader>>failed()
312                     .withError(RpcError.ErrorType.APPLICATION, String.format("Message processing failed : %s", deviceDataException.getError()), deviceDataException)
313                     .build();
314             replyFuture.set(rpcResult);
315             messageSpy.spyMessage(deviceDataException.getClass(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
316             try {
317                 requestContext.close();
318             } catch (final Exception e) {
319                 LOG.error("Closing RequestContext failed: ", e);
320             }
321         } else {
322             LOG.error("Can't find request context registered for xid : {}", xid.getValue());
323         }
324     }
325
326     @Override
327     public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
328         //TODO: will be defined later
329     }
330
331     @Override
332     public synchronized void processPortStatusMessage(final PortStatusMessage portStatus) {
333         messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
334         final TranslatorKey translatorKey = new TranslatorKey(portStatus.getVersion(), PortGrouping.class.getName());
335         final MessageTranslator<PortGrouping, FlowCapableNodeConnector> messageTranslator = translatorLibrary.lookupTranslator(translatorKey);
336         final FlowCapableNodeConnector flowCapableNodeConnector = messageTranslator.translate(portStatus, this, null);
337
338         final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
339         if (portStatus.getReason().equals(PortReason.OFPPRADD) || portStatus.getReason().equals(PortReason.OFPPRMODIFY)) {
340             // because of ADD status node connector has to be created
341             final NodeConnectorBuilder nConnectorBuilder = new NodeConnectorBuilder().setKey(iiToNodeConnector.getKey());
342             nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
343             nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
344             writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
345         } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
346             addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
347         }
348     }
349
350     private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
351         final InstanceIdentifier<Node> iiToNodes = deviceState.getNodeInstanceIdentifier();
352         final BigInteger dataPathId = deviceState.getFeatures().getDatapathId();
353         final NodeConnectorId nodeConnectorId = NodeStaticReplyTranslatorUtil.nodeConnectorId(dataPathId.toString(), portNo, version);
354         return iiToNodes.child(NodeConnector.class, new NodeConnectorKey(nodeConnectorId));
355     }
356
357     @Override
358     public void processPacketInMessage(final PacketInMessage packetInMessage) {
359         messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
360         final TranslatorKey translatorKey = new TranslatorKey(packetInMessage.getVersion(), PacketIn.class.getName());
361         final MessageTranslator<PacketInMessage, PacketReceived> messageTranslator = translatorLibrary.lookupTranslator(translatorKey);
362         final PacketReceived packetReceived = messageTranslator.translate(packetInMessage, this, null);
363         notificationService.publish(packetReceived);
364     }
365
366     @Override
367     public TranslatorLibrary oook() {
368         return translatorLibrary;
369     }
370
371     @Override
372     public void setTranslatorLibrary(final TranslatorLibrary translatorLibrary) {
373         this.translatorLibrary = translatorLibrary;
374     }
375
376     @Override
377     public synchronized HashedWheelTimer getTimer() {
378         return hashedWheelTimer;
379     }
380
381     @Override
382     public void close() throws Exception {
383         deviceState.setValid(false);
384
385         LOG.trace("Removing node {} from operational DS.", getDeviceState().getNodeId());
386         addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, getDeviceState().getNodeInstanceIdentifier());
387
388         deviceGroupRegistry.close();
389         deviceFlowRegistry.close();
390         deviceMeterRegistry.close();
391
392         if (primaryConnectionContext.getConnectionAdapter().isAlive()) {
393             primaryConnectionContext.setConnectionState(ConnectionContext.CONNECTION_STATE.RIP);
394             primaryConnectionContext.getConnectionAdapter().disconnect();
395         }
396         for (Map.Entry<Long, RequestContext> entry : requests.entrySet()) {
397             RequestContextUtil.closeRequestContextWithRpcError(entry.getValue(), DEVICE_DISCONNECTED);
398         }
399         for (ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
400             if (connectionContext.getConnectionAdapter().isAlive()) {
401                 connectionContext.getConnectionAdapter().disconnect();
402             }
403         }
404         for (DeviceContextClosedHandler deviceContextClosedHandler : closeHandlers) {
405             deviceContextClosedHandler.onDeviceContextClosed(this);
406         }
407
408     }
409
410     @Override
411     public synchronized void onDeviceDisconnected(final ConnectionContext connectionContext) {
412         if (this.getPrimaryConnectionContext().equals(connectionContext)) {
413             try {
414                 close();
415             } catch (Exception e) {
416                 LOG.trace("Error closing device context.");
417             }
418             if (null != deviceDisconnectedHandler) {
419                 deviceDisconnectedHandler.onDeviceDisconnected(connectionContext);
420             }
421         } else {
422             auxiliaryConnectionContexts.remove(connectionContext);
423         }
424     }
425
426
427     private class XidGenerator {
428
429         private final AtomicLong xid = new AtomicLong(0);
430
431         public Xid generate() {
432             return new Xid(xid.incrementAndGet());
433         }
434     }
435
436     @Override
437     public RequestContext extractNextOutstandingMessage(final long barrierXid) {
438         RequestContext nextMessage = null;
439         final Iterator<Long> keyIterator = requests.keySet().iterator();
440         if (keyIterator.hasNext()) {
441             final Long oldestXid = keyIterator.next();
442             if (oldestXid < barrierXid) {
443                 nextMessage = requests.remove(oldestXid);
444             }
445         }
446         return nextMessage;
447     }
448
449     @Override
450     public void setCurrentBarrierTimeout(final Timeout timeout) {
451         barrierTaskTimeout = timeout;
452     }
453
454     @Override
455     public Timeout getBarrierTaskTimeout() {
456         return barrierTaskTimeout;
457     }
458
459     @Override
460     public void setNotificationService(final NotificationProviderService notificationServiceParam) {
461         notificationService = notificationServiceParam;
462     }
463
464     @Override
465     public MessageSpy getMessageSpy() {
466         return messageSpy;
467     }
468
469     @Override
470     public void setDeviceDisconnectedHandler(final DeviceDisconnectedHandler deviceDisconnectedHandler) {
471         this.deviceDisconnectedHandler = deviceDisconnectedHandler;
472     }
473
474     @Override
475     public void addDeviceContextClosedHandler(final DeviceContextClosedHandler deviceContextClosedHandler) {
476         this.closeHandlers.add(deviceContextClosedHandler);
477     }
478 }