path: root/libclapi/cl_enqueue.c

/*
 * Copyright © 2012 Intel Corporation
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library. If not, see <http://www.gnu.org/licenses/>.
 *
 */
#include <stdlib.h>
#include <assert.h>
#include <malloc.h>
#include <pthread.h>
#include <string.h>
#include "cl_mutex.h"
#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 {
  cl_command_queue queue;
  pthread_t tid;
  pthread_cond_t cond;
  pthread_mutex_t mutex;
  cl_uint cookie;
  cl_bool quit;
  cl_bool in_exec;
  cl_command_queue_work_item work_items;
} _cl_command_queue_worker;
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)
{
  if (worker->work_items == NULL) {
    worker->work_items = it;
    it->prev = it->next = 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;
  }
}
/* 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 == it->next) { // Last one
    assert(it->prev == it);
    worker->work_items = NULL;
  } else {
    it->prev->next = it->next;
    it->next->prev = it->prev;
  }
}

static void set_work_item_status(cl_command_queue_work_item it, cl_int status)
{
  it->status = status;
  if (it->event)
    cl_event_set_status(it->event, status);
}


/* Return 0 means still not ready, 1 means ready, -1 means err and need to cancel. */
static cl_int check_work_item_ready(cl_command_queue_work_item item)
{
  int i;
  cl_event e;
  cl_int ret_val;
  cl_int status;

  if (item->depend_event_num == 0)
    return 1;

  assert(item->depend_events);
  ret_val = 1;
  for (i = 0; i < item->depend_event_num; i++) {
    e = item->depend_events[i];
    status = cl_event_get_status(e);

    if (status > CL_COMPLETE) {
      ret_val = 0;
    }

    if (status < 0) {
      return -1;
    }
  }

  return ret_val;
}

static void *worker_thread_function(void *Arg)
{
  cl_int ret;
  cl_command_queue_worker worker = (cl_command_queue_worker)Arg;
  cl_uint last_cookie = worker->cookie;
  cl_command_queue_work_item it, start_it;

  CL_MUTEX_LOCK(&worker->mutex);

  while (worker->quit == CL_FALSE) {
    while (worker->work_items == NULL || last_cookie == worker->cookie) {
      CL_COND_WAIT(&worker->cond, &worker->mutex);
      if (worker->quit) {
        CL_MUTEX_UNLOCK(&worker->mutex);
        return NULL;
      }
    }

    last_cookie = worker->cookie;
    if (worker->work_items != NULL) { // Do the real jobs.
      cl_command_queue_work_item ready_one = NULL;
      cl_int is_ready = 0;

      /* Get the first available one and execute it. */
      it = worker->work_items;
      start_it = it;
      while (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 (it == start_it)
          it = NULL;
      }

      if (ready_one == NULL) { // nothing new is ready, wait again.
        assert(is_ready == 0);
        continue;
      }

      /* We execute it without lock. */
      worker->in_exec = CL_TRUE;
      pthread_mutex_unlock(&worker->mutex);

      assert(is_ready != 0);
      if (is_ready < 0) // Error happend, just cancel.
        set_work_item_status(ready_one, -1);

      if (ready_one->status == CL_QUEUED) {
        ret = ready_one->submit(ready_one);
        set_work_item_status(ready_one, ret);
      }

      if (ready_one->status == CL_SUBMITTED) {
        ret = ready_one->run(ready_one);
        set_work_item_status(ready_one, ret);
      }

      if (ready_one->status == CL_RUNNING) {
        ret = ready_one->complete(ready_one);
        set_work_item_status(ready_one, ret);
      }

      cl_enqueue_destroy_work_item(worker->queue, ready_one);
      CL_MUTEX_LOCK(&worker->mutex);
      worker->in_exec = CL_FALSE;
      worker->cookie++; // something has changed.
      /* Some event should have changed, notify the thread block on event to check status. */
      CL_COND_BROADCAST(&worker->cond);
    }
  }

  CL_MUTEX_UNLOCK(&worker->mutex);

  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_command_queue_work_item it;
  int i;

  assert(queue);
  it = CL_CALLOC(1, sizeof(_cl_command_queue_work_item));
  if (it == NULL)
    return NULL;

  cl_retain_command_queue(queue);
  it->queue = queue;

  if (num_events_in_wait_list > 0) {
    assert(event_wait_list);
    it->depend_events = CL_CALLOC(num_events_in_wait_list, sizeof(cl_command_queue_work_item));
    if (it->depend_events == NULL) {
      CL_FREE(it);
      return NULL;
    }

    for (i = 0; i < num_events_in_wait_list; i++) {
      assert(!IS_MARKER_EVENT(event_wait_list[i]));
      assert(!IS_BARRIER_EVENT(event_wait_list[i]));
      cl_retain_event(event_wait_list[i]);
      it->depend_events[i] = event_wait_list[i];
    }

    it->depend_event_num = num_events_in_wait_list;
  }

  if (event) {
    cl_retain_event(event);
    it->event = event;
    assert(cl_event_get_status(event) == CL_QUEUED);
  }

  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;
  assert(queue->worker == NULL);

  worker = CL_CALLOC(1, sizeof(_cl_command_queue_worker));
  if (worker == NULL)
    return CL_OUT_OF_HOST_MEMORY;

  worker->cookie = 1; // start from 1
  worker->quit = CL_FALSE;
  worker->in_exec = CL_FALSE;
  worker->work_items = NULL;

  CL_MUTEX_INIT(&worker->mutex);
  CL_COND_INIT(&worker->cond);

  if (pthread_create(&worker->tid, NULL, worker_thread_function, worker)) {
    printf("Can not create worker thread for queue %p...\n", queue);
    CL_MUTEX_DESTROY(&worker->mutex);
    CL_COND_DESTROY(&worker->cond);
    CL_FREE(worker);
    return CL_OUT_OF_RESOURCES;
  }

  queue->worker = worker;
  worker->queue = queue;
  return CL_SUCCESS;
}

