racket-matrix-2012/os2.rkt

#lang racket/base
;; Virtualized operating system, this time with presence.

(require racket/set)
(require racket/match)
(require (only-in racket/list flatten))
(require "unify.rkt")

(provide nested-vm
	 ground-vm

	 (struct-out topic)
	 topic-publisher
	 topic-subscriber
	 co-roles
	 co-topics
	 topic-union

	 (struct-out handlers)

	 (except-out (struct-out transition) transition)
	 (rename-out [make-transition transition])
	 extend-transition

	 role
	 (except-out (struct-out add-role) add-role)
	 (rename-out [make-add-role add-role])
	 (except-out (struct-out delete-role) delete-role)
	 (rename-out [make-delete-role delete-role])
	 (except-out (struct-out send-message) send-message)
	 (rename-out [make-send-message send-message])
	 (except-out (struct-out spawn) spawn)
	 (rename-out [make-spawn spawn])
	 (except-out (struct-out kill) kill)
	 (rename-out [make-kill kill])

	 (struct-out at-meta-level)

	 ;; Aliases of structures themselves, because of shadowing of
	 ;; constructors/type-names.
	 (rename-out [topic <topic>])
	 (rename-out [handlers <handlers>])
	 (rename-out [transition <transition>])
	 (rename-out [add-role <add-role>])
	 (rename-out [delete-role <delete-role>])
	 (rename-out [send-message <send-message>])
	 (rename-out [spawn <spawn>])
	 (rename-out [kill <kill>])
	 (rename-out [at-meta-level <at-meta-level>])

	 ;; Reexports from unify.rkt for convenience
	 wild
	 wild?
	 non-wild?

	 ;; Reexports from racket/match for convenience
	 (all-from-out racket/match))

;; Endpoints are the units of deduplication.
;; Flows (in canonical form) are the units of presence.

;;---------------------------------------------------------------------------
;; Data definitions

;; A PID is an (arbitrary) VM-unique process identifier. Concretely,
;; it's an integer.

;; A PreEID is an arbitrary, process-chosen-and-supplied identifier
;; for an endpoint. It is to be equal?-comparable. It is to be unique
;; within the scope of a single process.

