import com.google.common.base.Verify;
import com.google.common.util.concurrent.AsyncFunction;
import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.FutureFallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
-import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import java.math.BigInteger;
-import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.CheckForNull;
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.ReadOnlyTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
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.handlers.DeviceTerminationPhaseHandler;
import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
+import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor;
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.ItemLifeCycleRegistry;
import org.opendaylight.openflowplugin.extension.api.core.extension.ExtensionConverterProvider;
import org.opendaylight.openflowplugin.extension.api.exception.ConversionException;
import org.opendaylight.openflowplugin.extension.api.path.MessagePath;
-import org.opendaylight.openflowplugin.impl.LifecycleConductor;
import org.opendaylight.openflowplugin.impl.common.ItemLifeCycleSourceImpl;
import org.opendaylight.openflowplugin.impl.common.NodeStaticReplyTranslatorUtil;
import org.opendaylight.openflowplugin.impl.device.listener.MultiMsgCollectorImpl;
import org.opendaylight.openflowplugin.impl.registry.flow.FlowRegistryKeyFactory;
import org.opendaylight.openflowplugin.impl.registry.group.DeviceGroupRegistryImpl;
import org.opendaylight.openflowplugin.impl.registry.meter.DeviceMeterRegistryImpl;
+import org.opendaylight.openflowplugin.impl.rpc.RpcContextImpl;
import org.opendaylight.openflowplugin.impl.util.DeviceInitializationUtils;
import org.opendaylight.openflowplugin.impl.util.MdSalRegistrationUtils;
import org.opendaylight.openflowplugin.openflow.md.core.session.SwitchConnectionCookieOFImpl;
private final ConnectionContext primaryConnectionContext;
private final DeviceState deviceState;
private final DataBroker dataBroker;
- private final HashedWheelTimer hashedWheelTimer;
private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
private final TransactionChainManager transactionChainManager;
private final DeviceFlowRegistry deviceFlowRegistry;
private final DeviceGroupRegistry deviceGroupRegistry;
private final DeviceMeterRegistry deviceMeterRegistry;
- private final Collection<DeviceTerminationPhaseHandler> closeHandlers = new HashSet<>();
private final PacketInRateLimiter packetInLimiter;
private final MessageSpy messageSpy;
private final ItemLifeCycleKeeper flowLifeCycleKeeper;
private NotificationPublishService notificationPublishService;
- private NotificationService notificationService;
private final OutboundQueue outboundQueueProvider;
private Timeout barrierTaskTimeout;
private final MessageTranslator<PortGrouping, FlowCapableNodeConnector> portStatusTranslator;
private final NodeId nodeId;
private volatile DEVICE_CONTEXT_STATE deviceCtxState;
+ private boolean isStatisticsRpcEnabled;
@VisibleForTesting
DeviceContextImpl(@Nonnull final ConnectionContext primaryConnectionContext,
@Nonnull final DeviceState deviceState,
@Nonnull final DataBroker dataBroker,
- @Nonnull final HashedWheelTimer hashedWheelTimer,
- @Nonnull final MessageSpy _messageSpy,
+ @Nonnull final LifecycleConductor conductor,
@Nonnull final OutboundQueueProvider outboundQueueProvider,
@Nonnull final TranslatorLibrary translatorLibrary,
final boolean switchFeaturesMandatory) {
this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
this.deviceState = Preconditions.checkNotNull(deviceState);
this.dataBroker = Preconditions.checkNotNull(dataBroker);
- this.hashedWheelTimer = Preconditions.checkNotNull(hashedWheelTimer);
+ Preconditions.checkNotNull(conductor);
this.outboundQueueProvider = Preconditions.checkNotNull(outboundQueueProvider);
- this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState);
+ this.transactionChainManager = new TransactionChainManager(dataBroker, deviceState, conductor);
auxiliaryConnectionContexts = new HashMap<>();
deviceFlowRegistry = new DeviceFlowRegistryImpl();
deviceGroupRegistry = new DeviceGroupRegistryImpl();
deviceMeterRegistry = new DeviceMeterRegistryImpl();
- messageSpy = _messageSpy;
+ messageSpy = conductor.getMessageIntelligenceAgency();
packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
/*initial*/ 1000, /*initial*/2000, messageSpy, REJECTED_DRAIN_FACTOR);
}
@Override
- public Long reservedXidForDeviceMessage() {
+ public Long reserveXidForDeviceMessage() {
return outboundQueueProvider.reserveEntry();
}
@Override
public void removeAuxiliaryConnectionContext(final ConnectionContext connectionContext) {
final SwitchConnectionDistinguisher connectionDistinguisher = createConnectionDistinguisher(connectionContext);
- if (null != connectionDistinguisher) {
- LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
- .getRemoteAddress(), nodeId);
- auxiliaryConnectionContexts.remove(connectionDistinguisher);
- }
+ LOG.debug("auxiliary connection dropped: {}, nodeId:{}", connectionContext.getConnectionAdapter()
+ .getRemoteAddress(), nodeId);
+ auxiliaryConnectionContexts.remove(connectionDistinguisher);
}
@Override
}
/* Routed RPC registration */
MdSalRegistrationUtils.registerMasterServices(getRpcContext(), DeviceContextImpl.this, OfpRole.BECOMEMASTER);
- getRpcContext().registerStatCompatibilityServices();
+
+ if (isStatisticsRpcEnabled) {
+ MdSalRegistrationUtils.registerStatCompatibilityServices(getRpcContext(), this,
+ notificationPublishService, new AtomicLong());
+ }
/* Prepare init info collecting */
getDeviceState().setDeviceSynchronized(false);
new AsyncFunction<Void, Boolean>() {
@Override
- public ListenableFuture<Boolean> apply(final Void input) throws Exception {
+ public ListenableFuture<Boolean> apply(@Nonnull final Void input) throws Exception {
getStatisticsContext().statListForCollectingInitialization();
return getStatisticsContext().gatherDynamicData();
}
LOG.warn(errMsg);
throw new IllegalStateException(errMsg);
}
- if (!input.booleanValue()) {
+ if (!input) {
final String errMsg = String.format("Get Initial Device %s information fails",
getDeviceState().getNodeId());
LOG.warn(errMsg);
@Override
public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
final InstanceIdentifier<T> path, final T data) throws Exception {
- transactionChainManager.writeToTransaction(store, path, data);
+ transactionChainManager.writeToTransaction(store, path, data, false);
+ }
+
+ @Override
+ public <T extends DataObject> void writeToTransactionWithParentsSlow(LogicalDatastoreType store, InstanceIdentifier<T> path, T data) throws Exception {
+ transactionChainManager.writeToTransaction(store, path, data, true);
}
@Override
if (itemLifecycleListener != null) {
//1. translate to general flow (table, priority, match, cookie)
final org.opendaylight.yang.gen.v1.urn.opendaylight.flow.service.rev130819.FlowRemoved flowRemovedNotification =
- flowRemovedTranslator.translate(flowRemoved, this, null);
+ flowRemovedTranslator.translate(flowRemoved, this.getDeviceState(), null);
//2. create registry key
final FlowRegistryKey flowRegKey = FlowRegistryKeyFactory.create(flowRemovedNotification);
//3. lookup flowId
@Override
public void processPortStatusMessage(final PortStatusMessage portStatus) {
messageSpy.spyMessage(portStatus.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH_PUBLISHED_SUCCESS);
- final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this, null);
+ final FlowCapableNodeConnector flowCapableNodeConnector = portStatusTranslator.translate(portStatus, this.getDeviceState(), null);
final KeyedInstanceIdentifier<NodeConnector, NodeConnectorKey> iiToNodeConnector = provideIIToNodeConnector(portStatus.getPortNo(), portStatus.getVersion());
try {
addDeleteToTxChain(LogicalDatastoreType.OPERATIONAL, iiToNodeConnector);
}
submitTransaction();
- } catch (Exception e) {
+ } catch (final Exception e) {
LOG.warn("Error processing port status message: {}", e.getMessage());
}
}
public void processPacketInMessage(final PacketInMessage packetInMessage) {
messageSpy.spyMessage(packetInMessage.getImplementedInterface(), MessageSpy.STATISTIC_GROUP.FROM_SWITCH);
final ConnectionAdapter connectionAdapter = getPrimaryConnectionContext().getConnectionAdapter();
- final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this, null);
+ final PacketReceived packetReceived = packetInTranslator.translate(packetInMessage, this.getDeviceState(), null);
if (packetReceived == null) {
LOG.debug("Received a null packet from switch {}", connectionAdapter.getRemoteAddress());
// publish
notificationPublishService.offerNotification(experimenterMessageFromDevBld.build());
} catch (final ConversionException e) {
- LOG.warn("Conversion of experimenter notification failed", e);
+ LOG.error("Conversion of experimenter notification failed", e);
}
}
return translatorLibrary;
}
- @Override
- public HashedWheelTimer getTimer() {
- return hashedWheelTimer;
- }
-
@Override
public synchronized void close() {
LOG.debug("closing deviceContext: {}, nodeId:{}",
return barrierTaskTimeout;
}
- @Override
- public void setNotificationService(final NotificationService notificationService) {
- this.notificationService = notificationService;
- }
-
@Override
public void setNotificationPublishService(final NotificationPublishService notificationPublishService) {
this.notificationPublishService = notificationPublishService;
}
@Override
- public void storeNodeConnectorRef(final Long portNumber, final NodeConnectorRef nodeConnectorRef) {
+ public void storeNodeConnectorRef(@Nonnull final Long portNumber, @Nonnull final NodeConnectorRef nodeConnectorRef) {
nodeConnectorCache.put(
Preconditions.checkNotNull(portNumber),
Preconditions.checkNotNull(nodeConnectorRef));
deviceMeterRegistry.close();
}
+ @Override
+ public void setStatisticsRpcEnabled(boolean isStatisticsRpcEnabled) {
+ this.isStatisticsRpcEnabled = isStatisticsRpcEnabled;
+ }
+
@Override
public DEVICE_CONTEXT_STATE getDeviceContextState() {
return deviceCtxState;
deviceState.setValid(false);
return transactionChainManager.shuttingDown();
}
+
+ @VisibleForTesting
+ TransactionChainManager getTransactionChainManager() {
+ return this.transactionChainManager;
+ }
}
Code Review / openflowplugin.git / blobdiff