import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import java.math.BigInteger;
-import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ExecutionException;
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.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueue;
import org.opendaylight.openflowplugin.api.openflow.connection.ConnectionContext;
-import org.opendaylight.openflowplugin.api.openflow.connection.ThrottledNotificationsOfferer;
+import org.opendaylight.openflowplugin.api.openflow.connection.OutboundQueueProvider;
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.RequestContext;
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.handlers.MultiMsgCollector;
import org.opendaylight.yangtools.yang.binding.DataObject;
import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
import org.opendaylight.yangtools.yang.binding.KeyedInstanceIdentifier;
-import org.opendaylight.yangtools.yang.common.RpcError;
-import org.opendaylight.yangtools.yang.common.RpcResult;
-import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class DeviceContextImpl implements DeviceContext {
private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
- public static final String DEVICE_DISCONNECTED = "Device disconnected.";
+
+ // TODO: watermarks should be derived from effective rpc limit (75%|95%)
+ private static final int PACKETIN_LOW_WATERMARK = 15000;
+ private static final int PACKETIN_HIGH_WATERMARK = 19000;
+ // TODO: drain factor should be parametrized
+ public static final float REJECTED_DRAIN_FACTOR = 0.25f;
private final ConnectionContext primaryConnectionContext;
private final DeviceState deviceState;
private final HashedWheelTimer hashedWheelTimer;
private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
private final TransactionChainManager txChainManager;
- private TranslatorLibrary translatorLibrary;
private final DeviceFlowRegistry deviceFlowRegistry;
private final DeviceGroupRegistry deviceGroupRegistry;
private final DeviceMeterRegistry deviceMeterRegistry;
- private Timeout barrierTaskTimeout;
- private NotificationService notificationService;
- private final MessageSpy messageSpy;
- private DeviceDisconnectedHandler deviceDisconnectedHandler;
private final Collection<DeviceContextClosedHandler> closeHandlers = new HashSet<>();
+ private final PacketInRateLimiter packetInLimiter;
+ private final MessageSpy messageSpy;
private NotificationPublishService notificationPublishService;
- private final ThrottledNotificationsOfferer throttledConnectionsHolder;
- private final BlockingQueue<PacketReceived> bumperQueue;
- private final OutboundQueue outboundQueueProvider;
-
- @Override
- public MultiMsgCollector getMultiMsgCollector() {
- return multiMsgCollector;
- }
-
- @Override
- public Long getReservedXid() {
- return outboundQueueProvider.reserveEntry();
- }
-
- private final MultiMsgCollector multiMsgCollector = new MultiMsgCollectorImpl();
-
+ private DeviceDisconnectedHandler deviceDisconnectedHandler;
+ private NotificationService notificationService;
+ private TranslatorLibrary translatorLibrary;
+ private OutboundQueue outboundQueueProvider;
+ private Timeout barrierTaskTimeout;
@VisibleForTesting
DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
@Nonnull final DeviceState deviceState,
@Nonnull final DataBroker dataBroker,
@Nonnull final HashedWheelTimer hashedWheelTimer,
- @Nonnull final MessageSpy _messageSpy,
- @Nonnull final ThrottledNotificationsOfferer throttledConnectionsHolder) {
+ @Nonnull final MessageSpy _messageSpy, OutboundQueueProvider outboundQueueProvider) {
this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
this.deviceState = Preconditions.checkNotNull(deviceState);
this.dataBroker = Preconditions.checkNotNull(dataBroker);
this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
- txChainManager = new TransactionChainManager(dataBroker, hashedWheelTimer, 500L, 500L);
+ this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
+ txChainManager = new TransactionChainManager(dataBroker, deviceState);
auxiliaryConnectionContexts = new HashMap<>();
deviceFlowRegistry = new DeviceFlowRegistryImpl();
deviceGroupRegistry = new DeviceGroupRegistryImpl();
deviceMeterRegistry = new DeviceMeterRegistryImpl();
messageSpy = _messageSpy;
- this.throttledConnectionsHolder = throttledConnectionsHolder;
- bumperQueue = new ArrayBlockingQueue<>(5000);
- multiMsgCollector.setDeviceReplyProcessor(this);
- outboundQueueProvider = Preconditions.checkNotNull(primaryConnectionContext.getOutboundQueueProvider());
+
+ this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
+ PACKETIN_LOW_WATERMARK, PACKETIN_HIGH_WATERMARK, messageSpy, REJECTED_DRAIN_FACTOR);
}
/**
* This method is called from {@link DeviceManagerImpl} only. So we could say "posthandshake process finish"
* and we are able to set a scheduler for an automatic transaction submitting by time (0,5sec).
*/
- void submitTransaction() {
- txChainManager.enableSubmit();
- txChainManager.submitTransaction();
+ void initialSubmitTransaction() {
+ txChainManager.initialSubmitWriteTransaction();
+ }
+
+ @Override
+ public Long getReservedXid() {
+ return outboundQueueProvider.reserveEntry();
}
@Override
public <M extends ChildOf<DataObject>> void onMessage(final M message, final RequestContext<?> requestContext) {
// TODO Auto-generated method stub
-
}
@Override
txChainManager.addDeleteOperationTotTxChain(store, path);
}
+ @Override
+ public boolean submitTransaction() {
+ return txChainManager.submitWriteTransaction();
+ }
+
@Override
public ConnectionContext getPrimaryConnectionContext() {
return primaryConnectionContext;
return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
}
-
@Override
public DeviceFlowRegistry getDeviceFlowRegistry() {
return deviceFlowRegistry;
@Override
public void processReply(final OfHeader ofHeader) {
- RpcResult<OfHeader> rpcResult;
if (ofHeader instanceof Error) {
messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
} else {
messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
}
-
}
@Override
}
}
- @Override
- public void processException(final Xid xid, final DeviceDataException deviceDataException) {
- messageSpy.spyMessage(deviceDataException.getClass(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
- }
-
@Override
public void processFlowRemovedMessage(final FlowRemoved flowRemoved) {
//TODO: will be defined later
} else if (portStatus.getReason().equals(PortReason.OFPPRDELETE)) {
addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
}
+ submitTransaction();
}
private KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> provideIIToNodeConnector(final long portNo, final short version) {
@Override
public void processPacketInMessage(final PacketInMessage packetInMessage) {
messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
+ final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
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 (packetReceived == null) {
+ LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
return;
}
- ListenableFuture<?> listenableFuture = notificationPublishService.offerNotification(packetReceived);
- if (NotificationPublishService.REJECTED.equals(listenableFuture)) {
- LOG.debug("notification offer rejected");
- messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
- } else if (listenableFuture.isDone()) {
- Object x = null;
- try {
- x = listenableFuture.get();
- messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
- } catch (InterruptedException e) {
- LOG.debug("notification offer interrupted: {}", e.getMessage());
- LOG.trace("notification offer interrupted..", e);
- } catch (ExecutionException e) {
- LOG.debug("notification offer failed: {}", e.getMessage());
- LOG.trace("notification offer failed..", e);
- } finally {
- if (null == x) {
- messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
- }
- }
- } else {
- messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
+ if (!packetInLimiter.acquirePermit()) {
+ LOG.debug("Packet limited");
+ // TODO: save packet into emergency slot if possible
+ // FIXME: some other counter
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
+ return;
}
- }
-
- private void applyThrottling(final PacketReceived packetReceived, final ConnectionAdapter connectionAdapter) {
- final InetSocketAddress remoteAddress = connectionAdapter.getRemoteAddress();
- LOG.debug("Notification offer refused by notification service.");
- messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
- connectionAdapter.setAutoRead(false);
- LOG.debug("Throttling ingress for {}", remoteAddress);
- final ListenableFuture<Void> queueDone;
-
- // adding first notification
- bumperQueue.offer(packetReceived);
- synchronized (bumperQueue) {
- queueDone = throttledConnectionsHolder.applyThrottlingOnConnection(bumperQueue);
+ final ListenableFuture<? extends Object> offerNotification = notificationPublishService.offerNotification(packetReceived);
+ if (NotificationPublishService.REJECTED.equals(offerNotification)) {
+ LOG.debug("notification offer rejected");
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
+ packetInLimiter.drainLowWaterMark();
+ packetInLimiter.releasePermit();
+ return;
}
- Futures.addCallback(queueDone, new FutureCallback<Void>() {
+
+ Futures.addCallback(offerNotification, new FutureCallback<Object>() {
@Override
- public void onSuccess(final Void result) {
- LOG.debug("Un - throttling ingress for {}", remoteAddress);
- connectionAdapter.setAutoRead(true);
+ public void onSuccess(final Object result) {
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
+ packetInLimiter.releasePermit();
}
@Override
public void onFailure(final Throwable t) {
- LOG.warn("failed to offer queued notification for {}: {}", remoteAddress, t.getMessage());
- LOG.debug("failed to offer queued notification for {}.. ", remoteAddress, t);
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
+ LOG.debug("notification offer failed: {}", t.getMessage());
+ LOG.trace("notification offer failed..", t);
+ packetInLimiter.releasePermit();
}
});
}
}
@Override
- public void close() {
+ 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();
- }
+ primaryConnectionContext.close();
for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
- if (connectionContext.getConnectionAdapter().isAlive()) {
- connectionContext.getConnectionAdapter().disconnect();
- }
+ connectionContext.close();
}
+
for (final DeviceContextClosedHandler deviceContextClosedHandler : closeHandlers) {
deviceContextClosedHandler.onDeviceContextClosed(this);
}
+
+ txChainManager.close();
}
@Override
public void onDeviceDisconnected(final ConnectionContext connectionContext) {
- if (this.getPrimaryConnectionContext().equals(connectionContext)) {
+ if (getPrimaryConnectionContext().equals(connectionContext)) {
try {
close();
} catch (final Exception e) {
@Override
public void addDeviceContextClosedHandler(final DeviceContextClosedHandler deviceContextClosedHandler) {
- this.closeHandlers.add(deviceContextClosedHandler);
+ closeHandlers.add(deviceContextClosedHandler);
}
@Override
- public void startGatheringOperationsToOneTransaction() {
- txChainManager.startGatheringOperationsToOneTransaction();
+ public void onPublished() {
+ primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
+ for (final ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
+ switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
+ }
}
@Override
- public void commitOperationsGatheredInOneTransaction() {
- txChainManager.commitOperationsGatheredInOneTransaction();
+ public MultiMsgCollector getMultiMsgCollector(final RequestContext<List<MultipartReply>> requestContext) {
+ return new MultiMsgCollectorImpl(this, requestContext);
}
-
}