#lang racket/base

;; Virtualized operating system.

(require racket/match)
(require racket/list)
(require "functional-queue.rkt")

(provide
 ;; Waiting for messages
 (struct-out subscription)
 (struct-out message-handler)

 ;; Kernel requests
 (struct-out kernel-mode-transition)

 ;; Constructing, accessing and running VMs
 make-vm
 vm?
 run-vm

 ;; Grounding out the infinite tower of VMs
 default-pattern-predicate
 ground-vm
 )

;; Each VM hosts 0 or more *multiplexed* processes. Each process has
;; its own state record. In between schedulings, a process consists of
;; 1 or more message handlers. A message handler is a pair of a message
;; recogniser and a procedure taking a message and a process state to a
;; new process state.
;;
;; Each VM provides a *communication bus* for its processes to
;; use. The communication bus is the only form of IPC the VM provides.
;;
;; Some processes *relay* messages out from the VM to other
;; VMs. Because the "tree" of VMs so formed has to be a tree - See
;; Shivers & Might 2006 for a discussion of this - we gather together
;; all the interactions between the supervenient VM and its support VM
;; into a single channel of communication. The relaying processes are,
;; in effect, device-drivers, providing application-specific
;; communication services to other processes in the VM.
;;
;; We split processes into "user" processes, permitted only to spawn
;; other user processes and send messages on the VM's bus, and
;; "kernel" processes, permitted also to spawn other kernel processes
;; and send messages to the VM's container.
;;
;; Time plays an interesting role in a distributed system: if the
;; medium messages are sent through isn't cooperative enough to let us
;; know of a failed conversational participant, our only recourse is
;; /timeout/. Therefore, we require every level of the machine to
;; support timeouts, though we do not require such timeouts to be tied
;; to real, wall-clock time: simulated time is just fine. This helps
;; with testability.
;;
;; Racket's alarm-evt is almost the right design for timeouts: its
;; synchronisation value should be the (or some) value of the clock
;; after the asked-for time. That way it serves as timeout and
;; clock-reader in one.

;; VMs are parameterised over:
;;  - the type of messages carried on the bus, Message
;;  - the type of patterns over Messages, MessagePattern
;;  - the type of messages to other VMs, MetaMessage
;;  - the type of patterns over MetaMessages, MetaMessagePattern

;; A PatternPredicate is a (MessagePattern Message -> Boolean), used
;; to match a message against a pattern.

