Entry tags:

lisp

Multiprocessing lisp evaluations
I sometimes need to evaluate the same form with different parameters repeatedly. Such as

(job 1)
(job 2)
(job 3)
...

When these are time consuming, I'd like to take advantage of the two processors I have in my computer. As the jobs are independent of each other (no communication), I only need to maintain a job queue, have each processor pick off the front of the queue, and then place the results in an accessible place before getting the next job. Since SBCL has threads, at least for Linux on x86 and amd64, I should be able to use this mechanism to build the job queue. Following Rochkind Section 5.17, I have written the following:

(defparameter *job-lock* (sb-thread:make-mutex :name "job lock"))
(defparameter *results* (list nil))
(defvar *end-of-jobs* (make-symbol "EOJ"))
(defvar *jobs* nil)

(defun worker (job)
  (let ((my-job nil)
	(more-jobs t))
    (loop while more-jobs
       do
       (sb-thread:with-mutex (*job-lock*)
	 (setf more-jobs (or *jobs*))
	 (setf my-job (when more-jobs (pop *jobs*))))
       (when my-job			; there is a job to be done
	 (if (eq my-job *end-of-jobs*)
	     (setf more-jobs nil)
	     (let ((my-results (apply job my-job))) ; call the job outside the mutex
	       (sb-thread:with-mutex (*job-lock*)   ; save results
		 (push my-results *results*))))))))

(defun run-tasks (job dataset number-of-workers)
  "The job is a function that takes one non-null argument.
   The dataset is a list of arglist sets for the job.
   The number-of-workers is the number of workers desired, 
   presumably the number of processors available."
  (setf *jobs* (make-list number-of-workers :initial-element *end-of-jobs*)
	*results* (list nil))
  (dolist (ds dataset) (push ds *jobs*))
  (let ((threads (list nil)))
    (loop repeat number-of-workers
       do (push (sb-thread:make-thread (lambda () (worker job))) threads))
    (dolist (thread (butlast threads)) (sb-thread:join-thread thread))
    (butlast *results*)))

Try this example:

(defun job (x) (list x (+ (loop for i from 1 to 2000000 sum (let ((p (* i x))) (1- (expt p (/ p))))))))
(run-tasks #'job '((1) (2) (3) (4) (5) (6) (7) (8) (9) (10)) 2)
((1 105.73587) (2 58.11023) (3 41.245914) (4 31.676012) (5 26.053244)
 (6 22.305136) (7 19.607342) (8 17.642727) (9 15.433696) (10 14.060191))

Unfortunately, after I coded this up and tried it on my actual function, I found that that function was not thread safe, due to use of a foreign library that wasn't thread safe.

(defstruct locked-list
  (list nil)
  (lock (sb-thread:make-mutex)))

(defun worker (fun jobs)
  (loop for job = (sb-thread:with-mutex ((locked-list-lock jobs))
                    (if (locked-list-list jobs)
                        (pop (locked-list-list jobs))
                        (loop-finish)))
        collect (apply fun job)))

(defun run-tasks (fun dataset number-of-workers)
  (loop with jobs = (make-locked-list :list dataset)
        for thread in (loop repeat number-of-workers
                            collect (sb-thread:make-thread (lambda ()
                                                             (worker fun jobs))))
        appending (sb-thread:join-thread thread)))

<pre style="overflow:auto">
;; your wide Lisp code...
</pre>