Added SILC Thread Queue API

[crypto.git] / lib / silcutil / silcthreadqueue.c

diff --git a/lib/silcutil/silcthreadqueue.c b/lib/silcutil/silcthreadqueue.c
new file mode 100644 (file)
index 0000000..1ca9ef3
--- /dev/null
+++ b/lib/silcutil/silcthreadqueue.c
@@ -0,0 +1,178 @@
+/*
+
+  silcthreadqueue.c
+
+  Author: Pekka Riikonen <priikone@silcnet.org>
+
+  Copyright (C) 2008 Pekka Riikonen
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of the GNU General Public License as published by
+  the Free Software Foundation; version 2 of the License.
+
+  This program 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 General Public License for more details.
+
+*/
+
+#include "silcruntime.h"
+
+/************************** Types and definitions ***************************/
+
+/* Thread queue context */
+struct SilcThreadQueueStruct {
+  SilcDList queue;             /* The queue */
+  SilcMutex lock;              /* Queue lock */
+  SilcCond cond;               /* Condition for waiting */
+  SilcAtomic32 connected;      /* Number of connected threads */
+};
+
+/************************** SILC Thread Queue API ***************************/
+
+/* Allocate thread queue */
+
+SilcThreadQueue silc_thread_queue_alloc(void)
+{
+  SilcThreadQueue queue;
+
+  queue = silc_calloc(1, sizeof(*queue));
+  if (!queue)
+    return NULL;
+
+  SILC_LOG_DEBUG(("Allocated thread queue %p", queue));
+
+  if (!silc_mutex_alloc(&queue->lock)) {
+    silc_free(queue);
+    return NULL;
+  }
+
+  if (!silc_cond_alloc(&queue->cond)) {
+    silc_mutex_free(queue->lock);
+    silc_free(queue);
+    return NULL;
+  }
+
+  queue->queue = silc_dlist_init();
+  if (!queue->queue) {
+    silc_cond_free(queue->cond);
+    silc_mutex_free(queue->lock);
+    silc_free(queue);
+    return NULL;
+  }
+
+  silc_atomic_init32(&queue->connected, 1);
+
+  return queue;
+}
+
+/* Connect current thread to queue */
+
+void silc_thread_queue_connect(SilcThreadQueue queue)
+{
+  silc_atomic_add_int32(&queue->connected, 1);
+}
+
+/* Disconnect current thread from queue */
+
+void silc_thread_queue_disconnect(SilcThreadQueue queue)
+{
+  if (silc_atomic_sub_int32(&queue->connected, 1) > 0)
+    return;
+
+  /* Free queue */
+  SILC_LOG_DEBUG(("Free thread queue %p", queue));
+  silc_cond_free(queue->cond);
+  silc_mutex_free(queue->lock);
+  silc_dlist_uninit(queue->queue);
+  silc_atomic_uninit32(&queue->connected);
+  silc_free(queue);
+}
+
+/* Push data to queue */
+
+void silc_thread_queue_push(SilcThreadQueue queue, void *data)
+{
+  if (silc_unlikely(!data))
+    return;
+
+  SILC_LOG_DEBUG(("Push data %p to thread queue %p", data, queue));
+
+  silc_mutex_lock(queue->lock);
+  silc_dlist_start(queue->queue);
+  silc_dlist_insert(queue->queue, data);
+  silc_cond_broadcast(queue->cond);
+  silc_mutex_unlock(queue->lock);
+}
+
+/* Get data or wait if wanted or return NULL. */
+
+void *silc_thread_queue_pop(SilcThreadQueue queue, SilcBool block)
+{
+  void *data;
+
+  if (block)
+    return silc_thread_queue_timed_pop(queue, 0);
+
+  silc_mutex_lock(queue->lock);
+
+  silc_dlist_start(queue->queue);
+  data = silc_dlist_get(queue->queue);
+  if (data)
+    silc_dlist_del(queue->queue, data);
+
+  SILC_LOG_DEBUG(("Pop data %p from thread queue %p", data, queue));
+
+  silc_mutex_unlock(queue->lock);
+
+  return data;
+}
+
+/* Get data or wait for a while */
+
+void *silc_thread_queue_timed_pop(SilcThreadQueue queue,
+                                 int timeout_msec)
+{
+  void *data;
+
+  silc_mutex_lock(queue->lock);
+
+  silc_dlist_start(queue->queue);
+  while ((data = silc_dlist_get(queue->queue)) == SILC_LIST_END) {
+    if (!silc_cond_timedwait(queue->cond, queue->lock, timeout_msec))
+      break;
+    silc_dlist_start(queue->queue);
+  }
+
+  if (data)
+    silc_dlist_del(queue->queue, data);
+
+  SILC_LOG_DEBUG(("Pop data %p from thread queue %p", data, queue));
+
+  silc_mutex_unlock(queue->lock);
+
+  return data;
+}
+
+/* Pop entire queue */
+
+SilcDList silc_thread_queue_pop_list(SilcThreadQueue queue, SilcBool block)
+{
+  SilcDList list;
+
+  silc_mutex_lock(queue->lock);
+
+  if (block)
+    while (silc_dlist_count(queue->queue) == 0)
+      silc_cond_wait(queue->cond, queue->lock);
+
+  list = queue->queue;
+  queue->queue = silc_dlist_init();
+
+  silc_mutex_unlock(queue->lock);
+
+  silc_dlist_start(list);
+
+  return list;
+}