35 |
:export (par-map |
:export (par-map |
36 |
par-for-each |
par-for-each |
37 |
n-par-map |
n-par-map |
38 |
n-par-for-each) |
n-par-for-each |
39 |
|
n-for-each-par-map) |
40 |
:re-export (future-ref) |
:re-export (future-ref) |
41 |
:export-syntax (begin-thread |
:export-syntax (begin-thread |
42 |
parallel |
parallel |
101 |
(loop))))) |
(loop))))) |
102 |
futures))))) |
futures))))) |
103 |
|
|
104 |
|
;;; The following procedure is motivated by the common and important |
105 |
|
;;; case where a lot of work should be done (not too much) in parallel |
106 |
|
;;; but the results need to be handled serially (for example when |
107 |
|
;;; writing them to a file). |
108 |
|
;;; |
109 |
|
(define (n-for-each-par-map n s-proc p-proc . arglists) |
110 |
|
"Using N parallel processes, apply S-PROC in serial order on the results |
111 |
|
of applying P-PROC on ARGLISTS." |
112 |
|
(let* ((m (make-mutex)) |
113 |
|
(futures '()) |
114 |
|
(no-result '(no-value)) |
115 |
|
(results (make-list (length (car arglists)) no-result)) |
116 |
|
(result results)) |
117 |
|
(do ((i 0 (+ 1 i))) |
118 |
|
((= i n) |
119 |
|
(for-each future-ref futures)) |
120 |
|
(set! futures |
121 |
|
(cons (future |
122 |
|
(let loop () |
123 |
|
(lock-mutex m) |
124 |
|
(cond ((null? results) |
125 |
|
(unlock-mutex m)) |
126 |
|
((not (eq? (car results) no-result)) |
127 |
|
(let ((arg (car results))) |
128 |
|
;; stop others from choosing to process results |
129 |
|
(set-car! results no-result) |
130 |
|
(unlock-mutex m) |
131 |
|
(s-proc arg) |
132 |
|
(lock-mutex m) |
133 |
|
(set! results (cdr results)) |
134 |
|
(unlock-mutex m) |
135 |
|
(loop))) |
136 |
|
((null? result) |
137 |
|
(unlock-mutex m)) |
138 |
|
(else |
139 |
|
(let ((args (map car arglists)) |
140 |
|
(my-result result)) |
141 |
|
(set! arglists (map cdr arglists)) |
142 |
|
(set! result (cdr result)) |
143 |
|
(unlock-mutex m) |
144 |
|
(set-car! my-result (apply p-proc args)) |
145 |
|
(loop)))))) |
146 |
|
futures))))) |
147 |
|
|
148 |
(define (thread-handler tag . args) |
(define (thread-handler tag . args) |
149 |
(fluid-set! the-last-stack #f) |
(fluid-set! the-last-stack #f) |
150 |
(let ((n (length args)) |
(let ((n (length args)) |