test/extra166y/LongMapReduceDemo.java

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain, as explained at
 * http://creativecommons.org/licenses/publicdomain
 */

import jsr166y.*;
import extra166y.*;
import static extra166y.Ops.*;
import java.util.*;
import java.util.concurrent.*;

public class LongMapReduceDemo {
    static final int NCPU = Runtime.getRuntime().availableProcessors();
    static final Random rng = new Random();

    /**
     * Sequential version, for performance comparison
     */
    static long seqMapReduce(long[] array, 
                            LongOp mapper,
                            LongReducer reducer,
                            long base) {
        long n = array.length;
        long x = base;
        for (int i = 0; i < n; ++i) 
            x = reducer.op(x, mapper.op(array[i]));
        return x;
    }

    // sample functions
    static final class GetNext implements LongOp {
        public long op(long seed) {
            long x = seed;
            x ^= x << 13; 
            x ^= x >>> 7; 
            x ^= (x << 17);
            return x;
        }
    }

    static final class Accum implements LongReducer {
        public long op(long x, long y) {
            return x + y;
        }
    }

    /** for time conversion */
    static final long NPS = (1000L * 1000 * 1000);

    public static void main(String[] args) throws Exception {
        int n = 1 << 18;
        int reps = 1 << 8;
        long[] array = new long[n];
        for (int i = 0; i < n; ++i) 
            array[i] = i + 1L;
        ForkJoinPool fjp = new ForkJoinPool(1);
        ParallelLongArray pa = ParallelLongArray.createUsingHandoff(array, fjp);
        final GetNext getNext = new GetNext();
        final Accum accum = new Accum();
        final long zero = 0L;
        long last, now;
        double elapsed;
        for (int j = 0; j < 2; ++j) {
            long rseed = rng.nextLong();
            resetSeeds(array, rseed);
            long seqsum = 0;
            last = System.nanoTime();
            for (int k = 0; k < reps; ++k) {
                seqsum += seqMapReduce(array, getNext, accum, zero);
                long tmp = array[k];
                array[k] = array[n - k - 1];
                array[n - k - 1] = tmp;
            }
            now = System.nanoTime();
            elapsed = (double)(now - last) / NPS;
            last = now;
            System.out.printf("sequential:    %7.3f\n", elapsed);
            for (int i = 2; i <= NCPU; i <<= 1) {
                resetSeeds(array, rseed);
                long sum = 0;
                fjp.setParallelism(i);
                last = System.nanoTime();
                for (int k = 0; k < reps; ++k) {
                    sum += pa.withMapping(getNext).reduce(accum, zero);
                    long tmp = array[k];
                    array[k] = array[n - k - 1];
                    array[n - k - 1] = tmp;
                }
                now = System.nanoTime();
                elapsed = (double)(now - last) / NPS;
                last = now;
                System.out.printf("poolSize %3d:  %7.3f\n", fjp.getParallelism(), elapsed);
                if (sum != seqsum) throw new Error("checksum");
            }
            for (int i = NCPU; i >= 1; i >>>= 1) {
                resetSeeds(array, rseed);
                long sum = 0;
                fjp.setParallelism(i);
                last = System.nanoTime();
                for (int k = 0; k < reps; ++k) {
                    sum += pa.withMapping(getNext).reduce(accum, zero);
                    long tmp = array[k];
                    array[k] = array[n - k - 1];
                    array[n - k - 1] = tmp;
                }
                now = System.nanoTime();
                elapsed = (double)(now - last) / NPS;
                last = now;
                System.out.printf("poolSize %3d:  %7.3f\n", fjp.getParallelism(), elapsed);
                if (sum != seqsum) throw new Error("checksum");
            }
        }
        fjp.shutdownNow();
        fjp.awaitTermination(1, TimeUnit.SECONDS);
        Thread.sleep(100);
    }

