concurrent-fun.lisp

(require :sb-concurrency)
(defpackage :demo (:use :cl :sb-thread :sb-concurrency :sb-ext))

(in-package :demo)

(defun make-atomic-boolean (&optional (init nil))
  (list init))

(defun atomic-boolean-get (ab)
  (barrier (:read) 
    (car ab)))

(defun atomic-boolean-set! (new-value ab)
  (compare-and-swap (car ab) (not new-value) new-value))

(defun worker-thread (q keep-running)
  (let ((name (thread-name *current-thread*)))
    (format t "Started thread ~A ~%" name)
    (loop
       (when (not (atomic-boolean-get keep-running)) 
         (return))
       (multiple-value-bind (o) (dequeue q)
         (when (not (null o))
             (format t "Thread ~A received ~A ~%" name o)))
       (sleep 0.1))
    (format t "Thread ~A shutting down ~%" name)))

(defun make-workers (q keep-running name n)
  (loop 
     for i from 1 to n
     collect (make-thread 
              (lambda ()
                (worker-thread q keep-running))
              :name (format nil "~A-worker-~A" name i))))

(defun make-threadpool-simulator (num-threads)
  (make-instance 'threadpool-simulator :num-threads num-threads))

(defclass threadpool-simulator ()
  ((keep-running 
    :initform (make-atomic-boolean t))
   (q
    :initform (make-queue))
   (num-threads
    :initarg :num-threads
    :initform (error "Must supply number of threads"))
   (threads)))

(defmethod initialize-instance :after ((simulator threadpool-simulator) &key)
  (with-slots (q keep-running num-threads threads) simulator 
    (setf threads (make-workers q
                                keep-running
                                "simulator"
                                num-threads))))

(defmethod simulator-send (o (simulator threadpool-simulator))
  (with-slots (q) simulator
    (enqueue o q)))

(defmethod simulator-stop ((simulator threadpool-simulator))
  (with-slots (keep-running threads) simulator
    (atomic-boolean-set! nil keep-running)
    (mapcar #'join-thread threads)))