X-Git-Url: https://jxself.org/git/?p=8sync.git;a=blobdiff_plain;f=8sync%2Fagenda.scm;h=da578a77f25a0ad839ba80f379ded4bac102d697;hp=da398458db34096d2893d61286456ee9ad156ab1;hb=f8924934cac11ad55419bd4ba6c6732eefe05908;hpb=d059d67fdc1e547f6d55b2883e5b0cc2f14f0380 diff --git a/8sync/agenda.scm b/8sync/agenda.scm index da39845..da578a7 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. ;;; @@ -23,10 +23,11 @@ #: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 @@ -53,29 +54,17 @@ 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 - - - make-port-remove-request port-remove-request port-remove-request? - port-remove-request-port - run-it wrap wrap-apply run run-at run-delay - %run %run-at %run-delay %port-request %port-remove-request - %8sync-run %8sync-run-at %8sync-run-delay - %8sync-port %8sync-port-remove + 8sync-delay + 8sync-run 8sync-run-at 8sync-run-delay + 8sync + 8sleep - catch-8sync catch-%8sync - ;; used for introspecting the error, but a method for making ;; is not exposed wrapped-exception? @@ -112,14 +101,13 @@ (define-immutable-record-type (make-agenda-intern queue prompt-tag - read-port-map write-port-map except-port-map + read-port-map write-port-map schedule time catch-handler pre-unwind-handler) agenda? (queue 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) @@ -136,15 +124,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 + read-port-map write-port-map schedule time catch-handler pre-unwind-handler)) @@ -443,50 +430,20 @@ Will produce (0 . 0) instead of a negative number, if needed." (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)) - -(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)) - -(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) @@ -497,131 +454,36 @@ 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 port port-request-args ...) - (make-async-request - (lambda (kont) - (list (make-port-request port port-request-args ...) - (make-run-request kont #f))))) - -(define-syntax-rule (%port-remove-request port) - (make-async-request - (lambda (kont) - (list (make-port-remove-request port) - (make-run-request kont #f))))) - - -;; 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 ...))) - -(define-syntax-rule (%8sync-port-remove rest ...) - "Sugar for (%8sync (%port-remove-request ...))" - (%8sync (%port-remove-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 + ;; What's with returning #f to kont? + ;; Otherwise we sometimes get errors like + ;; "Zero values returned to single-valued continuation"" + (wrap (kont #f)) #f) + (make-run-request (lambda () body ...) #f)))))) + +;; TODO: Rewrite when we move to this being just `sleep'. +(define (8sleep time) + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (make-run-request (lambda () (kont #f)) (tdelta time)))))) + +;; Voluntarily yield execution +(define (yield) ; @@: should this be define-inlinable? + "Voluntarily yield execution to the scheduler." + (8sync-abort-to-prompt + (make-async-request + (lambda (kont) + (make-run-request (lambda () (kont #f)) #f))))) ;;; Execution of agenda, and current agenda @@ -659,7 +521,7 @@ Also handles sleeping when all we have to do is wait on the schedule." (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 @@ -671,23 +533,20 @@ Also handles sleeping when all we have to do is wait on the schedule." (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)) @@ -706,8 +565,7 @@ Also handles sleeping when all we have to do is wait on the schedule." (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 (or (ports-to-select?) ;; select doubles as sleep... @@ -715,50 +573,60 @@ Also handles sleeping when all we have to do is wait on the schedule." (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) - ;; @@: Should we remove if #f? - (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 (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-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))) - (= 0 (hash-count (const #t) (agenda-except-port-map agenda))))) + (= 0 (hash-count (const #t) (agenda-write-port-map agenda))))) (define* (start-agenda agenda #:key ;; @@: Should we make stop-on-nothing-to-do ;; the default stop-condition? - stop-condition + (stop-condition stop-on-nothing-to-do) (get-time gettimeofday) - (handle-ports update-agenda-from-select!)) + (handle-ports update-agenda-from-select!) + ;; For live hacking madness, etc + (post-run-hook #f)) ;; TODO: Document fields "Start up the AGENDA" + (install-suspendable-ports!) (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))) + ;; @@: This relies on mutation at present on the queue, in the rare + ;; event it's used. If we ever switch to something more immutable, + ;; it should return a new modified agenda instead. + (if post-run-hook + (post-run-hook agenda)) (if (and stop-condition (stop-condition agenda)) 'done (let* ((agenda @@ -814,6 +682,18 @@ Also handles sleeping when all we have to do is wait on the schedule." (or pre-unwind-handler (lambda _ #f))) (begin body ...))) +(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" @@ -821,7 +701,11 @@ based on the results" (call-with-prompt (agenda-prompt-tag agenda) (lambda () - (parameterize ((%current-agenda agenda)) + (parameterize ((%current-agenda agenda) + ;; @@: Couldn't we just parameterize this at the start of + ;; the agenda...? + (current-read-waiter wait-for-readable) + (current-write-waiter wait-for-writable)) (maybe-catch-all ((agenda-catch-handler agenda) (agenda-pre-unwind-handler agenda)) @@ -856,11 +740,15 @@ based on the results" (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)) + ((? 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