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.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.TreeMap;
-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.device.DeviceContext;
import org.opendaylight.openflowplugin.api.openflow.device.DeviceState;
import org.opendaylight.openflowplugin.api.openflow.device.MessageTranslator;
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.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;
private final DeviceState deviceState;
private final DataBroker dataBroker;
private final HashedWheelTimer hashedWheelTimer;
- private final Map<Long, RequestContext> requests = new TreeMap<>();
-
private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
private final TransactionChainManager txChainManager;
private TranslatorLibrary translatorLibrary;
private DeviceDisconnectedHandler deviceDisconnectedHandler;
private final Collection<DeviceContextClosedHandler> closeHandlers = new HashSet<>();
private NotificationPublishService notificationPublishService;
- private final ThrottledNotificationsOfferer throttledConnectionsHolder;
- private final BlockingQueue<PacketReceived> bumperQueue;
private final OutboundQueue outboundQueueProvider;
+ private final MultiMsgCollector multiMsgCollector = new MultiMsgCollectorImpl();
+
+ private volatile int outstandingNotificationsAmount = 0;
+ private volatile boolean filteringPacketIn = false;
+ private final Object throttlingLock = new Object();
+ private int filteringHighWaterMark = 0;
@Override
public MultiMsgCollector getMultiMsgCollector() {
return outboundQueueProvider.reserveEntry();
}
- private final MultiMsgCollector multiMsgCollector = new MultiMsgCollectorImpl();
-
-
@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) {
this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
this.deviceState = Preconditions.checkNotNull(deviceState);
this.dataBroker = Preconditions.checkNotNull(dataBroker);
deviceGroupRegistry = new DeviceGroupRegistryImpl();
deviceMeterRegistry = new DeviceMeterRegistryImpl();
messageSpy = _messageSpy;
- this.throttledConnectionsHolder = throttledConnectionsHolder;
- bumperQueue = new ArrayBlockingQueue<>(5000);
multiMsgCollector.setDeviceReplyProcessor(this);
outboundQueueProvider = Preconditions.checkNotNull(primaryConnectionContext.getOutboundQueueProvider());
}
}
@Override
- public <M extends ChildOf<DataObject>> void onMessage(final M message, final RequestContext requestContext) {
+ public <M extends ChildOf<DataObject>> void onMessage(final M message, final RequestContext<?> requestContext) {
// TODO Auto-generated method stub
-
}
@Override
return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
}
- @Override
- 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) {
- synchronized (requests) {
- requests.put(xid.getValue(), requestFutureContext);
- }
- }
-
- @Override
- public RequestContext unhookRequestCtx(final Xid xid) {
- synchronized (requests) {
- return requests.remove(xid.getValue());
- }
- }
-
@Override
public DeviceFlowRegistry getDeviceFlowRegistry() {
return deviceFlowRegistry;
@Override
public void processReply(final OfHeader ofHeader) {
- final RequestContext requestContext = requests.remove(ofHeader.getXid());
- if (null != requestContext) {
- 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 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);
- }
-
- requestContext.setResult(rpcResult);
- try {
- requestContext.close();
- } catch (final Exception e) {
- LOG.warn("Closing RequestContext failed: {}", e.getMessage());
- LOG.debug("Closing RequestContext failed.. ", e);
- }
+ if (ofHeader instanceof Error) {
+ messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
} else {
- LOG.warn("Can't find request context registered for xid : {}. Type of reply: {}. From address: {}", ofHeader.getXid(), ofHeader.getClass().getName(),
- getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress());
+ messageSpy.spyMessage(ofHeader.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
}
}
@Override
public void processReply(final Xid xid, final List<MultipartReply> ofHeaderList) {
- final RequestContext requestContext;
- synchronized (requests) {
- requestContext = requests.remove(xid.getValue());
- }
- if (null != requestContext) {
- final RpcResult<List<MultipartReply>> rpcResult = RpcResultBuilder
- .<List<MultipartReply>>success()
- .withResult(ofHeaderList)
- .build();
- requestContext.setResult(rpcResult);
- for (final MultipartReply multipartReply : ofHeaderList) {
- messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
- }
-
- unhookRequestCtx(xid);
- try {
- requestContext.close();
- } catch (final Exception e) {
- LOG.warn("Closing RequestContext failed: {}", e.getMessage());
- LOG.debug("Closing RequestContext failed.. ", e);
- }
- } else {
- LOG.warn("Can't find request context registered for xid : {}. Type of reply: MULTIPART. From address: {}", xid.getValue(),
- getPrimaryConnectionContext().getConnectionAdapter().getRemoteAddress());
+ for (final MultipartReply multipartReply : ofHeaderList) {
+ messageSpy.spyMessage(multipartReply.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
}
}
@Override
public void processException(final Xid xid, final DeviceDataException deviceDataException) {
-
- LOG.trace("Processing exception for xid : {}", xid.getValue());
-
- final RequestContext requestContext = requests.remove(xid.getValue());
-
- if (null != requestContext) {
- final RpcResult<List<OfHeader>> rpcResult = RpcResultBuilder
- .<List<OfHeader>>failed()
- .withError(RpcError.ErrorType.APPLICATION, String.format("Message processing failed : %s", deviceDataException.getError()), deviceDataException)
- .build();
- requestContext.setResult(rpcResult);
- messageSpy.spyMessage(deviceDataException.getClass(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
- try {
- requestContext.close();
- } catch (final Exception e) {
- LOG.warn("Closing RequestContext failed: ", e);
- LOG.debug("Closing RequestContext failed..", e);
- }
- } else {
- LOG.warn("Can't find request context registered for xid : {}. Exception message {}",
- xid.getValue(), deviceDataException.getMessage());
- }
+ messageSpy.spyMessage(deviceDataException.getClass(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_FAILURE);
}
@Override
@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");
return;
}
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_TRANSLATE_SRC_FAILURE);
- ListenableFuture<?> listenableFuture = notificationPublishService.offerNotification(packetReceived);
- if (NotificationPublishService.REJECTED.equals(listenableFuture)) {
+ ListenableFuture<? extends Object> offerNotification = notificationPublishService.offerNotification(packetReceived);
+ synchronized (throttlingLock) {
+ outstandingNotificationsAmount += 1;
+ }
+ if (NotificationPublishService.REJECTED.equals(offerNotification)) {
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);
+ synchronized (throttlingLock) {
+ if (outstandingNotificationsAmount > 1 && !filteringPacketIn) {
+ connectionAdapter.setPacketInFiltering(true);
+ messageSpy.spyMessage(DeviceContext.class, MessageSpy.STATISTIC_GROUP.OFJ_BACKPRESSURE_ON);
+ filteringPacketIn = true;
+ filteringHighWaterMark = outstandingNotificationsAmount;
+ LOG.debug("PacketIn filtering on: {}, watermark: {}", connectionAdapter.getRemoteAddress(), outstandingNotificationsAmount);
}
}
- } else {
- messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
}
- }
-
- 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);
- }
- Futures.addCallback(queueDone, new FutureCallback<Void>() {
- @Override
- public void onSuccess(final Void result) {
- LOG.debug("Un - throttling ingress for {}", remoteAddress);
- connectionAdapter.setAutoRead(true);
- }
-
- @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);
- }
- });
+ Futures.addCallback(offerNotification,
+ new FutureCallback<Object>() {
+ @Override
+ public void onSuccess(final Object result) {
+ countdownFiltering();
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
+ }
+
+ @Override
+ public void onFailure(final Throwable t) {
+ countdownFiltering();
+ messageSpy.spyMessage(packetReceived.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_NOTIFICATION_REJECTED);
+ LOG.debug("notification offer failed: {}, outstanding: {}", t.getMessage(), outstandingNotificationsAmount);
+ LOG.trace("notification offer failed..", t);
+ }
+
+ private void countdownFiltering() {
+ synchronized (throttlingLock) {
+ outstandingNotificationsAmount -= 1;
+ if (outstandingNotificationsAmount == 0 && filteringPacketIn) {
+ connectionAdapter.setPacketInFiltering(false);
+ messageSpy.spyMessage(DeviceContext.class, MessageSpy.STATISTIC_GROUP.OFJ_BACKPRESSURE_OFF);
+
+ filteringPacketIn = false;
+ LOG.debug("PacketIn filtering off: {}, outstanding: {}", connectionAdapter.getRemoteAddress(), outstandingNotificationsAmount);
+ }
+ }
+ }
+ }
+ );
}
@Override
}
@Override
- public void close() throws Exception {
+ public void close() {
deviceState.setValid(false);
LOG.trace("Removing node {} from operational DS.", getDeviceState().getNodeId());
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
}
}
- @Override
- public RequestContext extractNextOutstandingMessage(final long barrierXid) {
- RequestContext nextMessage = null;
- 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 setCurrentBarrierTimeout(final Timeout timeout) {
barrierTaskTimeout = timeout;
txChainManager.commitOperationsGatheredInOneTransaction();
}
+ @Override
+ public void onPublished() {
+ primaryConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
+ for (ConnectionContext switchAuxConnectionContext : auxiliaryConnectionContexts.values()) {
+ switchAuxConnectionContext.getConnectionAdapter().setPacketInFiltering(false);
+ }
+ }
}