- initial import of CLTaskCompletionService.

 - fixed pool shutdown behavior. It will now wait till all started tasks finish to be able to release the queue context.

1 files changed, 62 insertions, 25 deletions

diff --git a/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java b/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java
index 68ca33f9..6e5d5e99 100644
--- a/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java
+++ b/test/com/jogamp/opencl/util/concurrent/CLMultiContextTest.java
@@ -19,7 +19,6 @@ 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;
@@ -70,29 +69,36 @@ public class CLMultiContextTest {
         + "    array[index]++;                                       \n"
         + "}                                                         \n";
 
-    private final class CLTestTask implements CLTask<CLSimpleQueueContext, Buffer> {
+    private final class CLTestTask implements CLTask<CLSimpleQueueContext, IntBuffer> {
 
-        private final Buffer data;
+        private final IntBuffer data;
 
-        public CLTestTask(Buffer buffer) {
+        public CLTestTask(IntBuffer buffer) {
             this.data = buffer;
         }
 
         @Override
-        public Buffer execute(CLSimpleQueueContext qc) {
+        public IntBuffer execute(CLSimpleQueueContext qc) {
             
             CLCommandQueue queue = qc.getQueue();
             CLContext context = qc.getCLContext();
             CLKernel kernel = qc.getKernel("compute");
 
-            CLBuffer<Buffer> buffer = null;
+//            System.out.println(Thread.currentThread().getName()+" / "+queue);
+            assertFalse(qc.isReleased());
+            assertFalse(queue.isReleased());
+            assertFalse(context.isReleased());
+            assertFalse(kernel.isReleased());
+
+            CLBuffer<IntBuffer> buffer = null;
             try{
+
                 buffer = context.createBuffer(data);
                 int gws = buffer.getCLCapacity();
 
                 kernel.putArg(buffer).putArg(gws).rewind();
 
-                queue.putWriteBuffer(buffer, true);
+                queue.putWriteBuffer(buffer, false);
                 queue.put1DRangeKernel(kernel, 0, gws, 0);
                 queue.putReadBuffer(buffer, true);
             }finally{
@@ -106,6 +112,16 @@ public class CLMultiContextTest {
 
     }
 
+    private List<CLTestTask> createTasks(IntBuffer data, int taskCount, int slice) {
+        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));
+        }
+        return tasks;
+    }
+
     @Test
     public void commandQueuePoolTest() throws InterruptedException, ExecutionException {
 
@@ -123,48 +139,62 @@ public class CLMultiContextTest {
             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));
-            }
+            List<CLTestTask> tasks = createTasks(data, taskCount, slice);
 
             out.println("invoking "+tasks.size()+" tasks on "+pool.getSize()+" queues");
 
             // blocking invoke
-            List<Future<Buffer>> results = pool.invokeAll(tasks);
+            List<Future<IntBuffer>> results = pool.invokeAll(tasks);
             assertNotNull(results);
             checkBuffer(1, data);
 
             // submit blocking emediatly
             for (CLTestTask task : tasks) {
-                Buffer ret = pool.submit(task).get();
+                IntBuffer ret = pool.submit(task).get();
                 assertNotNull(ret);
+                checkBuffer(2, ret);
             }
             checkBuffer(2, data);
 
             // submitAll using futures
-            List<Future<Buffer>> futures = pool.submitAll(tasks);
-            for (Future<Buffer> future : futures) {
-                Buffer ret = future.get();
+            List<Future<IntBuffer>> futures = pool.submitAll(tasks);
+            for (Future<IntBuffer> future : futures) {
+                IntBuffer ret = future.get();
                 assertNotNull(ret);
+                checkBuffer(3, ret);
             }
             checkBuffer(3, data);
 
             // switching contexts using different program
             factory = CLQueueContextFactory.createSimple(programSource.replaceAll("\\+\\+", "--"));
             pool.switchContext(factory);
-            List<Future<Buffer>> results2 = pool.invokeAll(tasks);
+            List<Future<IntBuffer>> results2 = pool.invokeAll(tasks);
             assertNotNull(results2);
             checkBuffer(2, data);
-            
+
+            // Note: we have to make sure that we don't resubmit old tasks at this point since
+            // we wait only for completion of a subset of tasks.
             // submit any
-            Buffer buffer = pool.invokeAny(tasks);
-            assertNotNull(buffer);
-            checkContains(1, data);
+            data = Buffers.newDirectIntBuffer(slice*taskCount);
+            tasks = createTasks(data, taskCount, slice);
+
+            IntBuffer ret1 = pool.invokeAny(tasks);
+            assertNotNull(ret1);
+            checkBuffer(-1, ret1);
+            checkContains(-1, data);
+
+            // completionservice take/any test
+            data = Buffers.newDirectIntBuffer(slice*taskCount);
+            tasks = createTasks(data, taskCount, slice);
+
+            CLTaskCompletionService<CLSimpleQueueContext, IntBuffer> service = new CLTaskCompletionService(pool);
+            for (CLTestTask task : tasks) {
+                service.submit(task);
+            }
+            IntBuffer ret2 = service.take().get();
+            assertNotNull(ret2);
+            checkBuffer(-1, ret2);
+            checkContains(-1, data);
 
             pool.release();
         }finally{
@@ -189,4 +219,11 @@ public class CLMultiContextTest {
         fail();
     }
 
+//    @Test
+    public void loadTest() throws InterruptedException, ExecutionException {
+        for (int i = 0; i < 40; i++) {
+            commandQueuePoolTest();
+        }
+    }
+
 }