/* KInterbasDB Python Package - Implementation of ThreadSafeFIFOQueue
 *
 * Version 3.3
 *
 * The following contributors hold Copyright (C) over their respective
 * portions of code (see license.txt for details):
 *
 * [Original Author (maintained through version 2.0-0.3.1):]
 *   1998-2001 [alex]  Alexander Kuznetsov   <alexan@users.sourceforge.net>
 * [Maintainers (after version 2.0-0.3.1):]
 *   2001-2002 [maz]   Marek Isalski         <kinterbasdb@maz.nu>
 *   2002-2007 [dsr]   David Rushby          <woodsplitter@rocketmail.com>
 * [Contributors:]
 *   2001      [eac]   Evgeny A. Cherkashin  <eugeneai@icc.ru>
 *   2001-2002 [janez] Janez Jere            <janez.jere@void.si>            */

#include "_kisupport_threadsafe_fifo_queue.h"

static int ThreadSafeFIFOQueue_cancel(ThreadSafeFIFOQueue *);

static int ThreadSafeFIFOQueue_init(ThreadSafeFIFOQueue *self) {
  /* !! THIS METHOD IS NOT THREAD-SAFE !!
   *    (Obviously--it's a constructor.) */
  boolean init_lock = FALSE;
  boolean init_not_empty = FALSE;

  self->cancelled = FALSE;
  self->closed = FALSE;
  self->head = NULL;
  self->tail = NULL;

  if (Mutex_init(&self->lock) != 0) { goto fail; }
  init_lock = TRUE;

  #ifdef PLATFORM_WINDOWS
    /* Manually reset event, initially non-signalled: */
    self->not_empty = CreateEvent(NULL, TRUE, FALSE, NULL);
    if (self->not_empty == INVALID_HANDLE_VALUE) { goto fail; }
  #else
    if (pthread_cond_init(&self->not_empty, NULL) != 0) { goto fail; }
  #endif
  init_not_empty = TRUE;

  return 0;

  fail:
    if (init_lock) {
      Mutex_close(&self->lock);
    }

    if (init_not_empty) {
      #ifdef PLATFORM_WINDOWS
        assert (self->not_empty != INVALID_HANDLE_VALUE);
        CloseHandle(self->not_empty);
        self->not_empty = INVALID_HANDLE_VALUE;
      #else
        pthread_cond_destroy(&self->not_empty);
      #endif
    }

    self->closed = TRUE;

    return -1;
} /* ThreadSafeFIFOQueue_init */

static int ThreadSafeFIFOQueue_close(ThreadSafeFIFOQueue *self) {
  /* !! THIS METHOD IS NOT THREAD-SAFE !!
   *    (It releases the synchronization objects, so it couldn't possibly
   *     be.) */
  assert (!self->closed);

  if (ThreadSafeFIFOQueue_cancel(self) != 0) { goto fail; }
  assert (self->cancelled);

  if (Mutex_close(&self->lock) != 0) { goto fail; }

  #ifdef PLATFORM_WINDOWS
    assert (self->not_empty != INVALID_HANDLE_VALUE);
    if (CloseHandle(self->not_empty) == 0) {
      /* MSDN says, "If the function fails, the return value is zero." */
      goto fail;
    } else {
      self->not_empty = INVALID_HANDLE_VALUE;
    }
  #else
    if (pthread_cond_destroy(&self->not_empty) != 0) { goto fail; }
  #endif

  self->closed = TRUE;
  return 0;

  fail:
    /* Even though this close attempt failed, the failure was so fundamental
     * that we can't expect to do better next time. */
    self->closed = TRUE;
    return -1;
} /* ThreadSafeFIFOQueue_close */

static LONG_LONG _ThreadSafeFIFOQueue_delete_internal_container_if_necessary(
    ThreadSafeFIFOQueue *self
  )
{ /* !! THIS METHOD IS NOT THREAD-SAFE !! */
  LONG_LONG n_items_flushed = 0;

  QueueNode *cur_node = (QueueNode *) self->head;
  while (cur_node != NULL) {
    QueueNode *next_node = (QueueNode *) cur_node->next;

    assert (cur_node->payload_del_func != NULL);
    cur_node->payload_del_func((void *) cur_node->payload);
    kimem_plain_free(cur_node);

    cur_node = next_node;
    ++n_items_flushed;
  }

  self->head = NULL;
  self->tail = NULL;

  #ifdef PLATFORM_WINDOWS
    assert (self->not_empty != INVALID_HANDLE_VALUE);
    ResetEvent(self->not_empty);
    assert (self->not_empty != INVALID_HANDLE_VALUE);
  #else
    /* No action is necessary on POSIX because we're using a condition variable
     * which offers atomic release-and-wait somewhat similar to the WinNT 4.0+
     * SignalObjectAndWait, so there's no need for a manually reset synch
     * obj. */
  #endif

  return n_items_flushed;
} /* _ThreadSafeFIFOQueue_delete_internal_container_if_necessary */

