uni/inf2440/hw3/MultiRadixPar.java

import java.util.*;
import java.util.concurrent.CyclicBarrier;
/***********************************************************
 * Oblig 3 - sekvensiell kode, INF2440 v2017.
 *            Ifi, Uio, Arne Maus
 * for store verdier av n > 100 m, kjør (f.eks):
 *     >java -Xmx16000m MultiRadix 1000000000
 ************************************************************/
class MultiRadixPar implements Sorter {
	int n;
	int [] a;
	final static int NUM_BIT = 7; // alle tall 6-11 .. finn ut hvilken verdi som er best
	int nThreads = Math.min(
		Runtime.getRuntime().availableProcessors(),
		16);

	public int[] sort(int[] arr) {
		return radixMulti(arr);
	}

	class PartAWorker implements Runnable {
		int[] a;
		int start;
		int end;
		int biggest;

		PartAWorker(int[] a, int start, int end) {
			this.a = a;
			this.start = start;
			this.end = end;
			this.biggest = a[start];
		}

		public void run() {
			for (int i = start + 1; i < end; ++i) {
				if (a[i] > biggest)
					biggest = a[i];
			}
		}
	}

	int [] radixMulti(int [] a) {
		long tt = System.nanoTime();
		// 1-5 digit radixSort of : a[]
		int max = a[0], numBit = 2, numDigits, n =a.length;
		int [] bit ;

		// a) finn max verdi i a[]
		int d = a.length / nThreads;
		Thread[] threads = new Thread[nThreads];
		PartAWorker[] workers = new PartAWorker[nThreads];
		for (int i = 0; i < nThreads; ++i) {
			PartAWorker w = workers[i] = new PartAWorker(
				a, d * i, i == nThreads - 1 ? a.length : d * (i + 1));

			Thread t = threads[i] = new Thread(w);
			t.start();
		}
		for (Thread t: threads) { try { t.join(); } catch (InterruptedException ex) {} }
		max = workers[0].biggest;
		for (int i = 1; i < nThreads; ++i) {
			if (workers[i].biggest > max)
				max = workers[i].biggest;
		}
		while (max >= (1L<<numBit) )numBit++; // antall binaere siffer i max

		// bestem antall bit i numBits sifre
		numDigits = Math.max(1, numBit/NUM_BIT);
		bit = new int[numDigits];
		int rest = (numBit%numDigits), sum =0;;

		// fordel bitene vi skal sortere paa jevnt
		for (int i = 0; i < bit.length; i++){
			bit[i] = numBit/numDigits;
			if ( rest-- > 0)  bit[i]++;
		}

		int[] t=a, b = new int [n];

		for (int i =0; i < bit.length; i++) {
			radixSort( a,b,bit[i],sum );    // i-te siffer fra a[] til b[]
			sum += bit[i];
			// swap arrays (pointers only)
			t = a;
			a = b;
			b = t;
		}
		if (bit.length%2 != 0 ) {
			// et odde antall sifre, kopier innhold tilbake til original a[] (nå b)
			System.arraycopy (a,0,b,0,a.length);
		}

		return a;
	} // end radixMulti

	class Worker extends Thread {
		int index;
		CyclicBarrier barrier;
		int[] a;
		int[] b;
		int[][] accumulated;
		int[] globalCount;
		int[][] count2d;
		int mask;
		int shift;
		int start;
		int end;
		boolean doD = false;

		int acumVal;
		int j;
		int[] count;

		Worker(
				int index, CyclicBarrier barrier,
				int[] a, int[] b, int[][] accumulated,
				int[] globalCount, int[][] count2d,
				int mask, int shift,
				int start, int end) {

			this.index = index;
			this.barrier = barrier;
			this.a = a;
			this.b = b;
			this.accumulated = accumulated;
			this.globalCount = globalCount;
			this.count2d = count2d;
			this.mask = mask;
			this.shift = shift;
			this.start = start;
			this.end = end;

			acumVal = 0;
			j = 0;
			count = new int[mask+1];
		}

		synchronized public void run() {
			// b) count=the frequency of each radix value in a
			for (int i = start; i < end; i++) {
				count[(a[i]>>> shift) & mask]++;
			}

			// c) Add up in 'count' - accumulated values
			for (int i = 0; i <= mask; i++) {
				count2d[index][i] = count[i];
				j = count[i];
				accumulated[index][i] = acumVal;
				acumVal += j;
			}

			// Wait for synchronization
			try { barrier.await(); } catch (Exception ex) {}
			while (!doD)
				try { wait(); } catch (InterruptedException ex) {}

			// d) move numbers in sorted order a to b
			int[] space = new int[mask + 1];
			for (int i = 0; i < mask + 1; ++i) {
				for (int j = 0; j < index; ++j) {
					space[i] += count2d[j][i];
				}
			}
			for (int i = start; i < end; i++) {
				int idx = (a[i] >>> shift) & mask;
				b[globalCount[idx] + space[idx]++] = a[i];
			}
		}

		synchronized public void startPartD() {
			this.doD = true;
			notify();
		}
	}

	/** Sort a[] on one digit ; number of bits = maskLen, shiftet up 'shift' bits */
	void radixSort ( int [] a, int [] b, int maskLen, int shift){
		int mask = (1 << maskLen) - 1;

		CyclicBarrier barrier = new CyclicBarrier(nThreads + 1);
		int[][] accumulated = new int[nThreads][mask + 1];
		int[][] count2d = new int[nThreads][mask + 1];
		int[] count = new int[mask + 1];

		// Create and start workers
		Worker[] workers = new Worker[nThreads];
		int d = a.length / nThreads;
		for (int i = 0; i < nThreads; ++i) {
			int start = d * i;
			int end = i == nThreads - 1 ? a.length : d * (i + 1);
			workers[i] = new Worker(
					i, barrier,
					a, b, accumulated,
					count, count2d,
					mask, shift,
					start, end);
			workers[i].start();
		}

		// Synchronize
		try { barrier.await(); } catch (Exception ex) {}

		// Sequentially fix count
		for (int i=0; i < count.length; ++i) {
			for (int j = 0; j < nThreads; ++j) {
				count[i] += accumulated[j][i];
			}
		}

		// Do part D
		for (Worker w: workers) {
			w.startPartD();
		}

		// Join threads
		for (int i = 0; i < nThreads; ++i) {
			try { workers[i].join(); } catch (InterruptedException ex) {}
		}
	}// end radixSort

	void testSort(int [] a){
		for (int i = 0; i< a.length-1;i++) {
			if (a[i] > a[i+1]){
				System.out.println("SorteringsFEIL på plass: "+i +" a["+i+"]:"+a[i]+" > a["+(i+1)+"]:"+a[i+1]);
				return;
			}
		}
	}// end simple sorteingstest
}// end SekvensiellRadix