Skip to content
Snippets Groups Projects
Commit 70384eb1 authored by Vladimir Sedach's avatar Vladimir Sedach
Browse files

Added support for using restarts to handle errors in future computations....

Added support for using restarts to handle errors in future computations. Thanks to Maciej Katafiasz for requesting this functionality.
parent 4f9c1e69
No related branches found
No related tags found
No related merge requests found
......@@ -6,15 +6,21 @@
(lock :reader lock :initform (make-lock "future lock"))
(computing-thread :accessor computing-thread :initform nil)
(wait-list :accessor wait-list :initform ())
(future-id :reader future-id :initarg :future-id)))
(future-id :reader future-id :initarg :future-id)
(restart-notifier :accessor restart-notifier :initform nil)
(error-descriptor :accessor error-descriptor :initform nil)
(proxy-restart :accessor proxy-restart :initform nil)))
(defun make-future (task future-id)
(make-instance 'future :task task :future-id future-id))
(defun %ready-to-yield? (future)
(or (slot-boundp future 'values) (error-descriptor future)))
(defun ready-to-yield? (future)
"Returns t if the future values have been computed, nil otherwise."
(with-lock-held ((lock future))
(slot-boundp future 'values)))
(%ready-to-yield? future)))
(defun force (future &rest values)
"If the future has not yet yielded a value, installs the given
......@@ -32,17 +38,20 @@ computation of the future)."
(abort-scheduled-future-task computing-thread (future-id future)))
(setf computing-thread nil
(wait-list future) nil
(task future) nil)))
(task future) nil
(restart-notifier future) nil
(error-descriptor future) nil
(proxy-restart future) nil)))
(defun select (&rest futures)
"Returns the first future that is ready to yield."
(let ((notifier (make-condition-variable))
(select-lock (make-lock "select lock")))
(let ((notifier (make-condition-variable :name "Eager Future2 select notifier"))
(select-lock (make-lock "Eager Future2 select lock")))
(with-lock-held (select-lock)
(let (any-computing?)
(dolist (future futures)
(with-lock-held ((lock future))
(when (slot-boundp future 'values)
(when (%ready-to-yield? future)
(return-from select future))
(when (and (not any-computing?) (computing-thread future))
(setf any-computing? t))
......@@ -50,9 +59,8 @@ computation of the future)."
(unless any-computing?
(schedule-future (first futures) :speculative)))
(loop (dolist (future futures)
(with-lock-held ((lock future))
(when (slot-boundp future 'values)
(return-from select future))))
(when (ready-to-yield? future)
(return-from select future)))
(condition-wait notifier select-lock)))))
(defun yield (future)
......@@ -67,14 +75,37 @@ another thread is currently computing the value of the future, blocks
until the value is available.
In case of an eager future, blocks until the value is available."
(tagbody
(with-lock-held ((lock future))
(if (slot-boundp future 'values)
(go done)
(if (computing-thread future)
(go select)
(progn (setf (computing-thread future) (current-thread))
(go compute)))))
select (select future) (go done)
compute (multiple-value-call #'force future (funcall (task future)))
done (return-from yield (values-list (%values future)))))
(tagbody (catch 'task-done
(let ((*computing-future* (future-id future)))
(with-lock-held ((lock future))
(cond ((slot-boundp future 'values) (return-from yield (values-list (%values future))))
((error-descriptor future) (go handle-error))
((computing-thread future) (go select))
(t (setf (computing-thread future) (current-thread)))))
(multiple-value-call #'force future (restart-case (funcall (task future))
(force-values (&rest values)
:report "Set future value"
:interactive (lambda () (list (eval (read))))
(values-list values))))))
(return-from yield (values-list (%values future)))
select (select future)
handle-error (with-lock-held ((lock future))
(if (slot-boundp future 'values)
(return-from yield (values-list (%values future)))
(progn
(catch 'go-select
(eval `(restart-case (error ,(car (error-descriptor future)))
(force-values (&rest values)
:report "Set future value"
:interactive (lambda () (list (eval (read))))
(setf (proxy-restart ,future) (cons 'force-values values)
(error-descriptor ,future) nil)
(condition-notify (restart-notifier ,future))
(throw 'go-select nil))
,@(loop for restart in (cdr (error-descriptor future)) collect
`(,(restart-name restart) (&rest args)
(setf (proxy-restart ,future) (cons ,restart args)
(error-descriptor ,future) nil)
(condition-notify (restart-notifier ,future))
(throw 'go-select nil))))))
(go select))))))
......@@ -22,13 +22,25 @@ If lazy, newly created futures are not computed until asked to yield their value
(flet ((get-future () (or (weak-pointer-value future-ptr) (throw 'task-done nil))))
(let ((*computing-future* (future-id (get-future))))
(with-lock-held ((lock (get-future)))
(if (slot-boundp (get-future) 'values)
(if (%ready-to-yield? (get-future))
(throw 'task-done nil)
(setf (computing-thread (get-future)) (current-thread))))
(finalize (get-future) (let ((thread (current-thread))
(future-id *computing-future*))
(lambda () (abort-scheduled-future-task thread future-id))))
(let ((values (multiple-value-list (funcall (task (get-future))))))
(let ((values
(let ((*debugger-hook*
(lambda (c old-hook)
(declare (ignore old-hook))
(with-lock-held ((lock (get-future)))
(setf (restart-notifier (get-future)) (make-condition-variable :name "Eager Future2 restart proxy CV")
(error-descriptor (get-future)) (cons c (compute-restarts c)))
(loop (condition-wait (restart-notifier (get-future)) (lock (get-future)))
(let ((proxy-restart (proxy-restart (get-future))))
(when proxy-restart
(apply #'invoke-restart (car proxy-restart) (cdr proxy-restart)))))))))
(restart-case (multiple-value-list (funcall (task (get-future))))
(force-values (&rest values) values)))))
(apply #'force (get-future) values)))))))
(defun schedule-future (future future-type)
......
......@@ -5,16 +5,19 @@
;; making futures
#:*default-future-type*
#:pcall
#:pexec
#:plet
;; dealing with futures
#:ready-to-yield?
#:yield
#:select
#:force
#:force-values ;; restart
;; thread pool management
#:thread-pool-size
#:advise-thread-pool-size
;; helper functions
#:pexec
#:plet
))
......@@ -13,15 +13,18 @@
(lambda ()
(unwind-protect
(catch 'die
(loop (with-lock-held (*thread-pool-lock*)
(incf *free-threads*)
(ignore-errors
(funcall (loop (let ((task (pop *waiting-tasks*)))
(if task
(progn
(decf *free-threads*)
(return task))
(condition-wait *leader-notifier* *thread-pool-lock*)))))))))
(let ((*debugger-hook* (lambda (c old-hook)
(declare (ignore c old-hook))
(throw 'continue nil))))
(loop (with-lock-held (*thread-pool-lock*)
(incf *free-threads*)
(catch 'continue
(funcall (loop (let ((task (pop *waiting-tasks*)))
(if task
(progn
(decf *free-threads*)
(return task))
(condition-wait *leader-notifier* *thread-pool-lock*))))))))))
(with-lock-held (*thread-counter-lock*) (decf *total-threads*))))
:name "Eager Future2 Worker")
(with-lock-held (*thread-counter-lock*) (incf *total-threads*)))
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment