import org.opendaylight.controller.md.sal.binding.api.NotificationPublishService;
import org.opendaylight.controller.md.sal.binding.api.ReadOnlyTransaction;
import org.opendaylight.controller.md.sal.common.api.data.LogicalDatastoreType;
-import org.opendaylight.controller.md.sal.common.api.data.TransactionChainClosedException;
import org.opendaylight.mdsal.singleton.common.api.ServiceGroupIdentifier;
import org.opendaylight.openflowjava.protocol.api.connection.ConnectionAdapter;
import org.opendaylight.openflowjava.protocol.api.keys.MessageTypeKey;
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.ClusterInitializationPhaseHandler;
import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService;
import org.opendaylight.openflowplugin.api.openflow.md.core.SwitchConnectionDistinguisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-/**
- *
- */
public class DeviceContextImpl implements DeviceContext, ExtensionConverterProviderKeeper{
private static final Logger LOG = LoggerFactory.getLogger(DeviceContextImpl.class);
private static final float LOW_WATERMARK_FACTOR = 0.75f;
// TODO: high water mark factor should be parametrized
private static final float HIGH_WATERMARK_FACTOR = 0.95f;
+ private boolean initialized;
private ConnectionContext primaryConnectionContext;
private final DeviceState deviceState;
private final DataBroker dataBroker;
private final Map<SwitchConnectionDistinguisher, ConnectionContext> auxiliaryConnectionContexts;
- private final TransactionChainManager transactionChainManager;
- private final DeviceFlowRegistry deviceFlowRegistry;
- private final DeviceGroupRegistry deviceGroupRegistry;
- private final DeviceMeterRegistry deviceMeterRegistry;
+ private TransactionChainManager transactionChainManager;
+ private DeviceFlowRegistry deviceFlowRegistry;
+ private DeviceGroupRegistry deviceGroupRegistry;
+ private DeviceMeterRegistry deviceMeterRegistry;
private final PacketInRateLimiter packetInLimiter;
private final MessageSpy messageSpy;
private final ItemLifeCycleKeeper flowLifeCycleKeeper;
private final ItemLifeCycleRegistry itemLifeCycleSourceRegistry;
private ExtensionConverterProvider extensionConverterProvider;
private final DeviceManager deviceManager;
-
+ private boolean skipTableFeatures;
private boolean switchFeaturesMandatory;
private final DeviceInfo deviceInfo;
private final ConvertorExecutor convertorExecutor;
private volatile CONTEXT_STATE state;
+ private ClusterInitializationPhaseHandler clusterInitializationPhaseHandler;
- public DeviceContextImpl(
+ DeviceContextImpl(
@Nonnull final ConnectionContext primaryConnectionContext,
@Nonnull final DataBroker dataBroker,
@Nonnull final MessageSpy messageSpy,
@Nonnull final TranslatorLibrary translatorLibrary,
@Nonnull final DeviceManager manager,
- final ConvertorExecutor convertorExecutor) {
- this.primaryConnectionContext = Preconditions.checkNotNull(primaryConnectionContext);
+ final ConvertorExecutor convertorExecutor,
+ final boolean skipTableFeatures) {
+ this.primaryConnectionContext = primaryConnectionContext;
this.deviceInfo = primaryConnectionContext.getDeviceInfo();
this.deviceState = new DeviceStateImpl();
- this.dataBroker = Preconditions.checkNotNull(dataBroker);
- this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo);
- auxiliaryConnectionContexts = new HashMap<>();
- deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier());
- deviceGroupRegistry = new DeviceGroupRegistryImpl();
- deviceMeterRegistry = new DeviceMeterRegistryImpl();
+ this.dataBroker = dataBroker;
+ this.auxiliaryConnectionContexts = new HashMap<>();
this.messageSpy = Preconditions.checkNotNull(messageSpy);
- this.deviceManager = Preconditions.checkNotNull(manager);
+ this.deviceManager = manager;
- packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
+ this.packetInLimiter = new PacketInRateLimiter(primaryConnectionContext.getConnectionAdapter(),
/*initial*/ 1000, /*initial*/2000, this.messageSpy, REJECTED_DRAIN_FACTOR);
this.translatorLibrary = translatorLibrary;
- portStatusTranslator = translatorLibrary.lookupTranslator(
+ this.portStatusTranslator = translatorLibrary.lookupTranslator(
new TranslatorKey(deviceInfo.getVersion(), PortGrouping.class.getName()));
- packetInTranslator = translatorLibrary.lookupTranslator(
+ this.packetInTranslator = translatorLibrary.lookupTranslator(
new TranslatorKey(deviceInfo.getVersion(), PacketIn.class.getName()));
- flowRemovedTranslator = translatorLibrary.lookupTranslator(
+ this.flowRemovedTranslator = translatorLibrary.lookupTranslator(
new TranslatorKey(deviceInfo.getVersion(), FlowRemoved.class.getName()));
- itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
- flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
- itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
+ this.itemLifeCycleSourceRegistry = new ItemLifeCycleRegistryImpl();
+ this.flowLifeCycleKeeper = new ItemLifeCycleSourceImpl();
+ this.itemLifeCycleSourceRegistry.registerLifeCycleSource(flowLifeCycleKeeper);
this.state = CONTEXT_STATE.INITIALIZATION;
this.convertorExecutor = convertorExecutor;
+ this.skipTableFeatures = skipTableFeatures;
+ this.initialized = false;
}
- /**
- * 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).
- */
@Override
public void initialSubmitTransaction() {
- transactionChainManager.initialSubmitWriteTransaction();
+ if (initialized) {
+ transactionChainManager.initialSubmitWriteTransaction();
+ }
}
@Override
public <T extends DataObject> void writeToTransaction(final LogicalDatastoreType store,
final InstanceIdentifier<T> path,
final T data){
- transactionChainManager.writeToTransaction(store, path, data, false);
+ if (initialized) {
+ transactionChainManager.writeToTransaction(store, path, data, false);
+ }
}
@Override
public <T extends DataObject> void writeToTransactionWithParentsSlow(final LogicalDatastoreType store,
final InstanceIdentifier<T> path,
final T data){
- transactionChainManager.writeToTransaction(store, path, data, true);
+ if (initialized) {
+ transactionChainManager.writeToTransaction(store, path, data, true);
+ }
}
@Override
- public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) throws TransactionChainClosedException {
- transactionChainManager.addDeleteOperationTotTxChain(store, path);
+ public <T extends DataObject> void addDeleteToTxChain(final LogicalDatastoreType store, final InstanceIdentifier<T> path) {
+ if (initialized) {
+ transactionChainManager.addDeleteOperationTotTxChain(store, path);
+ }
}
@Override
public boolean submitTransaction() {
- return transactionChainManager.submitWriteTransaction();
+ return initialized && transactionChainManager.submitWriteTransaction();
}
@Override
}
@Override
- public ConnectionContext getAuxiliaryConnectiobContexts(final BigInteger cookie) {
+ public ConnectionContext getAuxiliaryConnectionContexts(final BigInteger cookie) {
return auxiliaryConnectionContexts.get(new SwitchConnectionCookieOFImpl(cookie.longValue()));
}
}
submitTransaction();
} catch (final Exception e) {
- LOG.warn("Error processing port status message: ", e);
+ LOG.warn("Error processing port status message for port {} on device {} : {}", portStatus.getPortNo(),
+ getDeviceInfo().getNodeId().toString(), e);
}
}
@Override
public synchronized void shutdownConnection() {
- LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Shutdown method for node {}", getDeviceInfo().getLOGValue());
+ }
if (CONTEXT_STATE.TERMINATION.equals(getState())) {
LOG.debug("DeviceCtx for Node {} is in termination process.", getDeviceInfo().getLOGValue());
return;
}
- setState(CONTEXT_STATE.TERMINATION);
if (ConnectionContext.CONNECTION_STATE.RIP.equals(getPrimaryConnectionContext().getConnectionState())) {
LOG.debug("ConnectionCtx for Node {} is in RIP state.", getDeviceInfo().getLOGValue());
return;
}
- /* Terminate Auxiliary Connection */
+
+ // Terminate Auxiliary Connection
for (final ConnectionContext connectionContext : auxiliaryConnectionContexts.values()) {
LOG.debug("Closing auxiliary connection {}", connectionContext.getNodeId());
connectionContext.closeConnection(false);
}
- /* Terminate Primary Connection */
+
+ // Terminate Primary Connection
getPrimaryConnectionContext().closeConnection(true);
- /* Close all Group Registry */
- deviceGroupRegistry.close();
- deviceFlowRegistry.close();
- deviceMeterRegistry.close();
+
+ // Close all datastore registries
+ if (initialized) {
+ deviceGroupRegistry.close();
+ deviceFlowRegistry.close();
+ deviceMeterRegistry.close();
+ }
}
@Override
public ListenableFuture<Void> shuttingDownDataStoreTransactions() {
- return transactionChainManager.shuttingDown();
+ return initialized
+ ? this.transactionChainManager.shuttingDown()
+ : Futures.immediateFuture(null);
}
@VisibleForTesting
this.state = state;
}
- @Override
- public void startupClusterServices() throws ExecutionException, InterruptedException {
- LOG.debug("Initializing transaction chain manager for node {}", getDeviceInfo().getLOGValue());
- this.transactionChainManager.activateTransactionManager();
- LOG.debug("Waiting to get node {} information", getDeviceInfo().getLOGValue());
- DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor);
- }
-
@Override
public ListenableFuture<Void> stopClusterServices(boolean deviceDisconnected) {
- return this.transactionChainManager.deactivateTransactionManager();
+ return initialized
+ ? this.transactionChainManager.deactivateTransactionManager()
+ : Futures.immediateFuture(null);
}
@Override
@Override
public void putLifecycleServiceIntoTxChainManager(final LifecycleService lifecycleService){
- this.transactionChainManager.setLifecycleService(lifecycleService);
+ if (initialized) {
+ this.transactionChainManager.setLifecycleService(lifecycleService);
+ }
}
@Override
this.primaryConnectionContext = connectionContext;
this.onPublished();
}
+
+ @Override
+ public boolean isSkipTableFeatures() {
+ return this.skipTableFeatures;
+ }
+
+ @Override
+ public void setLifecycleInitializationPhaseHandler(final ClusterInitializationPhaseHandler handler) {
+ this.clusterInitializationPhaseHandler = handler;
+ }
+
+ @Override
+ public boolean onContextInstantiateService(final ConnectionContext connectionContext) {
+
+ if (getPrimaryConnectionContext().getConnectionState().equals(ConnectionContext.CONNECTION_STATE.RIP)) {
+ LOG.warn("Connection on device {} was interrupted, will stop starting master services.", deviceInfo.getLOGValue());
+ return false;
+ }
+
+ LOG.info("Starting device context cluster services for node {}", deviceInfo.getLOGValue());
+
+ lazyTransactionManagerInitialization();
+
+ this.transactionChainManager.activateTransactionManager();
+
+ try {
+ DeviceInitializationUtils.initializeNodeInformation(this, switchFeaturesMandatory, this.convertorExecutor);
+ } catch (ExecutionException | InterruptedException e) {
+ LOG.warn("Device {} cannot be initialized: ", deviceInfo.getLOGValue(), e);
+ return false;
+ }
+
+ return this.clusterInitializationPhaseHandler.onContextInstantiateService(getPrimaryConnectionContext());
+ }
+
+ @VisibleForTesting
+ void lazyTransactionManagerInitialization() {
+ if (!this.initialized) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Transaction chain manager for node {} created", deviceInfo.getLOGValue());
+ }
+ this.transactionChainManager = new TransactionChainManager(dataBroker, deviceInfo);
+ this.deviceFlowRegistry = new DeviceFlowRegistryImpl(dataBroker, deviceInfo.getNodeInstanceIdentifier());
+ this.deviceGroupRegistry = new DeviceGroupRegistryImpl();
+ this.deviceMeterRegistry = new DeviceMeterRegistryImpl();
+ this.initialized = true;
+ }
+ }
}
Code Review / openflowplugin.git / blobdiff