Passing objects between threads

Passing objects between threads

[Panicz Maciej Godek]

Hi,
is there any easy way to create a channel (queue) that could be used to
communicate between threads? In particular, if the queue is empty, I would
like the consumer to wait until something appears in it (pretty much like
the channels in Clojure)

regs,
Panicz

Re: Passing objects between threads

[Diogo F. S. Ramos]

[-- Attachment #1: Type: text/plain, Size: 396 bytes --]

> Hi,
> is there any easy way to create a channel (queue) that could be used to
> communicate between threads? In particular, if the queue is empty, I would
> like the consumer to wait until something appears in it (pretty much like
> the channels in Clojure)

I don't know about Clojure, but I've written the attached program a
while ago.  It worked for my purposes, but it wasn't tested much.


[-- Attachment #2: channel.scm --]
[-- Type: application/octet-stream, Size: 4023 bytes --]

;;; channel.scm --- Data structure for threads synchronization

;; Copyright (C) 2014 Diogo F. S. Ramos <dfsr@riseup.net>

;; This program is free software: you can redistribute it and/or
;; modify it under the terms of the GNU General Public License as
;; published by the Free Software Foundation, either version 3 of the
;; License, or (at your option) any later version.

;; This program 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
;; General Public License for more details.

;; You should have received a copy of the GNU General Public License
;; along with this program.  If not, see
;; <http://www.gnu.org/licenses/>.

;;; Commentary:

;; Channel is a data structure for synchronization between threads.

;; A thread can put or get values from a channel.  If there isn't
;; another thread at the other end, the thread will block until other
;; thread appears.

;; Multiple threads can put and get values from a single channel but
;; only two at a time can communicate.

;;; Code:

(define-module (concurrency channel)
  #:use-module (srfi srfi-9)
  #:use-module (srfi srfi-9 gnu)
  #:use-module (ice-9 threads)
  #:export (make-channel channel? channel-put channel-get))

;;; Values used to know if there is something on the channel
(define no-value '(no-value))
(define receiver-waiting '(receiver-waiting))

(define-record-type <channel>
  (%make-channel receiver-mutex sender-mutex mutex cv v)
  channel?
  (receiver-mutex receiver-mutex)
  (sender-mutex sender-mutex)
  (mutex mutex)
  (cv cv)
  (v channel-value set-channel-value!))

(set-record-type-printer! <channel>
                          (lambda (record port)
                            (display "#<channel>" port)))

(define (make-channel)
  "Return a channel."
  (%make-channel (make-mutex) (make-mutex) (make-mutex)
                 (make-condition-variable) no-value))

(define (channel-get ch)
  "Get a value from the channel CH.

If there is no value availabe, it will block the caller until there is
one."
  ;; Block reiceivers so only one can go in
  (with-mutex (receiver-mutex ch)
    (take-value ch)))

(define (channel-put ch v)
  "Put a value into the channel CH.

If there is no one waiting for a value, it will block until a getter
appears."
  ;; Block senders so only one can go in
  (with-mutex (sender-mutex ch)
    (put-value ch v)))

(define (sender-waiting? ch)
  (not (eq? (channel-value ch) no-value)))

(define (receiver-waiting? ch)
  (eq? (channel-value ch) receiver-waiting))

(define (take-value ch)
  "Take a value from the channel.

It blocks the thread if there is no thread waiting with a value."
  ;; There should be only one thread getting a value from the channel
  (with-mutex (mutex ch)
    (when (not (sender-waiting? ch))
      ;; Add data to the channel which represents a receiver is
      ;; waiting and block until there is a sender with data in the
      ;; channel.
      (set-channel-value! ch receiver-waiting)
      (wait-condition-variable (cv ch) (mutex ch)))
    (let ((r (channel-value ch)))
      ;; Retrieve data from the channel and release the sender, which
      ;; is waiting.
      (set-channel-value! ch no-value)  ;clean channel's value
      (signal-condition-variable (cv ch))
      r)))

(define (put-value ch v)
  "Put a value into the channel.

It blocks the thread if there is no thread waiting for a value."
  ;; There should be only one thread putting a value into the channel
  (lock-mutex (mutex ch))
  (cond ((receiver-waiting? ch)
         (set-channel-value! ch v)
         ;; Signal to the receiver there is data in the channel
         (signal-condition-variable (cv ch))
         ;; wait for receiver to release us
         (unlock-mutex (mutex ch) (cv ch)))
        (else
         (set-channel-value! ch v)
         ;; wait for a receiver
         (wait-condition-variable (cv ch) (mutex ch))
         (unlock-mutex (mutex ch)))))

Re: Passing objects between threads

[Chaos Eternal]

As part of my (dont know when to finish) guile-termite project, I have a
thread-mailbox impletement.
https://github.com/ChaosEternal/guile-termite/tree/master/thread-mailbox

On Sat, Sep 10, 2016 at 6:36 PM Diogo F. S. Ramos <dfsr@riseup.net> wrote:

> > Hi,
> > is there any easy way to create a channel (queue) that could be used to
> > communicate between threads? In particular, if the queue is empty, I
> would
> > like the consumer to wait until something appears in it (pretty much like
> > the channels in Clojure)
>
> I don't know about Clojure, but I've written the attached program a
> while ago.  It worked for my purposes, but it wasn't tested much.
>
>

Re: Passing objects between threads

[Panicz Maciej Godek]

guile-termite looks like a great project, fingers crossed :)
thanks!

2016-09-10 13:18 GMT+02:00 Chaos Eternal <chaoseternal@shlug.org>:

> As part of my (dont know when to finish) guile-termite project, I have a
> thread-mailbox impletement.
> https://github.com/ChaosEternal/guile-termite/tree/master/thread-mailbox
>
> On Sat, Sep 10, 2016 at 6:36 PM Diogo F. S. Ramos <dfsr@riseup.net> wrote:
>
>> > Hi,
>> > is there any easy way to create a channel (queue) that could be used to
>> > communicate between threads? In particular, if the queue is empty, I
>> would
>> > like the consumer to wait until something appears in it (pretty much
>> like
>> > the channels in Clojure)
>>
>> I don't know about Clojure, but I've written the attached program a
>> while ago.  It worked for my purposes, but it wasn't tested much.
>>
>>

Re: Passing objects between threads

[Panicz Maciej Godek]

Looks very nice, thanks.
I'll certainly draw some inspiration from it!

2016-09-10 12:35 GMT+02:00 Diogo F. S. Ramos <dfsr@riseup.net>:

> > Hi,
> > is there any easy way to create a channel (queue) that could be used to
> > communicate between threads? In particular, if the queue is empty, I
> would
> > like the consumer to wait until something appears in it (pretty much like
> > the channels in Clojure)
>
> I don't know about Clojure, but I've written the attached program a
> while ago.  It worked for my purposes, but it wasn't tested much.
>
>

Re: Passing objects between threads

[Chris Vine]

On Sat, 10 Sep 2016 11:37:55 +0200
Panicz Maciej Godek <godek.maciek@gmail.com> wrote:
> Hi,
> is there any easy way to create a channel (queue) that could be used
> to communicate between threads? In particular, if the queue is empty,
> I would like the consumer to wait until something appears in it
> (pretty much like the channels in Clojure)

I haven't used Clojure channels, but what you describe is a traditional
blocking asynchronous queue.  Since guile has POSIX condition variables
they are trivial to make.  Here is one simple implementation, which
allows multiple writers and multiple readers.  async-queue-pop! blocks a
reader until something is available in the queue.  If there are
multiple readers, which reader awakens on any write is for the thread
scheduler. A loop is included in async-queue-pop! because POSIX
condition variables can spuriously wake up, so you have to test the
condition (whether the queue is empty or not) on any wake up before
returning.

You will find loads of literature and tutorials on this pattern, using
condition variables.


******************************************

(use-modules (srfi srfi-9)
             (ice-9 q)
             (ice-9 threads))

(define-record-type <async-queue>
  (_make-async-queue mutex cond q)
  async-queue?
  (mutex mutex-get)
  (cond cond-get)
  (q q-get))

   
(define (make-async-queue)
  (_make-async-queue (make-mutex)
                     (make-condition-variable)
                     (make-q)))

(define (async-queue-pop! a-q)
  (let ([mutex (mutex-get a-q)])
    (with-mutex mutex
      (let ([q (q-get a-q)])
        (let loop ()
          (if (q-empty? q)
              (begin
                (wait-condition-variable (cond-get a-q) mutex)
                (loop))
              (deq! q)))))))

(define (async-queue-push! a-q item)
  (with-mutex (mutex-get a-q)
    (enq! (q-get a-q) item))
  (signal-condition-variable (cond-get a-q)))

Re: Passing objects between threads

[Panicz Maciej Godek]

I agree, it is a simple concept, and easily implementable using condition
variables. However, it would probably be nicer if guile provided some
standard solution, instead of forcing every programmer to coming up with
their own ones.

I've noticed that there's a module (ice-9 occam-channel) bundled with
Guile, but unfortunately it is undocumented.

Maybe it is a problem with the Guile community that -- despite some
attempts like Guildhall -- we didn't manage to create anything like CPAN or
pip. (I think that the Chicken community was more successful in that regard)


2016-09-10 16:30 GMT+02:00 Chris Vine <chris@cvine.freeserve.co.uk>:

> On Sat, 10 Sep 2016 11:37:55 +0200
> Panicz Maciej Godek <godek.maciek@gmail.com> wrote:
> > Hi,
> > is there any easy way to create a channel (queue) that could be used
> > to communicate between threads? In particular, if the queue is empty,
> > I would like the consumer to wait until something appears in it
> > (pretty much like the channels in Clojure)
>
> I haven't used Clojure channels, but what you describe is a traditional
> blocking asynchronous queue.  Since guile has POSIX condition variables
> they are trivial to make.  Here is one simple implementation, which
> allows multiple writers and multiple readers.  async-queue-pop! blocks a
> reader until something is available in the queue.  If there are
> multiple readers, which reader awakens on any write is for the thread
> scheduler. A loop is included in async-queue-pop! because POSIX
> condition variables can spuriously wake up, so you have to test the
> condition (whether the queue is empty or not) on any wake up before
> returning.
>
> You will find loads of literature and tutorials on this pattern, using
> condition variables.
>
>
> ******************************************
>
> (use-modules (srfi srfi-9)
>              (ice-9 q)
>              (ice-9 threads))
>
> (define-record-type <async-queue>
>   (_make-async-queue mutex cond q)
>   async-queue?
>   (mutex mutex-get)
>   (cond cond-get)
>   (q q-get))
>
>
> (define (make-async-queue)
>   (_make-async-queue (make-mutex)
>                      (make-condition-variable)
>                      (make-q)))
>
> (define (async-queue-pop! a-q)
>   (let ([mutex (mutex-get a-q)])
>     (with-mutex mutex
>       (let ([q (q-get a-q)])
>         (let loop ()
>           (if (q-empty? q)
>               (begin
>                 (wait-condition-variable (cond-get a-q) mutex)
>                 (loop))
>               (deq! q)))))))
>
> (define (async-queue-push! a-q item)
>   (with-mutex (mutex-get a-q)
>     (enq! (q-get a-q) item))
>   (signal-condition-variable (cond-get a-q)))
>
>
>

Re: Passing objects between threads

[Chris Vine]

On Sat, 10 Sep 2016 22:16:22 +0200
Panicz Maciej Godek <godek.maciek@gmail.com> wrote:
> I agree, it is a simple concept, and easily implementable using
> condition variables. However, it would probably be nicer if guile
> provided some standard solution, instead of forcing every programmer
> to coming up with their own ones.
> 
> I've noticed that there's a module (ice-9 occam-channel) bundled with
> Guile, but unfortunately it is undocumented.
> 
> Maybe it is a problem with the Guile community that -- despite some
> attempts like Guildhall -- we didn't manage to create anything like
> CPAN or pip. (I think that the Chicken community was more successful
> in that regard)

Unnecessary and overly complex abstractions are undesirable when
dealing with native threads, because to use threads safely you need to
know what the implementation is doing behind your back.  (And I would
say they are undesirable in other cases as well, but not everyone may
agree with that.)

Some thread abstractions I like - I think Go channels are quite a nice
light weight wrapper over thread-safe and type-safe queues.  Generally
though I don't mind guile's lack of support for this given that it is
so simple to do yourself.

On your last point, I think by and large that guile has been reasonably
successful in its ecosystem.  There is a large set of libraries and
scheme extensions that come with guile out of the box, and plenty of
third party libraries on github and similar sites as well.  I also want
to inspect code I put on my system - if I do not know of the author and
trust her code, I definitely want to look at the code first, not just
install it over the internet.

I am not sure that internet distribution sites of the kind you mention
are particularly helpful unless they are curated, and there isn't
usually the manpower to do that for open source code.  Perl has waned
considerably notwithstanding CPAN.  (I am also suspicious of sites
which advertise themselves as: "A guild is an association of
independent craftspeople.  A guildhall is where they meet.  This
Guildhall aims to make a virtual space for Guile wizards and
journeyfolk to share code."  WTF.)

Chris

Re: Passing objects between threads

[Andy Wingo]

On Sat 10 Sep 2016 11:37, Panicz Maciej Godek <godek.maciek@gmail.com> writes:

> is there any easy way to create a channel (queue) that could be used to
> communicate between threads? In particular, if the queue is empty, I would
> like the consumer to wait until something appears in it (pretty much like
> the channels in Clojure)

If you are using kernel threads, you can use something like:

  https://lists.gnu.org/archive/html/guile-devel/2012-02/msg00033.html

If you are using a user-space library for lightweight concurrency like
8sync or something, they usually have a channel abstraction.  There is
one in https://github.com/wingo/fibers, for example (and in that one,
the channels are themselves threadsafe and can be used for passing data
between fibers running on different kernel threads).  A new release is
coming later this evening but it requires the latest development Guile.

I think more generally, now is a time for a lot of experimentation in
the concurrency side of Guile; I look forward to seeing what people
settle on :)

Andy

Re: Passing objects between threads

[Panicz Maciej Godek]

2016-09-11 20:34 GMT+02:00 Andy Wingo <wingo@pobox.com>:

> On Sat 10 Sep 2016 11:37, Panicz Maciej Godek <godek.maciek@gmail.com>
> writes:
>
> > is there any easy way to create a channel (queue) that could be used to
> > communicate between threads? In particular, if the queue is empty, I
> would
> > like the consumer to wait until something appears in it (pretty much like
> > the channels in Clojure)
>
> If you are using kernel threads, you can use something like:
>
>   https://lists.gnu.org/archive/html/guile-devel/2012-02/msg00033.html
>
> If you are using a user-space library for lightweight concurrency like
> 8sync or something, they usually have a channel abstraction.  There is
> one in https://github.com/wingo/fibers, for example (and in that one,
> the channels are themselves threadsafe and can be used for passing data
> between fibers running on different kernel threads).  A new release is
> coming later this evening but it requires the latest development Guile.
>
> Hah, both look rather complicated. I eventually followed Chris' advice and
developed "my own" queues, which turned out to be a bit simplified variant
of his code:

(use-modules (ice-9 nice-9) (ice-9 q) (ice-9 threads))

(define (make-channel)
  `(,(make-q) ,(make-mutex) ,(make-condition-variable)))

(define (channel? c)
  (and-let* (((queue mutex guard) c)
             ((q? queue))
             ((mutex? mutex))
             ((condition-variable? guard)))))

(define (send! data #;through channel)
  (let (((queue mutex guard) channel))
    (with-mutex mutex
      (enq! queue data)
      (signal-condition-variable guard))))

(define (receive! #;from channel)
  (let (((queue mutex guard) channel))
    (with-mutex mutex
      (while (q-empty? queue)
        (wait-condition-variable guard mutex))
      (deq! queue))))


> I think more generally, now is a time for a lot of experimentation in
> the concurrency side of Guile; I look forward to seeing what people
> settle on :)


I came up with an interface for thread communication, but I don't know how
to implement it -- maybe you (or someone else) would be able to help, or at
least warn not to go that way.

I am trying to extend my SLAYER framework to support reactive programming.
I came up with the following architecture: the event system gathers events
and puts them into a queue, for example, pressing key k causes a '(key-down
k) event to appear.

(define EVENT-QUEUE (make-channel))

(set-event-handler! (lambda (event) (send! event #;through channel)))

Then there is this update function, which takes the current state of the
world, and an event, and returns an updated event.

Thus, the evolution of the world can be understood as the fold-left on the
event queue:

(define (channel-fold update state channel)
  (let ((message (receive! #;from channel)))
    (channel-fold update (update state message) channel)))

However, if I put things that way, I have no access to the current state,
because it is only a transient variable being passed to the tail-recursive
call.

I could of course communicate between the threads using a global variable
and assignment, but this is nasty. Instead I thought of the following
interface: I would like to have two functions, let's call them "exposed"
and "snapshot". The first one would behave like identity (or values),
except that its evaluation would cause the snapshot value of the current
thread to be set. The second function would be used to retrieve the
snapshot value, so that it could be accessed from other threads.

For example, I could define the evolution thread in the following way:

(define evolution
  (call-with-new-thread
   (λ ()
     (channel-fold
      (λ (state action)
        (let ((state* (update state #;with action)))
          (exposed state*)))
      initial-state
      EVENT-QUEUE))))

Then, I could access the state variable using the snapshot procedure, so
that -- for example -- I could use it in the drawing routine:

(set-display-procedure!
 (λ ()
   (draw! (visualization (snapshot evolution)))))

What I would like about that solution is that even if the world was updated
much faster than the drawing routine could handle, the system could remain
responsive. (I also think that this interface would be nice with
time-constrained optimization algorithms, so that they could make the best
known solution available at any time)

However, the problem is that -- while procedures have their properties --
there seems to be nothing similar in the realm of threads. Normally in
similar cases I could use a thread-indexed hash table, but in the context
of multi-threading it would probably need to be protected by a mutex, and
it would likely slow things down.

Any suggestions?

Also, if the data is immutable, and only one thread does assignments, is it
OK to assume that variable reference and assignment are atomic operations,
or do I need mutices anyway?

Re: Passing objects between threads

[Chris Vine]

On Mon, 12 Sep 2016 21:58:02 +0200
Panicz Maciej Godek <godek.maciek@gmail.com> wrote:
[snip]
> Hah, both look rather complicated. I eventually followed Chris'
> advice and  
> developed "my own" queues, which turned out to be a bit simplified
> variant of his code:
> 
> (use-modules (ice-9 nice-9) (ice-9 q) (ice-9 threads))
> 
> (define (make-channel)
>   `(,(make-q) ,(make-mutex) ,(make-condition-variable)))
> 
> (define (channel? c)
>   (and-let* (((queue mutex guard) c)
>              ((q? queue))
>              ((mutex? mutex))
>              ((condition-variable? guard)))))
> 
> (define (send! data #;through channel)
>   (let (((queue mutex guard) channel))
>     (with-mutex mutex
>       (enq! queue data)
>       (signal-condition-variable guard))))
> 
> (define (receive! #;from channel)
>   (let (((queue mutex guard) channel))
>     (with-mutex mutex
>       (while (q-empty? queue)
>         (wait-condition-variable guard mutex))
>       (deq! queue))))

All I would say about this, if you want to use an asynchronous queue of
this kind as a "channel", is that "channels" often provide for
throttling of the write side so the producer(s) can't overwhelm the
consumer(s).  This might comprise a hard limit to the size of the queue
with a condition variable applied to the writer as well as the reader
(Andy's solution), but there are other ways of doing it, such as
inserting writer waits of increasing duration where the size of the
queue starts to grow too large.

> > I think more generally, now is a time for a lot of experimentation
> > in the concurrency side of Guile; I look forward to seeing what
> > people settle on :)  
> 
> 
> I came up with an interface for thread communication, but I don't
> know how to implement it -- maybe you (or someone else) would be able
> to help, or at least warn not to go that way.
> 
> I am trying to extend my SLAYER framework to support reactive
> programming. I came up with the following architecture: the event
> system gathers events and puts them into a queue, for example,
> pressing key k causes a '(key-down k) event to appear.
> 
> (define EVENT-QUEUE (make-channel))
> 
> (set-event-handler! (lambda (event) (send! event #;through channel)))
> 
> Then there is this update function, which takes the current state of
> the world, and an event, and returns an updated event.
> 
> Thus, the evolution of the world can be understood as the fold-left
> on the event queue:
> 
> (define (channel-fold update state channel)
>   (let ((message (receive! #;from channel)))
>     (channel-fold update (update state message) channel)))
> 
> However, if I put things that way, I have no access to the current
> state, because it is only a transient variable being passed to the
> tail-recursive call.
> 
> I could of course communicate between the threads using a global
> variable and assignment, but this is nasty. Instead I thought of the
> following interface: I would like to have two functions, let's call
> them "exposed" and "snapshot". The first one would behave like
> identity (or values), except that its evaluation would cause the
> snapshot value of the current thread to be set. The second function
> would be used to retrieve the snapshot value, so that it could be
> accessed from other threads.
> 
> For example, I could define the evolution thread in the following way:
> 
> (define evolution
>   (call-with-new-thread
>    (λ ()
>      (channel-fold
>       (λ (state action)
>         (let ((state* (update state #;with action)))
>           (exposed state*)))
>       initial-state
>       EVENT-QUEUE))))
> 
> Then, I could access the state variable using the snapshot procedure,
> so that -- for example -- I could use it in the drawing routine:
> 
> (set-display-procedure!
>  (λ ()
>    (draw! (visualization (snapshot evolution)))))
> 
> What I would like about that solution is that even if the world was
> updated much faster than the drawing routine could handle, the system
> could remain responsive. (I also think that this interface would be
> nice with time-constrained optimization algorithms, so that they
> could make the best known solution available at any time)
> 
> However, the problem is that -- while procedures have their
> properties -- there seems to be nothing similar in the realm of
> threads. Normally in similar cases I could use a thread-indexed hash
> table, but in the context of multi-threading it would probably need
> to be protected by a mutex, and it would likely slow things down.
> 
> Any suggestions?
> 
> Also, if the data is immutable, and only one thread does assignments,
> is it OK to assume that variable reference and assignment are atomic
> operations, or do I need mutices anyway?

I don't have sufficient grasp of what you want and how slayer operates
to help you with this.  When I want to multiplex a number of "channels"
onto a single receiver/event-queue thread, I use this:

https://github.com/ChrisVine/guile-a-sync/wiki/coroutines
https://github.com/ChrisVine/guile-a-sync/wiki/event-loop

(On the second reference, scroll down and see in particular the
await-task-in-thread! and await-task! procedures, and the other await*
procedures).  Each a-sync or compose-a-sync block constructs a
coroutine which operates as a kind of channel or fibre with 'await'
semantics - a single thread can wait concurrently on any number of them.
However, this doesn't look as if it is what you are after.

Chris

Re: Passing objects between threads

[Panicz Maciej Godek]

>
> I could of course communicate between the threads using a global variable
> and assignment, but this is nasty. Instead I thought of the following
> interface: I would like to have two functions, let's call them "exposed"
> and "snapshot". The first one would behave like identity (or values),
> except that its evaluation would cause the snapshot value of the current
> thread to be set. The second function would be used to retrieve the
> snapshot value, so that it could be accessed from other threads.
>
> For example, I could define the evolution thread in the following way:
>
> (define evolution
>   (call-with-new-thread
>    (λ ()
>      (channel-fold
>       (λ (state action)
>         (let ((state* (update state #;with action)))
>           (exposed state*)))
>       initial-state
>       EVENT-QUEUE))))
>
> Then, I could access the state variable using the snapshot procedure, so
> that -- for example -- I could use it in the drawing routine:
>
> (set-display-procedure!
>  (λ ()
>    (draw! (visualization (snapshot evolution)))))
>
> What I would like about that solution is that even if the world was
> updated much faster than the drawing routine could handle, the system could
> remain responsive. (I also think that this interface would be nice with
> time-constrained optimization algorithms, so that they could make the best
> known solution available at any time)
>
> However, the problem is that -- while procedures have their properties --
> there seems to be nothing similar in the realm of threads. Normally in
> similar cases I could use a thread-indexed hash table, but in the context
> of multi-threading it would probably need to be protected by a mutex, and
> it would likely slow things down.
>
> Any suggestions?
>

I have found that SRFI-18 provides a pair
thread-specific/thread-specific-set! that allows me to implement that
interface. Hooray!