PF3/hw2/Ex3/src/parallel/ThreadPoolPrimeCounter.java

package parallel;

import util.Prime;
import java.util.List;
import util.PrimeCounter;
import java.util.ArrayList;
import java.util.concurrent.*;

public class ThreadPoolPrimeCounter implements PrimeCounter {

	private final int numThreads;

	public ThreadPoolPrimeCounter(int numThreads) {
		this.numThreads = numThreads;
	}

	public int countPrimes(int[] nums) throws Exception {
		final ExecutorService e = Executors.newFixedThreadPool(numThreads);
		final int step = (int) Math.ceil((double) nums.length / numThreads);
		final List<Future<Integer>> results = new ArrayList<>(numThreads);

		for (int i = 0; i < nums.length; i += step) {
			results.add(e.submit(new CountPrimes(nums, i, i + step)));
		}

		e.shutdown();

		int count = 0;
		for (Future<Integer> result : results) {
			count += result.get();
		}
		return count;
	}

	private static class CountPrimes implements Callable<Integer> {
		private final int[] nums;
		private final int low, high;

		private CountPrimes(int[] nums, int low, int high) {
			this.nums = nums;
			this.low = low;
			this.high = Math.min(high, nums.length);
		}

		@Override
		public Integer call() throws Exception {
			int count = 0;
			for (int i = low; i < high; i++) {
				if (Prime.isPrime(nums[i])) {
					count++;
				}
			}
			return count;
		}
	}
}