parallel: Add parallel-map that works with SRFI-18

Makefile

@@ -39,7 +39,6 @@ package: retropikzel/${LIBRARY}/LICENSE retropikzel/${LIBRARY}/VERSION retropikz
 		--authors=${AUTHOR} \
 		--doc=${DOCFILE} \
 		--description="${DESCRIPTION}" \
-		--test=${TESTFILE} \
 		${LIBRARY_FILE}
 
 ${PKG}: package

retropikzel/parallel-srfi-18.scm

@@ -1,19 +1,100 @@
 (define thread-proc (lambda (x) #t))
 
-(define (new-thread)
+(define (new-thread name)
-  (make-thread
+  (let
-    (lambda ()
+    ((thread
-      (thread-specific-set!
+       (make-thread
-        (current-thread)
+         (lambda ()
-        (thread-proc (thread-specific (current-thread)))))))
+           (letrec*
+             ((runner
+                (lambda ()
+                  (when (list? (thread-specific (current-thread)))
+                    (for-each
+                      (lambda (job)
+                          (let ((lst (list-ref job 0))
+                                (index (list-ref job 1))
+                                (proc (list-ref job 2))
+                                (result (list-ref job 3))
+                                (getter (list-ref job 4))
+                                (setter (list-ref job 5)))
+                            (guard (condition
+                                     (else
+                                       (display "Error in thread: ")
+                                       (write (thread-name (current-thread)))
+                                       (display ", error: ")
+                                       (write condition)
+                                       (newline)
+                                       (list-set! threads
+                                                  (string->number (thread-name (current-thread)))
+                                                  (new-thread (thread-name (current-thread))))))
+                              (setter result
+                                         index
+                                         (apply proc
+                                                (list (getter lst index)))))))
+                      (thread-specific (current-thread)))
+                    (thread-specific-set! (current-thread) #f))
+                  (runner))))
+             (runner)))
+         name)))
+    (thread-specific-set! thread #f)
+    (thread-start! thread)
+    thread))
 
-(define threads (make-list thread-count (new-thread)))
+(define threads
+  (letrec*
+    ((looper
+       (lambda (count result)
+         (if (>= count thread-count)
+           result
+           (looper (+ count 1)
+                   (append result (list (new-thread (number->string count)))))))))
+    (looper 0 '())))
 
 (define-syntax parallel-map
   (syntax-rules ()
     ((_ env (l args body ...) lst)
-     (let* ((lst-length (length lst))
+     (letrec*
-            (thread-proc (eval `(lambda args body ...)
+       ((result (make-list (length lst) #f))
-                               (apply environment 'env))))
+        (lst-length (length lst))
-       (map thread-proc lst)
+        (loop-length (- (length lst) 2))
-       ))))
+        (proc (eval `(lambda args body ...) (apply environment env)))
+        (job-queues (make-vector thread-count '()))
+        (index 0)
+        (waited? #f)
+        (waiter (lambda ()
+                  (for-each
+                    (lambda (thread)
+                      (if (not (thread-specific thread))
+                        (set! waited? #t)
+                        (set! waited? #f)))
+                    threads)
+                  (if (not waited?) (waiter))))
+        (queu-builder
+          (lambda (index)
+            (when (< index lst-length)
+              (for-each
+                (lambda (thread)
+                  (when (< index lst-length)
+                    (let ((thread-index (string->number (thread-name thread))))
+                      (vector-set! job-queues
+                                   thread-index
+                                   (append (vector-ref job-queues thread-index)
+                                           (list (list lst
+                                                       index
+                                                       proc
+                                                       result
+                                                       list-ref
+                                                       list-set!)))))
+                    (set! index (+ index 1))))
+                threads)
+              (queu-builder index)))))
+       (queu-builder 0)
+       (for-each
+         (lambda (thread)
+           (thread-specific-set!
+             thread
+             (vector-ref job-queues
+                         (string->number (thread-name thread)))))
+         threads)
+       (waiter)
+       result))))

retropikzel/parallel.scm

@@ -1,6 +1,4 @@
 (define-syntax parallel-map
   (syntax-rules ()
     ((_ env (l args body ...) lst)
-     (let* ((lst-length (length lst))
+     (map (eval '(lambda args body ...) (apply environment env)) lst))))
-            (proc (eval `(lambda args body ...) (apply environment 'env))))
-       (map proc lst)))))

retropikzel/parallel.sld

@@ -3,14 +3,12 @@
   (import (scheme base)
           (scheme write)
           (scheme eval)
-          (retropikzel hardware-info)
+          (retropikzel hardware-info))
-          (retropikzel purer))
   (include "parallel/shared.scm")
   (cond-expand
     ((library (srfi 18))
      (import (srfi 18))
      (include "parallel-srfi-18.scm"))
-    (else
+    (else (include "parallel.scm"))
-      (import (scheme base))
+    )
-      (include "parallel.scm")))
   (export parallel-map))

retropikzel/parallel/README.md

@@ -1 +1 @@
-WIP
+Utilities to process things in parallel easily

retropikzel/parallel/test.scm

@@ -1,9 +1,23 @@
+(import (scheme base)
+        (scheme write)
+        (retropikzel tap)
+        (retropikzel debug)
+        (retropikzel parallel)
+        (srfi 19)
+        (srfi 64))
 
-(write (parallel-map
+(test-runner-current (tap-runner))
-         ((scheme base))
+
-         (lambda (i)
+(define lst '(1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16))
-           (+ i 1))
+
-         '(1 2 3 4 5 6 7 8 9 10)))
+(write (parallel-map '((scheme base)) (lambda (i) (+ i 1)) lst))
+(newline)
+(newline)
+(write (parallel-map '((scheme base)) (lambda (i) (+ i 1)) lst))
+(newline)
+(newline)
+(write (parallel-map '((scheme base)) (lambda (i) (+ i 1)) lst))
+(newline)
 (newline)