Crew

https://github.com/wlandau/crew
Admin
Crew is an R framework that extends the mirai scheduler with a unifying interface for deploying...
Tokens:5,550
Snippets:40
Trust Score:9.8
Update:6 months ago
Show doc for...
Tokens:
Raw
### Create and Manage a Local Controller

Source: https://context7.com/wlandau/crew/llms.txt

Demonstrates how to initialize a local crew controller with specific worker configurations, submit a task, and retrieve results. It covers the lifecycle from starting the controller to terminating resources.

```r
library(crew)

# Create a controller with 4 workers that idle timeout after 60 seconds
controller <- crew_controller_local(
  name = "my_controller",
  workers = 4,
  seconds_idle = 60,
  seconds_launch = 30,
  tasks_max = Inf,
  reset_globals = TRUE,
  crashes_max = 5L
)

# Start the controller
controller$start()

# Submit a task
controller$push(
  name = "compute_sqrt",
  command = sqrt(x),
  data = list(x = 16)
)

# Wait for all tasks to complete
controller$wait(mode = "all")

# Retrieve the result
result <- controller$pop()
print(result$result[[1]])

# Clean up resources
controller$terminate()
```

--------------------------------

### Custom Launcher Plugin for Unix Nice Priority (R)

Source: https://context7.com/wlandau/crew/llms.txt

Illustrates how to create a custom launcher plugin by extending the `crew_class_launcher` abstract class. This example defines a 'nice_launcher_class' that launches workers using the Unix 'nice' command to set a lower process priority. It includes a helper function `crew_controller_nice` to easily create controllers using this custom launcher.

```r
library(crew)
library(R6)

# Custom launcher for Unix processes with nice priority
nice_launcher_class <- R6::R6Class(
  classname = "nice_launcher_class",
  inherit = crew::crew_class_launcher,
  public = list(
    launch_worker = function(call) {
      bin <- file.path(R.home("bin"), "Rscript")
      processx::process$new(
        command = "nice",
        args = c("-n", "10", bin, self$r_arguments, "-e", call),
        cleanup = FALSE
      )
    }
  )
)

# Helper function to create controller with custom launcher
crew_controller_nice <- function(
  name = NULL,
  workers = 1L,
  seconds_idle = 300,
  ...
) {
  client <- crew::crew_client()
  launcher <- nice_launcher_class$new(
    name = name,
    workers = workers,
    seconds_idle = seconds_idle
  )
  controller <- crew::crew_controller(
    client = client,
    launcher = launcher
  )
  controller$validate()
  controller
}

```

--------------------------------

### TLS Configuration for Secure Worker Connections (R)

Source: https://context7.com/wlandau/crew/llms.txt

Demonstrates how to configure Transport Layer Security (TLS) for encrypted communication between the crew controller and its workers. It shows both automatic TLS setup and custom configuration using user-provided certificates and private keys. This enhances security for distributed or remote worker setups.

```r
library(crew)

# Automatic TLS (recommended for most cases)
tls_auto <- crew_tls(mode = "automatic")

controller <- crew_controller_local(
  workers = 2,
  tls = tls_auto
)

# Custom TLS with your own certificates
tls_custom <- crew_tls(
  mode = "custom",
  key = "/path/to/private_key.pem",
  password = Sys.getenv("TLS_KEY_PASSWORD"),  # Don't hardcode!
  certificates = c("/path/to/certificate.pem", "/path/to/ca_cert.pem")
)

controller_secure <- crew_controller_local(
  workers = 2,
  host = "192.168.1.100",  # Specific network interface
  tls = tls_custom
)

controller_secure$start()
# ... use controller ...
controller_secure$terminate()
```

--------------------------------

### Initialize and execute tasks with crew controller

Source: https://context7.com/wlandau/crew/llms.txt

This snippet demonstrates how to initialize a crew controller with specific worker settings, push a command for asynchronous execution, and retrieve the result. It highlights the basic lifecycle of starting, pushing, waiting, and terminating a controller.

```R
controller <- crew_controller_nice(workers = 2, seconds_idle = 60)
controller$start()
controller$push(command = Sys.getpid())
controller$wait()
print(controller$pop()$result[[1]])
controller$terminate()
```

--------------------------------

### Install Latest GitHub Version of Crew (R)

Source: https://github.com/wlandau/crew/blob/main/CONTRIBUTING.md

This R code snippet demonstrates how to install the latest development version of the 'crew' package directly from GitHub. This is often a prerequisite for reporting bugs to ensure the issue hasn't already been fixed.

```r
remotes::install_github("wlandau/crew")
```

--------------------------------

### Install crew package

Source: https://github.com/wlandau/crew/blob/main/README.md

