Re: Passing objects between threads

[Chris Vine <chris@cvine.freeserve.co.uk>]

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