parallel-raw

(defun reset-parallelism-variables
  nil
  (send-die-to-worker-threads)
  (when *reset-core-count-too*
    (setf *core-count* (core-count-raw))
    (setf *unassigned-and-active-work-count-limit*
      (* 4 *core-count*)))
  (setf *idle-thread-count*
    (make-atomically-modifiable-counter 0))
  (setf *idle-core-count*
    (make-atomically-modifiable-counter (1- *core-count*)))
  (setf *unassigned-and-active-work-count*
    (make-atomically-modifiable-counter 1))
  (setf *total-work-count*
    (make-atomically-modifiable-counter 1))
  (setf *work-queue* nil)
  (reset-lock *work-queue-lock*)
  (setf *check-work-and-core-availability* (make-semaphore))
  (setf *check-core-availability-for-resuming*
    (make-semaphore))
  (setf *throwable-worker-thread* nil)
  (setf *total-parallelism-piece-historical-count* 0)
  (setf *reset-parallelism-variables* nil)
  t)

(defun eval-and-save-result
  (work)
  (assert work)
  (let ((result-array (parallelism-piece-result-array work)) (array-index (parallelism-piece-array-index work))
      (closure (parallelism-piece-closure work)))
    (setf (aref result-array array-index)
      (cons t (funcall closure)))))

(defun pop-work-and-set-thread
  nil
  (let ((work (with-lock *work-queue-lock*
         (when (consp *work-queue*) (pop *work-queue*)))) (thread (current-thread)))
    (when work
      (assert thread)
      (assert (parallelism-piece-thread-array work))
      (setf (aref (parallelism-piece-thread-array work)
          (parallelism-piece-array-index work))
        thread))
    work))

(defun consume-work-on-work-queue-when-there
  nil
  (catch :worker-thread-no-longer-needed (let ((*throwable-worker-thread* t) (*default-hs* nil))
      (declare (special *default-hs*))
      (loop (loop while
          (not (and *work-queue*
              (< 0 (atomically-modifiable-counter-read *idle-core-count*))))
          do
          (when (not (wait-on-semaphore *check-work-and-core-availability*
                :timeout 15))
            (throw :worker-thread-no-longer-needed nil)))
        (unwind-protect-disable-interrupts-during-cleanup (when (<= 0 (atomic-decf *idle-core-count*))
            (catch :result-no-longer-needed (let ((work nil))
                (unwind-protect-disable-interrupts-during-cleanup (progn (without-interrupts (setq work (pop-work-and-set-thread))
                      (when work (atomic-decf *idle-thread-count*)))
                    (when work
                      (eval-and-save-result work)
                      (let* ((thread-array (parallelism-piece-thread-array work)) (result-array (parallelism-piece-result-array work))
                          (array-index (parallelism-piece-array-index work))
                          (throw-siblings-when-function (parallelism-piece-throw-siblings-when-function work)))
                        (setf (aref thread-array array-index) nil)
                        (when throw-siblings-when-function
                          (funcall throw-siblings-when-function
                            (aref result-array array-index))))))
                  (when work
                    (let* ((semaphore-to-signal-as-last-act (parallelism-piece-semaphore-to-signal-as-last-act work)) (thread-array (parallelism-piece-thread-array work))
                        (array-index (parallelism-piece-array-index work)))
                      (incf *total-parallelism-piece-historical-count*)
                      (setf (aref thread-array array-index) nil)
                      (atomic-incf *idle-thread-count*)
                      (atomic-decf *total-work-count*)
                      (atomic-decf *unassigned-and-active-work-count*)
                      (assert (semaphorep semaphore-to-signal-as-last-act))
                      (signal-semaphore semaphore-to-signal-as-last-act)))))))
          (atomic-incf *idle-core-count*)
          (signal-semaphore *check-work-and-core-availability*)
          (signal-semaphore *check-core-availability-for-resuming*)))))
  (atomic-decf *idle-thread-count*))

