multi-threading-raw

other

(in-package "ACL2")

other

(defstruct (atomically-modifiable-counter (:constructor make-atomically-modifiable-counter-raw))
  (val 0)
  (lock (make-mutex :name "counter lock")))

make-atomically-modifiable-counterfunction

(defun make-atomically-modifiable-counter
  (initial-value)
  (make-atomically-modifiable-counter-raw :val initial-value))

define-atomically-modifiable-countermacro

(defmacro define-atomically-modifiable-counter
  (name initial-value)
  `(defvar ,NAME
    (make-atomically-modifiable-counter ,INITIAL-VALUE)))

atomically-modifiable-counter-readmacro

(defmacro atomically-modifiable-counter-read
  (counter)
  `(atomically-modifiable-counter-val ,COUNTER))

atomic-incfmacro

(defmacro atomic-incf
  (x)
  `(with-recursive-lock ((atomically-modifiable-counter-lock ,X))
    nil
    (incf (atomically-modifiable-counter-val ,X))))

atomic-incf-multiplemacro

(defmacro atomic-incf-multiple
  (counter count)
  `(with-recursive-lock ((atomically-modifiable-counter-lock ,COUNTER))
    nil
    (incf (atomically-modifiable-counter-val ,COUNTER) ,COUNT)))

atomic-decfmacro

(defmacro atomic-decf
  (x)
  `(with-recursive-lock ((atomically-modifiable-counter-lock ,X))
    nil
    (decf (atomically-modifiable-counter-val ,X))))

lockpfunction

(defun lockp (x) (typep x 'mutex))

reset-lockmacro

(defmacro reset-lock
  (bound-symbol)
  `(setq ,BOUND-SYMBOL (make-lock ,(SYMBOL-NAME BOUND-SYMBOL))))

run-threadfunction

(defun run-thread
  (name fn-symbol &rest args)
  (make-thread (lambda nil (apply fn-symbol args))
    :name name))

interrupt-threadfunction

(defun interrupt-thread
  (thread function &rest args)
  (if args
    (error "Passing arguments to interrupt-thread not supported in SBCL.")
    (interrupt-thread thread function)))

kill-threadfunction

(defun kill-thread
  (thread)
  (terminate-thread thread))

all-threadsfunction

(defun all-threads nil (list-all-threads))

current-threadfunction

(defun current-thread nil *current-thread*)

thread-waitfunction

(defun thread-wait
  (fn &rest args)
  (loop while
    (not (apply fn args))
    do
    (sleep 0.05)))

with-potential-timeoutmacro

(defmacro with-potential-timeout
  (body &key timeout)
  `(if ,TIMEOUT
    (handler-case (with-deadline (:seconds ,TIMEOUT)
        ,BODY)
      (timeout nil))
    ,BODY))

make-condition-variablefunction

(defun make-condition-variable
  nil
  (make-waitqueue))

signal-condition-variablemacro

(defmacro signal-condition-variable
  (cv)
  `(condition-notify ,CV))

broadcast-condition-variablemacro

