X-Git-Url: https://jxself.org/git/?p=8sync.git;a=blobdiff_plain;f=8sync%2Fagenda.scm;h=f20fffe889507a936a1e13df3d70f51c82c5a2ac;hp=c2614ef471cef514b14f3a7878a1523ea453c603;hb=ac6b7ab9cd36a5960f8289d6225a1c187debe777;hpb=9bb036f18457a6c2d24343e21e6950f9df5043e7 diff --git a/8sync/agenda.scm b/8sync/agenda.scm index c2614ef..f20fffe 100644 --- a/8sync/agenda.scm +++ b/8sync/agenda.scm @@ -53,8 +53,6 @@ schedule-segments-split schedule-extract-until! add-segments-contents-to-queue! - %8sync - make-run-request run-request? run-request-proc run-request-when @@ -64,11 +62,18 @@ port-request-port port-request-read port-request-write port-request-except + + make-port-remove-request port-remove-request port-remove-request? + port-remove-request-port + run-it wrap wrap-apply run run-at run-delay - %port-request %run %run-at %run-delay + 8sync 8sync-delay + 8sync-run 8sync-run-at 8sync-run-delay + 8sync-port 8sync-port-remove + 8sync-nowait - catch-8sync catch-%8sync + catch-8sync ;; used for introspecting the error, but a method for making ;; is not exposed @@ -78,6 +83,8 @@ print-error-and-continue + stop-on-nothing-to-do + %current-agenda start-agenda agenda-run-once)) @@ -132,7 +139,7 @@ Generally done automatically for the user through (make-agenda)." (schedule (make-schedule)) (time (gettimeofday)) (catch-handler #f) - (pre-unwind-handler #f)) + (pre-unwind-handler print-error-and-continue)) ;; TODO: document arguments "Make a fresh agenda." (make-agenda-intern queue prompt @@ -453,25 +460,28 @@ Will produce (0 . 0) instead of a negative number, if needed." (define port-request make-port-request) +(define-record-type + (make-port-remove-request port) + port-remove-request? + (port port-remove-request-port)) + +(define port-remove-request make-port-remove-request) + ;;; Asynchronous escape to run things ;;; ================================= -(define-syntax-rule (%8sync async-request) - "Run BODY asynchronously at a prompt, passing args to make-future. - -Runs things asynchronously (8synchronously?)" - (propagate-%async-exceptions - (abort-to-prompt (current-agenda-prompt) - async-request))) +(define-syntax-rule (8sync-abort-to-prompt async-request) + (abort-to-prompt (current-agenda-prompt) + async-request)) ;; Async port request and run-request meta-requests (define (make-async-request proc) "Wrap PROC in an async-request The purpose of this is to make sure that users don't accidentally -return the wrong thing via (%8sync) and trip themselves up." +return the wrong thing via (8sync) and trip themselves up." (cons '*async-request* proc)) (define (setup-async-request resume-kont async-request) @@ -482,7 +492,7 @@ return the wrong thing via (%8sync) and trip themselves up." ;; TODO: deliver more helpful errors depending on what the user ;; returned (_ (throw 'invalid-async-request - "Invalid request passed back via an (%8sync) procedure." + "Invalid request passed back via an (8sync) procedure." async-request)))) (define-record-type @@ -501,64 +511,99 @@ return the wrong thing via (%8sync) and trip themselves up." (wrapped-exception-stacks body-result)) body-result))) -(define-syntax-rule (%run body ...) - (%run-at body ... #f)) - -(define-syntax-rule (%run-at body ... when) - ;; Send an asynchronous request to apply a continuation to the - ;; following function, then handle that as a request to the agenda - (make-async-request - (lambda (kont) - ;; We're making a run request - (make-run-request - ;; Wrapping the following execution to run... - (wrap - ;; Once we get the result from the inner part, we'll resume - ;; this continuation, but first - ;; @@: Is this running immediately, or queueing the result - ;; after evaluation for the next agenda tick? It looks - ;; like evaluating immediately. Is that what we want? - (kont - ;; Any unhandled errors are caught - (let ((exception-stack #f)) - (catch #t - ;; Run the actual code the user requested - (lambda () - body ...) - ;; If something bad happened and we didn't catch it, - ;; we'll wrap it up in such a way that the continuation - ;; can address it - (lambda (key . args) - (cond - ((eq? key '8sync-caught-error) - (match args - ((orig-key orig-args orig-stacks) - (make-wrapped-exception - orig-key orig-args - (cons exception-stack orig-stacks))))) - (else - (make-wrapped-exception key args - (list exception-stack))))) - (lambda _ - (set! exception-stack (make-stack #t 1 0))))))) - when)))) - -(define-syntax-rule (%run-delay body ... delay-time) - (%run-at body ... (tdelta delay-time))) - -(define-syntax-rule (%port-request add-this-port port-request-args ...) - (make-async-request - (lambda (kont) - (list (make-port-request port-request-args ...) - (make-run-request kont))))) - -;; TODO -(define-syntax-rule (%run-with-return return body ...) - (make-async-request - (lambda (kont) - (let ((return kont)) - (lambda () - body ...))))) +(define-syntax 8sync + (syntax-rules () + "Run BODY asynchronously (8synchronously?) at a prompt, then return. + +Possibly specify WHEN as the second argument." + ((8sync body) + (8sync-run body)) + ((8sync body when) + (8sync-run-at body when)))) + +(define-syntax-rule (8sync-run body ...) + (8sync-run-at body ... #f)) + +(define-syntax-rule (8sync-run-at body ... when) + (propagate-%async-exceptions + (8sync-abort-to-prompt + ;; Send an asynchronous request to apply a continuation to the + ;; following function, then handle that as a request to the agenda + (make-async-request + (lambda (kont) + ;; We're making a run request + (make-run-request + ;; Wrapping the following execution to run... + (wrap + ;; Once we get the result from the inner part, we'll resume + ;; this continuation, but first + ;; @@: Is this running immediately, or queueing the result + ;; after evaluation for the next agenda tick? It looks + ;; like evaluating immediately. Is that what we want? + (kont + ;; Any unhandled errors are caught + (let ((exception-stack #f)) + (catch #t + ;; Run the actual code the user requested + (lambda () + body ...) + ;; If something bad happened and we didn't catch it, + ;; we'll wrap it up in such a way that the continuation + ;; can address it + (lambda (key . args) + (cond + ((eq? key '8sync-caught-error) + (match args + ((orig-key orig-args orig-stacks) + (make-wrapped-exception + orig-key orig-args + (cons exception-stack orig-stacks))))) + (else + (make-wrapped-exception key args + (list exception-stack))))) + (lambda _ + (set! exception-stack (make-stack #t 1 0))))))) + when)))))) + +(define-syntax-rule (8sync-run-delay body ... delay-time) + (8sync-run-at body ... (tdelta delay-time))) + +(define-syntax-rule (8sync-delay args ...) + (8sync-run-delay args ...)) + +(define-syntax-rule (8sync-port port port-request-args ...) + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (list (make-port-request port port-request-args ...) + (make-run-request + ;; What's with returning #f to kont? + ;; Otherwise we sometimes get errors like + ;; "Zero values returned to single-valued continuation"" + (wrap (kont #f)) #f)))))) + +(define-syntax-rule (8sync-port-remove port) + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (list (make-port-remove-request port) + (make-run-request + ;; See comment in 8sync-port + (wrap (kont #f)) #f)))))) + + +;; TODO: Write (%run-immediately) + +(define-syntax-rule (8sync-nowait body) + "Run body asynchronously but ignore its result... +forge ahead in our current function!" + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (list (make-run-request + ;; See comment in 8sync-port + (wrap (kont #f)) #f) + (make-run-request (lambda () body) #f)))))) (define-syntax-rule (catch-8sync exp (handler-key handler) ...) (catch '8sync-caught-error @@ -572,10 +617,6 @@ return the wrong thing via (%8sync) and trip themselves up." (else (raise '8sync-caught-error orig-key orig-args orig-stacks)))))) -;; Alias...? -(define-syntax-rule (catch-%8sync rest ...) - (catch-8sync rest ...)) - ;;; Execution of agenda, and current agenda @@ -584,7 +625,9 @@ return the wrong thing via (%8sync) and trip themselves up." (define %current-agenda (make-parameter #f)) (define (update-agenda-from-select! agenda) - "Potentially (select) on ports specified in agenda, adding items to queue" + "Potentially (select) on ports specified in agenda, adding items to queue. + +Also handles sleeping when all we have to do is wait on the schedule." (define (hash-keys hash) (hash-map->list (lambda (k v) k) hash)) (define (get-wait-time) @@ -661,7 +704,9 @@ return the wrong thing via (%8sync) and trip themselves up." (has-items? agenda-write-port-map) (has-items? agenda-except-port-map))) - (if (ports-to-select?) + (if (or (ports-to-select?) + ;; select doubles as sleep... + (not (schedule-empty? (agenda-schedule agenda)))) (update-agenda) agenda)) @@ -669,6 +714,7 @@ return the wrong thing via (%8sync) and trip themselves up." "Update an agenda for a port-request" (define (handle-selector request-selector port-map-selector) (if (request-selector port-request) + ;; @@: Should we remove if #f? (hash-set! (port-map-selector agenda) (port-request-port port-request) (request-selector port-request)))) @@ -677,8 +723,27 @@ return the wrong thing via (%8sync) and trip themselves up." (handle-selector port-request-except agenda-except-port-map)) +(define (agenda-handle-port-remove-request! agenda port-remove-request) + "Update an agenda for a port-remove-request" + (let ((port (port-remove-request-port port-remove-request))) + (hash-remove! (agenda-read-port-map agenda) port) + (hash-remove! (agenda-write-port-map agenda) port) + (hash-remove! (agenda-except-port-map agenda) port))) + + +(define (stop-on-nothing-to-do agenda) + (and (q-empty? (agenda-queue agenda)) + (schedule-empty? (agenda-schedule agenda)) + (= 0 (hash-count (const #t) (agenda-read-port-map agenda))) + (= 0 (hash-count (const #t) (agenda-write-port-map agenda))) + (= 0 (hash-count (const #t) (agenda-except-port-map agenda))))) + + (define* (start-agenda agenda - #:key stop-condition + #:key + ;; @@: Should we make stop-on-nothing-to-do + ;; the default stop-condition? + (stop-condition stop-on-nothing-to-do) (get-time gettimeofday) (handle-ports update-agenda-from-select!)) ;; TODO: Document fields @@ -708,6 +773,7 @@ return the wrong thing via (%8sync) and trip themselves up." (agenda-queue agenda)) (loop agenda)))))) + (define (print-error-and-continue key . args) "Frequently used as pre-unwind-handler for agenda" (cond @@ -781,11 +847,14 @@ based on the results" (#f (enq! next-queue (run-request-proc run-request)))))))) (define (handle-individual result) + ;; @@: Could maybe optimize by switching to an explicit cond... (match result ((? run-request? new-proc) (enqueue new-proc)) ((? port-request? port-request) (agenda-handle-port-request! agenda port-request)) + ((? port-remove-request? port-remove-request) + (agenda-handle-port-remove-request! agenda port-remove-request)) ;; do nothing (_ #f))) ;; @@: We might support delay-wrapped procedures here