I’ve written a program for solving a problem using standard single-threaded code. However, it looks like it could be recast as a multi-threaded problem using divide & conquer. Since this is a first attempt at multi-thread programming (and debugging), I thought it might be advisable to separate the problem itself from the parallel management. The following is a multi-threaded divide & conquer skeleton, which hopefully the actual problem could fit into. I would appreciate a code check and any relevant advice about how to better code multi-threaded applications like this.
The skeleton uses the Lparallel library with queues for communicating among the threads. There is a main thread which initializes the queues & threads, submits the problem, and then blocks waiting for a solution. The problem is submitted to a pool of threads, whose number can range up to one less than the number of hardware processors available. Each idle thread blocks until a (sub)problem becomes available on the problems queue, which it may then then capture. If the problem size is above a minimal problem-size-cutoff, the thread splits off a subproblem (randomly to simulate the uncertainty in how much future processing the subproblem will require). The subproblem is pushed onto the problems queue so another idle thread can deal with it subsequently, and the thread then continues processing with what’s remaining of the problem after the split. When a thread finishes working on its problem, it reports that it is now idle via the idles queue. When any thread (namely, the last thread running) observes that all threads are idle, it reports problem complete via the done queue.
To run the following code, simply switch to the :test package and execute (main) after loading. The printout shows when any thread receives a problem (its size—eg, 34), and if it is split, what the random split is—eg, (13 21). If it is not split because it is too small, the printout just shows the simulated processing of what’s remaining.
Secondary questions: 1) The function main uses up a whole thread just waiting for problem completion. Is there a way to use this thread too? 2) Can the number of queues or queue accesses be reduced? 3) Would a so-called sequence diagram be the best way to verify correctness? Ie, is this worth learning, or is there a better way. (Ps: The program seems to run OK, but I’m not even sure how to simulate the duration of a thread.) Thanks for any time you can spare.
EDITS 3/30/19: 1) Replaced with-temp-kernel (1) with (setf lparallel:*kernel* (lparallel:make-kernel *num-threads*)) and (lparallel:end-kernel) to follow library recommended way to start and stop lparallel kernel. 2) Added tracing queues for each thread to collect each thread’s problem solution progress for debugging. 3) Added epilog code to main to printout the trace of each thread after kernel shutdown, in order to avoid printing interference during shutdown.
#-:lparallel (ql:quickload :lparallel)
(defpackage :test (:use :cl :lparallel :lparallel.queue))
(in-package :test)
(defparameter *num-threads* 2) ;available problem-solving threads (excludes main thread-0)
(defparameter *problem* 100) ;estimated size of the main problem
(defparameter *problem-size-cutoff* 5) ;don't split small problems
(defun thread (problems idles trace done)
"Definition for a problem-solving thread."
(loop do
(let ((problem (pop-queue problems))) ;blocks until a problem is available
(pop-queue idles) ;one (this) thread is no longer idle
(push-queue problem trace) ;track this thread's solution of its problem
(loop do
(when (and (> problem (* 2 *problem-size-cutoff*)) ;don't split if small problem
(>= (queue-count idles) 1)) ;ie, some threads are idle
(let ((n (+ (random (- problem (* 2 *problem-size-cutoff*)))
*problem-size-cutoff*))) ;split problem
(sleep .01) ;time to run splitting algorithm
(push-queue n problems) ;n is random size of subproblem split
(setf problem (- problem n)) ;remaining problem size after split
(push-queue (list n problem) trace))) ;track this thread's splitting
(decf problem) (sleep .1) ;work more on current problem
(push-queue problem trace) ;track this thread's incremental progress
(when (= problem 0) ;is this thread's task complete?
(push-queue t idles) ;this thread now idle
(if (= (queue-count idles) *num-threads*) ;all threads idle
(progn (push-queue t done) ;signal all done to main
(return-from thread))
(return))))))) ;return from inner loop to get new problem
(defun main ()
"The main consumer of parallel thread processing."
(setf lparallel:*kernel* (lparallel:make-kernel *num-threads*))
(let ((c (make-channel))
(problems (make-queue)) ;holds current problems to be solved
(idles (make-queue)) ;holds a t for each current idle thread
(traces nil) ;holds a trace of the progress of each thread for debugging
(done (make-queue))) ;signals t when all processing is done
(loop for i from 1 to *num-threads* ;get all threads up & running
do (let ((trace (make-queue))) ;track the progress of a thread using a queue
(push trace traces)
(submit-task c #'thread problems idles trace done)
(push-queue t idles))) ;each t in idles means a thread is currently idle
(push-queue *problem* problems) ;send the main problem to the thread pool
(pop-queue done) ;block until thread pool reports done
(let ((list-of-traces ;collect each thread's trace before shutting down kernel
(map 'list (lambda (trace)
(loop until (queue-empty-p trace)
collect (pop-queue trace)))
traces)))
(lparallel:end-kernel)
(loop for rtrace in (reverse list-of-traces) ;printout each thread's trace
do (print rtrace))))) ;final print to avoid interference with shutdown
