download: Don't report the progress too fast.

* guix/utils.scm (<progress-reporter>): New record type.
(call-with-progress-reporter): New procedure.
* guix/build/download.scm (dump-port*, rate-limited, progress-reporter/file):
New procedures.
(ftp-fetch, http-fetch): Use 'dump-port*'.
(progress-proc): Remove procedure.
* guix/scripts/substitute.scm (progress-report-port): Rewrite in terms of
<progress-reporter>.
(process-substitution): Adjust accordingly.

3 files changed, 161 insertions, 99 deletions

diff --git a/guix/build/download.scm b/guix/build/download.scm
index bcf22663b0..9490f48055 100644
--- a/guix/build/download.scm
+++ b/guix/build/download.scm
@@ -27,6 +27,7 @@
   #:use-module (guix base64)
   #:use-module (guix ftp-client)
   #:use-module (guix build utils)
+  #:use-module (guix utils)
   #:use-module (rnrs io ports)
   #:use-module (rnrs bytevectors)
   #:use-module (srfi srfi-1)
@@ -45,7 +46,7 @@
             url-fetch
             byte-count->string
             current-terminal-columns
-            progress-proc
+            progress-reporter/file
             uri-abbreviation
             nar-uri-abbreviation
             store-path-abbreviation))
@@ -148,65 +149,97 @@ Otherwise return STORE-PATH."
    (define time-monotonic time-tai))
   (else #t))
 
-(define* (progress-proc file size
-                        #:optional (log-port (current-output-port))
-                        #:key (abbreviation basename))
-  "Return a procedure to show the progress of FILE's download, which is SIZE
-bytes long.  The returned procedure is suitable for use as an argument to
-`dump-port'.  The progress report is written to LOG-PORT, with ABBREVIATION
-used to shorten FILE for display."
-  ;; XXX: Because of <http://bugs.gnu.org/19939> this procedure is often not
-  ;; called as frequently as we'd like too; this is especially bad with Nginx
-  ;; on hydra.gnu.org, which returns whole nars as a single chunk.
-  (let ((start-time #f))
-    (let-syntax ((with-elapsed-time
-                     (syntax-rules ()
-                       ((_ elapsed body ...)
-                        (let* ((now     (current-time time-monotonic))
-                               (elapsed (and start-time
-                                             (duration->seconds
-                                              (time-difference now
-                                                               start-time)))))
-                          (unless start-time
-                            (set! start-time now))
-                          body ...)))))
+
+;; TODO: replace '(@ (guix build utils) dump-port))'.
+(define* (dump-port* in out
+                     #:key (buffer-size 16384)
+                     (reporter (make-progress-reporter noop noop noop)))
+  "Read as much data as possible from IN and write it to OUT, using chunks of
+BUFFER-SIZE bytes.  After each successful transfer of BUFFER-SIZE bytes or
+less, report the total number of bytes transferred to the REPORTER, which
+should be a <progress-reporter> object."
+  (define buffer
+    (make-bytevector buffer-size))
+
+  (call-with-progress-reporter reporter
+    (lambda (report)
+      (let loop ((total 0)
+                 (bytes (get-bytevector-n! in buffer 0 buffer-size)))
+        (or (eof-object? bytes)
+            (let ((total (+ total bytes)))
+              (put-bytevector out buffer 0 bytes)
+              (report total)
+              (loop total (get-bytevector-n! in buffer 0 buffer-size))))))))
+
+(define (rate-limited proc interval)
+  "Return a procedure that will forward the invocation to PROC when the time
+elapsed since the previous forwarded invocation is greater or equal to
+INTERVAL (a time-duration object), otherwise does nothing and returns #f."
+  (let ((previous-at #f))
+    (lambda args
+      (let* ((now (current-time time-monotonic))
+             (forward-invocation (lambda ()
+                                   (set! previous-at now)
+                                   (apply proc args))))
+        (if previous-at
+            (let ((elapsed (time-difference now previous-at)))
+              (if (time>=? elapsed interval)
+                  (forward-invocation)
+                  #f))
+            (forward-invocation))))))
+
+(define* (progress-reporter/file file size
+                                 #:optional (log-port (current-output-port))
+                                 #:key (abbreviation basename))
+  "Return a <progress-reporter> object to show the progress of FILE's download,
+which is SIZE bytes long.  The progress report is written to LOG-PORT, with
+ABBREVIATION used to shorten FILE for display."
+  (let ((start-time (current-time time-monotonic))
+        (transferred 0))
+    (define (render)
+      "Write the progress report to LOG-PORT."
+      (define elapsed
+        (duration->seconds
+         (time-difference (current-time time-monotonic) start-time)))
       (if (number? size)
-          (lambda (transferred cont)
-            (with-elapsed-time elapsed
-              (let* ((%          (* 100.0 (/ transferred size)))
-                     (throughput (if elapsed
-                                     (/ transferred elapsed)
-                                     0))
-                     (left       (format #f " ~a  ~a"
-                                         (abbreviation file)
-                                         (byte-count->string size)))
-                     (right      (format #f "~a/s ~a ~a~6,1f%"
-                                         (byte-count->string throughput)
-                                         (seconds->string elapsed)
-                                         (progress-bar %) %)))
-                (display "\r\x1b[K" log-port)
-                (display (string-pad-middle left right
-                                            (current-terminal-columns))
-                         log-port)
-                (flush-output-port log-port)
-                (cont))))
-          (lambda (transferred cont)
-            (with-elapsed-time elapsed
-              (let* ((throughput (if elapsed
-                                     (/ transferred elapsed)
-                                     0))
-                     (left       (format #f " ~a"
-                                         (abbreviation file)))
-                     (right      (format #f "~a/s ~a | ~a transferred"
-                                         (byte-count->string throughput)
-                                         (seconds->string elapsed)
-                                         (byte-count->string transferred))))
-                (display "\r\x1b[K" log-port)
-                (display (string-pad-middle left right
-                                            (current-terminal-columns))
-                         log-port)
-                (flush-output-port log-port)
-                (cont))))))))
+          (let* ((%  (* 100.0 (/ transferred size)))
+                 (throughput (/ transferred elapsed))
+                 (left       (format #f " ~a  ~a"
+                                     (abbreviation file)
+                                     (byte-count->string size)))
+                 (right      (format #f "~a/s ~a ~a~6,1f%"
+                                     (byte-count->string throughput)
+                                     (seconds->string elapsed)
+                                     (progress-bar %) %)))
+            (display "\r\x1b[K" log-port)
+            (display (string-pad-middle left right
+                                        (current-terminal-columns))
+                     log-port)
+            (flush-output-port log-port))
+          (let* ((throughput (/ transferred elapsed))
+                 (left       (format #f " ~a"
+                                     (abbreviation file)))
+                 (right      (format #f "~a/s ~a | ~a transferred"
+                                     (byte-count->string throughput)
+                                     (seconds->string elapsed)
+                                     (byte-count->string transferred))))
+            (display "\r\x1b[K" log-port)
+            (display (string-pad-middle left right
+                                        (current-terminal-columns))
+                     log-port)
+            (flush-output-port log-port))))
+
+    (progress-reporter
+     (start render)
+     ;; Report the progress every 300ms or longer.
+     (report
+      (let ((rate-limited-render
+             (rate-limited render (make-time time-monotonic 300000000 0))))
+        (lambda (value)
+          (set! transferred value)
+          (rate-limited-render))))
+     ;; Don't miss the last report.
+     (stop render))))
 
 (define* (uri-abbreviation uri #:optional (max-length 42))
   "If URI's string representation is larger than MAX-LENGTH, return an
@@ -264,9 +297,10 @@ out if the connection could not be established in less than TIMEOUT seconds."
                          (dirname (uri-path uri)))))
     (call-with-output-file file
       (lambda (out)
-        (dump-port in out
-                   #:buffer-size %http-receive-buffer-size
-                   #:progress (progress-proc (uri-abbreviation uri) size))))
+        (dump-port* in out
+                    #:buffer-size %http-receive-buffer-size
+                    #:reporter (progress-reporter/file
+                                (uri-abbreviation uri) size))))
 
     (ftp-close conn))
     (newline)
@@ -755,10 +789,10 @@ certificates; otherwise simply ignore them."
            (lambda (p)
              (if (port? bv-or-port)
                  (begin
-                   (dump-port bv-or-port p
-                              #:buffer-size %http-receive-buffer-size
-                              #:progress (progress-proc (uri-abbreviation uri)
-                                                        size))
+                   (dump-port* bv-or-port p
+                               #:buffer-size %http-receive-buffer-size
+                               #:reporter (progress-reporter/file
+                                           (uri-abbreviation uri) size))
                    (newline))
                  (put-bytevector p bv-or-port))))
          file))
@@ -863,8 +897,8 @@ otherwise simply ignore them."
                               hashes))
                 content-addressed-mirrors))
 
-  ;; Make this unbuffered so 'progress-proc' works as expected.  _IOLBF means
-  ;; '\n', not '\r', so it's not appropriate here.
+  ;; Make this unbuffered so 'progress-report/file' works as expected.  _IOLBF
+  ;; means '\n', not '\r', so it's not appropriate here.
   (setvbuf (current-output-port) _IONBF)
 
   (setvbuf (current-error-port) _IOLBF)
@@ -879,8 +913,4 @@ otherwise simply ignore them."
                file url)
        #f))))
 
-;;; Local Variables:
-;;; eval: (put 'with-elapsed-time 'scheme-indent-function 1)
-;;; End:
-
 ;;; download.scm ends here
diff --git a/guix/scripts/substitute.scm b/guix/scripts/substitute.scm
index 61eff18869..3dcf42d0d1 100755
--- a/guix/scripts/substitute.scm
+++ b/guix/scripts/substitute.scm
@@ -34,7 +34,8 @@
   #:use-module ((guix build utils) #:select (mkdir-p dump-port))
   #:use-module ((guix build download)
                 #:select (current-terminal-columns
-                          progress-proc uri-abbreviation nar-uri-abbreviation
+                          progress-reporter/file
+                          uri-abbreviation nar-uri-abbreviation
                           (open-connection-for-uri
                            . guix:open-connection-for-uri)
                           close-connection
@@ -814,23 +815,25 @@ was found."
                                 (= (string-length file) 32)))))
               (narinfo-cache-directories directory)))
 
-(define (progress-report-port report-progress port)
-  "Return a port that calls REPORT-PROGRESS every time something is read from
-PORT.  REPORT-PROGRESS is a two-argument procedure such as that returned by
-`progress-proc'."
-  (define total 0)
-  (define (read! bv start count)
-    (let ((n (match (get-bytevector-n! port bv start count)
-               ((? eof-object?) 0)
-               (x x))))
-      (set! total (+ total n))
-      (report-progress total (const n))
-      ;; XXX: We're not in control, so we always return anyway.
-      n))
-
-  (make-custom-binary-input-port "progress-port-proc"
-                                 read! #f #f
-                                 (cut close-connection port)))
+(define (progress-report-port reporter port)
+  "Return a port that continuously reports the bytes read from PORT using
+REPORTER, which should be a <progress-reporter> object."
+  (match reporter
+    (($ <progress-reporter> start report stop)
+     (let* ((total 0)
+            (read! (lambda (bv start count)
+                     (let ((n (match (get-bytevector-n! port bv start count)
+                                ((? eof-object?) 0)
+                                (x x))))
+                       (set! total (+ total n))
+                       (report total)
+                       n))))
+       (start)
+       (make-custom-binary-input-port "progress-port-proc"
+                                      read! #f #f
+                                      (lambda ()
+                                        (close-connection port)
+                                        (stop)))))))
 
 (define-syntax with-networking
   (syntax-rules ()
@@ -947,12 +950,11 @@ DESTINATION as a nar file.  Verify the substitute against ACL."
                           (dl-size  (or download-size
                                         (and (equal? comp "none")
                                              (narinfo-size narinfo))))
-                          (progress (progress-proc (uri->string uri)
-                                                   dl-size
-                                                   (current-error-port)
-                                                   #:abbreviation
-                                                   nar-uri-abbreviation)))
-                     (progress-report-port progress raw)))
+                          (reporter (progress-reporter/file
+                                     (uri->string uri) dl-size
+                                     (current-error-port)
+                                     #:abbreviation nar-uri-abbreviation)))
+                     (progress-report-port reporter raw)))
                   ((input pids)
                    (decompressed-port (and=> (narinfo-compression narinfo)
                                              string->symbol)
@@ -961,8 +963,8 @@ DESTINATION as a nar file.  Verify the substitute against ACL."
       (restore-file input destination)
       (close-port input)
 
-      ;; Skip a line after what 'progress-proc' printed, and another one to
-      ;; visually separate substitutions.
+      ;; Skip a line after what 'progress-reporter/file' printed, and another
+      ;; one to visually separate substitutions.
       (display "\n\n" (current-error-port))
 
       (every (compose zero? cdr waitpid) pids))))
