move worker to cl

4 files changed, 89 insertions, 15 deletions

diff --git a/include/cl_command_queue.h b/include/cl_command_queue.h
index c4b93638..7f673f00 100644
--- a/include/cl_command_queue.h
+++ b/include/cl_command_queue.h
@@ -47,6 +47,7 @@ typedef struct _cl_command_queue_work_item {
   cl_command_queue queue;
   cl_event* depend_events;
   cl_uint depend_event_num;
+  cl_bool queued;
   void* pdata;
 } _cl_command_queue_work_item;
 typedef _cl_command_queue_work_item* cl_command_queue_work_item;
diff --git a/include/cl_driver.h b/include/cl_driver.h
index a5d0630a..895d047c 100644
--- a/include/cl_driver.h
+++ b/include/cl_driver.h
@@ -20,6 +20,7 @@
 
 #include "CL/cl.h"
 #include "cl_kernel.h"
+#include "cl_command_queue.h"
 
 /* We put a header to identify each object. This will make the programmer life
  * easy if objects are wrongly used in the API
@@ -77,6 +78,7 @@ typedef struct _cl_driver {
   cl_int (*event_changed)(cl_event event, cl_command_queue queue, cl_int status);
   cl_int (*wait_for_events)(cl_uint num_events, cl_event* events, cl_command_queue queue);
 
+  cl_int (*destroy_work_item)(cl_command_queue queue, cl_command_queue_work_item item);
 
 
 
diff --git a/libclapi/cl_enqueue.c b/libclapi/cl_enqueue.c
index d7dba5c8..5509b6c3 100644
--- a/libclapi/cl_enqueue.c
+++ b/libclapi/cl_enqueue.c
@@ -24,6 +24,8 @@
 #include "cl_alloc.h"
 #include "cl_internals.h"
 #include "cl_command_queue.h"
+#include "cl_driver.h"
+#include "cl_device_id.h"
 #include "cl_event.h"
 
 typedef struct _cl_command_queue_worker {
@@ -41,20 +43,28 @@ typedef _cl_command_queue_worker* cl_command_queue_worker;
 /* Call this must with worker->mutex locked. */
 static void insert_one_work_item(cl_command_queue_worker worker, cl_command_queue_work_item it)
 {
-  it->next = worker->work_items;
-  if (worker->work_items != NULL)
+  if (worker->work_items == NULL) {
+    worker->work_items = it;
+    it->prev = it->next = it;
+    worker->work_items = it;
+    return;
+  } else {
+    worker->work_items->prev->next = it;
+    it->prev = worker->work_items->prev;
+    it->next = worker->work_items;
     worker->work_items->prev = it;
-  worker->work_items = it;
+  }
 }
 /* Call this must with worker->mutex locked. */
 static void delete_one_work_item(cl_command_queue_worker worker, cl_command_queue_work_item it)
 {
-  if (it->prev)
+  if (it->prev == it->next) { // Last one
+    assert(it->prev == it);
+    worker->work_items = NULL;
+  } else {
     it->prev->next = it->next;
-  if (it->next)
     it->next->prev = it->prev;
-  if (worker->work_items == it)
-    worker->work_items = it->next;
+  }
 }
 
 static void set_work_item_status(cl_command_queue_work_item it, cl_int status)
