From da3f737079b80f4550fb3997b3ee89b570837348 Mon Sep 17 00:00:00 2001
From: Tony Garnock-Jones <tonygarnockjones@gmail.com>
Date: Sun, 29 Apr 2018 23:16:07 +0100
Subject: [PATCH] imperative-syndicate versions

---
 README.md                                     | 10 +-
 broadcast-latency-imperative-syndicate.rkt    | 92 +++++++++++++++++++
 ...server-imperative-syndicate-tcp-driver.rkt | 64 +++++++++++++
 external-latency.rkt                          |  4 +
 internal-latency-imperative-syndicate.rkt     | 87 ++++++++++++++++++
 observe-all-imperative-syndicate.rkt          | 58 ++++++++++++
 observe-some-imperative-syndicate.rkt         | 82 +++++++++++++++++
 run-all.sh                                    |  6 ++
 sum-all-imperative-syndicate.rkt              | 56 +++++++++++
 9 files changed, 456 insertions(+), 3 deletions(-)
 create mode 100644 broadcast-latency-imperative-syndicate.rkt
 create mode 100644 echo-server-imperative-syndicate-tcp-driver.rkt
 create mode 100644 internal-latency-imperative-syndicate.rkt
 create mode 100644 observe-all-imperative-syndicate.rkt
 create mode 100644 observe-some-imperative-syndicate.rkt
 create mode 100644 sum-all-imperative-syndicate.rkt

diff --git a/README.md b/README.md
index 3ef7bc5..5596d57 100644
--- a/README.md
+++ b/README.md
@@ -10,6 +10,7 @@
     - echoserver.erl is a matching ping server in Erlang
     - echo-server-minimart-tcp-driver.rkt is a server using Minimart's TCP driver
     - echo-server-prospect-tcp-driver.rkt is a server using Prospect's TCP driver
+    - echo-server-imperative-syndicate-tcp-driver.rkt is a server using Imperative Syndicate's TCP driver
     - echo-server.rkt is a server using Minimart, but eschewing the TCP driver
     - plain-racket-server.rkt is a server using built-in Racket threads
     - uvserver.c is a server written in C with libuv
@@ -20,11 +21,14 @@
       symbols instead of fixnums to identify peers.
     - internal-latency-prospect.rkt is the same, but using Prospect
       instead of Minimart.
+    - internal-latency-imperative-syndicate.rkt is the same, but using
+      Imperative Syndicate instead of Minimart.
 
  - observe-all-minimart.rkt, observe-all-prospect.rkt,
-   sum-all-minimart.rkt and sum-all-prospect.rkt measure the costs of
-   presence notification and processing in Minimart and Prospect
-   worlds, respectively.
+   observe-all-imperative-syndicate.rkt, sum-all-minimart.rkt,
+   sum-all-prospect.rkt, and sum-all-imperative-syndicate.rkt measure
+   the costs of presence notification and processing in Minimart,
+   Prospect, and Imperative Syndicate worlds, respectively.
 
  - pingpong.rkt and pingpong.erl are simple measurements of Racket's
    and Erlang's built-in thread communication latency, respectively.
