127 |
broadcast; If tag of most recent message is at least that large, |
broadcast; If tag of most recent message is at least that large, |
128 |
this is up-to-date." |
this is up-to-date." |
129 |
(>= (get-last-tag) (aref *up-to-date-tag* 0))) |
(>= (get-last-tag) (aref *up-to-date-tag* 0))) |
130 |
(defmacro result-task () |
(defmacro result-task-input () |
131 |
"Returns task corresponding to most recent result (most recent message) |
"Returns task-input corresponding to most recent result (most recent message) |
132 |
returned." |
returned." |
133 |
`(aref pending (get-last-tag))) |
`(aref pending (get-last-tag))) |
134 |
|
|
149 |
(aref pending curr-tag))) |
(aref pending curr-tag))) |
150 |
(setf (aref pending curr-tag) task) |
(setf (aref pending curr-tag) task) |
151 |
(setf (aref slave-tags dest) curr-tag) |
(setf (aref slave-tags dest) curr-tag) |
152 |
,@(if trace `((if ,trace (format t "TRACE: -> ~d: ~s~%" dest task)))) |
,@(if trace `((if ,trace (format t "TRACE: -> ~d : ~s~%" dest task)))) |
153 |
(decf num-free-processors) |
(decf num-free-processors) |
154 |
(send-message task dest curr-tag))) |
(send-message task dest curr-tag))) |
155 |
|
|
214 |
(funcall generate-task-input)))))) |
(funcall generate-task-input)))))) |
215 |
(send-task task (pop free-processors) trace)) |
(send-task task (pop free-processors) trace)) |
216 |
(when (= num-free-processors num-slaves) (return)) ; no task, no pending |
(when (= num-free-processors num-slaves) (return)) ; no task, no pending |
217 |
(let ((result (receive-message))) ; Get a result |
(let ((task-out (receive-message))) ; Get a task-out |
218 |
(if trace |
(if trace |
219 |
(format t "TRACE: ~d ->: ~s~%" (MPI-Status-source) result)) |
(format t "TRACE: <= ~d: ~s~%" (MPI-Status-source) task-out)) |
220 |
(if (eq (result-task) DEAD-SLAVE) |
(if (eq (result-task-input) DEAD-SLAVE) |
221 |
(progn |
(progn |
222 |
(setq result nil) |
(setq task-out nil) |
223 |
(if trace |
(if trace |
224 |
(format t "DEAD SLAVE tried to return result. Killed.~%"))) |
(format t |
225 |
|
"DEAD SLAVE tried to return task output. Killed.~%"))) |
226 |
(progn ; BETTER TO DELAY THIS AND DO ONLY IF NEEDED (a la TOP-C) |
(progn ; BETTER TO DELAY THIS AND DO ONLY IF NEEDED (a la TOP-C) |
227 |
(setq free-processors ; free-processors is FIFO queue |
(setq free-processors ; free-processors is FIFO queue |
228 |
(nconc free-processors (list (MPI-status-source)))) |
(nconc free-processors (list (MPI-status-source)))) |
229 |
(incf num-free-processors))) |
(incf num-free-processors))) |
230 |
(setf (aref slave-tags (MPI-status-source)) NO-TAG) |
(setf (aref slave-tags (MPI-status-source)) NO-TAG) |
231 |
(when result |
(when task-out |
232 |
(block try-result |
(block try-result |
233 |
(let ((task-result |
(let ((task-action |
234 |
(if check-task-result |
(if check-task-result |
235 |
(funcall check-task-result result) t))) |
(funcall check-task-result |
236 |
(if (and (consp task-result) |
(result-task-input) task-out) |
237 |
(eq (first task-result) 'continuation)) |
t))) ; default is t = UPDATE |
238 |
|
(if (and (consp task-action) |
239 |
|
(eq (first task-action) 'continuation)) |
240 |
;;ask same processor to extend computation |
;;ask same processor to extend computation |
241 |
(progn (setf (aref pending (get-last-tag)) nil) |
(progn (setf (result-task-input) nil) |
242 |
(send-task task-result |
(send-task task-action |
243 |
(first (last free-processors)) trace) |
(first (last free-processors)) trace) |
244 |
(setf free-processors (nbutlast free-processors)) |
(setf free-processors (nbutlast free-processors)) |
245 |
(return-from try-result)) ; skip update |
(return-from try-result)) ; skip update |
246 |
(case task-result |
(case task-action |
247 |
;;REDO: resend to same processor that did result. |
;;REDO: resend to same processor that did task-out. |
248 |
;;IF RESENDING TO SAME PROC., CAN JUST RETURN ? |
;;IF RESENDING TO SAME PROC., CAN JUST RETURN ? |
249 |
;; AND LET do-task REMEMBER ORIGINAL TASK |
;; AND LET do-task REMEMBER ORIGINAL TASK |
250 |
((? REDO) (send-task (aref pending (get-last-tag)) |
((? REDO) (send-task (result-task-input) |
251 |
(first (last free-processors)) trace) |
(first (last free-processors)) trace) |
252 |
(setf free-processors (nbutlast free-processors)) |
(setf free-processors (nbutlast free-processors)) |
253 |
;;(get-last-tag) is set for last REC'D msg, only |
;;(get-last-tag) is set for last REC'D msg, only |
254 |
(setf (aref pending (get-last-tag)) nil) |
(setf (result-task-input) nil) |
255 |
(return-from try-result)) ; skip update |
(return-from try-result)) ; skip update |
256 |
((nil NO-ACTION) ; NO-ACTION |
((nil NO-ACTION) ; NO-ACTION |
257 |
(setf (aref pending (get-last-tag)) nil) |
(setf (result-task-input) nil) |
258 |
(return-from try-result)) ; skip update |
(return-from try-result)) ; skip update |
259 |
((t UPDATE) ; UPDATE |
((t UPDATE) ; UPDATE |
260 |
;; else assume result is up-to-date |
;; else assume task-out is up-to-date |
|
(setf (aref pending (get-last-tag)) nil) |
|
261 |
(when update-shared-data |
(when update-shared-data |
262 |
;;broadcast-tag will cause slave in master-slave |
;;broadcast-tag will cause slave in master-slave |
263 |
;; mode to recognize update-shared-data request |
;; mode to recognize update-shared-data request |
264 |
(progn (broadcast-message result) |
(progn (broadcast-message (result-task-input)) |
265 |
(funcall update-shared-data result))) |
(broadcast-message task-out) |
266 |
(when set-task (funcall set-task result)) |
(funcall update-shared-data |
267 |
|
(result-task-input) |
268 |
|
task-out))) |
269 |
|
(setf (result-task-input) nil) |
270 |
|
(when set-task (funcall set-task task-out)) |
271 |
(setf (aref *up-to-date-tag* 0) curr-tag)) |
(setf (aref *up-to-date-tag* 0) curr-tag)) |
272 |
(otherwise (error ":check-task-result must return one ~ |
(otherwise (error ":check-task-result must return one ~ |
273 |
of '(REDO NO-ACTION UPDATE).")))))))) |
of '(REDO NO-ACTION UPDATE).")))))))) |
309 |
;;THIS ASSUMES SLAVE IS ALSO RUNNING (master-slave ...) |
;;THIS ASSUMES SLAVE IS ALSO RUNNING (master-slave ...) |
310 |
;; mpi::... required, since reader would otherwise put in curr package |
;; mpi::... required, since reader would otherwise put in curr package |
311 |
(broadcast-message "#.(throw 'mpi::done 'done)")) |
(broadcast-message "#.(throw 'mpi::done 'done)")) |
312 |
(let ((result)) ; else this is a slave |
(let ((task-in)) ; else this is a slave |
313 |
(catch 'mpi::done |
(catch 'mpi::done |
314 |
(loop |
(loop |
315 |
(setq result (receive-message 0)) |
(setq task-in (receive-message 0)) |
316 |
(if (equal result '(quote %%mpi-internal-slave-escape%%)) (return)) |
(if (equal task-in '(quote %%mpi-internal-slave-escape%%)) |
317 |
(if (equal result '(quote %%mpi-internal-slave-confirm%%)) |
(return)) |
318 |
|
(if (equal task-in '(quote %%mpi-internal-slave-confirm%%)) |
319 |
(send-message 'mpi-slave-confirmed 0 (get-last-tag)) |
(send-message 'mpi-slave-confirmed 0 (get-last-tag)) |
320 |
(if (/= (get-last-tag) (the fixnum broadcast-tag)) |
(if (/= (get-last-tag) (the fixnum broadcast-tag)) |
321 |
(send-message (funcall do-task result) 0 (get-last-tag)) ;reply |
(send-message (funcall do-task task-in) 0 (get-last-tag));reply |
322 |
(when update-shared-data ; broadcast doesn't require reply |
(progn |
323 |
(funcall update-shared-data result)))))))) |
(let* ((task-out (receive-message 0))) |
324 |
|
(when update-shared-data ; broadcast doesn't require reply |
325 |
|
(funcall update-shared-data task-in task-out)))))))))) |
326 |
(when buffer-size |
(when buffer-size |
327 |
(adjust-array (get '*msg-buf* 'character) old-buffer-size :static t)) |
(adjust-array (get '*msg-buf* 'character) old-buffer-size :static t)) |
328 |
(setf (first *master-slave-time*) |
(setf (first *master-slave-time*) |