branch wip-offload updated: tmp2

[Mathieu Othacehe]

This is an automated email from the git hooks/post-receive script.

mothacehe pushed a commit to branch wip-offload
in repository guix-cuirass.

The following commit(s) were added to refs/heads/wip-offload by this push:
     new 062f148  tmp2
062f148 is described below

commit 062f14861ac5b80b455210323f6553fc15d4f6f2
Author: Mathieu Othacehe <othacehe@gnu.org>
AuthorDate: Fri Dec 11 17:46:02 2020 +0100

    tmp2
---
 src/cuirass/base.scm          |  1 +
 src/cuirass/database.scm      |  6 ++++++
 src/cuirass/remote-server.scm | 29 ++++++++++++++++++++++++-----
 src/cuirass/remote-worker.scm | 15 ++++++---------
 src/cuirass/remote.scm        | 12 ++++++++----
 src/cuirass/templates.scm     |  2 +-
 6 files changed, 46 insertions(+), 19 deletions(-)

diff --git a/src/cuirass/base.scm b/src/cuirass/base.scm
index 64bf524..3a87fc3 100644
--- a/src/cuirass/base.scm
+++ b/src/cuirass/base.scm
@@ -662,6 +662,7 @@ updating the database accordingly."
            (db-update-build-status! drv (build-status failed)))
          (log-message "bogus build-failed event for '~a'" drv)))
     (('workers workers)
+     (db-clear-workers)
      (for-each (lambda (worker)
                  (db-add-worker (sexp->worker worker)))
                workers))
diff --git a/src/cuirass/database.scm b/src/cuirass/database.scm
index 3baaf61..971cbd0 100644
--- a/src/cuirass/database.scm
+++ b/src/cuirass/database.scm
@@ -85,6 +85,7 @@
             db-get-build-products
             db-add-worker
             db-get-workers
+            db-clear-workers
             db-get-evaluation-summary
             db-get-checkouts
             read-sql-file
@@ -1447,3 +1448,8 @@ SELECT name, address, systems, last_seen from Workers"))
                       (systems (with-input-from-string systems read))
                       (last-seen last-seen))
                      workers)))))))
+
+(define (db-clear-workers)
+  "Remove all workers from Workers table."
+  (with-db-writer-worker-thread db
+    (sqlite-exec db "DELETE FROM Workers;")))
diff --git a/src/cuirass/remote-server.scm b/src/cuirass/remote-server.scm
index f9d3be1..0dcfed7 100644
--- a/src/cuirass/remote-server.scm
+++ b/src/cuirass/remote-server.scm
@@ -154,8 +154,10 @@ Start a remote build server.\n"))
   "Return #t if there is some available work for the worker with the given
 NAME and #f otherwise."
   (let* ((worker (hash-ref %workers name))
-         (systems (worker-systems worker))
-         (queues (find-system-queues systems)))
+         (systems (and worker (worker-systems worker)))
+         (queues (if systems
+                     (find-system-queues systems)
+                     '())))
     (not (null? queues))))
 
 (define (pop-random-build name)
@@ -169,6 +171,20 @@ work for the worker with the given NAME."
          (queues (find-system-queues systems)))
     (q-pop! (random-queue queues))))
 
