contrib/sb-concurrency/queue.lisp

   1 ;;;; Lock-free FIFO queues, from "An Optimistic Approach to Lock-Free FIFO
   2 ;;;; Queues" by Edya Ladan-Mozes and Nir Shavit.
   3 ;;;;
   4 ;;;; Written by Nikodemus Siivola for SBCL.
   5 ;;;;
   6 ;;;; This software is part of the SBCL system. See the README file for
   7 ;;;; more information.
   8 ;;;;
   9 ;;;; This software is derived from the CMU CL system, which was written at
  10 ;;;; Carnegie Mellon University and released into the public domain. The
  11 ;;;; software is in the public domain and is provided with absolutely no
  12 ;;;; warranty. See the COPYING and CREDITS files for more information.
  13
  14 (in-package :sb-concurrency)
  15
  16 (defconstant +dummy+ '.dummy.)
  17
  18 (declaim (inline make-node))
  19 (defstruct node
  20   value
  21   (prev nil :type (or null node))
  22   (next nil :type (or null node)))
  23
  24 (declaim (inline %make-queue))
  25 (defstruct (queue (:constructor %make-queue (head tail name))
  26                   (:copier nil)
  27                   (:predicate queuep))
  28   "Lock-free thread safe queue."
  29   (head (error "No HEAD.") :type node)
  30   (tail (error "No TAIL.") :type node)
  31   (name nil))
  32
  33 (setf (documentation 'queuep 'function)
  34       "Returns true if argument is a QUEUE, NIL otherwise."
  35       (documentation 'queue-name 'function)
  36       "Name of a QUEUE. Can be assingned to using SETF. Queue names
  37 can be arbitrary printable objects, and need not be unique.")
  38
  39 (defun make-queue (&key name initial-contents)
  40   "Returns a new QUEUE with NAME and contents of the INITIAL-CONTENTS
  41 sequence enqueued."
  42   (let* ((dummy (make-node :value +dummy+))
  43          (queue (%make-queue dummy dummy name)))
  44     (flet ((enc-1 (x)
  45              (enqueue x queue)))
  46       (declare (dynamic-extent #'enc-1))
  47       (map nil #'enc-1 initial-contents))
  48     queue))
  49
  50 (defun enqueue (value queue)
  51   "Adds VALUE to the end of QUEUE. Returns VALUE."
  52   (let ((node (make-node :value value)))
  53     (loop for tail = (queue-tail queue)
  54           do (setf (node-next node) tail)
  55              (when (eq tail (sb-ext:compare-and-swap (queue-tail queue) tail node))
  56                (setf (node-prev tail) node)
  57                (return value)))))
  58
  59 (defun dequeue (queue)
  60   "Retrieves the oldest value in QUEUE and returns it as the primary value,
  61 and T as secondary value. If the queue is empty, returns NIL as both primary
  62 and secondary value."
  63   (tagbody
  64    :continue
  65      (let* ((head (queue-head queue))
  66             (tail (queue-tail queue))
  67             (first-node-prev (node-prev head))
  68             (val (node-value head)))
  69        (barrier (:read))
  70        (when (eq head (queue-head queue))
  71          (cond ((not (eq val +dummy+))
  72                 (if (eq tail head)
  73                     (let ((dummy (make-node :value +dummy+ :next tail)))
  74                       (when (eq tail (sb-ext:compare-and-swap (queue-tail queue)
  75                                                               tail dummy))
  76                         (setf (node-prev head) dummy))
  77                       (go :continue))
  78                     (when (null first-node-prev)
  79                       (fixList queue tail head)
  80                       (go :continue)))
  81                 (when (eq head (sb-ext:compare-and-swap (queue-head queue)
  82                                                         head first-node-prev))
  83                   ;; These assignment is not present in the paper, but are
  84                   ;; equivalent to the free(head.ptr) call there.
  85                   ;;
  86                   ;; First we unlink the HEAD from the queue -- the code in
  87                   ;; the paper leaves the dangling pointer in place.
  88                   ;;
  89                   ;; Then we NIL out the slots in HEAD to help the GC,
  90                   ;; otherwise conservativism might lead to massive chains of
  91                   ;; nodes being retained.
  92                   (setf (node-next first-node-prev) nil
  93                         (node-prev head) nil
  94                         (node-next head) nil
  95                         (node-value head) nil)
  96                   (return-from dequeue (values val t))))
  97                ((eq tail head)
  98                 (return-from dequeue (values nil nil)))
  99                ((null first-node-prev)
 100                 (fixList queue tail head)
 101                 (go :continue))
 102                (t
 103                 (sb-ext:compare-and-swap (queue-head queue)
 104                                          head first-node-prev)))))
 105      (go :continue)))
 106
 107 (defun fixlist (queue tail head)
 108   (let ((current tail))
 109     (loop while (and (eq head (queue-head queue)) (not (eq current head)))
 110           do (let ((next (node-next current)))
 111                (when (not next)
 112                  (return-from fixlist nil))
 113                (let ((nextNodePrev (node-prev next)))
 114                  (when (not (eq nextNodePrev current))
 115                    (setf (node-prev next) current))
 116                  (setf current next))))))
 117
 118 (defun list-queue-contents (queue)
 119   "Returns the contents of QUEUE as a list without removing them from the
 120 QUEUE. Mainly useful for manual examination of queue state."
 121   (let (all)
 122     (labels ((walk (node)
 123                ;; Since NEXT pointers are always right, traversing from tail
 124                ;; to head is safe.
 125                (let ((value (node-value node))
 126                      (next (node-next node)))
 127                  (unless (eq +dummy+ value)
 128                    (push value all))
 129                  (when next
 130                    (walk next)))))
 131       (walk (queue-tail queue)))
 132     all))
 133
 134 (defun queue-count (queue)
 135   "Returns the number of objects in QUEUE. Mainly useful for manual
 136 examination of queue state, and in PRINT-OBJECT methods: inefficient as it
 137 walks the entire queue."
 138   (let ((n 0))
 139     (declare (unsigned-byte n))
 140     (labels ((walk (node)
 141                (let ((value (node-value node))
 142                      (next (node-next node)))
 143                  (unless (eq +dummy+ value)
 144                    (incf n))
 145                  (when next
 146                    (walk next)))))
 147       (walk (queue-tail queue))
 148       n)))
 149
 150 (defun queue-empty-p (queue)
 151   "Returns T if QUEUE is empty, NIL otherwise."
 152   (let* ((head (queue-head queue))
 153          (tail (queue-tail queue))
 154          (val (node-value head)))
 155     (and (eq head tail) (eq val +dummy+))))