X-Git-Url: https://jxself.org/git/?p=8sync.git;a=blobdiff_plain;f=8sync%2Fagenda.scm;h=4a126a4e1fddbcf140748f62b7b9861f7d5e03e2;hp=aded0d56f2230a84c31c11d8b23bec5c4f440377;hb=2255f6996a4fbe708b235760816a6dd49959fb6d;hpb=ed0ef531945b62463f3dcd529930b73b06fc7ba4 diff --git a/8sync/agenda.scm b/8sync/agenda.scm index aded0d5..4a126a4 100644 --- a/8sync/agenda.scm +++ b/8sync/agenda.scm @@ -1,5 +1,5 @@ ;;; 8sync --- Asynchronous programming for Guile -;;; Copyright (C) 2015 Christopher Allan Webber +;;; Copyright (C) 2015, 2016 Christopher Allan Webber ;;; ;;; This file is part of 8sync. ;;; @@ -19,31 +19,20 @@ (define-module (8sync agenda) #:use-module (srfi srfi-1) #:use-module (srfi srfi-9) - #:use-module (srfi srfi-9 gnu) #:use-module (ice-9 q) #:use-module (ice-9 match) #:use-module (ice-9 receive) + #:use-module (ice-9 suspendable-ports) #:export ( make-agenda agenda? agenda-queue agenda-prompt-tag - agenda-read-port-map agenda-write-port-map agenda-except-port-map + agenda-read-port-map agenda-write-port-map agenda-schedule make-async-prompt-tag list->q make-q* - - make-time-segment time-segment? - time-segment-time time-segment-queue - - time< time= time<= time-delta+ - time-minus time-plus - - - make-time-delta tdelta time-delta? - time-delta-sec time-delta-usec - make-schedule schedule? schedule-add! schedule-empty? @@ -53,24 +42,15 @@ schedule-segments-split schedule-extract-until! add-segments-contents-to-queue! - %8sync - make-run-request run-request? run-request-proc run-request-when - - make-port-request port-request port-request? - port-request-port - port-request-read port-request-write port-request-except - run-it wrap wrap-apply run run-at run-delay - %run %run-at %run-delay %port-request - %8sync-run %8sync-run-at %8sync-run-delay %8sync-port + 8sync + 8sleep 8yield - catch-8sync catch-%8sync - ;; used for introspecting the error, but a method for making ;; is not exposed wrapped-exception? @@ -79,11 +59,10 @@ print-error-and-continue - %current-agenda - start-agenda agenda-run-once)) + stop-on-nothing-to-do -;; @@: Using immutable agendas here, so wouldn't it make sense to -;; replace this queue stuff with using pfds based immutable queues? + %current-agenda-prompt + run-agenda agenda-run-once!)) ;;; Agenda definition @@ -96,25 +75,26 @@ ;;; operation (from which they can be returned later) ;;; - a mapping of ports to various handler procedures ;;; -;;; The goal, eventually, is for this all to be immutable and functional. +;;; @@: Is this next part deprecated? +;;; The goal, maybe eventually, is for this all to be immutable and functional. ;;; However, we aren't there yet. Some tricky things: ;;; - The schedule needs to be immutable, yet reasonably efficient. ;;; - Need to use immutable queues (ijp's pfds library?) ;;; - Modeling reading from ports as something repeatable, ;;; and with reasonable separation from functional components? -(define-immutable-record-type +;; TODO: Tear out the immutable agenda aspect until we're actually ready +;; to use it. +(define-record-type (make-agenda-intern queue prompt-tag - read-port-map write-port-map except-port-map - schedule time catch-handler pre-unwind-handler) + read-port-map write-port-map + schedule catch-handler pre-unwind-handler) agenda? - (queue agenda-queue) + (queue agenda-queue set-agenda-queue!) (prompt-tag agenda-prompt-tag) (read-port-map agenda-read-port-map) (write-port-map agenda-write-port-map) - (except-port-map agenda-except-port-map) (schedule agenda-schedule) - (time agenda-time) (catch-handler agenda-catch-handler) (pre-unwind-handler agenda-pre-unwind-handler)) @@ -129,26 +109,14 @@ Generally done automatically for the user through (make-agenda)." (prompt (make-prompt-tag)) (read-port-map (make-hash-table)) (write-port-map (make-hash-table)) - (except-port-map (make-hash-table)) (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 - read-port-map write-port-map except-port-map - schedule time - catch-handler pre-unwind-handler)) - -(define (current-agenda-prompt) - "Get the prompt for the current agenda; signal an error if there isn't one." - (let ((current-agenda (%current-agenda))) - (if (not current-agenda) - (throw - 'no-current-agenda - "Can't get current agenda prompt if there's no current agenda!") - (agenda-prompt-tag current-agenda)))) + read-port-map write-port-map + schedule catch-handler pre-unwind-handler)) ;; helper for making queues for an agenda (define (list->q lst) @@ -181,32 +149,6 @@ Generally done automatically for the user through (make-agenda)." (time time-segment-time) (queue time-segment-queue)) -;; @@: This seems to be the same as srfi-18's seconds->time procedure? -;; Maybe double check and switch to that? (Thanks amz3!) - -(define (time-from-float-or-fraction time) - "Produce a (sec . usec) pair from TIME, a float or fraction" - (let* ((mixed-whole (floor time)) - (mixed-rest (- time mixed-whole)) ; float or fraction component - (sec mixed-whole) - (usec (floor (* 1000000 mixed-rest)))) - (cons (inexact->exact sec) (inexact->exact usec)))) - -(define (time-segment-right-format time) - "Ensure TIME is in the right format. - -The right format means (second . microsecond). -If an integer, will convert appropriately." - ;; TODO: add floating point / rational number support. - (match time - ;; time is already a cons of second and microsecnd - (((? integer? s) . (? integer? u)) time) - ;; time was just an integer (just the second) - ((? integer? _) (cons time 0)) - ((or (? rational? _) (? inexact? _)) - (time-from-float-or-fraction time)) - (_ (throw 'invalid-time "Invalid time" time)))) - (define* (make-time-segment time #:optional (queue (make-q))) "Make a time segment of TIME and QUEUE @@ -236,24 +178,7 @@ run (time-segment-right-format) first." (or (time< time1 time2) (time= time1 time2))) - -(define-record-type - (make-time-delta-intern sec usec) - time-delta? - (sec time-delta-sec) - (usec time-delta-usec)) - -(define* (make-time-delta time) - "Make a of SEC seconds and USEC microseconds. - -This is used primarily so the agenda can recognize RUN-REQUEST objects -which are meant to delay computation" - (match (time-segment-right-format time) - ((sec . usec) - (make-time-delta-intern sec usec)))) - -(define tdelta make-time-delta) - +;; @@: Maybe we should use floor/ here? (define (time-carry-correct time) "Corrects/handles time microsecond carry. Will produce (0 . 0) instead of a negative number, if needed." @@ -269,25 +194,19 @@ Will produce (0 . 0) instead of a negative number, if needed." (+ (cdr time) 1000000)))) (else time))) -(define (time-delta+ time time-delta) - "Increment a TIME by the value of TIME-DELTA" - (time-carry-correct - (cons (+ (car time) (time-delta-sec time-delta)) - (+ (cdr time) (time-delta-usec time-delta))))) - (define (time-minus time1 time2) "Subtract TIME2 from TIME1" (time-carry-correct (cons (- (car time1) (car time2)) (- (cdr time1) (cdr time2))))) +;; @@: Unused? (define (time-plus time1 time2) "Add TIME1 and TIME2" (time-carry-correct (cons (+ (car time1) (car time2)) (+ (cdr time1) (cdr time2))))) - (define-record-type (make-schedule-intern segments) schedule? @@ -310,41 +229,40 @@ Will produce (0 . 0) instead of a negative number, if needed." ;; a more functional setup? (define (schedule-add! schedule time proc) "Mutate SCHEDULE, adding PROC at an appropriate time segment for TIME" - (let ((time (time-segment-right-format time))) - (define (new-time-segment) - (let ((new-segment - (make-time-segment time))) - (enq! (time-segment-queue new-segment) proc) - new-segment)) - (define (loop segments) - (define (segment-equals-time? segment) - (time= time (time-segment-time segment))) - - (define (segment-more-than-time? segment) - (time< time (time-segment-time segment))) - - ;; We could switch this out to be more mutate'y - ;; and avoid the O(n) of space... is that over-optimizing? - (match segments - ;; If we're at the end of the list, time to make a new - ;; segment... - ('() (cons (new-time-segment) '())) - ;; If the segment's time is exactly our time, good news - ;; everyone! Let's append our stuff to its queue - (((? segment-equals-time? first) rest ...) - (enq! (time-segment-queue first) proc) - segments) - ;; If the first segment is more than our time, - ;; ours belongs before this one, so add it and - ;; start consing our way back - (((? segment-more-than-time? first) rest ...) - (cons (new-time-segment) segments)) - ;; Otherwise, build up recursive result - ((first rest ... ) - (cons first (loop rest))))) - (set-schedule-segments! - schedule - (loop (schedule-segments schedule))))) + (define (new-time-segment) + (let ((new-segment + (make-time-segment time))) + (enq! (time-segment-queue new-segment) proc) + new-segment)) + (define (loop segments) + (define (segment-equals-time? segment) + (time= time (time-segment-time segment))) + + (define (segment-more-than-time? segment) + (time< time (time-segment-time segment))) + + ;; We could switch this out to be more mutate'y + ;; and avoid the O(n) of space... is that over-optimizing? + (match segments + ;; If we're at the end of the list, time to make a new + ;; segment... + ('() (cons (new-time-segment) '())) + ;; If the segment's time is exactly our time, good news + ;; everyone! Let's append our stuff to its queue + (((? segment-equals-time? first) rest ...) + (enq! (time-segment-queue first) proc) + segments) + ;; If the first segment is more than our time, + ;; ours belongs before this one, so add it and + ;; start consing our way back + (((? segment-more-than-time? first) rest ...) + (cons (new-time-segment) segments)) + ;; Otherwise, build up recursive result + ((first rest ... ) + (cons first (loop rest))))) + (set-schedule-segments! + schedule + (loop (schedule-segments schedule)))) (define (schedule-empty? schedule) "Check if the SCHEDULE is currently empty" @@ -352,31 +270,30 @@ Will produce (0 . 0) instead of a negative number, if needed." (define (schedule-segments-split schedule time) "Does a multiple value return of time segments before/at and after TIME" - (let ((time (time-segment-right-format time))) - (define (segment-is-now? segment) - (time= (time-segment-time segment) time)) - (define (segment-is-before-now? segment) - (time< (time-segment-time segment) time)) - - (let loop ((segments-before '()) - (segments-left (schedule-segments schedule))) - (match segments-left - ;; end of the line, return - ('() - (values (reverse segments-before) '())) - - ;; It's right now, so time to stop, but include this one in before - ;; but otherwise return - (((? segment-is-now? first) rest ...) - (values (reverse (cons first segments-before)) rest)) - - ;; This is prior or at now, so add it and keep going - (((? segment-is-before-now? first) rest ...) - (loop (cons first segments-before) rest)) - - ;; Otherwise it's past now, just return what we have - (segments-after - (values segments-before segments-after)))))) + (define (segment-is-now? segment) + (time= (time-segment-time segment) time)) + (define (segment-is-before-now? segment) + (time< (time-segment-time segment) time)) + + (let loop ((segments-before '()) + (segments-left (schedule-segments schedule))) + (match segments-left + ;; end of the line, return + ('() + (values (reverse segments-before) '())) + + ;; It's right now, so time to stop, but include this one in before + ;; but otherwise return + (((? segment-is-now? first) rest ...) + (values (reverse (cons first segments-before)) rest)) + + ;; This is prior or at now, so add it and keep going + (((? segment-is-before-now? first) rest ...) + (loop (cons first segments-before) rest)) + + ;; Otherwise it's past now, just return what we have + (segments-after + (values segments-before segments-after))))) (define (schedule-extract-until! schedule time) "Extract all segments until TIME from SCHEDULE, and pop old segments off" @@ -429,50 +346,35 @@ Will produce (0 . 0) instead of a negative number, if needed." "Run BODY at WHEN" (make-run-request (wrap body ...) when)) -;; @@: Is it okay to overload the term "delay" like this? -;; Would `run-in' be better? (define-syntax-rule (run-delay body ... delay-time) "Run BODY at DELAY-TIME time from now" - (make-run-request (wrap body ...) (tdelta delay-time))) - - -;; A request to set up a port with at least one of read, write, except -;; handling processes - -(define-record-type - (make-port-request-intern port read write except) - port-request? - (port port-request-port) - (read port-request-read) - (write port-request-write) - (except port-request-except)) + (make-run-request (wrap body ...) (delayed-time delay-time))) -(define* (make-port-request port #:key read write except) - (if (not (or read write except)) - (throw 'no-port-handler-given "No port handler given.\n")) - (make-port-request-intern port read write except)) +;;; Procedures that are delimited continuations being resumed are +;;; wrapped in a . This is so that we don't accidentally +;;; wrap in another catch statement every time we resume them, which +;;; would be a drag. -(define port-request make-port-request) +(define-record-type + (kontinue kont) + kontinue? + (kont kontinue-kont)) ;;; 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) @@ -483,129 +385,65 @@ 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 - (make-wrapped-exception key args stacks) - wrapped-exception? - (key wrapped-exception-key) - (args wrapped-exception-args) - (stacks wrapped-exception-stacks)) - -(define-syntax-rule (propagate-%async-exceptions body) - (let ((body-result body)) - (if (wrapped-exception? body-result) - (throw '8sync-caught-error - (wrapped-exception-key body-result) - (wrapped-exception-args body-result) - (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))))) - -;; Sugar -(define-syntax-rule (%8sync-run rest ...) - "Sugar for (%8sync (%run ...))" - (%8sync (%run rest ...))) - -(define-syntax-rule (%8sync-run-at rest ...) - "Sugar for (%8sync (%run-at ...))" - (%8sync (%run-at rest ...))) - -(define-syntax-rule (%8sync-run-delay rest ...) - "Sugar for (%8sync (%run-delay ...))" - (%8sync (%run-delay rest ...))) - -(define-syntax-rule (%8sync-port rest ...) - "Sugar for (%8sync (%port-request ...))" - (%8sync (%port-request rest ...))) - - -;; TODO: Write (%run-immediately) - -;; TODO -(define-syntax-rule (%run-with-return return body ...) - (make-async-request - (lambda (kont) - (let ((return kont)) - (lambda () - body ...))))) - -(define-syntax-rule (catch-8sync exp (handler-key handler) ...) - (catch '8sync-caught-error - (lambda () - exp) - (lambda (_ orig-key orig-args orig-stacks) - (cond - ((or (eq? handler-key #t) - (eq? orig-key handler-key)) - (apply handler orig-stacks orig-args)) ... - (else (raise '8sync-caught-error - orig-key orig-args orig-stacks)))))) - -;; Alias...? -(define-syntax-rule (catch-%8sync rest ...) - (catch-8sync rest ...)) - +(define-syntax-rule (8sync 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 + (kontinue kont) #f) + (make-run-request (lambda () body ...) #f)))))) + +(define (delayed-time in-secs) + "Calculate a cons of '(sec . usec) of IN-SECS from current time" + (define cur-time (gettimeofday)) + (define cur-secs (car cur-time)) + (define cur-usecs (cdr cur-time)) + (define (convert-non-integer) + (define next-time-in-usecs + (+ (* (+ in-secs cur-secs) ; add our seconds to current seconds + 1000000) ; and turn into usecs + cur-usecs)) ; then add in current usecs + ;; convert into sec / usec pair + (receive (secs usecs) + (floor/ next-time-in-usecs 1000000) + (cons secs (floor usecs)))) + (define (convert-integer) + (cons (+ in-secs cur-secs) cur-usecs)) + (if (integer? in-secs) + (convert-integer) + (convert-non-integer))) + +;; TODO: Rewrite when we move to this being just `sleep'. +(define (8sleep secs) + "Like sleep, but asynchronous." + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (make-run-request (kontinue kont) (delayed-time secs)))))) + +;; Voluntarily yield execution +(define (8yield) + "Voluntarily yield execution to the scheduler." + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (make-run-request (kontinue kont) #f))))) ;;; Execution of agenda, and current agenda ;;; ======================================= -(define %current-agenda (make-parameter #f)) +(define %current-agenda-prompt (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) @@ -616,7 +454,7 @@ return the wrong thing via (%8sync) and trip themselves up." ((not (q-empty? (agenda-queue agenda))) (cons 0 0)) (soonest-time ; ie, the agenda is non-empty - (let* ((current-time (agenda-time agenda))) + (let* ((current-time (gettimeofday))) (if (time<= soonest-time current-time) ;; Well there's something due so let's select ;; (this avoids a (possible?) race condition chance) @@ -632,7 +470,7 @@ return the wrong thing via (%8sync) and trip themselves up." (lambda () (select (hash-keys (agenda-read-port-map agenda)) (hash-keys (agenda-write-port-map agenda)) - (hash-keys (agenda-except-port-map agenda)) + '() sec usec)) (lambda (key . rest-args) (match rest-args @@ -644,23 +482,20 @@ return the wrong thing via (%8sync) and trip themselves up." (lambda (initial-procs) (fold (lambda (port prev) - (cons (lambda () - ((hash-ref port-map port) port)) - prev)) + (define proc (hashq-ref port-map port)) + ;; Now that we've selected on this port, it can be removed + (hashq-remove! port-map port) + (cons proc prev)) initial-procs ports))) (match (do-select) - ((read-ports write-ports except-ports) - ;; @@: Come on, we can do better than append ;P + ((read-ports write-ports except-ports) ; except-ports not used ((compose (ports->procs read-ports (agenda-read-port-map agenda)) (ports->procs write-ports - (agenda-write-port-map agenda)) - (ports->procs - except-ports - (agenda-except-port-map agenda))) + (agenda-write-port-map agenda))) '())))) (define (update-agenda) (let ((procs-to-run (get-procs-to-run)) @@ -679,55 +514,64 @@ return the wrong thing via (%8sync) and trip themselves up." (selector agenda)) 0))) (or (has-items? agenda-read-port-map) - (has-items? agenda-write-port-map) - (has-items? agenda-except-port-map))) + (has-items? agenda-write-port-map))) - (if (ports-to-select?) + (if (or (ports-to-select?) + ;; select doubles as sleep... + (not (schedule-empty? (agenda-schedule agenda)))) (update-agenda) agenda)) -(define (agenda-handle-port-request! agenda port-request) - "Update an agenda for a port-request" - (define (handle-selector request-selector port-map-selector) - (if (request-selector port-request) - (hash-set! (port-map-selector agenda) - (port-request-port port-request) - (request-selector port-request)))) - (handle-selector port-request-read agenda-read-port-map) - (handle-selector port-request-write agenda-write-port-map) - (handle-selector port-request-except agenda-except-port-map)) - - -(define* (start-agenda agenda - #:key stop-condition - (get-time gettimeofday) - (handle-ports update-agenda-from-select!)) +(define-record-type + (make-read-request port proc) + read-request? + (port read-request-port) + (proc read-request-proc)) + +(define-record-type + (make-write-request port proc) + write-request? + (port write-request-port) + (proc write-request-proc)) + +(define (agenda-handle-read-request! agenda read-request) + "Handle , which is a request to add this port to the poll/select +on suspendable ports." + (hashq-set! (agenda-read-port-map agenda) + (read-request-port read-request) + (read-request-proc read-request))) + +(define (agenda-handle-write-request! agenda write-request) + (hashq-set! (agenda-write-port-map agenda) + (write-request-port write-request) + (write-request-proc write-request))) + +(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))))) + + +(define* (run-agenda agenda + #:key (stop-condition stop-on-nothing-to-do) + ;; For live hacking madness, etc + (post-run-hook #f)) ;; TODO: Document fields "Start up the AGENDA" - (let loop ((agenda agenda)) - (let ((agenda - ;; @@: Hm, maybe here would be a great place to handle - ;; select'ing on ports. - ;; We could compose over agenda-run-once and agenda-read-ports - (agenda-run-once agenda))) - (if (and stop-condition (stop-condition agenda)) - 'done - (let* ((agenda - ;; We have to update the time after ports handled, too - ;; because it may have changed after a select - (set-field - (handle-ports - ;; Adjust the agenda's time just in time - ;; We do this here rather than in agenda-run-once to make - ;; agenda-run-once's behavior fairly predictable - (set-field agenda (agenda-time) (get-time))) - (agenda-time) (get-time)))) - ;; Update the agenda's current queue based on - ;; currently applicable time segments - (add-segments-contents-to-queue! - (schedule-extract-until! (agenda-schedule agenda) (agenda-time agenda)) - (agenda-queue agenda)) - (loop agenda)))))) + (install-suspendable-ports!) + (parameterize ((%current-agenda-prompt (agenda-prompt-tag agenda)) + (current-read-waiter wait-for-readable) + (current-write-waiter wait-for-writable)) + (while (not (stop-condition agenda)) + (agenda-run-once! agenda) + (update-agenda-from-select! agenda) + (add-segments-contents-to-queue! + (schedule-extract-until! (agenda-schedule agenda) (gettimeofday)) + (agenda-queue agenda)) + (if post-run-hook + (post-run-hook agenda)))) + 'done) (define (print-error-and-continue key . args) "Frequently used as pre-unwind-handler for agenda" @@ -764,18 +608,36 @@ return the wrong thing via (%8sync) and trip themselves up." (or pre-unwind-handler (lambda _ #f))) (begin body ...))) -(define (agenda-run-once agenda) +(define (wait-for-readable port) + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (make-read-request port (wrap (kont #f))))))) + +(define (wait-for-writable port) + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (make-write-request port (wrap (kont #f))))))) + +(define (agenda-run-once! agenda) "Run once through the agenda, and produce a new agenda based on the results" + ;; @@: Maybe rename proc -> run-this ? (define (call-proc proc) (call-with-prompt (agenda-prompt-tag agenda) (lambda () - (parameterize ((%current-agenda agenda)) - (maybe-catch-all - ((agenda-catch-handler agenda) - (agenda-pre-unwind-handler agenda)) - (proc)))) + (if (kontinue? proc) + ;; Resume continuation. + ;; We need to pass in #f, otherwise we sometimes get errors like + ;; "Zero values returned to single-valued continuation"" + ((kontinue-kont proc) #f) + ;; Otherwise call the procedure and set up error catching. + (maybe-catch-all + ((agenda-catch-handler agenda) + (agenda-pre-unwind-handler agenda)) + (proc)))) (lambda (kont async-request) (setup-async-request kont async-request)))) @@ -786,35 +648,29 @@ based on the results" (proc-result (call-proc proc)) (enqueue (lambda (run-request) - (define (schedule-at! time proc) - (schedule-add! (agenda-schedule agenda) time proc)) (let ((request-time (run-request-when run-request))) - (match request-time - ((? time-delta? time-delta) - (let ((time (time-delta+ (agenda-time agenda) - time-delta))) - (schedule-at! time (run-request-proc run-request)))) - ((? integer? sec) - (let ((time (cons sec 0))) - (schedule-at! time (run-request-proc run-request)))) - (((? integer? sec) . (? integer? usec)) - (schedule-at! request-time (run-request-proc run-request))) - (#f - (enq! next-queue (run-request-proc run-request)))))))) + (if request-time + (schedule-add! (agenda-schedule agenda) request-time + (run-request-proc run-request)) + (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)) + ((? read-request? read-request) + (agenda-handle-read-request! agenda read-request)) + ((? write-request? write-request) + (agenda-handle-write-request! agenda write-request)) ;; do nothing + ;; Remember, we don't throw an error here because procedures can + ;; return a run request, eg with run-it, at the end of their + ;; evaluation to keep looping. + ;; @@: Though is this really a useful feature? (_ #f))) ;; @@: We might support delay-wrapped procedures here (match proc-result ((results ...) (for-each handle-individual results)) (one-result (handle-individual one-result))))) - ;; TODO: Alternately, we could return the next-queue - ;; along with changes to be added to the schedule here? - ;; Return new agenda, with next queue set - (set-field agenda (agenda-queue) next-queue))) + (set-agenda-queue! agenda next-queue)))