MULTIPLE-THREADS

multiple-threads.asd

(defsystem "multiple-threads"
  :description "Multithreading Utility Library"
  :version "1.0"
  :author "Chris Dusto <cricket_lover@live.com>"
  :depends-on ("common-structures")
  :components ((:file "multiple-threads")))

multiple-threads.lisp

(defpackage :multiple-threads
  (:nicknames :mt)
  (:use :common-lisp :common-structures :sb-thread)
  (:export :with-multiple-threads))
(in-package :multiple-threads)
(defmacro with-multiple-threads
  ((n-threads worker-fn &key add-work get-work n-commands finished wait-work)
   &body body)
  (let ((work (gensym "work-"))
        (thread-id (gensym "thread-id-"))
        (command-buf (gensym "command-buf-"))
        (command-lock (gensym "command-lock-"))
        (command-queue (gensym "command-queue-"))
        (result-buf (gensym "result-buf-"))
        (result-lock (gensym "result-lock-"))
        (result-queue (gensym "result-queue-"))
        (n-active (gensym "n-active-"))
        (helper-loop-block-name (gensym "outer-loop-"))
        (is-finished (or finished (gensym "is-finished-"))))
    `(let ((,command-buf (make-instance 'cs:queue))
           (,command-lock (sb-thread:make-mutex))
           (,command-queue (sb-thread:make-waitqueue))
           (,result-buf nil)
           (,result-lock (sb-thread:make-mutex))
           (,result-queue (sb-thread:make-waitqueue))
           (,is-finished nil)
           (,n-active 0))
       (dotimes (,thread-id ,n-threads)
         (sb-thread:make-thread
           (lambda (,thread-id)
             (sb-thread:with-mutex
               (,result-lock)
               (incf ,n-active))
             (unwind-protect
               (block ,helper-loop-block-name
                 (loop
                   (let ((,work
                           (sb-thread:with-mutex
                             (,command-lock)
                             (loop
                               (let ((command (cs:buffer ,command-buf)))
                                 (when command
                                   (return command)))
                               (when ,is-finished
                                 (return-from ,helper-loop-block-name))
                               (sb-thread:condition-wait
                                 ,command-queue ,command-lock)))))
                     (unless ,work
                       (return))
                     (let ((result
                             (funcall ,worker-fn ,thread-id ,work)))
                       (sb-thread:with-mutex
                         (,result-lock)
                         (push (cons ,work result) ,result-buf)
                         (sb-thread:condition-notify ,result-queue))))))
               ; FIXME Why is this _needed_ to stop deadlocks?
               ; TODO Make a testing program for this BUG!
               (sb-thread:with-mutex
                 (,command-lock)
                 (sb-thread:condition-notify ,command-queue))
               (sb-thread:with-mutex
                 (,result-lock)
                 (decf ,n-active)
                 (sb-thread:condition-notify ,result-queue))))
           :arguments (list ,thread-id)))
       (sb-thread:with-mutex
         (,result-lock)
         (loop
           (flet
             ,(remove-if
                #'null
                (list
                  (when add-work
                    `(,add-work
                       (work)
                       (when work
                         (sb-thread:with-mutex
                           (,command-lock)
                           (setf (cs:buffer ,command-buf) work)
                           (sb-thread:condition-notify ,command-queue)))))
                  (when finished
                    `((setf ,finished)
                      (v)
                      (setf ,finished v)
                      (when v
                        (sb-thread:with-mutex
                          (,command-lock)
                          (sb-thread:condition-notify ,command-queue)))
                      v))
                  (when get-work
                    `(,get-work () (pop ,result-buf)))
                  (when n-commands
                    `(,n-commands () (cs:n-elems ,command-buf)))
                  (when wait-work
                    `(,wait-work () (sb-thread:condition-wait ,result-queue ,result-lock)))))
             ,@body)
           (when (zerop ,n-active)
             (return)))
         nil))))

test.lisp

(require :multiple-threads)
(let ((my-iter 20))
  (mt:with-multiple-threads
    (4 (lambda (tid work-order)
         (declare (ignore tid))
         (sleep (random 3.0))
         (apply #'+ work-order))
     :add-work add-work
     :get-work get-work
     :n-commands n-cmds
     :finished finished
     :wait-work wait-work)
    (unless finished
      (loop while (and (< (n-cmds) 5)
                       (plusp my-iter)) do
            (decf my-iter)
            (add-work
              (loop for i from 0 below 3 collect (random 10))))
      (when (zerop my-iter)
        (setf finished t)))
    (wait-work)
    (loop for my-work = (get-work)
          while my-work
          do (format t "~S -> ~S~%" (car my-work) (cdr my-work)))))