TCP in a pub/sub style

presence/NOTES-tcp.txt

@@ -0,0 +1,11 @@
+15:01 < tonyg> samth: even tcp has this automatic-resource-allocation thing 
+               that we described yesterday for rudybot-big-bang! to open a 
+               connection, advertise (you your-port target-ip target-port) and 
+               subscribe (target-ip target-port you your-port)
+15:01 < tonyg> and to listen, advertise (you your-port * *) and subscribe (* * 
+               you your-port)
+15:02 < tonyg> note that where the wildcards are permitted varies between the 
+               languages used for ad patterns and sub patterns!
+15:02 < tonyg> (uh, and the driver uses presence/directory to auto-open and 
+               -close the underlying sockets)
+15:35 < tonyg> Oh UDP has the same pattern (though less restrictive)

presence/conversation-socket.rkt

@@ -5,139 +5,162 @@
 (require racket/match)
 (require racket/tcp)
 (require "conversation.rkt")
+(require "standard-thread.rkt")
 
-(provide (struct-out set-options)
+(provide (struct-out tcp-option)
-	 tcp-server-actor
+	 (struct-out tcp-credit)
-	 tcp-client-actor)
+	 (struct-out tcp-data)
+	 (struct-out tcp-eof)
+	 (struct-out tcp-address)
+	 (struct-out tcp-stream)
+	 flip-stream
+	 wild-address
+	 wild-stream
+	 tcp-driver)
 
