From ddef41c97ffd530c6fb6e5b457e08a91a38270a4 Mon Sep 17 00:00:00 2001
From: Andrey Kurilov <andrey.kurilov@emc.com>
Date: Mon, 22 Aug 2016 14:25:42 +0300
Subject: [PATCH] intermediate statistic feature fix

---
 .../impl/load/executor/LoadExecutorBase.java    | 17 ++++++++---------
 1 file changed, 8 insertions(+), 9 deletions(-)

diff --git a/src/main/java/com/emc/mongoose/core/impl/load/executor/LoadExecutorBase.java b/src/main/java/com/emc/mongoose/core/impl/load/executor/LoadExecutorBase.java
index 4bc2fca5b3..40f54b7d7a 100644
--- a/src/main/java/com/emc/mongoose/core/impl/load/executor/LoadExecutorBase.java
+++ b/src/main/java/com/emc/mongoose/core/impl/load/executor/LoadExecutorBase.java
@@ -100,7 +100,7 @@ public abstract class LoadExecutorBase<T extends Item>
 		counterSubm = new AtomicLong(0),
 		countRej = new AtomicLong(0),
 		counterResults = new AtomicLong(0);
-	private final AtomicInteger currConcurrencyLevel = new AtomicInteger(0);
+	private final AtomicInteger currConcurrency = new AtomicInteger(0);
 	protected T lastItem;
 	protected final Object state = new Object();
 	protected final ItemBuffer<T> itemOutBuff;
@@ -360,13 +360,12 @@ protected IoStats createIntermediateStats() {
 	//
 	@Override
 	public boolean isFullThrottleEntered() {
-		return totalThreadCount - currConcurrencyLevel.get() < totalThreadCount * fullLoadThreshold;
+		return currConcurrency.get() > totalThreadCount * fullLoadThreshold;
 	}
 	//
 	@Override
 	public boolean isFullThrottleExited() {
-		return isShutdown.get() &&
-			totalThreadCount - currConcurrencyLevel.get() > totalThreadCount * fullLoadThreshold;
+		return isShutdown.get() && currConcurrency.get() < totalThreadCount * fullLoadThreshold;
 	}
 	////////////////////////////////////////////////////////////////////////////////////////////////
 	protected LoadExecutorBase(
@@ -813,7 +812,7 @@ protected void logTrace(
 	}
 	//
 	protected final void incrementBusyThreadCount() {
-		currConcurrencyLevel.incrementAndGet();
+		currConcurrency.incrementAndGet();
 	}
 	//
 	@Override
@@ -843,7 +842,7 @@ public void ioTaskCompleted(final IoTask<T> ioTask)
 		if(nodeBalancer != null) {
 			nodeBalancer.markTaskFinish(nodeAddr);
 		}
-		currConcurrencyLevel.decrementAndGet();
+		currConcurrency.decrementAndGet();
 		//
 		if(IoTask.Status.SUCC == status) {
 			// update the metrics with success
@@ -894,7 +893,7 @@ public int ioTaskCompletedBatch(
 		//
 		final int n = to - from;
 		if(n > 0) {
-			currConcurrencyLevel.addAndGet(-n);
+			currConcurrency.addAndGet(-n);
 			if(storageNodeAddrs != null) {
 				final String nodeAddr = ioTasks.get(from).getNodeAddr();
 				nodeBalancer.markTasksFinish(nodeAddr, n);
@@ -970,7 +969,7 @@ public int ioTaskCompletedBatch(
 	//
 	protected void ioTaskCancelled(final int n) {
 		LOG.debug(Markers.MSG, "{}: I/O task canceled", hashCode());
-		currConcurrencyLevel.decrementAndGet();
+		currConcurrency.decrementAndGet();
 		ioStats.markFail(n);
 		if(medIoStats != null && medIoStats.isStarted()) {
 			medIoStats.markFail(n);
@@ -979,7 +978,7 @@ protected void ioTaskCancelled(final int n) {
 	}
 	//
 	protected void ioTaskFailed(final int n, final Throwable e) {
-		currConcurrencyLevel.decrementAndGet();
+		currConcurrency.decrementAndGet();
 		ioStats.markFail(n);
 		if(medIoStats != null && medIoStats.isStarted()) {
 			medIoStats.markFail();