1 files changed, 106 insertions, 0 deletions
diff --git a/guix-qa-frontpage/utils.scm b/guix-qa-frontpage/utils.scm
index 00e0ed7..abaed30 100644
--- a/guix-qa-frontpage/utils.scm
+++ b/guix-qa-frontpage/utils.scm
@@ -17,5 +17,111 @@
 ;;; <http://www.gnu.org/licenses/>.
 
 (define-module (guix-qa-frontpage utils)
+  #:use-module (srfi srfi-1)
+  #:use-module (srfi srfi-71)
+  #:use-module (ice-9 match)
+  #:use-module (fibers)
+  #:use-module (fibers channels)
   #:use-module (guix-build-coordinator utils fibers)
+  #:export (fiberize
+            fibers-map
+            fibers-for-each)
   #:re-export (with-fibers-port-timeouts))
+
+(define* (fiberize proc #:key (parallelism 1))
+  (let ((channel (make-channel)))
+    (for-each
+     (lambda _
+       (spawn-fiber
+        (lambda ()
+          (while #t
+            (let ((reply-channel args (car+cdr
+                                       (get-message channel))))
+              (put-message
+               reply-channel
+               (with-exception-handler
+                   (lambda (exn)
+                     (cons 'exception exn))
+                 (lambda ()
+                   (with-throw-handler #t
+                     (lambda ()
+                       (call-with-values
+                           (lambda ()
+                             (apply proc args))
+                         (lambda vals
+                           (cons 'result vals))))
+                     (lambda _
+                       (backtrace))))
+                 #:unwind? #t)))))
+        #:parallel? #t))
+     (iota parallelism))
+
+    (lambda args
+      (let ((reply-channel (make-channel)))
+        (put-message channel (cons reply-channel args))
+        (match (get-message reply-channel)
+          (('result . vals) (apply values vals))
+          (('exception . exn) (raise-exception exn)))))))
+
+(define (fibers-map proc . lists)
+  (let ((channels
+         (apply
+          map
+          (lambda args
+            (let ((channel (make-channel)))
+              (spawn-fiber
+               (lambda ()
+                 (put-message
+                  channel
+                  (with-exception-handler
+                      (lambda (exn)
+                        (cons 'exception exn))
+                    (lambda ()
+                      (with-throw-handler #t
+                        (lambda ()
+                          (call-with-values
+                              (lambda ()
+                                (apply proc args))
+                            (lambda val
+                              (cons 'result val))))
+                        (lambda _
+                          (backtrace))))
+                    #:unwind? #t))))
+              channel))
+          lists)))
+    (map
+     (match-lambda
+       (('result . val) val)
+       (('exception . exn) (raise-exception exn)))
+     (map get-message channels))))
+
+(define (fibers-for-each proc . lists)
+  ;; Like split-at, but don't care about the order of the resulting lists, and
+  ;; don't error if the list is shorter than i elements
+  (define (split-at* lst i)
+    (let lp ((l lst) (n i) (acc '()))
+      (if (or (<= n 0) (null? l))
+          (values (reverse! acc) l)
+          (lp (cdr l) (- n 1) (cons (car l) acc)))))
+
+  ;; As this can be called with lists with tens of thousands of items in them,
+  ;; batch the
+  (define batch-size 20)
+  (define (get-batch lists)
+    (let ((split-lists
+           (map (lambda (lst)
+                  (let ((batch rest (split-at* lst batch-size)))
+                    (cons batch rest)))
+                lists)))
+      (values (map car split-lists)
+              (map cdr split-lists))))
+
+  (let loop ((lists lists))
+    (call-with-values
+        (lambda ()
+          (get-batch lists))
+      (lambda (batch rest)
+        (apply fibers-map proc batch)
+        (unless (null? (car rest))
+          (loop  rest)))))
+  *unspecified*)