Commands to install the crew package from various sources including CRAN, GitHub, and R-universe.

```R
install.packages("crew")
remotes::install_github("wlandau/crew")
install.packages("crew", repos = "https://wlandau.r-universe.dev")
```

--------------------------------

### Resource Metrics Logging with Autometric (R)

Source: https://context7.com/wlandau/crew/llms.txt

Configures crew to log CPU and memory usage of its workers using the 'autometric' package. This snippet shows how to set up metrics collection intervals and output paths (e.g., stdout or a file directory), and how to configure local worker logging options like log directory and combining stdout/stderr. It also includes an example of running a memory-intensive task.

```r
library(crew)

# Configure metrics logging
metrics <- crew_options_metrics(
  path = "/dev/stdout",     # Log to stdout (Unix)
  # path = "logs/metrics",  # Or log to directory
  seconds_interval = 5      # Log every 5 seconds
)

# Configure local worker options
local_opts <- crew_options_local(
  log_directory = "logs/workers",  # Worker stdout/stderr logs
  log_join = TRUE                   # Combine stdout and stderr
)

controller <- crew_controller_local(
  workers = 4,
  options_metrics = metrics,
  options_local = local_opts
)

controller$start()

# Run computationally intensive tasks
controller$map(
  command = {
    # Memory-intensive work
    big_data <- matrix(rnorm(1e6), nrow = 1000)
    colMeans(big_data)
  },
  iterate = list(task_id = 1:10)
)

controller$terminate()

# Read and visualize metrics (requires autometric package)
# library(autometric)
# data <- autometric::log_read("logs/metrics")
# autometric::log_plot(data)
```

--------------------------------

### Lint Package Code with lintr (R)

Source: https://github.com/wlandau/crew/blob/main/CONTRIBUTING.md

This R code snippet shows how to use the `lint_package()` function from the 'lintr' package to check code formatting according to the tidyverse style guide. This is a recommended step for ensuring code quality before contributing.

```r
lintr::lint_package()
```

--------------------------------

### Asynchronous Task Execution with walk()

Source: https://context7.com/wlandau/crew/llms.txt

Demonstrates how to submit multiple tasks asynchronously using the walk method. This allows the main R process to continue executing other code while background workers process the iterations.

```R
library(crew)

controller <- crew_controller_local(workers = 4, seconds_idle = 10)
controller$start()

process_item <- function(id, multiplier) {
  Sys.sleep(0.2)
  id * multiplier
}

controller$walk(
  command = process_item(id, multiplier),
  iterate = list(id = 1:10),
  data = list(process_item = process_item, multiplier = 5),
  verbose = TRUE
)

print("Tasks submitted, doing other work...")
controller$wait(mode = "all")
results <- controller$collect(error = "stop")
print(as.numeric(results$result))
controller$terminate()
```

--------------------------------

### Submit Asynchronous Tasks with push()

Source: https://context7.com/wlandau/crew/llms.txt

Explains how to submit individual tasks with custom data, global variables, and package dependencies. It also details the structure of the returned task object, including status, results, and execution metadata.

```r
library(crew)

controller <- crew_controller_local(workers = 2, seconds_idle = 10)
controller$start()

# Push a task with data and globals
my_function <- function(x, y) x + y

controller$push(
  name = "addition_task",
  command = my_function(a, b),
  data = list(a = 10, b = 20),
  globals = list(my_function = my_function),
  packages = c("dplyr"),
  seed = 123,
  algorithm = "Mersenne-Twister",
  seconds_timeout = 300,
  scale = TRUE,
  throttle = TRUE
)

# Wait and collect result
controller$wait(mode = "all")
task <- controller$pop()

controller$terminate()
```

--------------------------------

### POST /controller/local

Source: https://context7.com/wlandau/crew/llms.txt

Initializes a local controller to manage worker processes for asynchronous task execution.

```APIDOC
## POST /controller/local

### Description
Creates and starts a local controller object to manage a pool of R workers for distributed task processing.

### Method
POST

### Endpoint
crew_controller_local()

### Parameters
#### Request Body
- **name** (string) - Required - Unique identifier for the controller.
- **workers** (integer) - Required - Number of worker processes to maintain.
- **seconds_idle** (numeric) - Optional - Time in seconds before an idle worker exits.
- **seconds_launch** (numeric) - Optional - Max wait time for worker initialization.
- **tasks_max** (numeric) - Optional - Max tasks per worker before recycling.

### Request Example
{
  "name": "my_controller",
  "workers": 4,
  "seconds_idle": 60
}

### Response
#### Success Response (200)
- **controller** (object) - The initialized controller instance.

#### Response Example
{
  "status": "started",
  "controller": "my_controller"
}
```

--------------------------------

