Multiprocess multithreaded computation with SBCL

shared.lisp

#!/usr/bin/sbcl --script
(require :sb-posix)
(require :sb-bsd-sockets)

;;; Globals
;; IPC
(defvar *pid* nil)
(defvar *socket-path* "/tmp/socket")
(defvar *socket* nil)
;; MT
(defvar *running* nil)
(defvar *workers* nil)
(defvar *shared-resource* nil)


;;; Business check
(defun signal-finish ()
  (sb-ext:atomic-update *running* #'1-))
  
(defun not-busy-p ()
  (zerop *running*))


;;; Result collector
(defun status ()
  (format t "Common(~a): ~{~a~^ ~}~%" *pid* *shared-resource*))

(defun add (elem)
  (sb-ext:atomic-update *shared-resource* #'cons elem))
  

;;; Workers
(defun make-workers (count args)
  (setf *running* count)
  (setf *workers* (loop for i below count collect (apply #'make-worker (nth i args)))))

(defun join-workers (&optional (workers *workers*))
  (mapc #'sb-thread:join-thread workers))


;;; Util
(defun color-formatter (color)
  (format nil "~a[~am~~?~a[m" #\Esc color #\Esc))

(defun value (prefix inc)
  (format nil "~?/~a/~a" (color-formatter (if (zerop *pid*) "1;31" "1;32")) `("~a" (,(if (zerop *pid*) "P" "C"))) prefix inc))
  
(defun make-worker (prefix count delay)
  (sb-thread:make-thread
    #'(lambda ()
        (loop for i below count do
            (add (value prefix i))
            (sleep delay))
        (signal-finish))))
        

;;; Fork
(defun mt-computation ()
  (format t "The process ~a is starting.~%" *pid*)
  (make-workers 2
    `((,(format nil "~?" (color-formatter "1;33") '("~a" (a))) 100 0.0001)
      (,(format nil "~?" (color-formatter "1;36") '("~a" (b))) 10 0.001)))
  (loop named shower
    do (status) (sleep 0.001)
    when (not-busy-p) do (return-from shower))
  (join-workers)
  (format t "The process ~a is stopping.~%" *pid*))
  
(defun mp-computation ()
  (when (probe-file *socket-path*) (delete-file *socket-path*))
  (setf *pid* (sb-posix:fork))
  
  (setf *socket* (make-instance 'sb-bsd-sockets:local-socket :type :stream))
  (restart-case
    (handler-bind ((sb-bsd-sockets:address-in-use-error #'(lambda (condition) (invoke-restart 'connect))))
      (sb-bsd-sockets:socket-bind *socket* *socket-path*)
      (sb-bsd-sockets:socket-listen *socket* 0)
      (setf *socket* (sb-bsd-sockets:socket-accept *socket*)))
    (connect () (sb-bsd-sockets:socket-connect *socket* *socket-path*)))

  (if (zerop *pid*)
    (format t "~a" (sb-bsd-sockets:socket-receive *socket* nil 100))
    (sb-bsd-sockets:socket-send *socket* (format nil "The child ~a is starting.~%" *pid*) nil))
  
  (mt-computation)

  (if (zerop *pid*)
    (format t "~a" (sb-bsd-sockets:socket-receive *socket* nil 100))
    (sb-bsd-sockets:socket-send *socket* (format nil "The child ~a is stopping.~%" *pid*) nil))
  
  (sb-bsd-sockets:socket-close *socket*))


;;; Main
(mp-computation)