OPNFLWPLUG-983 Group and flow removal stats are not reported in order

[openflowplugin.git] / openflowplugin-impl / src / main / java / org / opendaylight / openflowplugin / impl / statistics / StatisticsContextImpl.java

diff --git a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsContextImpl.java b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsContextImpl.java
index 9b246c4a709784530f70b90b0ff9de3fc468fdf7..724531b94f87c2a0028065a3e045e7337e376140 100644 (file)
--- a/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsContextImpl.java
+++ b/openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/statistics/StatisticsContextImpl.java
@@ -66,8 +66,8 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
     private final long maximumPollingDelay;
     private final boolean isUsingReconciliationFramework;
     private final AtomicBoolean schedulingEnabled = new AtomicBoolean(true);
-    private final AtomicReference<ListenableFuture<Boolean>> lastDataGathering = new AtomicReference<>();
-    private final AtomicReference<StatisticsPollingService> statisticsPollingService = new AtomicReference<>();
+    private final AtomicReference<ListenableFuture<Boolean>> lastDataGatheringRef = new AtomicReference<>();
+    private final AtomicReference<StatisticsPollingService> statisticsPollingServiceRef = new AtomicReference<>();
     private List<MultipartType> collectingStatType;
     private StatisticsGatheringService<T> statisticsGatheringService;
     private StatisticsGatheringOnTheFlyService<T> statisticsGatheringOnTheFlyService;
@@ -108,8 +108,8 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
     }
 
     @Override
-    public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher contextChainMastershipWatcher) {
-        this.contextChainMastershipWatcher = contextChainMastershipWatcher;
+    public void registerMastershipWatcher(@Nonnull final ContextChainMastershipWatcher newWatcher) {
+        this.contextChainMastershipWatcher = newWatcher;
     }
 
     @Override
@@ -155,10 +155,6 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
             statListForCollecting.add(MultipartType.OFPMPTABLE);
         }
 
-        if (devState.isFlowStatisticsAvailable()) {
-            statListForCollecting.add(MultipartType.OFPMPFLOW);
-        }
-
         if (devState.isGroupAvailable()) {
             statListForCollecting.add(MultipartType.OFPMPGROUPDESC);
             statListForCollecting.add(MultipartType.OFPMPGROUP);
@@ -169,6 +165,10 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
             statListForCollecting.add(MultipartType.OFPMPMETER);
         }
 
+        if (devState.isFlowStatisticsAvailable()) {
+            statListForCollecting.add(MultipartType.OFPMPFLOW);
+        }
+
         if (devState.isPortStatisticsAvailable()) {
             statListForCollecting.add(MultipartType.OFPMPPORTSTATS);
         }
@@ -209,7 +209,7 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
             return Futures.immediateFuture(Boolean.TRUE);
         }
 
-        return this.lastDataGathering.updateAndGet(future -> {
+        return this.lastDataGatheringRef.updateAndGet(future -> {
             // write start timestamp to state snapshot container
             StatisticsGatheringUtils.markDeviceStateSnapshotStart(deviceInfo, deviceContext);
 
@@ -221,7 +221,8 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
             // build statistics gathering future
             final ListenableFuture<Boolean> newDataGathering = collectingStatType.stream()
                     .reduce(lastDataGathering, this::statChainFuture,
-                        (listenableFuture, asyn) -> Futures.transformAsync(listenableFuture, result -> asyn));
+                        (listenableFuture, asyn) -> Futures.transformAsync(listenableFuture, result -> asyn,
+                                MoreExecutors.directExecutor()));
 
             // write end timestamp to state snapshot container
             Futures.addCallback(newDataGathering, new FutureCallback<Boolean>() {
@@ -266,7 +267,7 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
                                       getDeviceInfo(), multipartType, deviceContext, deviceContext, convertorExecutor,
                                       statisticsWriterProvider, executorService) : Futures
                     .immediateFuture(Boolean.FALSE);
-        });
+        }, MoreExecutors.directExecutor());
     }
 
     private void startGatheringData() {
@@ -283,19 +284,19 @@ class StatisticsContextImpl<T extends OfHeader> implements StatisticsContext {
 
         schedulingEnabled.set(true);
         statisticsPollingService.startAsync();
-        this.statisticsPollingService.set(statisticsPollingService);
+        this.statisticsPollingServiceRef.set(statisticsPollingService);
     }
 
     private ListenableFuture<Void> stopGatheringData() {
         LOG.info("Stopping running statistics gathering for node {}", deviceInfo);
         cancelLastDataGathering();
 
-        return Optional.ofNullable(statisticsPollingService.getAndSet(null)).map(StatisticsPollingService::stop)
+        return Optional.ofNullable(statisticsPollingServiceRef.getAndSet(null)).map(StatisticsPollingService::stop)
                 .orElseGet(() -> Futures.immediateFuture(null));
     }
 
     private void cancelLastDataGathering() {
-        final ListenableFuture<Boolean> future = lastDataGathering.getAndSet(null);
+        final ListenableFuture<Boolean> future = lastDataGatheringRef.getAndSet(null);
 
         if (Objects.nonNull(future) && !future.isDone() && !future.isCancelled()) {
             future.cancel(true);