WIP

1 files changed, 76 insertions, 9 deletions

diff --git a/guix-build-coordinator/agent.scm b/guix-build-coordinator/agent.scm
index 424d808..e5ba2c5 100644
--- a/guix-build-coordinator/agent.scm
+++ b/guix-build-coordinator/agent.scm
@@ -20,6 +20,7 @@
 
 (define-module (guix-build-coordinator agent)
   #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-9)
   #:use-module (srfi srfi-11)
   #:use-module (srfi srfi-19)
   #:use-module (ice-9 match)
@@ -43,6 +44,11 @@
   #:use-module (guix-build-coordinator agent-messaging abstract)
   #:export (run-agent))
 
+(define-record-type <upload-progress>
+  (make-upload-progress bytes-sent)
+  upload-progress?
+  (bytes-sent    upload-progress-bytes-sent  set-upload-progress-bytes-sent!))
+
 (define (run-agent uuid
                    coordinator-interface
                    systems
@@ -95,6 +101,61 @@
       (write-textfile metrics-registry
                       metrics-file)))
 
+  (define upload-slots
+    (make-vector 2 #f))
+
+  (define queued-uploads
+    '())
+
+  (define uploads-mutex
+    (make-mutex))
+  (define uploads-condition-variable
+    (make-condition-variable))
+
+  (define (with-upload-slot lgr file p)
+    (define (report-bytes-sent bytes)
+      (peek "UPLOAD SLOTS" upload-slots)
+      (peek "QUEUED UPLOADS" queued-uploads)
+      (display
+       (simple-format #f "bytes sent: ~A\n" bytes)))
+
+    (lock-mutex uploads-mutex)
+
+    (set! queued-uploads
+          (cons file queued-uploads))
+
+    (let loop ()
+      (let ((free-index
+             (any (lambda (index)
+                    (eq? (vector-ref upload-slots index)
+                         #f))
+                  (iota (vector-length upload-slots)
+                        0))))
+
+        (if free-index
+            (begin
+              (vector-set! upload-slots
+                           free-index
+                           (make-upload-progress 0))
+
+              (set! queued-uploads
+                    (delete file queued-uploads string=?))
+
+              (unlock-mutex uploads-mutex)
+
+              (call-with-values (lambda () (p report-bytes-sent))
+                (lambda vals
+                  (signal-condition-variable uploads-condition-variable)
+
+                  (apply values vals))))
+            (begin
+              (wait-condition-variable
+               uploads-condition-variable
+               uploads-mutex
+               (+ 15 (time-second (current-time))))
+
+              (loop))))))
+
   (define (process-job build)
     (let ((build-id (assoc-ref build "uuid"))
           (derivation-name (or (assoc-ref build "derivation_name")
@@ -143,7 +204,8 @@
                                           build-id
                                           derivation-name
                                           end-time
-                                          submit-outputs?)
+                                          submit-outputs?
+                                          with-upload-slot)
                       (post-build-failure lgr
                                           coordinator-interface
                                           build-id
@@ -579,7 +641,8 @@ but the guix-daemon claims it's unavailable"
 
 (define (post-build-success lgr coordinator-interface
                             build-id derivation end-time
-                            submit-outputs?)
+                            submit-outputs?
+                            with-upload-slot)
   (define output-details
     (map
      (match-lambda
@@ -670,13 +733,17 @@ but the guix-daemon claims it's unavailable"
       #:unwind? #t))
 
   (define (submit-one-output output-name output)
-    (log-msg lgr 'INFO
-             build-id ": submitting output "
-             (derivation-output-path output))
-    (submit-output coordinator-interface
-                   build-id output-name
-                   (derivation-output-path output)
-                   #:log (build-log-procedure lgr build-id)))
+    (with-upload-slot
+     lgr
+     (derivation-output-path output)
+     (lambda (report-bytes-sent)
+       (log-msg lgr 'INFO
+                build-id ": submitting output "
+                (derivation-output-path output))
+       (submit-output coordinator-interface
+                      build-id output-name
+                      (derivation-output-path output)
+                      #:log (build-log-procedure lgr build-id)))))
 
   (if submit-outputs?
       (begin