stable http transfer

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
+            (lambda (connection request)
-             file-dispatcher
+              (let* ([path (url-path-string (request-uri request))]
-             not-found-dispatcher)]
+                     [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")

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)))))

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.

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.
 }

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
+  (call-with-http-timeout
-   (string->url location)
+   10
-   (lambda (url)
+   (λ (_progress!)
-     (get-pure-port url '() #:redirections 3))
+     (call/input-url
-   (lambda (in)
+      (string->url location)
-     (xml->xexpr (document-element (read-xml in))))))
+      (λ (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)
+      (call-with-http-timeout
-                    (http-sendrecv host
+       10
-                                   (url-request-target url-value)
+       (λ (_progress!)
-                                   #:ssl? ssl?
+         (let-values ([(status response-headers in)
-                                   #:port port
+                       (http-sendrecv host
-                                   #:method #"POST"
+                                      (url-request-target url-value)
-                                   #:headers headers
+                                      #:ssl? ssl?
-                                   #:data body)])
+                                      #:port port
-        (let ([code (http-status-code status)]
+                                      #:method #"POST"
-              [response-body (read-response-body in)])
+                                      #:headers headers
-          (dbg-upnp-soap "SOAP response action=~a status=~a bytes=~a" action-name code (bytes-length response-body))
+                                      #:data body)])
-          (values code response-headers response-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