1 files changed, 59 insertions, 60 deletions

diff --git a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java
index be811a114..8074fe6fe 100644
--- a/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java
+++ b/cps-ncmp-service/src/main/java/org/onap/cps/ncmp/api/inventory/sync/ModuleSyncWatchdog.java
@@ -21,17 +21,16 @@
 
 package org.onap.cps.ncmp.api.inventory.sync;
 
+import java.util.Collection;
+import java.util.HashSet;
 import java.util.List;
-import java.util.concurrent.ConcurrentMap;
+import java.util.Map;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.TimeUnit;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.onap.cps.ncmp.api.impl.event.lcm.LcmEventsCmHandleStateHandler;
 import org.onap.cps.ncmp.api.impl.yangmodels.YangModelCmHandle;
-import org.onap.cps.ncmp.api.inventory.CmHandleState;
-import org.onap.cps.ncmp.api.inventory.CompositeState;
-import org.onap.cps.ncmp.api.inventory.InventoryPersistence;
-import org.onap.cps.ncmp.api.inventory.LockReasonCategory;
-import org.springframework.beans.factory.annotation.Qualifier;
+import org.onap.cps.spi.model.DataNode;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -40,75 +39,75 @@ import org.springframework.stereotype.Component;
 @Component
 public class ModuleSyncWatchdog {
 
-    private static final boolean MODEL_SYNC_IN_PROGRESS = false;
-    private static final boolean MODEL_SYNC_DONE = true;
-
-    private final InventoryPersistence inventoryPersistence;
-
     private final SyncUtils syncUtils;
+    private final BlockingQueue<DataNode> moduleSyncWorkQueue;
+    private final Map<String, Object> moduleSyncStartedOnCmHandles;
+    private final ModuleSyncTasks moduleSyncTasks;
 
-    private final ModuleSyncService moduleSyncService;
-
-    @Qualifier("moduleSyncSemaphores")
-    private final ConcurrentMap<String, Boolean> moduleSyncSemaphores;
-
-    private final LcmEventsCmHandleStateHandler lcmEventsCmHandleStateHandler;
+    private static final int MODULE_SYNC_BATCH_SIZE = 100;
+    private static final long PREVENT_CPU_BURN_WAIT_TIME_MILLIS = 10;
+    private static final String VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP = "Started";
 
     /**
      * Execute Cm Handle poll which changes the cm handle state from 'ADVISED' to 'READY'.
      */
-    @Scheduled(fixedDelayString = "${timers.advised-modules-sync.sleep-time-ms:30000}")
-    public void executeAdvisedCmHandlePoll() {
-        syncUtils.getAdvisedCmHandles().forEach(advisedCmHandle -> {
-            final String cmHandleId = advisedCmHandle.getId();
-            if (hasPushedIntoSemaphoreMap(cmHandleId)) {
-                log.debug("executing module sync on {}", cmHandleId);
-                final CompositeState compositeState = inventoryPersistence.getCmHandleState(cmHandleId);
-                try {
-                    moduleSyncService.deleteSchemaSetIfExists(advisedCmHandle);
-                    moduleSyncService.syncAndCreateSchemaSetAndAnchor(advisedCmHandle);
-                    lcmEventsCmHandleStateHandler.updateCmHandleState(advisedCmHandle, CmHandleState.READY);
-                    updateModuleSyncSemaphoreMap(cmHandleId);
-                } catch (final Exception e) {
-                    syncUtils.updateLockReasonDetailsAndAttempts(compositeState,
-                            LockReasonCategory.LOCKED_MODULE_SYNC_FAILED, e.getMessage());
-                    setCmHandleStateLocked(advisedCmHandle, compositeState.getLockReason());
-                }
-                log.debug("{} is now in {} state", cmHandleId, compositeState.getCmHandleState().name());
-            } else {
-                log.debug("{} already processed by another instance", cmHandleId);
-            }
-        });
-        log.debug("No Cm-Handles currently found in an ADVISED state");
+    @Scheduled(fixedDelayString = "${timers.advised-modules-sync.sleep-time-ms:5000}")
+    public void moduleSyncAdvisedCmHandles() {
+        populateWorkQueueIfNeeded();
+        while (!moduleSyncWorkQueue.isEmpty()) {
+            final Collection<DataNode> nextBatch = prepareNextBatch();
+            moduleSyncTasks.performModuleSync(nextBatch);
+            preventBusyWait();
+        }
     }
 
     /**
-     * Execute Cm Handle poll which changes the cm handle state from 'LOCKED' to 'ADVISED'.
+     * Find any failed (locked) cm handles and change state back to 'ADVISED'.
      */
     @Scheduled(fixedDelayString = "${timers.locked-modules-sync.sleep-time-ms:300000}")
-    public void executeLockedCmHandlePoll() {
-        final List<YangModelCmHandle> lockedCmHandles = syncUtils.getModuleSyncFailedCmHandles();
-        for (final YangModelCmHandle lockedCmHandle : lockedCmHandles) {
-            final CompositeState compositeState = lockedCmHandle.getCompositeState();
-            final boolean isReadyForRetry = syncUtils.isReadyForRetry(compositeState);
-            if (isReadyForRetry) {
-                log.debug("Reset cm handle {} state to ADVISED to re-attempt module-sync", lockedCmHandle.getId());
-                lcmEventsCmHandleStateHandler.updateCmHandleState(lockedCmHandle, CmHandleState.ADVISED);
-            }
-        }
+    public void resetPreviouslyFailedCmHandles() {
+        final List<YangModelCmHandle> failedCmHandles = syncUtils.getModuleSyncFailedCmHandles();
+        moduleSyncTasks.resetFailedCmHandles(failedCmHandles);
     }
 
-    private void setCmHandleStateLocked(final YangModelCmHandle advisedCmHandle,
-            final CompositeState.LockReason lockReason) {
-        advisedCmHandle.getCompositeState().setLockReason(lockReason);
-        lcmEventsCmHandleStateHandler.updateCmHandleState(advisedCmHandle, CmHandleState.LOCKED);
+    private void preventBusyWait() {
+        // This method isn't really needed until CPS-1200 Performance Improvement: Watchdog Parallel execution
+        // but leaving here to minimize impacts on this class for that Jira
+        try {
+            TimeUnit.MILLISECONDS.sleep(PREVENT_CPU_BURN_WAIT_TIME_MILLIS);
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
     }
 
-    private void updateModuleSyncSemaphoreMap(final String cmHandleId) {
-        moduleSyncSemaphores.replace(cmHandleId, MODEL_SYNC_DONE);
+    private void populateWorkQueueIfNeeded() {
+        if (moduleSyncWorkQueue.isEmpty()) {
+            final List<DataNode> advisedCmHandles = syncUtils.getAdvisedCmHandles();
+            for (final DataNode advisedCmHandle : advisedCmHandles) {
+                if (!moduleSyncWorkQueue.offer(advisedCmHandle)) {
+                    log.warn("Unable to add cm handle {} to the work queue", advisedCmHandle.getLeaves().get("id"));
+                }
+            }
+        }
     }
 
-    private boolean hasPushedIntoSemaphoreMap(final String cmHandleId) {
-        return moduleSyncSemaphores.putIfAbsent(cmHandleId, MODEL_SYNC_IN_PROGRESS) == null;
+    private Collection<DataNode> prepareNextBatch() {
+        final Collection<DataNode> nextBatchCandidates = new HashSet<>(MODULE_SYNC_BATCH_SIZE);
+        final Collection<DataNode> nextBatch = new HashSet<>(MODULE_SYNC_BATCH_SIZE);
+        moduleSyncWorkQueue.drainTo(nextBatchCandidates, MODULE_SYNC_BATCH_SIZE);
+        log.debug("nextBatchCandidates size : {}", nextBatchCandidates.size());
+        for (final DataNode batchCandidate : nextBatchCandidates) {
+            final String cmHandleId = String.valueOf(batchCandidate.getLeaves().get("id"));
+            final boolean alreadyAddedToInProgressMap = VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP
+                .equals(moduleSyncStartedOnCmHandles.putIfAbsent(cmHandleId, VALUE_FOR_HAZELCAST_IN_PROGRESS_MAP));
+            if (alreadyAddedToInProgressMap) {
+                log.debug("module sync for {} already in progress by other instance", cmHandleId);
+            } else {
+                nextBatch.add(batchCandidate);
+            }
+        }
+        log.debug("nextBatch size : {}", nextBatch.size());
+        return nextBatch;
     }
+
 }