(defun spawn-worker-threads-if-needed
  nil
  (loop while
    (< (atomically-modifiable-counter-read *idle-thread-count*)
      *max-idle-thread-count*)
    do
    (progn (atomic-incf *idle-thread-count*)
      (run-thread "Worker thread"
        'consume-work-on-work-queue-when-there))))

(defun add-work-list-to-queue
  (work-list)
  (let ((work-list-length (length work-list)))
    (with-lock *work-queue-lock*
      (setf *work-queue* (nconc *work-queue* work-list)))
    (atomic-incf-multiple *total-work-count* work-list-length)
    (atomic-incf-multiple *unassigned-and-active-work-count*
      work-list-length)
    (dotimes (i work-list-length)
      (signal-semaphore *check-work-and-core-availability*))))

(defun combine-array-results-into-list
  (result-array current-position acc)
  (if (< current-position 0)
    acc
    (combine-array-results-into-list result-array
      (1- current-position)
      (cons (cdr (aref result-array current-position)) acc))))

(defun remove-thread-array-from-work-queue-rec
  (work-queue thread-array array-positions-left)
  (cond ((eql array-positions-left 0) work-queue)
    ((atom work-queue) nil)
    ((eq thread-array
       (parallelism-piece-thread-array (car work-queue))) (progn (atomic-decf *total-work-count*)
        (atomic-decf *unassigned-and-active-work-count*)
        (assert (semaphorep (parallelism-piece-semaphore-to-signal-as-last-act (car work-queue))))
        (signal-semaphore (parallelism-piece-semaphore-to-signal-as-last-act (car work-queue)))
        (remove-thread-array-from-work-queue-rec (cdr work-queue)
          thread-array
          (1- array-positions-left))))
    (t (cons (car work-queue)
        (remove-thread-array-from-work-queue-rec (cdr work-queue)
          thread-array
          (1- array-positions-left))))))

(defun remove-thread-array-from-work-queue
  (thread-array)
  (without-interrupts (with-lock *work-queue-lock*
      (setf *work-queue*
        (remove-thread-array-from-work-queue-rec *work-queue*
          thread-array
          (length thread-array))))))

(defun terminate-siblings
  (thread-array)
  (remove-thread-array-from-work-queue thread-array)
  (throw-threads-in-array thread-array
    (1- (length thread-array))))

(defun generate-work-list-from-closure-list-rec
  (thread-array result-array
    children-done-semaphore
    closure-list
    current-position
    &optional
    throw-siblings-when-function)
  (if (atom closure-list)
    (assert (equal current-position (length thread-array)))
    (cons (make-parallelism-piece :thread-array thread-array
        :result-array result-array
        :array-index current-position
        :semaphore-to-signal-as-last-act children-done-semaphore
        :closure (car closure-list)
        :throw-siblings-when-function throw-siblings-when-function)
      (generate-work-list-from-closure-list-rec thread-array
        result-array
        children-done-semaphore
        (cdr closure-list)
        (1+ current-position)
        throw-siblings-when-function))))

(defun generate-work-list-from-closure-list
  (closure-list &optional terminate-early-function)
  (let* ((closure-count (length closure-list)) (thread-array (make-array closure-count :initial-element nil))
      (result-array (make-array closure-count :initial-element nil))
      (children-done-semaphore (make-semaphore)))
    (progn (assert (semaphorep children-done-semaphore))
      (mv (generate-work-list-from-closure-list-rec thread-array
          result-array
          children-done-semaphore
          closure-list
          0
          (if terminate-early-function
            (lambda (x)
              (when (funcall terminate-early-function (cdr x))
                (terminate-siblings thread-array)))
            nil))
        thread-array
        result-array
        children-done-semaphore))))

(defun pargs-parallelism-buffer-has-space-available
  nil
  (< (atomically-modifiable-counter-read *unassigned-and-active-work-count*)
    *unassigned-and-active-work-count-limit*))

(defun not-too-many-pieces-of-parallelism-work-already-in-existence
  nil
  (< (atomically-modifiable-counter-read *total-work-count*)
    (f-get-global 'total-parallelism-work-limit
      *the-live-state*)))

(defun parallelism-resources-available
  nil
  (and (f-get-global 'parallel-execution-enabled *the-live-state*)
    (pargs-parallelism-buffer-has-space-available)
    (not-too-many-pieces-of-parallelism-work-already-in-existence)))

(defun throw-threads-in-array
  (thread-array current-position)
  (assert thread-array)
  (when (<= 0 current-position)
    (let ((current-thread (aref thread-array current-position)))
      (when current-thread
        (interrupt-thread current-thread
          (lambda nil
            (when (aref thread-array current-position)
              (throw :result-no-longer-needed nil))))))
    (throw-threads-in-array thread-array (1- current-position))))

(defun decrement-children-left
  (children-left-ptr semaphore-notification-obj)
  (when (semaphore-notification-status semaphore-notification-obj)
    (decf (aref children-left-ptr 0))
    (clear-semaphore-notification-status semaphore-notification-obj)))

(defun wait-for-children-to-finish
  (semaphore children-left-ptr semaphore-notification-obj)
  (assert children-left-ptr)
  (when (<= 1 (aref children-left-ptr 0))
    (assert (not (semaphore-notification-status semaphore-notification-obj)))
    (unwind-protect-disable-interrupts-during-cleanup (wait-on-semaphore semaphore
        :notification semaphore-notification-obj)
      (decrement-children-left children-left-ptr
        semaphore-notification-obj))
    (wait-for-children-to-finish semaphore
      children-left-ptr
      semaphore-notification-obj)))

(defun wait-for-resumptive-parallelism-resources
  nil
  (loop while
    (<= (atomically-modifiable-counter-read *idle-core-count*)
      (- *core-count*))
    do
    (wait-on-semaphore *check-core-availability-for-resuming*))
  (atomic-incf *unassigned-and-active-work-count*)
  (atomic-decf *idle-core-count*))

(defun early-terminate-children-and-rewait
  (children-done-semaphore children-left-ptr
    semaphore-notification-obj
    thread-array)
  (when (< 0 (aref children-left-ptr 0))
    (remove-thread-array-from-work-queue thread-array)
    (throw-threads-in-array thread-array
      (1- (length thread-array)))
    (wait-for-children-to-finish children-done-semaphore
      children-left-ptr
      semaphore-notification-obj)))

(defun prepare-to-wait-for-children
  nil
  (atomic-incf *idle-core-count*)
  (signal-semaphore *check-work-and-core-availability*)
  (signal-semaphore *check-core-availability-for-resuming*)
  (atomic-decf *unassigned-and-active-work-count*))

(defun parallelize-closure-list
  (closure-list &optional terminate-early-function)
  (assert (and (consp closure-list) (cdr closure-list)))
  (let ((work-list-setup-p nil) (semaphore-notification-obj (make-semaphore-notification))
      (children-left-ptr (make-array 1 :initial-element (length closure-list))))
    (mv-let (work-list thread-array
        result-array
        children-done-semaphore)
      (generate-work-list-from-closure-list closure-list
        terminate-early-function)
      (assert (semaphorep children-done-semaphore))
      (unwind-protect-disable-interrupts-during-cleanup (progn (without-interrupts (spawn-worker-threads-if-needed)
            (setq work-list-setup-p
              (progn (add-work-list-to-queue work-list) t))
            (prepare-to-wait-for-children))
          (wait-for-children-to-finish children-done-semaphore
            children-left-ptr
            semaphore-notification-obj))
        (progn (when work-list-setup-p
            (early-terminate-children-and-rewait children-done-semaphore
              children-left-ptr
              semaphore-notification-obj
              thread-array)
            (wait-for-resumptive-parallelism-resources)
            (assert (eq (aref children-left-ptr 0) 0)))))
      (combine-array-results-into-list result-array
        (1- (length result-array))
        nil))))

(defun parallelize-fn
  (parent-fun-name arg-closures
    &optional
    terminate-early-function)
  (assert (cdr arg-closures))
  (let ((parallelize-closures-res (parallelize-closure-list arg-closures
         terminate-early-function)))
    (if (or (equal parent-fun-name 'and-list)
        (equal parent-fun-name 'or-list))
      (funcall parent-fun-name parallelize-closures-res)
      (apply parent-fun-name parallelize-closures-res))))

(defun parallelism-condition
  (gran-form-exists gran-form)
  (if gran-form-exists
    `(and (parallelism-resources-available) ,GRAN-FORM)
    '(parallelism-resources-available)))

(defun plet-doublets
  (bindings bsym n)
  (cond ((endp bindings) nil)
    (t (cons (list (caar bindings) (list 'nth n bsym))
        (plet-doublets (cdr bindings) bsym (1+ n))))))

(defun make-closures
  (bindings)
  (if (endp bindings)
    nil
    (cons `(function (lambda nil ,(CADAR BINDINGS)))
      (make-closures (cdr bindings)))))

(defun identity-list (&rest rst) rst)

(defun make-list-until-non-declare
  (remaining-list acc)
  (if (not (caar-is-declarep remaining-list))
    (mv (reverse acc) remaining-list)
    (make-list-until-non-declare (cdr remaining-list)
      (cons (car remaining-list) acc))))

(defun parse-additional-declare-forms-for-let
  (x)
  (mv-let (declare-forms body)
    (make-list-until-non-declare (cdr x) nil)
    (mv (car x) declare-forms body)))

(defun signal-semaphores
  (sems)
  (cond ((endp sems) nil)
    (t (signal-semaphore (car sems))
      (signal-semaphores (cdr sems)))))

(defun number-of-active-threads-aux
  (threads acc)
  (declare (ignore threads acc))
  0)

(defun number-of-active-threads
  nil
  (number-of-active-threads-aux (all-threads) 0))

(defun number-of-threads-waiting-on-a-child-aux
  (threads acc)
  (declare (ignore threads acc))
  0)

(defun number-of-threads-waiting-on-a-child
  nil
  (number-of-threads-waiting-on-a-child-aux (all-threads) 0))

(defun future-queue-length
  nil
  (+ (- *last-slot-saved* *last-slot-taken*)
    *threads-waiting-for-starting-core*))

(defun total-number-of-threads nil (length (all-threads)))

(defun acl2p-sum-list1
  (lst acc)
  (cond ((endp lst) acc)
    (t (acl2p-sum-list1 (cdr lst) (+ (car lst) acc)))))

(defun acl2p-sum-list (lst) (acl2p-sum-list1 lst 0))

(defun average-future-queue-size
  nil
  (* 1.0
    (/ (acl2p-sum-list *future-queue-length-history*)
      (length *future-queue-length-history*))))

(defun print-interesting-parallelism-variables-str
  nil
  (incf *refresh-rate-indicator*)
  (setf *future-queue-length-history*
    (cons (future-queue-length) *future-queue-length-history*))
  (concatenate 'string
    (format nil
      "  Printing stats related to executing proofs in parallel.~% ")
    (value-of-symbol *idle-future-core-count*)
    (value-of-symbol *idle-future-resumptive-core-count*)
    (value-of-symbol *idle-future-thread-count*)
    (value-of-symbol *threads-waiting-for-starting-core*)
    (value-of-symbol number-of-idle-threads-and-threads-waiting-for-a-starting-core)
    (value-of-symbol total-number-of-threads)
    (format nil "~% ")
    (value-of-symbol *unassigned-and-active-future-count*)
    (value-of-symbol *unassigned-and-active-work-count-limit*)
    (value-of-symbol *total-future-count*)
    (value-of-symbol total-parallelism-work-limit)
    (format nil "~% ")
    (value-of-symbol number-of-active-threads)
    (value-of-symbol number-of-threads-waiting-on-a-child)
    (format nil "~% ")
    (value-of-symbol *last-slot-taken*)
    (value-of-symbol *last-slot-saved*)
    (value-of-symbol future-queue-length)
    (value-of-symbol average-future-queue-size)
    (format nil "~% ")
    (value-of-symbol *resource-based-parallelizations*)
    (value-of-symbol *resource-based-serializations*)
    (value-of-symbol *resource-and-timing-based-parallelizations*)
    (value-of-symbol *resource-and-timing-based-serializations*)
    (value-of-symbol *futures-resources-available-count*)
    (value-of-symbol *futures-resources-unavailable-count*)
    (format nil "~% ")
    (format nil
      " Printing stats related to aborting futures.~% ")
    (value-of-symbol *aborted-futures-total*)
    (value-of-symbol *aborted-futures-via-throw*)
    (value-of-symbol *aborted-futures-via-flag*)
    (value-of-symbol *almost-aborted-future-count*)
    (format nil "~% ")
    (value-of-symbol *refresh-rate-indicator*)))

(defun print-interesting-parallelism-variables
  nil
  (format t (print-interesting-parallelism-variables-str)))