Get the broker working.

Add support for encoding prefab structs as JSON objects. Remove linkage & at-meta from patches inside broker dataspace. This is ugly, and deserves to be revisited in future. Fix a bug where using trie-prune-branch was incorrect, and trie-subtract should have been used instead. Factor out support/struct.rkt.
2016-05-10 00:25:50 -04:00 · 2016-05-10 00:25:50 -04:00 · bbca582b98
parent abc844c964
commit bbca582b98
7 changed files with 132 additions and 53 deletions
--- a/racket/syndicate/actor.rkt
+++ b/racket/syndicate/actor.rkt
@ -20,6 +20,8 @@

         actor-body->spawn-action

+         patch-without-linkage
+
         ;;----------------------------------------
         (struct-out actor-state)
         pretty-print-actor-state
@ -156,6 +158,19 @@
 ;; Projection for observing LinkActive.
 (define link-active-projection (link-active ? (?!)))

+;; Assertions for patch-without-linkage to remove. TODO: this is gross.
+(define linkage-assertions
+  (trie-union-all #:combiner (lambda (v1 v2) (trie-success #t))
+                  (list (pattern->trie #t (link-active ? ?))
+                        (pattern->trie #t (observe (link-active ? ?)))
+                        (pattern->trie #t (link-result ? ? ?))
+                        (pattern->trie #t (observe (link-result ? ? ?))))))
+
+;; Patch -> Patch
+;; Remove linkage-related assertions.
+(define (patch-without-linkage p)
+  (patch-pruned-by p linkage-assertions))
+
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Producing Instruction side-effects

--- a/racket/syndicate/broker/protocol.rkt
+++ b/racket/syndicate/broker/protocol.rkt
@ -12,25 +12,55 @@
 (require racket/match)
 (require "../main.rkt")
 (require "../tset.rkt")
+(require "../support/struct.rkt")

 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
 ;; Wire protocol representation of events and actions

+(define (revive-prefabs j)
+  (match j
+    [(? list?) (map revive-prefabs j)]
+    [(? hash?)
+     (define type (hash-ref j '@type #f))
+     (define fields (hash-ref j 'fields #f))
+     (if (and type fields (= (hash-count j) 2))
+         (apply (struct-type-make-constructor
+                 (prefab-key->struct-type (string->symbol type) (length fields)))
+                (map revive-prefabs fields))
+         (for/hasheq [((k v) (in-hash j))] (values k (revive-prefabs v))))]
+    [_ j]))
+
+(define (pickle-prefabs j)
+  (match j
+    [(? list?) (map pickle-prefabs j)]
+    [(? hash?) (for/hasheq [((k v) (in-hash j))] (values k (pickle-prefabs v)))]
+    [(? non-object-struct?)
+     (hasheq '@type (symbol->string (struct-type-name (struct->struct-type j)))
+             'fields (map pickle-prefabs (cdr (vector->list (struct->vector j)))))]
+    [_ j]))
+
 (define only-peer (datum-tset 'peer))

 (define (drop j)
-  (match j
+  (match (revive-prefabs j)
    ["ping" 'ping]
    ["pong" 'pong]
-    [`("patch" ,pj) (jsexpr->patch pj (lambda (v) only-peer))]
+    [`("patch" ,pj) (jsexpr->patch pj
+                                   (lambda (v) only-peer)
+                                   (lambda (arity t)
+                                     (prefab-key->struct-type (string->symbol t) arity)))]
    [`("message" ,body) (message body)]))

 (define (lift j)
-  (match j
-    ['ping "ping"]
-    ['pong "pong"]
-    [(? patch? p) `("patch" ,(patch->jsexpr p (lambda (v) #t)))]
-    [(message body) `("message" ,body)]))
+  (pickle-prefabs
+   (match j
+     ['ping "ping"]
+     ['pong "pong"]
+     [(? patch? p) `("patch" ,(patch->jsexpr p (lambda (v) #t)))]
+     [(message body) `("message" ,body)])))
+
+(require racket/trace)
+(trace drop lift)

 (define drop-json-action drop)
 (define lift-json-event lift)
--- a/racket/syndicate/broker/server.rkt
+++ b/racket/syndicate/broker/server.rkt
@ -8,6 +8,7 @@
 (require net/rfc6455)
 (require (except-in "../main.rkt" dataspace assert))
 (require "../actor.rkt")
+(require "../trie.rkt")
 (require "../demand-matcher.rkt")
 (require "../drivers/timer.rkt")
 (require "../drivers/websocket.rkt")
@ -34,7 +35,7 @@

         (arm-ping-timer!)

-         (until (retracted (advertise (websocket-message c server-id _)))
+         (until (retracted (advertise (websocket-message c server-id _)) #:meta-level 1)
           (assert (advertise (websocket-message server-id c _)) #:meta-level 1)

           (on (message (timer-expired c _) #:meta-level 1)
@ -43,15 +44,28 @@

           (on (message (websocket-message c server-id $data) #:meta-level 1)
               (match (drop-json-action (string->jsexpr data))
+                 ['ping (send-event 'pong)]
+                 ['pong (void)]
                 [(? patch? p) (patch! (patch-without-at-meta p))]
                 [(message (at-meta _)) (void)]
                 [(message body) (send! body)]))

           (on-event
-            [(? patch? p) (send-event (patch-without-at-meta p))]
+            [(? patch? p) (send-event (clean-patch p))]
            [(message (at-meta _)) #f]
            [(? message? m) (send-event m)]))))

+(define stuff-to-prune
+  (trie-union-all #:combiner (lambda (v1 v2) (trie-success #t))
+                  (list (pattern->trie #t (at-meta ?))
+                        (pattern->trie #t (observe (at-meta ?))))))
+
+(define (clean-patch p)
+  ;; TODO: this is gross. Linkage shouldn't be visible, and there
+  ;; should be some clean way of getting rid of observe(atMeta(...))
+  ;; and so on.
+  (patch-without-linkage (patch-pruned-by p stuff-to-prune)))
+
 ;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

 (module+ main
--- a/racket/syndicate/drivers/websocket.rkt
+++ b/racket/syndicate/drivers/websocket.rkt
@ -158,10 +158,10 @@
    (match e
      [(message (at-meta (websocket-incoming-message _ (? eof-object?))))
       (shutdown-connection! state)]
-      [(message (at-meta (websocket-incoming-message _ (? bytes? bs))))
+      [(message (at-meta (websocket-incoming-message _ bytes-or-string)))
       (transition state (message (websocket-message (connection-state-remote-addr state)
                                                     (connection-state-local-addr state)
-                                                     bs)))]
+                                                     bytes-or-string)))]
      [(message (websocket-message _ _ m))
       (ws-send! (connection-state-c state) m)
       #f]
--- a/racket/syndicate/patch.rkt
+++ b/racket/syndicate/patch.rkt
@ -20,6 +20,7 @@
         label-patch
         limit-patch
         limit-patch/routing-table
+         patch-pruned-by
         patch-without-at-meta
         only-meta-tset
         compute-aggregate-patch
@ -71,6 +72,8 @@
 (define observe-parenthesis (open-parenthesis 1 struct:observe))
 (define at-meta-parenthesis (open-parenthesis 1 struct:at-meta))

+(define at-meta-everything (pattern->trie #t (at-meta ?)))
+
 (define (patch-empty? p)
  (and (patch? p)
       (trie-empty? (patch-added p))
@ -135,11 +138,15 @@
         (trie-intersect out bound
 			 #:combiner (lambda (v1 v2) (empty-tset-guard (tset-intersect v1 v2))))))

+;; Completely ignores success-values in t.
+(define (patch-pruned-by p t)
+  (match-define (patch added removed) p)
+  (patch (trie-subtract #:combiner (lambda (v1 v2) trie-empty) added t)
+         (trie-subtract #:combiner (lambda (v1 v2) trie-empty) removed t)))
+
 ;; Removes at-meta assertions from the given patch.
 (define (patch-without-at-meta p)
-  (match-define (patch added removed) p)
-  (patch (trie-prune-branch added at-meta-parenthesis)
-         (trie-prune-branch removed at-meta-parenthesis)))
+  (patch-pruned-by p at-meta-everything))

 (define only-meta-tset (datum-tset 'meta))

@ -288,7 +295,7 @@

 (define (jsexpr->patch pj
                       jsexpr->success
-                       [lookup-struct-type (lambda (t) #f)]
+                       [lookup-struct-type (lambda (arity t) #f)]
                       #:deserialize-atom [deserialize-atom values])
  (match-define (list ij oj) pj)
  (patch (jsexpr->trie ij jsexpr->success lookup-struct-type #:deserialize-atom deserialize-atom)
--- a/racket/syndicate/support/struct.rkt
+++ b/racket/syndicate/support/struct.rkt
@ -0,0 +1,28 @@
+#lang racket/base
+;; Misc support routines not (yet) provided by racket itself
+
+(provide non-object-struct?
+         struct-type-name
+         struct->struct-type)
+
+(require (only-in racket/class object?))
+
+;; Any -> Boolean
+;; Racket objects are structures, so we reject them explicitly for
+;; now, leaving them opaque to unification.
+(define (non-object-struct? x)
+  (and (struct? x)
+       (not (object? x))))
+
+;; struct-type -> Symbol
+;; Extract just the name of the given struct-type.
+(define (struct-type-name st)
+  (define-values (name x2 x3 x4 x5 x6 x7 x8) (struct-type-info st))
+  name)
+
+;; Structure -> StructType
+;; Errors when given any struct that isn't completely transparent/prefab.
+(define (struct->struct-type p)
+  (define-values (t skipped?) (struct-info p))
+  (when skipped? (error 'struct->struct-type "Cannot reflect on struct instance ~v" p))
+  t)
--- a/racket/syndicate/trie.rkt
+++ b/racket/syndicate/trie.rkt
@ -50,7 +50,7 @@
         trie-append
         trie-relabel

-         trie-prune-branch
+         ;; trie-prune-branch
         trie-step

         projection->pattern
@ -75,11 +75,11 @@
 (require racket/match)
 (require (only-in racket/list append-map make-list))
 (require (only-in racket/port call-with-output-string with-output-to-string))
-(require (only-in racket/class object?))
 (require "canonicalize.rkt")
 (require "treap.rkt")
 (require "tset.rkt")
 (require "hash-order.rkt")
+(require "support/struct.rkt")

 (module+ test
  (require rackunit)
@ -241,12 +241,6 @@
            [(_ 'vector) '>]
            [(_ _) (hash-order (struct-type-name a-type) (struct-type-name b-type))])]))

-;; struct-type -> Symbol
-;; Extract just the name of the given struct-type.
-(define (struct-type-name st)
-  (define-values (name x2 x3 x4 x5 x6 x7 x8) (struct-type-info st))
-  name)
-
 ;; (Treap OpenParenthesis Trie)
 (define empty-omap (treap-empty open-parenthesis-order))

@ -404,20 +398,6 @@
 (define (pattern->trie v . ps)
  (pattern->trie* v ps))

-;; Structure -> StructType
-;; Errors when given any struct that isn't completely transparent/prefab.
-(define (struct->struct-type p)
-  (define-values (t skipped?) (struct-info p))
-  (when skipped? (error 'struct->struct-type "Cannot reflect on struct instance ~v" p))
-  t)
-
-;; Any -> Boolean
-;; Racket objects are structures, so we reject them explicitly for
-;; now, leaving them opaque to unification.
-(define (non-object-struct? x)
-  (and (struct? x)
-       (not (object? x))))
-
 ;; (A B -> B) B (Vectorof A) -> B
 (define (vector-foldr kons knil v)
  (for/fold [(acc knil)] [(elem (in-vector v (- (vector-length v) 1) -1 -1))]
@ -643,17 +623,22 @@
                     [#f trie-empty]
                     [result (success result)]))))

-;; Trie (U OpenParenthesis Sigma) -> Trie
-;; Outright removes tries reachable from m via edges labelled with key.
-;; Useful for removing (at-meta *) when the success value along that
-;; branch doesn't matter.
-(define (trie-prune-branch m key)
-  (match* (m key)
-    [((branch os w h) (open-parenthesis arity _))
-     (canonicalize (collapse (struct-copy branch m [opens (rupdate arity w os key trie-empty)])))]
-    [((branch os w h) _)
-     (canonicalize (collapse (struct-copy branch m [sigmas (rupdate 0 w h key trie-empty)])))]
-    [(_ _) m]))
+;; DANGEROUS: doesn't adjust any wild edge. So if you give it m=★, it
+;; will give you the wrong answer. Note that trie-step uses
+;; rlookup-open, which deals with the wild edges, so doesn't have this
+;; problem.
+;;
+;; ;; Trie (U OpenParenthesis Sigma) -> Trie
+;; ;; Outright removes tries reachable from m via edges labelled with key.
+;; ;; Useful for removing (at-meta *) when the success value along that
+;; ;; branch doesn't matter.
+;; (define (trie-prune-branch m key)
+;;   (match* (m key)
+;;     [((branch os w h) (open-parenthesis arity _))
+;;      (canonicalize (collapse (struct-copy branch m [opens (rupdate arity w os key trie-empty)])))]
+;;     [((branch os w h) _)
+;;      (canonicalize (collapse (struct-copy branch m [sigmas (rupdate 0 w h key trie-empty)])))]
+;;     [(_ _) m]))

 ;; Trie (U OpenParenthesis Sigma) -> Trie
 (define (trie-step m key)
@ -1027,13 +1012,13 @@
    [(? struct-type?)
     (string-append ":" (symbol->string (struct-type-name type)))]))

-;; String (String -> (Option struct-type)) -> ParenType
-(define (string->paren-type s lookup-struct-type)
+;; Natural String (String -> (Option struct-type)) -> ParenType
+(define (string->paren-type arity s lookup-struct-type)
  (match s
    ["L" 'list]
    ["V" 'vector]
    [_ (if (char=? (string-ref s 0) #\:)
-           (or (lookup-struct-type (substring s 1))
+           (or (lookup-struct-type arity (substring s 1))
               (error 'string->paren-type "Unexpected struct type name ~v" (substring s 1)))
           (error 'string->paren-type "Invalid paren-type string representation ~v" s))]))

@ -1060,7 +1045,7 @@
 ;; Deserializes a matcher from a JSON expression.
 (define (jsexpr->trie j
                      jsexpr->success
-                      [lookup-struct-type (lambda (t) #f)]
+                      [lookup-struct-type (lambda (arity t) #f)]
                      #:deserialize-atom [deserialize-atom values])
  (let walk ((j j))
    (match j
@ -1071,7 +1056,7 @@
        (collapse
         (branch (for/fold [(acc empty-omap)] [(jopen (in-list jopens))]
                   (match-define (list arity type-str vj) jopen)
-                   (define type (string->paren-type type-str lookup-struct-type))
+                   (define type (string->paren-type arity type-str lookup-struct-type))
                   (treap-insert acc (canonical-open-parenthesis arity type) (walk vj)))
                 (walk jwild)
                 (for/fold [(acc empty-smap)] [(jsigma (in-list jsigmas))]