ice-9/runq.scm

   1 ;;;; runq.scm --- the runq data structure
   2 ;;;;
   3 ;;;;    Copyright (C) 1996, 2001, 2006 Free Software Foundation, Inc.
   4 ;;;;
   5 ;;;; This library is free software; you can redistribute it and/or
   6 ;;;; modify it under the terms of the GNU Lesser General Public
   7 ;;;; License as published by the Free Software Foundation; either
   8 ;;;; version 2.1 of the License, or (at your option) any later version.
   9 ;;;;
  10 ;;;; This library is distributed in the hope that it will be useful,
  11 ;;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 ;;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13 ;;;; Lesser General Public License for more details.
  14 ;;;;
  15 ;;;; You should have received a copy of the GNU Lesser General Public
  16 ;;;; License along with this library; if not, write to the Free Software
  17 ;;;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
  18 ;;;;
  19
  20 ;;; Commentary:
  21
  22 ;;; One way to schedule parallel computations in a serial environment is
  23 ;;; to explicitly divide each task up into small, finite execution time,
  24 ;;; strips.  Then you interleave the execution of strips from various
  25 ;;; tasks to achieve a kind of parallelism.  Runqs are a handy data
  26 ;;; structure for this style of programming.
  27 ;;;
  28 ;;; We use thunks (nullary procedures) and lists of thunks to represent
  29 ;;; strips.  By convention, the return value of a strip-thunk must either
  30 ;;; be another strip or the value #f.
  31 ;;;
  32 ;;; A runq is a procedure that manages a queue of strips.  Called with no
  33 ;;; arguments, it processes one strip from the queue.  Called with
  34 ;;; arguments, the arguments form a control message for the queue.  The
  35 ;;; first argument is a symbol which is the message selector.
  36 ;;;
  37 ;;; A strip is processed this way: If the strip is a thunk, the thunk is
  38 ;;; called -- if it returns a strip, that strip is added back to the
  39 ;;; queue.  To process a strip which is a list of thunks, the CAR of that
  40 ;;; list is called.  After a call to that CAR, there are 0, 1, or 2 strips
  41 ;;; -- perhaps one returned by the thunk, and perhaps the CDR of the
  42 ;;; original strip if that CDR is not nil.  The runq puts whichever of
  43 ;;; these strips exist back on the queue.  (The exact order in which
  44 ;;; strips are put back on the queue determines the scheduling behavior of
  45 ;;; a particular queue -- it's a parameter.)
  46
  47 ;;; Code:
  48
  49 (define-module (ice-9 runq)
  50   :use-module (ice-9 q)
  51   :export (runq-control make-void-runq make-fair-runq
  52            make-exclusive-runq make-subordinate-runq-to strip-sequence
  53            fair-strip-subtask))
  54
  55 ;;;;
  56 ;;;     (runq-control q msg . args)
  57 ;;;
  58 ;;;             processes in the default way the control messages that
  59 ;;;             can be sent to a runq.  Q should be an ordinary
  60 ;;;             Q (see utils/q.scm).
  61 ;;;
  62 ;;;             The standard runq messages are:
  63 ;;;
  64 ;;;             'add! strip0 strip1...          ;; to enqueue one or more strips
  65 ;;;             'enqueue! strip0 strip1...      ;; to enqueue one or more strips
  66 ;;;             'push! strip0 ...               ;; add strips to the front of the queue
  67 ;;;             'empty?                         ;; true if it is
  68 ;;;             'length                         ;; how many strips in the queue?
  69 ;;;             'kill!                          ;; empty the queue
  70 ;;;             else                            ;; throw 'not-understood
  71 ;;;
  72 (define (runq-control q msg . args)
  73   (case msg
  74     ((add!)                     (for-each (lambda (t) (enq! q t)) args) '*unspecified*)
  75     ((enqueue!)                 (for-each (lambda (t) (enq! q t)) args) '*unspecified*)
  76     ((push!)                    (for-each (lambda (t) (q-push! q t)) args) '*unspecified*)
  77     ((empty?)                   (q-empty? q))
  78     ((length)                   (q-length q))
  79     ((kill!)                    (set! q (make-q)))
  80     (else                       (throw 'not-understood msg args))))
  81
  82 (define (run-strip thunk) (catch #t thunk (lambda ign (warn 'runq-strip thunk ign) #f)))
  83
  84 ;;;;
  85 ;;; make-void-runq
  86 ;;;
  87 ;;; Make a runq that discards all messages except "length", for which
  88 ;;; it returns 0.
  89 ;;;
  90 (define (make-void-runq)
  91   (lambda opts
  92     (and opts
  93         (apply-to-args opts
  94           (lambda (msg . args)
  95             (case msg
  96               ((length)         0)
  97               (else             #f)))))))
  98
  99 ;;;;
 100 ;;;     (make-fair-runq)
 101 ;;;
 102 ;;;             Returns a runq procedure.
 103 ;;;             Called with no arguments, the procedure processes one strip from the queue.
 104 ;;;             Called with arguments, it uses runq-control.
 105 ;;;
 106 ;;;             In a fair runq, if a strip returns a new strip X, X is added
 107 ;;;             to the end of the queue, meaning it will be the last to execute
 108 ;;;             of all the remaining procedures.
 109 ;;;
 110 (define (make-fair-runq)
 111   (letrec ((q (make-q))
 112            (self
 113             (lambda ctl
 114               (if ctl
 115                   (apply runq-control q ctl)
 116                   (and (not (q-empty? q))
 117                        (let ((next-strip (deq! q)))
 118                          (cond
 119                           ((procedure? next-strip)      (let ((k (run-strip next-strip)))
 120                                                           (and k (enq! q k))))
 121                           ((pair? next-strip) (let ((k (run-strip (car next-strip))))
 122                                                 (and k (enq! q k)))
 123                                               (if (not (null? (cdr next-strip)))
 124                                                   (enq! q (cdr next-strip)))))
 125                          self))))))
 126     self))
 127
 128
 129 ;;;;
 130 ;;;     (make-exclusive-runq)
 131 ;;;
 132 ;;;             Returns a runq procedure.
 133 ;;;             Called with no arguments, the procedure processes one strip from the queue.
 134 ;;;             Called with arguments, it uses runq-control.
 135 ;;;
 136 ;;;             In an exclusive runq, if a strip W returns a new strip X, X is added
 137 ;;;             to the front of the queue, meaning it will be the next to execute
 138 ;;;             of all the remaining procedures.
 139 ;;;
 140 ;;;             An exception to this occurs if W was the CAR of a list of strips.
 141 ;;;             In that case, after the return value of W is pushed onto the front
 142 ;;;             of the queue, the CDR of the list of strips is pushed in front
 143 ;;;             of that (if the CDR is not nil).   This way, the rest of the thunks
 144 ;;;             in the list that contained W have priority over the return value of W.
 145 ;;;
 146 (define (make-exclusive-runq)
 147   (letrec ((q (make-q))
 148            (self
 149             (lambda ctl
 150               (if ctl
 151                   (apply runq-control q ctl)
 152                   (and (not (q-empty? q))
 153                        (let ((next-strip (deq! q)))
 154                          (cond
 155                           ((procedure? next-strip)      (let ((k (run-strip next-strip)))
 156                                                           (and k (q-push! q k))))
 157                           ((pair? next-strip) (let ((k (run-strip (car next-strip))))
 158                                                 (and k (q-push! q k)))
 159                                               (if (not (null? (cdr next-strip)))
 160                                                   (q-push! q (cdr next-strip)))))
 161                          self))))))
 162     self))
 163
 164
 165 ;;;;
 166 ;;;     (make-subordinate-runq-to superior basic-inferior)
 167 ;;;
 168 ;;;             Returns a runq proxy for the runq basic-inferior.
 169 ;;;
 170 ;;;             The proxy watches for operations on the basic-inferior that cause
 171 ;;;             a transition from a queue length of 0 to a non-zero length and
 172 ;;;             vice versa.   While the basic-inferior queue is not empty,
 173 ;;;             the proxy installs a task on the superior runq.  Each strip
 174 ;;;             of that task processes N strips from the basic-inferior where
 175 ;;;             N is the length of the basic-inferior queue when the proxy
 176 ;;;             strip is entered.  [Countless scheduling variations are possible.]
 177 ;;;
 178 (define (make-subordinate-runq-to superior-runq basic-runq)
 179   (let ((runq-task (cons #f #f)))
 180     (set-car! runq-task
 181               (lambda ()
 182                 (if (basic-runq 'empty?)
 183                     (set-cdr! runq-task #f)
 184                     (do ((n (basic-runq 'length) (1- n)))
 185                         ((<= n 0)                #f)
 186                       (basic-runq)))))
 187     (letrec ((self
 188               (lambda ctl
 189                 (if (not ctl)
 190                     (let ((answer (basic-runq)))
 191                       (self 'empty?)
 192                       answer)
 193                     (begin
 194                       (case (car ctl)
 195                         ((suspend)              (set-cdr! runq-task #f))
 196                         (else                   (let ((answer (apply basic-runq ctl)))
 197                                                   (if (and (not (cdr runq-task)) (not (basic-runq 'empty?)))
 198                                                       (begin
 199                                                         (set-cdr! runq-task runq-task)
 200                                                         (superior-runq 'add! runq-task)))
 201                                                   answer))))))))
 202       self)))
 203
 204 ;;;;
 205 ;;;     (define fork-strips (lambda args args))
 206 ;;;             Return a strip that starts several strips in
 207 ;;;             parallel.   If this strip is enqueued on a fair
 208 ;;;             runq, strips of the parallel subtasks will run
 209 ;;;             round-robin style.
 210 ;;;
 211 (define fork-strips (lambda args args))
 212
 213
 214 ;;;;
 215 ;;;     (strip-sequence . strips)
 216 ;;;
 217 ;;;             Returns a new strip which is the concatenation of the argument strips.
 218 ;;;
 219 (define ((strip-sequence . strips))
 220   (let loop ((st (let ((a strips)) (set! strips #f) a)))
 221     (and (not (null? st))
 222          (let ((then ((car st))))
 223            (if then
 224                (lambda () (loop (cons then (cdr st))))
 225                (lambda () (loop (cdr st))))))))
 226
 227
 228 ;;;;
 229 ;;;     (fair-strip-subtask . initial-strips)
 230 ;;;
 231 ;;;             Returns a new strip which is the synchronos, fair,
 232 ;;;             parallel execution of the argument strips.
 233 ;;;
 234 ;;;
 235 ;;;
 236 (define (fair-strip-subtask . initial-strips)
 237   (let ((st (make-fair-runq)))
 238     (apply st 'add! initial-strips)
 239     st))
 240
 241 ;;; runq.scm ends here