2 files changed, 67 insertions, 0 deletions
diff --git a/lib/mqueue.c b/lib/mqueue.c
index e1433245..f252586e 100644
--- a/lib/mqueue.c
+++ b/lib/mqueue.c
@@ -603,6 +603,59 @@ done:
 } ;
 
 /*------------------------------------------------------------------------------
+ * Revoke message(s)
+ *
+ * Revokes all messages, or only messages whose arg0 matches the given value.
+ * (If the given value is NULL revokes everything.)
+ *
+ * Revokes by calling mqb_dispatch_destroy().
+ *
+ * During a revoke() operation more items may be enqueued, but no other mqueue
+ * operations may be performed.  Enqueued items may promptly be revoked, except
+ * for priority items if the revoke operation has already moved past the last
+ * priority item.
+ */
+extern void
+mqueue_revoke(mqueue_queue mq, void* arg0)
+{
+  mqueue_block mqb ;
+  mqueue_block prev ;
+
+  qpt_mutex_lock(&mq->mutex) ;
+
+  prev = NULL ;
+  while (1)
+    {
+      if (prev == NULL)
+        mqb = mq->head ;
+      else
+        mqb = prev->next ;
+
+      if (mqb == NULL)
+        break ;
+
+      if ((arg0 == NULL) || (arg0 == mqb->arg0))
+        {
+          if (prev == NULL)
+            mq->head   = mqb->next ;
+          else
+            prev->next = mqb->next ;
+
+          if (mq->tail == mqb)
+            mq->tail = prev ;
+
+          qpt_mutex_unlock(&mq->mutex) ;
+            mqb_dispatch_destroy(mqb) ;
+          qpt_mutex_lock(&mq->mutex) ;
+        }
+      else
+        prev = mqb ;
+    } ;
+
+  qpt_mutex_unlock(&mq->mutex) ;
+} ;
+
+/*------------------------------------------------------------------------------
  * No longer waiting for a signal  -- does nothing if !qpthreads_enabled.
  *
  * Returns true <=> signal has been kicked
diff --git a/lib/mqueue.h b/lib/mqueue.h
index a4f5d91d..a6edb4d7 100644
--- a/lib/mqueue.h
+++ b/lib/mqueue.h
@@ -225,12 +225,18 @@ mqueue_enqueue(mqueue_queue mq, mqueue_block mqb, int priority) ;
 extern mqueue_block
 mqueue_dequeue(mqueue_queue mq, int wait, void* arg) ;
 
+extern void
+mqueue_revoke(mqueue_queue mq, void* arg0) ;
+
 extern int
 mqueue_done_waiting(mqueue_queue mq, mqueue_thread_signal mtsig) ;
 
 extern void
 mqueue_local_enqueue(mqueue_local_queue lmq, mqueue_block mqb) ;
 
+Inline mqueue_block
+mqueue_local_head(mqueue_local_queue lmq) ;
+
 extern mqueue_block
 mqueue_local_dequeue(mqueue_local_queue lmq) ;
 
@@ -257,6 +263,8 @@ extern void mqb_push_argv_u(mqueue_block mqb, mqb_uint_t u) ;
 extern void mqb_push_argv_array(mqueue_block mqb, unsigned n, void** array) ;
 
 Inline void mqb_dispatch(mqueue_block mqb, mqb_flag_t flag) ;
+Inline void mqb_dispatch_action(mqueue_block mqb) ;
+Inline void mqb_dispatch_destroy(mqueue_block mqb) ;
 
 Inline void* mqb_get_arg0(mqueue_block mqb) ;
 Inline void* mqb_get_args(mqueue_block mqb) ;
@@ -278,6 +286,12 @@ extern void** mqb_pop_argv_array(mqueue_block mqb) ;
  * The Inline functions.
  */
 
+Inline mqueue_block
+mqueue_local_head(mqueue_local_queue lmq)
+{
+  return lmq->head ;
+} ;
+
 /* Set operations.      */
 
 Inline void