diff --git a/broadcast-latency-imperative-syndicate.rkt b/broadcast-latency-imperative-syndicate.rkt
new file mode 100644
index 0000000..0ee5e76
--- /dev/null
+++ b/broadcast-latency-imperative-syndicate.rkt
@@ -0,0 +1,92 @@
+#lang imperative-syndicate/test-implementation
+;; Measurement of *broadcast* message delivery latency.
+
+(require logbook)
+
+(message-struct ping (src dst timestamp))
+(message-struct pong (timestamp))
+
+(define (send-ping! src dst)
+  (send! (ping src dst (current-inexact-milliseconds))))
+
+(define (run #:echoer-count [echoer-count 100]
+	     #:run-time [run-time 10000])
+  (define total-latency 0)
+  (define total-roundtrips 0)
+  (define boot-start-time (current-inexact-milliseconds))
+  (define run-start-time #f)
+
+  (define (rate-at count)
+    ;; count is the number of roundtrips
+    ;; each roundtrip involves (+ echoer-count 1) messages
+    ;; we want messages per second
+    (/ (* count (+ echoer-count 1)) ;; echoer-count pings and one pong per roundtrip
+       (/ total-latency 1000.0) ;; latency in seconds
+       ))
+
+  (test-case
+      [(for [(id (in-range echoer-count))]
+         (spawn (on (message (ping $src $dst $stamp))
+                    (when (= dst id) (send! (pong stamp))))))
+
+       (let ((src echoer-count)
+             (dst 0))
+         (spawn (on-start
+                 (set! run-start-time (current-inexact-milliseconds))
+                 (send-ping! src dst))
+
+                (on (message (pong $start-time))
+                    (define stop-time (current-inexact-milliseconds))
+                    ;; TODO: is there a way of reducing the measurement error here,
+                    ;; perhaps by recording against run-start-time instead of start-time, somehow?
+                    ;; TODO: first, characterize the measurement error
+                    (define delta (- stop-time start-time))
+                    (set! total-latency (+ total-latency delta))
+                    (set! total-roundtrips (+ total-roundtrips 1))
+                    (when (zero? (modulo total-roundtrips 10000))
+                      (log-info "After ~a roundtrips, ~a milliseconds; ~a Hz"
+                                total-roundtrips
+                                total-latency
+                                (rate-at total-roundtrips)))
+                    (when (< (- stop-time run-start-time) run-time)
+                      (send-ping! src dst)))))])
+  (values total-roundtrips
+          (rate-at total-roundtrips)
+          (- run-start-time boot-start-time)))
+
+(module+ main
+  (define t 2000)
+  (define E (standard-logbook-entry (default-logbook #:verbose? #t) "minimart" "broadcast-latency-imperative-syndicate"))
+  (define T (logbook-table E "broadcast-latency"
+			   #:column-spec '(number-of-echoers
+					   secs/msg
+					   msgs/sec
+					   boot-delay-ms
+					   secs/process-booted
+					   roundtrip-count
+					   run-duration-ms)))
+  ;; Warmup
+  (let ()
+    (run #:echoer-count 1 #:run-time 1000)
+    (run #:echoer-count 10 #:run-time 1000)
+    (void))
+  ;; Real run
+  (for ((n
+         (list* 1 2 5
+                (let loop ((n 10))
+                  (if (>= n 30000)
+                      '()
+                      (cons (inexact->exact (round n))
+                            (loop (* n (sqrt (sqrt 2))))))))
+	 ))
+    (collect-garbage)
+    (collect-garbage)
+    (collect-garbage)
+    (define-values (count v boot-delay-ms) (run #:echoer-count n #:run-time t))
+    (write-logbook-datum! T (list n
+				  (/ 1.0 v)
+				  v
+				  boot-delay-ms
+				  (/ (/ boot-delay-ms 1000.0) n)
+				  count
+				  t))))
diff --git a/echo-server-imperative-syndicate-tcp-driver.rkt b/echo-server-imperative-syndicate-tcp-driver.rkt
new file mode 100644
index 0000000..ec32c41
--- /dev/null
+++ b/echo-server-imperative-syndicate-tcp-driver.rkt
@@ -0,0 +1,64 @@
+#lang imperative-syndicate
+
+(require racket/port)
+(require logbook)
+(require racket/cmdline)
+
+(require/activate imperative-syndicate/drivers/timer)
+(require/activate imperative-syndicate/drivers/tcp)
+
+(begin-for-declarations
+  (define server-entry-name #f)
+  (define server-entry-type #f)
+
+  (command-line #:program "echo-server-prospect-tcp-driver.rkt"
+                #:once-each
+                ["--logbook-entry-name" name
+                 "set logbook entry name to use when recording run statistics"
+                 (set! server-entry-name name)]
+                ["--logbook-entry-type" type
+                 "set logbook entry type to use"
+                 (set! server-entry-type type)])
+
+  (when (not server-entry-type)
+    (error 'echo-server "Please supply the --logbook-entry-type command-line argument."))
+
+  (define L (default-logbook))
+  (define E (logbook-entry L "minimart" server-entry-name server-entry-type))
+  (define Tmem (logbook-table E "server-memory-use" #:column-spec '(time-seconds memory-use)))
+  (define Tconn (logbook-table E "server-connections" #:column-spec '(time-seconds connection-count)))
+
+  (define connection-count 0)
+  (define first-connection-seen? #f)
+
+  (define statistics-poll-interval 2000))
+
+(define (statistician)
+  (spawn #:name 'statistician
+         (field [sample-deadline (+ (current-inexact-milliseconds) statistics-poll-interval)])
+         (on (asserted (later-than (sample-deadline)))
+             ;; (collect-garbage)
+             ;; (collect-garbage)
+             ;; (collect-garbage)
+             (define now (sample-deadline))
+             (write-logbook-datum! Tmem (list (/ now 1000.0) (current-memory-use)))
+             (write-logbook-datum! Tconn (list (/ now 1000.0) connection-count))
+             (when (and first-connection-seen? (zero? connection-count))
+               (exit 0))
+             (sample-deadline (+ (sample-deadline) statistics-poll-interval)))))
+
+(define (listener port-number)
+  (spawn #:name 'listener
+         (during/spawn (tcp-connection $id (tcp-listener port-number))
+           #:name (list 'connection id)
+           (assert (tcp-accepted id))
+           (on-start
+            (set! connection-count (+ connection-count 1))
+            (set! first-connection-seen? #t))
+           (on-stop
+            (set! connection-count (- connection-count 1)))
+           (on (message (tcp-in id $bs))
+               (send! (tcp-out id bs))))))
+
+(listener 5999)
+(statistician)
diff --git a/external-latency.rkt b/external-latency.rkt
index c7b48b7..ea33036 100644
--- a/external-latency.rkt
+++ b/external-latency.rkt
@@ -27,6 +27,9 @@
 	      ["--prospect+tcp"
                "use prospect server with TCP driver"
                (set! server-variation 'prospect+tcp)]
+	      ["--imperative-syndicate+tcp"
+               "use imperative-syndicate server with TCP driver"
+               (set! server-variation 'imperative-syndicate+tcp)]
 	      ["--racket" "use plain racket server" (set! server-variation 'racket)]
 	      ["--other" name "use other server" (set! server-variation name)])
 
@@ -64,6 +67,7 @@
      ['minimart (format-racket-server-command-line "echo-server.rkt")]
      ['minimart+tcp (format-racket-server-command-line "echo-server-minimart-tcp-driver.rkt")]
      ['prospect+tcp (format-racket-server-command-line "echo-server-prospect-tcp-driver.rkt")]
+     ['imperative-syndicate+tcp (format-racket-server-command-line "echo-server-imperative-syndicate-tcp-driver.rkt")]
      ['racket (format-racket-server-command-line "plain-racket-server.rkt")]
      ['erlang
       (define erlang-version-command
diff --git a/internal-latency-imperative-syndicate.rkt b/internal-latency-imperative-syndicate.rkt
new file mode 100644
index 0000000..8b7bf8e
--- /dev/null
+++ b/internal-latency-imperative-syndicate.rkt
@@ -0,0 +1,87 @@
+#lang imperative-syndicate/test-implementation
+;; Measurement of message delivery latency.
+
+(require logbook)
+
+(message-struct ping (src dst timestamp))
+
+(define (send-ping! src dst)
+  (send! (ping src dst (current-inexact-milliseconds))))
+
+(define (run #:echoer-count [echoer-count 100]
+	     #:run-time [run-time 10000])
+  (define total-latency 0)
+  (define total-roundtrips 0)
+  (define boot-start-time (current-inexact-milliseconds))
+  (define run-start-time #f)
+
+  (define (rate-at count)
+    (/ (* count 2) ;; two messages per roundtrip
+       (/ total-latency 1000.0) ;; latency in seconds
+       ))
+
+  (test-case
+      [(for [(id (in-range echoer-count))]
+         (spawn (on (message (ping $src id $stamp))
+                    (send! (ping id src stamp)))))
+
+       (let ((src echoer-count)
+             (dst 0))
+         (spawn (on-start
+                 (set! run-start-time (current-inexact-milliseconds))
+                 (send-ping! src dst))
+
+                (on (message (ping dst src $start-time))
+                    (define stop-time (current-inexact-milliseconds))
+                    ;; TODO: is there a way of reducing the measurement error here,
+                    ;; perhaps by recording against run-start-time instead of start-time, somehow?
+                    ;; TODO: first, characterize the measurement error
+                    (define delta (- stop-time start-time))
+                    (set! total-latency (+ total-latency delta))
+                    (set! total-roundtrips (+ total-roundtrips 1))
+                    (when (zero? (modulo total-roundtrips 10000))
+                      (log-info "After ~a roundtrips, ~a milliseconds; ~a Hz"
+                                total-roundtrips
+                                total-latency
+                                (rate-at total-roundtrips)))
+                    (when (< (- stop-time run-start-time) run-time)
+                      (send-ping! src dst)))))])
+
+  (values total-roundtrips (rate-at total-roundtrips) (- run-start-time boot-start-time)))
+
+(module+ main
+  (define t 2000)
+  (define E (standard-logbook-entry (default-logbook #:verbose? #t) "minimart" "internal-latency-imperative-syndicate"))
+  (define T (logbook-table E "internal-latency"
+			   #:column-spec '(number-of-echoers
+					   secs/msg
+					   msgs/sec
+					   boot-delay-ms
+					   secs/process-booted
+					   roundtrip-count
+					   run-duration-ms)))
+  ;; Warmup
+  (let ()
+    (run #:echoer-count 1 #:run-time 1000)
+    (run #:echoer-count 10 #:run-time 1000)
+    (void))
+  ;; Real run
+  (for ((n
+         (list* 1 2 5
+                (let loop ((n 10))
+                  (if (>= n 30000)
+                      '()
+                      (cons (inexact->exact (round n))
+                            (loop (* n (sqrt (sqrt 2))))))))
+	 ))
+    (collect-garbage)
+    (collect-garbage)
+    (collect-garbage)
+    (define-values (count v boot-delay-ms) (run #:echoer-count n #:run-time t))
+    (write-logbook-datum! T (list n
+				  (/ 1.0 v)
+				  v
+				  boot-delay-ms
+				  (/ (/ boot-delay-ms 1000.0) n)
+				  count
+				  t))))
diff --git a/observe-all-imperative-syndicate.rkt b/observe-all-imperative-syndicate.rkt
new file mode 100644
index 0000000..650e50a
--- /dev/null
+++ b/observe-all-imperative-syndicate.rkt
@@ -0,0 +1,58 @@
+#lang imperative-syndicate/test-implementation
+;; Measurement of presence processing.
+;; Peers observe each other, but do not process the resulting routing events.
+
+(require logbook)
+
+(assertion-struct presence (id))
+
+(define (run #:peer-count [peer-count 100])
+  (define event-count 0)
+
+  (define-values (results cpu-time wall-clock-time gc-time)
+    (time-apply
+     (lambda ()
+       (test-case
+           [(for [(id (in-range peer-count))]
+              (spawn (assert (presence id))
+                     (on (asserted (presence $peer))
+                         (set! event-count (+ event-count 1)))))]))
+     '()))
+
+  (values event-count (max 1 (- cpu-time gc-time))))
+
+(module+ main
+  (define E (standard-logbook-entry (default-logbook #:verbose? #t) "minimart" "observe-all-imperative-syndicate"))
+  (define T (logbook-table E "presence-processing"
+			   #:column-spec '(number-of-peers
+                                           secs/routing-update
+                                           routing-updates/sec
+                                           secs/peer
+                                           peers/sec
+                                           event-count
+                                           run-duration-ms)))
+  ;; Warmup
+  (let ()
+    (run #:peer-count 1)
+    (run #:peer-count 10)
+    (void))
+  ;; Real run
+  (for ((n
+         (list* 1 2 5
+                (let loop ((n 10))
+                  (if (>= n 1000)
+                      '()
+                      (cons (inexact->exact (round n))
+                            (loop (* n (sqrt (sqrt 2))))))))
+	 ))
+    (collect-garbage)
+    (collect-garbage)
+    (collect-garbage)
+    (define-values (event-count run-duration-ms) (run #:peer-count n))
+    (write-logbook-datum! T (list n
+                                  (/ (/ run-duration-ms 1000.0) event-count)
+                                  (/ event-count (/ run-duration-ms 1000.0))
+                                  (/ (/ run-duration-ms 1000.0) n)
+                                  (/ n (/ run-duration-ms 1000.0))
+				  event-count
+                                  run-duration-ms))))
diff --git a/observe-some-imperative-syndicate.rkt b/observe-some-imperative-syndicate.rkt
new file mode 100644
index 0000000..06c948a
--- /dev/null
+++ b/observe-some-imperative-syndicate.rkt
@@ -0,0 +1,82 @@
+#lang imperative-syndicate/test-implementation
+;; Measurement of presence processing.
+;; Peers observe just one distinguished peer, and do not process resulting routing events.
+
+(require logbook)
+
+(provide run)
+
+(assertion-struct client ())
+
+(define (run #:peer-count [peer-count 100] #:client-first? [client-first? #f])
+  (define event-count 0)
+
+  (define-values (results cpu-total-ms wall-ms cpu-gc-ms)
+    (time-apply
+     (lambda ()
+       (test-case
+           [(define (spawn-client)
+              (spawn (assert (client))))
+
+            (define (spawn-server id)
+              (spawn (on (asserted (client))
+                         (set! event-count (+ event-count 1)))))
+
+            (define (spawn-servers)
+              (for [(id (in-range (- peer-count 1)))]
+                (spawn-server id)))
+
+            (if client-first?
+                (begin (spawn-client) (spawn-servers))
+                (begin (spawn-servers) (spawn-client)))]))
+     '()))
+
+  (values event-count (max 1 (- cpu-total-ms cpu-gc-ms))))
+
+(module+ main
+  (define E (standard-logbook-entry (default-logbook #:verbose? #t) "minimart" "observe-some-imperative-syndicate"))
+  (define T-client-first (logbook-table E "presence-processing-narrow-client-first"
+                                        #:column-spec '(number-of-peers
+                                                        secs/routing-update
+                                                        routing-updates/sec
+                                                        secs/peer
+                                                        peers/sec
+                                                        event-count
+                                                        run-duration-ms)))
+  (define T-client-last (logbook-table E "presence-processing-narrow-client-last"
+                                       #:column-spec '(number-of-peers
+                                                       secs/routing-update
+                                                       routing-updates/sec
+                                                       secs/peer
+                                                       peers/sec
+                                                       event-count
+                                                       run-duration-ms)))
+  ;; Warmup
+  (let ()
+    (run #:peer-count 1)
+    (run #:peer-count 10)
+    (void))
+  ;; Real run
+  (for ((n
+         #;(list 5)
+         (list* 2 5
+                (let loop ((n 10))
+                  (if (>= n 100000)
+                      '()
+                      (cons (inexact->exact (round n))
+                            (loop (* n (sqrt (sqrt 2))))))))
+	 ))
+    (collect-garbage)
+    (collect-garbage)
+    (collect-garbage)
+    (for ((client-first? (list #t #f)))
+      (define-values (event-count run-duration-ms)
+        (run #:peer-count n #:client-first? client-first?))
+      (write-logbook-datum! (if client-first? T-client-first T-client-last)
+                            (list n
+                                  (/ (/ run-duration-ms 1000.0) event-count)
+                                  (/ event-count (/ run-duration-ms 1000.0))
+                                  (/ (/ run-duration-ms 1000.0) n)
+                                  (/ n (/ run-duration-ms 1000.0))
+                                  event-count
+                                  run-duration-ms)))))
diff --git a/run-all.sh b/run-all.sh
index 9d234ec..f56cdd9 100755
--- a/run-all.sh
+++ b/run-all.sh
@@ -14,14 +14,20 @@ racket external-latency.rkt --max-waypoint $MAX_WAYPOINT --uv
 ## racket external-latency.rkt --max-waypoint $MAX_MINIMART_WAYPOINT --minimart ## don't bother
 racket external-latency.rkt --max-waypoint $MAX_MINIMART_WAYPOINT --minimart+tcp
 racket external-latency.rkt --max-waypoint $MAX_WAYPOINT --prospect+tcp
+racket external-latency.rkt --max-waypoint $MAX_WAYPOINT --imperative-syndicate+tcp
 racket external-latency.rkt --max-waypoint $MAX_WAYPOINT --racket
 
 racket internal-latency.rkt
 racket internal-latency-prospect.rkt
+racket internal-latency-imperative-syndicate.rkt
+
 racket broadcast-latency-prospect.rkt
+racket broadcast-latency-imperative-syndicate.rkt
 
 racket observe-all-minimart.rkt
 racket observe-all-prospect.rkt
+racket observe-all-imperative-syndicate.rkt
 
 racket observe-some-minimart.rkt
 racket observe-some-prospect.rkt
+racket observe-some-imperative-syndicate.rkt
diff --git a/sum-all-imperative-syndicate.rkt b/sum-all-imperative-syndicate.rkt
new file mode 100644
index 0000000..16768b0
--- /dev/null
+++ b/sum-all-imperative-syndicate.rkt
@@ -0,0 +1,56 @@
+#lang imperative-syndicate/test-implementation
+;; Measurement of presence processing.
+;; Peers observe each other, AND ALSO process the resulting routing events.
+
+(require logbook)
+
+(assertion-struct presence (id))
+
+(define (run #:peer-count [peer-count 100])
+  (define event-count 0)
+  (define start-time (current-inexact-milliseconds))
+  (test-case [(for [(id (in-range peer-count))]
+                (spawn (assert (presence id))
+                       (field [peer-count 0])
+                       (on (asserted (presence $peer))
+                           (set! event-count (+ event-count 1))
+                           (peer-count (+ (peer-count) 1)))))])
+  (define stop-time (current-inexact-milliseconds))
+  (define delta (- stop-time start-time))
+  (values event-count delta))
+
+(module+ main
+  (define E (standard-logbook-entry (default-logbook #:verbose? #t) "minimart" "sum-all-imperative-syndicate"))
+  (define T (logbook-table E "presence-processing"
+			   #:column-spec '(number-of-peers
+                                           secs/event
+                                           events/sec
+                                           secs/peer
+                                           peers/sec
+                                           event-count
+                                           run-duration-ms)))
+  ;; Warmup
+  (let ()
+    (run #:peer-count 1)
+    (run #:peer-count 10)
+    (void))
+  ;; Real run
+  (for ((n
+         (list* 1 2 5
+                (let loop ((n 10))
+                  (if (>= n 1000)
+                      '()
+                      (cons (inexact->exact (round n))
+                            (loop (* n (sqrt (sqrt 2))))))))
+	 ))
+    (collect-garbage)
+    (collect-garbage)
+    (collect-garbage)
+    (define-values (event-count run-duration-ms) (run #:peer-count n))
+    (write-logbook-datum! T (list n
+                                  (/ (/ run-duration-ms 1000.0) event-count)
+                                  (/ event-count (/ run-duration-ms 1000.0))
+                                  (/ (/ run-duration-ms 1000.0) n)
+                                  (/ n (/ run-duration-ms 1000.0))
+				  event-count
+                                  run-duration-ms))))