diff options
5 files changed, 168 insertions, 32 deletions
diff --git a/src/com/jogamp/opencl/util/CLMultiContext.java b/src/com/jogamp/opencl/util/CLMultiContext.java index f588fcef..f74c0a35 100644 --- a/src/com/jogamp/opencl/util/CLMultiContext.java +++ b/src/com/jogamp/opencl/util/CLMultiContext.java @@ -41,6 +41,13 @@ public class CLMultiContext implements CLResource { * Creates a multi context with all devices of the specified platforms and types. */ public static CLMultiContext create(CLPlatform[] platforms, CLDevice.Type... types) { + + if(platforms == null) { + throw new NullPointerException("platform list was null"); + }else if(platforms.length == 0) { + throw new IllegalArgumentException("platform list was empty"); + } + List<CLDevice> devices = new ArrayList<CLDevice>(); for (CLPlatform platform : platforms) { devices.addAll(asList(platform.listCLDevices(types))); @@ -54,6 +61,10 @@ public class CLMultiContext implements CLResource { */ public static CLMultiContext create(Collection<CLDevice> devices) { + if(devices.isEmpty()) { + throw new IllegalArgumentException("device list was empty"); + } + Map<CLPlatform, List<CLDevice>> platformDevicesMap = filterPlatformConflicts(devices); // create contexts diff --git a/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java b/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java index b80f09e6..a6bbe4d0 100644 --- a/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java +++ b/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java @@ -15,6 +15,7 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; /** * A multithreaded fixed size pool of OpenCL command queues. @@ -23,7 +24,7 @@ import java.util.concurrent.ThreadFactory; * instead of {@link Callable}s. * @author Michael Bien */ -public class CLCommandQueuePool implements CLResource { +public class CLCommandQueuePool<C extends CLQueueContext> implements CLResource { private List<CLQueueContext> contexts; private ExecutorService excecutor; @@ -55,11 +56,11 @@ public class CLCommandQueuePool implements CLResource { this.excecutor = Executors.newFixedThreadPool(contexts.size(), new QueueThreadFactory(contexts)); } - public static CLCommandQueuePool create(CLQueueContextFactory factory, CLMultiContext mc, CLCommandQueue.Mode... modes) { + public static <C extends CLQueueContext> CLCommandQueuePool<C> create(CLQueueContextFactory<C> factory, CLMultiContext mc, CLCommandQueue.Mode... modes) { return create(factory, mc.getDevices(), modes); } - public static CLCommandQueuePool create(CLQueueContextFactory factory, Collection<CLDevice> devices, CLCommandQueue.Mode... modes) { + public static <C extends CLQueueContext> CLCommandQueuePool<C> create(CLQueueContextFactory<C> factory, Collection<CLDevice> devices, CLCommandQueue.Mode... modes) { List<CLCommandQueue> queues = new ArrayList<CLCommandQueue>(devices.size()); for (CLDevice device : devices) { queues.add(device.createCommandQueue(modes)); @@ -67,21 +68,43 @@ public class CLCommandQueuePool implements CLResource { return create(factory, queues); } - public static CLCommandQueuePool create(CLQueueContextFactory factory, Collection<CLCommandQueue> queues) { + public static <C extends CLQueueContext> CLCommandQueuePool create(CLQueueContextFactory<C> factory, Collection<CLCommandQueue> queues) { return new CLCommandQueuePool(factory, queues); } - public <R> Future<R> submit(CLTask<R> task) { + /** + * @see ExecutorService#submit(java.util.concurrent.Callable) + */ + public <R> Future<R> submit(CLTask<? extends C, R> task) { return excecutor.submit(new TaskWrapper(task, finishAction)); } - public <R> List<Future<R>> invokeAll(Collection<CLTask<R>> tasks) throws InterruptedException { - List<TaskWrapper<R>> wrapper = new ArrayList<TaskWrapper<R>>(tasks.size()); - for (CLTask<R> task : tasks) { - wrapper.add(new TaskWrapper<R>(task, finishAction)); - } + /** + * @see ExecutorService#invokeAll(java.util.Collection) + */ + public <R> List<Future<R>> invokeAll(Collection<? extends CLTask<? super C, R>> tasks) throws InterruptedException { + List<TaskWrapper<C, R>> wrapper = wrapTasks(tasks); return excecutor.invokeAll(wrapper); } + + /** + * @see ExecutorService#invokeAll(java.util.Collection, long, java.util.concurrent.TimeUnit) + */ + public <R> List<Future<R>> invokeAll(Collection<? extends CLTask<? super C, R>> tasks, long timeout, TimeUnit unit) throws InterruptedException { + List<TaskWrapper<C, R>> wrapper = wrapTasks(tasks); + return excecutor.invokeAll(wrapper, timeout, unit); + } + + private <R> List<TaskWrapper<C, R>> wrapTasks(Collection<? extends CLTask<? super C, R>> tasks) { + List<TaskWrapper<C, R>> wrapper = new ArrayList<TaskWrapper<C, R>>(tasks.size()); + for (CLTask<? super C, R> task : tasks) { + if(task == null) { + throw new NullPointerException("at least one task was null"); + } + wrapper.add(new TaskWrapper<C, R>(task, finishAction)); + } + return wrapper; + } /** * Switches the context of all queues - this operation can be expensive. @@ -171,35 +194,41 @@ public class CLCommandQueuePool implements CLResource { this.index = 0; } - public synchronized Thread newThread(Runnable r) { + public synchronized Thread newThread(Runnable runnable) { + + SecurityManager sm = System.getSecurityManager(); + ThreadGroup group = (sm != null)? sm.getThreadGroup() : Thread.currentThread().getThreadGroup(); + CLQueueContext queue = context.get(index); - return new QueueThread(queue, index++); + QueueThread thread = new QueueThread(group, runnable, queue, index++); + thread.setDaemon(true); + + return thread; } } private static class QueueThread extends Thread { private final CLQueueContext context; - public QueueThread(CLQueueContext context, int index) { - super("queue-worker-thread-"+index+"["+context+"]"); + public QueueThread(ThreadGroup group, Runnable runnable, CLQueueContext context, int index) { + super(group, runnable, "queue-worker-thread-"+index+"["+context+"]"); this.context = context; - this.setDaemon(true); } } - private static class TaskWrapper<T> implements Callable<T> { + private static class TaskWrapper<C extends CLQueueContext, R> implements Callable<R> { - private final CLTask<T> task; + private final CLTask<? super C, R> task; private final FinishAction mode; - public TaskWrapper(CLTask<T> task, FinishAction mode) { + public TaskWrapper(CLTask<? super C, R> task, FinishAction mode) { this.task = task; this.mode = mode; } - public T call() throws Exception { + public R call() throws Exception { CLQueueContext context = ((QueueThread)Thread.currentThread()).context; - T result = task.run(context); + R result = task.execute((C)context); if(mode.equals(FinishAction.FLUSH)) { context.queue.flush(); }else if(mode.equals(FinishAction.FINISH)) { diff --git a/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java b/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java index 3956f93d..11b86889 100644 --- a/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java +++ b/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java @@ -4,6 +4,7 @@ package com.jogamp.opencl.util.concurrent; import com.jogamp.opencl.CLCommandQueue; +import com.jogamp.opencl.CLContext; import com.jogamp.opencl.CLKernel; import com.jogamp.opencl.CLProgram; import com.jogamp.opencl.CLResource; @@ -24,6 +25,10 @@ public abstract class CLQueueContext implements CLResource { return queue; } + public CLContext getCLContext() { + return queue.getContext(); + } + public static class CLSimpleQueueContext extends CLQueueContext { public final CLProgram program; @@ -39,6 +44,10 @@ public abstract class CLQueueContext implements CLResource { return kernels; } + public CLKernel getKernel(String name) { + return kernels.get(name); + } + public CLProgram getProgram() { return program; } diff --git a/src/com/jogamp/opencl/util/concurrent/CLTask.java b/src/com/jogamp/opencl/util/concurrent/CLTask.java index ff0f7614..0cfd24a5 100644 --- a/src/com/jogamp/opencl/util/concurrent/CLTask.java +++ b/src/com/jogamp/opencl/util/concurrent/CLTask.java @@ -8,11 +8,11 @@ package com.jogamp.opencl.util.concurrent; * A task executed on a command queue. * @author Michael Bien */ -public interface CLTask<R> { +public interface CLTask<C extends CLQueueContext, R> { /** * Runs the task on a queue and returns a result. */ - R run(CLQueueContext context); + R execute(C context); } diff --git a/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java b/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java index 7a1ed7aa..e5bcb1c5 100644 --- a/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java +++ b/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java @@ -3,14 +3,24 @@ */ package com.jogamp.opencl.util.concurrent; +import com.jogamp.common.nio.Buffers; +import com.jogamp.opencl.CLBuffer; +import com.jogamp.opencl.CLCommandQueue; import com.jogamp.opencl.CLContext; import com.jogamp.opencl.CLDevice; +import com.jogamp.opencl.CLKernel; import com.jogamp.opencl.CLPlatform; +import com.jogamp.opencl.util.concurrent.CLQueueContext.CLSimpleQueueContext; import com.jogamp.opencl.util.concurrent.CLQueueContextFactory.CLSimpleContextFactory; +import java.nio.IntBuffer; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; import org.junit.Rule; import org.junit.rules.MethodRule; import org.junit.rules.Timeout; import com.jogamp.opencl.util.CLMultiContext; +import java.nio.Buffer; +import java.util.ArrayList; import java.util.List; import org.junit.Test; @@ -52,27 +62,97 @@ public class CLMultiContextTest { } private final static String programSource = - "kernel void vectorAdd(global const int* a, global const int* b, global int* c, int iNumElements) { \n" - + " int iGID = get_global_id(0); \n" - + " if (iGID >= iNumElements) { \n" - + " return; \n" - + " } \n" - + " c[iGID] = a[iGID] + b[iGID]; \n" - + "} \n"; + "kernel void increment(global int* array, int numElements) { \n" + + " int index = get_global_id(0); \n" + + " if (index >= numElements) { \n" + + " return; \n" + + " } \n" + + " array[index]++; \n" + + "} \n"; + + private final class CLTestTask implements CLTask<CLSimpleQueueContext, Buffer> { + + private final Buffer data; + + public CLTestTask(Buffer buffer) { + this.data = buffer; + } + + public Buffer execute(CLSimpleQueueContext qc) { + + CLCommandQueue queue = qc.getQueue(); + CLContext context = qc.getCLContext(); + CLKernel kernel = qc.getKernel("increment"); + + CLBuffer<Buffer> buffer = null; + try{ + buffer = context.createBuffer(data); + int gws = buffer.getCLCapacity(); + + kernel.putArg(buffer).putArg(gws).rewind(); + + queue.putWriteBuffer(buffer, true); + queue.put1DRangeKernel(kernel, 0, gws, 0); + queue.putReadBuffer(buffer, true); + }finally{ + if(buffer != null) { + buffer.release(); + } + } + + return data; + } + + } @Test - public void commandQueuePoolTest() { + public void commandQueuePoolTest() throws InterruptedException, ExecutionException { CLMultiContext mc = CLMultiContext.create(CLPlatform.listCLPlatforms()); try { CLSimpleContextFactory factory = CLQueueContextFactory.createSimple(programSource); - CLCommandQueuePool pool = CLCommandQueuePool.create(factory, mc); + CLCommandQueuePool<CLSimpleQueueContext> pool = CLCommandQueuePool.create(factory, mc); assertTrue(pool.getSize() > 0); + + final int slice = 64; + final int tasksPerQueue = 10; + final int taskCount = pool.getSize() * tasksPerQueue; + + IntBuffer data = Buffers.newDirectIntBuffer(slice*taskCount); + + List<CLTestTask> tasks = new ArrayList<CLTestTask>(taskCount); + + for (int i = 0; i < taskCount; i++) { + IntBuffer subBuffer = Buffers.slice(data, i*slice, slice); + assertEquals(slice, subBuffer.capacity()); + tasks.add(new CLTestTask(subBuffer)); + } + + out.println("invoking "+tasks.size()+" tasks on "+pool.getSize()+" queues"); + + pool.invokeAll(tasks); + checkBuffer(1, data); + + + for (CLTestTask task : tasks) { + pool.submit(task).get(); + } + checkBuffer(2, data); + + + List<Future<Buffer>> futures = new ArrayList<Future<Buffer>>(taskCount); + for (CLTestTask task : tasks) { + futures.add(pool.submit(task)); + } + for (Future<Buffer> future : futures) { + future.get(); + } + checkBuffer(3, data); - pool.switchContext(factory); +// pool.switchContext(factory); pool.release(); }finally{ @@ -80,4 +160,11 @@ public class CLMultiContextTest { } } + private void checkBuffer(int expected, IntBuffer data) { + while(data.hasRemaining()) { + assertEquals(expected, data.get()); + } + data.rewind(); + } + } |