CL concurrent worker functions

workers.lisp

(defpackage workers
  (:use :cl)
  (:export #:list->simple-cqueue
           #:simple-cqueue->list
           #:join-workers
           #:join-consumers
           #:wmap
           #:wfilter
           #:wforeach))

(in-package :workers)

(defun list->simple-cqueue (list)
  (let ((q (queues:make-queue :simple-cqueue)))
    (loop :for it :in list :do (queues:qpush q it))
    q))

(defun simple-cqueue->list (q)
  (loop :with it :and ok
        :do (setf (values it ok) (queues:qpop q))
        :while ok :collect it))

(defmacro join-workers (workers &body body)
  `(mapcar #'bt:join-thread
    (loop :repeat ,workers
          :collect (bt:make-thread (lambda () ,@body)))))

(defmacro queue-consumer (q &body proc)
  `(loop :with it :and ok
         :do (setf (values it ok) (queues:qpop ,q))
         :while ok :do (funcall ,(car proc) it)))

(defmacro join-consumers (workers q &body proc)
  `(join-workers ,workers (queue-consumer ,q ,(car proc))))

(defun wmap (workers proc seq)
  (let ((in (list->simple-cqueue seq))
        (out (queues:make-queue :simple-cqueue)))
    (join-consumers workers in (lambda (it) (queues:qpush out (funcall proc it))))
    (simple-cqueue->list out)))

(defun wfilter (workers proc seq)
  (let ((in (list->simple-cqueue seq))
        (out (queues:make-queue :simple-cqueue)))
    (join-consumers workers in (lambda (it) (when (funcall proc it) (queues:qpush out it))))
    (simple-cqueue->list out)))

(defun wforeach (workers proc seq)
  (let ((in (list->simple-cqueue seq)))
    (join-consumers workers in (lambda (it) (funcall proc it)))))