This package provides an unbounded, imperative queue that is supposed to
be thread safe and asynchronous exception safe. It is essentially the
same communication abstraction as
that each channel is split into separate sending and receiving ends,
ReceivePort respectively. This has at least two
- Program behavior can be more finely constrained via the type system.
- Channels can have zero
ReceivePortsassociated with them. Messages written to such a channel disappear into the aether and can be garbage collected. Note that
ReceivePortscan be subsequently attached to such a channel via
A channel can have multiple
ReceivePorts. This results in a publish-
subscribe pattern of communication: every element will be delivered to
every port. Alternatively, multiple threads can read from a single
port. This results in a push-pull pattern of communication, similar to
ZeroMQ: every element will be delivered to exactly one thread. Of
course both can be used together to form hybrid patterns of
A channel can only have one
SendPort. However multiple threads can
can safely write to a single port, allowing effects similar to multiple
Some of the tradeoffs of
split-channel compared to Cloud Haskell's
split-channelis restricted to in-process communications only.
split-channelhas no restriction on the type of values that may be communicated.
- There is a quasi-duality between the two approaches: Cloud Haskell's
ReceivePortsare special whereas
, which allows for considerably more efficient publish-subscribe communications than supported by Cloud Haskell at the present time.
- data SendPort a
- data ReceivePort a
- new :: IO (SendPort a, ReceivePort a)
- newSendPort :: IO (SendPort a)
- send :: SendPort a -> a -> IO ()
- receive :: ReceivePort a -> IO a
- sendMany :: SendPort a -> [a] -> IO ()
- listen :: SendPort a -> IO (ReceivePort a)
- duplicate :: ReceivePort a -> IO (ReceivePort a)
- split :: SendPort a -> IO (ReceivePort a, SendPort a)
- fold :: (a -> b -> b) -> ReceivePort a -> IO b
- unsafeFold :: (a -> b -> b) -> ReceivePort a -> IO b
SendPorts represent one end of the channel. There is only one
SendPort per channel, though it can be used from multiple threads.
Messages can be sent to the channel using
ReceivePorts represent the other end of a channel. A channel
can have many
ReceivePorts, which all receive the same messages in
a publish/subscribe like manner. A single
ReceivePort can be used
from multiple threads, where every message will be delivered to a
single thread in a push/pull like manner. Use
receive to fetch
messages from the channel.
Creates a new channel and a
(SendPort, ReceivePort) pair representing
the two sides of the channel.
Produces a new channel that initially has zero
Any elements written to this channel before a reader is
will be eligible for garbage collection.
Send an element to a channel. This is asynchronous and does not block.
Fetch an element from a channel. If no element is available, it blocks
until one is. Can be used in conjunction with
Atomically send many messages at once. Note that this function forces the spine of the list beforehand to minimize the critical section, which also helps prevent exceptions at inopportune times. Trying to send an infinite list will never send anything, though it will allocate a lot of memory trying to do so.
Create a new
ReceivePort attached the same channel as a given
ReceivePort starts out empty, and remains so
until more elements are written to the
Create a new
ReceivePort attached to the same channel as another
ReceivePort. These two ports will receive the same messages.
Any messages in the channel that have not been consumed by the
existing port will also appear in the new port.
This function associates a brand new channel with a existing send port, returning a new receive port associated with the existing send port and a new send port associated with the existing receive ports.
A possible use case is to transparently replace the backend of a service without affecting the clients of that service. For example, we might use it along the following lines:
swapService :: SendPort Request -> IO () swapService s = do (r', s') <- split s send s' ShutdownRequest forkNewService r'
This is not a good solution in all cases. For example, the service might consist of multiple threads, and maybe some of those send internal messages on the same channel as the clients. It would probably be a bug to change the destination that those internal messages go.
MVar would introduce an extra layer of
indirection, but also allows you to be selective about which senders
observe the effect. The clients would use an
MVar (SendPort RequestOrInternalMessage) whereas the internal
threads would use the
SendPort RequestOrInternalMessage directly,
without going through the MVar.
A right fold over a receiver, a generalization of
getChanContents = fold (:). Note that the type of
implies that the folding function needs to be sufficienctly non-strict,
otherwise the result cannot be productive.