Skip to contents

AsyncBackend

Source: R/AsyncBackend.R
AsyncBackend.Rd

This is a concrete implementation of the abstract class Backend that implements the BackendService interface. This backend executes tasks in parallel asynchronously (i.e., without blocking the main R session) on a parallel::makeCluster() cluster created in a background R session.

See also

BackendService, Backend, SyncBackend, ProgressTrackingContext, and TaskState.

Super classes

parabar::BackendService -> parabar::Backend -> AsyncBackend

Active bindings

task_state

A list of logical values indicating the state of the task execution. See the TaskState class for more information on how the statues are determined. The following statuses are available:

  • task_not_started: Indicates whether the backend is free. TRUE signifies that no task has been started and the backend is free to deploy.

  • task_is_running: Indicates whether a task is currently running on the backend.

  • task_is_completed: Indicates whether a task has finished executing. TRUE signifies that the output of the task has not been fetched. Calling the method get_option() will move the output from the background R session to the main R session. Once the output has been fetched, the backend is free to deploy another task.

session_state

A list of logical values indicating the state of the background session managing the cluster. See the SessionState class for more information on the available statuses. The following statuses are available:

  • session_is_starting: Indicates whether the session is starting.

  • session_is_idle: Indicates whether the session is idle.

  • session_is_busy: Indicates whether the session is busy. A session is busy when a task is running or when the output of a task has not been fetched into the main R session. See the task_state field.

  • session_is_finished: Indicates whether the session was closed.

Methods

Public methods

Method new()

Create a new AsyncBackend object.

Usage

AsyncBackend$new()

Returns

An object of class AsyncBackend.

Method start()

Start the backend.

Usage

AsyncBackend$start(specification)

Arguments

specification

An object of class Specification that contains the backend configuration.

Returns

This method returns void. The resulting backend must be stored in the .cluster private field on the Backend abstract class, and accessible to any concrete backend implementations via the active binding cluster.

Method stop()

Stop the backend.

Usage

AsyncBackend$stop()

Returns

This method returns void.

Method clear()

Remove all objects from the backend. This function is equivalent to calling rm(list = ls(all.names = TRUE)) on each node in the backend.

Usage

AsyncBackend$clear()

Returns

This method returns void.

Method peek()

Inspect the backend for variables available in the .GlobalEnv.

Usage

AsyncBackend$peek()

Returns

This method returns a list of character vectors, where each element corresponds to a node in the backend. The character vectors contain the names of the variables available in the .GlobalEnv on each node.

Method export()

Export variables from a given environment to the backend.

Usage

AsyncBackend$export(variables, environment)

Arguments

variables

A character vector of variable names to export.

environment

An environment object from which to export the variables.

Returns

This method returns void.

Method evaluate()

Evaluate an arbitrary expression on the backend.

Usage

AsyncBackend$evaluate(expression)

Arguments

expression

An unquoted expression to evaluate on the backend.

Returns

This method returns the result of the expression evaluation.

Method sapply()

Run a task on the backend akin to parallel::parSapply().

Usage

AsyncBackend$sapply(x, fun, ...)

Arguments

x

An atomic vector or list to pass to the fun function.

fun

A function to apply to each element of x.

...

Additional arguments to pass to the fun function.

Returns

This method returns void. The output of the task execution must be stored in the private field .output on the Backend abstract class, and is accessible via the get_output() method.

Method lapply()

Run a task on the backend akin to parallel::parLapply().

Usage

AsyncBackend$lapply(x, fun, ...)

Arguments

x

An atomic vector or list to pass to the fun function.

fun

A function to apply to each element of x.

...

Additional arguments to pass to the fun function.

Returns

This method returns void. The output of the task execution must be stored in the private field .output on the Backend abstract class, and is accessible via the get_output() method.

Method apply()

Run a task on the backend akin to parallel::parApply().

Usage

AsyncBackend$apply(x, margin, fun, ...)

Arguments

x

An array to pass to the fun function.

margin

A numeric vector indicating the dimensions of x the fun function should be applied over. For example, for a matrix, margin = 1 indicates applying fun rows-wise, margin = 2 indicates applying fun columns-wise, and margin = c(1, 2) indicates applying fun element-wise. Named dimensions are also possible depending on x. See parallel::parApply() and base::apply() for more details.

fun

A function to apply to x according to the margin.

...

Additional arguments to pass to the fun function.

Returns

This method returns void. The output of the task execution must be stored in the private field .output on the Backend abstract class, and is accessible via the get_output() method.

Method get_output()

Get the output of the task execution.

Usage

AsyncBackend$get_output(wait = FALSE)

Arguments

wait

A logical value indicating whether to wait for the task to finish executing before fetching the results. Defaults to FALSE. See the Details section for more information.

Details

This method fetches the output of the task execution after calling the sapply() method. It returns the output and immediately removes it from the backend. Subsequent calls to this method will throw an error if no additional tasks have been executed in the meantime. This method should be called after the execution of a task.

If wait = TRUE, the method will block the main process until the backend finishes executing the task and the results are available. If wait = FALSE, the method will immediately attempt to fetch the results from the background R session, and throw an error if the task is still running.

Returns

A vector, matrix, or list of the same length as x, containing the results of the fun. The output format differs based on the specific operation employed. Check out the documentation for the apply operations of parallel::parallel for more information.

Method clone()

The objects of this class are cloneable with this method.

Usage

AsyncBackend$clone(deep = FALSE)

Arguments

deep

Whether to make a deep clone.

Examples

# Create a specification object.
specification <- Specification$new()

# Set the number of cores.
specification$set_cores(cores = 2)

# Set the cluster type.
specification$set_type(type = "psock")

# Create an asynchronous backend object.
backend <- AsyncBackend$new()

# Start the cluster on the backend.
backend$start(specification)

# Check if there is anything on the backend.
backend$peek()
#> [[1]]
#> character(0)
#> 
#> [[2]]
#> character(0)
#> 

# Create a dummy variable.
name <- "parabar"

# Export the variable to the backend.
backend$export("name")

# Remove variable from current environment.
rm(name)

# Run an expression on the backend, using the exported variable `name`.
backend$evaluate({
    # Print the name.
    print(paste0("Hello, ", name, "!"))
})
#> [[1]]
#> [1] "Hello, parabar!"
#> 
#> [[2]]
#> [1] "Hello, parabar!"
#> 

# Run a task in parallel (i.e., approx. 2.5 seconds).
backend$sapply(
    x = 1:10,
    fun = function(x) {
        # Sleep a bit.
        Sys.sleep(0.5)

        # Compute something.
        output <- x + 1

        # Return the result.
        return(output)
    }
)

# Right know the main process is free and the task is executing on a `psock`
# cluster started in a background `R` session.

# Trying to get the output immediately will throw an error, indicating that the
# task is still running.
try(backend$get_output())
#> Error : A task is currently running.

# However, we can block the main process and wait for the task to complete
# before fetching the results.
backend$get_output(wait = TRUE)
#>  [1]  2  3  4  5  6  7  8  9 10 11

# Clear the backend.
backend$clear()

# Check that there is nothing on the cluster.
backend$peek()
#> [[1]]
#> character(0)
#> 
#> [[2]]
#> character(0)
#> 

# Stop the backend.
backend$stop()

# Check that the backend is not active.
backend$active
#> [1] FALSE