(in-package :cl-user)

(use-package :sb-thread)

;; Test robustness of semaphore acquisition and notification
;; with asynchronous thread termination.

(let ((test-name '(:semaphore-notification :wait-on-semaphore)))
  (format t "~&::: Running ~A~%" test-name)
  (let ((sem (make-semaphore))
        (ok nil)
        (n 0)                 ; counter for recording acquisition sequence
	(notes (list :dummy)) ; records successful semaphore acquisitions
        (incs (list :dummy))  ; records successful sleep && incf
	(kills (list :dummy)))
    ;; In CRITICAL, WAIT-ON-SEMAPHORE and SLEEP can be interrupted by
    ;; TERMINATE-THREAD below. But the PUSH, INCF in case of success
    ;; and the SIGNAL-SEMAPHORE cleanup cannot be interrupted.
    ;;
    ;; INCS, as opposed to the threads' return values, is used to
    ;; record successful acquisitions of the semaphore because
    ;; TERMINATE-THREAD below may abort a thread after CRITICAL has
    ;; left WITHOUT-INTERRUPTS.
    (flet ((critical (i)
             (let ((note (make-semaphore-notification)))
               (sb-sys:without-interrupts
                 (unwind-protect
                      (let ((sleep (ecase i
				     ((1) 0.01)
				     ((2 3) 0.1))))
                        (sb-sys:with-local-interrupts
                          (wait-on-semaphore sem :notification note)
			  (atomic-push (list (thread-name *current-thread*) n (get-internal-real-time)) (cdr notes))
                          (sleep sleep))
			(atomic-push (list (thread-name *current-thread*) n (get-internal-real-time)) (cdr incs))
                        (values (incf n) sleep))
                   ;; Re-increment on exit if we decremented it.
                   (when (semaphore-notification-status note)
                     (signal-semaphore sem)))))))
      ;; Create /parallel/ threads trying to acquire and then release the
      ;; semaphore. Try to asynchronously abort every second thread
      ;; at some random point of its execution.
      (let* ((threads (loop for i from 1 upto 3
                            collect (make-thread
                                     #'critical :name (format nil "T~A" i) :arguments (list i))))
             (safe '())
	     (safe-remaining '())
             (unsafe '())
             (interruptor (make-thread (lambda ()
                                         (loop until ok)
                                         (let (x)
                                           (loop :for thread :in threads
					      :for i :from 1
					      :do
					      (assert (string= (format nil "T~A" i) (thread-name thread)))
                                             (cond (x
                                                    (push thread unsafe)
                                                    (let ((sleep (ecase i
								   ((2) 0.01))))
						      (sleep sleep)
						      (let ((t1 (get-internal-real-time)))
							(multiple-value-bind (result err)
							    (ignore-errors
							      (terminate-thread thread))
							  (declare (ignorable result err))
							  (atomic-push (list (thread-name thread)
									     (not err) 
									     t1 
									     (get-internal-real-time))
								       (cdr kills))
							  (princ (if err #\E #\-))))))
                                                   (t
						    (princ #\.)
                                                    (push thread safe)))
                                             (setf x (not x))))))))
        (signal-semaphore sem)
        (setf ok t)
        (join-thread interruptor)
	(terpri)
	(setf safe-remaining safe)
	(sleep 0.1)
	(loop
	   :with safe-remaining-old
	   :for count :from 0.1 :below 10 :by 0.1
	   :do (setf safe-remaining-old safe-remaining
		     safe-remaining
		     (remove-if-not #'(lambda (thread)
					(eq :busy
					    (join-thread thread
							 :timeout 0
							 :default :busy)))
				    safe-remaining))
	   :when (and (not (eql (length safe-remaining-old)
				(length safe-remaining)))
		      (or safe-remaining
			  (> count 0.1)))
	   :do (format t "[~4,1Fs] ~D safe remaining~@[: ~{~A~^, ~}~]~%"
		       count (length safe-remaining)
		       (mapcar #'thread-name (reverse safe-remaining)))
	   :while safe-remaining
	   :do (sleep 0.1))
	(when safe-remaining
	  (format t "~&*** Notes:")
	  (mapc #'pprint (sort (copy-list (cdr notes)) #'< :key #'third))
	  (format t "~&*** Inc's:")
	  (mapc #'pprint (sort (copy-list (cdr incs)) #'< :key #'third))
	  (format t "~&*** Kills:")
	  (mapc #'pprint (sort (copy-list (cdr kills)) #'< :key #'third))
	  (terpri) (force-output)
	  (dolist (thread (reverse safe-remaining))
	    (let ((err
		   (second
		    (multiple-value-list
		     (ignore-errors
		       (interrupt-thread thread #'sb-debug:backtrace))))))
	      (cond (err
		     (format *ERROR-OUTPUT* "~&*** ERROR: ~A~%" err)
		     (force-output *ERROR-OUTPUT*))
		    (t
		     (sleep 3)))))
	  (exit :code 1 :abort t))
        ;; Within the "unsafe" set, some threads may have acquired
        ;; the semaphore before being interrupted.
        (let* ((unsafe/successful (intersection (mapcar #'car (cdr incs)) unsafe))
               (ns (mapcar #'second (cdr incs))))
          ;; The number of successful "unsafe" threads and the
          ;; (deterministic) size of the "safe" set have to add up to
          ;; the total number of acquisitions.
          (assert (= n (+ (length unsafe/successful) (length safe))))
          ;; The acquisition of the semaphore serializes access to N,
          ;; therefore the recorded values of N should increase in
          ;; increments of 1.
          (assert (equal (reverse ns)
                         (loop for i below (length ns) collect i)))
          ;; Sanity check: both sets should be non-empty
          (assert safe)
          (assert unsafe)))))
  (format t "~&::: Success ~A~%" test-name))
(exit :code 0)