Replaced my custom concurrent counter with an AtomicInteger

This commit is contained in:
Lennart Hensler committed 2015-03-19 21:58:22 +01:00
1 parent 33a8a1d71c
commit 2cebb082f9
1 file changed
+6 -44
@@ -6,8 +6,7 @@ import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Level;
import java.util.logging.Logger;
@@ -22,7 +21,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
private final Set<Processor<ResultType, ?>> resultReceivers;
private final ExecutorService executor;
private final UnfinishedInstructionsCounter unfinishedInstructionsCounter;
private final AtomicInteger unfinishedInstructionsCounter;
private boolean isFinished = false;
private boolean gotAborted = false;
@@ -31,7 +30,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
super(inputType, resultType);
this.executor = executor;
this.resultReceivers = new HashSet<Processor<ResultType, ?>>(resultReceivers);
unfinishedInstructionsCounter = new UnfinishedInstructionsCounter();
unfinishedInstructionsCounter = new AtomicInteger();
}
@Override
@@ -53,11 +52,11 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
onFailure(e);
}
} finally {
AbstractPartitioningParallelProcessor.this.unfinishedInstructionsCounter.decrement();
AbstractPartitioningParallelProcessor.this.unfinishedInstructionsCounter.getAndDecrement();
}
}
};
unfinishedInstructionsCounter.increment();
unfinishedInstructionsCounter.getAndIncrement();
try {
executor.execute(instructionWrapper);
} catch (RejectedExecutionException exc){
@@ -124,7 +123,7 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
}
private boolean areUnfinishedInstructionsLeft() {
return unfinishedInstructionsCounter.getUnfinishedInstructionsAmount() > 0;
return unfinishedInstructionsCounter.get() > 0;
}
protected void tellResultReceiversToFinish() {
@@ -161,41 +160,4 @@ public abstract class AbstractPartitioningParallelProcessor<InputType, WorkingTy
protected abstract void setAdditionalData(AdditionalResultDataBuilder additionalDataBuilder);
/**
* Thread safe class to manage, if there are unfinished instructions.
*/
private class UnfinishedInstructionsCounter {
private final Lock instructionsAmountLock;
private int unfinishedInstructionsAmount;
public UnfinishedInstructionsCounter() {
instructionsAmountLock = new ReentrantLock();
}
public void increment() {
instructionsAmountLock.lock();
try {
unfinishedInstructionsAmount++;
} finally {
instructionsAmountLock.unlock();
}
}
public void decrement() {
instructionsAmountLock.lock();
try {
unfinishedInstructionsAmount--;
unfinishedInstructionsAmount = Math.max(0, unfinishedInstructionsAmount);
} finally {
instructionsAmountLock.unlock();
}
}
public int getUnfinishedInstructionsAmount() {
return unfinishedInstructionsAmount;
}
}
}