Initial translation of proxy and network-query to os2

This commit is contained in:
Tony Garnock-Jones 2012-05-03 16:00:43 -04:00
parent f466fd6a05
commit f79bf86111
2 changed files with 243 additions and 237 deletions

View File

@ -5,10 +5,10 @@
(require "api.rkt") (require "api.rkt")
(require "codec.rkt") (require "codec.rkt")
(require "zonedb.rkt") (require "zonedb.rkt")
(require "../racket-matrix/os-big-bang.rkt") (require "../racket-matrix/os2.rkt")
(require "../racket-matrix/os-udp.rkt") (require "../racket-matrix/os2-udp.rkt")
(require "../racket-matrix/os-timer.rkt") (require "../racket-matrix/os2-timer.rkt")
(require "os-dns.rkt") (require "os2-dns.rkt")
(provide network-query (provide network-query
(struct-out network-reply)) (struct-out network-reply))
@ -214,13 +214,13 @@
;; network-query : UdpAddress Question DomainName NEListOf<DomainName> UniqueId -> BootK ;; network-query : UdpAddress Question DomainName NEListOf<DomainName> UniqueId -> BootK
(define (network-query s q zone-origin server-names unique-id) (define (network-query s q zone-origin server-names unique-id)
(os-big-bang/transition (lambda (self-pid)
(try-next-server (network-query-state (network-request s q zone-origin server-names unique-id) (try-next-server (network-query-state (network-request s q zone-origin server-names unique-id)
first-timeout first-timeout
(hash) (hash)
'() '()
#f #f
server-names)))) server-names))))
(define (try-next-server w) (define (try-next-server w)
(match w (match w
@ -248,31 +248,31 @@
[current-name current-name] [current-name current-name]
[remaining-names remaining-names]) [remaining-names remaining-names])
(send-message subq) (send-message subq)
(subscribe/fresh subscription-id (role/fresh subq-id (topic-subscriber (answered-question subq (wild)))
(message-handlers w #:state w
[(answered-question (== subq) ans) [(answered-question (== subq) ans)
(define ips (define ips
(map make-dns-address (set->list (extract-addresses current-name ans)))) (map make-dns-address (set->list (extract-addresses current-name ans))))
(extend-transition (extend-transition
(try-next-server (struct-copy network-query-state w (try-next-server (struct-copy network-query-state w
[known-addresses (hash-set known-addresses [known-addresses (hash-set known-addresses
current-name current-name
ips)] ips)]
[remaining-addresses ips])) [remaining-addresses ips]))
(unsubscribe subscription-id))])))))] (delete-role subq-id))]))))]
[(network-query-state req timeout _ (cons current-ip remaining-ips) _ _) [(network-query-state req timeout _ (cons current-ip remaining-ips) _ _)
(define rpc-id (gensym 'network-query/allocate-query-id)) (define rpc-id (gensym 'network-query/allocate-query-id))
(transition w (transition w
(send-message `(request ,rpc-id allocate-query-id)) (send-message `(request ,rpc-id allocate-query-id))
(subscribe rpc-id (role rpc-id (topic-subscriber `(reply ,rpc-id ,(wild)))
(message-handlers w #:state w
[`(reply ,(== rpc-id) ,id) [`(reply ,(== rpc-id) ,id)
(extend-transition (send-request (struct-copy network-query-state w (extend-transition (send-request (struct-copy network-query-state w
[remaining-addresses remaining-ips]) [remaining-addresses remaining-ips])
id id
timeout timeout
current-ip) current-ip)
(unsubscribe rpc-id))])))])) (delete-role rpc-id))]))]))
(define (on-answer w ans server-ip) (define (on-answer w ans server-ip)
(match ans (match ans
@ -309,30 +309,32 @@
(transition w (transition w
(send-message (dns-request query s server-ip)) (send-message (dns-request query s server-ip))
(send-message (set-timer subscription-id (* timeout 1000) 'relative)) (send-message (set-timer subscription-id (* timeout 1000) 'relative))
(subscribe subscription-id (role subscription-id
(message-handlers w (set (topic-subscriber (timer-expired subscription-id (wild)))
[(timer-expired (== subscription-id) _) (topic-subscriber (dns-reply (wild) (wild) s)))
(log-debug (format "Timed out ~v ~v to ~v ~v after ~v seconds" #:state w
q query-id [(timer-expired (== subscription-id) _)
zone-origin server-ip (log-debug (format "Timed out ~v ~v to ~v ~v after ~v seconds"
timeout)) q query-id
(extend-transition (try-next-server w) zone-origin server-ip
(unsubscribe subscription-id) timeout))
(send-message (list 'release-query-id query-id)))] (extend-transition (try-next-server w)
[(dns-reply reply-message source (== s)) (delete-role subscription-id)
;; TODO: maybe receive only specifically from the queried IP address? (send-message (list 'release-query-id query-id)))]
(log-debug [(dns-reply reply-message source (== s))
(format ;; TODO: maybe receive only specifically from the queried IP address?
"Answer to ~v from ~v ~v in ~v ms~n-- Answers: ~v~n-- Authorities: ~v~n-- Additional: ~v" (log-debug
q zone-origin server-ip (format
(inexact->exact (round (- (current-inexact-milliseconds) start-time))) "Answer to ~v from ~v ~v in ~v ms~n-- Answers: ~v~n-- Authorities: ~v~n-- Additional: ~v"
(dns-message-answers reply-message) q zone-origin server-ip
(dns-message-authorities reply-message) (inexact->exact (round (- (current-inexact-milliseconds) start-time)))
(dns-message-additional reply-message))) (dns-message-answers reply-message)
(if (not (= (dns-message-id reply-message) (dns-message-id query))) (dns-message-authorities reply-message)
w (dns-message-additional reply-message)))
(extend-transition (on-answer w (if (not (= (dns-message-id reply-message) (dns-message-id query)))
(filter-dns-reply q reply-message zone-origin) w
server-ip) (extend-transition (on-answer w
(unsubscribe subscription-id) (filter-dns-reply q reply-message zone-origin)
(send-message (list 'release-query-id query-id))))])))) server-ip)
(delete-role subscription-id)
(send-message (list 'release-query-id query-id))))])))

360
proxy.rkt
View File

@ -11,11 +11,10 @@
(require "zonedb.rkt") (require "zonedb.rkt")
(require "network-query.rkt") (require "network-query.rkt")
(require "resolver.rkt") (require "resolver.rkt")
(require "../racket-matrix/os.rkt") (require "../racket-matrix/os2.rkt")
(require "../racket-matrix/os-big-bang.rkt") (require "../racket-matrix/os2-udp.rkt")
(require "../racket-matrix/os-udp.rkt") (require "../racket-matrix/os2-timer.rkt")
(require "../racket-matrix/os-timer.rkt") (require "os2-dns.rkt")
(require "os-dns.rkt")
(require racket/pretty) (require racket/pretty)
@ -27,77 +26,72 @@
;; start-proxy : UInt16 CompiledZone CompiledZone -> Void ;; start-proxy : UInt16 CompiledZone CompiledZone -> Void
(define (start-proxy port-number zone roots-only) (define (start-proxy port-number zone roots-only)
(define server-addr (udp-listener port-number))
(define client-addr (udp-handle 'dns-client))
(define boot-server (log-info "Ready.")
(os-big-bang 'no-state/boot-server
(spawn dns-spy)
(spawn (timer-relay 'timer-relay:dns))
(spawn (query-id-allocator))
(send-meta-message `(request create-server-socket (udp new ,port-number 512)))
(subscribe/fresh wait-id
(meta-message-handlers w
[`(reply create-server-socket ,s)
(transition w
(unsubscribe wait-id)
(send-meta-message
`(request create-client-socket (udp new 0 512)))
(client-socket-waiter s))]))))
(define (client-socket-waiter s) (ground-vm
(subscribe/fresh wait-id (transition 'no-state
(meta-message-handlers w ;;(spawn udp-spy)
[`(reply create-client-socket ,c) (spawn udp-driver #:debug-name 'udp-driver)
(log-info "Ready.") (spawn (timer-driver 'timer-driver) #:debug-name 'timer-driver)
(transition w (spawn (nested-vm
(unsubscribe wait-id) 'dns-vm
(spawn (dns-read-driver s)) (transition 'no-state
(spawn (dns-write-driver s)) (spawn dns-spy #:debug-name 'dns-spy)
(spawn (dns-read-driver c)) (spawn (timer-relay 'timer-relay:dns) #:debug-name 'timer-relay)
(spawn (dns-write-driver c)) (spawn (query-id-allocator) #:debug-name 'query-id-allocator)
(spawn (packet-dispatcher s)) (spawn (dns-read-driver server-addr) #:debug-name 'server-dns-reader)
(spawn (question-dispatcher zone roots-only c)))]))) (spawn (dns-write-driver server-addr) #:debug-name 'server-dns-writer)
(spawn (dns-read-driver client-addr) #:debug-name 'client-dns-reader)
(ground-vm (os-big-bang (void) (spawn (dns-write-driver client-addr) #:debug-name 'client-dns-writer)
;;(spawn udp-spy) (spawn (packet-dispatcher server-addr) #:debug-name 'packet-dispatcher)
(spawn udp-driver) (spawn (question-dispatcher zone roots-only client-addr)
(spawn (timer-driver 'timer-driver)) #:debug-name 'question-dispatcher)))
(spawn (nested-vm boot-server))))) #:debug-name 'dns-vm))))
(define (query-id-allocator) (define (query-id-allocator)
;; TODO: track how many are allocated and throttle requests if too ;; TODO: track how many are allocated and throttle requests if too
;; many are in flight ;; many are in flight
(os-big-bang (set) ;; SetOf<UInt16>, all active query IDs (transition (set) ;; SetOf<UInt16>, all active query IDs
(subscribe 'query-id-request-handler (role 'query-id-request-handler (topic-subscriber `(request ,(wild) allocate-query-id))
(message-handlers allocated #:state allocated
[`(request ,reply-addr allocate-query-id) [`(request ,reply-addr allocate-query-id)
(let recheck () (let recheck ()
(define n (random 65536)) (define n (random 65536))
(if (set-member? allocated n) (if (set-member? allocated n)
(recheck) (recheck)
(transition (set-add allocated n) (transition (set-add allocated n)
(send-message `(reply ,reply-addr ,n)))))] (send-message `(reply ,reply-addr ,n)))))])
[`(release-query-id ,n) (role 'query-id-release-handler (topic-subscriber `(release-query-id ,(wild)))
(transition (set-remove allocated n))])))) #:state allocated
[`(release-query-id ,n)
(transition (set-remove allocated n))])))
(define (packet-dispatcher s) (define (packet-dispatcher s)
(os-big-bang (set) ;; SetOf<ActiveRequest> (transition (set) ;; SetOf<ActiveRequest>
(subscribe 'packet-dispatcher (role 'error-logger (topic-subscriber (bad-dns-packet (wild) (wild) (wild) (wild)))
(message-handlers old-active-requests #:state old-active-requests
[(? bad-dns-packet? p) [p
(log-error (pretty-format p)) (log-error (pretty-format p))
;; TODO: ^ perhaps use metalevel events? perhaps don't bother though ;; TODO: ^ perhaps use metalevel events? perhaps don't bother though
old-active-requests] old-active-requests])
[(and r (dns-request m source (== s))) ;; We only listen for requests on our server socket (role 'request-booter (topic-subscriber (dns-request (wild) (wild) s))
(define req-id (active-request source (dns-message-id m))) #:state old-active-requests
;; TODO: when we have presence/error-handling, remove req-id [(and r (dns-request m source (== s))) ;; We only listen for requests on our server socket
;; from active requests once request-handler pseudothread exits. (define req-id (active-request source (dns-message-id m)))
(if (set-member? old-active-requests req-id) ;; TODO: when we have presence/error-handling, remove req-id
old-active-requests ;; ignore retransmitted duplicates ;; from active requests once request-handler pseudothread exits.
(transition (set-add old-active-requests req-id) (if (set-member? old-active-requests req-id)
(spawn (packet-relay req-id r))))] old-active-requests ;; ignore retransmitted duplicates
[(and r (dns-reply m (== s) sink)) (transition (set-add old-active-requests req-id)
(define req-id (active-request sink (dns-message-id m))) (spawn (packet-relay req-id r) #:debug-name (list 'packet-relay req-id))))])
(set-remove old-active-requests req-id)])))) (role 'reply-cleanup (topic-subscriber (dns-reply (wild) s (wild)))
#:state old-active-requests
[(and r (dns-reply m (== s) sink))
(define req-id (active-request sink (dns-message-id m)))
(set-remove old-active-requests req-id)])))
(define (packet-relay req-id request) (define (packet-relay req-id request)
(match-define (dns-request request-message request-source request-sink) request) (match-define (dns-request request-message request-source request-sink) request)
@ -126,39 +120,40 @@
(match (dns-message-questions request-message) (match (dns-message-questions request-message)
['() ['()
;; No questions! ;; No questions!
(os-big-bang 'no-state/packet-relay (transition 'no-state/packet-relay
(send-message (answer->reply #f (empty-complete-answer))))] (send-message (answer->reply #f (empty-complete-answer))))]
[(cons original-question _) [(cons original-question _)
;; At least one question ;; At least one question
(log-debug (format "Looking up ~v with query id ~v" (log-debug (format "Looking up ~v with query id ~v"
original-question (dns-message-id request-message))) original-question (dns-message-id request-message)))
(os-big-bang 'no-state/packet-relay (transition 'no-state/packet-relay
(send-message original-question) (send-message original-question)
(subscribe/fresh wait-id (role/fresh wait-id (topic-subscriber (answered-question original-question (wild)))
(message-handlers w #:state w
[(answered-question (== original-question) answer) [(answered-question (== original-question) answer)
(log-debug (format "Final answer to ~v with query id ~v is ~v" (log-debug (format "Final answer to ~v with query id ~v is ~v"
original-question original-question
(dns-message-id request-message) (dns-message-id request-message)
answer)) answer))
(transition w (transition w
(unsubscribe wait-id) (delete-role wait-id)
(send-message (answer->reply original-question answer)))])))])) (send-message (answer->reply original-question answer)))]))]))
(define (glueless-question-handler roots-only-zone q client-sock) (define (glueless-question-handler roots-only-zone q client-sock)
;; Restart q, an overly-glueless question, from the roots. ;; Restart q, an overly-glueless question, from the roots.
(define restarted-question (restart-question q)) (define restarted-question (restart-question q))
(os-big-bang 'no-state (transition 'no-state
(subscribe/fresh relay (role/fresh relay (topic-subscriber (answered-question restarted-question (wild)))
(message-handlers w #:state w
[(answered-question (== restarted-question) ans) [(answered-question (== restarted-question) ans)
;; We got the answer to our restarted question; now transform ;; We got the answer to our restarted question; now transform
;; it into an answer to the original question, to unblock the ;; it into an answer to the original question, to unblock the
;; original questioner. ;; original questioner.
(transition w (transition w
(unsubscribe relay) (delete-role relay)
(send-message (answered-question q ans)))])) (send-message (answered-question q ans)))])
(spawn (question-handler roots-only-zone restarted-question client-sock)))) (spawn (question-handler roots-only-zone restarted-question client-sock)
#:debug-name (list 'glueless-question-handler-inner restarted-question))))
(define (question-dispatcher seed-zone roots-only client-sock) (define (question-dispatcher seed-zone roots-only client-sock)
(define (transition-and-set-timers new-zone timers) (define (transition-and-set-timers new-zone timers)
@ -167,60 +162,67 @@
(match-define (cons name ttl) timerspec) (match-define (cons name ttl) timerspec)
(send-message (set-timer (list 'check-dns-expiry name) (* ttl 1000) 'relative))))) (send-message (set-timer (list 'check-dns-expiry name) (* ttl 1000) 'relative)))))
(define-values (cleaned-seed-zone initial-timers) (zone-expire seed-zone)) (define-values (cleaned-seed-zone initial-timers) (zone-expire seed-zone))
(os-big-bang/transition (extend-transition (transition-and-set-timers cleaned-seed-zone initial-timers)
(extend-transition (transition-and-set-timers cleaned-seed-zone initial-timers) ;; TODO: consider deduping questions here too?
;; TODO: consider deduping questions here too? (role 'debug-dumper (topic-subscriber `(debug-dump))
(subscribe 'question-handler-factory #:state zone
(message-handlers zone [`(debug-dump)
[`(debug-dump) (with-output-to-file "zone-proxy.zone"
(with-output-to-file "zone-proxy.zone" (lambda ()
(lambda () (write-bytes (bit-string->bytes (zone->bit-string zone))))
(write-bytes (bit-string->bytes (zone->bit-string zone)))) #:mode 'binary
#:mode 'binary #:exists 'replace)
#:exists 'replace) (with-output-to-file "zone-proxy.dump"
(with-output-to-file "zone-proxy.dump" (lambda ()
(lambda () (display "----------------------------------------------------------------------\n")
(display "----------------------------------------------------------------------\n") (display (seconds->date (current-seconds)))
(display (seconds->date (current-seconds))) (newline)
(newline) (for* ([(name rrmap) zone] [(rr expiry) rrmap])
(for* ([(name rrmap) zone] [(rr expiry) rrmap]) (write (list rr expiry))
(write (list rr expiry)) (newline))
(newline)) (newline))
(newline)) #:mode 'text
#:mode 'text #:exists 'append)
#:exists 'append) (with-output-to-file "zone-proxy.debug"
(with-output-to-file "zone-proxy.debug" (lambda ()
(lambda () (display "----------------------------------------------------------------------\n")
(display "----------------------------------------------------------------------\n") (display (seconds->date (current-seconds)))
(display (seconds->date (current-seconds))) (newline)
(newline) (pretty-write current-ground-transition))
(pretty-write current-ground-transition)) #:mode 'text
#:mode 'text #:exists 'append)
#:exists 'append) zone])
zone] (role 'question-handler-factory (topic-subscriber (question (wild) (wild) (wild) (wild)))
[(? question? q) #:state zone
(transition zone [(? question? q)
(cond (transition zone
[(question-cyclic? q) (cond
(log-warning (format "Cyclic question ~v" q)) [(question-cyclic? q)
(send-message (answered-question q (empty-complete-answer)))] (log-warning (format "Cyclic question ~v" q))
[(question-too-glueless? q) (send-message (answered-question q (empty-complete-answer)))]
(log-warning (format "Overly-glueless question ~v" q)) [(question-too-glueless? q)
(spawn (glueless-question-handler roots-only q client-sock))] (log-warning (format "Overly-glueless question ~v" q))
[else (spawn (glueless-question-handler roots-only q client-sock)
(spawn (question-handler zone q client-sock))]))] #:debug-name (list 'glueless-question-handler-outer q))]
[(network-reply _ answer) [else
(define-values (new-zone timers) (incorporate-complete-answer answer zone)) (spawn (question-handler zone q client-sock)
(transition-and-set-timers new-zone timers)] #:debug-name (list 'question-handler q))]))])
[(timer-expired (list 'check-dns-expiry name) now-msec) (role 'network-reply-snoop (topic-subscriber (network-reply (wild) (wild)))
(zone-expire-name zone name (/ now-msec 1000.0))]))))) #:state zone
[(network-reply _ answer)
(define-values (new-zone timers) (incorporate-complete-answer answer zone))
(transition-and-set-timers new-zone timers)])
(role 'timer-expiry-handler
(topic-subscriber (timer-expired (list 'check-dns-expiry (wild)) (wild)))
#:state zone
[(timer-expired (list 'check-dns-expiry name) now-msec)
(zone-expire-name zone name (/ now-msec 1000.0))])))
(struct question-state (zone q client-sock nameservers-tried retry-count) #:prefab) (struct question-state (zone q client-sock nameservers-tried retry-count) #:prefab)
(struct expanding-cnames (q accumulator remaining-count) #:prefab) (struct expanding-cnames (q accumulator remaining-count) #:prefab)
(define (question-handler zone q client-sock) (define (question-handler zone q client-sock)
(os-big-bang/transition (retry-question (question-state zone q client-sock (set) 0)))
(retry-question (question-state zone q client-sock (set) 0))))
(define (send-empty-reply w q) (define (send-empty-reply w q)
(transition w (send-message (answered-question q (empty-complete-answer))))) (transition w (send-message (answered-question q (empty-complete-answer)))))
@ -248,33 +250,34 @@
q q
zone-origin zone-origin
(map rr-rdata (set->list nameserver-rrs)) (map rr-rdata (set->list nameserver-rrs))
referral-id)) referral-id)
(subscribe referral-id #:debug-name (list 'network-query q))
(message-handlers w (role referral-id (topic-subscriber (network-reply referral-id (wild)))
[(network-reply (== referral-id) #f) ;; name-error/NXDOMAIN #:state w
(transition w [(network-reply (== referral-id) #f) ;; name-error/NXDOMAIN
(unsubscribe referral-id) (transition w
(send-message (answered-question q #f)))] (delete-role referral-id)
[(network-reply (== referral-id) ans) (send-message (answered-question q #f)))]
(define-values (new-zone ignored-timers) (incorporate-complete-answer ans zone)) [(network-reply (== referral-id) ans)
(when (log-level? (current-logger) 'debug) (define-values (new-zone ignored-timers) (incorporate-complete-answer ans zone))
(log-debug (format "Referral ~v results in origin ~v:~n" (when (log-level? (current-logger) 'debug)
referral-id zone-origin)) (log-debug (format "Referral ~v results in origin ~v:~n"
(for ([k (set-union (list->set (hash-keys zone)) referral-id zone-origin))
(list->set (hash-keys new-zone)))] (for ([k (set-union (list->set (hash-keys zone))
#:when (in-bailiwick? k zone-origin)) (list->set (hash-keys new-zone)))]
(log-debug (format "Old ~v ~v~nNew ~v ~v" #:when (in-bailiwick? k zone-origin))
k (hash-ref zone k 'missing) (log-debug (format "Old ~v ~v~nNew ~v ~v"
k (hash-ref new-zone k 'missing)))) k (hash-ref zone k 'missing)
(log-debug "=-=-=-=-=-=")) k (hash-ref new-zone k 'missing))))
(define nameserver-names (for/set ([rr nameserver-rrs]) (rr-rdata rr))) (log-debug "=-=-=-=-=-="))
(extend-transition (define nameserver-names (for/set ([rr nameserver-rrs]) (rr-rdata rr)))
(retry-question (struct-copy question-state w (extend-transition
[nameservers-tried (set-union nameservers-tried (retry-question (struct-copy question-state w
nameserver-names)] [nameservers-tried (set-union nameservers-tried
[zone new-zone] nameserver-names)]
[retry-count (+ old-retry-count 1)])) [zone new-zone]
(unsubscribe referral-id))])))] [retry-count (+ old-retry-count 1)]))
(delete-role referral-id))]))]
[(? complete-answer? ans) [(? complete-answer? ans)
(transition w (send-message (answered-question q ans)))] (transition w (send-message (answered-question q ans)))]
[(partial-answer base cnames) [(partial-answer base cnames)
@ -283,17 +286,18 @@
;; TODO: record chains of CNAMEs to avoid pathologically-long chains ;; TODO: record chains of CNAMEs to avoid pathologically-long chains
(define cname-q (cname-question cname q)) (define cname-q (cname-question cname q))
(list (send-message cname-q) (list (send-message cname-q)
(subscribe/fresh subscription-id (role/fresh subscription-id
(message-handlers (expanding-cnames q acc remaining) (topic-subscriber (answered-question cname-q (wild)))
[(answered-question (== cname-q) ans) #:state (expanding-cnames q acc remaining)
(define new-acc (if ans (merge-answers acc ans) acc)) [(answered-question (== cname-q) ans)
(define new-remaining (- remaining 1)) (define new-acc (if ans (merge-answers acc ans) acc))
(define new-w (expanding-cnames q new-acc new-remaining)) (define new-remaining (- remaining 1))
(transition new-w (define new-w (expanding-cnames q new-acc new-remaining))
(unsubscribe subscription-id) (transition new-w
(if (zero? new-remaining) (delete-role subscription-id)
(send-message (answered-question q new-acc)) (if (zero? new-remaining)
'()))])))) (send-message (answered-question q new-acc))
'()))])))
cnames))])])) cnames))])]))
(require "test-rrs.rkt") (require "test-rrs.rkt")