[no subject]

[Mathieu Othacehe]

branch: wip-zmq
commit 576211fce24893b523544372716de08258528b04
Author: Mathieu Othacehe <othacehe@gnu.org>
AuthorDate: Thu Mar 11 07:59:32 2021 +0100

    tmp
---
 src/cuirass/remote-server.scm |  3 ++-
 src/cuirass/remote.scm        | 18 ++++++++++++++++++
 2 files changed, 20 insertions(+), 1 deletion(-)

diff --git a/src/cuirass/remote-server.scm b/src/cuirass/remote-server.scm
index 4cd7f53..00c85ce 100644
--- a/src/cuirass/remote-server.scm
+++ b/src/cuirass/remote-server.scm
@@ -387,7 +387,8 @@ frontend to the workers connected through the TCP backend."
                                   (zmq-empty-delimiter)
                                   (string->bv message)))))))
                (if (need-fetching? command)
-                   (zmq-message-send fetch-socket fetch-msg)
+                   (zmq-message-send* fetch-socket fetch-msg
+                                      ZMQ_DONTWAIT)
                    (read-worker-exp rest
                                     #:reply-worker reply-worker))))))
         (db-remove-unresponsive-workers (%worker-timeout))
diff --git a/src/cuirass/remote.scm b/src/cuirass/remote.scm
index 7a59a3d..7c21fb0 100644
--- a/src/cuirass/remote.scm
+++ b/src/cuirass/remote.scm
@@ -70,6 +70,7 @@
 
             zmq-poll*
             zmq-message-receive*
+            zmq-message-send*
             zmq-socket-ready?
             zmq-empty-delimiter
 
@@ -397,6 +398,20 @@ retries a call to PROC."
 
   safe)
 
+(define (EAGAIN-safe proc)
+  "Return a variant of PROC that catches EAGAIN 'zmq-error' exceptions and
+retries a call to PROC."
+  (define (safe . args)
+    (catch 'zmq-error
+      (lambda ()
+        (apply proc args))
+      (lambda (key errno . rest)
+        (if (= errno EAGAIN)
+            (log-message "Sending message would block.")
+            (apply throw key errno rest)))))
+
+  safe)
+
 (define zmq-poll*
   ;; Return a variant of ZMQ-POLL that catches EINTR errors.
   (EINTR-safe zmq-poll))
@@ -404,6 +419,9 @@ retries a call to PROC."
 (define zmq-message-receive*
   (EINTR-safe zmq-message-receive))
 
+(define zmq-message-send*
+  (EAGAIN-safe zmq-message-send))
+
 (define (zmq-socket-ready? items socket)
   "Return #t if the given SOCKET is part of ITEMS, a list returned by a
 'zmq-poll' call, return #f otherwise."