    static void resetSeeds(long[] array, long s) {
        for (int i = 0; i < array.length; ++i)
            array[i] = s++;
    }
    
}
Revision:	1.1
Committed:	Sun Nov 1 22:00:35 2009 UTC (14 years, 7 months ago) by dl
Branch:	MAIN
Log Message:	Move tests for extra166y
#	Content
1	/*
2	* Written by Doug Lea with assistance from members of JCP JSR-166
3	* Expert Group and released to the public domain, as explained at
4	* http://creativecommons.org/licenses/publicdomain
5	*/
6
7	import jsr166y.*;
8	import extra166y.*;
9	import static extra166y.Ops.*;
10	import java.util.*;
11	import java.util.concurrent.*;
12
13	public class LongMapReduceDemo {
14	static final int NCPU = Runtime.getRuntime().availableProcessors();
15	static final Random rng = new Random();
16
17	/**
18	* Sequential version, for performance comparison
19	*/
20	static long seqMapReduce(long[] array,
21	LongOp mapper,
22	LongReducer reducer,
23	long base) {
24	long n = array.length;
25	long x = base;
26	for (int i = 0; i < n; ++i)
27	x = reducer.op(x, mapper.op(array[i]));
28	return x;
29	}
30
31	// sample functions
32	static final class GetNext implements LongOp {
33	public long op(long seed) {
34	long x = seed;
35	x ^= x << 13;
36	x ^= x >>> 7;
37	x ^= (x << 17);
38	return x;
39	}
40	}
41
42	static final class Accum implements LongReducer {
43	public long op(long x, long y) {
44	return x + y;
45	}
46	}
47
48	/** for time conversion */
49	static final long NPS = (1000L * 1000 * 1000);
50
51	public static void main(String[] args) throws Exception {
52	int n = 1 << 18;
53	int reps = 1 << 8;
54	long[] array = new long[n];
55	for (int i = 0; i < n; ++i)
56	array[i] = i + 1L;
57	ForkJoinPool fjp = new ForkJoinPool(1);
58	ParallelLongArray pa = ParallelLongArray.createUsingHandoff(array, fjp);
59	final GetNext getNext = new GetNext();
60	final Accum accum = new Accum();
61	final long zero = 0L;
62	long last, now;
63	double elapsed;
64	for (int j = 0; j < 2; ++j) {
65	long rseed = rng.nextLong();
66	resetSeeds(array, rseed);
67	long seqsum = 0;
68	last = System.nanoTime();
69	for (int k = 0; k < reps; ++k) {
70	seqsum += seqMapReduce(array, getNext, accum, zero);
71	long tmp = array[k];
72	array[k] = array[n - k - 1];
73	array[n - k - 1] = tmp;
74	}
75	now = System.nanoTime();
76	elapsed = (double)(now - last) / NPS;
77	last = now;
78	System.out.printf("sequential: %7.3f\n", elapsed);
79	for (int i = 2; i <= NCPU; i <<= 1) {
80	resetSeeds(array, rseed);
81	long sum = 0;
82	fjp.setParallelism(i);
83	last = System.nanoTime();
84	for (int k = 0; k < reps; ++k) {
85	sum += pa.withMapping(getNext).reduce(accum, zero);
86	long tmp = array[k];
87	array[k] = array[n - k - 1];
88	array[n - k - 1] = tmp;
89	}
90	now = System.nanoTime();
91	elapsed = (double)(now - last) / NPS;
92	last = now;
93	System.out.printf("poolSize %3d: %7.3f\n", fjp.getParallelism(), elapsed);
94	if (sum != seqsum) throw new Error("checksum");
95	}
96	for (int i = NCPU; i >= 1; i >>>= 1) {
97	resetSeeds(array, rseed);
98	long sum = 0;
99	fjp.setParallelism(i);
100	last = System.nanoTime();
101	for (int k = 0; k < reps; ++k) {
102	sum += pa.withMapping(getNext).reduce(accum, zero);
103	long tmp = array[k];
104	array[k] = array[n - k - 1];
105	array[n - k - 1] = tmp;
106	}
107	now = System.nanoTime();
108	elapsed = (double)(now - last) / NPS;
109	last = now;
110	System.out.printf("poolSize %3d: %7.3f\n", fjp.getParallelism(), elapsed);
111	if (sum != seqsum) throw new Error("checksum");
112	}
113	}
114	fjp.shutdownNow();
115	fjp.awaitTermination(1, TimeUnit.SECONDS);
116	Thread.sleep(100);
117	}
118
119	static void resetSeeds(long[] array, long s) {
120	for (int i = 0; i < array.length; ++i)
121	array[i] = s++;
122	}
123
124	}