LOCAL void cl_command_queue_worker_destroy(cl_command_queue queue)
{
  cl_command_queue_worker worker = NULL;
  assert(queue->worker != NULL);

  worker = queue->worker;
  CL_MUTEX_LOCK(&worker->mutex);
  if (worker->quit) { // already destroy the queue?
    CL_MUTEX_UNLOCK(&worker->mutex);
    assert(0);
    return;
  }

  worker->quit = 1;
  CL_COND_BROADCAST(&worker->cond);
  CL_MUTEX_UNLOCK(&worker->mutex);

  pthread_join(worker->tid, NULL);

  CL_MUTEX_DESTROY(&worker->mutex);
  CL_COND_DESTROY(&worker->cond);

  /* We will wait for finish before destroy the command queue. */
  if (worker->work_items) {
    printf("There are still some enqueued works in the queue %p when this queue is destroyed,"
           " this may cause very serious problems.\n", queue);
    assert(0);
  }

  CL_FREE(worker);
  queue->worker = NULL;
}

LOCAL cl_int cl_enqueue_wait_for_flush(cl_command_queue queue)
{
  cl_command_queue_worker worker = NULL;
  cl_command_queue_work_item it, start_it;

  assert(queue->worker != NULL);
  worker = queue->worker;

  CL_MUTEX_LOCK(&worker->mutex);


  while (1) {
    cl_bool need_to_wait = CL_FALSE;
    it = worker->work_items;
    start_it = it;

    if (worker->quit) { // already destroy the queue?
      CL_MUTEX_UNLOCK(&worker->mutex);
      return CL_INVALID_COMMAND_QUEUE;
    }

    while (it) {
      if (it->status > CL_SUBMITTED) {// Need not to lock, ready has been deleted from list
        need_to_wait = CL_TRUE;
        break;
      }

      it = it->next;
      if (it == start_it)
        it = NULL;
    }

    if (need_to_wait) {
      CL_COND_WAIT(&worker->cond, &worker->mutex);
    } else {
      break;
    }
  }

  CL_MUTEX_UNLOCK(&worker->mutex);
  return CL_SUCCESS;
}

LOCAL cl_int cl_enqueue_wait_for_finish(cl_command_queue queue)
{
  cl_command_queue_worker worker = NULL;

  assert(queue->worker != NULL);
  worker = queue->worker;

  CL_MUTEX_LOCK(&worker->mutex);

  while (1) {
    if (worker->quit) { // already destroy the queue?
      CL_MUTEX_UNLOCK(&worker->mutex);
      return CL_INVALID_COMMAND_QUEUE;
    }

    if (worker->work_items == NULL && worker->in_exec == CL_FALSE)
      break;

    CL_COND_WAIT(&worker->cond, &worker->mutex);
  }

  CL_MUTEX_UNLOCK(&worker->mutex);
  return CL_SUCCESS;
}

LOCAL void cl_enqueue_notify_event_changed(cl_command_queue queue)
{
  cl_command_queue_worker worker = NULL;

  assert(queue->worker != NULL);
  worker = queue->worker;

  CL_MUTEX_LOCK(&worker->mutex);
  if (worker->quit) { // already destroy the queue?
    CL_MUTEX_UNLOCK(&worker->mutex);
    return;
  }

  worker->cookie++;
  CL_COND_BROADCAST(&worker->cond);
  CL_MUTEX_UNLOCK(&worker->mutex);
}

LOCAL cl_int cl_enqueue_wait_for_events(cl_command_queue queue, const cl_event* events, int num)
{
  cl_command_queue_worker worker = NULL;
  cl_bool need_to_wait = CL_FALSE;
  cl_int status;
  int i;

  assert(queue->worker != NULL);
  worker = queue->worker;

  CL_MUTEX_LOCK(&worker->mutex);

  while (1) {
    if (worker->quit) { // already destroy the queue?
      CL_MUTEX_UNLOCK(&worker->mutex);
      return CL_INVALID_COMMAND_QUEUE;
    }

    need_to_wait = CL_FALSE;
    for (i = 0; i < num; i++) {
      assert(events[i]);
      assert(events[i]->queue == queue);
      assert(!IS_MARKER_EVENT(events[i]));
      assert(!IS_BARRIER_EVENT(events[i]));

      status = cl_event_get_status(events[i]);
      if (status < CL_COMPLETE) {
        CL_MUTEX_UNLOCK(&worker->mutex);
        return CL_EXEC_STATUS_ERROR_FOR_EVENTS_IN_WAIT_LIST;
      }

      if (status > CL_COMPLETE) {
        need_to_wait = CL_TRUE;
      }
    }

    if (need_to_wait) {
      CL_COND_WAIT(&worker->cond, &worker->mutex);
    } else {
      // All events are ready, OK now.
      break;
    }
  }

  CL_MUTEX_UNLOCK(&worker->mutex);
  return CL_SUCCESS;
}