Channels

The main function is mizu_channel(). It creates a two-way shared-memory channel, with one lock-free ring per direction, and spawns a child R process connected to the other end. The child evaluates a quoted expression with ch bound to its side of the channel. Only a short join token is passed to the child on its command line; all data crosses through the rings.

library(mizu)

ch <- mizu_channel(quote(
  repeat {
    x <- mizu_recv(ch, timeout = 30)
    if (inherits(x, "mizu_sentinel")) break
    mizu_send(ch, x * 2)
  }
))

mizu_send(ch, 21)
mizu_recv(ch, timeout = 5)
#> [1] 42

mizu_send() publishes a message to the peer, and it is visible as soon as the call returns. mizu_recv() returns the next message. mizu_send_batch() and mizu_recv_batch() move a whole list of messages in one call. Use them at rates where the per-call overhead of R itself starts to matter.

A full ring, a timeout, an orderly close, and peer death are returned as class-tagged sentinel values, not errors. A receive loop tests for them with inherits(x, "mizu_sentinel"), or with the specific classes mizu_full, mizu_timeout, mizu_closed, mizu_peer_gone. mizu_is_sentinel() is the stricter test. It returns TRUE only for the sentinel objects that mizu itself returned, since a class test cannot tell a terminal state from a look-alike payload (a sentinel forwarded as a message arrives as an ordinary copy). No call needs an error handler. If the process at the other end dies, receives first drain the messages that it already published, then report mizu_peer_gone. mizu_alive(ch) reports at any time whether the peer still runs, without touching the rings. mizu_channel_info(ch) returns a read-only snapshot of the channel’s state: ring geometry and watermarks, the free-list and open caches, and the notify signals sent.

mizu_close() performs an orderly shutdown. It waits for the peer to finish reading its messages before it releases the shared resources:

mizu_close(ch)

If the peer’s code raises an error, the error crosses as data: a receive returns a mizu_error_remote condition object rather than signalling it. mizu_is_remote_error() tests for one, conditionMessage() leads with the remote type, and stop(x) signals it in your own session:

ch <- mizu_channel(quote(stop("boom")))
x <- mizu_recv(ch, timeout = 5)
mizu_is_remote_error(x)
#> [1] TRUE
conditionMessage(x)
#> [1] "simpleError: boom"
mizu_close(ch)

The peer also prints the error to its standard error, which appears on your console by default. The launcher argument controls how the peer is spawned. The default mizu_launcher() takes stdout and stderr arguments, so launcher = mizu_launcher(stderr = "peer.log") writes the error to a file instead. The same launcher spawns the R workers of mizu_pool().

Promises and handlers

In an event-driven program (a Shiny app, an httpuv server, anything built around an event loop), the channel can wait on a notify endpoint instead of blocking. When the peer publishes, the endpoint wakes the event loop, and later delivers the message on the main thread. There is no polling and no extra thread. This needs the later package, and promises for the promise forms.

mizu_recv_async() returns a promise without blocking. It resolves with the next message, the same object a blocking mizu_recv() would return, and rejects on close, peer death, or its own timeout with a classed error that carries the sentinel’s class name. mizu_recv_handler() is the stream form. It installs a function that the event loop calls with each burst of messages, and batch = TRUE delivers each burst as one list. Delivery is per burst, not per message. A promise costs tens of microseconds of R work against about a microsecond of transport, so use the batch handler for high message volumes. Every main-thread drain is bounded per event-loop turn and re-scheduled, so a saturated ring cannot starve the loop.

library(promises)
ch <- mizu_channel(quote(mizu_send(ch, mizu_recv(ch))))
mizu_recv_async(ch) %...>% print()
mizu_send(ch, 42L)
while (!later::loop_empty()) later::run_now(1) # pump the loop until the resolution lands
#> [1] 42
mizu_close(ch)

A channel can have only one of the two consumer forms at a time. Installing a handler while promises are pending, or requesting a promise while a handler is installed, is an error.