The goals of jobqueue are to:
- Run jobs in parallel on background processes.
- Allow jobs to be stopped at any point.
- Process job results with asynchronous callbacks.
Installation
# Install the latest stable version from CRAN: install.packages("jobqueue") # Or the development version from GitHub: install.packages("pak") pak::pak("cmmr/jobqueue")
Example
library(jobqueue) jq <- jobqueue() job <- jq$run({ paste('Hello', 'world!') }) job$result #> [1] "Hello world!"
Asynchronous Callbacks
Main articles:
vignette('results') and
vignette('hooks')
job <- jq$run( expr = { 42 }, hooks = list( 'created' = ~{ message("We're uid '", .$uid, "'.") }, '*' = ~{ message(' - ', .$state) })) #> We're uid 'J2'. #> - created #> - submitted #> - queued #> - starting #> - running #> - done job$on('done', ~{ message('result = ', .$result) }) #> result = 42
Converting to Promises
job <- jq$run({ 3.14 }) callback <- function (result) message('resolved with: ', result) job %...>% callback #> resolved with: 3.14 job %>% then(callback) #> resolved with: 3.14 as.promise(job)$then(callback) #> resolved with: 3.14
See also https://rstudio.github.io/promises/
Shiny Integration
function(input, output, session) { output$plot <- renderPlot({ jq$run({ read.table(url) }, list(url = input$url)) %...>% head(input$n) %...>% plot() }) }
See also https://rstudio.github.io/promises/articles/promises_06_shiny.html
Stopping Jobs
When a running job is stopped, the background process for it is terminated. Terminated background process are automatically replaced by new ones.
Stopped jobs will return a condition object of class 'interrupt' as their result.
Main article: vignette('stops')
Manually
job <- jq$run({ Sys.sleep(2); 'Zzzzz' }) job$stop() job$result #> <interrupt: job stopped by user>
A custom message can also be given, e.g. job$stop('my reason'), which
will be returned in the condition object.
Runtime Limits
job <- jq$run({ Sys.sleep(2); 'Zzzzz' }, timeout = 0.2) job$result #> <interrupt: total runtime exceeded 0.2 seconds>
Limits (in seconds) can be set on:
- the total 'submitted' to 'done' time:
timeout = 2 - on a per-state basis:
timeout = list(queued = 1, running = 2) - or both:
timeout = list(total = 3, queued = 2, running = 2)
Stop ID
New jobs will replace existing jobs with the same stop_id.
job1 <- jq$run({ Sys.sleep(1); 'A' }, stop_id = 123) job2 <- jq$run({ 'B' }, stop_id = 123) job1$result #> <superseded: duplicated stop_id> job2$result #> [1] "B"
Copy ID
New jobs will mirror the output of existing jobs with the same copy_id.
job1 <- jq$run({ Sys.sleep(1); 'A' }, copy_id = 456) job2 <- jq$run({ 'B' }, copy_id = 456) job1$result #> [1] "A" job2$result #> [1] "A"
Variables
Main article: vignette('eval')
jq2 <- jobqueue(globals = list(G = 8)) expr <- quote(c(x = x , y = y, G = G)) job <- jq2$run(expr, vars = list(x = 10, y = 2)) dput(job$result) #> c(x = 10, y = 2, G = 8) jq2$stop()