WIPwip-upload-slots

1 files changed, 54 insertions, 9 deletions

diff --git a/guix-build-coordinator/agent.scm b/guix-build-coordinator/agent.scm
index e5ba2c5..08af8bc 100644
--- a/guix-build-coordinator/agent.scm
+++ b/guix-build-coordinator/agent.scm
@@ -23,6 +23,7 @@
   #:use-module (srfi srfi-9)
   #:use-module (srfi srfi-11)
   #:use-module (srfi srfi-19)
+  #:use-module (srfi srfi-43)
   #:use-module (ice-9 match)
   #:use-module (ice-9 format)
   #:use-module (ice-9 futures)
@@ -45,8 +46,9 @@
   #:export (run-agent))
 
 (define-record-type <upload-progress>
-  (make-upload-progress bytes-sent)
+  (make-upload-progress file bytes-sent)
   upload-progress?
+  (file          upload-progress-file)
   (bytes-sent    upload-progress-bytes-sent  set-upload-progress-bytes-sent!))
 
 (define (run-agent uuid
@@ -113,11 +115,45 @@
     (make-condition-variable))
 
   (define (with-upload-slot lgr file p)
+    (define upload-progress-record
+      (make-upload-progress file 0))
+
+    (define last-progress-update-bytes-sent 0)
+
     (define (report-bytes-sent bytes)
-      (peek "UPLOAD SLOTS" upload-slots)
-      (peek "QUEUED UPLOADS" queued-uploads)
-      (display
-       (simple-format #f "bytes sent: ~A\n" bytes)))
+      (let ((bytes-now-sent
+             (+ (upload-progress-bytes-sent upload-progress-record)
+                bytes)))
+
+        (set-upload-progress-bytes-sent!
+         upload-progress-record
+         bytes-now-sent)
+
+        (when (> bytes-now-sent
+                 (+ last-progress-update-bytes-sent 50000000))
+          (set! last-progress-update-bytes-sent
+                bytes-now-sent)
+
+          (let ((uploads-count
+                 (vector-count upload-slots
+                               (lambda (slot)
+                                 (not (eq? #f slot)))))
+                (queued-uploads-count
+                 (length queued-uploads)))
+
+            (peek "UPLOAD SLOTS" upload-slots)
+            (peek "QUEUED UPLOADS" queued-uploads)
+
+            (vector-for-each
+             (lambda (_ upload-progress)
+               (display
+                (simple-format #f "~A: bytes sent: ~A\n"
+                               (upload-progress-file upload-progress)
+                               (rationalize (exact->inexact
+                                             (/ (upload-progress-bytes-sent
+                                                 upload-progress)
+                                                1000000))
+                                            0.1)))))))))
 
     (lock-mutex uploads-mutex)
 
@@ -127,8 +163,10 @@
     (let loop ()
       (let ((free-index
              (any (lambda (index)
-                    (eq? (vector-ref upload-slots index)
-                         #f))
+                    (if (eq? (vector-ref upload-slots index)
+                             #f)
+                        index
+                        #f))
                   (iota (vector-length upload-slots)
                         0))))
 
@@ -136,7 +174,7 @@
             (begin
               (vector-set! upload-slots
                            free-index
-                           (make-upload-progress 0))
+                           upload-progress-record)
 
               (set! queued-uploads
                     (delete file queued-uploads string=?))
@@ -145,6 +183,12 @@
 
               (call-with-values (lambda () (p report-bytes-sent))
                 (lambda vals
+                  (with-mutex uploads-mutex
+                    (peek "FREEING SLOT" free-index)
+                    (vector-set! upload-slots
+                                 free-index
+                                 #f))
+
                   (signal-condition-variable uploads-condition-variable)
 
                   (apply values vals))))
@@ -743,7 +787,8 @@ but the guix-daemon claims it's unavailable"
        (submit-output coordinator-interface
                       build-id output-name
                       (derivation-output-path output)
-                      #:log (build-log-procedure lgr build-id)))))
+                      #:log (build-log-procedure lgr build-id)
+                      #:report-bytes-sent report-bytes-sent))))
 
   (if submit-outputs?
       (begin