Source code

001/*
002 * To change this template, choose Tools | Templates
003 * and open the template in the editor.
004 */
005
006package com.nativelibs4java.opencl.util;
007
008import org.bridj.Pointer;
009import static org.bridj.Pointer.*;
010
011import java.io.IOException;
012import java.util.Random;
013import java.util.concurrent.ExecutorService;
014import java.util.concurrent.Executors;
015import java.util.concurrent.TimeUnit;
016import java.util.logging.Level;
017import java.util.logging.Logger;
018
019import com.nativelibs4java.opencl.CLBuildException;
020import com.nativelibs4java.opencl.CLContext;
021import com.nativelibs4java.opencl.CLEvent;
022import com.nativelibs4java.opencl.CLBuffer;
023import com.nativelibs4java.opencl.JavaCL;
024import com.nativelibs4java.opencl.CLQueue;
025import com.nativelibs4java.opencl.CLMem.Usage;
026
027/**
028 * Parallel Random numbers generator (<a href="http://en.wikipedia.org/wiki/Xorshift">Xorshift</a> adapted to work in parallel).<br>
029 * This class was designed as a drop-in replacement for java.util.Random (albeit with a more limited API) :
030 * <pre>{@code
031 * ParallelRandom r = new ParallelRandom();
032 * r.setPreload(true); // faster
033 * while (true) {
034 *     System.out.println(r.nextDouble());
035 * }
036 * }</pre>
037 * <br>
038 * It is also possible to get entire batches of random integers with {@link ParallelRandom#next()} or {@link ParallelRandom#next(Pointer)}.<br>
039 * The preload feature precomputes a new batch in background as soon as one starts to consume numbers from the current batch.
040 * @author ochafik
041 */
042public class ParallelRandom {
043
044    protected final XORShiftRandom randomProgram;
045    //private final IntBuffer outputBuffer;
046    //private IntBuffer mappedOutputBuffer;
047    protected final CLQueue queue;
048    protected final CLContext context;
049    protected final int parallelSize;
050    protected final int[] globalWorkSizes;
051        
052        protected int consumedInts = 0;
053
054        boolean preload;
055        CLEvent preloadEvent;
056        protected CLBuffer<Integer> seeds, output;
057        Pointer<Integer> lastData;
058        boolean isDataFresh;
059        
060        public ParallelRandom() throws IOException {
061                this(JavaCL.createBestContext().createDefaultQueue(), 32 * 1024, System.currentTimeMillis());
062        }
063    public ParallelRandom(CLQueue queue, int parallelSize, final long seed) throws IOException {
064        try {
065            this.queue = queue;
066            this.context = queue.getContext();
067            randomProgram = new XORShiftRandom(context);
068            this.parallelSize = parallelSize;
069
070            int seedsNeededByWorkItem = 4;
071            //int generatedNumbersByWorkItemIteration = 1;
072            int maxUnits = queue.getDevice().getMaxComputeUnits();
073            int unitsFactor = maxUnits < 10 ? 1 : 16;
074            int scheduledWorkItems = maxUnits * unitsFactor;
075            //int countByWorkItem = parallelSize / scheduledWorkItems;
076            if (scheduledWorkItems > parallelSize / seedsNeededByWorkItem) {
077                scheduledWorkItems = parallelSize / seedsNeededByWorkItem;
078                scheduledWorkItems += parallelSize % seedsNeededByWorkItem;
079            }
080            //int iterationsByWorkItem = parallelCount / (generatedNumbersByWorkItemIteration * scheduledWorkItems);
081            globalWorkSizes = new int[] { scheduledWorkItems };
082
083            //int lws = 1;//(int)queue.getDevice().getMaxWorkGroupSize();
084            //if (lws > 32)
085            //    lws = 32;
086            //localWorkSizes = new int[] { lws };
087
088            randomProgram.getProgram().defineMacro("NUMBERS_COUNT", parallelSize);
089            randomProgram.getProgram().defineMacro("WORK_ITEMS_COUNT", scheduledWorkItems);
090
091            final int nSeeds = seedsNeededByWorkItem * parallelSize;
092            final Pointer<Integer> seedsBuf = allocateInts(nSeeds).order(context.getKernelsDefaultByteOrder());
093            initSeeds(seedsBuf, seed);
094            //println(seedsBuf);
095            this.seeds = context.createBuffer(Usage.InputOutput, seedsBuf, true);
096            //this.lastOutputData = NIOUtils.directInts(parallelSize, context.getKernelsDefaultByteOrder());
097            this.output = context.createBuffer(Usage.Output, Integer.class, parallelSize);
098        } catch (InterruptedException ex) {
099            Logger.getLogger(ParallelRandom.class.getName()).log(Level.SEVERE, null, ex);
100            throw new RuntimeException("Failed to initialized parallel random", ex);
101        }
102    }
103        
104        static final int floatMask = 0x00ffffff;
105        static final double floatDivid = (1 << 24);
106        //static final int mask = (1 << 30) - 1;
107        //static final double divid = (1 << 30);
108
109        public int nextInt() {
110                waitForData(1);
111                return lastData.get(consumedInts++);            
112        }
113        
114        /**
115         * If true, a new batch of parallel random numbers is automatically precomputed in background as soon as one starts to consume numbers from the current batch (this gives faster random numbers, at the risk of computing more values than needed)
116         */
117        public synchronized boolean isPreload() {
118                return preload;
119        }
120        /**
121         * If true, a new batch of parallel random numbers is automatically precomputed in background as soon as one starts to consume numbers from the current batch (this gives faster random numbers, at the risk of computing more values than needed)
122         */
123        public synchronized void setPreload(boolean preload) throws CLBuildException {
124                this.preload = preload;
125                if (preload && preloadEvent == null) {
126                        if (lastData == null) {
127                                preloadEvent = randomProgram.gen_numbers(queue, seeds, output, globalWorkSizes, null);
128                        } else if (consumedInts > 0) {
129                                preload();
130                        }
131                }
132        }
133        private synchronized CLEvent preload() throws CLBuildException {
134                return preloadEvent = randomProgram.gen_numbers(queue, seeds, output, globalWorkSizes, null, preloadEvent);
135        }
136        private synchronized void waitForData(int n) {
137                try {
138                        if (lastData == null) {
139                                //lastOutputData = NIOUtils.directInts(parallelSize, context.getKernelsDefaultByteOrder());
140                                if (preloadEvent == null)
141                                        preloadEvent = randomProgram.gen_numbers(queue, seeds, output, globalWorkSizes, null);
142                                        
143                                readLastOutputData();
144                        }
145                        if (consumedInts > parallelSize - n) {
146                                preload().waitFor();
147                                consumedInts = 0;
148                                readLastOutputData();
149                        }
150                        if (preload && preloadEvent == null)
151                                preload();
152                } catch (CLBuildException ex) {
153                        throw new RuntimeException(ex);
154                }
155        }
156        private synchronized void readLastOutputData() {
157                if (lastData == null)
158                        lastData = output.read(queue, preloadEvent);
159                else
160                        output.read(queue, lastData, true, preloadEvent);
161                preloadEvent = null;
162        }
163        public long nextLong() {
164        return (((long)nextInt()) << 32) | nextInt();
165    }
166        
167        private static final int intSignMask = 1 << 31;
168        public int nextInt(int n) {
169        if (n <= 0)
170            throw new IllegalArgumentException("n must be positive");
171
172        if ((n & -n) == n)  // i.e., n is a power of 2
173            return (int)((n * (long)(nextInt() & intSignMask)) >> 31);
174
175        int bits, val;
176        do {
177            bits = nextInt() & intSignMask;
178            val = bits % n;
179        } while (bits - val + (n-1) < 0);
180        return val;
181    }
182
183        public float nextFloat() {
184                return (float)((nextInt() & floatMask) / floatDivid);
185        }
186        
187        private static final int doubleMask = (1 << 27) - 1;
188        private static final double doubleDivid = 1L << 53;
189        
190        public double nextDouble() {
191                return (((long)(nextInt() & doubleMask) << 27) | (nextInt() & doubleMask)) / doubleDivid;
192        }
193
194        public CLBuffer<Integer> getSeeds() {
195                return seeds;
196        }
197        public CLQueue getQueue() {
198                return queue;
199        }
200        
201    /**
202     * Number of random numbers generated at each call of {@link ParallelRandom#next() } or {@link ParallelRandom#next(Pointer) }<br>
203     * The numbers might not all be generated exactly in parallel, the level of parallelism is implementation-dependent.
204     * @return size of each buffer returned by {@link ParallelRandom#next() }
205     */
206    public int getParallelSize() {
207        return parallelSize;
208    }
209    
210    public synchronized CLEvent doNext() {
211        try {
212            //if (mappedOutputBuffer != null) {
213            //    //output.unmap(queue, mappedOutputBuffer);
214            //    mappedOutputBuffer = null;
215            //}
216            return randomProgram.gen_numbers(queue, seeds, //parallelSize,
217                    output, globalWorkSizes, null);
218        } catch (CLBuildException ex) {
219            Logger.getLogger(ParallelRandom.class.getName()).log(Level.SEVERE, null, ex);
220            throw new RuntimeException("Failed to compile the random number generation routine", ex);
221        }
222    }
223
224    /**
225     * Copies the next {@link ParallelRandom#getParallelSize() } random integers in the provided output buffer
226     * @param output
227     */
228    public synchronized void next(Pointer<Integer> output) {
229        CLEvent evt = doNext();
230        this.output.read(queue, output, true, evt);
231    }
232
233    
234    /**
235     * Returns a direct NIO buffer containing the next {@link ParallelRandom#getParallelSize() } random integers.<br>
236     * This buffer is read only and will only be valid until any of the "next" method is called again.
237     * @return output buffer of capacity ; see {@link ParallelRandom#getParallelSize() }
238     */
239    public synchronized Pointer<Integer> next() {
240        CLEvent evt = doNext();
241        //queue.finish(); evt = null;
242        //return outputBuffer;
243        //return (mappedOutputBuffer = output.map(queue, MapFlags.Read, evt)).asReadOnlyBuffer();
244        return output.read(queue, evt);
245    }
246
247    private void initSeeds(final Pointer<Integer> seedsBuf, final long seed) throws InterruptedException {
248        final long nSeeds = seedsBuf.getValidElements();
249
250        long start = System.nanoTime();
251
252        // TODO benchmark threshold :
253        boolean parallelize = nSeeds > 10000;
254        //parallelize = false;
255        if (parallelize) {
256            Random random = new Random(seed);
257            for (long i = nSeeds; i-- != 0;)
258                seedsBuf.set(i, random.nextInt());
259        } else {
260            // Parallelize seeds initialization
261            final int nThreads = Runtime.getRuntime().availableProcessors();// * 2;
262            ExecutorService service = Executors.newFixedThreadPool(nThreads);
263            for (int i = 0; i < nThreads; i++) {
264                final int iThread = i;
265                service.execute(new Runnable() {
266
267                    public void run() {
268                        long n = nSeeds / nThreads;
269                        long offset = n * iThread;
270                        Random random = new Random(seed + iThread);// * System.currentTimeMillis());
271                        if (iThread == nThreads - 1)
272                            n += nSeeds - n * nThreads;
273                        
274                        for (long i = n; i-- != 0;)
275                            seedsBuf.set(offset++, random.nextInt());
276                    }
277                });
278            }
279            service.shutdown();
280            service.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
281        }
282
283        long time = System.nanoTime() - start;
284        Logger.getLogger(ParallelRandom.class.getName()).log(Level.INFO, "Initialization of " + nSeeds + " seeds took " + (time/1000000) + " ms (" + (time / (double)nSeeds) + " ns per seed)");
285    }
286}