#lang racket/base
;; Core implementation of network actors and Network Calculus (NC) communication API.

(require racket/set)
(require racket/match)
(require racket/list)
(require "route.rkt")
(require "gestalt.rkt")
(require "functional-queue.rkt")
(require (only-in web-server/private/util exn->string))

(provide (struct-out routing-update)
	 (struct-out message)
	 (struct-out quit)
	 (struct-out process)
	 (struct-out transition)

	 ;; imported from route.rkt:
	 ?
	 wildcard?
	 ?!
	 (struct-out capture)
	 pretty-print-matcher
	 matcher-key-set
	 matcher-key-set/single
	 projection->pattern
	 matcher-empty?

	 sub
	 pub
	 gestalt-accepts?

	 spawn
	 send
	 feedback
	 spawn-world
	 deliver-event
	 transition-bind
	 sequence-transitions
	 log-events-and-actions?
	 routing-implementation)

;; A PID is a number uniquely identifying a Process within a World.
;; Note that PIDs are only meaningful within the context of their
;; World: they are not global Process identifiers.

;; (Parameterof (Listof PID))
;; Path to the active leaf in the process tree. The car end is the
;; leaf; the cdr end, the root. Used for debugging purposes.
(define pid-stack (make-parameter '()))

;; (Parameterof Boolean)
;; True when Worlds should log their internal actions for use in
;; debugging.
(define log-events-and-actions? (make-parameter #f))

;; TODO: support +Inf.0 as a level number

;; An Event is a communication from a World to a Process contained
;; within it. One of
;;  - (routing-update Gestalt), description of change in the sender's interests/subscriptions
;;  - (message Any Nat Boolean), a (multicast, in general) message sent by an actor
;; A message's (feedback?) field is #f when it is a message
;; originating from an advertiser/publisher and terminating with a
;; subscriber, and #t in the opposite case.
(struct routing-update (gestalt) #:prefab)
(struct message (body meta-level feedback?) #:prefab)

;; An Action is a communication from a Process to its containing
;; World, instructing the World to take some action on the Process's
;; behalf. One of
;;  - an Event: change in the Process's interests, or message from the Process
;;  - a Process: instruction to spawn a new process as described
;;  - (quit): instruction to terminate the sending process
(struct quit () #:prefab)

;; A PendingEvent is a description of a set of Events to be
;; communicated to a World's Processes. In naïve implementations of
;; NC, there is no distinction between Events and PendingEvents; here,
;; we must ensure that the buffering delay doesn't affect the Gestalts
;; communicated in routing-update Events, so a special record is used
;; to capture the appropriate Gestalt environment.
;;  - (pending-routing-update Gestalt Gestalt (Option PID))
(struct pending-routing-update (aggregate affected-subgestalt known-target) #:prefab)

;; A Process (a.k.a. Actor) describes a single actor in a World.
;;  - (process Gestalt Behavior Any)
;; The Gestalt describes the current interests of the Process: either
;; those it was spawned with, or the most recent interests from a
;; routing-update Action.
(struct process (gestalt behavior state) #:transparent)

;; A World (a.k.a Configuration) is the state of an actor representing
;; a group of communicating Processes. The term is also used from time
;; to time to denote the actor having a World as its state and
;; world-handle-event as its Behavior.
(struct world (next-pid			;; PID, for next-spawned process
	       pending-event-queue	;; (Queueof PendingEvent)
	       runnable-pids		;; (Setof PID), non-inert processes
	       partial-gestalt		;; Gestalt, from local processes only; maps to PID
	       full-gestalt		;; Gestalt, union of partial- and downward-gestalts
	       process-table		;; (HashTable PID Process)
	       downward-gestalt		;; Gestalt, representing interests of outside world
	       process-actions		;; (Queueof (Pairof PID Action))
	       ) #:transparent)

;; A Behavior is a ((Option Event) Any -> Transition): a function
;; mapping an Event (or, in the #f case, a poll signal) and a
;; Process's current state to a Transition.
;;
;; A Transition is either
;;  - #f, a signal from a Process that it is inert and need not be
;;        scheduled until some Event relevant to it arrives; or,
;;  - a (transition Any (Constreeof Action)), a new Process state to
;;        be held by its World and a sequence of Actions for the World
;;        to take on the transitioning Process's behalf.
(struct transition (state actions) #:transparent)

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Protocol and utilities

;; sub : Pattern [#:meta-level Nat] [#:level Nat] -> Gestalt
;; pub : Pattern [#:meta-level Nat] [#:level Nat] -> Gestalt
;;
;; Construct atomic Gestalts representing subscriptions/advertisements
;; matching the given pattern, at the given meta-level and level.
;; These are frequently used in combination with gestalt-union when
;; building spawn and routing-update Actions.
(define (sub p #:meta-level [ml 0] #:level [l 0]) (simple-gestalt #f p l ml))
(define (pub p #:meta-level [ml 0] #:level [l 0]) (simple-gestalt #t p l ml))

;; Gestalt Any -> Boolean
;; True iff m falls within the set of messages represented by the Gestalt.
(define (gestalt-accepts? g m)
  (match-define (message b ml f?) m)
  (not (set-empty? (gestalt-match-value g b ml f?))))

;; Behavior Any [Gestalt] -> Action
;; Constructs a spawn Action for a new process with the given behavior
;; and state. If a Gestalt is supplied, the new process will begin its
;; existence with the corresponding subscriptions/advertisements/
;; conversational-responsibilities.
(define (spawn behavior state [gestalt (gestalt-empty)]) (process gestalt behavior state))

;; send : Any [#:meta-level Nat] -> Action
;; feedback : Any [#:meta-level Nat] -> Action
;;
;; Each constructs an Action that will deliver a body to peers at the
;; given meta-level. (send) constructs messages that will be delivered
;; to subscribers; (feedback), to advertisers.
(define (send body #:meta-level [ml 0]) (message body ml #f))
(define (feedback body #:meta-level [ml 0]) (message body ml #t))

;; Action* -> Action
;; Constructs an action which causes the creation of a new World
;; Process. The given actions will be taken by a primordial process
;; running in the context of the new World.
(define (spawn-world . boot-actions)
  (spawn world-handle-event
	 (enqueue-actions (world 0
				 (make-queue)
				 (set)
				 (gestalt-empty)
				 (gestalt-empty)
				 (hash)
				 (gestalt-empty)
				 (make-queue))
			  -1
			  boot-actions)))

;; Any -> Boolean; type predicates for Event and Action respectively.
(define (event? x) (or (routing-update? x) (message? x)))
(define (action? x) (or (event? x) (process? x) (quit? x)))

;; (Any -> Transition) Transition -> Transition
;; A kind of monad-ish bind operator: threads the state in t0 through
;; k, appending the action sequence from t0 with that from the result
;; of calling k.
;; TODO: sort out exactly how #f should propagate here
(define (transition-bind k t0)
  (match-define (transition state0 actions0) t0)
  (match (k state0)
    [(transition state1 actions1) (transition state1 (cons actions0 actions1))]
    [#f t0]))

;; Transition (Any -> Transition)* -> Transition
;; Each step is a function from state to Transition. The state in t0
;; is threaded through the steps; the action sequences are appended.
(define (sequence-transitions t0 . steps)
  (foldl transition-bind t0 steps))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; Trigger guards

;; TriggerGuards wrap process Behavior and state, only passing through
;; routing-update Events to their contained process behavior/state if
;; there has been a change. All other Events go straight through.
;;
;;  - (trigger-guard Gestalt Behavior Any)
;;
;; The structural similarity to Processes is meaningful: a Process
;; describes the current interests of the Process, as well as its
;; behavior. A TriggerGuard describes the current interests of the
;; Process's *environment*, and doesn't bother passing on a
;; routing-update unless the change is non-zero.
(struct trigger-guard (gestalt handler state) #:transparent)

;; Behavior :> (Option Event) TriggerGuard -> Transition
;; Inspects the given event: if it is a routing update, the contained
;; Gestalt is compared to the TriggerGuard's record of the previous
;; Gestalt from the environment, and only if it is different is it
;; passed on.
(define (trigger-guard-handle e s0)
  (match-define (trigger-guard old-gestalt handler old-state) s0)
  (define (deliver s)
    (match (handler e old-state)
      [#f
       (if (eq? s s0) #f (transition s '()))]
      [(transition new-state actions)
       (transition (struct-copy trigger-guard s [state new-state]) actions)]))
  (match e
    [(routing-update new-gestalt)
     (if (equal? new-gestalt old-gestalt)
	 #f
	 (deliver (struct-copy trigger-guard s0 [gestalt new-gestalt])))]
    [_ (deliver s0)]))

;; Process -> Process
;; Wraps a Process in a TriggerGuard.
(define (trigger-guard-process p)
  (match-define (process _ b s) p)
  (struct-copy process p [behavior trigger-guard-handle] [state (trigger-guard #f b s)]))

;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
;; World implementation

;; Each time a world is handed an event from its environment, it:
;;  1. dispatches PendingEvents
;;      a. removing them one-at-a-time from the queue
;;      b. converting them to Events and dispatching them to processes
;;      c. updating process states and accumulating Actions in the queue
;;      d. any process that returned non-#f is considered "non-idle" for step 3.
;;  2. performs Actions
;;      a. removing them one-at-a-time from the queue
;;      b. interpreting them
;;      c. updating World state and accumulating PendingEvents in the queue
;;  3. steps non-idle processes
;;      a. runs through the runnable-pids set of processes accumulated in 1d. above
;;      b. any process that returned non-#f is put in the "non-idle" set for next time
;;  4. yields updated World state and world Actions to the environment.
;;
;; Note that routing-update Actions are queued as
;; pending-routing-update structures in order to preserve and
;; communicate transient Gestalt states to Processes. In addition, the
;; known-target field of a pending-routing-update structure is used to
;; provide NC's initial Gestalt signal to a newly-spawned process.
;;
;; TODO: should step 3 occur before step 1?

;; World PID (Constreeof Action) -> World
;; Stores actions taken by PID for later interpretation.
(define (enqueue-actions w pid actions)
  (struct-copy world w
    [process-actions (queue-append-list (world-process-actions w)
					(filter-map (lambda (a) (and (action? a) (cons pid a)))
						    (flatten actions)))]))

;; World -> Boolean
;; True if the World has no further reductions it can take.
;;
;; The code is written to maintain the runnable-pids set carefully, to
;; ensure we can locally decide whether we're inert or not without
;; having to search the whole deep process tree.
(define (inert? w)
  (and (queue-empty? (world-pending-event-queue w))
       (queue-empty? (world-process-actions w))
       (set-empty? (world-runnable-pids w))))

;; Event PID Process -> Transition
;; Delivers the event to the process, catching any exceptions it
;; throws and converting them to quit Actions.
(define (deliver-event e pid p)
  (parameterize ((pid-stack (cons pid (pid-stack))))
    (when (and (log-events-and-actions?) e)
      (log-info "EVENT ~a: ~v --> ~v ~v"
		(pid-stack)
		e
		(process-behavior p)
		(if (world? (process-state p))
		    "#<world>"
		    (process-state p))))
    (with-handlers ([(lambda (exn) #t)
		     (lambda (exn)
		       (log-error "Process ~a died with exception:\n~a"
				  (pid-stack)
				  (exn->string exn))
		       (transition (process-state p) (list (quit))))])
      (match (with-continuation-mark 'minimart-process
				     pid ;; TODO: debug-name, other user annotation
				     ((process-behavior p) e (process-state p)))
	[#f #f] ;; inert.
	[(? transition? t) t] ;; potentially runnable.
	[x
	 (log-error "Process ~a returned non-#f, non-transition: ~v" (pid-stack) x)
	 (transition (process-state p) (list (quit)))]))))

;; World PID -> World
;; Marks the given PID as not-provably-inert.
(define (mark-pid-runnable w pid)
  (struct-copy world w [runnable-pids (set-add (world-runnable-pids w) pid)]))

;; PID Transition World -> World
;; Examines the given Transition, updating PID's Process's state and
;; enqueueing Actions for later interpretation. When the Transition is
;; non-#f, PID's Process may wish to take further internal reductions,
;; so we mark it as runnable.
(define (apply-transition pid t w)
  (match t
    [#f w]
    [(transition new-state new-actions)
     (let* ((w (transform-process pid w
				  (lambda (p)
				    (when (and (log-events-and-actions?)
					       (not (null? (flatten new-actions))))
				      (log-info "ACTIONS ~a: ~v <-- ~v ~v"
						(cons pid (pid-stack))
						new-actions
						(process-behavior p)
						(if (world? new-state)
						    "#<world>"
						    new-state)))
				    (struct-copy process p [state new-state])))))
       (enqueue-actions (mark-pid-runnable w pid) pid new-actions))]))

;; PendingEvent World -> World
;; Enqueue a PendingEvent for later interpretation and dispatch.
(define (enqueue-pending-event e w)
  (struct-copy world w [pending-event-queue (enqueue (world-pending-event-queue w) e)]))

;; World -> Transition
;; Examines all queued actions, interpreting them, updating World
;; state, and possibly causing the World to send Actions for
;; interpretation to its own containing World in turn.
(define (perform-actions w)
  (for/fold ([t (transition (struct-copy world w [process-actions (make-queue)]) '())])
      ((entry (in-list (queue->list (world-process-actions w)))))
    (match-define (cons pid a) entry)
    (transition-bind (perform-action pid a) t)))

;; World -> Transition
;; Interprets queued PendingEvents, delivering resulting Events to Processes.
(define (dispatch-pending-events w)
  (transition (for/fold ([w (struct-copy world w [pending-event-queue (make-queue)])])
		  ((e (in-list (queue->list (world-pending-event-queue w)))))
		(dispatch-pending-event e w))
              '()))

;; PID World (Process -> Process) -> World
;; Extracts a Process by PID, maps fp over it, and stores the result back into the table.
(define (transform-process pid w fp)
  (define pt (world-process-table w))
  (match (hash-ref pt pid)
    [#f w]
    [p (struct-copy world w [process-table (hash-set pt pid (fp p))])]))

;; World -> World
;; Updates the World's cached copy of the union of its partial- and downward-gestalts.
(define (update-full-gestalt w)
  (struct-copy world w [full-gestalt
			(gestalt-union (world-partial-gestalt w) (world-downward-gestalt w))]))

;; World Gestalt (Option PID) -> World
;; Constructs and enqueues a PendingEvent describing a change to the
;; World's gestalt falling within the relevant-gestalt *subset* of it.
(define (issue-local-routing-update w relevant-gestalt known-target)
  (enqueue-pending-event (pending-routing-update (world-full-gestalt w)
						 relevant-gestalt
						 known-target)
			 w))

;; World Gestalt (Option PID) -> Transition
;; Communicates a change in World's gestalt falling within the
;; relevant-gestalt *subset* of it both to local Processes and to the
;; World's own containing World.
(define (issue-routing-update w relevant-gestalt known-target)
  (transition (issue-local-routing-update w relevant-gestalt known-target)
              (routing-update (drop-gestalt (world-partial-gestalt w)))))

;; World Gestalt Gestalt (Option PID) -> Transition
;; Communicates a change in the World gestalt corresponding to a
;; change in a single Process's gestalt. The old-gestalt is what the
;; Process used to be interested in; new-gestalt is its new interests.
(define (apply-and-issue-routing-update w old-gestalt new-gestalt known-target)
  (define new-partial
    (gestalt-union (gestalt-erase-path (world-partial-gestalt w) old-gestalt) new-gestalt))
  (issue-routing-update (update-full-gestalt (struct-copy world w [partial-gestalt new-partial]))
			(gestalt-union old-gestalt new-gestalt)
			known-target))

;; PID Action -> World -> Transition
;; Interprets a single Action performed by PID, updating World state
;; and possibly causing the World to take externally-visible Actions
;; as a result.
(define ((perform-action pid a) w)
  (match a
    [(? process? new-p)
     (let* ((new-pid (world-next-pid w))
	    (new-p (trigger-guard-process new-p))
	    (new-gestalt (label-gestalt (process-gestalt new-p) new-pid))
	    (new-p (struct-copy process new-p [gestalt new-gestalt]))
	    (w (struct-copy world w
		 [next-pid (+ new-pid 1)]
		 [process-table (hash-set (world-process-table w) new-pid new-p)])))
       ;; (log-info "Spawned process ~a ~v ~v"
       ;; 		 (cons new-pid (pid-stack))
       ;; 		 (process-behavior new-p)
       ;; 		 (process-state new-p))
       (apply-and-issue-routing-update w (gestalt-empty) new-gestalt new-pid))]
    [(quit)
     (define pt (world-process-table w))
     (define p (hash-ref pt pid (lambda () #f)))
     (if p
	 (let* ((w (struct-copy world w [process-table (hash-remove pt pid)])))
	   ;; (log-info "Process ~a terminating; ~a processes remain"
	   ;; 	     (cons pid (pid-stack))
	   ;; 	     (hash-count (world-process-table w)))
	   (apply-and-issue-routing-update w (process-gestalt p) (gestalt-empty) #f))
	 (transition w '()))]
    [(routing-update gestalt)
     (define pt (world-process-table w))
     (define p (hash-ref pt pid (lambda () #f)))
     (if p
	 (let* ((old-gestalt (process-gestalt p))
		(new-gestalt (label-gestalt gestalt pid))
		(new-p (struct-copy process p [gestalt new-gestalt]))
		(w (struct-copy world w [process-table (hash-set pt pid new-p)])))
	   (apply-and-issue-routing-update w old-gestalt new-gestalt #f))
	 (transition w '()))]
    [(message body meta-level feedback?)
     (if (zero? meta-level)
	 (transition (enqueue-pending-event a w) '())
	 (transition w (message body (- meta-level 1) feedback?)))]))

;; PendingEvent World -> World
;; Interprets a PendingEvent, delivering the resulting Event(s) to Processes.
(define (dispatch-pending-event e w)
  (match e
    [(message body meta-level feedback?)
     (define pids (gestalt-match-value (world-partial-gestalt w) body meta-level feedback?))
     (define pt (world-process-table w))
     (for/fold ([w w]) [(pid (in-set pids))]
       (apply-transition pid (deliver-event e pid (hash-ref pt pid)) w))]
    [(pending-routing-update g affected-subgestalt known-target)
     (define affected-pids (gestalt-match affected-subgestalt g))
     (define pt (world-process-table w))
     (for/fold ([w w])
	 [(pid (in-set (if known-target (set-add affected-pids known-target) affected-pids)))]
       (match (hash-ref pt pid (lambda () #f))
	 [#f w]
	 [p (define g1 (gestalt-filter g (process-gestalt p)))
	    (apply-transition pid (deliver-event (routing-update g1) pid p) w)]))]))

;; World -> Transition
;; Polls the non-provably-inert processes identified by the
;; runnable-pids set (by sending them #f instead of an Event).
;;
;; N.B.: We also effectively compute whether this entire World is
;; inert here.
;;
;; This is roughly the "schedule" rule of the Network Calculus.
(define (step-children w)
  (define runnable-pids (world-runnable-pids w))
  (if (set-empty? runnable-pids)
      #f ;; world is inert.
      (transition (for/fold ([w (struct-copy world w [runnable-pids (set)])])
		      [(pid (in-set runnable-pids))]
		    (define p (hash-ref (world-process-table w) pid (lambda () #f)))
		    (if (not p) w (apply-transition pid (deliver-event #f pid p) w)))
		  '()))) ;; world needs another check to see if more can happen.

;; Behavior :> (Option Event) World -> Transition
;; World's behavior function. Lifts and dispatches an incoming event
;; to contained Processes.
(define (world-handle-event e w)
  (if (or e (not (inert? w)))
      (sequence-transitions (transition (inject-event e w) '())
			    dispatch-pending-events
			    perform-actions
			    (lambda (w) (or (step-children w) (transition w '()))))
      (step-children w)))

;; Event World -> World
;; Translates an event from the World's container into PendingEvents
;; suitable for its own contained Processes.
(define (inject-event e w)
  (match e
    [#f w]
    [(routing-update g)
     (define old-downward (world-downward-gestalt w))
     (define new-downward (lift-gestalt (label-gestalt g 'out)))
     (issue-local-routing-update (update-full-gestalt
				  (struct-copy world w [downward-gestalt new-downward]))
				 (gestalt-union old-downward new-downward)
				 #f)]
    [(message body meta-level feedback?)
     (enqueue-pending-event (message body (+ meta-level 1) feedback?) w)]))

;; Symbol
;; Describes the routing implementation, for use in profiling, debugging etc.
(define routing-implementation 'fastrouting)