-(struct set-options (new-values) #:prefab)
+(struct tcp-option (name value) #:prefab)
+(struct tcp-credit (amount) #:prefab)
+(struct tcp-data (inbound? chunk) #:prefab)
+(struct tcp-eof () #:prefab)
 
-(define (socket-name role s)
+;; The address of a TCP endpoint.
-  (define-values (local-ip local-port remote-ip remote-port)
+;;
-    (tcp-addresses s #t))
+;; The field host may be set to #f, in which case it means "some (or
-  (list role local-ip local-port remote-ip remote-port))
+;; all) local interface(s)". If port is a number, it's treated as a
+;; TCP port directly; otherwise, it's treated as a local handle for an
+;; automatically-allocated port.
+(struct tcp-address (host port) #:prefab)
 
-(define (option-value options key [missing-value #f])
+;; Identifies a (unidirectional) TCP stream. A single TCP connection
-  (cond
+;; has two of these, one in each direction.
-   ((assq key options) => cadr)
+(struct tcp-stream (source sink) #:prefab)
-   (else missing-value)))
 
-(define (tcp-server-actor room options . tcp-listener-args)
+(define (socket-name s)
-  (define listener (apply tcp-listen tcp-listener-args))
+  (call-with-values (lambda () (tcp-addresses s #t)) list))
-  (define name (socket-name 'listener listener))
-  (thread (lambda ()
-	    (define handle (join-room room name))
-	    (log-info (format "Listening on ~v" name))
-	    (let loop ((owner #f)
-		       (remaining-credit (option-value options 'initial-accept-credit 0)))
-	      (sync (handle-evt (send handle disconnected-evt)
-				(lambda (reason)
-				  (log-error (format "~v: conversation closed: ~v" name reason))
-				  (tcp-close listener)))
-		    (handle-evt (send handle listen-evt)
-				(match-lambda
-				 ((arrived who)
-				  (log-info (format "~v: New owner: ~v" name who))
-				  (loop who remaining-credit))
-				 ((departed who why)
-				  (if (equal? owner who)
-				      (begin (log-info (format "~v: Owner departed, closing" name))
-					     (tcp-close listener))
-				      (loop owner remaining-credit)))
-				 ((says _ (credit _ amount) _)
-				  (define new-credit (+ remaining-credit amount))
-				  (log-info (format "~v: Credit now ~v" name new-credit))
-				  (loop owner new-credit))
-				 (unexpected
-				  (log-warning (format "~v: Ignoring message: ~v" name unexpected))
-				  (loop owner remaining-credit))))
-		    (if (positive? remaining-credit)
-			(handle-evt (tcp-accept-evt listener)
-				    (match-lambda
-				     ((list i o)
-				      (send handle say
-					    (tcp-socket-actor 'inbound-connection options i o)
-					    'accepted)
-				      (loop owner (- remaining-credit 1)))))
-			never-evt)))))
-  room)
 
-(define (tcp-client-actor room options . tcp-connect-args)
+(define (flip-stream s)
-  (define-values (i o) (apply tcp-connect tcp-connect-args))
+  (tcp-stream (tcp-stream-sink s) (tcp-stream-source s)))
-  (tcp-socket-actor 'outbound-connection options i o))
 
-(define (tcp-socket-actor role options i o [room (make-room)])
+(define (wild-address) (tcp-address (wild) (wild)))
-  (define name (socket-name role i))
+(define (wild-stream) (tcp-stream (wild-address) (wild-address)))
-  (define (close-ports)
+
-    (close-input-port i)
+(define (tcp-driver room)
-    (close-output-port o))
+  (standard-thread
-  (define (compute-terminator options)
+   (lambda ()
-    ;; See read-line-evt and friends.
+     (define handle (join-room room 'DRIVER))
-    (option-value options 'line-terminator 'any))
+     (send handle assert! (topic-publisher (wild-stream) #:virtual? #t))
-  (define (compute-read-evt options)
+     (let loop ()
-    (define read-mode (option-value options 'read-mode 'bytes))
+       (match (send handle listen)
-    (case read-mode
+	 [(arrived (topic 'subscriber
-      ((bytes) (values (lambda (credit) (read-bytes-evt credit i))
+			  (tcp-stream (tcp-address (? wild?) (? wild?))
-		       bytes-length))
+				      (and local-address (tcp-address #f local-handle)))
-      ((lines) (values (lambda (credit) (read-line-evt i (compute-terminator options)))
+			  _))
-		       (lambda (v) 1)))
+	  ;;(write `(listening on ,local-handle)) (newline)
-      ((bytes-lines) (values (lambda (credit) (read-bytes-line-evt i (compute-terminator options)))
+	  (define listener (tcp-listen (if (number? local-handle) local-handle 0) 4 #t))
-			     (lambda (v) 1)))
+	  (standard-thread (lambda () (tcp-listen-driver room local-address listener)))
-      (else (error 'tcp-socket-actor "Illegal read-evt mode ~v" read-mode))))
+	  (loop)]
-  (thread (lambda ()
+	 [(arrived (topic 'subscriber
-	    (define handle (join-room room name))
+			  (and name (tcp-stream (tcp-address remote-host remote-port)
-	    (log-info (format "~v: New connection" name))
+						(tcp-address #f local-handle)))
-	    (with-handlers
+			  _))
-		((exn? (lambda (e)
+	  (when (not (number? local-handle))
-			 (close-ports)
+	    ;;(write `(connecting to ,remote-host ,remote-port with handle ,local-handle)) (newline)
-			 (raise e))))
+	    (define-values (i o)
-	      (let loop ((options options)
+	      (tcp-connect remote-host remote-port
-			 (peer-count 0)
+			   #f (if (number? local-handle) local-handle #f)))
-			 (remaining-credit (option-value options 'initial-read-credit 0)))
+	    (standard-thread (lambda () (tcp-read-driver room name i)))
-		;;(write `(connection-loop ,options ,peer-count ,remaining-credit)) (newline)
+	    (standard-thread (lambda () (tcp-write-driver room (flip-stream name) o))))
-		(sync (handle-evt (send handle disconnected-evt)
+	  (loop)]
-				  (lambda (reason)
+	 [_
-				    (log-error (format "~v: conversation closed: ~v" name reason))))
+	  (loop)])))))
-		      (handle-evt (send handle listen-evt)
+
-				  (match-lambda
+(define (tcp-read-driver room name i)
-				   ((arrived _)
+  (define handle (join-room room 'READER))
-				    (loop options (+ peer-count 1) remaining-credit))
+  (send handle assert! (topic-publisher name))
-				   ((departed _ _)
+  (let loop ((credit 0) (mode 'line))
-				    (if (= peer-count 1)
+    (sync (wrap-evt (send handle listen-evt)
-					(log-info (format "~v: Last peer departed" name))
+		    (lambda (m)
-					(loop options (- peer-count 1) remaining-credit)))
+		      (match m
-				   ((says _ (credit _ amount) _)
+			[(departed (topic 'subscriber (== name) _) _)
-				    (loop options peer-count (+ remaining-credit amount)))
+			 ;;(write `(closing reader ,name)) (newline)
-				   ((says _ (? eof-object?) _)
+			 (tcp-abandon-port i)]
-				    (close-output-port o)
+			[(says (topic 'subscriber (== name) _) (tcp-option 'mode new-mode))
-				    (loop options peer-count remaining-credit))
+			 (loop credit new-mode)]
-				   ((says _ (? bytes? bs) _)
+			[(says (topic 'subscriber (== name) _) (tcp-credit amount))
-				    ;; TODO: credit flow the other way?
+			 (loop (+ credit amount) mode)]
-				    (write-bytes bs o)
+			[_
-				    (flush-output o)
+			 (loop credit mode)])))
-				    (loop options peer-count remaining-credit))
+	  (if (positive? credit)
-				   ((says _ (? string? s) _)
+	      (match mode
-				    ;; TODO: credit flow the other way?
+		['line (handle-evt (read-line-evt i 'any)
-				    (write-string s o)
+				   (lambda (v)
-				    (flush-output o)
+				     (cond
-				    (loop options peer-count remaining-credit))
+				      [(eof-object? v)
-				   ((says _ (set-options new-values) _)
+				       (send handle say (topic-publisher name) (tcp-eof))
-				    (loop new-values peer-count remaining-credit))
+				       (loop 0 mode)]
-				   (unexpected
+				      [else
-				    (log-warning (format "~v: Ignoring message: ~v"
+				       ;;(write `(relaying line ,v from ,name)) (newline)
-							 name unexpected))
+				       (send handle say (topic-publisher name) (tcp-data #t v))
-				    (loop options peer-count remaining-credit))))
+				       (loop (- credit 1) mode)])))]
-		      (if (positive? remaining-credit)
+		['bytes (handle-evt (read-bytes-evt credit i)
-			  (let-values (((e-maker credit-adjuster) (compute-read-evt options)))
+				    (lambda (v)
-			    (handle-evt (e-maker remaining-credit)
+				      (cond
-					(lambda (v)
+				       [(eof-object? v)
-					  (if (eof-object? v)
+					(send handle say (topic-publisher name) (tcp-eof))
-					      (begin (send handle say v 'eof)
+					(loop 0 mode)]
-						     (loop options peer-count 0))
+				       [else
-					      (begin (send handle say v 'data)
+					;;(write `(relaying bytes ,v from ,name)) (newline)
-						     (loop options peer-count
+					(send handle say (topic-publisher name) (tcp-data #t v))
-							   (- remaining-credit
+					(loop (- credit (bytes-length v)) mode)])))])
-							      (credit-adjuster v))))))))
+	      never-evt))))
-			  never-evt)))
+
-	      (close-ports))))
+(define (tcp-write-driver room name o)
-  room)
+  (define handle (join-room room 'WRITER))
+  (send handle assert! (topic-subscriber name))
+  (let loop ((mode 'string))
+    (sync (wrap-evt (send handle listen-evt)
+		    (lambda (m)
+		      (match m
+			[(departed (topic 'publisher (== name) _) _)
+			 ;;(write `(closing writer ,name)) (newline)
+			 (tcp-abandon-port o)]
+			[(says (topic 'publisher (== name) _) (tcp-option 'mode new-mode))
+			 (loop new-mode)]
+			[(says (topic 'publisher (== name) _) (tcp-data #f v))
+			 ;;(write `(writing ,v to ,name)) (newline)
+			 (define credit-amount
+			   (match mode
+			     ['string (write-string v o)
+				      1]
+			     ['bytes (write-bytes v o)]))
+			 (flush-output o)
+			 (send handle say (topic-subscriber name) (tcp-credit credit-amount))
+			 (loop mode)]
+			[_
+			 (loop mode)]))))))
+
+(define (tcp-listen-driver room local-address listener)
+  (define handle (join-room room 'LISTENER))
+  (define name (tcp-stream (wild-address) local-address))
+  (send handle assert! (topic-publisher name #:virtual? #t))
+  (let loop ()
+    (sync (wrap-evt (send handle listen-evt)
+		    (lambda (m)
+		      (match m
+			[(departed (topic 'subscriber (== name) _) _)
+			 ;;(write `(closing listener ,name)) (newline)
+			 (tcp-close listener)]
+			[_
+			 ;;(write `(listener heard ,m)) (newline)
+			 (loop)])))
+	  (wrap-evt (tcp-accept-evt listener)
+		    (lambda (v)
+		      (match-define (list i o) v)
+		      (match-define (list _ _ remote-host remote-port) (socket-name i))
+		      (define new-stream
+			(tcp-stream (tcp-address remote-host remote-port) local-address))
+		      (standard-thread
+		       (lambda () (tcp-read-driver room new-stream i)))
+		      (standard-thread
+		       (lambda () (tcp-write-driver room (flip-stream new-stream) o)))
+		      (loop))))))

presence/test-conversation-socket.rkt

@@ -5,69 +5,91 @@
 
 (require "conversation.rkt")
 (require "conversation-socket.rkt")
+(require "standard-thread.rkt")
 
 (define pool (make-room 'everybody))
 
-(define (handle-connection sock quit-proc)
+(struct groupchat (utterance) #:prefab)
-  (join-room pool)
+
-  (define h (join-room sock))
+(define (session inbound-stream quit-proc)
-  (match (send h listen)
+  (define h (join-room pool 'SESSION))
-    ((arrived peer-name)
+  (define outbound-stream (flip-stream inbound-stream))
-     (let loop ()
+  (send h assert! (topic-publisher outbound-stream))
-       (send h say "Ready>> ")
+  (send h assert! (topic-subscriber inbound-stream))
-       (sync (handle-evt (send h listen-evt)
+  (send h assert! (topic-publisher groupchat))
-			 (match-lambda
+  (send h assert! (topic-subscriber groupchat))
-			  ((says _ _ 'eof)
+  (define (send-text s) (send h say (topic-publisher outbound-stream) (tcp-data #f s)))
-			   (send h say "OK, bye\n"))
+  (define (issue-credit) (send h say (topic-subscriber inbound-stream) (tcp-credit 1)))
-			  ((says _ "quit" 'data)
+  (issue-credit)
-			   (send h say (credit peer-name 1))
+  (let loop ((prompt? #t))
-			   (quit-proc)
+    (when prompt? (send-text "Ready>> "))
-			   (send h say "OK, will quit accepting\n")
+    (match (send h listen)
-			   (loop))
+      [(says (topic 'publisher (== inbound-stream) _) (tcp-eof))
-			  ((says _ what 'data)
+       (send-text "OK, bye\n")]
-			   (write what)
+      [(says (topic 'publisher (== inbound-stream) _) (tcp-data #t "quit"))
-			   (newline)
+       (issue-credit)
-			   (send h say (credit #f 1))
+       (quit-proc)
-			   (send h say "Carry on\n")
+       (send-text "OK, will quit accepting\n")
-			   (loop))
+       (loop #t)]
-			  ((departed _ _) (void))
+      [(says (topic 'publisher (== inbound-stream) _) (tcp-data #t what))
-			  (else (loop))))
+       (write `(someone said ,what))
-	     (handle-evt (send h disconnected-evt) void))))))
+       (newline)
+       (issue-credit)
+       (send-text "Carry on\n")
+       (send h say (topic-publisher groupchat) (groupchat what))
+       (loop #t)]
+      [(says (topic 'publisher (== groupchat) _) (groupchat what))
+       (send-text (string-append "CHAT: " what "\n"))
+       (loop #t)]
+      [(departed _ _)
+       (void)]
+      [_
+       (loop #f)])))
 
 (define (listen port-no)
-  (define r (make-room))
+  (standard-thread
-  (tcp-server-actor r
+   (lambda ()
-		    `((initial-accept-credit 1)
+     (define h (join-room pool 'LISTEN-THREAD))
-		      (read-mode lines)
+     (define server-address (tcp-address #f port-no))
-		      (initial-read-credit 1))
+     (send h assert! (topic-subscriber (tcp-stream (wild-address) server-address)))
-		    port-no)
+     (define (quit-proc) (send h depart))
-  (define h (join-room r 'main))
-  (match (send h listen)
-    ((arrived listener-name)
      (let loop ()
        (match (send h listen)
-	 ((says _ sock 'accepted)
+	 [(arrived
-	  (thread (lambda ()
+	   (topic 'publisher
-		    (handle-connection sock
+		  (and inbound-stream (tcp-stream (tcp-address (? non-wild?) (? non-wild?))
-				       (lambda ()
+						  (== server-address)))
-					 (send h depart 'told-to-quit)))))
+		  _))
-	  (send h say (credit listener-name 1)))
+	  (write `(starting session for ,inbound-stream)) (newline)
-	 (unexpected
+	  (standard-thread (lambda () (session inbound-stream quit-proc)))
-	  (write `(unexpected ,unexpected))
+	  (loop)]
-	  (newline)))
+	 [_
-       (loop)))))
+	  (loop)])))))
 
-(thread (lambda ()
+;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;
-	  (join-room pool)
+
-	  (listen 5001)))
+(tcp-driver pool)
+
+(define port-number 5001)
+(display "Listening on port ")
+(display port-number)
+(newline)
+
+(listen port-number)
 
 (define (wait-until-pool-empty)
-  (define h (join-room pool))
+  (define h (join-room pool 'WAITER))
-  (let loop ((count 0))
+  (send h assert! (topic-publisher (wild) #:virtual? #t))
+  (send h assert! (topic-subscriber (wild) #:virtual? #t))
+  (let loop ((show-count #t) (count 0))
+    (when show-count
+      (write `(pool has ,count members)) (newline))
     (match (send h listen)
-      ((arrived _) (loop (+ count 1)))
+      [(arrived x)
-      ((departed _ _) (if (= count 1)
+       (write `(,x arrived in pool)) (newline)
+       (loop #t (+ count 1))]
+      [(departed _ _) (if (= count 1)
 			  'done
-			  (loop (- count 1))))
+			  (loop #t (- count 1)))]
-      (_ (loop count)))))
+      [_ (loop #f count)])))
 (wait-until-pool-empty)