import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import java.math.BigInteger;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.Nonnull;
import org.opendaylight.controller.md.sal.binding.api.DataBroker;
+import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
+import org.opendaylight.controller.md.sal.binding.api.NotificationService;
import org.opendaylight.controller.md.sal.binding.api.ReadTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
+import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
+import org.opendaylight.openflowplugin.api.openflow.connection.ThrottledConnectionsHolder;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
import org.opendaylight.openflowplugin.api.openflow.device.TranslatorLibrary;
import org.opendaylight.openflowplugin.api.openflow.device.Xid;
import org.opendaylight.openflowplugin.api.openflow.device.exception.DeviceDataException;
+import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceContextClosedHandler;
+import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceDisconnectedHandler;
import org.opendaylight.openflowplugin.api.openflow.device.listener.OpenflowMessageListenerFacade;
-import org.opendaylight.openflowplugin.api.openflow.flow.registry.DeviceFlowRegistry;
import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
import org.opendaylight.openflowplugin.api.openflow.md.core.TranslatorKey;
+import org.opendaylight.openflowplugin.api.openflow.registry.flow.DeviceFlowRegistry;
+import org.opendaylight.openflowplugin.api.openflow.registry.group.DeviceGroupRegistry;
+import org.opendaylight.openflowplugin.api.openflow.registry.meter.DeviceMeterRegistry;
+import org.opendaylight.openflowplugin.api.openflow.statistics.ofpspecific.MessageSpy;
import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
-import org.opendaylight.openflowplugin.impl.flow.registry.DeviceFlowRegistryImpl;
+import org.opendaylight.openflowplugin.impl.registry.flow.DeviceFlowRegistryImpl;
+import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
+import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
+import org.opendaylight.openflowplugin.impl.services.RequestContextUtil;
import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.inventory.rev130819.FlowCapableNodeConnector;
import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorId;
import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsData;
import org.opendaylight.yang.gen.v1.urn.opendaylight.port.statistics.rev131214.FlowCapableNodeConnectorStatisticsDataBuilder;
-import org.opendaylight.yang.gen.v1.urn.opendaylight.table.types.rev131026.TableFeatures;
import org.opendaylight.yangtools.yang.binding.ChildOf;
import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
public class DeviceContextImpl implements DeviceContext {
private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
+ public static final String DEVICE_DISCONNECTED = "Device disconnected.";
private final ConnectionContext primaryConnectionContext;
private final DeviceState deviceState;
private final DataBroker dataBroker;
private final XidGenerator xidGenerator;
private final HashedWheelTimer hashedWheelTimer;
- private Map<Long, RequestContext> requests = new TreeMap<>();
+ private final Map<Long, RequestContext> requests = new TreeMap<>();
private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
private final TransactionChainManager txChainManager;
private TranslatorLibrary translatorLibrary;
private OpenflowMessageListenerFacade openflowMessageListenerFacade;
private final DeviceFlowRegistry deviceFlowRegistry;
+ private final DeviceGroupRegistry deviceGroupRegistry;
+ private final DeviceMeterRegistry deviceMeterRegistry;
private Timeout barrierTaskTimeout;
- private NotificationProviderService notificationService;
+ private NotificationService notificationService;
+ private final MessageSpy<Class> messageSpy;
+ private DeviceDisconnectedHandler deviceDisconnectedHandler;
+ private final List<DeviceContextClosedHandler> closeHandlers = new ArrayList<>();
+ private NotificationPublishService notificationPublishService;
+ private final ThrottledConnectionsHolder throttledConnectionsHolder;
+
@VisibleForTesting
DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
- @Nonnull final DeviceState deviceState, @Nonnull final DataBroker dataBroker,
- @Nonnull final HashedWheelTimer hashedWheelTimer) {
+ @Nonnull final DeviceState deviceState,
+ @Nonnull final DataBroker dataBroker,
+ @Nonnull final HashedWheelTimer hashedWheelTimer,
+ @Nonnull final MessageSpy _messageSpy,
+ @Nonnull final ThrottledConnectionsHolder throttledConnectionsHolder) {
this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
this.deviceState = Preconditions.checkNotNull(deviceState);
this.dataBroker = Preconditions.checkNotNull(dataBroker);
this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
xidGenerator = new XidGenerator();
- txChainManager = new TransactionChainManager(dataBroker, hashedWheelTimer, 500L);
+ txChainManager = new TransactionChainManager(dataBroker, hashedWheelTimer, 500L, 500L);
auxiliaryConnectionContexts = new HashMap<>();
- requests = new HashMap<>();
deviceFlowRegistry = new DeviceFlowRegistryImpl();
+ deviceGroupRegistry = new DeviceGroupRegistryImpl();
+ deviceMeterRegistry = new DeviceMeterRegistryImpl();
+ messageSpy = _messageSpy;
+ this.throttledConnectionsHolder = throttledConnectionsHolder;
}
/**
* and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
*/
void submitTransaction() {
+ txChainManager.enableSubmit();
txChainManager.submitTransaction();
- txChainManager.enableCounter();
}
@Override
@Override
public void addAuxiliaryConenctionContext(final ConnectionContext connectionContext) {
- final SwitchConnectionDistinguisher connectionDistinguisher = new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
+ final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
auxiliaryConnectionContexts.put(connectionDistinguisher, connectionContext);
}
+ private SwitchConnectionDistinguisher createConnectionDistinguisher(final ConnectionContext connectionContext) {
+ return new SwitchConnectionCookieOFImpl(connectionContext.getFeatures().getAuxiliaryId());
+ }
+
@Override
public void removeAuxiliaryConenctionContext(final ConnectionContext connectionContext) {
// TODO Auto-generated method stub
txChainManager.addDeleteOperationTotTxChain(store, path);
}
- @Override
- public TableFeatures getCapabilities() {
- // TODO Auto-generated method stub
- return null;
- }
-
@Override
public ConnectionContext getPrimaryConnectionContext() {
return primaryConnectionContext;
}
@Override
- public Map<Long, RequestContext> getRequests() {
- return requests;
+ public RequestContext lookupRequest(final Xid xid) {
+ synchronized (requests) {
+ return requests.get(xid.getValue());
+ }
+ }
+
+ @Override
+ public int getNumberOfOutstandingRequests() {
+ synchronized (requests) {
+ return requests.size();
+ }
}
@Override
public void hookRequestCtx(final Xid xid, final RequestContext requestFutureContext) {
- // TODO Auto-generated method stub
- requests.put(xid.getValue(), requestFutureContext);
+ synchronized (requests) {
+ requests.put(xid.getValue(), requestFutureContext);
+ }
+ }
+
+ @Override
+ public RequestContext unhookRequestCtx(final Xid xid) {
+ synchronized (requests) {
+ return requests.remove(xid.getValue());
+ }
}
@Override
return deviceFlowRegistry;
}
+ @Override
+ public DeviceGroupRegistry getDeviceGroupRegistry() {
+ return deviceGroupRegistry;
+ }
+
+ @Override
+ public DeviceMeterRegistry getDeviceMeterRegistry() {
+ return deviceMeterRegistry;
+ }
+
@Override
public void processReply(final OfHeader ofHeader) {
- final RequestContext requestContext = getRequests().get(ofHeader.getXid());
+ final RequestContext requestContext = requests.remove(ofHeader.getXid());
if (null != requestContext) {
final SettableFuture replyFuture = requestContext.getFuture();
- getRequests().remove(ofHeader.getXid());
RpcResult<OfHeader> rpcResult;
if (ofHeader instanceof Error) {
+ //TODO : this is the point, where we can discover that add flow operation failed and where we should
+ //TODO : remove this flow from deviceFlowRegistry
final Error error = (Error) ofHeader;
- final String message = "Operation on device failed";
+ final String message = "Operation on device failed with xid " + ofHeader.getXid() + ".";
rpcResult = RpcResultBuilder
.<OfHeader>failed()
.withError(RpcError.ErrorType.APPLICATION, message, new DeviceDataException(message, error))
.build();
+ messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
} else {
rpcResult = RpcResultBuilder
.<OfHeader>success()
.withResult(ofHeader)
.build();
+ messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
}
replyFuture.set(rpcResult);
try {
requestContext.close();
} catch (final Exception e) {
- LOG.error("Closing RequestContext failed: ", e);
+ LOG.warn("Closing RequestContext failed: {}", e.getMessage());
+ LOG.debug("Closing RequestContext failed.. ", e);
}
} else {
- LOG.error("Can't find request context registered for xid : {}. Type of reply: {}. From address: {}", ofHeader.getXid(), ofHeader.getClass().getName(),
+ LOG.warn("Can't find request context registered for xid : {}. Type of reply: {}. From address: {}", ofHeader.getXid(), ofHeader.getClass().getName(),
getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress());
}
}
@Override
public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
- final RequestContext requestContext = getRequests().get(xid.getValue());
+ final RequestContext requestContext;
+ synchronized (requests) {
+ requestContext = requests.remove(xid.getValue());
+ }
if (null != requestContext) {
final SettableFuture replyFuture = requestContext.getFuture();
- getRequests().remove(xid.getValue());
final RpcResult<List<MultipartReply>> rpcResult = RpcResultBuilder
.<List<MultipartReply>>success()
.withResult(ofHeaderList)
.build();
replyFuture.set(rpcResult);
+ for (final MultipartReply multipartReply : ofHeaderList) {
+ messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
+ }
+
try {
requestContext.close();
} catch (final Exception e) {
- LOG.error("Closing RequestContext failed: ", e);
+ LOG.warn("Closing RequestContext failed: {}", e.getMessage());
+ LOG.debug("Closing RequestContext failed.. ", e);
}
} else {
- LOG.error("Can't find request context registered for xid : {}. Type of reply: MULTIPART. From address: {}", xid.getValue(),
+ LOG.warn("Can't find request context registered for xid : {}. Type of reply: MULTIPART. From address: {}", xid.getValue(),
getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress());
}
}
LOG.trace("Processing exception for xid : {}", xid.getValue());
- final RequestContext requestContext = getRequests().get(xid.getValue());
+ final RequestContext requestContext = requests.remove(xid.getValue());
if (null != requestContext) {
final SettableFuture replyFuture = requestContext.getFuture();
- getRequests().remove(xid.getValue());
final RpcResult<List<OfHeader>> rpcResult = RpcResultBuilder
.<List<OfHeader>>failed()
.withError(RpcError.ErrorType.APPLICATION, String.format("Message processing failed : %s", deviceDataException.getError()), deviceDataException)
.build();
replyFuture.set(rpcResult);
+ messageSpy.spyMessage(deviceDataException.getClass(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
try {
requestContext.close();
} catch (final Exception e) {
- LOG.error("Closing RequestContext failed: ", e);
+ LOG.warn("Closing RequestContext failed: ", e);
+ LOG.debug("Closing RequestContext failed..", e);
}
} else {
- LOG.error("Can't find request context registered for xid : {}", xid.getValue());
+ LOG.warn("Can't find request context registered for xid : {}. Exception message {}",
+ xid.getValue(), deviceDataException.getMessage());
}
}
@Override
public void processPortStatusMessage(final PortStatusMessage portStatus) {
+ messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
final TranslatorKey translatorKey = new TranslatorKey(portStatus.getVersion(), PortGrouping.class.getName());
final MessageTranslator<PortGrouping, FlowCapableNodeConnector> messageTranslator = translatorLibrary.lookupTranslator(translatorKey);
final FlowCapableNodeConnector flowCapableNodeConnector = messageTranslator.translate(portStatus, this, null);
nConnectorBuilder.addAugmentation(FlowCapableNodeConnectorStatisticsData.class, new FlowCapableNodeConnectorStatisticsDataBuilder().build());
nConnectorBuilder.addAugmentation(FlowCapableNodeConnector.class, flowCapableNodeConnector);
writeToTransaction(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector, nConnectorBuilder.build());
- } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE) ) {
+ } else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
}
}
@Override
public void processPacketInMessage(final PacketInMessage packetInMessage) {
- final TranslatorKey translatorKey = new TranslatorKey(packetInMessage.getVersion(), PacketIn.class.getName());
- final MessageTranslator<PacketInMessage, PacketReceived> messageTranslator = translatorLibrary.lookupTranslator(translatorKey);
- final PacketReceived packetReceived = messageTranslator.translate(packetInMessage, this, null);
- notificationService.publish(packetReceived);
+ messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
+ final ConnectionAdapter connectionAdapter = this.getPrimaryConnectionContext().getConnectionAdapter();
+ if (connectionAdapter.isAutoRead()) {
+ final TranslatorKey translatorKey = new TranslatorKey(packetInMessage.getVersion(), PacketIn.class.getName());
+ final MessageTranslator<PacketInMessage, PacketReceived> messageTranslator = translatorLibrary.lookupTranslator(translatorKey);
+ final PacketReceived packetReceived = messageTranslator.translate(packetInMessage, this, null);
+
+ if (packetReceived != null) {
+ messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_OUT_SUCCESS);
+ } else {
+ messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
+ }
+ if (!notificationPublishService.offerNotification(packetReceived)) {
+ LOG.debug("Notification offer refused by notification service.");
+ messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
+ LOG.debug("Notification offer refused by notification service.");
+ connectionAdapter.setAutoRead(false);
+ LOG.debug("Throttling primary connection for {}", connectionAdapter.getRemoteAddress());
+ this.throttledConnectionsHolder.storeThrottledConnection(connectionAdapter);
+ } else {
+ messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
+ }
+ }
+
+
}
@Override
return hashedWheelTimer;
}
+ @Override
+ public void close() throws Exception {
+ deviceState.setValid(false);
+
+ LOG.trace("Removing node {} from operational DS.", getDeviceState().getNodeId());
+ addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, getDeviceState().getNodeInstanceIdentifier());
+
+ deviceGroupRegistry.close();
+ deviceFlowRegistry.close();
+ deviceMeterRegistry.close();
+
+ if (primaryConnectionContext.getConnectionAdapter().isAlive()) {
+ primaryConnectionContext.setConnectionState(ConnectionContext.CONNECTION_STATE.RIP);
+ primaryConnectionContext.getConnectionAdapter().disconnect();
+ }
+ for (final Map.Entry<Long, RequestContext> entry : requests.entrySet()) {
+ RequestContextUtil.closeRequestContextWithRpcError(entry.getValue(), DEVICE_DISCONNECTED);
+ }
+ for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
+ if (connectionContext.getConnectionAdapter().isAlive()) {
+ connectionContext.getConnectionAdapter().disconnect();
+ }
+ }
+ for (final DeviceContextClosedHandler deviceContextClosedHandler : closeHandlers) {
+ deviceContextClosedHandler.onDeviceContextClosed(this);
+ }
+
+ }
+
+ @Override
+ public void onDeviceDisconnected(final ConnectionContext connectionContext) {
+ if (this.getPrimaryConnectionContext().equals(connectionContext)) {
+ try {
+ close();
+ } catch (final Exception e) {
+ LOG.trace("Error closing device context.");
+ }
+ if (null != deviceDisconnectedHandler) {
+ deviceDisconnectedHandler.onDeviceDisconnected(connectionContext);
+ }
+ } else {
+ final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
+ auxiliaryConnectionContexts.remove(connectionDistinguisher);
+ }
+ }
+
private class XidGenerator {
@Override
public RequestContext extractNextOutstandingMessage(final long barrierXid) {
RequestContext nextMessage = null;
- final Iterator<Long> keyIterator = requests.keySet().iterator();
- if (keyIterator.hasNext()) {
- final Long oldestXid = keyIterator.next();
- if (oldestXid < barrierXid) {
- nextMessage = requests.remove(oldestXid);
+ synchronized (requests) {
+ final Iterator<Long> keyIterator = requests.keySet().iterator();
+ if (keyIterator.hasNext()) {
+ final Long oldestXid = keyIterator.next();
+ if (oldestXid < barrierXid) {
+ nextMessage = requests.remove(oldestXid);
+ }
}
}
return nextMessage;
}
@Override
- public void setNotificationService(final NotificationProviderService notificationServiceParam) {
+ public void setNotificationService(final NotificationService notificationServiceParam) {
notificationService = notificationServiceParam;
}
+
+ @Override
+ public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
+ this.notificationPublishService = notificationPublishService;
+ }
+
+ @Override
+ public MessageSpy getMessageSpy() {
+ return messageSpy;
+ }
+
+ @Override
+ public void setDeviceDisconnectedHandler(final DeviceDisconnectedHandler deviceDisconnectedHandler) {
+ this.deviceDisconnectedHandler = deviceDisconnectedHandler;
+ }
+
+ @Override
+ public void addDeviceContextClosedHandler(final DeviceContextClosedHandler deviceContextClosedHandler) {
+ this.closeHandlers.add(deviceContextClosedHandler);
+ }
+
+ @Override
+ public void startGatheringOperationsToOneTransaction() {
+ txChainManager.startGatheringOperationsToOneTransaction();
+ }
+
+ @Override
+ public void commitOperationsGatheredInOneTransaction() {
+ txChainManager.commitOperationsGatheredInOneTransaction();
+ }
+
}
Code Review / openflowplugin.git / blobdiff