X-Git-Url: https://jxself.org/git/?p=8sync.git;a=blobdiff_plain;f=8sync%2Fagenda.scm;h=af5ef3ce31b516bbeb2d9df05404926e9d0fa336;hp=8ffe6f18a489a94520f0a1058f0d9a8e3ca1da8a;hb=e7161a37ba465d41bb9fde3ba1d3a7a3f4f8da42;hpb=2425363fe06d00cebd8f9c8ec9ae0a8c61716958 diff --git a/8sync/agenda.scm b/8sync/agenda.scm index 8ffe6f1..af5ef3c 100644 --- a/8sync/agenda.scm +++ b/8sync/agenda.scm @@ -1,21 +1,22 @@ -;; Copyright (C) 2015 Christopher Allan Webber - -;; This library is free software; you can redistribute it and/or -;; modify it under the terms of the GNU Lesser General Public -;; License as published by the Free Software Foundation; either -;; version 3 of the License, or (at your option) any later version. -;; -;; This library is distributed in the hope that it will be useful, -;; but WITHOUT ANY WARRANTY; without even the implied warranty of -;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -;; Lesser General Public License for more details. -;; -;; You should have received a copy of the GNU Lesser General Public -;; License along with this library; if not, write to the Free Software -;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA -;; 02110-1301 USA - -(define-module (eightsync agenda) +;;; 8sync --- Asynchronous programming for Guile +;;; Copyright (C) 2015 Christopher Allan Webber +;;; +;;; This file is part of 8sync. +;;; +;;; 8sync is free software: you can redistribute it and/or modify it +;;; under the terms of the GNU Lesser General Public License as +;;; published by the Free Software Foundation, either version 3 of the +;;; License, or (at your option) any later version. +;;; +;;; 8sync is distributed in the hope that it will be useful, +;;; but WITHOUT ANY WARRANTY; without even the implied warranty of +;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +;;; GNU Lesser General Public License for more details. +;;; +;;; You should have received a copy of the GNU Lesser General Public +;;; License along with 8sync. If not, see . + +(define-module (8sync agenda) #:use-module (srfi srfi-1) #:use-module (srfi srfi-9) #:use-module (srfi srfi-9 gnu) @@ -30,6 +31,8 @@ make-async-prompt-tag + list->q make-q* + make-time-segment time-segment? time-segment-time time-segment-queue @@ -50,8 +53,6 @@ schedule-segments-split schedule-extract-until! add-segments-contents-to-queue! - %sync 8sync - make-run-request run-request? run-request-proc run-request-when @@ -61,11 +62,28 @@ port-request-port port-request-read port-request-write port-request-except - run-it wrap run run-at run-delay + + 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 - 8port-request 8run 8run-at 8run-delay + %8sync %8sync-delay + %8sync-run %8sync-run-at %8sync-run-delay + %8sync-port %8sync-port-remove + catch-8sync catch-%8sync + + ;; used for introspecting the error, but a method for making + ;; is not exposed + wrapped-exception? + wrapped-exception-key wrapped-exception-args + wrapped-exception-stacks + + print-error-and-continue + + stop-on-nothing-to-do + %current-agenda start-agenda agenda-run-once)) @@ -93,7 +111,7 @@ (define-immutable-record-type (make-agenda-intern queue prompt-tag read-port-map write-port-map except-port-map - schedule time) + schedule time catch-handler pre-unwind-handler) agenda? (queue agenda-queue) (prompt-tag agenda-prompt-tag) @@ -101,7 +119,9 @@ (write-port-map agenda-write-port-map) (except-port-map agenda-except-port-map) (schedule agenda-schedule) - (time agenda-time)) + (time agenda-time) + (catch-handler agenda-catch-handler) + (pre-unwind-handler agenda-pre-unwind-handler)) (define (make-async-prompt-tag) "Make an async prompt tag for an agenda. @@ -116,12 +136,15 @@ Generally done automatically for the user through (make-agenda)." (write-port-map (make-hash-table)) (except-port-map (make-hash-table)) (schedule (make-schedule)) - (time (gettimeofday))) + (time (gettimeofday)) + (catch-handler #f) + (pre-unwind-handler #f)) ;; TODO: document arguments "Make a fresh agenda." (make-agenda-intern queue prompt read-port-map write-port-map except-port-map - schedule time)) + 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." @@ -132,6 +155,19 @@ Generally done automatically for the user through (make-agenda)." "Can't get current agenda prompt if there's no current agenda!") (agenda-prompt-tag current-agenda)))) +;; helper for making queues for an agenda +(define (list->q lst) + "Makes a queue composed of LST items" + (let ((q (make-q))) + (for-each + (lambda (x) + (enq! q x)) + lst) + q)) + +(define (make-q* . args) + "Makes a queue and populates it with this invocation's ARGS" + (list->q args)) ;;; Schedule @@ -150,6 +186,17 @@ 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. @@ -161,6 +208,8 @@ If an integer, will convert appropriately." (((? 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))) @@ -199,12 +248,14 @@ run (time-segment-right-format) first." (sec time-delta-sec) (usec time-delta-usec)) -(define* (make-time-delta sec #:optional (usec 0)) +(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 " - (make-time-delta-intern sec usec)) +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) @@ -233,13 +284,13 @@ Will produce (0 . 0) instead of a negative number, if needed." "Subtract TIME2 from TIME1" (time-carry-correct (cons (- (car time1) (car time2)) - (- (cdr time2) (cdr time2))))) + (- (cdr time1) (cdr time2))))) (define (time-plus time1 time2) "Add TIME1 and TIME2" (time-carry-correct (cons (+ (car time1) (car time2)) - (+ (cdr time2) (cdr time2))))) + (+ (cdr time1) (cdr time2))))) (define-record-type @@ -367,6 +418,12 @@ Will produce (0 . 0) instead of a negative number, if needed." (lambda () body ...)) +(define-syntax-rule (wrap-apply body) + "Wrap possibly multi-value function in a procedure, applies all arguments" + (lambda args + (apply body args))) + + ;; @@: Do we really want `body ...' here? ;; what about just `body'? (define-syntax-rule (run body ...) @@ -402,64 +459,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 ;;; ================================= -;; The future's in futures - -(define (make-future call-first on-success on-fail on-error) - ;; TODO: add error stuff here - (lambda () - (let ((call-result (call-first))) - ;; queue up calling the - (run (on-success call-result))))) - -(define (agenda-on-error agenda) - (const #f)) - -(define (agenda-on-fail agenda) - (const #f)) - -(define* (request-future call-first on-success - #:key - (agenda (%current-agenda)) - (on-fail (agenda-on-fail agenda)) - (on-error (agenda-on-error agenda)) - (when #f)) - ;; TODO: error handling - ;; do we need some distinction between expected, catchable errors, - ;; and unexpected, uncatchable ones? Probably...? - (make-run-request - (make-future call-first on-success on-fail on-error) - when)) - -(define-syntax-rule (%sync async-request) - "Run BODY asynchronously at a prompt, passing args to make-future. - -Pronounced `eight-sync' despite the spelling. - -%sync was chosen because (async) was already taken and could lead to -errors, and this version of asynchronous code uses a prompt, so the `a' -character becomes a `%' prompt! :) - -The % and 8 characters kind of look similar... hence this library's -name! (That, and the pun 'eight-synchronous' programming.) -There are 8sync aliases if you prefer that name." +(define-syntax-rule (%8sync-abort-to-prompt async-request) (abort-to-prompt (current-agenda-prompt) async-request)) -(define-syntax-rule (8sync args ...) - "Alias for %sync" - (%sync args ...)) - ;; 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) @@ -470,43 +491,134 @@ 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 (%sync) procedure." + "Invalid request passed back via an (%8sync) procedure." async-request)))) -(define-syntax-rule (%run body ...) - (%run-at body ... #f)) - -(define-syntax-rule (%run-at body ... when) - (make-async-request - (lambda (kont) - (make-run-request - (wrap - (kont - (begin body ...))) - 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 ...))))) - -;; Aliases -(define-syntax-rule (8run args ...) (%run args ...)) -(define-syntax-rule (8run-at args ...) (%run-at args ...)) -(define-syntax-rule (8run-delay args ...) (%run-delay args ...)) -(define-syntax-rule (8port-request args ...) (%port-request args ...)) +(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 %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-immediate 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 body #f)))))) + +(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 ...)) @@ -516,7 +628,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) @@ -539,10 +653,17 @@ return the wrong thing via (8sync) and trip themselves up." ;; TODO: support usecond wait time too (match (get-wait-time) ((sec . usec) - (select (hash-keys (agenda-read-port-map agenda)) - (hash-keys (agenda-write-port-map agenda)) - (hash-keys (agenda-except-port-map agenda)) - sec usec)))) + (catch 'system-error + (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 + ((_ _ _ (EINTR)) + '(() () ())) + (_ (error "Unhandled error in select!" key rest-args)))))))) (define (get-procs-to-run) (define (ports->procs ports port-map) (lambda (initial-procs) @@ -586,7 +707,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)) @@ -594,6 +717,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)))) @@ -602,8 +726,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 (get-time gettimeofday) (handle-ports update-agenda-from-select!)) ;; TODO: Document fields @@ -633,6 +776,42 @@ 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 + ((eq? key '8sync-caught-error) + (match args + ((orig-key orig-args stacks) + (display "\n*** Caught async exception. ***\n") + (format (current-error-port) + "* Original key '~s and arguments: ~s *\n" + orig-key orig-args) + (display "* Caught stacks below (ending with original) *\n\n") + (for-each + (lambda (s) + (display-backtrace s (current-error-port)) + (newline (current-error-port))) + stacks)))) + (else + (format (current-error-port) + "\n*** Caught exception with key '~s and arguments: ~s ***\n" + key args) + (display-backtrace (make-stack #t 1 0) + (current-error-port)) + (newline (current-error-port))))) + +(define-syntax-rule (maybe-catch-all (catch-handler pre-unwind-handler) + body ...) + (if (or catch-handler pre-unwind-handler) + (catch + #t + (lambda () + body ...) + (or catch-handler (lambda _ #f)) + (or pre-unwind-handler (lambda _ #f))) + (begin body ...))) + (define (agenda-run-once agenda) "Run once through the agenda, and produce a new agenda based on the results" @@ -641,7 +820,10 @@ based on the results" (agenda-prompt-tag agenda) (lambda () (parameterize ((%current-agenda agenda)) - (proc))) + (maybe-catch-all + ((agenda-catch-handler agenda) + (agenda-pre-unwind-handler agenda)) + (proc)))) (lambda (kont async-request) (setup-async-request kont async-request)))) @@ -668,11 +850,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