;; A EID is an (arbitrary) VM-unique endpoint identifier. Concretely,
;; it's an (eid PID PreEID). As a consequence of the scope of PreEIDs,
;; EIDs shouldn't be visible outside the scope of the owning process.
(struct eid (pid pre-eid) #:transparent)

;; One endpoint, one topic (which may be changed over time!), with the
;; caveat that as we are at present unable to represent true topic
;; unions, we actually store a *set* of topics against each
;; endpoint. The (current!) topic for the endpoint is to be taken as
;; the union of all the members in the set.

;; A Flow is a Topic that comes from the intersection of two dual
;; topics.

;; A sent message includes a "body" and a "role", and is equivalent to
;; a non-virtual topic with that role and with the given "body" as a
;; pattern. In a sense, topics quite literally are patterns over
;; entire messages.

;; A QuasiQueue<X> is a list of Xs in *reversed* order.

(struct vm (processes ;; Hash<PID, Process>
	    endpoints ;; Hash<EID, Endpoint>
	    next-process-id ;; PID
	    pending-actions ;; QuasiQueue<(cons PID Action)>
	    ) #:transparent)

;; An Endpoint is an (endpoint EID Set<Topic> Handlers), representing a
;; facet of a process responsible for playing a particular role (the
;; topic) in a conversation.
(struct endpoint (id topics handlers) #:transparent)

;; A Process is an Exists State . (process PID State Set<EID>),
;; representing a VM process and its collection of active endpoints.
(struct process (id state endpoints) #:transparent)

;; A Topic is a (topic Role Pattern Boolean), describing an Endpoint's
;; role in a conversation.
(struct topic (role pattern virtual?) #:prefab)

;; BootK = either (PID -> Transition) or Transition
;; InterruptK = State -> Transition
;; TrapK<X> = X -> InterruptK

;; (handlers Maybe<TrapK<Topic>>
;;           Maybe<TrapK<Topic * Reason>>
;;           Maybe<TrapK<Topic * Message>>)
(struct handlers (presence absence message) #:transparent)

;; actions is a plain old ordered ConsTreeOf<Action>, not a
;; QuasiQueue.
(struct transition (state actions) #:transparent)

;; Preactions.
;; Ks are various TrapKs or #f, signifying lack of interest.
;;
;; (add-role PreEID (or Topic Set<Topic>) Handlers)
(struct add-role (pre-eid topics handlers) #:prefab)
;;
;; (delete-role PreEID Any)
(struct delete-role (pre-eid reason) #:prefab)
;;
;; (send-message Any Role)
(struct send-message (body role) #:prefab)
;;
;; (spawn BootK Maybe<TrapK<PID>>)
(struct spawn (main k) #:prefab)
;;
;; (kill Maybe<PID> Any)
(struct kill (pid reason) #:prefab)

;; An Action is either a Preaction or an (at-meta-level Preaction).
(struct at-meta-level (preaction) #:prefab)

;;---------------------------------------------------------------------------
;; role macro

(require (for-syntax syntax/parse))
(require (for-syntax racket/base))
(define-syntax role
  (lambda (stx)
    (syntax-parse stx
      [(_ pre-eid topics-expr
	  #:state state-pattern
	  (~or (~optional (~seq #:on-presence presence) #:name "#:on-presence handler")
	       (~optional (~seq #:on-absence absence) #:name "#:on-absence handler")
	       (~optional (~seq #:topic topic) #:defaults ([topic #'t0]) #:name "#:topic")
	       (~optional (~seq #:reason reason) #:defaults ([reason #'r0]) #:name "#:reason"))
	  ...
	  [message-pattern clause-body ...]
	  ...)
       (define-syntax-rule (build-handler args e-attr)
	 (if (not (attribute e-attr))
	     #'#f
	     #`(lambda args (match-lambda [state-pattern e-attr]))))
       (with-syntax ([presence-handler (build-handler (topic) presence)]
		     [absence-handler (build-handler (topic reason) absence)]
		     [message-handler #'(lambda (topic message-body)
					  (lambda (state)
					    (match state
					      [state-pattern
					       (match message-body
						 [message-pattern clause-body ...]
						 ...
						 [_ state])])))])
	 #'(add-role pre-eid
		     topics-expr
		     (handlers presence-handler absence-handler message-handler)))])))

;;---------------------------------------------------------------------------
;; Smarter constructors for transitions and preactions.

(define (make-transition state . actions) (transition state actions))
(define make-add-role add-role) ;; no special treatment required at present
(define (make-delete-role pre-eid [reason #f]) (delete-role pre-eid reason))
(define (make-send-message body [role 'publisher]) (send-message body role))
(define (make-spawn main [k #f]) (spawn main k))
(define (make-kill [pid #f] #:reason [reason #f]) (kill pid reason))

(define (extend-transition t . more-actions)
  (match t
    [(transition state actions) (transition state (list actions more-actions))]
    [state (transition state more-actions)]))

;;---------------------------------------------------------------------------
;; Topics and roles

(define (topic-publisher pattern #:virtual? [virtual? #f])
  (topic 'publisher pattern virtual?))

(define (topic-subscriber pattern #:virtual? [virtual? #f])
  (topic 'subscriber pattern virtual?))

(define (co-roles r)
  (case r
    [(publisher) '(subscriber)]
    [(subscriber) '(publisher)]
    [else #f]))

(define (co-topics t)
  (for/list ([co-role (co-roles (topic-role t))])
    (struct-copy topic t [role co-role])))

(define (topic-union . ts)
  (unless (andmap topic? ts)
    (error 'topic-union "Expects topics as arguments, but given ~v" ts))
  (list->set ts))

(define (refine-topic remote-topic new-pattern)
  (struct-copy topic remote-topic [pattern new-pattern]))

(define (roles-intersect? l r)
  (memq l (co-roles r)))

;; Both left and right must be canonicalized.
(define (topic-intersection left right)
  (and (roles-intersect? (topic-role left) (topic-role right))
       (mgu-canonical (freshen (topic-pattern left)) (freshen (topic-pattern right)))))

;; True iff the flow between remote-topic and local-topic should be
;; visible to the local peer. This is the case when either local-topic
;; is virtual (in which case everything is seen) or otherwise if
;; remote-topic is also not virtual.
(define (flow-visible? local-topic remote-topic)
  (or (topic-virtual? local-topic)
      (not (topic-virtual? remote-topic))))

;;---------------------------------------------------------------------------
;; Core virtualizable virtual machine.

(define (make-vm boot)
  (vm (hash)
      (hash)
      0
      (list (cons -1 (spawn boot #f)))))

(define (run-vm state)
  (let loop ((remaining-actions (reverse (vm-pending-actions state)))
	     (state (struct-copy vm state [pending-actions '()]))
	     (outbound-actions '()))
    (match remaining-actions
      ['()
       (transition (collect-dead-processes state) (reverse outbound-actions))]
      [(cons (cons pid action) rest)
       (match action
	 [(at-meta-level preaction)
	  (define transformed-preaction (transform-meta-action pid preaction))
	  (loop rest state (cons transformed-preaction outbound-actions))]
	 [preaction
	  (loop rest (perform-action pid preaction state) outbound-actions)])])))

(define (collect-dead-processes state)
  (struct-copy vm state
	       [processes (for/hash ([(pid p) (in-hash (vm-processes state))]
				     #:when (or (not (set-empty? (process-endpoints p)))
						(ormap (lambda (entry) (= (car entry) pid))
						       (vm-pending-actions state))))
			    (values pid p))]))

(define (send-to-user failure-proc f . args)
  (with-handlers ([exn:fail? failure-proc])
    (apply f args)))

(define (ensure-topic-union t)
  (cond [(topic? t) (set t)]
	[(set? t) t]
	[else
	 (error 'ensure-topic-union
		"Expected either a single topic or a set of topics; got ~v"
		t)]))

(define (perform-action pid preaction state)
  (match preaction
    [(add-role pre-eid topics hs) (do-subscribe pid pre-eid (ensure-topic-union topics) hs state)]
    [(delete-role pre-eid reason) (do-unsubscribe pid pre-eid reason state)]
    [(send-message body role) (route-and-deliver role body state)]
    [(spawn main k) (do-spawn pid main k state)]
    [(kill pid-to-kill reason) (do-kill (or pid-to-kill pid) reason state)]))

(define (do-subscribe pid pre-eid topics hs state)
  (cond
   [(hash-has-key? (vm-processes state) pid)
    (define new-eid (eid pid pre-eid))
    (define new-endpoint (endpoint new-eid topics hs))
    (let* ((state (notify-route-additions state new-endpoint))
	   (state (generic-update-process state pid (add-process-eid new-eid)))
	   (state (install-endpoint state new-eid new-endpoint)))
      state)]
   [else state]))

(define (generic-update-process state pid updater)
  (struct-copy vm state [processes (hash-update (vm-processes state) pid updater)]))

(define ((add-process-eid new-eid) p)
  (struct-copy process p [endpoints (set-add (process-endpoints p) new-eid)]))

(define ((remove-process-eid old-eid) p)
  (struct-copy process p [endpoints (set-remove (process-endpoints p) old-eid)]))

(define (install-endpoint state new-eid new-endpoint)
  (struct-copy vm state [endpoints (hash-set (vm-endpoints state) new-eid new-endpoint)]))

(define (uninstall-endpoint state old-eid)
  (struct-copy vm state [endpoints (hash-remove (vm-endpoints state) old-eid)]))

(define (notify-route-additions state new-endpoint)
  (match-define (endpoint (eid pid _) topics (handlers presence-handler _ _)) new-endpoint)
  (for*/fold ([state state])
      ([(matching-pid p) (in-hash (vm-processes state))]
       [matching-eid (in-set (process-endpoints p))]
       [e (in-value (hash-ref (vm-endpoints state) matching-eid))]
       [matching-topics (in-value (endpoint-topics e))]
       [matching-topic (in-set matching-topics)]
       [topic (in-set topics)]
       [flow-pattern (in-value (topic-intersection topic matching-topic))]
       #:when flow-pattern)
    (define inbound-flow (refine-topic matching-topic flow-pattern))
    (define outbound-flow (refine-topic topic flow-pattern))
    (define e-presence-handler (handlers-presence (endpoint-handlers e)))
    (let* ((state (if (flow-visible? topic inbound-flow)
		      (run-trapk state pid presence-handler inbound-flow)
		      state))
	   (state (if (flow-visible? matching-topic outbound-flow)
		      (run-trapk state matching-pid e-presence-handler outbound-flow)
		      state)))
      state)))

(define (do-unsubscribe pid pre-eid reason state)
  (define old-eid (eid pid pre-eid))
  (cond
   [(hash-has-key? (vm-endpoints state) old-eid)
    (define old-endpoint (hash-ref (vm-endpoints state) old-eid))
    (let* ((state (generic-update-process state pid (remove-process-eid old-eid)))
	   (state (uninstall-endpoint state old-eid))
	   (state (notify-route-deletions state old-endpoint reason)))
      state)]
   [else state]))

(define (notify-route-deletions state old-endpoint reason)
  (define removed-topics (endpoint-topics old-endpoint))
  (for*/fold ([state state])
      ([(matching-pid p) (in-hash (vm-processes state))]
       [matching-eid (in-set (process-endpoints p))]
       [e (in-value (hash-ref (vm-endpoints state) matching-eid))]
       [matching-topics (in-value (endpoint-topics e))]
       [matching-topic (in-set matching-topics)]
       [removed-topic (in-set removed-topics)]
       [flow-pattern (in-value (topic-intersection removed-topic matching-topic))]
       #:when flow-pattern)
    (define outbound-flow (refine-topic removed-topic flow-pattern))
    (define absence-handler (handlers-absence (endpoint-handlers e)))
    (run-trapk state matching-pid absence-handler outbound-flow reason)))

(define (route-and-deliver role body state)
  (define message-topic (topic role body #f))
  (define endpoints
    (for*/set ([(matching-pid p) (in-hash (vm-processes state))]
	       [matching-eid (in-set (process-endpoints p))]
	       [e (in-value (hash-ref (vm-endpoints state) matching-eid))]
	       [matching-topics (in-value (endpoint-topics e))]
	       [matching-topic (in-set matching-topics)]
	       [flow-pattern (in-value (topic-intersection message-topic matching-topic))]
	       #:when flow-pattern)
      e))
  (for/fold ([state state]) ([e (in-set endpoints)])
    (match-define (endpoint (eid pid _) _ (handlers _ _ message-handler)) e)
    (run-trapk state pid message-handler message-topic body)))

(define (do-spawn spawning-pid main k state)
  (define new-pid (vm-next-process-id state))
  (match-define (transition initial-state initial-actions)
    (cond
     [(procedure? main) (send-to-user (lambda (e) (transition #f (kill #f e))) main new-pid)]
     [(transition? main) main]))
  (define initial-process (process new-pid initial-state (set)))
  (define spawned-state
    (struct-copy vm (enqueue-actions state new-pid initial-actions)
		 [processes (hash-set (vm-processes state) new-pid initial-process)]
		 [next-process-id (+ new-pid 1)]))
  (run-trapk spawned-state spawning-pid k new-pid))

(define (print-kill pid-to-kill reason)
  (cond
   [(eq? reason #f) (printf "PID ~v exited normally~n" pid-to-kill)]
   [(exn? reason) ((error-display-handler)
		   (format "PID ~v exited with exception~n~a" pid-to-kill (exn-message reason))
		   reason)]
   [else (printf "PID ~v exited with reason: ~a~n" pid-to-kill reason)]))

(define (do-kill pid-to-kill reason state)
  (cond
   [(hash-has-key? (vm-processes state) pid-to-kill)
    (print-kill pid-to-kill reason)
    (define dying-endpoints (process-endpoints (hash-ref (vm-processes state) pid-to-kill)))
    (let* ((state (for/fold ([state state]) ([eid (in-set dying-endpoints)])
		    (do-unsubscribe pid-to-kill (eid-pre-eid eid) reason state))))
      (struct-copy vm state [processes (hash-remove (vm-processes state) pid-to-kill)]))]
   [else state]))

(define (run-trapk state pid trap-k . args)
  (if trap-k
      (let ((failure-proc (lambda (e) (lambda (process-state)
					(transition process-state (kill #f e))))))
	(run-ready state pid (apply send-to-user failure-proc trap-k args)))
      state))

(define (maybe-transition->transition t)
  (cond [(transition? t) t]
	[else (transition t '())]))

(define (run-ready state pid interrupt-k)
  (define old-state (process-state (hash-ref (vm-processes state) pid)))
  (match-define (transition new-state actions)
    (maybe-transition->transition
     (send-to-user (lambda (e) (transition old-state (kill #f e))) interrupt-k old-state)))
  (generic-update-process (enqueue-actions state pid actions)
			  pid
			  (lambda (p) (struct-copy process p [state new-state]))))

(define (preaction? a)
  (or (add-role? a)
      (delete-role? a)
      (send-message? a)
      (spawn? a)
      (kill? a)))

(define (action? a)
  (or (preaction? a)
      (and (at-meta-level? a)
	   (preaction? (at-meta-level-preaction a)))))

(define (valid-action? pid a)
  (cond
   [(action? a)]
   [(eq? a #f) #f] ;; skip falses in action ConsTrees
   [(void? a) #f] ;; skip voids in action ConsTrees
   [else (log-warning (format "Illegal action ~v from pid ~v" a pid))
	 #f]))

(define (enqueue-actions state pid actions)
  (define flat-actions (for/list ([a (flatten actions)] #:when (valid-action? pid a)) (cons pid a)))
  (struct-copy vm state
	       [pending-actions (append (reverse flat-actions) (vm-pending-actions state))]))

(define (((wrap-trapk pid trapk) . args) state)
  (apply run-trapk state pid trapk args))

(define (transform-meta-action pid preaction)
  (match preaction
    [(add-role pre-eid topics hs)
     (add-role (eid pid pre-eid)
	       topics
	       (handlers (wrap-trapk pid (handlers-presence hs))
			 (wrap-trapk pid (handlers-absence hs))
			 (wrap-trapk pid (handlers-message hs))))]
    [(delete-role pre-eid reason)
     (delete-role (eid pid pre-eid) reason)]
    [(? send-message? p) p]
    [(spawn main k)
     (spawn main (wrap-trapk pid k))]
    [(? kill? p) p]))

(define (nested-vm boot)
  (lambda (self-pid) (run-vm (make-vm boot))))

(define (ground-vm boot)
  (let loop ((state (make-vm boot)))
    (match (run-vm state)
      [(transition state actions)
       (when (not (null? actions))
	 (error 'ground-vm "Cannot process meta-actions because no further meta-level exists: ~v"
		actions))
       (define waiting? (null? (vm-pending-actions state)))
       (define active-events
	 (for*/fold ([acc '()])
	     ([(eid e) (in-hash (vm-endpoints state))]
	      [active-topic (in-set (endpoint-topics e))])
	   (match active-topic
	     [(topic 'subscriber (cons (? evt? evt) _) #f)
	      (cons (wrap-evt evt (lambda (message)
				    (lambda (state)
				      (route-and-deliver 'publisher
							 (cons evt message)
							 state))))
		    acc)]
	     [_ acc])))
       (if (and waiting? (null? active-events))
	   'done ;; About to block, and nothing can wake us
	   (let ((interruptk (apply sync
				    (if waiting?
					never-evt
					(wrap-evt always-evt (lambda (dummy) values)))
				    active-events)))
	     (loop (interruptk state))))])))