From d36d382e2025563bb639af0ca377fef562070fcc Mon Sep 17 00:00:00 2001 From: Hans Dijkema Date: Fri, 30 Oct 2026 10:41:56 +0100 Subject: [PATCH] stable http transfer --- media-file-server.rkt | 31 ++++++- private/http-transfer.rkt | 120 ++++++++++++++++++++++++++++ scribblings/intro.scrbl | 8 ++ scribblings/media-file-server.scrbl | 4 + service.rkt | 43 +++++----- 5 files changed, 185 insertions(+), 21 deletions(-) create mode 100644 private/http-transfer.rkt diff --git a/media-file-server.rkt b/media-file-server.rkt index c1a6816..7e03b74 100644 --- a/media-file-server.rkt +++ b/media-file-server.rkt @@ -207,10 +207,17 @@ #"text/plain; charset=utf-8" '() (list #"Not found\n"))))] + [dispatch-files (sequence:make file-dispatcher not-found-dispatcher)] [dispatcher - (sequence:make - file-dispatcher - not-found-dispatcher)] + (lambda (connection request) + (let* ([path (url-path-string (request-uri request))] + [file (call-with-semaphore lock + (lambda () (hash-ref publications path #f)))]) + (if file + (log-media-transfer + request file + (lambda () (dispatch-files connection request))) + (not-found-dispatcher connection request))))] [confirmation (make-async-channel)] [stop (serve @@ -234,6 +241,24 @@ #f)))) +;;; Logs each published GET/HEAD request around the complete file dispatch. +;;; No publication lock is held while the renderer reads its response. +(define (log-media-transfer request file send!) + (let* ([started (current-inexact-monotonic-milliseconds)] + [completed? #f] + [range (headers-assq* #"Range" (request-headers/raw request))] + [client (request-client-ip request)] + [method (bytes->string/utf-8 (request-method request))]) + (dynamic-wind + (lambda () + (info-upnp-file-server "Media transfer started: client=~a method=~a file=~a range=~a" + client method file (if range (header-value range) "full"))) + (lambda () (send!) (set! completed? #t)) + (lambda () + (info-upnp-file-server "Media transfer ~a: client=~a method=~a file=~a elapsed-ms=~a" + (if completed? "sent" "interrupted") client method file + (round (- (current-inexact-monotonic-milliseconds) started))))))) + (define (dlna-response-headers _path) (list (header #"Accept-Ranges" #"bytes") diff --git a/private/http-transfer.rkt b/private/http-transfer.rkt new file mode 100644 index 0000000..fef38a1 --- /dev/null +++ b/private/http-transfer.rkt @@ -0,0 +1,120 @@ +#lang racket/base + +(require racket/async-channel) + +(provide call-with-http-timeout) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +; goal : Bound HTTP inactivity while allowing transfers of any duration. +; pre : Seconds is positive. Proc consumes its HTTP ports before returning +; and calls progress! whenever new response bytes arrive. +; post : The worker has exited and its HTTP resources have been closed. +; result : Proc's values, or its exception, or an inactivity timeout error. +; internals: Activity restarts sync/timeout. On cancellation, a break unwinds +; semaphore and dynamic-wind cleanup before the custodian closes +; remaining ports. No timed-out worker can continue changing state. +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +(define (call-with-http-timeout seconds proc) + (let* ((custodian (make-custodian)) + (activity (make-async-channel 1)) + (result (make-async-channel)) + (finished? #f) + (worker + (parameterize ((current-custodian custodian)) + (thread + (λ () + (with-handlers ((exn:break? void)) + (parameterize-break #t + (with-handlers + ((exn? (λ (exception) + (async-channel-put result (cons #f exception))))) + (let ((values-list + (call-with-values + (λ () (proc (λ () (async-channel-put activity #t)))) + list))) + (async-channel-put result (cons #t values-list))))))))))) + (dynamic-wind + void + (λ () + (let loop () + (let ((event (sync/timeout seconds + (handle-evt result (λ (value) (cons 'result value))) + (handle-evt activity (λ (_) 'progress))))) + (cond + ((not event) + (error 'http-transfer "No HTTP progress for ~a seconds" seconds)) + ((eq? event 'progress) (loop)) + (else + (set! finished? #t) + (if (cadr event) + (apply values (cddr event)) + (raise (cddr event)))))))) + (λ () + (unless (or finished? (thread-dead? worker)) + (break-thread worker)) + (thread-wait worker) + (custodian-shutdown-all custodian))))) + +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; +;; Tests for module http-transfer.rkt +;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;; + +(module+ test + (require rackunit racket/file) + + (check-equal? (call-with-values + (λ () (call-with-http-timeout 1 (λ (_) (values 1 2)))) list) + '(1 2)) + (check-exn #rx"original failure" + (λ () (call-with-http-timeout 1 (λ (_) (error "original failure"))))) + + (test-case "continuing progress allows a transfer longer than its idle limit" + (check-equal? + (call-with-http-timeout + 0.5 + (λ (progress!) + (let loop ((chunks 6)) + (unless (zero? chunks) + (sleep 0.15) + (progress!) + (loop (- chunks 1)))) + 'downloaded)) + 'downloaded)) + + (test-case "stalled HTTP workers release locks and close their ports" + (let ((lock (make-semaphore 1)) + (output #f) + (target (make-temporary-file "http-stall-~a"))) + (dynamic-wind + void + (λ () + (check-exn + #rx"No HTTP progress" + (λ () + (parameterize-break #f + (call-with-http-timeout + 0.1 + (λ (_) + (set! output (open-output-file target #:exists 'truncate)) + (call-with-semaphore lock (λ () (sync never-evt)))))))) + (check-true (semaphore-try-wait? lock)) + (check-true (port-closed? output))) + (λ () (delete-file target))))) + + (test-case "breaking the caller also unwinds its HTTP worker" + (let* ((ready (make-semaphore)) + (lock (make-semaphore 1)) + (caller + (thread + (λ () + (with-handlers ((exn:break? void)) + (call-with-http-timeout + 5 + (λ (_) + (call-with-semaphore + lock + (λ () (semaphore-post ready) (sync never-evt)))))))))) + (semaphore-wait ready) + (break-thread caller) + (check-not-false (sync/timeout 2 caller)) + (check-true (semaphore-try-wait? lock))))) diff --git a/scribblings/intro.scrbl b/scribblings/intro.scrbl index 59cb4d1..bbd2bff 100644 --- a/scribblings/intro.scrbl +++ b/scribblings/intro.scrbl @@ -31,3 +31,11 @@ The public interface is divided into layers: A normal application can use the device-oriented modules without dealing with SSDP, SOAP, service control URLs, or DIDL-Lite XML directly. The generic layer remains available for inspection and vendor-specific extensions. + +SOAP HTTP exchanges and service-description requests have a ten-second +response deadline. A stalled renderer +therefore raises a transport failure instead of indefinitely occupying a +player's transport lock. The private @tt{call-with-http-timeout} procedure +runs the exchange in a worker, propagates its values or exception and closes +its HTTP resources. Its @tt{progress!} callback can restart the inactivity +wait for streaming callers; SOAP calls use a single bounded exchange. diff --git a/scribblings/media-file-server.scrbl b/scribblings/media-file-server.scrbl index 7dc5298..86c3ed7 100644 --- a/scribblings/media-file-server.scrbl +++ b/scribblings/media-file-server.scrbl @@ -15,6 +15,10 @@ underlying Racket file dispatcher streams files and supports both @tt{HEAD} requests and HTTP byte ranges. Byte ranges are important for media-renderer probing and seeking. +At info log level, each published transfer reports its start and completion or +interruption, including client address, HTTP method, requested range and elapsed +time. No publication lock is held while the response is being streamed. + @defproc[(media-file-server? [value any/c]) boolean?]{ Returns @racket[#t] when @racket[value] is a media file server. } diff --git a/service.rkt b/service.rkt index 9d30191..671295f 100644 --- a/service.rkt +++ b/service.rkt @@ -13,6 +13,7 @@ racket/port racket/string xml + "private/http-transfer.rkt" "private/model.rkt" "private/xml.rkt") @@ -155,12 +156,15 @@ (define (read-xml-url location) (dbg-upnp-soap "Reading UPnP XML description from ~a" location) - (call/input-url - (string->url location) - (lambda (url) - (get-pure-port url '() #:redirections 3)) - (lambda (in) - (xml->xexpr (document-element (read-xml in)))))) + (call-with-http-timeout + 10 + (λ (_progress!) + (call/input-url + (string->url location) + (λ (url) + (get-pure-port url '() #:redirections 3)) + (λ (in) + (xml->xexpr (document-element (read-xml in)))))))) (define (read-service-actions service) (dbg-upnp-soap "Reading actions for service ~a" (upnp-service-service-type service)) @@ -298,18 +302,21 @@ "controlURL has no host" "controlURL" control-url)) (dbg-upnp-soap "SOAP request action=~a service=~a control-url=~a arguments=~s bytes=~a" action-name service-type control-url arguments (bytes-length body)) - (let-values ([(status response-headers in) - (http-sendrecv host - (url-request-target url-value) - #:ssl? ssl? - #:port port - #:method #"POST" - #:headers headers - #:data body)]) - (let ([code (http-status-code status)] - [response-body (read-response-body in)]) - (dbg-upnp-soap "SOAP response action=~a status=~a bytes=~a" action-name code (bytes-length response-body)) - (values code response-headers response-body)))))) + (call-with-http-timeout + 10 + (λ (_progress!) + (let-values ([(status response-headers in) + (http-sendrecv host + (url-request-target url-value) + #:ssl? ssl? + #:port port + #:method #"POST" + #:headers headers + #:data body)]) + (let ([code (http-status-code status)] + [response-body (read-response-body in)]) + (dbg-upnp-soap "SOAP response action=~a status=~a bytes=~a" action-name code (bytes-length response-body)) + (values code response-headers response-body)))))))) (define (bytes->xexpr body) (call-with-input-bytes