Make the parallel operations more continuous

Instead of batching the list items, change the batch size to a
parallelism limit and run up to that many fibers. When the processing
of one list item finishes, another will then start immediately after,
rather than when the whole batch is finished.

These changes also make the fibers-map and fibers-for-each operations
work with vectors as well as lists.

knots/parallelism.scm

@@ -35,25 +35,6 @@
             fibers-parallel
             fibers-let))
 
-;; 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 (get-batch batch-size 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))))
-
 (define (defer-to-parallel-fiber thunk)
   (let ((reply (make-channel)))
     (spawn-fiber
@@ -85,46 +66,86 @@
         (apply values result)))
      responses)))
 
-(define (fibers-batch-map proc batch-size . lists)
+(define (fibers-batch-map proc parallelism-limit . lists)
-  (let loop ((lists lists)
+  (define vecs (map (lambda (list-or-vec)
-             (result '()))
+                      (if (vector? list-or-vec)
-    (let ((batch
+                          list-or-vec
-           rest
+                          (list->vector list-or-vec)))
-           (get-batch batch-size lists)))
+                    lists))
-      (if (any null? batch)
+
-          result
+  (define vecs-length
-          (let ((response-channels
+    (vector-length (first vecs)))
-                 (apply map
+
-                        (lambda args
+  (define result-vec
-                          (defer-to-parallel-fiber
+    (make-vector vecs-length))
-                            (lambda ()
+
-                              (apply proc args))))
+  (let loop ((next-to-process-index 0)
-                        batch)))
+             (channel-indexes '()))
-            (loop rest
+    (if (and (eq? #f next-to-process-index)
-                  (append! result
+             (null? channel-indexes))
-                           (apply fetch-result-of-defered-thunks
+        (if (vector? (first lists))
-                                  response-channels))))))))
+            result-vec
+            (vector->list result-vec))
+
+        (if (or (= (length channel-indexes)
+                   (min parallelism-limit vecs-length))
+                (eq? #f next-to-process-index))
+            (let ((new-index
+                   new-channel-indexes
+                   (perform-operation
+                    (apply
+                     choice-operation
+                     (map
+                      (lambda (index)
+                        (wrap-operation
+                         (get-operation
+                          (vector-ref result-vec index))
+                         (lambda (result)
+                           (match result
+                             (('exception . exn)
+                              (raise-exception exn))
+                             (_
+                              (vector-set! result-vec
+                                           index
+                                           (first result))
+
+                              (values next-to-process-index
+                                      (lset-difference =
+                                                       channel-indexes
+                                                       (list index))))))))
+                      channel-indexes)))))
+              (loop new-index
+                    new-channel-indexes))
+
+            (loop (if (= (+ 1 next-to-process-index)
+                         vecs-length)
+                      #f
+                      (+ 1 next-to-process-index))
+                  (begin
+                    (vector-set!
+                     result-vec
+                     next-to-process-index
+                     (defer-to-parallel-fiber
+                       (lambda ()
+                         (apply proc
+                                (map (lambda (vec)
+                                       (vector-ref vec next-to-process-index))
+                                     vecs)))))
+                    (cons next-to-process-index
+                          channel-indexes)))))))
 
 (define (fibers-map proc . lists)
   (apply fibers-batch-map proc 20 lists))
 
-(define (fibers-batch-for-each proc batch-size . lists)
+(define (fibers-batch-for-each proc parallelism-limit . lists)
-  (let loop ((lists lists))
+  (apply fibers-batch-map
-    (let ((batch
+         (lambda args
-           rest
+           (apply proc args)
-           (get-batch batch-size lists)))
+           *unspecified*)
-      (if (any null? batch)
+         parallelism-limit
-          *unspecified*
+         lists)
-          (let ((response-channels
+
-                 (apply map
+  *unspecified*)
-                        (lambda args
-                          (defer-to-parallel-fiber
-                            (lambda ()
-                              (apply proc args))))
-                        batch)))
-            (apply fetch-result-of-defered-thunks
-                   response-channels)
-            (loop rest))))))
 
 (define (fibers-for-each proc . lists)
   (apply fibers-batch-for-each proc 20 lists))

tests/parallelism.scm

@@ -3,6 +3,7 @@
              (unit-test)
              (knots parallelism))
 
+;; Test fibers-map
 (run-fibers-for-tests
  (lambda ()
    (assert-equal
@@ -12,4 +13,34 @@
                 (* 2 i))
               (iota 34))))))
 
+;; Test fibers-batch-map with a large batch size
+(run-fibers-for-tests
+ (lambda ()
+   (assert-equal
+    1122
+    (apply + (fibers-batch-map
+              (lambda (i)
+                (* 2 i))
+              100
+              (iota 34))))))
+
+;; Test fibers-map with vectors
+(run-fibers-for-tests
+ (lambda ()
+   (assert-equal
+    1122
+    (apply + (vector->list
+              (fibers-map
+               (lambda (i)
+                 (* 2 i))
+               (list->vector (iota 34))))))))
+
+;; Test fibers-for-each
+(run-fibers-for-tests
+ (lambda ()
+   (fibers-for-each
+    (lambda (i)
+      (* 2 i))
+    (iota 34))))
+
 (display "parallelism test finished successfully\n")