Order messages to the writer thread channel

So that they're processed in a first come first served manor.

1 files changed, 26 insertions, 1 deletions

diff --git a/guix-build-coordinator/utils/fibers.scm b/guix-build-coordinator/utils/fibers.scm
index 4f6431c..293429c 100644
--- a/guix-build-coordinator/utils/fibers.scm
+++ b/guix-build-coordinator/utils/fibers.scm
@@ -1,4 +1,5 @@
 (define-module (guix-build-coordinator utils fibers)
+  #:use-module (ice-9 q)
   #:use-module (ice-9 match)
   #:use-module (ice-9 atomic)
   #:use-module (ice-9 threads)
@@ -31,7 +32,9 @@
             port-read-timeout-error?
             port-write-timeout-error?
             with-fibers-timeout
-            with-fibers-port-timeouts)
+            with-fibers-port-timeouts
+
+            make-queueing-channel)
   #:replace (retry-on-error))
 
 (define %worker-thread-args
@@ -545,3 +548,25 @@ If already in the worker thread, call PROC immediately."
    (append
     args
     (list #:sleep-impl sleep))))
+
+(define (make-queueing-channel channel)
+  (define queue (make-q))
+
+  (let ((queue-channel (make-channel)))
+    (spawn-fiber
+     (lambda ()
+       (while #t
+         (if (q-empty? queue)
+             (enq! queue
+                   (perform-operation
+                    (get-operation queue-channel)))
+             (let ((front (q-front queue)))
+               (perform-operation
+                (choice-operation
+                 (wrap-operation (get-operation queue-channel)
+                                 (lambda (val)
+                                   (enq! queue val)))
+                 (wrap-operation (put-operation channel front)
+                                 (lambda _
+                                   (q-pop! queue))))))))))
+    queue-channel))