### Implementing Backup Controllers for Crash Recovery

Source: https://context7.com/wlandau/crew/llms.txt

Configures a primary controller with a backup controller to handle task retries automatically. If a task fails on the primary, it can be routed to a higher-resource backup.

```R
library(crew)

high_memory <- crew_controller_local(name = "high_memory", workers = 1, crashes_max = 2L)
default <- crew_controller_local(name = "default", workers = 4, crashes_max = 3L, backup = high_memory)

group <- crew_controller_group(default, high_memory)
group$start()

group$push(name = "risky_task", command = { result <- heavy_computation(); result }, data = list(heavy_computation = heavy_computation))

group$wait(mode = "all")
result <- group$pop()
print(result$controller)
group$terminate()
```

--------------------------------

### Coordinating Controller Groups

Source: https://context7.com/wlandau/crew/llms.txt

Shows how to group multiple controllers with different configurations to route tasks to specific worker pools. This is useful for managing heterogeneous workloads.

```R
library(crew)

fast_controller <- crew_controller_local(name = "fast", workers = 8, seconds_idle = 5)
slow_controller <- crew_controller_local(name = "slow", workers = 2, seconds_idle = 60, tasks_max = 1L)

group <- crew_controller_group(fast_controller, slow_controller)
group$start()

group$push(name = "quick_task", command = sqrt(4), controller = "fast")
group$push(name = "heavy_task", command = { Sys.sleep(2); mean(1:1000) }, controller = "slow")

group$wait(mode = "all")
result <- group$pop()
print(result$controller)
group$summary()
group$terminate()
```

--------------------------------

### POST /controller/map

Source: https://context7.com/wlandau/crew/llms.txt

Applies a command to multiple inputs synchronously and collects results.

```APIDOC
## POST /controller/map

### Description
Executes a command across a list of inputs in parallel and returns a tibble containing the results and status of each task.

### Method
POST

### Endpoint
controller$map()

### Parameters
#### Request Body
- **command** (expression) - Required - The R code to execute.
- **iterate** (list) - Required - Named list of inputs to map over.
- **data** (list) - Optional - Constant data across all tasks.

### Request Example
{
  "command": "x^2",
  "iterate": {"x": [1, 2, 3]}
}

### Response
#### Success Response (200)
- **results** (tibble) - A table containing the result, status, and metadata for each input.

#### Response Example
{
  "results": [{"input": 1, "result": 1}, {"input": 2, "result": 4}, {"input": 3, "result": 9}]
}
```

--------------------------------

### Monitoring Local Crew Processes (R)

Source: https://context7.com/wlandau/crew/llms.txt

Provides functionality to monitor and manage local crew processes, including dispatchers, workers, and daemons. It demonstrates how to list active processes by their Process IDs (PIDs) and how to terminate specific processes using the monitor object. This is useful for debugging and managing the lifecycle of crew components.

```r
library(crew)

# Create a monitor for local processes
monitor <- crew_monitor_local()

# List all local mirai dispatcher processes
dispatchers <- monitor$dispatchers()
print(dispatchers)
#> [1] 12345 12346

# List all local crew worker processes
workers <- monitor$workers()
print(workers)
#> [1] 12350 12351 12352 12353

# List mirai daemon processes
daemons <- monitor$daemons()
print(daemons)
#> integer(0)

# Terminate specific processes by PID
if (length(workers) > 0) {
  monitor$terminate(pids = workers[1])
}

# Verify termination
monitor$workers()
```

--------------------------------

### POST /controller/push

Source: https://context7.com/wlandau/crew/llms.txt

Submits an individual task to the controller for asynchronous execution.

```APIDOC
## POST /controller/push

### Description
Submits a specific command to the worker pool for execution, including data, global variables, and package dependencies.

### Method
POST

### Endpoint
controller$push()

### Parameters
#### Request Body
- **name** (string) - Required - Task identifier.
- **command** (expression) - Required - The R code to execute.
- **data** (list) - Optional - Data environment for the task.
- **globals** (list) - Optional - Global variables required by the command.
- **packages** (vector) - Optional - Packages to load on the worker.

### Request Example
{
  "name": "addition_task",
  "command": "a + b",
  "data": {"a": 10, "b": 20}
}

### Response
#### Success Response (200)
- **task_id** (string) - Unique task identifier.

#### Response Example
{
  "status": "queued",
  "task_id": "addition_task"
}
```

--------------------------------

### Shiny with Promises for Responsive UI (R)

Source: https://context7.com/wlandau/crew/llms.txt

Integrates promises with Shiny to enable immediate UI updates while background tasks are processed. It uses the crew package to manage parallel task execution and collect results asynchronously, preventing the UI from freezing. Dependencies include 'promises', 'shiny', and 'crew'.

