Some kind of for-each-par-map

[wingo]

Some kind of for-each-par-map

[amirouche]

I almost managed to make it work with a fiber thread managing two queue: a) one for available workers b) another for pending work. See the commit at:

amirouche/guile-babelia@d52b996

The trouble is that there is an error inside fibers. When I comment some code in fibers related to gc, it completes:

modified   fibers/channels.scm
@@ -2,17 +2,17 @@
 
 ;;;; Copyright (C) 2016 Andy Wingo <[email protected]>
 ;;;; Copyright (C) 2017 Christopher Allan Webber <[email protected]>
-;;;; 
+;;;;
 ;;;; This library is free software; you can redistribute it and/or
 ;;;; modify it under the terms of the GNU Lesser General Public
 ;;;; License as published by the Free Software Foundation; either
 ;;;; version 3 of the License, or (at your option) any later version.
-;;;; 
+;;;;
 ;;;; This library is distributed in the hope that it will be useful,
 ;;;; but WITHOUT ANY WARRANTY; without even the implied warranty of
 ;;;; MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 ;;;; Lesser General Public License for more details.
-;;;; 
+;;;;
 ;;;; You should have received a copy of the GNU Lesser General Public
 ;;;; License along with this library; if not, write to the Free Software
 ;;;; Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
@@ -111,13 +111,13 @@ with a receiver fiber to send @var{message} over @var{channel}."
             (not (eq? put-flag get-flag)))))
        ;; First, publish this put operation.
        (enqueue! putq-box (vector put-flag resume-put message))
-       ;; Next, possibly clear off any garbage from queue.
-       (when (= (counter-decrement! putq-gc-counter) 0)
-         (dequeue-filter! putq-box
-                          (match-lambda
-                            (#(flag resume)
-                             (not (eq? (atomic-box-ref flag) 'S)))))
-         (counter-reset! putq-gc-counter))
+       ;; ;; Next, possibly clear off any garbage from queue.
+       ;; (when (= (counter-decrement! putq-gc-counter) 0)
+       ;;   (dequeue-filter! putq-box
+       ;;                    (match-lambda
+       ;;                      (#(flag resume)
+       ;;                       (not (eq? (atomic-box-ref flag) 'S)))))
+       ;;   (counter-reset! putq-gc-counter))
        ;; In the try phase, we scanned the getq for a get operation,
        ;; but we were unable to perform any of them.  Since then,
        ;; there might be a new get operation on the queue.  However

Otherwise, here is the error I get:

In ice-9/boot-9.scm:
  1717:10 19 (with-exception-handler _ _ #:unwind? _ # _)
In babelia/okvs/wiredtiger.scm:
   225:35 18 (_)
In babelia/web/api/search.scm:
    27:17 17 (_ _)
In babelia/okvs/fts.scm:
     38:2 16 (fts-query #<<okvs> cnx: #<bytestructure 0x7fcb947b98a…> …)
In babelia/pool.scm:
   102:23 15 (pool-for-each-par-map #<procedure 7fcb94725930 at bab…> …)
In srfi/srfi-1.scm:
   592:29 14 (map1 (#vu8(1 110 215 60 222 161 24 55 230 135 127 …) …))
   592:29 13 (map1 (#vu8(1 110 215 61 13 76 209 241 71 238 92 # …) …))
   592:29 12 (map1 (#vu8(1 110 215 61 64 218 117 117 223 42 14 # …) …))
   592:29 11 (map1 (#vu8(1 110 215 61 115 64 62 192 7 163 119 # …) …))
   592:29 10 (map1 (#vu8(1 110 215 61 119 240 184 60 210 59 248 …) …))
   592:17  9 (map1 (#vu8(1 110 215 61 128 165 104 193 243 154 49 …) …))
In babelia/pool.scm:
     74:4  8 (_ _)
In fibers/operations.scm:
   176:10  7 (perform-operation _)
In fibers/channels.scm:
    116:9  6 (block-fn #<atomic-box 7fcb9421ecb0 value: S> _ #<proce…>)
In fibers/deque.scm:
    91:11  5 (dequeue-filter! #<atomic-box 7fcb942b9360 value: (() …> …)
In unknown file:
           4 (filter #<procedure 7fcb94892ec0 at fibers/channels.sc…> …)
In fibers/channels.scm:
   117:26  3 (_ _)
In ice-9/boot-9.scm:
  1650:16  2 (raise-exception _ #:continuable? _)
In babelia/okvs/wiredtiger.scm:
   224:13  1 (_ _)
In ice-9/boot-9.scm:
  1650:16  0 (raise-exception _ #:continuable? _)
ice-9/boot-9.scm:1650:16: In procedure raise-exception:
Throw to key `match-error' with args `("match" "no matching pattern" #(#<atomic-box 7fcb9421ecb0 value: S> #<procedure resume (thunk)> (work #<procedure 7fcb961ec800 at babelia/pool.scm:102:52 ()> . #<<channel> getq: #<atomic-box 7fcb9421ed30 value: (())> getq-gc-counter: #<atomic-box 7fcb9421ed20 value: 42> putq: #<atomic-box 7fcb9421ed00 value: (())> putq-gc-counter: #<atomic-box 7fcb9421ecf0 value: 42>>)))'.

[aconchillo]

Fixed in with #32.