X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=openflowplugin-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fopenflowplugin%2Fimpl%2Fstatistics%2FStatisticsManagerImpl.java;h=6d726ad071f13cb10c1b30aa4575795f200c9502;hb=277d201d5db47620b29a6a69fd99aec539e537eb;hp=b6dbb56c61d8bd33b153c2e47a1a8c289d63caf0;hpb=86c852307c5484cc6a5545453b8309315a389af7;p=openflowplugin.git diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java index b6dbb56c61..6d726ad071 100644 --- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java +++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsManagerImpl.java @@ -9,26 +9,37 @@ package org.opendaylight.openflowplugin.impl.statistics; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Optional; import com.google.common.base.Preconditions; import com.google.common.base.Verify; import com.google.common.collect.Iterators; import com.google.common.util.concurrent.FutureCallback; 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 io.netty.util.TimerTask; +import java.util.Iterator; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CancellationException; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nonnull; import org.opendaylight.controller.sal.binding.api.BindingAwareBroker; import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry; import org.opendaylight.openflowplugin.api.openflow.device.DeviceContext; import org.opendaylight.openflowplugin.api.openflow.device.DeviceInfo; +import org.opendaylight.openflowplugin.api.openflow.device.DeviceState; import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceInitializationPhaseHandler; import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceTerminationPhaseHandler; -import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleConductor; +import org.opendaylight.openflowplugin.api.openflow.lifecycle.LifecycleService; import org.opendaylight.openflowplugin.api.openflow.rpc.ItemLifeCycleSource; import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsContext; import org.opendaylight.openflowplugin.api.openflow.statistics.StatisticsManager; -import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeId; +import org.opendaylight.openflowplugin.openflow.md.core.sal.convertor.ConvertorExecutor; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.ChangeStatisticsWorkModeInput; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutput; import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.openflowplugin.sm.control.rev150812.GetStatisticsWorkModeOutputBuilder; @@ -40,26 +51,17 @@ import org.opendaylight.yangtools.yang.common.RpcResultBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.CheckForNull; -import javax.annotation.Nonnull; -import java.util.Iterator; -import java.util.concurrent.CancellationException; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Future; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; - public class StatisticsManagerImpl implements StatisticsManager, StatisticsManagerControlService { private static final Logger LOG = LoggerFactory.getLogger(StatisticsManagerImpl.class); private static final long DEFAULT_STATS_TIMEOUT_SEC = 50L; + private final ConvertorExecutor convertorExecutor; private DeviceInitializationPhaseHandler deviceInitPhaseHandler; private DeviceTerminationPhaseHandler deviceTerminPhaseHandler; - private final ConcurrentMap contexts = new ConcurrentHashMap<>(); + private final ConcurrentMap contexts = new ConcurrentHashMap<>(); private static final long basicTimerDelay = 3000; private static long currentTimerDelay = basicTimerDelay; @@ -67,62 +69,49 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag private StatisticsWorkMode workMode = StatisticsWorkMode.COLLECTALL; private final Semaphore workModeGuard = new Semaphore(1, true); - private boolean shuttingDownStatisticsPolling; + private boolean isStatisticsPollingEnabled; private BindingAwareBroker.RpcRegistration controlServiceRegistration; - private final LifecycleConductor conductor; + private final HashedWheelTimer hashedWheelTimer; @Override public void setDeviceInitializationPhaseHandler(final DeviceInitializationPhaseHandler handler) { deviceInitPhaseHandler = handler; } - public StatisticsManagerImpl(@CheckForNull final RpcProviderRegistry rpcProviderRegistry, - final boolean shuttingDownStatisticsPolling, - final LifecycleConductor lifecycleConductor) { + public StatisticsManagerImpl(final RpcProviderRegistry rpcProviderRegistry, + final boolean isStatisticsPollingEnabled, + final HashedWheelTimer hashedWheelTimer, + final ConvertorExecutor convertorExecutor) { Preconditions.checkArgument(rpcProviderRegistry != null); + this.convertorExecutor = convertorExecutor; this.controlServiceRegistration = Preconditions.checkNotNull(rpcProviderRegistry.addRpcImplementation( StatisticsManagerControlService.class, this)); - this.shuttingDownStatisticsPolling = shuttingDownStatisticsPolling; - this.conductor = lifecycleConductor; + this.isStatisticsPollingEnabled = isStatisticsPollingEnabled; + this.hashedWheelTimer = hashedWheelTimer; } @Override - public void onDeviceContextLevelUp(final DeviceInfo deviceInfo) throws Exception { + public void onDeviceContextLevelUp(final DeviceInfo deviceInfo, final LifecycleService lifecycleService) throws Exception { - final DeviceContext deviceContext = Preconditions.checkNotNull(conductor.getDeviceContext(deviceInfo.getNodeId())); - - final StatisticsContext statisticsContext = new StatisticsContextImpl(deviceInfo.getNodeId(), shuttingDownStatisticsPolling, conductor); - Verify.verify(contexts.putIfAbsent(deviceInfo.getNodeId(), statisticsContext) == null, "StatisticsCtx still not closed for Node {}", deviceInfo.getNodeId()); - - deviceContext.getDeviceState().setDeviceSynchronized(true); - deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo); + final StatisticsContext statisticsContext = new StatisticsContextImpl(deviceInfo, isStatisticsPollingEnabled, lifecycleService, convertorExecutor, this); + Verify.verify(contexts.putIfAbsent(deviceInfo, statisticsContext) == null, "StatisticsCtx still not closed for Node {}", deviceInfo.getLOGValue()); + lifecycleService.setStatContext(statisticsContext); + deviceInitPhaseHandler.onDeviceContextLevelUp(deviceInfo, lifecycleService); } @VisibleForTesting - void pollStatistics(final DeviceContext deviceContext, - final StatisticsContext statisticsContext, - final TimeCounter timeCounter) { - - final NodeId nodeId = deviceContext.getDeviceState().getNodeId(); + void pollStatistics(final DeviceState deviceState, + final StatisticsContext statisticsContext, + final TimeCounter timeCounter, + final DeviceInfo deviceInfo) { if (!statisticsContext.isSchedulingEnabled()) { - LOG.debug("Disabling statistics scheduling for device: {}", nodeId); - return; - } - - if (!deviceContext.getDeviceState().isValid()) { - LOG.debug("Session is not valid for device: {}", nodeId); - return; - } - - if (!deviceContext.getDeviceState().isStatisticsPollingEnabled()) { - LOG.debug("Statistics polling is currently disabled for device: {}", nodeId); - scheduleNextPolling(deviceContext, statisticsContext, timeCounter); + LOG.debug("Disabled statistics scheduling for device: {}", deviceInfo.getNodeId().getValue()); return; } - LOG.debug("POLLING ALL STATISTICS for device: {}", nodeId); + LOG.debug("POLLING ALL STATISTICS for device: {}", deviceInfo.getNodeId()); timeCounter.markStart(); final ListenableFuture deviceStatisticsCollectionFuture = statisticsContext.gatherDynamicData(); Futures.addCallback(deviceStatisticsCollectionFuture, new FutureCallback() { @@ -130,51 +119,49 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag public void onSuccess(final Boolean o) { timeCounter.addTimeMark(); calculateTimerDelay(timeCounter); - scheduleNextPolling(deviceContext, statisticsContext, timeCounter); + scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter); } @Override public void onFailure(@Nonnull final Throwable throwable) { timeCounter.addTimeMark(); - LOG.warn("Statistics gathering for single node was not successful: {}", throwable.getMessage()); - LOG.trace("Statistics gathering for single node was not successful.. ", throwable); + LOG.warn("Statistics gathering for single node was not successful: {}", throwable); calculateTimerDelay(timeCounter); if (throwable instanceof CancellationException) { - /** This often happens when something wrong with akka or DS, so closing connection will help to restart device **/ - conductor.closeConnection(deviceContext.getDeviceState().getNodeId()); + /* This often happens when something wrong with akka or DS, so closing connection will help to restart device **/ + contexts.get(deviceInfo).getLifecycleService().closeConnection(); } else { - scheduleNextPolling(deviceContext, statisticsContext, timeCounter); + scheduleNextPolling(deviceState, deviceInfo, statisticsContext, timeCounter); } } }); final long averageTime = TimeUnit.MILLISECONDS.toSeconds(timeCounter.getAverageTimeBetweenMarks()); - final long STATS_TIMEOUT_SEC = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC; - final TimerTask timerTask = new TimerTask() { - - @Override - public void run(final Timeout timeout) throws Exception { - if (!deviceStatisticsCollectionFuture.isDone()) { - LOG.info("Statistics collection for node {} still in progress even after {} secs", nodeId, STATS_TIMEOUT_SEC); - deviceStatisticsCollectionFuture.cancel(true); - } + final long statsTimeoutSec = averageTime > 0 ? 3 * averageTime : DEFAULT_STATS_TIMEOUT_SEC; + final TimerTask timerTask = timeout -> { + if (!deviceStatisticsCollectionFuture.isDone()) { + LOG.info("Statistics collection for node {} still in progress even after {} secs", deviceInfo.getLOGValue(), statsTimeoutSec); + deviceStatisticsCollectionFuture.cancel(true); } }; - conductor.newTimeout(timerTask, STATS_TIMEOUT_SEC, TimeUnit.SECONDS); + hashedWheelTimer.newTimeout(timerTask, statsTimeoutSec, TimeUnit.SECONDS); } - private void scheduleNextPolling(final DeviceContext deviceContext, + private void scheduleNextPolling(final DeviceState deviceState, + final DeviceInfo deviceInfo, final StatisticsContext statisticsContext, final TimeCounter timeCounter) { - LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceContext.getDeviceState().getNodeId()); - if (!shuttingDownStatisticsPolling) { - final Timeout pollTimeout = conductor.newTimeout(new TimerTask() { - @Override - public void run(final Timeout timeout) throws Exception { - pollStatistics(deviceContext, statisticsContext, timeCounter); - } - }, currentTimerDelay, TimeUnit.MILLISECONDS); + LOG.debug("SCHEDULING NEXT STATISTICS POLLING for device: {}", deviceInfo.getNodeId()); + if (!isStatisticsPollingEnabled) { + final Timeout pollTimeout = hashedWheelTimer.newTimeout( + timeout -> pollStatistics( + deviceState, + statisticsContext, + timeCounter, + deviceInfo), + currentTimerDelay, + TimeUnit.MILLISECONDS); statisticsContext.setPollTimeout(pollTimeout); } } @@ -202,13 +189,13 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag } @Override - public void onDeviceContextLevelDown(final DeviceContext deviceContext) { - final StatisticsContext statisticsContext = contexts.remove(deviceContext.getDeviceState().getNodeId()); + public void onDeviceContextLevelDown(final DeviceInfo deviceInfo) { + final StatisticsContext statisticsContext = contexts.remove(deviceInfo); if (null != statisticsContext) { - LOG.trace("Removing device context from stack. No more statistics gathering for device: {}", deviceContext.getDeviceState().getNodeId()); + LOG.debug("Removing device context from stack. No more statistics gathering for device: {}", deviceInfo.getLOGValue()); statisticsContext.close(); } - deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceContext); + deviceTerminPhaseHandler.onDeviceContextLevelDown(deviceInfo); } @Override @@ -225,13 +212,15 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag if (workModeGuard.tryAcquire()) { final StatisticsWorkMode targetWorkMode = input.getMode(); if (!workMode.equals(targetWorkMode)) { - shuttingDownStatisticsPolling = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode); + isStatisticsPollingEnabled = StatisticsWorkMode.FULLYDISABLED.equals(targetWorkMode); // iterate through stats-ctx: propagate mode - for (final StatisticsContext statisticsContext : contexts.values()) { - final DeviceContext deviceContext = statisticsContext.getDeviceContext(); + for (Map.Entry entry : contexts.entrySet()) { + final DeviceInfo deviceInfo = entry.getKey(); + final StatisticsContext statisticsContext = entry.getValue(); + final DeviceContext deviceContext = statisticsContext.getLifecycleService().getDeviceContext(); switch (targetWorkMode) { case COLLECTALL: - scheduleNextPolling(deviceContext, statisticsContext, new TimeCounter()); + scheduleNextPolling(deviceContext.getDeviceState(), deviceInfo, statisticsContext, new TimeCounter()); for (final ItemLifeCycleSource lifeCycleSource : deviceContext.getItemLifeCycleSourceRegistry().getLifeCycleSources()) { lifeCycleSource.setItemLifecycleListener(null); } @@ -262,43 +251,38 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag } @Override - public void startScheduling(final NodeId nodeId) { - if (shuttingDownStatisticsPolling) { - LOG.info("Statistics are shut down for device: {}", nodeId); + public void startScheduling(final DeviceInfo deviceInfo) { + if (isStatisticsPollingEnabled) { + LOG.info("Statistics are shut down for device: {}", deviceInfo.getNodeId()); return; } - final StatisticsContext statisticsContext = contexts.get(nodeId); + final StatisticsContext statisticsContext = contexts.get(deviceInfo); if (statisticsContext == null) { - LOG.warn("Statistics context not found for device: {}", nodeId); + LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId()); return; } if (statisticsContext.isSchedulingEnabled()) { - LOG.debug("Statistics scheduling is already enabled for device: {}", nodeId); + LOG.debug("Statistics scheduling is already enabled for device: {}", deviceInfo.getNodeId()); return; } - LOG.info("Scheduling statistics poll for device: {}", nodeId); - final DeviceContext deviceContext = conductor.getDeviceContext(nodeId); - - if (deviceContext == null) { - LOG.warn("Device context not found for device: {}", nodeId); - return; - } + LOG.info("Scheduling statistics poll for device: {}", deviceInfo.getNodeId()); statisticsContext.setSchedulingEnabled(true); - scheduleNextPolling(deviceContext, statisticsContext, new TimeCounter()); + final DeviceState deviceState = contexts.get(deviceInfo).getLifecycleService().getDeviceContext().getDeviceState(); + scheduleNextPolling(deviceState, deviceInfo, statisticsContext, new TimeCounter()); } @Override - public void stopScheduling(final NodeId nodeId) { - LOG.debug("Stopping statistics scheduling for device: {}", nodeId); - final StatisticsContext statisticsContext = contexts.get(nodeId); + public void stopScheduling(final DeviceInfo deviceInfo) { + LOG.debug("Stopping statistics scheduling for device: {}", deviceInfo.getNodeId()); + final StatisticsContext statisticsContext = contexts.get(deviceInfo); if (statisticsContext == null) { - LOG.warn("Statistics context not found for device: {}", nodeId); + LOG.warn("Statistics context not found for device: {}", deviceInfo.getNodeId()); return; } @@ -321,4 +305,5 @@ public class StatisticsManagerImpl implements StatisticsManager, StatisticsManag public void setDeviceTerminationPhaseHandler(final DeviceTerminationPhaseHandler handler) { this.deviceTerminPhaseHandler = handler; } + }