ice-9/runq.scm

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