;;; Timers

(defun start-timer ()
  (get-internal-real-time))

(defun stop-timer (timer)
  (/ (float (- (get-internal-real-time) timer))
     internal-time-units-per-second))

(defmacro timeit (func)
  `(let* ((timer (start-timer))
	  (rslt ,func))
     (format t "~a~%" (stop-timer timer))
     rslt))

;;; Threads
(require "alexandria")
(require "sb-concurrency")
(use-package :alexandria)
(use-package :sb-concurrency)
(use-package :sb-thread) ; Seems I do not need to require this one

(defmacro domsg ((var mbox) &body body)
  (with-gensyms (start)
    `(let (,var)
       (block nil
	 (tagbody
	    ,start
	    (setq ,var (receive-message ,mbox))
	    ,@body
	    (if ,var (go ,start)))))))

(defun source (dst)
  (do ((k 1 (1+ k))) ((= k 100000) (send-message dst nil))
    (send-message dst (/ (* (1+ k) k) 2))))

(defun make-pairs (src dst)
  (let (buffer)
    (domsg (chunk src)
      (if chunk
	  (if buffer
	      (send-message dst (cons (pop buffer) chunk))
	      (push chunk buffer))
	  (send-message dst nil)))))

(defun split (src dst1 dst2)
  (domsg (chunk src)
    (send-message dst1 chunk)
    (send-message dst2 chunk)))

(defun multiply (src dst)
  (domsg (chunk src)
    (if chunk
	(destructuring-bind (x . y) chunk
	  (send-message dst (* x y)))
	(send-message dst nil))))

(defun toy-gcd (src dst)
  (labels ((helper (x y)
	     (if (< x y)
		 (helper y x)
		 (if (= 0 y) (send-message dst x)
		     (helper y (mod x y))))))
    (domsg (chunk src)
      (if chunk
	  (destructuring-bind (x . y) chunk
	    (helper x y))
	  (send-message dst nil)))))

(defun combine (src1 src2 dst)
  (domsg (chunk src1)
    (if chunk
	(send-message dst (cons chunk (receive-message src2)))
	(send-message dst nil))))

(defun divide (src dst)
  (domsg (chunk src)
    (if chunk
	(destructuring-bind (x . y) chunk
	  (send-message dst (/ x y)))
	(send-message dst nil))))

(defun get-digits (src dst)
  (labels ((helper (x)
	     (unless (= 0 x)
	       (helper (floor x 10))
	       (send-message dst (mod x 10)))))
    (domsg (x src)
      (if x
	  (helper x)
	  (send-message dst nil)))))

(defun sink (src)
  (domsg (chunk src)))

(defun pipeline ()
  (let ((c1 (make-mailbox))
	(c2 (make-mailbox))
	(c3 (make-mailbox))
	(c4 (make-mailbox))
	(c5 (make-mailbox))
	(c6 (make-mailbox))
	(c7 (make-mailbox))
	(c8 (make-mailbox))
	(c9 (make-mailbox)))
    (let ((ft (make-thread (lambda () (sink c9)))))
      (make-thread (lambda () (get-digits c8 c9)))
      (make-thread (lambda () (divide c7 c8)))
      (make-thread (lambda () (combine c5 c6 c7)))
      (make-thread (lambda () (toy-gcd c4 c6)))
      (make-thread (lambda () (multiply c3 c5)))
      (make-thread (lambda () (split c2 c3 c4)))
      (make-thread (lambda () (make-pairs c1 c2)))
      (unwind-protect
	   (progn
	     (source c1)
	     (join-thread ft))
	(send-message c1 nil)))))

(timeit (dotimes (i 100) (pipeline)))