summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorMichael Bien <mbien@fh-landshut.de>2011-05-07 02:11:44 +0200
committerMichael Bien <mbien@fh-landshut.de>2011-05-07 02:11:44 +0200
commit19cc9195c73002f84c153a1ffc60f00408e1176e (patch)
tree2be66b79e071e1acddabf89eae3dd440435f26a4
parent8df524bf292051455005869ddfcfcc761af576e1 (diff)
introduced CLQueueContext and its factory - WIP.
-rw-r--r--src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java82
-rw-r--r--src/com/jogamp/opencl/util/concurrent/CLQueueContext.java52
-rw-r--r--src/com/jogamp/opencl/util/concurrent/CLQueueContextFactory.java46
-rw-r--r--src/com/jogamp/opencl/util/concurrent/CLTask.java5
-rw-r--r--test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java35
5 files changed, 180 insertions, 40 deletions
diff --git a/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java b/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java
index 92828e9..ee6dc86 100644
--- a/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java
+++ b/src/com/jogamp/opencl/util/concurrent/CLCommandQueuePool.java
@@ -18,7 +18,7 @@ import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
/**
- * A multithreaded pool of OpenCL command queues.
+ * A multithreaded fixed size pool of OpenCL command queues.
* It serves as a multiplexer distributing tasks over N queues.
* The usage of this pool is similar to {@link ExecutorService} but it uses {@link CLTask}s
* instead of {@link Callable}s.
@@ -26,29 +26,37 @@ import java.util.concurrent.ThreadFactory;
*/
public class CLCommandQueuePool implements CLResource {
- private final List<CLCommandQueue> queues;
- private final ExecutorService excecutor;
+ private final List<CLQueueContext> contexts;
+ private ExecutorService excecutor;
private FinishAction finishAction = FinishAction.DO_NOTHING;
- private CLCommandQueuePool(Collection<CLCommandQueue> queues) {
- this.queues = Collections.unmodifiableList(new ArrayList<CLCommandQueue>(queues));
- this.excecutor = Executors.newFixedThreadPool(queues.size(), new QueueThreadFactory(this.queues));
+ private CLCommandQueuePool(Collection<CLQueueContext> contexts) {
+ this.contexts = Collections.unmodifiableList(new ArrayList<CLQueueContext>(contexts));
+ initExecutor();
}
- public static CLCommandQueuePool create(CLMultiContext mc, CLCommandQueue.Mode... modes) {
- return create(mc.getDevices(), modes);
+ private void initExecutor() {
+ this.excecutor = Executors.newFixedThreadPool(contexts.size(), new QueueThreadFactory(contexts));
}
- public static CLCommandQueuePool create(Collection<CLDevice> devices, CLCommandQueue.Mode... modes) {
+ public static CLCommandQueuePool create(CLQueueContextFactory factory, CLMultiContext mc, CLCommandQueue.Mode... modes) {
+ return create(factory, mc.getDevices(), modes);
+ }
+
+ public static CLCommandQueuePool create(CLQueueContextFactory factory, Collection<CLDevice> devices, CLCommandQueue.Mode... modes) {
List<CLCommandQueue> queues = new ArrayList<CLCommandQueue>(devices.size());
for (CLDevice device : devices) {
queues.add(device.createCommandQueue(modes));
}
- return create(queues);
+ return create(factory, queues);
}
- public static CLCommandQueuePool create(Collection<CLCommandQueue> queues) {
- return new CLCommandQueuePool(queues);
+ public static CLCommandQueuePool create(CLQueueContextFactory factory, Collection<CLCommandQueue> queues) {
+ List<CLQueueContext> contexts = new ArrayList<CLQueueContext>(queues.size());
+ for (CLCommandQueue queue : queues) {
+ contexts.add(factory.setup(queue, null));
+ }
+ return new CLCommandQueuePool(contexts);
}
public <R> Future<R> submit(CLTask<R> task) {
@@ -66,18 +74,18 @@ public class CLCommandQueuePool implements CLResource {
/**
* Calls {@link CLCommandQueue#flush()} on all queues.
*/
- public void flush() {
- for (CLCommandQueue queue : queues) {
- queue.flush();
+ public void flushQueues() {
+ for (CLQueueContext context : contexts) {
+ context.queue.flush();
}
}
/**
* Calls {@link CLCommandQueue#finish()} on all queues.
*/
- public void finish() {
- for (CLCommandQueue queue : queues) {
- queue.finish();
+ public void finishQueues() {
+ for (CLQueueContext context : contexts) {
+ context.queue.finish();
}
}
@@ -85,16 +93,20 @@ public class CLCommandQueuePool implements CLResource {
* Releases all queues.
*/
public void release() {
- for (CLCommandQueue queue : queues) {
- queue.finish().release();
- }
excecutor.shutdown();
+ for (CLQueueContext context : contexts) {
+ context.queue.finish().release();
+ }
}
/**
* Returns the command queues used in this pool.
*/
public List<CLCommandQueue> getQueues() {
+ List<CLCommandQueue> queues = new ArrayList<CLCommandQueue>(contexts.size());
+ for (CLQueueContext context : contexts) {
+ queues.add(context.queue);
+ }
return queues;
}
@@ -102,7 +114,7 @@ public class CLCommandQueuePool implements CLResource {
* Returns the size of this pool (number of command queues).
*/
public int getSize() {
- return queues.size();
+ return contexts.size();
}
public FinishAction getFinishAction() {
@@ -115,31 +127,31 @@ public class CLCommandQueuePool implements CLResource {
@Override
public String toString() {
- return getClass().getSimpleName()+" [queues: "+queues.size()+" on finish: "+finishAction+"]";
+ return getClass().getSimpleName()+" [queues: "+contexts.size()+" on finish: "+finishAction+"]";
}
private static class QueueThreadFactory implements ThreadFactory {
- private final List<CLCommandQueue> queues;
+ private final List<CLQueueContext> context;
private int index;
- private QueueThreadFactory(List<CLCommandQueue> queues) {
- this.queues = queues;
+ private QueueThreadFactory(List<CLQueueContext> queues) {
+ this.context = queues;
this.index = 0;
}
public synchronized Thread newThread(Runnable r) {
- CLCommandQueue queue = queues.get(index);
+ CLQueueContext queue = context.get(index);
return new QueueThread(queue, index++);
}
}
private static class QueueThread extends Thread {
- private final CLCommandQueue queue;
- public QueueThread(CLCommandQueue queue, int index) {
- super("queue-worker-thread-"+index+"["+queue+"]");
- this.queue = queue;
+ private final CLQueueContext context;
+ public QueueThread(CLQueueContext context, int index) {
+ super("queue-worker-thread-"+index+"["+context+"]");
+ this.context = context;
this.setDaemon(true);
}
}
@@ -155,12 +167,12 @@ public class CLCommandQueuePool implements CLResource {
}
public T call() throws Exception {
- CLCommandQueue queue = ((QueueThread)Thread.currentThread()).queue;
- T result = task.run(queue);
+ CLQueueContext context = ((QueueThread)Thread.currentThread()).context;
+ T result = task.run(context);
if(mode.equals(FinishAction.FLUSH)) {
- queue.flush();
+ context.queue.flush();
}else if(mode.equals(FinishAction.FINISH)) {
- queue.finish();
+ context.queue.finish();
}
return result;
}
diff --git a/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java b/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java
new file mode 100644
index 0000000..fef0047
--- /dev/null
+++ b/src/com/jogamp/opencl/util/concurrent/CLQueueContext.java
@@ -0,0 +1,52 @@
+/*
+ * Created on Friday, May 06 2011 21:02
+ */
+package com.jogamp.opencl.util.concurrent;
+
+import com.jogamp.opencl.CLCommandQueue;
+import com.jogamp.opencl.CLKernel;
+import com.jogamp.opencl.CLProgram;
+import com.jogamp.opencl.CLResource;
+import java.util.Map;
+
+/**
+ * @author Michael Bien
+ */
+public abstract class CLQueueContext implements CLResource {
+
+ public final CLCommandQueue queue;
+
+ public CLQueueContext(CLCommandQueue queue) {
+ this.queue = queue;
+ }
+
+ public CLCommandQueue getQueue() {
+ return queue;
+ }
+
+ public static class CLSimpleQueueContext extends CLQueueContext {
+
+ public final CLProgram program;
+ public final Map<String, CLKernel> kernels;
+
+ public CLSimpleQueueContext(CLCommandQueue queue, CLProgram program) {
+ super(queue);
+ this.program = program;
+ this.kernels = program.createCLKernels();
+ }
+
+ public Map<String, CLKernel> getKernels() {
+ return kernels;
+ }
+
+ public CLProgram getProgram() {
+ return program;
+ }
+
+ public void release() {
+ throw new UnsupportedOperationException("Not supported yet.");
+ }
+
+ }
+
+}
diff --git a/src/com/jogamp/opencl/util/concurrent/CLQueueContextFactory.java b/src/com/jogamp/opencl/util/concurrent/CLQueueContextFactory.java
new file mode 100644
index 0000000..64fdfbc
--- /dev/null
+++ b/src/com/jogamp/opencl/util/concurrent/CLQueueContextFactory.java
@@ -0,0 +1,46 @@
+/*
+ * Created onSaturday, May 07 2011 00:40
+ */
+package com.jogamp.opencl.util.concurrent;
+
+import com.jogamp.opencl.CLCommandQueue;
+import com.jogamp.opencl.CLProgram;
+
+/**
+ *
+ * @author Michael Bien
+ */
+public abstract class CLQueueContextFactory<C extends CLQueueContext> {
+
+ /**
+ * Creates a new queue context for the given queue.
+ * @param old the old context or null.
+ */
+ public abstract C setup(CLCommandQueue queue, CLQueueContext old);
+
+
+ /**
+ * Creates a simple context factory producing single program contexts.
+ * @param source sourcecode of a OpenCL program.
+ */
+ public static CLSimpleContextFactory createSimple(String source) {
+ return new CLSimpleContextFactory(source);
+ }
+
+ public static class CLSimpleContextFactory extends CLQueueContextFactory<CLQueueContext.CLSimpleQueueContext> {
+
+ private final String source;
+
+ public CLSimpleContextFactory(String source) {
+ this.source = source;
+ }
+
+ @Override
+ public CLQueueContext.CLSimpleQueueContext setup(CLCommandQueue queue, CLQueueContext old) {
+ CLProgram program = queue.getContext().createProgram(source).build(queue.getDevice());
+ return new CLQueueContext.CLSimpleQueueContext(queue, program);
+ }
+
+ }
+
+}
diff --git a/src/com/jogamp/opencl/util/concurrent/CLTask.java b/src/com/jogamp/opencl/util/concurrent/CLTask.java
index ebecb93..ff0f761 100644
--- a/src/com/jogamp/opencl/util/concurrent/CLTask.java
+++ b/src/com/jogamp/opencl/util/concurrent/CLTask.java
@@ -3,7 +3,6 @@
*/
package com.jogamp.opencl.util.concurrent;
-import com.jogamp.opencl.CLCommandQueue;
/**
* A task executed on a command queue.
@@ -12,8 +11,8 @@ import com.jogamp.opencl.CLCommandQueue;
public interface CLTask<R> {
/**
- * Runs the task on a queue and returns its result.
+ * Runs the task on a queue and returns a result.
*/
- R run(CLCommandQueue queue);
+ R run(CLQueueContext context);
}
diff --git a/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java b/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java
index 8e96daf..f076324 100644
--- a/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java
+++ b/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java
@@ -6,6 +6,7 @@ package com.jogamp.opencl.util.concurrent;
import com.jogamp.opencl.CLContext;
import com.jogamp.opencl.CLDevice;
import com.jogamp.opencl.CLPlatform;
+import com.jogamp.opencl.util.concurrent.CLQueueContextFactory.CLSimpleContextFactory;
import org.junit.Rule;
import org.junit.rules.MethodRule;
import org.junit.rules.Timeout;
@@ -22,8 +23,8 @@ import static java.lang.System.*;
*/
public class CLMultiContextTest {
- @Rule
- public MethodRule methodTimeout= new Timeout(10000);
+// @Rule
+// public MethodRule methodTimeout= new Timeout(10000);
@Test
public void createMultiContextTest() {
@@ -50,5 +51,35 @@ public class CLMultiContextTest {
}
+ private final static String programSource =
+ " // OpenCL Kernel Function for element by element vector addition \n"
+ + "kernel void vectorAdd(global const int* a, global const int* b, global int* c, int iNumElements) { \n"
+ + " // get index in global data array \n"
+ + " int iGID = get_global_id(0); \n"
+ + " // bound check (equivalent to the limit on a 'for' loop for standard/serial C code \n"
+ + " if (iGID >= iNumElements) { \n"
+ + " return; \n"
+ + " } \n"
+ + " // add the vector elements \n"
+ + " c[iGID] = a[iGID] + b[iGID]; \n"
+ + "} \n";
+
+ @Test
+ public void commandQueuePoolTest() {
+
+ CLMultiContext mc = CLMultiContext.create(CLPlatform.listCLPlatforms());
+
+ try {
+
+ CLSimpleContextFactory factory = CLQueueContextFactory.createSimple(programSource);
+ CLCommandQueuePool pool = CLCommandQueuePool.create(factory, mc);
+
+ assertTrue(pool.getSize() > 0);
+
+ pool.release();
+ }finally{
+ mc.release();
+ }
+ }
}