package parallel;

import util.Prime;
import java.util.List;
import util.PrimeCounter;
import java.util.ArrayList;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;

public class CallablePrimeCounter implements PrimeCounter {

	private final int numThreads;

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

	public int countPrimes(int[] nums) throws Exception {
		int count = 0;
		List<FutureTask<Integer>> futureTasks = new ArrayList<>(numThreads);
		int size = (int) Math.ceil((double) nums.length / numThreads);
		for (int i = 0; i < numThreads; i++) {
			FutureTask<Integer> future = 
					new FutureTask<>(new CountPrimes(nums, i * size, (i + 1) * size));
			futureTasks.add(future);
			new Thread(future).start();
		}
		for (FutureTask<Integer> task : futureTasks) {
			count += task.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;
		}
	}
}