;; A VM is a (vm ListBagOf<Suspension>
;;               QueueOf<Message> ;; TODO: make unordered?
;;               QueueOf<MetaMessage> ;; TODO: make unordered?
;;               QueueOf<BootK>).
(struct vm (suspensions
	    pending-messages
	    pending-meta-messages
	    pending-processes
	    pattern-predicate) #:transparent)

;; A TrapK<X> is a X -> InterruptK, representing a suspended process
;; waiting for some information from the VM before it can continue.

;; An InterruptK is a
;;   ProcessState -> KernelModeTransition
;; representing a suspended process that can run instantly without
;; waiting for more information from the VM. The input is the state of
;; the process, and the output is the information passed back to the
;; VM when the process yields the CPU.

;; A BootK is a ( -> KernelModeTransition), representing either a
;; fresh process or a previously-suspended process just about to
;; resume.

;; A Subscription is a
;; (subscription ProcessState
;;               Maybe<InterruptK>
;;               ListBagOf<MessageHandler>
;;               ListBagOf<MetaMessageHandler>).
;; To poll the kernel, include a non-#f InterruptK.
(struct subscription (state
		      k
		      message-handlers
		      meta-message-handlers) #:transparent)

;; A Suspension is a
;; (suspension ProcessState
;;             Maybe<InterruptK>
;;             ListBagOf<MessageHandler>
;;             Map<HID,MetaMessageHandler>).
(struct suspension (state
		    k
		    message-handlers
		    meta-message-handlers) #:transparent)

;; A HID is a per-VM unique value, used to identify specific
;; MetaMessageHandlers. Here, we use gensyms, though an alternative
;; (and purer) approach would be to keep a counter in the VM and use
;; that to construct IDs.

;; A MessageHandler is one of
;; -- (message-handler MessagePattern TrapK<Message>)
(struct message-handler (pattern k) #:transparent)

;; A KernelModeTransition is a
;; (kernel-mode-transition Subscription
;;                         ListBagOf<Message>
;;			   ListBagOf<MetaMessage>
;;                         ListBagOf<BootK>)
;; representing the subscription for the transitioning process, a list
;; of messages to emit at both this VM's and its container's level,
;; and a list of new processes to create and schedule.
(struct kernel-mode-transition (subscription
				messages
				meta-messages
				new-processes) #:transparent)

;; A ListBagOf<X> is a ListOf<X> with the additional constraint that
;; order isn't meaningful.

;; TODO: is timeout really primitive? If so, isn't presence primitive?
;; TODO: what about metatimeout?
;; TODO: what about spawn-meta-process etc? Come back to this later.
;; TODO: enforce user-mode restrictions
;; TODO: timeouts

;; BootK -> VM
(define (make-vm boot #:pattern-predicate [pattern-predicate default-pattern-predicate])
  (vm (list)
      (make-queue)
      (make-queue)
      (enqueue (make-queue) boot)
      pattern-predicate))

;; VM -> KernelModeTransition
;; (A kind of Meta-InterruptK)
(define (run-vm state)
  (let* ((state (requeue-pollers state))
	 (state (run-runnables state))
	 (state (dispatch-messages state))
	 (meta-messages (queue->list (vm-pending-meta-messages state)))
	 (meta-handlers (append-map extract-downward-meta-message-handlers (vm-suspensions state)))
	 (poller-k (and (should-poll? state) run-vm)) ;; only block if there's nothing left to do
	 (state (struct-copy vm state [pending-meta-messages (make-queue)])))
    (kernel-mode-transition (subscription state poller-k meta-handlers '())
			    meta-messages
			    '()
			    '())))

(define (requeue-pollers state)
  (foldl (lambda (susp state)
	   (if (suspension-polling? susp)
	       (enqueue-runnable (lambda () ((suspension-k susp) (suspension-state susp))) state)
	       (enqueue-suspension susp state)))
	 (struct-copy vm state [suspensions '()])
	 (vm-suspensions state)))

(define (run-runnables state)
  (foldl (lambda (r state) (perform-transition (r) state))
	 (struct-copy vm state [pending-processes (make-queue)])
	 (queue->list (vm-pending-processes state))))

(define (dispatch-messages state)
  (foldl dispatch-message
	 (struct-copy vm state [pending-messages (make-queue)])
	 (queue->list (vm-pending-messages state))))

(define (extract-downward-meta-message-handlers susp)
  (for/list ([(hid mmh) (in-hash (suspension-meta-message-handlers susp))])
    (message-handler (message-handler-pattern mmh) (dispatch-meta-message hid))))

(define (extract-upward-meta-message-handlers susp)
  (for/list ([(hid mmh) (in-hash (suspension-meta-message-handlers susp))])
    (message-handler hid (message-handler-k mmh))))

(define (((dispatch-meta-message hid) message) state)
  (run-vm
   (foldl (match-suspension message
			    (lambda (handler-hid message) (equal? hid handler-hid))
			    extract-upward-meta-message-handlers)
	  (struct-copy vm state [suspensions '()])
	  (vm-suspensions state))))

;; KernelModeTransition VM -> VM
(define (perform-transition transition state)
  (match transition
    [(kernel-mode-transition new-subscription
			     messages
			     meta-messages
			     new-processes)
     (let* ((state (foldl enqueue-message state messages))
	    (state (foldl enqueue-runnable state new-processes))
	    (state (enqueue-suspension (subscription->suspension new-subscription) state))
	    (state (foldl enqueue-meta-message state meta-messages)))
       state)]
    [other
     (error 'vm "Processes must return a kernel-mode-transition struct; got ~v" other)]))

(define (subscription->suspension sub)
  (match-define (subscription ps k mhs mmhs) sub)
  (suspension ps k mhs (for/hash ([mmh mmhs]) (values (gensym 'hid) mmh))))

(define (enqueue-message message state)
  (struct-copy vm state [pending-messages (enqueue (vm-pending-messages state) message)]))

(define (enqueue-runnable r state)
  (struct-copy vm state [pending-processes (enqueue (vm-pending-processes state) r)]))

(define (enqueue-suspension susp state)
  (match susp
    [(suspension _ #f '() (? (lambda (h) (zero? (hash-count h)))))
     ;; dead process because no continuations offered
     state]
    [(suspension _ _ _ _)
     (struct-copy vm state [suspensions (cons susp (vm-suspensions state))])]))

(define (enqueue-meta-message message state)
  (struct-copy vm state [pending-meta-messages (enqueue (vm-pending-meta-messages state) message)]))

(define (dispatch-message message state)
  (foldl (match-suspension message
			   (vm-pattern-predicate state)
			   suspension-message-handlers)
	 (struct-copy vm state [suspensions '()])
	 (vm-suspensions state)))

(define ((match-suspension message apply-pattern handlers-getter) susp state)
  (let search-handlers ((message-handlers (handlers-getter susp)))
    (cond
     [(null? message-handlers)
      ;; No handler matched this message. Put the suspension
      ;; back on the list for some future message.
      (enqueue-suspension susp state)]
     [(apply-pattern (message-handler-pattern (car message-handlers)) message)
      (define trapk (message-handler-k (car message-handlers)))
      (define interruptk (trapk message))
      (perform-transition (interruptk (suspension-state susp)) state)]
     [else
      (search-handlers (cdr message-handlers))])))

(define (suspension-polling? susp)
  (not (eq? (suspension-k susp) #f)))

;; VM -> Boolean
;; When should a VM block? When it has no runnables, no pending
;; messages, and no polling suspensions. Otherwise, it should poll.
(define (should-poll? state)
  (or (not (queue-empty? (vm-pending-processes state)))
      (not (queue-empty? (vm-pending-messages state)))
      (ormap suspension-polling? (vm-suspensions state))))

;;---------------------------------------------------------------------------

(define (nested-vm-inert? sub)
  (match sub
    [(subscription (vm _ (? queue-empty?) (? queue-empty?) (? queue-empty?) _) #f '() '())
     ;; Inert iff not waiting for any messages or metamessages, and
     ;; with no internal work left to do.
     #t]
    [_ #f]))

(define default-pattern-predicate
  (lambda (p m) (p m)))

;; PatternPredicate ( -> KernelModeTransition ) -> Void
;; In this context,
;;   Message = a thunk
;;   MessagePattern = evt?
;;   MetaMessage, MetaMessagePattern = not defined because there's no outer level
;; Runs its argument VM until it becomes (provably) inert.
(define (ground-vm boot #:pattern-predicate [pattern-predicate default-pattern-predicate])
  (let loop ((transition (run-vm (make-vm boot #:pattern-predicate pattern-predicate))))
    (for-each (lambda (thunk) (thunk)) (kernel-mode-transition-messages transition))
    (when (not (nested-vm-inert? (kernel-mode-transition-subscription transition)))
      (match transition
	[(kernel-mode-transition (subscription new-state
					       polling-k
					       message-handlers
					       '())
				 _
				 '()
				 '())
	 (define inbound-messages
	   (map (match-lambda [(message-handler e k) (wrap-evt e (lambda (v) (cons v k)))])
		message-handlers))
	 (match-define (cons inbound-value inbound-continuation)
	   (apply sync
		  (wrap-evt (if polling-k always-evt never-evt)
			    (lambda (v) (cons (void)
					      (lambda (dummy) polling-k))))
		  inbound-messages))
	 (loop ((inbound-continuation inbound-value) new-state))]
	[_
	 (error 'ground-vm
		"Outermost VM may not spawn new siblings or send or receive metamessages")]))))