@@ -97,13 +107,12 @@ static cl_int check_work_item_ready(cl_command_queue_work_item item)
 static void *worker_thread_function(void *Arg)
 {
 
-#if 0
   cl_command_queue_worker worker = (cl_command_queue_worker)Arg;
   CL_MUTEX_LOCK(&worker->mutex);
   cl_uint last_cookie = worker->cookie;
-  cl_command_queue_work_item it;
+  cl_command_queue_work_item it, start_it;
 
-  while (worker->quit == false) {
+  while (worker->quit == CL_FALSE) {
     while (worker->work_items == NULL || last_cookie == worker->cookie) {
       CL_COND_WAIT(&worker->cond, &worker->mutex);
       if (worker->quit) {
@@ -119,16 +128,18 @@ static void *worker_thread_function(void *Arg)
 
       /* Get the first available one and execute it. */
       it = worker->work_items;
-      while(it) {
+      start_it = it;
+      while(it && it != start_it) {
         if ((is_ready = check_work_item_ready(it)) != 0) { // error or ready
           ready_one = it;
           delete_one_work_item(worker, it);
+          it->queued = CL_FALSE;
           break;
         }
         it = it->next;
       }
 
-      if (readyOne == NULL) { // nothing new is ready, wait again.
+      if (ready_one == NULL) { // nothing new is ready, wait again.
         assert(is_ready == 0);
         continue;
       }
@@ -141,6 +152,7 @@ static void *worker_thread_function(void *Arg)
       if (is_ready < 0) // Error happend, just cancel.
         set_work_item_status(ready_one, -1);
 
+#if 0
       if (ready_one->status == CL_QUEUE) {
         if (readyOne->submit() == CL_SUCCESS) {
           set_work_item_status(ready_one, CL_SUBMITTED);
@@ -164,8 +176,9 @@ static void *worker_thread_function(void *Arg)
           set_work_item_status(ready_one, -1);
         }
       }
+#endif
 
-      fffffffffffffff(readyOne);
+      cl_enqueue_destroy_work_item(worker->queue, ready_one);
       CL_MUTEX_LOCK(&worker->mutex);
       worker->in_exec = CL_FALSE;
       worker->cookie++; // something has changed.
@@ -175,13 +188,45 @@ static void *worker_thread_function(void *Arg)
   }
 
   CL_MUTEX_UNLOCK(&worker->mutex);
-#endif
 
   return NULL;
 }
 
+LOCAL cl_int cl_enqueue_queue_work_item(cl_command_queue queue, cl_command_queue_work_item item)
+{
+  cl_command_queue_worker worker;
+
+  assert(queue);
+  assert(item);
+  assert(item->queue == queue); //Should belong to this queue.
+  if (item->event) {
+    assert(item->event->queue == queue); //Should belong to this queue.
+  }
+
+  worker = queue->worker;
+
+  CL_MUTEX_LOCK(&worker->mutex);
+
+  if (worker->quit) { // already destroy the queue?
+    CL_MUTEX_UNLOCK(&worker->mutex);
+    return CL_INVALID_COMMAND_QUEUE;
+  }
+
+  if (item->status == CL_COMPLETE) {
+    CL_MUTEX_UNLOCK(&worker->mutex);
+    return CL_SUCCESS;
+  }
+
+  insert_one_work_item(worker, item);
+  item->queued = CL_TRUE;
+  worker->cookie++;
+  CL_COND_BROADCAST(&worker->cond);
+  CL_MUTEX_UNLOCK(&worker->mutex);
+  return CL_SUCCESS;
+}
+
 LOCAL cl_command_queue_work_item cl_enqueue_create_work_item(cl_command_queue queue,
-    cl_uint num_events_in_wait_list, const cl_event *event_wait_list, cl_event *event)
+    cl_uint num_events_in_wait_list, const cl_event *event_wait_list, cl_event event)
 {
   cl_command_queue_work_item it;
   int i;
@@ -219,9 +264,29 @@ LOCAL cl_command_queue_work_item cl_enqueue_create_work_item(cl_command_queue qu
   }
 
   it->status = CL_QUEUED;
+  it->queued = CL_FALSE;
   return it;
 }
 
+LOCAL void cl_enqueue_destroy_work_item(cl_command_queue queue, cl_command_queue_work_item item)
+{
+  int i;
+  assert(item->queue = queue);
+  assert(item->queued == CL_FALSE);
+
+  queue->device->driver->destroy_work_item(queue, item);
+
+  if (item->depend_events) {
+    assert(item->depend_event_num > 0);
+    for (i = 0; i < item->depend_event_num; i++) {
+      cl_release_event(item->depend_events[i]);
+    }
+  }
+
+  if (item->event)
+    cl_release_event(item->event);
+}
+
 LOCAL cl_int cl_command_queue_worker_init(cl_command_queue queue)
 {
   cl_command_queue_worker worker = NULL;
diff --git a/libclapi/cl_internals.h b/libclapi/cl_internals.h
index 7290f33d..eaa728dd 100644
--- a/libclapi/cl_internals.h
+++ b/libclapi/cl_internals.h
@@ -487,5 +487,11 @@ extern void cl_release_command_queue(cl_command_queue queue);
 extern cl_int cl_retain_event(cl_event e);
 extern cl_int cl_event_get_status(cl_event event);
 extern cl_int cl_event_set_status(cl_event event, cl_int status);
+extern cl_int cl_enqueue_queue_work_item(cl_command_queue queue, cl_command_queue_work_item item);
+extern cl_command_queue_work_item cl_enqueue_create_work_item(cl_command_queue queue,
+    cl_uint num_events_in_wait_list, const cl_event *event_wait_list, cl_event event);
+extern void cl_enqueue_destroy_work_item(cl_command_queue queue, cl_command_queue_work_item item);
+extern cl_int cl_command_queue_worker_init(cl_command_queue queue);
+extern void cl_command_queue_worker_destroy(cl_command_queue queue);
 
 #endif /* __CL_INTERNALS_H__ */