static int ThreadSafeFIFOQueue_flush(ThreadSafeFIFOQueue *self,
    LONG_LONG *n_items_flushed
  )
{
  int res = -1;

  if (Mutex_lock(&self->lock) != 0) { goto exit; }
  /* Critical section within these brackets: */
  {
    if (self->cancelled) { goto unlock; }

    *n_items_flushed =
      _ThreadSafeFIFOQueue_delete_internal_container_if_necessary(self);
    res = 0;
  }
  unlock:
    if (Mutex_unlock(&self->lock) != 0) {
      res = -1;
      *n_items_flushed = -1;
      goto exit;
    }
    /* Fall through to exit: */

  exit:
    return res;
} /* ThreadSafeFIFOQueue_flush */

static int ThreadSafeFIFOQueue_cancel(ThreadSafeFIFOQueue *self) {
  int res = -1;

  if (Mutex_lock(&self->lock) != 0) { goto exit; }
  /* Critical section within these brackets: */
  {
    if (self->cancelled) {
      res = 0; /* If already cancelled, that's okay. */
      goto unlock;
    }

    self->cancelled = TRUE;

    _ThreadSafeFIFOQueue_delete_internal_container_if_necessary(self);
    assert (self->head == NULL);

    /* Even though we just removed all available items, it's necessary to wake
     * any waiting threads because queue cancellation is a "fatal" operation
     * (unlike queue flush). */
    #ifdef PLATFORM_WINDOWS
      SetEvent(self->not_empty);
    #else
      pthread_cond_broadcast(&self->not_empty);
    #endif

    res = 0;
  }
  unlock:
    if (Mutex_unlock(&self->lock) != 0) { goto exit; }

  exit:
    return res;
} /* ThreadSafeFIFOQueue_cancel */

static boolean ThreadSafeFIFOQueue_is_cancelled(ThreadSafeFIFOQueue *self) {
  /* Obviously, if the result is FALSE, there's no guarantee that it will still
   * be valid by the time it's returned to the caller. */
  return self->cancelled;
} /* ThreadSafeFIFOQueue_is_cancelled */

static int ThreadSafeFIFOQueue_put(ThreadSafeFIFOQueue *self,
    void *payload, QueueNodeDelFunc payload_del_func
  )
{
  int res = -1;

  QueueNode *node = (QueueNode *) kimem_plain_malloc(sizeof(QueueNode));
  if (node == NULL) { goto exit; }

  node->payload = payload;
  node->payload_del_func = payload_del_func;
  node->next = NULL;

  if (Mutex_lock(&self->lock) != 0) { goto exit; }
  /* Critical section within these brackets: */
  {
    if (self->cancelled) { goto unlock; }

    if (self->head == NULL) {
      assert (self->tail == NULL);

      self->head = self->tail = node;
    } else {
      assert (self->tail != NULL);

      self->tail->next = node;
      self->tail = node;
    }
    /* node has been attached to the ThreadSafeFIFOQueue's container; we're no
     * longer responsible for its memory: */
    node = NULL;

    /* On Windows, we're using the "broadcast model" with a manually reset
     * Event so as to avoid the lost notification problem without needing
     * SignalObjectAndWait (available on >= NT 4.0 only) or a timed-semi-busy-
     * wait loop.
     * On POSIX, pthread_cond_wait guarantees the atomicity of the
     * mutex-release-and-condition-wait, and we *know* that only one item will
     * be added to the queue at a time, so the lost notification problem
     * doesn't exist, and we can use the more efficient "signal model". */
    #ifdef PLATFORM_WINDOWS
      assert (self->not_empty != INVALID_HANDLE_VALUE);
      SetEvent(self->not_empty);
    #else
      pthread_cond_signal(&self->not_empty);
    #endif

    res = 0;
  }
  unlock:
    if (Mutex_unlock(&self->lock) != 0) {
      res = -1;
      goto exit;
    }

  exit:
    if (res != 0) {
      if (node != NULL) {
        /* Since this method didn't succeed, we never assumed responsibility
         * for the payload's memory. */
        kimem_plain_free(node);
      }
    }

    return res;
} /* ThreadSafeFIFOQueue_put */

