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/publicdomain/zero/1.0/
 */

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);
            x ^= x << 13;
            x ^= x >>> 7;
            x ^= (x << 17);
            x ^= x << 13;
            x ^= x >>> 7;
            x ^= (x << 17);
            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 << 10;
        long[] array = new long[n];
        for (int i = 0; i < n; ++i)
            array[i] = i + 1L;
        ForkJoinPool fjp = new ForkJoinPool();
        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 = NCPU; i >= 1; i >>>= 1) {
                resetSeeds(array, rseed);
                long sum = 0;
                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 = 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");
            }
        }
        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.4
Committed:	Tue Mar 15 19:47:04 2011 UTC (13 years, 1 month ago) by jsr166
Branch:	MAIN
CVS Tags:	release-1_7_0, HEAD
Changes since 1.3:	+1 -1 lines
Log Message:	Update Creative Commons license URL in legal notices
#	User	Rev	Content
1	dl	1.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	jsr166	1.4	* http://creativecommons.org/publicdomain/zero/1.0/
5	dl	1.1	*/
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	jsr166	1.2	static long seqMapReduce(long[] array,
21	dl	1.1	LongOp mapper,
22			LongReducer reducer,
23			long base) {
24			long n = array.length;
25			long x = base;
26	jsr166	1.2	for (int i = 0; i < n; ++i)
27	dl	1.1	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	jsr166	1.2	x ^= x << 13;
36			x ^= x >>> 7;
37	dl	1.1	x ^= (x << 17);
38	dl	1.3	x ^= x << 13;
39			x ^= x >>> 7;
40			x ^= (x << 17);
41			x ^= x << 13;
42			x ^= x >>> 7;
43			x ^= (x << 17);
44			x ^= x << 13;
45			x ^= x >>> 7;
46			x ^= (x << 17);
47	dl	1.1	return x;
48			}
49			}
50
51			static final class Accum implements LongReducer {
52			public long op(long x, long y) {
53			return x + y;
54			}
55			}
56
57			/** for time conversion */
58			static final long NPS = (1000L * 1000 * 1000);
59
60			public static void main(String[] args) throws Exception {
61			int n = 1 << 18;
62	dl	1.3	int reps = 1 << 10;
63	dl	1.1	long[] array = new long[n];
64	jsr166	1.2	for (int i = 0; i < n; ++i)
65	dl	1.1	array[i] = i + 1L;
66	dl	1.3	ForkJoinPool fjp = new ForkJoinPool();
67	dl	1.1	ParallelLongArray pa = ParallelLongArray.createUsingHandoff(array, fjp);
68			final GetNext getNext = new GetNext();
69			final Accum accum = new Accum();
70			final long zero = 0L;
71			long last, now;
72			double elapsed;
73			for (int j = 0; j < 2; ++j) {
74			long rseed = rng.nextLong();
75			resetSeeds(array, rseed);
76			long seqsum = 0;
77			last = System.nanoTime();
78			for (int k = 0; k < reps; ++k) {
79			seqsum += seqMapReduce(array, getNext, accum, zero);
80			long tmp = array[k];
81			array[k] = array[n - k - 1];
82			array[n - k - 1] = tmp;
83			}
84			now = System.nanoTime();
85			elapsed = (double)(now - last) / NPS;
86			last = now;
87			System.out.printf("sequential: %7.3f\n", elapsed);
88	dl	1.3	for (int i = NCPU; i >= 1; i >>>= 1) {
89	dl	1.1	resetSeeds(array, rseed);
90			long sum = 0;
91			last = System.nanoTime();
92			for (int k = 0; k < reps; ++k) {
93			sum += pa.withMapping(getNext).reduce(accum, zero);
94			long tmp = array[k];
95			array[k] = array[n - k - 1];
96			array[n - k - 1] = tmp;
97			}
98			now = System.nanoTime();
99			elapsed = (double)(now - last) / NPS;
100			last = now;
101			System.out.printf("poolSize %3d: %7.3f\n", fjp.getParallelism(), elapsed);
102			if (sum != seqsum) throw new Error("checksum");
103			}
104	dl	1.3	for (int i = 2; i <= NCPU; i <<= 1) {
105	dl	1.1	resetSeeds(array, rseed);
106			long sum = 0;
107	dl	1.3	// fjp.setParallelism(i);
108	dl	1.1	last = System.nanoTime();
109			for (int k = 0; k < reps; ++k) {
110			sum += pa.withMapping(getNext).reduce(accum, zero);
111			long tmp = array[k];
112			array[k] = array[n - k - 1];
113			array[n - k - 1] = tmp;
114			}
115			now = System.nanoTime();
116			elapsed = (double)(now - last) / NPS;
117			last = now;
118			System.out.printf("poolSize %3d: %7.3f\n", fjp.getParallelism(), elapsed);
119			if (sum != seqsum) throw new Error("checksum");
120			}
121			}
122			fjp.shutdownNow();
123			fjp.awaitTermination(1, TimeUnit.SECONDS);
124			Thread.sleep(100);
125			}
126
127			static void resetSeeds(long[] array, long s) {
128			for (int i = 0; i < array.length; ++i)
129			array[i] = s++;
130			}
131	jsr166	1.2
132	dl	1.1	}