8sync/agenda.scm

   1 ;;; 8sync --- Asynchronous programming for Guile
   2 ;;; Copyright (C) 2015, 2016 Christopher Allan Webber <cwebber@dustycloud.org>
   3 ;;;
   4 ;;; This file is part of 8sync.
   5 ;;;
   6 ;;; 8sync is free software: you can redistribute it and/or modify it
   7 ;;; under the terms of the GNU Lesser General Public License as
   8 ;;; published by the Free Software Foundation, either version 3 of the
   9 ;;; License, or (at your option) any later version.
  10 ;;;
  11 ;;; 8sync is distributed in the hope that it will be useful,
  12 ;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
  13 ;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14 ;;; GNU Lesser General Public License for more details.
  15 ;;;
  16 ;;; You should have received a copy of the GNU Lesser General Public
  17 ;;; License along with 8sync.  If not, see <http://www.gnu.org/licenses/>.
  18
  19 (define-module (8sync agenda)
  20   #:use-module (srfi srfi-1)
  21   #:use-module (srfi srfi-9)
  22   #:use-module (srfi srfi-9 gnu)
  23   #:use-module (ice-9 q)
  24   #:use-module (ice-9 match)
  25   #:use-module (ice-9 receive)
  26   #:use-module (ice-9 suspendable-ports)
  27   #:export (<agenda>
  28             make-agenda agenda?
  29             agenda-queue agenda-prompt-tag
  30             agenda-read-port-map agenda-write-port-map
  31             agenda-schedule
  32
  33             make-async-prompt-tag
  34
  35             list->q make-q*
  36
  37             <time-segment>
  38             make-time-segment time-segment?
  39             time-segment-time time-segment-queue
  40
  41             time< time= time<= time-delta+
  42             time-minus time-plus
  43
  44             <time-delta>
  45             make-time-delta tdelta time-delta?
  46             time-delta-sec time-delta-usec
  47
  48             <schedule>
  49             make-schedule schedule?
  50             schedule-add! schedule-empty?
  51             schedule-segments
  52             schedule-soonest-time
  53
  54             schedule-segments-split schedule-extract-until!
  55             add-segments-contents-to-queue!
  56
  57             <run-request>
  58             make-run-request run-request?
  59             run-request-proc run-request-when
  60
  61             run-it wrap wrap-apply run run-at run-delay
  62
  63             8sync-delay
  64             8sync-run 8sync-run-at 8sync-run-delay
  65             8sync
  66             8sleep
  67
  68             ;; used for introspecting the error, but a method for making
  69             ;; is not exposed
  70             wrapped-exception?
  71             wrapped-exception-key wrapped-exception-args
  72             wrapped-exception-stacks
  73
  74             print-error-and-continue
  75
  76             stop-on-nothing-to-do
  77
  78             %current-agenda
  79             start-agenda agenda-run-once))
  80
  81 (install-suspendable-ports!)
  82
  83 ;; @@: Using immutable agendas here, so wouldn't it make sense to
  84 ;;   replace this queue stuff with using pfds based immutable queues?
  85
  86 \f
  87 ;;; Agenda definition
  88 ;;; =================
  89
  90 ;;; The agenda consists of:
  91 ;;;  - a queue of immediate items to handle
  92 ;;;  - sheduled future events to be added to a future queue
  93 ;;;  - a tag by which running processes can escape for some asynchronous
  94 ;;;    operation (from which they can be returned later)
  95 ;;;  - a mapping of ports to various handler procedures
  96 ;;;
  97 ;;; The goal, eventually, is for this all to be immutable and functional.
  98 ;;; However, we aren't there yet.  Some tricky things:
  99 ;;;  - The schedule needs to be immutable, yet reasonably efficient.
 100 ;;;  - Need to use immutable queues (ijp's pfds library?)
 101 ;;;  - Modeling reading from ports as something repeatable,
 102 ;;;    and with reasonable separation from functional components?
 103
 104 (define-immutable-record-type <agenda>
 105   (make-agenda-intern queue prompt-tag
 106                       read-port-map write-port-map
 107                       schedule time catch-handler pre-unwind-handler)
 108   agenda?
 109   (queue agenda-queue)
 110   (prompt-tag agenda-prompt-tag)
 111   (read-port-map agenda-read-port-map)
 112   (write-port-map agenda-write-port-map)
 113   (schedule agenda-schedule)
 114   (time agenda-time)
 115   (catch-handler agenda-catch-handler)
 116   (pre-unwind-handler agenda-pre-unwind-handler))
 117
 118 (define (make-async-prompt-tag)
 119   "Make an async prompt tag for an agenda.
 120
 121 Generally done automatically for the user through (make-agenda)."
 122   (make-prompt-tag "prompt"))
 123
 124 (define* (make-agenda #:key
 125                       (queue (make-q))
 126                       (prompt (make-prompt-tag))
 127                       (read-port-map (make-hash-table))
 128                       (write-port-map (make-hash-table))
 129                       (schedule (make-schedule))
 130                       (time (gettimeofday))
 131                       (catch-handler #f)
 132                       (pre-unwind-handler print-error-and-continue))
 133   ;; TODO: document arguments
 134   "Make a fresh agenda."
 135   (make-agenda-intern queue prompt
 136                       read-port-map write-port-map
 137                       schedule time
 138                       catch-handler pre-unwind-handler))
 139
 140 (define (current-agenda-prompt)
 141   "Get the prompt for the current agenda; signal an error if there isn't one."
 142   (let ((current-agenda (%current-agenda)))
 143     (if (not current-agenda)
 144         (throw
 145          'no-current-agenda
 146          "Can't get current agenda prompt if there's no current agenda!")
 147         (agenda-prompt-tag current-agenda))))
 148
 149 ;; helper for making queues for an agenda
 150 (define (list->q lst)
 151   "Makes a queue composed of LST items"
 152   (let ((q (make-q)))
 153     (for-each
 154      (lambda (x)
 155        (enq! q x))
 156      lst)
 157     q))
 158
 159 (define (make-q* . args)
 160   "Makes a queue and populates it with this invocation's ARGS"
 161   (list->q args))
 162
 163 \f
 164 ;;; Schedule
 165 ;;; ========
 166
 167 ;;; This is where we handle timed events for the future
 168
 169 ;; This section totally borrows from the ideas in SICP
 170 ;; <3 <3 <3
 171
 172 ;; NOTE: time is a cons of (seconds . microseconds)
 173
 174 (define-record-type <time-segment>
 175   (make-time-segment-intern time queue)
 176   time-segment?
 177   (time time-segment-time)
 178   (queue time-segment-queue))
 179
 180 ;; @@: This seems to be the same as srfi-18's seconds->time procedure?
 181 ;;   Maybe double check and switch to that?  (Thanks amz3!)
 182
 183 (define (time-from-float-or-fraction time)
 184   "Produce a (sec . usec) pair from TIME, a float or fraction"
 185   (let* ((mixed-whole (floor time))
 186          (mixed-rest (- time mixed-whole))  ; float or fraction component
 187          (sec mixed-whole)
 188          (usec (floor (* 1000000 mixed-rest))))
 189     (cons (inexact->exact sec) (inexact->exact usec))))
 190
 191 (define (time-segment-right-format time)
 192   "Ensure TIME is in the right format.
 193
 194 The right format means (second . microsecond).
 195 If an integer, will convert appropriately."
 196   ;; TODO: add floating point / rational number support.
 197   (match time
 198     ;; time is already a cons of second and microsecnd
 199     (((? integer? s) . (? integer? u)) time)
 200     ;; time was just an integer (just the second)
 201     ((? integer? _) (cons time 0))
 202     ((or (? rational? _) (? inexact? _))
 203      (time-from-float-or-fraction time))
 204     (_ (throw 'invalid-time "Invalid time" time))))
 205
 206 (define* (make-time-segment time #:optional (queue (make-q)))
 207   "Make a time segment of TIME and QUEUE
 208
 209 No automatic conversion is done, so you might have to
 210 run (time-segment-right-format) first."
 211   (make-time-segment-intern time queue))
 212
 213 (define (time< time1 time2)
 214   "Check if TIME1 is less than TIME2"
 215   (cond ((< (car time1)
 216             (car time2))
 217          #t)
 218         ((and (= (car time1)
 219                  (car time2))
 220               (< (cdr time1)
 221                  (cdr time2)))
 222          #t)
 223         (else #f)))
 224
 225 (define (time= time1 time2)
 226   "Check whether TIME1 and TIME2 are equivalent"
 227   (and (= (car time1) (car time2))
 228        (= (cdr time1) (cdr time2))))
 229
 230 (define (time<= time1 time2)
 231   "Check if TIME1 is less than or equal to TIME2"
 232   (or (time< time1 time2)
 233       (time= time1 time2)))
 234
 235
 236 (define-record-type <time-delta>
 237   (make-time-delta-intern sec usec)
 238   time-delta?
 239   (sec time-delta-sec)
 240   (usec time-delta-usec))
 241
 242 (define* (make-time-delta time)
 243   "Make a <time-delta> of SEC seconds and USEC microseconds.
 244
 245 This is used primarily so the agenda can recognize RUN-REQUEST objects
 246 which are meant to delay computation"
 247   (match (time-segment-right-format time)
 248     ((sec . usec)
 249      (make-time-delta-intern sec usec))))
 250
 251 (define tdelta make-time-delta)
 252
 253 (define (time-carry-correct time)
 254   "Corrects/handles time microsecond carry.
 255 Will produce (0 . 0) instead of a negative number, if needed."
 256   (cond ((>= (cdr time) 1000000)
 257          (cons
 258           (+ (car time) 1)
 259           (- (cdr time) 1000000)))
 260         ((< (cdr time) 0)
 261          (if (= (car time) 0)
 262              '(0 0)
 263              (cons
 264               (- (car time) 1)
 265               (+ (cdr time) 1000000))))
 266         (else time)))
 267
 268 (define (time-delta+ time time-delta)
 269   "Increment a TIME by the value of TIME-DELTA"
 270   (time-carry-correct
 271    (cons (+ (car time) (time-delta-sec time-delta))
 272          (+ (cdr time) (time-delta-usec time-delta)))))
 273
 274 (define (time-minus time1 time2)
 275   "Subtract TIME2 from TIME1"
 276   (time-carry-correct
 277    (cons (- (car time1) (car time2))
 278          (- (cdr time1) (cdr time2)))))
 279
 280 (define (time-plus time1 time2)
 281   "Add TIME1 and TIME2"
 282   (time-carry-correct
 283    (cons (+ (car time1) (car time2))
 284          (+ (cdr time1) (cdr time2)))))
 285
 286
 287 (define-record-type <schedule>
 288   (make-schedule-intern segments)
 289   schedule?
 290   (segments schedule-segments set-schedule-segments!))
 291
 292 (define* (make-schedule #:optional segments)
 293   "Make a schedule, optionally pre-composed of SEGMENTS"
 294   (make-schedule-intern (or segments '())))
 295
 296 (define (schedule-soonest-time schedule)
 297   "Return a cons of (sec . usec) for next time segement, or #f if none"
 298   (let ((segments (schedule-segments schedule)))
 299     (if (eq? segments '())
 300         #f
 301         (time-segment-time (car segments)))))
 302
 303 ;; TODO: This code is reasonably easy to read but it
 304 ;;   mutates AND is worst case of O(n) in both space and time :(
 305 ;;   but at least it'll be reasonably easy to refactor to
 306 ;;   a more functional setup?
 307 (define (schedule-add! schedule time proc)
 308   "Mutate SCHEDULE, adding PROC at an appropriate time segment for TIME"
 309   (let ((time (time-segment-right-format time)))
 310     (define (new-time-segment)
 311       (let ((new-segment
 312              (make-time-segment time)))
 313         (enq! (time-segment-queue new-segment) proc)
 314         new-segment))
 315     (define (loop segments)
 316       (define (segment-equals-time? segment)
 317         (time= time (time-segment-time segment)))
 318
 319       (define (segment-more-than-time? segment)
 320         (time< time (time-segment-time segment)))
 321
 322       ;; We could switch this out to be more mutate'y
 323       ;; and avoid the O(n) of space... is that over-optimizing?
 324       (match segments
 325         ;; If we're at the end of the list, time to make a new
 326         ;; segment...
 327         ('() (cons (new-time-segment) '()))
 328         ;; If the segment's time is exactly our time, good news
 329         ;; everyone!  Let's append our stuff to its queue
 330         (((? segment-equals-time? first) rest ...)
 331          (enq! (time-segment-queue first) proc)
 332          segments)
 333         ;; If the first segment is more than our time,
 334         ;; ours belongs before this one, so add it and
 335         ;; start consing our way back
 336         (((? segment-more-than-time? first) rest ...)
 337          (cons (new-time-segment) segments))
 338         ;; Otherwise, build up recursive result
 339         ((first rest ... )
 340          (cons first (loop rest)))))
 341     (set-schedule-segments!
 342      schedule
 343      (loop (schedule-segments schedule)))))
 344
 345 (define (schedule-empty? schedule)
 346   "Check if the SCHEDULE is currently empty"
 347   (eq? (schedule-segments schedule) '()))
 348
 349 (define (schedule-segments-split schedule time)
 350   "Does a multiple value return of time segments before/at and after TIME"
 351   (let ((time (time-segment-right-format time)))
 352     (define (segment-is-now? segment)
 353       (time= (time-segment-time segment) time))
 354     (define (segment-is-before-now? segment)
 355       (time< (time-segment-time segment) time))
 356
 357     (let loop ((segments-before '())
 358                (segments-left (schedule-segments schedule)))
 359       (match segments-left
 360         ;; end of the line, return
 361         ('()
 362          (values (reverse segments-before) '()))
 363
 364         ;; It's right now, so time to stop, but include this one in before
 365         ;; but otherwise return
 366         (((? segment-is-now? first) rest ...)
 367          (values (reverse (cons first segments-before)) rest))
 368
 369         ;; This is prior or at now, so add it and keep going
 370         (((? segment-is-before-now? first) rest ...)
 371          (loop (cons first segments-before) rest))
 372
 373         ;; Otherwise it's past now, just return what we have
 374         (segments-after
 375          (values segments-before segments-after))))))
 376
 377 (define (schedule-extract-until! schedule time)
 378   "Extract all segments until TIME from SCHEDULE, and pop old segments off"
 379   (receive (segments-before segments-after)
 380       (schedule-segments-split schedule time)
 381     (set-schedule-segments! schedule segments-after)
 382     segments-before))
 383
 384 (define (add-segments-contents-to-queue! segments queue)
 385   (for-each
 386    (lambda (segment)
 387      (let ((seg-queue (time-segment-queue segment)))
 388        (while (not (q-empty? seg-queue))
 389          (enq! queue (deq! seg-queue)))))
 390    segments))
 391
 392
 393 \f
 394 ;;; Request to run stuff
 395 ;;; ====================
 396
 397 (define-record-type <run-request>
 398   (make-run-request proc when)
 399   run-request?
 400   (proc run-request-proc)
 401   (when run-request-when))
 402
 403 (define* (run-it proc #:optional when)
 404   "Make a request to run PROC (possibly at WHEN)"
 405   (make-run-request proc when))
 406
 407 (define-syntax-rule (wrap body ...)
 408   "Wrap contents in a procedure"
 409   (lambda ()
 410     body ...))
 411
 412 (define-syntax-rule (wrap-apply body)
 413   "Wrap possibly multi-value function in a procedure, applies all arguments"
 414   (lambda args
 415     (apply body args)))
 416
 417
 418 ;; @@: Do we really want `body ...' here?
 419 ;;   what about just `body'?
 420 (define-syntax-rule (run body ...)
 421   "Run everything in BODY but wrap in a convenient procedure"
 422   (make-run-request (wrap body ...) #f))
 423
 424 (define-syntax-rule (run-at body ... when)
 425   "Run BODY at WHEN"
 426   (make-run-request (wrap body ...) when))
 427
 428 ;; @@: Is it okay to overload the term "delay" like this?
 429 ;;   Would `run-in' be better?
 430 (define-syntax-rule (run-delay body ... delay-time)
 431   "Run BODY at DELAY-TIME time from now"
 432   (make-run-request (wrap body ...) (tdelta delay-time)))
 433
 434
 435 \f
 436 ;;; Asynchronous escape to run things
 437 ;;; =================================
 438
 439 (define-syntax-rule (8sync-abort-to-prompt async-request)
 440   (abort-to-prompt (current-agenda-prompt)
 441                    async-request))
 442
 443 ;; Async port request and run-request meta-requests
 444 (define (make-async-request proc)
 445   "Wrap PROC in an async-request
 446
 447 The purpose of this is to make sure that users don't accidentally
 448 return the wrong thing via (8sync) and trip themselves up."
 449   (cons '*async-request* proc))
 450
 451 (define (setup-async-request resume-kont async-request)
 452   "Complete an async request for agenda-run-once's continuation handling"
 453   (match async-request
 454     (('*async-request* . async-setup-proc)
 455      (async-setup-proc resume-kont))
 456     ;; TODO: deliver more helpful errors depending on what the user
 457     ;;   returned
 458     (_ (throw 'invalid-async-request
 459               "Invalid request passed back via an (8sync) procedure."
 460               async-request))))
 461
 462 (define-syntax-rule (8sync body ...)
 463   "Run body asynchronously but ignore its result...
 464 forge ahead in our current function!"
 465   (8sync-abort-to-prompt
 466    (make-async-request
 467     (lambda (kont)
 468       (list (make-run-request
 469              ;; What's with returning #f to kont?
 470              ;; Otherwise we sometimes get errors like
 471              ;; "Zero values returned to single-valued continuation""
 472              (wrap (kont #f)) #f)
 473             (make-run-request (lambda () body ...) #f))))))
 474
 475 ;; TODO: Rewrite when we move to this being just `sleep'.
 476 (define (8sleep time)
 477   (8sync-abort-to-prompt
 478    (make-async-request
 479     (lambda (kont)
 480       (make-run-request (lambda () (kont #f)) (tdelta time))))))
 481
 482 ;; Voluntarily yield execution
 483 (define (yield)  ; @@: should this be define-inlinable?
 484   "Voluntarily yield execution to the scheduler."
 485   (8sync-abort-to-prompt
 486    (make-async-request
 487     (lambda (kont)
 488       (make-run-request (lambda () (kont #f)) #f)))))
 489
 490 \f
 491 ;;; Execution of agenda, and current agenda
 492 ;;; =======================================
 493
 494 (define %current-agenda (make-parameter #f))
 495
 496 (define (update-agenda-from-select! agenda)
 497   "Potentially (select) on ports specified in agenda, adding items to queue.
 498
 499 Also handles sleeping when all we have to do is wait on the schedule."
 500   (define (hash-keys hash)
 501     (hash-map->list (lambda (k v) k) hash))
 502   (define (get-wait-time)
 503     ;; TODO: we need to figure this out based on whether there's anything
 504     ;;   in the queue, and if not, how long till the next scheduled item
 505     (let ((soonest-time (schedule-soonest-time (agenda-schedule agenda))))
 506       (cond
 507        ((not (q-empty? (agenda-queue agenda)))
 508         (cons 0 0))
 509        (soonest-time    ; ie, the agenda is non-empty
 510         (let* ((current-time (agenda-time agenda)))
 511           (if (time<= soonest-time current-time)
 512               ;; Well there's something due so let's select
 513               ;; (this avoids a (possible?) race condition chance)
 514               (cons 0 0)
 515               (time-minus soonest-time current-time))))
 516        (else
 517         (cons #f #f)))))
 518   (define (do-select)
 519     ;; TODO: support usecond wait time too
 520     (match (get-wait-time)
 521       ((sec . usec)
 522        (catch 'system-error
 523          (lambda ()
 524            (select (hash-keys (agenda-read-port-map agenda))
 525                    (hash-keys (agenda-write-port-map agenda))
 526                    '()
 527                    sec usec))
 528          (lambda (key . rest-args)
 529            (match rest-args
 530              ((_ _ _ (EINTR))
 531               '(() () ()))
 532              (_ (error "Unhandled error in select!" key rest-args))))))))
 533   (define (get-procs-to-run)
 534     (define (ports->procs ports port-map)
 535       (lambda (initial-procs)
 536         (fold
 537          (lambda (port prev)
 538            (define proc (hashq-ref port-map port))
 539            ;; Now that we've selected on this port, it can be removed
 540            (hashq-remove! port-map port)
 541            (cons proc prev))
 542          initial-procs
 543          ports)))
 544     (match (do-select)
 545       ((read-ports write-ports except-ports) ; except-ports not used
 546        ((compose (ports->procs
 547                   read-ports
 548                   (agenda-read-port-map agenda))
 549                  (ports->procs
 550                   write-ports
 551                   (agenda-write-port-map agenda)))
 552         '()))))
 553   (define (update-agenda)
 554     (let ((procs-to-run (get-procs-to-run))
 555           (q (agenda-queue agenda)))
 556       (for-each
 557        (lambda (proc)
 558          (enq! q proc))
 559        procs-to-run))
 560     agenda)
 561   (define (ports-to-select?)
 562     (define (has-items? selector)
 563       ;; @@: O(n)
 564       ;;    ... we could use hash-for-each and a continuation to jump
 565       ;;    out with a #t at first indication of an item
 566       (not (= (hash-count (const #t)
 567                           (selector agenda))
 568               0)))
 569     (or (has-items? agenda-read-port-map)
 570         (has-items? agenda-write-port-map)))
 571
 572   (if (or (ports-to-select?)
 573           ;; select doubles as sleep...
 574           (not (schedule-empty? (agenda-schedule agenda))))
 575       (update-agenda)
 576       agenda))
 577
 578 (define-record-type <read-request>
 579   (make-read-request port proc)
 580   read-request?
 581   (port read-request-port)
 582   (proc read-request-proc))
 583
 584 (define-record-type <write-request>
 585   (make-write-request port proc)
 586   write-request?
 587   (port write-request-port)
 588   (proc write-request-proc))
 589
 590 (define (agenda-handle-read-request! agenda read-request)
 591   "Handle <read-request>, which is a request to add this port to the poll/select
 592 on suspendable ports."
 593   (hashq-set! (agenda-read-port-map agenda)
 594               (read-request-port read-request)
 595               (read-request-proc read-request)))
 596
 597 (define (agenda-handle-write-request! agenda write-request)
 598   (hashq-set! (agenda-write-port-map agenda)
 599               (write-request-port write-request)
 600               (write-request-proc write-request)))
 601
 602 (define (stop-on-nothing-to-do agenda)
 603   (and (q-empty? (agenda-queue agenda))
 604        (schedule-empty? (agenda-schedule agenda))
 605        (= 0 (hash-count (const #t) (agenda-read-port-map agenda)))
 606        (= 0 (hash-count (const #t) (agenda-write-port-map agenda)))))
 607
 608
 609 (define* (start-agenda agenda
 610                        #:key
 611                        ;; @@: Should we make stop-on-nothing-to-do
 612                        ;;   the default stop-condition?
 613                        (stop-condition stop-on-nothing-to-do)
 614                        (get-time gettimeofday)
 615                        (handle-ports update-agenda-from-select!)
 616                        ;; For live hacking madness, etc
 617                        (post-run-hook #f))
 618   ;; TODO: Document fields
 619   "Start up the AGENDA"
 620   (let loop ((agenda agenda))
 621     (let ((agenda
 622            ;; @@: Hm, maybe here would be a great place to handle
 623            ;;   select'ing on ports.
 624            ;;   We could compose over agenda-run-once and agenda-read-ports
 625            (agenda-run-once agenda)))
 626       ;; @@: This relies on mutation at present on the queue, in the rare
 627       ;;   event it's used.  If we ever switch to something more immutable,
 628       ;;   it should return a new modified agenda instead.
 629       (if post-run-hook
 630           (post-run-hook agenda))
 631       (if (and stop-condition (stop-condition agenda))
 632           'done
 633           (let* ((agenda
 634                   ;; We have to update the time after ports handled, too
 635                   ;; because it may have changed after a select
 636                   (set-field
 637                    (handle-ports
 638                     ;; Adjust the agenda's time just in time
 639                     ;; We do this here rather than in agenda-run-once to make
 640                     ;; agenda-run-once's behavior fairly predictable
 641                     (set-field agenda (agenda-time) (get-time)))
 642                    (agenda-time) (get-time))))
 643             ;; Update the agenda's current queue based on
 644             ;; currently applicable time segments
 645             (add-segments-contents-to-queue!
 646              (schedule-extract-until! (agenda-schedule agenda) (agenda-time agenda))
 647              (agenda-queue agenda))
 648             (loop agenda))))))
 649
 650
 651 (define (print-error-and-continue key . args)
 652   "Frequently used as pre-unwind-handler for agenda"
 653   (cond
 654    ((eq? key '8sync-caught-error)
 655     (match args
 656       ((orig-key orig-args stacks)
 657        (display "\n*** Caught async exception. ***\n")
 658        (format (current-error-port)
 659                "* Original key '~s and arguments: ~s *\n"
 660                orig-key orig-args)
 661        (display "* Caught stacks below (ending with original) *\n\n")
 662        (for-each
 663         (lambda (s)
 664           (display-backtrace s (current-error-port))
 665           (newline (current-error-port)))
 666         stacks))))
 667    (else
 668     (format (current-error-port)
 669             "\n*** Caught exception with key '~s and arguments: ~s ***\n"
 670             key args)
 671     (display-backtrace (make-stack #t 1 0)
 672                        (current-error-port))
 673     (newline (current-error-port)))))
 674
 675 (define-syntax-rule (maybe-catch-all (catch-handler pre-unwind-handler)
 676                                      body ...)
 677   (if (or catch-handler pre-unwind-handler)
 678       (catch
 679         #t
 680         (lambda ()
 681           body ...)
 682         (or catch-handler (lambda _ #f))
 683         (or pre-unwind-handler (lambda _ #f)))
 684       (begin body ...)))
 685
 686 (define (wait-for-readable port)
 687   (8sync-abort-to-prompt
 688    (make-async-request
 689     (lambda (kont)
 690       (make-read-request port (wrap (kont #f)))))))
 691
 692 (define (wait-for-writable port)
 693   (8sync-abort-to-prompt
 694    (make-async-request
 695     (lambda (kont)
 696       (make-write-request port (wrap (kont #f)))))))
 697
 698 (define (agenda-run-once agenda)
 699   "Run once through the agenda, and produce a new agenda
 700 based on the results"
 701   (define (call-proc proc)
 702     (call-with-prompt
 703      (agenda-prompt-tag agenda)
 704      (lambda ()
 705        (parameterize ((%current-agenda agenda)
 706                       ;; @@: Couldn't we just parameterize this at the start of
 707                       ;;   the agenda...?
 708                       (current-read-waiter wait-for-readable)
 709                       (current-write-waiter wait-for-writable))
 710          (maybe-catch-all
 711           ((agenda-catch-handler agenda)
 712            (agenda-pre-unwind-handler agenda))
 713           (proc))))
 714      (lambda (kont async-request)
 715        (setup-async-request kont async-request))))
 716
 717   (let ((queue (agenda-queue agenda))
 718         (next-queue (make-q)))
 719     (while (not (q-empty? queue))
 720       (let* ((proc (q-pop! queue))
 721              (proc-result (call-proc proc))
 722              (enqueue
 723               (lambda (run-request)
 724                 (define (schedule-at! time proc)
 725                   (schedule-add! (agenda-schedule agenda) time proc))
 726                 (let ((request-time (run-request-when run-request)))
 727                   (match request-time
 728                     ((? time-delta? time-delta)
 729                      (let ((time (time-delta+ (agenda-time agenda)
 730                                               time-delta)))
 731                        (schedule-at! time (run-request-proc run-request))))
 732                     ((? integer? sec)
 733                      (let ((time (cons sec 0)))
 734                        (schedule-at! time (run-request-proc run-request))))
 735                     (((? integer? sec) . (? integer? usec))
 736                      (schedule-at! request-time (run-request-proc run-request)))
 737                     (#f
 738                      (enq! next-queue (run-request-proc run-request))))))))
 739         (define (handle-individual result)
 740           ;; @@: Could maybe optimize by switching to an explicit cond...
 741           (match result
 742             ((? run-request? new-proc)
 743              (enqueue new-proc))
 744             ((? read-request? read-request)
 745              (agenda-handle-read-request! agenda read-request))
 746             ((? write-request? write-request)
 747              (agenda-handle-write-request! agenda write-request))
 748             ;; do nothing
 749             ;; Remember, we don't throw an error here because procedures can
 750             ;; return a run request, eg with run-it, at the end of their
 751             ;; evaluation to keep looping.
 752             ;; @@: Though is this really a useful feature?
 753             (_ #f)))
 754         ;; @@: We might support delay-wrapped procedures here
 755         (match proc-result
 756           ((results ...)
 757            (for-each handle-individual results))
 758           (one-result (handle-individual one-result)))))
 759     ;; TODO: Alternately, we could return the next-queue
 760     ;;   along with changes to be added to the schedule here?
 761     ;; Return new agenda, with next queue set
 762     (set-field agenda (agenda-queue) next-queue)))