+(define (remove-unresponsive-workers!)
+  (let ((unresponsive
+         (hash-fold (lambda (key value old)
+                      (let* ((last-seen (worker-last-seen value))
+                             (diff (- (current-time) last-seen)))
+                        (if (> diff (%worker-timeout))
+                            (cons key old)
+                            old)))
+                    '()
+                    %workers)))
+    (for-each (lambda (worker)
+                (hash-remove! %workers worker))
+              unresponsive)))
+
 (define* (read-client-exp client-socket client exp)
   "Read the given EXP sent by CLIENT."
   (catch 'system-error
@@ -199,8 +215,10 @@ work for the worker with the given NAME."
 (define* (read-worker-exp exp #:key reply-worker)
   "Read the given EXP sent by a worker.  REPLY-WORKER is a procedure that can
 be used to reply to the worker."
-  (define (update-workers! worker proc)
-    (let* ((worker* (sexp->worker worker))
+  (define (update-workers! base-worker proc)
+    (let* ((worker* (worker
+                     (inherit (sexp->worker base-worker))
+                     (last-seen (current-time))))
            (name (worker-name worker*)))
       (proc name)
       (hash-set! %workers name worker*)))
@@ -421,7 +439,7 @@ frontend to the workers connected through the TCP backend."
     ;; Do not use the built-in zmq-proxy as we want to edit the envelope of
     ;; frontend messages before forwarding them to the backend.
     (let loop ()
-      (let ((items (zmq-poll* poll-items)))
+      (let ((items (zmq-poll* poll-items 1000)))
         ;; client -> remote-server.
         (when (zmq-socket-ready? items client-socket)
           (match (zmq-get-msg-parts-bytevector client-socket)
@@ -454,6 +472,7 @@ frontend to the workers connected through the TCP backend."
           (let ((msg (zmq-get-msg-parts-bytevector fetch-socket)))
             (zmq-send-msg-parts-bytevector client-socket msg)))
 
+        (remove-unresponsive-workers!)
         (loop)))))
 
 
diff --git a/src/cuirass/remote-worker.scm b/src/cuirass/remote-worker.scm
index bc68fbe..3917574 100644
--- a/src/cuirass/remote-worker.scm
+++ b/src/cuirass/remote-worker.scm
@@ -196,16 +196,13 @@ command.  REPLY is a procedure that can be used to reply 
to this server."
     (('no-build)
      #t)))
 
-(define (worker-ping base-worker service)
+(define (worker-ping worker service)
   (define (ping socket)
-    (let ((worker (worker
-                   (inherit base-worker)
-                   (last-seen (current-time)))))
-      (zmq-send-msg-parts-bytevector
-       socket
-       (list (make-bytevector 0)
-             (string->bv
-              (zmq-worker-ping (worker->sexp worker)))))))
+    (zmq-send-msg-parts-bytevector
+     socket
+     (list (make-bytevector 0)
+           (string->bv
+            (zmq-worker-ping (worker->sexp worker))))))
 
   (call-with-new-thread
    (lambda ()
diff --git a/src/cuirass/remote.scm b/src/cuirass/remote.scm
index dbf3cd5..f9b040e 100644
--- a/src/cuirass/remote.scm
+++ b/src/cuirass/remote.scm
@@ -31,6 +31,7 @@
   #:use-module (srfi srfi-26)
   #:use-module (ice-9 match)
   #:use-module (ice-9 rdelim)
+  #:use-module (ice-9 threads)
   #:export (worker
             worker?
             worker-address
@@ -40,6 +41,7 @@
             worker->sexp
             sexp->worker
             generate-worker-name
+            %worker-timeout
 
             publish-server
             add-substitute-url
@@ -136,6 +138,9 @@
   "Return the service name of the server."
   (string-append (gethostname) "-" (random-string 4)))
 
+(define %worker-timeout
+  (make-parameter 60))
+
 
 ;;;
 ;;; Store publishing.
@@ -159,8 +164,8 @@
   "This procedure starts a publishing server listening on PORT in a new
 process and returns the pid of the forked process.  Use PUBLIC-KEY and
 PRIVATE-KEY to sign narinfos."
-  (match (primitive-fork)
-    (0
+  (call-with-new-thread
+   (lambda ()
      (parameterize ((%public-key public-key)
                     (%private-key private-key))
        (with-store store
@@ -172,8 +177,7 @@ PRIVATE-KEY to sign narinfos."
                 (socket (open-server-socket socket-address)))
            (run-publish-server socket store
                                #:compressions
-                               (list %default-gzip-compression))))))
-    (pid pid)))
+                               (list %default-gzip-compression))))))))
 
 
 ;;;
diff --git a/src/cuirass/templates.scm b/src/cuirass/templates.scm
index 5e1965a..4e46434 100644
--- a/src/cuirass/templates.scm
+++ b/src/cuirass/templates.scm
@@ -1043,7 +1043,7 @@ completed builds divided by the time required to build 
them.")
       (table
        (@ (class "table table-sm table-hover table-striped"))
        ,@(if (null? builds)
-             `((th (@ (scope "col")) "No elements here."))
+             `((th (@ (scope "col")) "Idle"))
              `((thead (tr (th (@ (scope "col")) "ID")
                           (th (@ (scope "col")) "Job")
                           (th (@ (scope "col")) "Queued at")