Esteban U C Castro

;;;
;;; Coroutine library, inspired by Stackless Python's API.
;;; (www.stackless.com)
;;;
;;; This has been tested in Gambit-C 4.0 beta 20.  AFAIK it should run on any
;;; R5RS compliant Scheme, or maybe earlier, since it doesn't use macros, and
;;; it has no external dependencies.  Please bear with the verbosity.
;;;
;;; (c) 2007 Esteban U. Caama�o Castro
;;;
;;; This code is in the public domain.
;;;

;;; Exported symbols (these are set! below, in the implementation scope).

@@; (add-next-task! thunk) -> <undefined>
;
@@; Schedule a new task to be run right after the current one yields or is
@@; blocked.
;
; If you'd rather switch to that one immediately, just (yield) right after
; calling this.
;
; If you'd rather schedule it after other waiting tasks, just (yield)
; first thing in the thunk.
(define add-next-task! #f)

@@; (last-task?) -> bool
;
@@; Whether this one is the only nonblocked task.
(define last-task? #f)

@@; (yield) -> <undefined>
;
@@; Pause the current task and resume the next scheduled one.
;
; This task will be resumed after all the other scheduled tasks have had a
; chance to run.
(define yield #f)

@@; (make-channel) -> channel object
;
; channel constructor.
;
@@; A channel is the facility to communicate and synchronize between tasks.
(define make-channel #f)

@@; (send-to-channel! channel value) -> <undefined>
;
; If there are any tasks waiting to receive data from this channel, this call
; returns immediately, and the oldest of the waiting tasks will be resumed
; next after the current task is paused or blocked.
;
; If there are no such waiting tasks, the current task will be blocked until
; other task calls receive-from-channel! on this channel.
;
@@; In either case, the value passed to this procedure is the value that the
@@; receiving task will get.
(define send-to-channel! #f)

@@; (receive-from-channel! channel) -> some value
;
; If there are any tasks waiting to send data to this channel, this call
; returns immediately, and the oldest of such tasks will be scheduled to be
; resumed right after the current task is paused or blocked.
;
; If there are no such waiting tasks, the current task will be blocked until
; other task calls send-to-channel! on this channel.
;
@@; In either case this procedure returns the value passed to the corresponding
@@; send-to-channel! call.
(define receive-from-channel! #f)

@@; (channel-sending? channel) -> bool
;
@@; Whether there are any tasks waiting to send data through this channel.
(define channel-sending? #f)

@@; (channel-receiving? channel) -> bool
;
@@; Whether there are any tasks waiting to receive data from this channel.
(define channel-receiving? #f)

;;; Utility.

; (run-tasks) -> <undefined>
;
; This just yields until there are no other non-blocked tasks.
;
; This is usually done in the main task: you set up initial tasks, let them
; run, and when they all are done this call returns and you clean up.
(define (run-tasks)
  (if (last-task?)
    'done
    (begin
      (yield)
      (run-tasks))))

;;; Implementation.
;;;
@@;;; Ignore this on first reading; just search forward for
@@;;;     "Implementation of exported symbols."

(let ()

  (define (make-task thunk)
    (list thunk #f #f))

  (define (task-continuation t)
    (car t))

  (define (set-task-continuation! t k)
    (set-car! t k))

  (define (prev-task t)
    (cadr t))

  (define (set-prev-task! t u)
    (set-car! (cdr t) u))

  (define (next-task t)
    (caddr t))

  (define (set-next-task! t u)
    (set-car! (cddr t) u))

  (define (remove-task! t)
    (set-prev-task! (next-task t) (prev-task t))
    (set-next-task! (prev-task t) (next-task t)))

  (define (resume-next!)
    (set! current (next-task current))
    ((task-continuation current))
    ; The task has exited normally.
    ; If it was the last task, we just exit,
    ; otherwise, we remove this one and continue with the next.
    (cond
      ((last-task?)
        'bye)
      (else
        (remove-task! current)
        (resume-next!))))

  (define (channel-state ch)
    (car ch))

  (define (channel-queue ch)
    (cdr ch))

  (define (set-channel-state! ch state)
    (set-car! ch state))

  (define (add-channel! ch op x)
    (set-channel-state! ch op)
    (queue-push! (channel-queue ch) x))

  (define (push-channel-sender! ch sender)
    ; XXX: assert I'm idle or sending.
    (add-channel! ch 'sending sender))

  (define (push-channel-receiver! ch receiver)
    ; XXX: assert I'm idle or receiving.
    (add-channel! ch 'receiving receiver))

  (define (pop-from-channel! ch)
    (let ((q (channel-queue ch)))
      (let ((ret (queue-pop! q)))
        (if (queue-empty? q)
          (set-channel-state! ch 'idle)
          #f)
        ret)))

  (define (pop-channel-sender! ch)
    ; XXX: assert I'm sending.
    (pop-from-channel! ch))

  (define (pop-channel-receiver! ch)
    ; XXX: assert I'm receiving.
    (pop-from-channel! ch))

  (define (make-sender k val)
    (cons k val))

  (define (sender-continuation sender)
    (car sender))

  (define (sender-value sender)
    (cdr sender))

  (define (block!)
    (if (last-task?)
      (error "I can't block the last remaining task! Who will wake it up?")
      (begin
        (remove-task! current)
        (resume-next!))))

  ;;; FIFO Queue.
  ;;;
  ;;; I should really use libraries or put this in a separate module, but I
  ;;; wanted this to be self contained so it's more convenient to give it a
  ;;; try.

  (define (make-queue)
    (cons '() '()))

  (define (queue-empty? q)
    (null? (queue-first q)))

  (define (queue-first q)
    (car q))

  (define (queue-last q)
    (cdr q))

  (define (set-queue-first! q el)
    (set-car! q el))

  (define (set-queue-last! q el)
    (set-cdr! q el))

  (define (make-queue-el val prev next)
    (list val prev next))

  (define (queue-el-value e)
    (car e))

  (define (queue-el-prev e)
    (cadr e))

  (define (set-queue-el-prev! el prev)
    (set-car! (cdr el) prev))

  (define (queue-el-next e)
    (caddr e))

  (define (set-queue-el-next! el next)
    (set-car! (cddr el) next))

  (define (queue-push! q val)
    (let ((el (make-queue-el val '() (queue-first q))))
        (if (queue-empty? q)
          (set-queue-last! q el)
          (set-queue-el-prev! (queue-first q) el))
        (set-queue-first! q el)))

  (define (queue-pop! q)
    (let ((last-el (queue-last q)))
      (set-queue-last! q (queue-el-prev last-el))
      (if (null? (queue-last q))
        (set-queue-first! q '())
        (begin
          ; Break references to let old items be garbage collected.
          (set-queue-el-next! (queue-last q) '())
          (set-queue-el-prev! last-el '())))
      (queue-el-value last-el)))

  ;;; Main task.

  (define current #f)
  (set! current (make-task #f))
  (set-prev-task! current current)
  (set-next-task! current current)

  ;;; Implementation of exported symbols.

@@  (set! add-next-task!
@@    (lambda (thunk)
@@      (let ((t (make-task thunk)))
@@        (set-prev-task! t current)
@@        (set-next-task! t (next-task current))
@@        (set-prev-task! (next-task current) t)
@@        (set-next-task! current t))))

  (set! make-channel
    (lambda ()
      (cons 'idle (make-queue))))

@@  (set! yield
@@    (lambda ()
@@      (call-with-current-continuation
@@        (lambda (k)
@@          (set-task-continuation! current k)
@@          (resume-next!)))))

  (set! last-task?
    (lambda ()
      (eq? (next-task current) current)))

@@  (set! send-to-channel!
@@    (lambda (ch val)
@@      (cond
@@        ((channel-receiving? ch)
@@          (let ((receiver-continuation (pop-channel-receiver! ch)))
@@            (add-next-task! (lambda () (receiver-continuation val)))))
@@        (else
@@          (call-with-current-continuation
@@            (lambda (k)
@@              (push-channel-sender! ch (make-sender k val))
@@              (block!)))))))

@@  (set! receive-from-channel!
@@    (lambda (ch)
@@      (cond
@@        ((channel-sending? ch)
@@          (let ((sender (pop-channel-sender! ch)))
@@            (add-next-task! (sender-continuation sender))
@@            (sender-value sender)))
@@        (else
@@          (call-with-current-continuation
@@            (lambda (k)
@@              (push-channel-receiver! ch k)
@@              (block!)))))))

  (set! channel-sending?
    (lambda (ch)
      (eq? (channel-state ch) 'sending)))

  (set! channel-receiving?
    (lambda (ch)
      (eq? (channel-state ch) 'receiving))))


;;; Debug.

(define (test-task)
  (let ((ch (make-channel)))
    (add-next-task!
      (lambda ()
        (display "yielding")
        (newline)
        (yield)
        (display "sending A")
        (newline)
        (send-to-channel! ch 42)
        (display "done sending A")
        (newline)))
    (add-next-task!
      (lambda ()
        (display "receiving X")
        (newline)
        (display (receive-from-channel! ch))
        (newline)))
    (add-next-task!
      (lambda ()
        (display "sending B")
        (newline)
        (send-to-channel! ch 'muac)
        (display "sent B")
        (newline)))
    ch))