(lispkit thread channel)

Library (lispkit thread channel) implements channels for communicating, coordinating and synchronizing threads of execution. LispKit channels are based on the channel abstraction provided by the Go programming language.

LispKit channels are thread-safe FIFO buffers for synchronizing communication between multiple threads. The current implementation supports multiple simultaneous receives and sends. It allows channels to be either synchronous or asynchronous by providing buffering capabilities. Furthermore, the library supports timeouts via channel timers and channel tickers.

The main differences compared to channels in the Go programming language are:

Channels do not have any type information.
Sending to a channel that gets closed does not panic, it unblocks all senders immediately with the fail flag set to non-#f.
Closing an already closed channel does not result in an error.
There is support for choosing what channels to select on at runtime via channel-select*.

Channels

(channel? obj)

Returns #t if obj is a channel, otherwise #f is returned.

(make-channel) (make-channel capacity)

Returns a new channel with a buffer size of capacity. If capacity is 0, the channel is synchronous and all its operations will block until a remote client sends/receives messages. Channels with a buffer capacity > 0 are asynchronous, but block if the buffer is exhausted.

(channel-send! channel msg)

Sends message msg to channel. channel-send! blocks if the capacity of channel is exhausted. channel-send! returns the fail flag of the send operation, i.e. #f is returned if the send operation succeeded.

(channel-receive! channel) (channel-receive! channel none)

Receives a message from channel and returns the message. If there is no message available, channel-receive! blocks. If the receive operation fails, none is returned, if provided. The default for none is #f.

Receives a message from channel and returns the message. If there is no message available, channel-try-receive! returns none, if provided. The default for none is #f.

Procedure channel-select* allows selecting channels that are chosen programmatically. It takes input that looks like this:

(channel-select*
  `((,chan1 meta1)         ; receive
    (,chan2 meta2 message) ; send
    (,chan3 meta3) ...))

channel-select* returns three values msg, fail, and meta, where msg is the message that was sent over the channel, fail is #t if the channel was closed and #f otherwise, and meta is the datum supplied in the arguments.

For example, if a message arrived on chan3 above, meta would be meta3 in that case. This allows one to see which channel a message came from, i.e. if you supply metadata that is the channel itself.

This is a channel switch that will send or receive on at most one channel, picking whichever clause is able to complete soonest. If no clause is ready, channel-select will block until one does, unless else is specified which will execute its body instead of blocking. Multiple send and receive clauses can be specified interchangeably, but only one clause will trigger and get executed. Example:

(channel-select
  ((chan1 -> msg fail)
     (if fail
         (print "chan1 closed!")
         (print "chan1 says " msg)))
  ((chan2 -> msg fail)
     (if fail
         (print "chan2 closed!")
         (print "chan2 says " msg))))

Receive clauses have the form ((chan -> msg [fail]) body ...). They execute body with msg bound to the message object and fail bound to a boolean flag indicating failure. Receiving from a closed channel immediately completes with this fail flag set to non-#f.

Send clauses have the form ((chan <- msg [fail]) body ...). They execute body after msg has been sent to a receiver, successfully buffered onto the channel, or if channel was closed. Sending to a closed channel immediately completes with the fail flag set to #f.

A send or receive clause on a closed channel with no fail-flag binding specified will immediately return void without executing body. This can be combined with recursion like this:

;; loop forever until either chan1 or chan2 closes
(let loop ()
   (channel-select
     ((chan1 -> msg)
        (display* "chan1 says " msg) (loop))
     ((chan2 <- 123)
        (display* "chan2 got  " 123) (loop))))

Or like this:

;; loop forever until chan1 closes. replacing chan2 is
;; important to avoid busy-wait!
(let loop ((chan2 chan2))
  (channel-select
    ((chan1 -> msg)
       (display* "chan1 says " msg)
       (loop chan2))
    ((chan2 -> msg fail)
       (if fail
           (begin
             (display* "chan2 closed, keep going")
             ;; create new forever-blocking channel
             (loop (make-channel 0)))
           (begin
             (display* "chan2 says " msg)
             (loop chan2))))))

channel-select returns the return value of the executed clause's body. To do a non-blocking receive, you can do the following:

(channel-select
  ((chan1 -> msg fail) (if fail #!eof msg))
  (else 'eagain))

channel-range continuously waits for messages to arrive on channel. Once a message msg is available, body ... gets executed and channel-range waits again for the next message to arrive. channel-range does not terminate unless channel is closed. The following statement is equivalent:

(let ((chan channel))
  (let loop ()
    (channel-select
      ((chan -> msg fail)
        (unless fail (begin body ...)(loop))))))

Closes channel. This will unblock existing receivers and senders waiting for an operation on channel with thir fail flag set to a non-\#f value. All future receivers and senders will also immdiately unblock in this way, so there is a risk to run into busy-loops.

The optional fail flag of channel-close can be used to specify an alternative to the default #t. As this value is given to all receivers and senders of channel, the fail flag can be used as a "broadcast" mechanism. fail flag must not be set to #f though, as that would indicate a successful message transaction.

Closing an already closed channel will results in its fail flag being updated.

Timers

Returns #t if obj is a channel timer as provided by this library. Otherwise timer? returns #f.

next is a thunk returning three values: when-next, data, and fail. when-next is when to trigger the next time, expressed in seconds since January 1, 1970 TAI (e.g. computed via (current-second)), data is the payload returned when the triggers (it's usually the time in seconds when it triggers), and fail refers to a fail flag, which is usually #f for timers.

next will be called exaclty once on every timeout and once at "startup" and can thus mutate its own private state. next is called within a timer mutex lock and thus does not need to be synchronized.

Returns a timer channel that will "send" a single message after duration seconds after its creation. The message is the current-second value at the time of the timeout, i.e. not when the message was received. Receiving more than once on an timer channel will block indefinitely or deadlock the second time.

(channel-select
  ((chan1 -> msg)
     (display* "chan1 says " msg))
  (((timer 1) -> when)
     (display* "chan1 took too long")))

You cannot send to or close a timer channel. Creating timers is a relatively cheap operation. Timers may be garbage-collected before the timer triggers. Creating a timer does not spawn a new thread.

Returns a ticker channel that will "send" a message every duration seconds. The message is the current-second value at the time of the tick, i.e. not when it was received.

Stops a ticker channel, i.e. the channel will stop sending "tick" messages.

Last updated 13 days ago