⟦7804b7e42⟧

TextFile

; This is -*- SCHEME -*- code

;	This is part of the Butterfly fly SCHEME basic band load
;	so that people can tell that they actually have more than
;	processor.
;
;	Note that this doesn't actually start the FUTURE system
;	until someone does a (futures-on)

(declare (usual-integrations)
	 (compilable-primitive-functions
	  (= eq?)
	  set-fixed-objects-vector!
	  vector-set!
	  get-external-number
	  get-work
	  (catch call-with-current-continuation)))

(define put-work (make-primitive-procedure 'put-work))
(define global-interrupt (make-primitive-procedure 'global-interrupt))

(define TOUCH (make-primitive-procedure 'TOUCH))
(define SET-CAR-IF-EQ?! (make-primitive-procedure 'SET-CAR-IF-EQ?!))
(define SET-CDR-IF-EQ?! (make-primitive-procedure 'SET-CDR-IF-EQ?!))
(define VECTOR-SET-IF-EQ?! (make-primitive-procedure 'VECTOR-SET-IF-EQ?!))
(define SET-CXR-IF-EQ?! (make-primitive-procedure 'SET-CXR-IF-EQ?!))
(define FUTURE? (make-primitive-procedure 'FUTURE?))
(define FUTURE-REF (make-primitive-procedure 'FUTURE-REF))
(define FUTURE-SET! (make-primitive-procedure 'FUTURE-SET!))
(define FUTURE-SIZE (make-primitive-procedure 'FUTURE-SIZE))
(define LOCK-FUTURE! (make-primitive-procedure 'LOCK-FUTURE!))
(define UNLOCK-FUTURE! (make-primitive-procedure 'UNLOCK-FUTURE!))
\f


(define n-interpreters (make-primitive-procedure 'N-INTERPRETERS))
(define my-processor-number (make-primitive-procedure 'MY-PROCESSOR-NUMBER))
(define my-interpreter-number 
  (make-primitive-procedure 'MY-INTERPRETER-NUMBER))

(define statistics-package
  (make-package statistics-package
   ((get-statistics
     (make-primitive-procedure 'get-statistics #!true))
    (stat-names '((CONTENTION-COUNT      . 0)
		  (GC-MASTER-IDLE-TIME   . 1)
		  (GC-SLAVE-IDLE-TIME    . 2)
		  (GC-LOOP-TIME          . 3)
		  (GC-DAEMON-TIME        . 4)))
    (This-load))

   (define (load-statistics)
     (set! this-load (get-statistics)))
   
   (define (get-statistic name)
     (if (unassigned? this-load)
	 (load-statistics))
     (vector-ref this-load (cdr (assq name stat-names))))

   (define (clear-statistics)
     (set! this-load))))

(define load-statistics (access load-statistics statistics-package))
(define get-statistic (access get-statistic statistics-package))
(define clear-statistics (access clear-statistics statistics-package))
\f


; Slots in a future
; First the ones the microcode knows about explicitly
(define FUTURE-DETERMINED-SLOT 0)
(define FUTURE-LOCK-SLOT 1)
(define FUTURE-VALUE-SLOT 2)
(define FUTURE-QUEUE-SLOT 2)
; Then the ones referenced only by the runtime system
(define FUTURE-PROCESS-SLOT 3)
(define FUTURE-DELAYED?-SLOT 4)
(define FUTURE-ORIG-CODE-SLOT 5)
(define FUTURE-PROCESS-PRIVATE-SLOT 6)
\f


; Some useful macros for dealing with atomicity.  Notice that
;;DEFINE-MACRO happens when the text is turned into code (i.e.
;;at syntax time), while ADD-SYNTAX! happens only when the program
;;is actually executed.  So both are used when this file uses the
;;macro, but only ADD-SYNTAX! is used for user macros which
;;are not referenced here.

; ATOMIC takes a list of expression and guarantees that they
;;are done without interrupts.

(define-macro (atomic . expressions)
  `(WITHOUT-INTERRUPTS
    (LAMBDA () . ,expressions)))

(add-syntax! 'ATOMIC
  (macro expressions
  `(WITHOUT-INTERRUPTS
    (LAMBDA () . ,expressions))))

; DEFINE-ATOMIC is like the procedural version of DEFINE, except
;;that the body is wrapped in WITHOUT-INTERRUPTS.

(define-macro (define-atomic arg-template . body)
  `(DEFINE ,arg-template (ATOMIC . ,body)))

(add-syntax! 'DEFINE-ATOMIC
  (macro (arg-template . body)
    `(DEFINE ,arg-template (ATOMIC . ,body))))
\f


; LOCKING-FUTURE is the same as ATOMIC except that it also wraps
;;a LOCK-FUTURE! and UNLOCK-FUTURE! around the expression(s).
;;LOCKED? is a flag which can be used in BODY -- it will be #!true
;;if the future is still valid (you hang until you can lock it),
;;or #!false if it has been spliced out.

(define-macro (LOCKING-FUTURE FUTURE LOCKED? . BODY)
  `(WITH-FUTURE-LOCKED ,future
      (LAMBDA (,locked?) . ,body)))

(add-syntax! 'LOCKING-FUTURE
  (macro (FUTURE LOCKED? . BODY)
    `(WITH-FUTURE-LOCKED ,future
       (LAMBDA (,locked?) . ,body))))

(define-macro (WITH-STATE STATE . BODY)
  `(CATCH (LAMBDA (,state) . ,body)))

(DEFINE-ATOMIC (with-future-locked future thunk)
  (if (lock-future! future)
      (let ((result (thunk #!true)))
	(unlock-future! future)
	result)
      (thunk #!false)))

(define (determine! future value #!optional keep-slot?)
  (let ((result (touch value)))
    (LOCKING-FUTURE future was-still-a-future?
      (if was-still-a-future?
	  (let ((known? (future-ref future FUTURE-DETERMINED-SLOT))
		(waiters (future-ref future FUTURE-QUEUE-SLOT)))
	    (if (eq? known? #!true)
		(error "Future cannot be determined twice."
		       future))
	    (future-set! future FUTURE-VALUE-SLOT result)
	    (if (unassigned? keep-slot?)
		(if (eq? known? #!false)
		    (future-set! future FUTURE-DETERMINED-SLOT
				 #!true))
		(future-set! future FUTURE-DETERMINED-SLOT
			     (if keep-slot? 'KEEP-SLOT #!true)))
	    ((access awaken! scheduler) waiters))
	  (error "Future cannot be determined twice." future)))
    result))
\f


(define scheduler
  (make-package scheduler
    ((Current-Future-Vector)  ; Process currently running
     (Start-Process)	      ; Default scheduler for FUTURE creation
     (Idle-Future)	      ; Future to wait until idle on
     (sti (make-primitive-procedure 'setup-timer-interrupt #!true))
     			      ; Procedure to initiate timer interrupt
     (drain-work-queue! (make-primitive-procedure 'drain-work-queue!))
     (preempting? #!false)    ; No timer currently set
     (Delta '()))	      ; Scheduling frequency, centi-seconds

    (DEFINE-ATOMIC (start-preempting interval)
      (if (not preempting?)
	  (sequence
	    (set! timer-interrupt
	      (lambda ()
		(let ((My-Task (Current-Future)))
		  (WITH-STATE me
		    (LOCKING-FUTURE My-Task I-am-running?
                      (if I-am-running?
			  (sequence
			    (future-set! My-Task FUTURE-PROCESS-SLOT me)
			    (put-work My-Task))
			  (sequence
			    (stop-preempting)
			    (bkpt "TIMER: Existential crisis!"))))
		    (next)))))
	    (set! preempting? #!true)
	    (set! delta interval)
	    (sti 0 interval))
	  (print "Already preempting when START-PREEMPTING called")))

    (DEFINE-ATOMIC (stop-preempting)
      (sti '() '())
      (set! preempting? #!false))

    (define make-future
      (let ((future-type (microcode-type 'FUTURE)))
 	(named-lambda (make-future orig-code name)
	  (primitive-set-type
	   future-type
	   (VECTOR #!FALSE				 ; No value yet
		   #!FALSE		                 ; Not locked
		   ((ACCESS MAKE-EMPTY-QUEUE SCHEDULER)) ; No waiters
		   orig-code				 ; How to resume
		   #!FALSE				 ; Not delayed
		   orig-code			         ; For debugging
		   (if (unbound? open-console-channel)
		       'no-console-channel
		       (vector (open-console-channel name))))))))
\f


    (define (spawn-process thunk doc #!optional start)
      (with-interrupt-mask INTERRUPT-MASK-GC-OK
	(lambda (old-interrupts)
	  (let ((object))
	    (set! object
              (make-future
	       (lambda (ignore) (set-interrupt-enables! old-interrupts)
		                (let ((result (thunk)))
				  (ATOMIC
				   (determine! object result)
				   (next))))
	       doc))
	    ((if (unassigned? start)
		 (access start-process scheduler)
		 start) object)
	    object))))
\f


    (define (Current-Future)
      (if (unassigned? Current-Future-Vector)
	  '()
	  (vector-ref Current-Future-Vector (My-Interpreter-Number))))

    (define (Set-Current-Future! Future)
      (vector-set! Current-Future-Vector (My-Interpreter-Number)
		   Future))

    (define (initialize-scheduler! #!optional interval default-scheduler)
      (pause-everything)		; Stop all processors & drain queue
      (set! Start-Process
	    (if (unassigned? default-scheduler)
		dfuture-scheduler
	        default-scheduler))
      (set! Current-Future-Vector (vector-cons (N-Interpreters) 'NO-FUTURE-YET))
      (set! Idle-Future (make-future 'NO-PROCESS "Idle-Loop"))
      (if (not (unassigned? interval)) (set! delta interval))
      (let ((fobj (get-fixed-objects-vector)))
	(vector-set! fobj (fixed-objects-vector-slot 'SCHEDULER)
		     await-future)
	(set-fixed-objects-vector! fobj))
      (Set-Current-Future! (make-future 'INITIAL-PROCESS "The Initial Process"))
      (global-interrupt
       1
       (lambda (IntCode IntEnb)
	 (set-interrupt-enables! IntEnb)
	 (next))
       (lambda () #!true))
      (or Delta 'NOT-PREEMPTIVE-SCHEDULING))
\f


; Scheduling support

    (define (next)
      (Set-Current-Future! 'WAITING-FOR-WORK)
      (run (get-work
	    (lambda ()
	      (stop-preempting)
	      (determine! Idle-Future 'DONE)
	      (set! Idle-Future (make-future 'NO-PROCESS "Idle Loop"))))))

    ;; RUN starts a process running

    (define (run future)
      (if (and delta preempting?) (sti 0 delta)) ; Full time interval
      (Set-Current-Future! future)
      ((future-set! future FUTURE-PROCESS-SLOT 'RUNNING) 'YOUR-TURN))

    ;; AWAKEN! is called with a queue (of processes waiting
    ;; for a future) and promotes them all to runnable status.

    (define (awaken! queue)
      (if (empty-queue? queue)
	  'DONE
	  (sequence
	    (let loop ()
	      (if (empty-queue? queue)
		  'DONE
		  (let ((next-item (dequeue! queue)))
		    (if (not (future? next-item))
			(bkpt "Non-future awakened!")
			(put-work next-item))
		    (loop))))
	    (if (and delta (not preempting?))
		(start-preempting delta)))))
\f


    ;; AWAIT-FUTURE suspends the current process and adds it to the
    ;; queue waiting for the specified future to get a value.

    (DEFINE-ATOMIC (await-future future)
      (WITH-STATE me
	((LOCKING-FUTURE future waiting-for-a-future?
	   (if (not (conjunction
		     waiting-for-a-future?
		     (null? (future-ref future
					FUTURE-DETERMINED-SLOT))))
	       (lambda () 'NO-NEED-TO-SUSPEND)
	       (let ((My-Task (Current-Future))
		     (next-process
		      (if (future-ref future FUTURE-DELAYED?-SLOT)
			  (lambda ()
			    (future-set! future FUTURE-DELAYED?-SLOT
					 #!false)
			    (run future))
			  next)))
		 (LOCKING-FUTURE My-Task I-am-running?
		   (if I-am-running?
		       (sequence
			 (future-set! My-Task FUTURE-PROCESS-SLOT me)
			 (enqueue! (future-ref future FUTURE-QUEUE-SLOT)
				   My-Task)
			 next-process)
		       (sequence	; I wasn't a legit. process
			 (print "AWAIT-FUTURE: Existential crisis!")
			 next)))))))))

\f



    ;; AWAIT-FUTURE-AFTER-ACTION suspends the current process after 
    ;; executing a thunk and adds it to the queue waiting for the
    ;; specified future to get a value.  Its purpose is to ensure
    ;; that the process is actually on the wait queue of the future
    ;; when the action takes place.  This prevents a race condition
    ;; which might cause problems if the action is intended to determine
    ;; the future (e.g., externally) and wake up the process.  In other words,
    ;; if the process has not been added to the wait queue when another
    ;; process or an external interrupt determines the future, the event 
    ;; would not wake up the process as intended.
    ;;
    ;; Since this is rather specialized code (it was added to support
    ;; the new console i/o system), many safeguards of the normal 
    ;; AWAIT-FUTURE are removed for speed.  Normally the future will
    ;; have been explicitly constructed by the user and is not being
    ;; determined by any other process.

    (define (await-future-after-action future action)
      (let ((My-Task (Current-Future)))
	(catch
	 (lambda (me)
	   (future-set! My-Task FUTURE-PROCESS-SLOT me)
	   (enqueue! (future-ref future FUTURE-QUEUE-SLOT)
		     My-Task)
	   (action)
	   (next)))))
\f


; Special scheduler operations

    ;; RESCHEDULE allows me to give up my processor slice and
    ;; wait until the scheduler gets back to me.

    (define-atomic (reschedule)
      (WITH-STATE me
	(future-set! (Current-Future) FUTURE-PROCESS-SLOT me)
	(put-work me)
	(next)))

    ;; WAIT-UNTIL-IDLE causes a process to just continue
    ;; going to sleep until there are no other active processes.

    (define (wait-until-idle) (touch idle-future))

    ;; DFUTURE-SCHEDULER is a future creation scheduler which
    ;; defers the child process and continues on with the parent.
    ;; Note that all creation schedulers are called as part of
    ;; the parent process, so this is the easy case.

    (DEFINE-ATOMIC (dfuture-scheduler future)
      (put-work future)
      (if (and delta (not preempting?)) (start-preempting delta))
      'CHILD-QUEUED-FOR-EXECUTION)

    ;; FUTURE-SCHEDULER is a future creation scheduler which
    ;; defers the parent process and continues on with the child.
    ;; This is a little harder than DFUTURE, since it is called
    ;; running as the parent.

    (DEFINE-ATOMIC (future-scheduler future)
      (WITH-STATE parent-process
	(let ((My-Future (Current-Future)))
	  (LOCKING-FUTURE My-Future Still-Runnable?
	    (if Still-Runnable?
		(sequence
		  (future-set! My-Future FUTURE-PROCESS-SLOT parent-process)
		  (put-work My-Future)
		  (if (and delta (not preempting?)) (start-preempting delta)))))
	  (run future))))

    ;; DELAY-SCHEDULER is a future creation scheduler which defers
    ;; execution of the newly created future until it is first
    ;; touched.

    (DEFINE-ATOMIC (delay-scheduler future)
      (future-set! future FUTURE-DELAYED?-SLOT #!true)
      'OK-I-DELAYED-IT)
\f


; Queue Abstraction
;
; -------------------------------
; | Tail Pointer | Head Pointer |
; -------------------------------
;        |             |
;        |             |
;        V             V    
;      -----  -----  -----
;      | |=|=>| |=|=>| |/| add new items by clobbering '()
;      -----  -----  -----
; remove from start of list
;
; The queue is empty when Tail=Head=#!NULL
; (thus it has one item when Tail=Head but they are not #!NULL)
;
; These operations assume that the caller has arranged for any
; desired ity.

    (define (make-empty-queue) (cons '() ()))
    (define queue-head-ptr car)
    (define queue-tail-ptr cdr)
    (define set-queue-head-ptr! set-car!)
    (define set-queue-tail-ptr! set-cdr!)

    (define (empty-queue? queue) (null? (queue-head-ptr queue)))

    (define (enqueue! queue object)
      (if (null? (queue-head-ptr queue))
	  (sequence
	   (set-queue-head-ptr! queue (list object))
	   (set-queue-tail-ptr! queue (queue-head-ptr queue)))
          (sequence
            (set-cdr! (queue-head-ptr queue) (list object))
            (set-queue-head-ptr! queue (cdr (queue-head-ptr queue))))))
    
    (define (dequeue! queue)
      (let ((current-tail (queue-tail-ptr queue)))
	(if (null? current-tail)
	    (error "Queue empty" queue)
	    (let ((result (car current-tail)))
	      (if (null? (cdr current-tail))
		  (sequence (set-queue-head-ptr! queue '())
			    (set-queue-tail-ptr! queue '()))
		(set-queue-tail-ptr! queue (cdr current-tail)))
	      result))))
\f


;; PAUSE-EVERYTHING is used to make every processor save its state
;; and all but the caller then go quiescent.  The value returned
;; by Pause-Everything is a procedure which will put the work queue
;; back to its initial state (modulo order of futures on the queue)
;
;  The state of the system consists of the current-future-vector
;  and the current queue.

  (DEFINE-ATOMIC (pause-everything)
    (stop-preempting)
    (if (unassigned? Current-Future-Vector)
	(lambda () 'DONE)
	(let ((drain-synch (make-synchronizer))
	      (proceed-synch (make-synchronizer)))
	  (global-interrupt
	   1
	   (lambda (int-code int-mask)
	     (if (eq? (stuff-yourself) 'GOT-STUFFED)
		 (sequence
		   (set-interrupt-enables! int-mask)
		   (await-synchrony drain-synch)
		   (await-synchrony proceed-synch)
		   (next))))
	   (lambda () #!TRUE))
	  (if (eq? (stuff-yourself) 'GOT-STUFFED)
	      (sequence
		(await-synchrony drain-synch)
		(let ((the-queue (drain-work-queue!))
		      (the-array Current-Future-Vector))
		  (set! Current-Future-Vector (vector-cons (N-Interpreters) 'PAUSED))
		  (Set-Current-Future! (make-future 'PAUSE-MASTER "Pause-Everyone"))
		  (await-synchrony proceed-synch)
		  (lambda ()
		    (if the-array
			(sequence
			  (release-state! the-array)
			  (release-state! the-queue)))
		    (set! the-array '())
		    (next))))
	      #!FALSE))))

;; STUFF-YOURSELF is really save processor state
;; put the current continuation into the current future

  (define (stuff-yourself)
    (WITH-STATE me
       (let ((basket (Current-Future)))
	 (LOCKING-FUTURE basket I-am-Running?
			 (if I-am-Running?
			     (future-set! basket FUTURE-PROCESS-SLOT me)))
	 'GOT-STUFFED)))

;; RELEASE-STATE! puts all the work in an array back on the queue

  (define (release-state! array)
    (let ((last (vector-size array)))
      (let release-next ((i 0))
	(if (< i last)
	    (let ((work-unit (vector-ref array i)))
	      (LOCKING-FUTURE work-unit work-to-do?
			      (conjunction
			       work-to-do?
			       (not (symbol? (future-ref work-unit FUTURE-PROCESS-SLOT)))
			       (put-work work-unit)))
	      (release-next (1+ i)))
	    'RESTARTED))))

  ))					; End of Make-Package for Scheduler
\f



; Export definitions to the world outside the scheduler

(define initialize-scheduler! (access initialize-scheduler! scheduler))
(define future-scheduler (access future-scheduler scheduler))
(define dfuture-scheduler (access dfuture-scheduler scheduler))
(define delay-scheduler (access delay-scheduler scheduler))
(define wait-until-idle (access wait-until-idle scheduler))
(define pause-everything (access pause-everything scheduler))
(define Current-Future (access Current-Future scheduler))

(define (futures-on #!optional slice)
  (if (unassigned? slice) (set! slice '()))
  (initialize-scheduler! slice dfuture-scheduler))
\f


(add-syntax! 'future
   (macro (expression #!optional doc user-scheduler)
     `((ACCESS SPAWN-PROCESS SCHEDULER)
       (LAMBDA () ,expression)			; Work to do
       ,(if (unassigned? doc)			; Documentation
	    (if (unbound? prin1-to-string)
		"NO DOCUMENTATION AVAILABLE"
		(prin1-to-string `,expression))
	    doc)
       ,@(if (unassigned? user-scheduler)	; Start-up procedure
	     '()
	     `(,user-scheduler)))))
DataMuseum.dk

DKUUG/EUUG Conference tapes

⟦7804b7e42⟧ TextFile

Derivation

TextFile
