並列和、並列プリフィクススキャン

ConcurrencyTest.java

public class ConcurrencyTest {

	public static void main(String[] args) {
		System.out.println("available processors "+ Runtime.getRuntime().availableProcessors());
		
		int[] a = new int[8000000];
		for(int i = 0 ; i < 800000 ; i++){
			a[i] =  (int) (10000 * Math.random());
		}
		PrefixScan s = new PrefixScan(a);
		s.serialInclusiveScan();
		s.parallelInclusiveScan();
		a = null;
		s = null;
		
		long[] b = new long[80000000];
		for(int i = 0 ; i < 80000000 ; i++){
			b[i] =  (long) (10 * Math.random());
		}
		Reduction r = new Reduction(b);
		System.out.println(r.parallelSummation());
		System.out.println(r.serialSummation());
	}

}

PrefixScan.java

import java.util.concurrent.*;

public class PrefixScan {
	private static final int NUM_THREADS = 4;
	private int[] a;
	private final int len;
	private long[] globalSum = new long[NUM_THREADS];
	private ExecutorService pool = Executors.newFixedThreadPool(NUM_THREADS);
	private final CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS + 1);
	private int[] partialSum = new int [NUM_THREADS];
	private int[] prefixSum  = new int [NUM_THREADS];
	
	public PrefixScan(int[] a){
		this.a = a.clone();
		len = a.length;
	}
	
	//参考
	//並行コンピューティング技法　Clay Breshears著　オライリージャパン　ISBN 978-4-87311-435-4
	//例6-8
	public int[] parallelInclusiveScan(){
		System.out.println("parallel calculation start");
		long pre = System.currentTimeMillis();
		
		for(int i = 0 ; i < NUM_THREADS ; i++){
			final int index = i;
			Runnable r = new Runnable() {
				@Override
				public void run() {
					int start = (int) (((float) len / NUM_THREADS) * index);
					int end   = (int) (((float) len / NUM_THREADS) * (index + 1));
					if(index == (NUM_THREADS - 1)) end = len;
					for(int i = start + 1 ; i < end ; i++){
						a[i] = a[i - 1] + a[i];
					}
					partialSum[index] = a[end - 1];
					
					try{
						barrier.await();
					}catch(Exception ex){
						ex.printStackTrace();
					}
				}
			};
			
			pool.execute(r);
		}
		
		try{
			barrier.await();
		}catch(Exception ex){
			ex.printStackTrace();;
		}
		
		prefixSum[0] = 0;
		for(int i = 1 ; i < NUM_THREADS ; i++)
			prefixSum[i] = prefixSum[i - 1] + partialSum[i - 1];
		
		for(int i = 0 ; i < NUM_THREADS ; i++){
			final int index = i;
			Runnable r = new Runnable() {
				@Override
				public void run() {
					int start = (int) (((float) len / NUM_THREADS) * index);
					int end   = (int) (((float) len / NUM_THREADS) * (index + 1));
					if(index == (NUM_THREADS - 1)) end = len;
					for(int i = start ; i < end ; i++){
						a[i] += prefixSum[index];
					}
					
					try{
						barrier.await();
					}catch(Exception ex){
						ex.printStackTrace();
					}
				}
			};
			
			pool.execute(r);
		}
		
		try{
			barrier.await();
		}catch(Exception ex){
			ex.printStackTrace();;
		}
		
		System.out.println("it took "+ (System.currentTimeMillis() - pre) +" ms");
		pool.shutdown();
		
		return a;
	}
	
	public int[] serialInclusiveScan(){
		System.out.println("serial calculation start");
		long pre = System.currentTimeMillis();
		int[] scan = new int[len];
		scan[0] = a[0];
		for(int i = 1 ; i < len ; i++)
			scan[i] = scan[i - 1] + a[i];
		System.out.println("it took "+ (System.currentTimeMillis() - pre) +" ms");
		return scan;
	}
	
	public int[] serialExclusiveScan(){
		int[] scan = new int[len];
		scan[0] = 0;
		for(int i = 1 ; i < len ; i++)
			scan[i] = scan[i - 1] + a[i - 1];
		return scan;
	}
}

Reduction.java

import java.util.concurrent.*;

public class Reduction {
	private static final int NUM_THREADS = 4;
	private final long[] a;
	private final int len;
	private long[] globalSum = new long[NUM_THREADS];
	private ExecutorService pool = Executors.newFixedThreadPool(NUM_THREADS);
	private final CyclicBarrier barrier = new CyclicBarrier(NUM_THREADS + 1);
	private final CyclicBarrier threadBarrier = new CyclicBarrier(NUM_THREADS);
	
	public Reduction(final long[] a){
		this.a = a;
		len = a.length;
	}
	
	//参考
	//並行コンピューティング技法　Clay Breshears著　オライリージャパン　ISBN 978-4-87311-435-4
	//例7-2
	public long parallelSummation(){
		System.out.println("parallel calculation start");
		long pre = System.currentTimeMillis();
		for(int i = 0 ; i < NUM_THREADS ; i ++){
			final int index = i;
			Runnable r = new Runnable() {
				@Override
				public void run(){
					int start, end;
					long partialSum = 0;
					start = (int) ((float)len/NUM_THREADS) * index;
					end = (int) ((float)len/NUM_THREADS) * (index + 1);
					for(int i = start ; i < end ; i++){
						partialSum += a[i];
					}
					globalSum[index] = partialSum;
					try{
						barrier.await();
					}catch(Exception ex){
						ex.printStackTrace();
					}
				}
			};
			pool.execute(r);
		}
		
		try{
			barrier.await();
		}catch(Exception ex){
			ex.printStackTrace();
		}

		for(int i = 0 ; i < NUM_THREADS ; i++){
			final int index = i;
			Runnable r = new Runnable() {
				@Override
				public void run() {
					int p2 = 2;
					for(int j = 1 ; j <= NUM_THREADS ; j *= 2){
						if(index % p2 == 0)
							if(index + j < NUM_THREADS)
								globalSum[index] += globalSum[index + j];
						try{
							threadBarrier.await();
						}catch(Exception ex){
							ex.printStackTrace();
						}
					}
					try{
						barrier.await();
					}catch(Exception ex){
						ex.printStackTrace();
					}
				}
			};
			pool.execute(r);
		}
		
		try{
			barrier.await();
		}catch(Exception ex){
			ex.printStackTrace();
		}
		System.out.println("it took "+ (System.currentTimeMillis() - pre) +" ms");
		pool.shutdown();
		return globalSum[0];
	}
	
	public long serialSummation(){
		System.out.println("serial calculation start");
		long pre = System.currentTimeMillis();
		long sum = 0;
		for(int i = 0 ; i < len ; i++){
			sum += a[i];
		}
		System.out.println("it took "+ (System.currentTimeMillis() - pre) +" ms");
		return sum;
	}
}