R/next.R

In mirai: Minimalist Async Evaluation Framework for R

# mirai ------------------------------------------------------------------------

#' Next >> Developer Interface
#'
#' `nextstream` retrieves the currently stored L'Ecuyer-CMRG RNG stream
#' for the specified compute profile and advances it to the next stream.
#'
#' These functions are exported for use by packages extending \pkg{mirai} with
#' alternative launchers of [daemon()] processes.
#'
#' For `nextstream`: Calling this function advances the stream stored within
#' the compute profile. This ensures that the next recursive stream is returned
#' on subsequent calls.
#'
#' @inheritParams mirai
#'
#' @return For `nextstream`: a length 7 integer vector, as given by
#'   `.Random.seed` when the L'Ecuyer-CMRG RNG is in use (may be passed directly
#'   to the `rs` argument of [daemon()]), or else NULL if a stream has not yet
#'   been created.
#'
#' @examples
#' daemons(sync = TRUE)
#' nextstream()
#' nextstream()
#'
#' nextget("url")
#'
#' daemons(0)
#'
#' @keywords internal
#' @export
#'
nextstream <- function(.compute = "default") next_stream(..[[.compute]])

#' Next >> Developer Interface
#'
#' `nextget` retrieves the specified item from the specified compute profile.
#'
#' @param x (character) item to retrieve: `"n"` (daemon count), `"dispatcher"`
#'   (dispatcher-to-host URL), `"url"` (daemon connection URL), or `"tls"`
#'   (client TLS configuration).
#'
#' @return For `nextget`: the requested item, or else NULL if not present.
#'
#' @keywords internal
#' @rdname nextstream
#' @export
#'
nextget <- function(x, .compute = "default") ..[[.compute]][[x]]

#' Next >> Developer Interface
#'
#' `nextcode` translates integer exit codes returned by [daemon()].
#'
#' @param xc (integer) return value from [daemon()].
#'
#' @return For `nextcode`: character string.
#'
#' @examples
#' nextcode(0L)
#' nextcode(1L)
#'
#' @keywords internal
#' @rdname nextstream
#' @export
#'
nextcode <- function(xc) {
  sprintf(
    "%d | Daemon %s",
    xc,
    switch(
      xc + 1L,
      "connection terminated",
      "idletime limit reached",
      "walltime limit reached",
      "task limit reached"
    )
  )
}

# internals --------------------------------------------------------------------

next_stream <- function(envir) {
  stream <- envir[["stream"]]
  if (is.integer(stream)) {
    `[[<-`(envir, "stream", parallel::nextRNGStream(stream))
  }
  stream
}

maybe_next_stream <- function(envir) {
  is.null(envir[["seed"]]) || return()
  next_stream(envir)
}