loopy.scm

   1 (define-module (loopy agenda)
   2   #:use-module (srfi srfi-9)
   3   #:use-module (srfi srfi-9 gnu)
   4   #:use-module (ice-9 q)
   5   #:use-module (ice-9 match)
   6   #:use-module (ice-9 receive)
   7   #:export (make-agenda
   8             agenda?
   9             agenda-queue agenda-prompt-tag
  10             agenda-port-pmapping agenda-schedule
  11
  12             make-async-prompt-tag
  13
  14             make-time-segment
  15             time-segment?
  16             time-segment-time time-segment-queue
  17
  18             time-< time-= time-<=
  19
  20             make-schedule
  21             schedule-add! schedule-empty?
  22             schedule-segments
  23
  24             schedule-segments-split schedule-extract-until!
  25
  26             make-port-mapping
  27             port-mapping-set! port-mapping-remove!
  28             port-mapping-empty? port-mapping-non-empty?
  29
  30             %current-agenda
  31             start-agenda agenda-run-once))
  32
  33 ;; @@: Using immutable agendas here, so wouldn't it make sense to
  34 ;;   replace this queue stuff with using pfds based immutable queues?
  35
  36 \f
  37 ;;; Agenda definition
  38 ;;; =================
  39
  40 ;;; The agenda consists of:
  41 ;;;  - a queue of immediate items to handle
  42 ;;;  - sheduled future events to be added to a future queue
  43 ;;;  - a tag by which running processes can escape for some asynchronous
  44 ;;;    operation (from which they can be returned later)
  45 ;;;  - a mapping of ports to various handler procedures
  46 ;;;
  47 ;;; The goal, eventually, is for this all to be immutable and functional.
  48 ;;; However, we aren't there yet.  Some tricky things:
  49 ;;;  - The schedule needs to be immutable, yet reasonably efficient.
  50 ;;;  - Need to use immutable queues (ijp's pfds library?)
  51 ;;;  - Modeling reading from ports as something repeatable,
  52 ;;;    and with reasonable separation from functional components?
  53
  54 (define-immutable-record-type <agenda>
  55   (make-agenda-intern queue prompt-tag port-mapping schedule)
  56   agenda?
  57   (queue agenda-queue)
  58   (prompt-tag agenda-prompt-tag)
  59   (port-mapping agenda-port-mapping)
  60   (schedule agenda-schedule))
  61
  62 (define (make-async-prompt-tag)
  63   (make-prompt-tag "prompt"))
  64
  65 (define* (make-agenda #:key
  66                       (queue (make-q))
  67                       (prompt (make-prompt-tag))
  68                       (port-mapping (make-port-mapping))
  69                       (schedule (make-schedule)))
  70   (make-agenda-intern queue prompt port-mapping schedule))
  71
  72
  73 \f
  74 ;;; Schedule
  75 ;;; ========
  76
  77 ;;; This is where we handle timed events for the future
  78
  79 ;; This section totally borrows from SICP
  80 ;; <3 <3 <3
  81
  82 ;; NOTE: time is a cons of (seconds . microseconds)
  83
  84 (define-record-type <time-segment>
  85   (make-time-segment-intern time queue)
  86   time-segment?
  87   (time time-segment-time)
  88   (queue time-segment-queue))
  89
  90 (define (time-segment-right-format time)
  91   (match time
  92     ;; time is already a cons of second and microsecnd
  93     (((? integer? s) . (? integer? u)) time)
  94     ;; time was just an integer (just the second)
  95     ((? integer? _) (cons time 0))
  96     (_ (throw 'invalid-time "Invalid time" time))))
  97
  98 (define* (make-time-segment time #:optional (queue (make-q)))
  99   (make-time-segment-intern time queue))
 100
 101 (define (time-< time1 time2)
 102   (cond ((< (car time1)
 103             (car time2))
 104          #t)
 105         ((and (= (car time1)
 106                  (car time2))
 107               (< (cdr time1)
 108                  (cdr time2)))
 109          #t)
 110         (else #f)))
 111
 112 (define (time-= time1 time2)
 113   (and (= (car time1) (car time2))
 114        (= (cdr time1) (cdr time2))))
 115
 116 (define (time-<= time1 time2)
 117   (or (time-< time1 time2)
 118       (time-= time1 time2)))
 119
 120 (define-record-type <schedule>
 121   (make-schedule-intern segments)
 122   schedule?
 123   (segments schedule-segments set-schedule-segments!))
 124
 125 (define* (make-schedule #:optional segments)
 126   (make-schedule-intern (or segments '())))
 127
 128 ;; TODO: This code is reasonably easy to read but it
 129 ;;   mutates AND is worst case of O(n) in both space and time :(
 130 ;;   but at least it'll be reasonably easy to refactor to
 131 ;;   a more functional setup?
 132 (define (schedule-add! time proc schedule)
 133   (let ((time (time-segment-right-format time)))
 134     (define (new-time-segment)
 135       (let ((new-segment
 136              (make-time-segment time)))
 137         (enq! (time-segment-queue new-segment) proc)
 138         new-segment))
 139     (define (loop segments)
 140       (define (segment-equals-time? segment)
 141         (time-= time (time-segment-time segment)))
 142
 143       (define (segment-more-than-time? segment)
 144         (time-< time (time-segment-time segment)))
 145
 146       ;; We could switch this out to be more mutate'y
 147       ;; and avoid the O(n) of space... is that over-optimizing?
 148       (match segments
 149         ;; If we're at the end of the list, time to make a new
 150         ;; segment...
 151         ('() (cons (new-time-segment) '()))
 152         ;; If the segment's time is exactly our time, good news
 153         ;; everyone!  Let's append our stuff to its queue
 154         (((? segment-equals-time? first) rest ...)
 155          (enq! (time-segment-queue first) proc)
 156          segments)
 157         ;; If the first segment is more than our time,
 158         ;; ours belongs before this one, so add it and
 159         ;; start consing our way back
 160         (((? segment-more-than-time? first) rest ...)
 161          (cons (new-time-segment) segments))
 162         ;; Otherwise, build up recursive result
 163         ((first rest ... )
 164          (cons first (loop rest)))))
 165     (set-schedule-segments!
 166      schedule
 167      (loop (schedule-segments schedule)))))
 168
 169 (define (schedule-empty? schedule)
 170   (eq? (schedule-segments schedule) '()))
 171
 172 (define (schedule-segments-split schedule time)
 173   "Does a multiple value return of time segments before/at and after TIME"
 174   (let ((time (time-segment-right-format time)))
 175     (define (segment-is-now? segment)
 176       (time-= (time-segment-time segment) time))
 177     (define (segment-is-before-now? segment)
 178       (time-< (time-segment-time segment) time))
 179
 180     (let loop ((segments-before '())
 181                (segments-left (schedule-segments schedule)))
 182       (match segments-left
 183         ;; end of the line, return
 184         ('()
 185          (values (reverse segments-before) '()))
 186
 187         ;; It's right now, so time to stop, but include this one in before
 188         ;; but otherwise return
 189         (((? segment-is-now? first) rest ...)
 190          (values (reverse (cons first segments-before)) rest))
 191
 192         ;; This is prior or at now, so add it and keep going
 193         (((? segment-is-before-now? first) rest ...)
 194          (loop (cons first segments-before) rest))
 195
 196         ;; Otherwise it's past now, just return what we have
 197         (segments-after
 198          (values segments-before segments-after))))))
 199
 200 (define (schedule-extract-until! schedule time)
 201   "Extract all segments until TIME from SCHEDULE, and pop old segments off"
 202   (receive (segments-before segments-after)
 203       (schedule-segments-split schedule time)
 204     (set-schedule-segments! schedule segments-after)
 205     segments-before))
 206
 207
 208 \f
 209 ;;; Port handling
 210 ;;; =============
 211
 212 (define (make-port-mapping)
 213   (make-hash-table))
 214
 215 (define* (port-mapping-set! port-mapping port #:optional read write except)
 216   "Sets port-mapping for reader / writer / exception handlers"
 217   (if (not (or read write except))
 218       (throw 'no-handlers-given "No handlers given for port" port))
 219   (hashq-set! port-mapping port
 220               `#(,read ,write ,except)))
 221
 222 (define (port-mapping-remove! port-mapping port)
 223   (hashq-remove! port-mapping port))
 224
 225 ;; TODO: This is O(n), I'm pretty sure :\
 226 ;; ... it might be worthwhile for us to have a
 227 ;;   port-mapping record that keeps a count of how many
 228 ;;   handlers (maybe via a promise?)
 229 (define (port-mapping-empty? port-mapping)
 230   "Is this port mapping empty?"
 231   (eq? (hash-count (const #t) port-mapping) 0))
 232
 233 (define (port-mapping-non-empty? port-mapping)
 234   "Whether this port-mapping contains any elements"
 235   (not (port-mapping-empty? port-mapping)))
 236
 237
 238 \f
 239 ;;; Execution of agenda, and current agenda
 240 ;;; =======================================
 241
 242 (define %current-agenda (make-parameter #f))
 243
 244 (define* (start-agenda agenda #:optional stop-condition)
 245   (let loop ((agenda agenda))
 246     (let ((new-agenda
 247            ;; @@: Hm, maybe here would be a great place to handle
 248            ;;   select'ing on ports.
 249            ;;   We could compose over agenda-run-once and agenda-read-ports
 250            (parameterize ((%current-agenda agenda))
 251              (agenda-run-once agenda))))
 252       (if (and stop-condition (stop-condition agenda))
 253           'done
 254           (loop new-agenda)))))
 255
 256 (define (agenda-run-once agenda)
 257   "Run once through the agenda, and produce a new agenda
 258 based on the results"
 259   (define (call-proc proc)
 260     (call-with-prompt
 261         (agenda-prompt-tag agenda)
 262       (lambda ()
 263         (proc))
 264       ;; TODO
 265       (lambda (k) k)))
 266
 267   (let ((queue (agenda-queue agenda))
 268         (next-queue (make-q)))
 269     (while (not (q-empty? queue))
 270       (let* ((proc (q-pop! queue))
 271              (proc-result (call-proc proc))
 272              (enqueue
 273               (lambda (new-proc)
 274                 (enq! next-queue new-proc))))
 275         ;; @@: We might support delay-wrapped procedures here
 276         (match proc-result
 277           ((? procedure? new-proc)
 278            (enqueue new-proc))
 279           (((? procedure? new-procs) ...)
 280            (for-each
 281             (lambda (new-proc)
 282               (enqueue new-proc))
 283             new-procs))
 284           ;; do nothing
 285           (_ #f))))
 286     ;; TODO: Selecting on ports would happen here?
 287     ;; Return new agenda, with next queue set
 288     (set-field agenda (agenda-queue) next-queue)))