diff --git a/guix/utils.scm b/guix/utils.scm
index 8fbbdc3563..de4aa65319 100644
--- a/guix/utils.scm
+++ b/guix/utils.scm
@@ -33,6 +33,7 @@
   #:autoload   (rnrs io ports) (make-custom-binary-input-port)
   #:use-module ((rnrs bytevectors) #:select (bytevector-u8-set!))
   #:use-module (guix memoization)
+  #:use-module (guix records)
   #:use-module ((guix build utils) #:select (dump-port mkdir-p))
   #:use-module ((guix build syscalls) #:select (mkdtemp! fdatasync))
   #:use-module (ice-9 format)
@@ -94,7 +95,13 @@
             call-with-decompressed-port
             compressed-output-port
             call-with-compressed-output-port
-            canonical-newline-port))
+            canonical-newline-port
+
+            <progress-reporter>
+            progress-reporter
+            make-progress-reporter
+            progress-reporter?
+            call-with-progress-reporter))
 
 
 ;;;
@@ -747,3 +754,26 @@ a location object."
   `((line     . ,(and=> (location-line loc) 1-))
     (column   . ,(location-column loc))
     (filename . ,(location-file loc))))
+
+
+;;;
+;;; Progress reporter.
+;;;
+
+(define-record-type* <progress-reporter>
+  progress-reporter make-progress-reporter progress-reporter?
+  (start   progress-reporter-start)     ; thunk
+  (report  progress-reporter-report)    ; procedure
+  (stop    progress-reporter-stop))     ; thunk
+
+(define (call-with-progress-reporter reporter proc)
+  "Start REPORTER for progress reporting, and call @code{(@var{proc} report)}
+with the resulting report procedure.  When @var{proc} returns, the REPORTER is
+stopped."
+  (match reporter
+    (($ <progress-reporter> start report stop)
+     (dynamic-wind start (lambda () (proc report)) stop))))
+
+;;; Local Variables:
+;;; eval: (put 'call-with-progress-reporter 'scheme-indent-function 1)
+;;; End: