I have some code in which fibers do not perform their work in parallel for some reason:
(define pool-initializer
(lambda* (#:key (parallelism (current-processor-count)))
(let ([channel-receive (make-channel)]
[scheduler (make-scheduler #:parallelism parallelism)])
(call-with-new-thread
(lambda ()
(run-fibers
(lambda ()
(let loop ([index parallelism])
(unless (zero? index)
(spawn-fiber (lambda () (worker index channel-receive)))
(loop (- index 1)))))
#:scheduler scheduler)))
(call-with-new-thread
(lambda ()
(work-distributor channel-receive)))
channel-receive)))
This procedure initializes a pool of fibers. The number of fibers depends on the parallelism keyword argument. Later on these fibers receive some work on a channel and return their result on a channel.
My understanding is, that fibers of the same scheduler should be able to run in parallel, if the scheduler's parallelism keyword argument is > 1. For some reason however, this does not happen and the fibers run sequentially, one after the other, when I give them work.
Why do they not run in parallel?
Is it because the scheduler is run inside a call-with-new-thread? Does this limit the parallelism to 1? I need to run it in call-with-new-thread, because I need to return the channel-receive, which will be input to the procedure, which gives work to a thing I called work-distributor. If I cannot run a scheduler in call-with-new-thread, how can I run a scheduler without blocking until all the fibers have completed their work?
So far I have only guesses, why my code does not do computation in parallel.
For a complete example of sequentially running fibers, I will paste my whole code below. It is actually based on @amirouche 's thread pool for his babelia project (I currently cannot find it on Github any longer and do not know where to look for it.), only that I am trying to use fibers instead of threads to perform work, hoping to take advantage of work stealing and work sharing as well as running more lightweight than threads.
The code:
(define-module (fibers-pool))
(use-modules
;; FIFO queue, not functional, using mutation
;; https://www.gnu.org/software/guile/manual/html_node/Queues.html
(ice-9 q)
(ice-9 match)
(ice-9 threads)
(rnrs exceptions)
(rnrs conditions)
;; fibers internals are needed for creating schedulers without running anything
;; in them immediately
(fibers)
(fibers channels)
(fibers internal))
(define displayln
(lambda (msg)
(display msg)
(newline)))
(define work-distributor
(lambda (channel-receive)
(let loop ([work-queue (make-q)]
[worker-channels-queue (make-q)])
(displayln "[WORK-DISTRIBUTOR]: work-distributor is listening for messages")
(display "[WORK-DISTRIBUTOR]: number of ready workers in queue: ")
(displayln (q-length worker-channels-queue))
(display "[WORK-DISTRIBUTOR]: number of works in queue: ")
(displayln (q-length work-queue))
(match (pk 'work-distributor-received-msg (get-message channel-receive))
[('worker-ready . channel-worker)
(displayln "[WORK-DISTRIBUTOR]: work-distributor received ready worker channel")
;; If there is no work for the ready worker, enqueue the worker,
;; otherwise give it work.
(cond
[(q-empty? work-queue)
(enq! worker-channels-queue channel-worker)]
[else
(let ([some-work (deq! work-queue)])
(put-message channel-worker (cons 'work some-work)))])
(loop work-queue worker-channels-queue)]
[('work . work)
;; ~work~ is always a pair of a thunk to be run and a return channel,
;; on which the result shall be put.
;; If there is no worker ready, enqueue the work, otherwise distribute
;; the work to a ready worker.
(cond
[(q-empty? worker-channels-queue)
(enq! work-queue work)]
[else
(let ([channel-worker (deq! worker-channels-queue)])
(put-message channel-worker (cons 'work work)))])
(loop work-queue worker-channels-queue)]
;; On any other message raise a condition.
[other
(raise
(condition
(make-error)
(make-message-condition "work-distributor received unrecognized message")
(make-irritants-condition (list other))))]))))
(define worker
(lambda (worker-index channel-receive)
(let ([channel-worker (make-channel)])
(displayln "[WORKER]: before worker message loop")
(let loop ()
;; Report as ready. Give my own channel to the work-distributor to let
;; it send me work.
(put-message channel-receive
(cons 'worker-ready
channel-worker))
;; Get messages sent to me by the distributor on my own channel.
(match (pk 'worker-got-msg (get-message channel-worker))
;; If I receive work, do the work and return it on the channel-return.
[('work . (thunk . channel-return))
;; Put the result on the return channel, so that anyone, who has the
;; a binding of the return channel, can access the result.
(put-message channel-return (thunk))
(loop)]
;; On any other message raise a condition.
[other
(raise
(condition
(make-error)
(make-message-condition "worker received unrecognized message")
(make-irritants-condition (list other))))])))))
(define pool-initializer
(lambda* (#:key (parallelism (current-processor-count)))
(let ([channel-receive (make-channel)]
[scheduler (make-scheduler #:parallelism parallelism)])
;; start as many workers as are desired
;; TODO: PROBLEM: ~run-fibers~ blocks. So we need a new thread to run the
;; fibers in a non-blocking way. LOOKUP: How to start fibers without
;; waiting for them to finish?
(call-with-new-thread
(lambda ()
(run-fibers
(lambda ()
(let loop ([index parallelism])
(unless (zero? index)
(display "[POOL INIT THREAD]: will spawn fiber ") (displayln index)
(spawn-fiber (lambda () (worker index channel-receive)))
;; We do not need to spawn new fibers in the same scheduler later. The
;; fibers should stay alive for the whole duration the program is
;; running.
(displayln "[POOL INIT THREAD]: fiber spawned")
(loop (- index 1)))))
#:scheduler scheduler)
(displayln "[POOL INIT]: pool init thread returning")))
(displayln "[POOL INIT]: will start work-distributor")
(call-with-new-thread
(lambda ()
(work-distributor channel-receive)))
;; Return the channel for receiving work, so that the outside context can
;; make use of it when calling ~publish~ to publish work.
channel-receive)))
(define publish
(lambda (work-as-thunk channel-receive)
;; The result of the computation can be taken from ~channel-return~.
(let ([channel-return (make-channel)])
;; Put work tagged as work on the receive channel of the work-distributor.
(let ([work-message (cons 'work (cons work-as-thunk channel-return))])
(display
(simple-format
#f "[PUBLISHER]: will publish the following work: ~a\n"
work-message))
(put-message channel-receive work-message))
(displayln "[PUBLISHER]: work published")
;; Return the ~channel-return~, so that the outside context can get
;; results from it.
channel-return)))
(define busy-work
(lambda ()
(let loop ([i 0])
(cond
[(< i 5e8) (loop (+ i 1))]
[else i]))))
;; Try it!
(define c-rec (pool-initializer #:parallelism 2))
(define c-ret-2 (publish (lambda () (busy-work)) c-rec))
(define c-ret-1 (publish (lambda () (busy-work)) c-rec))
(get-message c-ret-2)
(get-message c-ret-1)
On my machine, this runs in sequence, rather than in parallel.
I have some code in which fibers do not perform their work in parallel for some reason:
``` scheme
(define pool-initializer
(lambda* (#:key (parallelism (current-processor-count)))
(let ([channel-receive (make-channel)]
[scheduler (make-scheduler #:parallelism parallelism)])
(call-with-new-thread
(lambda ()
(run-fibers
(lambda ()
(let loop ([index parallelism])
(unless (zero? index)
(spawn-fiber (lambda () (worker index channel-receive)))
(loop (- index 1)))))
#:scheduler scheduler)))
(call-with-new-thread
(lambda ()
(work-distributor channel-receive)))
channel-receive)))
```
This procedure initializes a pool of fibers. The number of fibers depends on the `parallelism` keyword argument. Later on these fibers receive some work on a channel and return their result on a channel.
My understanding is, that fibers of the same scheduler should be able to run in parallel, if the scheduler's `parallelism` keyword argument is `> 1`. For some reason however, this does not happen and the fibers run sequentially, one after the other, when I give them work.
Why do they not run in parallel?
Is it because the scheduler is run inside a `call-with-new-thread`? Does this limit the parallelism to `1`? I need to run it in `call-with-new-thread`, because I need to return the `channel-receive`, which will be input to the procedure, which gives work to a thing I called `work-distributor`. If I cannot run a scheduler in `call-with-new-thread`, how can I run a scheduler without blocking until all the fibers have completed their work?
So far I have only guesses, why my code does not do computation in parallel.
For a complete example of sequentially running fibers, I will paste my whole code below. It is actually based on @amirouche 's thread pool for his babelia project (I currently cannot find it on Github any longer and do not know where to look for it.), only that I am trying to use fibers instead of threads to perform work, hoping to take advantage of work stealing and work sharing as well as running more lightweight than threads.
The code:
``` scheme
(define-module (fibers-pool))
(use-modules
;; FIFO queue, not functional, using mutation
;; https://www.gnu.org/software/guile/manual/html_node/Queues.html
(ice-9 q)
(ice-9 match)
(ice-9 threads)
(rnrs exceptions)
(rnrs conditions)
;; fibers internals are needed for creating schedulers without running anything
;; in them immediately
(fibers)
(fibers channels)
(fibers internal))
(define displayln
(lambda (msg)
(display msg)
(newline)))
(define work-distributor
(lambda (channel-receive)
(let loop ([work-queue (make-q)]
[worker-channels-queue (make-q)])
(displayln "[WORK-DISTRIBUTOR]: work-distributor is listening for messages")
(display "[WORK-DISTRIBUTOR]: number of ready workers in queue: ")
(displayln (q-length worker-channels-queue))
(display "[WORK-DISTRIBUTOR]: number of works in queue: ")
(displayln (q-length work-queue))
(match (pk 'work-distributor-received-msg (get-message channel-receive))
[('worker-ready . channel-worker)
(displayln "[WORK-DISTRIBUTOR]: work-distributor received ready worker channel")
;; If there is no work for the ready worker, enqueue the worker,
;; otherwise give it work.
(cond
[(q-empty? work-queue)
(enq! worker-channels-queue channel-worker)]
[else
(let ([some-work (deq! work-queue)])
(put-message channel-worker (cons 'work some-work)))])
(loop work-queue worker-channels-queue)]
[('work . work)
;; ~work~ is always a pair of a thunk to be run and a return channel,
;; on which the result shall be put.
;; If there is no worker ready, enqueue the work, otherwise distribute
;; the work to a ready worker.
(cond
[(q-empty? worker-channels-queue)
(enq! work-queue work)]
[else
(let ([channel-worker (deq! worker-channels-queue)])
(put-message channel-worker (cons 'work work)))])
(loop work-queue worker-channels-queue)]
;; On any other message raise a condition.
[other
(raise
(condition
(make-error)
(make-message-condition "work-distributor received unrecognized message")
(make-irritants-condition (list other))))]))))
(define worker
(lambda (worker-index channel-receive)
(let ([channel-worker (make-channel)])
(displayln "[WORKER]: before worker message loop")
(let loop ()
;; Report as ready. Give my own channel to the work-distributor to let
;; it send me work.
(put-message channel-receive
(cons 'worker-ready
channel-worker))
;; Get messages sent to me by the distributor on my own channel.
(match (pk 'worker-got-msg (get-message channel-worker))
;; If I receive work, do the work and return it on the channel-return.
[('work . (thunk . channel-return))
;; Put the result on the return channel, so that anyone, who has the
;; a binding of the return channel, can access the result.
(put-message channel-return (thunk))
(loop)]
;; On any other message raise a condition.
[other
(raise
(condition
(make-error)
(make-message-condition "worker received unrecognized message")
(make-irritants-condition (list other))))])))))
(define pool-initializer
(lambda* (#:key (parallelism (current-processor-count)))
(let ([channel-receive (make-channel)]
[scheduler (make-scheduler #:parallelism parallelism)])
;; start as many workers as are desired
;; TODO: PROBLEM: ~run-fibers~ blocks. So we need a new thread to run the
;; fibers in a non-blocking way. LOOKUP: How to start fibers without
;; waiting for them to finish?
(call-with-new-thread
(lambda ()
(run-fibers
(lambda ()
(let loop ([index parallelism])
(unless (zero? index)
(display "[POOL INIT THREAD]: will spawn fiber ") (displayln index)
(spawn-fiber (lambda () (worker index channel-receive)))
;; We do not need to spawn new fibers in the same scheduler later. The
;; fibers should stay alive for the whole duration the program is
;; running.
(displayln "[POOL INIT THREAD]: fiber spawned")
(loop (- index 1)))))
#:scheduler scheduler)
(displayln "[POOL INIT]: pool init thread returning")))
(displayln "[POOL INIT]: will start work-distributor")
(call-with-new-thread
(lambda ()
(work-distributor channel-receive)))
;; Return the channel for receiving work, so that the outside context can
;; make use of it when calling ~publish~ to publish work.
channel-receive)))
(define publish
(lambda (work-as-thunk channel-receive)
;; The result of the computation can be taken from ~channel-return~.
(let ([channel-return (make-channel)])
;; Put work tagged as work on the receive channel of the work-distributor.
(let ([work-message (cons 'work (cons work-as-thunk channel-return))])
(display
(simple-format
#f "[PUBLISHER]: will publish the following work: ~a\n"
work-message))
(put-message channel-receive work-message))
(displayln "[PUBLISHER]: work published")
;; Return the ~channel-return~, so that the outside context can get
;; results from it.
channel-return)))
(define busy-work
(lambda ()
(let loop ([i 0])
(cond
[(< i 5e8) (loop (+ i 1))]
[else i]))))
;; Try it!
(define c-rec (pool-initializer #:parallelism 2))
(define c-ret-2 (publish (lambda () (busy-work)) c-rec))
(define c-ret-1 (publish (lambda () (busy-work)) c-rec))
(get-message c-ret-2)
(get-message c-ret-1)
```
On my machine, this runs in sequence, rather than in parallel.