```r
library(promises)
library(shiny)
library(crew)

process_data <- function(x) {
  Sys.sleep(1)
  x * 2
}

ui <- fluidPage(
  actionButton("process", "Process Items"),
  textOutput("results")
)

server <- function(input, output, session) {
  controller <- crew_controller_local(workers = 4, seconds_idle = 10)
  controller$start()
  controller$autoscale()  # Enable background auto-scaling
  onStop(function() controller$terminate())

  totals <- reactiveValues(sum = 0, count = 0)

  collect_result <- function(ignore, controller, totals) {
    results <- controller$collect(error = "stop")
    if (!is.null(results)) {
      totals$sum <- totals$sum + sum(as.numeric(results$result))
      totals$count <- totals$count + nrow(results)
    }
  }

  observeEvent(input$process, {
    # Push tasks and chain promise for result collection
    for (i in 1:10) {
      controller$push(
        command = process_data(x),
        data = list(process_data = process_data, x = i)
      ) %...>% collect_result(controller, totals)
    }
  })

  output$results <- renderText({
    sprintf("Processed %d items, total: %d", totals$count, totals$sum)
  })
}

shinyApp(ui, server)
```

--------------------------------

### Check Package Health with devtools and goodpractice (R)

Source: https://github.com/wlandau/crew/blob/main/CONTRIBUTING.md

This R code snippet demonstrates how to run comprehensive package checks using `devtools::check()` and `goodpractice::gp()`. These functions help ensure the package is well-structured, follows best practices, and is ready for release.

```r
devtools::check()
goodpractice::gp()
```

--------------------------------

### Perform Synchronous Functional Mapping with map()

Source: https://context7.com/wlandau/crew/llms.txt

Shows how to apply a command to multiple inputs simultaneously using the map function. This method returns a tibble containing results and metadata for all tasks, simplifying batch processing.

```r
library(crew)

controller <- crew_controller_local(workers = 4, seconds_idle = 10)
controller$start()

# Define a computation function
slow_square <- function(x) {
  Sys.sleep(0.5)
  x^2
}

# Map over multiple inputs
results <- controller$map(
  command = slow_square(value) + offset,
  iterate = list(
    value = c(1, 2, 3, 4, 5)
  ),
  data = list(
    slow_square = slow_square
  ),
  globals = list(
    offset = 100
  ),
  names = "value",
  error = "stop",
  verbose = TRUE
)

print(results)
```

--------------------------------

### Calculate Package Code Coverage with covr (R)

Source: https://github.com/wlandau/crew/blob/main/CONTRIBUTING.md

This R code snippet illustrates how to calculate the code coverage of the package using the `covr::package_coverage()` function. High code coverage is important for ensuring that new or changed functionality is adequately tested.

```r
covr::package_coverage()
```

--------------------------------

### Cite the crew package in R

Source: https://github.com/wlandau/crew/blob/main/README.md

Provides the standard R command to retrieve citation information for the crew package and the corresponding BibTeX entry for LaTeX documentation.

```r
citation("crew")

# BibTeX entry:
# @Manual{,
#   title = {crew: A Distributed Worker Launcher Framework},
#   author = {William Michael Landau},
#   year = {2023},
#   note = {https://wlandau.github.io/crew/, https://github.com/wlandau/crew},
# }
```

--------------------------------

### Integrating Crew with Shiny Applications

Source: https://context7.com/wlandau/crew/llms.txt

Integrates crew into a Shiny application to perform non-blocking background computations. It uses reactive polling to update the UI once background tasks complete.

```R
library(shiny)
library(crew)

simulate_data <- function(n) { Sys.sleep(2); rnorm(n, mean = 100, sd = 15) }

ui <- fluidPage(actionButton("run", "Run Simulation"), textOutput("status"), plotOutput("histogram"))

server <- function(input, output, session) {
  controller <- crew_controller_local(workers = 4, seconds_idle = 30)
  controller$start()
  onStop(function() controller$terminate())
  
  sim_results <- reactiveVal(NULL)
  observeEvent(input$run, {
    controller$push(name = paste0("sim_", Sys.time()), command = simulate_data(n), data = list(simulate_data = simulate_data, n = 1000))
  })
  
  observe({
    invalidateLater(500)
    result <- controller$pop()
    if (!is.null(result) && result$status == "success") { sim_results(result$result[[1]]) }
  })
  
  output$status <- renderText({ invalidateLater(500); paste("Pending tasks:", controller$unresolved()) })
  output$histogram <- renderPlot({ req(sim_results()); hist(sim_results(), main = "Simulation Results", col = "steelblue") })
}

shinyApp(ui, server)
```
Tokens:
Raw