static WaitResult ThreadSafeFIFOQueue_get(ThreadSafeFIFOQueue *self,
    long timeout_millis, void **store_fetched_payload_here
  )
{
  QueueNode *node = NULL;
  WaitResult res = WR_WAIT_ERROR;

  #ifdef PLATFORM_WINDOWS
    #define _PLAT_WAIT_OK       WAIT_OBJECT_0
    #define _PLAT_WAIT_TIMEOUT  WAIT_TIMEOUT
    DWORD wait_res = _PLAT_WAIT_OK;
  #else
    #define _PLAT_WAIT_OK       0
    #define _PLAT_WAIT_TIMEOUT  ETIMEDOUT
    int wait_res = _PLAT_WAIT_OK;

    struct timespec abstime;
  #endif

  const boolean wait_is_timed = (boolean)
    (timeout_millis != WAIT_INFINITELY_LONG);
  boolean timed_out = FALSE;

  #ifndef PLATFORM_WINDOWS
    if (wait_is_timed) {
      millis_into_future_to_abstime(timeout_millis, &abstime);
    }
  #endif

  if (Mutex_lock(&self->lock) != 0) { goto exit; }

  while (!self->cancelled && self->head == NULL && !timed_out) {
    #ifdef PLATFORM_WINDOWS
      Mutex_unlock(&self->lock);

      assert (self->not_empty != INVALID_HANDLE_VALUE);
      wait_res = WaitForSingleObject(self->not_empty,
          (wait_is_timed ? timeout_millis : INFINITE)
        );
      assert (self->not_empty != INVALID_HANDLE_VALUE);

      Mutex_lock(&self->lock);
    #else
      if (!wait_is_timed) {
        wait_res = pthread_cond_wait(&self->not_empty, &self->lock);
      } else {
        wait_res = pthread_cond_timedwait(&self->not_empty, &self->lock,
            &abstime
          );
      }
    #endif

    timed_out = (boolean) (wait_res == _PLAT_WAIT_TIMEOUT);
  }

  if (self->cancelled) {
    res = WR_WAIT_CANCELLED;
    goto fail_and_unlock;
  } else if (timed_out) {
    res = WR_WAIT_TIMEOUT;
    goto unlock;
  } else if (wait_res != _PLAT_WAIT_OK) {
    res = WR_WAIT_ERROR;
    goto fail_and_unlock;
  }

  node = (QueueNode *) self->head;
  assert (node != NULL);

  self->head = self->head->next;

  if (self->head == NULL) {
    #ifdef PLATFORM_WINDOWS
      ResetEvent(self->not_empty);
    #else
      /* No action is necessary on POSIX because we're using a condition
       * variable which offers atomic release-and-wait somewhat similar to the
       * WinNT 4.0+ SignalObjectAndWait, so there's no need for a manually
       * reset synch obj. */
    #endif
  }

  if (node == self->tail) {
    /* self->head and self->tail were the same (meaning there was only one
     * item available). */
    assert (self->head == NULL);
    self->tail = NULL;
  }

  res = WR_WAIT_OK;
  /* Fall through to unlock: */

  unlock:
    Mutex_unlock(&self->lock);
    goto exit;

  fail_and_unlock:
    assert (res == WR_WAIT_ERROR || res == WR_WAIT_CANCELLED);
    Mutex_unlock(&self->lock);
    /* Fall through to exit: */

  exit:
    if (res == WR_WAIT_ERROR) {
      if (node != NULL) {
        assert (node->payload_del_func != NULL);
        node->payload_del_func((void *) node->payload);
        kimem_plain_free(node);
      }
    } else {
      if (node != NULL) {
        *store_fetched_payload_here = (void *) node->payload;
        kimem_plain_free(node);
      } else {
        *store_fetched_payload_here = NULL;
      }
    }

    return res;
} /* ThreadSafeFIFOQueue_get */