;; Producer/Consumer-Problem mit Puffer endlicher Größe,
;; mit asynchronem Message-Passing

(define-record-type :buffer
  (really-make-buffer space-channel
		      insert-request-channel insert-channel
		      remove-request-channel remove-channel
		      remove-ack-channel)
  buffer?
  (space-channel buffer-space-channel)
  (insert-request-channel buffer-insert-request-channel)
  (insert-channel buffer-insert-channel)
  (remove-request-channel buffer-remove-request-channel)
  (remove-channel buffer-remove-channel)
  (remove-ack-channel buffer-remove-ack-channel))

(define (make-buffer max-size)
  (let ((queue (make-finite-queue max-size))
	(space-channel (make-async-channel))
	(insert-request-channel (make-async-channel))
	(insert-channel (make-async-channel))
	(remove-channel (make-async-channel))
	(remove-request-channel (make-async-channel))
	(remove-ack-channel (make-async-channel)))
    (spawn
     (lambda ()
       (letrec ((insert
		 (lambda ()
		   (send-async space-channel #f)
		   (add-to-finite-queue! queue (receive-async insert-channel))
		   (switch)))
		(remove
		 (lambda ()
		   (send-async remove-channel
			       (remove-from-finite-queue! queue))
		   (receive-async remove-ack-channel)
		   (switch)))
		(empty
		 (lambda ()
		   (receive-async insert-request-channel)
		   (insert)))
		(full
		 (lambda ()
		   (receive-async remove-request-channel)
		   (remove)))
		(neither
		 (lambda ()
		   (receive-async-selective
		    (list
		     (cons insert-request-channel
			   (lambda (message)
			     (insert)))
		     (cons remove-request-channel
			   (lambda (message)
			     (remove)))))))
		(switch
		    (lambda ()
		      (cond
		       ((finite-queue-full? queue) (full))
		       ((finite-queue-empty? queue) (empty))
		       (else (neither))))))
	 (empty))))
    (really-make-buffer space-channel
			insert-request-channel insert-channel
			remove-request-channel remove-channel
			remove-ack-channel)))

(define (insert buffer value)
  (send-async (buffer-insert-request-channel buffer) value)
  (receive-async (buffer-space-channel buffer))
  (send-async (buffer-insert-channel buffer) value))

(define (remove buffer)
  (send-async (buffer-remove-request-channel buffer) #f)
  (let ((value
	 (receive-async (buffer-remove-channel buffer))))
    (send-async (buffer-remove-ack-channel buffer) #f)
    value))