(defmacro broadcast-condition-variable
  (cv)
  `(condition-broadcast ,CV)
  nil)

wait-on-condition-variablefunction

(defun wait-on-condition-variable
  (cv lock &key timeout)
  (with-potential-timeout (condition-wait cv lock)
    :timeout timeout)
  t)

other

(defstruct acl2-semaphore
  (lock (make-mutex))
  (cv (make-waitqueue))
  (count 0))

make-semaphorefunction

(defun make-semaphore
  (&optional name)
  (declare (ignore name))
  (make-acl2-semaphore))

semaphorepfunction

(defun semaphorep
  (semaphore)
  (and (acl2-semaphore-p semaphore)
    (typep (acl2-semaphore-lock semaphore) 'mutex)
    (typep (acl2-semaphore-cv semaphore) 'waitqueue)
    (integerp (acl2-semaphore-count semaphore))))

make-semaphore-notificationfunction

(defun make-semaphore-notification
  nil
  (make-array 1 :initial-element nil))

semaphore-notification-statusfunction

(defun semaphore-notification-status
  (semaphore-notification-object)
  (aref semaphore-notification-object 0))

clear-semaphore-notification-statusfunction

(defun clear-semaphore-notification-status
  (semaphore-notification-object)
  (setf (aref semaphore-notification-object 0) nil))

set-semaphore-notification-statusfunction

(defun set-semaphore-notification-status
  (semaphore-notification-object)
  (setf (aref semaphore-notification-object 0) t))

signal-semaphorefunction

(defun signal-semaphore
  (semaphore)
  (with-lock-raw (acl2-semaphore-lock semaphore)
    (without-interrupts (incf (acl2-semaphore-count semaphore))
      (signal-condition-variable (acl2-semaphore-cv semaphore))))
  nil)

wait-on-semaphorefunction

(defun wait-on-semaphore
  (semaphore &key notification timeout)
  (let ((received-signal nil))
    (with-lock-raw (acl2-semaphore-lock semaphore)
      (unwind-protect-disable-interrupts-during-cleanup (with-potential-timeout (progn (loop while
              (<= (acl2-semaphore-count semaphore) 0)
              do
              (wait-on-condition-variable (acl2-semaphore-cv semaphore)
                (acl2-semaphore-lock semaphore)))
            (setq received-signal t)
            t)
          :timeout timeout)
        (if received-signal
          (progn (decf (acl2-semaphore-count semaphore))
            (when notification
              (set-semaphore-notification-status notification)))
          (signal-condition-variable (acl2-semaphore-cv semaphore)))))
    (if timeout
      received-signal
      t)))

other

(defvar *throwable-worker-thread* nil)

throw-all-threads-in-listfunction

(defun throw-all-threads-in-list
  (thread-list)
  (if (endp thread-list)
    nil
    (progn (handler-case (interrupt-thread (car thread-list)
          (function (lambda nil
              (when *throwable-worker-thread*
                (throw :worker-thread-no-longer-needed nil)))))
        (interrupt-thread-error nil))
      (throw-all-threads-in-list (cdr thread-list)))))

kill-all-threads-in-listfunction

(defun kill-all-threads-in-list
  (thread-list)
  (if (endp thread-list)
    nil
    (progn (kill-thread (car thread-list))
      (kill-all-threads-in-list (cdr thread-list)))))

thread-namefunction

(defun thread-name (thread) (thread-name thread))

other

(defconstant *worker-thread-name* "Worker thread")

worker-threads1function

(defun worker-threads1
  (threads)
  (cond ((endp threads) nil)
    ((equal (thread-name (car threads)) *worker-thread-name*) (cons (car threads) (worker-threads1 (cdr threads))))
    (t (worker-threads1 (cdr threads)))))

worker-threadsfunction

(defun worker-threads nil (worker-threads1 (all-threads)))

all-worker-threads-are-deadfunction

(defun all-worker-threads-are-dead
  nil
  (<= (length (all-threads)) 1))

all-worker-threads-are-dead-or-resetfunction

(defun all-worker-threads-are-dead-or-reset
  nil
  (null (worker-threads)))

send-die-to-worker-threadsfunction

(defun send-die-to-worker-threads
  nil
  (throw-all-threads-in-list (worker-threads))
  (let ((round 0))
    (loop do
      (when (equal (mod round 10) 0)
        (throw-all-threads-in-list (worker-threads)))
      (sleep 0.03)
      (incf round)
      while
      (not (all-worker-threads-are-dead-or-reset)))))

kill-all-worker-threadsfunction

(defun kill-all-worker-threads
  nil
  (kill-all-threads-in-list (worker-threads))
  (thread-wait 'all-worker-threads-are-dead))

core-count-rawfunction

(defun core-count-raw
  (&optional (ctx nil) default)
  (let* ((count0 (getenv$-raw "ACL2_CORE_COUNT")) (count (and (stringp count0) (string-trim '(#\  #\	) count0))))
    (cond ((and count (not (equal count ""))) (multiple-value-bind (value posn)
          (parse-integer count :junk-allowed t)
          (cond ((and (integerp value)
               (< 0 value)
               (equal posn (length count))) value)
            (t (error "Invalid value for environment variable ~
                             ACL2_CORE_COUNT: ~a"
                count0)))))
      (t (if ctx
          (error "It is illegal to call cpu-core-count in this Common ~
                       Lisp implementation.")
          (or default 16))))))

other

(defvar *core-count*
  (core-count-raw)
  "The total number of CPU cores in the system.")

other

(defvar *unassigned-and-active-work-count-limit*
  (* 4 *core-count*))

other

(defconstant *max-idle-thread-count* (* 2 *core-count*))