Extended R dispatcher

[wlandau]

@shikokuchuo, I finally had a free moment to try the extended R dispatcher you sketched in https://github.com/shikokuchuo/mirai/tree/dispatcher. Really promising stuff! Just writing in to share my initial thoughts and requests.

Data

The crew launcher is an R6 object. Is there a way to send this object to the ext argument along with the function and its call? I tried to ship data in the closure of the function, but in the example below, f_data does not go along for the ride.

library(mirai)
f_data <- list2env(list(index = 0L), parent = emptyenv())
f <- function(envir = parent.frame()) {
  force(envir)
  tryCatch(
    f_data$index <- f_data$index + 1L,
    error = \(condition) writeLines(conditionMessage(condition), "error.txt")
  )
  saveRDS(as.list(envir), file.path("envir.rds"))
}
daemons(n = 2L, url = "ws://127.0.0.1:5700", ext = list(f, expression(f())))
task <- mirai(1)
daemons(0L)

Launching

crew uses information about the current task backlog to decide if it should launch workers, as well as how many workers to launch. I think the extended R dispatcher is a great opportunity to simplify the logic of this. Is there a way to get the number of tasks that have not yet been dispatched? Also, the existing serverfree vector looks useful. Is it FALSE if the server is currently running a task and TRUE otherwise?

I ran some simple code to see my options:

library(mirai)
f_data <- list2env(list(index = 0L), parent = emptyenv())
f <- function(envir = parent.frame()) {
  force(envir)
  saveRDS(as.list(envir), "envir.rds")
}
daemons(n = 2L, url = "ws://127.0.0.1:5700", ext = list(f, expression(f())))
urls <- nextget("urls")
lapply(urls, \(url) launch_local(url = url))
replicate(5L, mirai("task"))
daemons(0L)
str(readRDS("envir.rds")

Output below:

List of 48
 $ free       : int(0) 
 $ changes    : logi [1:2] FALSE FALSE
 $ activevec  : int [1:2] 0 0
 $ cv_values  : int [1:2] 0 0
 $ cmessage   :Class 'recvAio' <environment: 0x12cbc6828> 
 $ sockc      :Classes 'nanoSocket', 'nano' <externalptr> 
  ..- attr(*, "id")= int 4
  ..- attr(*, "state")= chr "opened"
  ..- attr(*, "protocol")= chr "rep"
  ..- attr(*, "dialer")=List of 1
  .. ..$ :Classes 'nanoDialer', 'nano' <externalptr> 
  .. .. ..- attr(*, "id")= int 2
  .. .. ..- attr(*, "url")= chr "ipc:///tmp/9848fd9af4e9f66881f6f44ac"
  .. .. ..- attr(*, "state")= chr "started"
  .. .. ..- attr(*, "socket")= int 4
 $ ctrchannel : logi TRUE
 $ req        :Class 'recvAio' <environment: 0x12cbc5708> 
 $ ctx        :<externalptr> 
 $ listener   :Classes 'nanoListener', 'nano' <externalptr> 
  ..- attr(*, "id")= int 2
  ..- attr(*, "url")= chr "ws://127.0.0.1:5700/2"
  ..- attr(*, "state")= chr "started"
  ..- attr(*, "socket")= int 3
 $ ncv        :Class 'conditionVariable' <externalptr> 
 $ nsock      :Classes 'nanoSocket', 'nano' <externalptr> 
  ..- attr(*, "id")= int 3
  ..- attr(*, "state")= chr "opened"
  ..- attr(*, "protocol")= chr "req"
  ..- attr(*, "listener")=List of 1
  .. ..$ :Classes 'nanoListener', 'nano' <externalptr> 
  .. .. ..- attr(*, "id")= int 2
  .. .. ..- attr(*, "url")= chr "ws://127.0.0.1:5700/2"
  .. .. ..- attr(*, "state")= chr "started"
  .. .. ..- attr(*, "socket")= int 3
 $ nurl       : chr "ws://127.0.0.1:5700/2"
 $ burl       : chr "ws://127.0.0.1:5700/2"
 $ i          : int 2
 $ envir      :<environment: 0x12cbc1270> 
 $ ports      : NULL
 $ baseurl    : Named chr [1:10] "ws://127.0.0.1:5700" "ws" "" "127.0.0.1:5700" ...
  ..- attr(*, "names")= chr [1:10] "rawurl" "scheme" "userinfo" "host" ...
 $ active     :List of 2
  ..$ :Class 'conditionVariable' <externalptr> 
  ..$ :Class 'conditionVariable' <externalptr> 
 $ servers    :List of 2
  ..$ :Classes 'nanoSocket', 'nano' <externalptr> 
  .. ..- attr(*, "id")= int 2
  .. ..- attr(*, "state")= chr "opened"
  .. ..- attr(*, "protocol")= chr "req"
  .. ..- attr(*, "listener")=List of 1
  .. .. ..$ :Classes 'nanoListener', 'nano' <externalptr> 
  .. .. .. ..- attr(*, "id")= int 1
  .. .. .. ..- attr(*, "url")= chr "ws://127.0.0.1:5700/1"
  .. .. .. ..- attr(*, "state")= chr "started"
  .. .. .. ..- attr(*, "socket")= int 2
  ..$ :Classes 'nanoSocket', 'nano' <externalptr> 
  .. ..- attr(*, "id")= int 3
  .. ..- attr(*, "state")= chr "opened"
  .. ..- attr(*, "protocol")= chr "req"
  .. ..- attr(*, "listener")=List of 1
  .. .. ..$ :Classes 'nanoListener', 'nano' <externalptr> 
  .. .. .. ..- attr(*, "id")= int 2
  .. .. .. ..- attr(*, "url")= chr "ws://127.0.0.1:5700/2"
  .. .. .. ..- attr(*, "state")= chr "started"
  .. .. .. ..- attr(*, "socket")= int 3
 $ queue      :List of 2
  ..$ :List of 2
  .. ..$ ctx:<externalptr> 
  .. ..$ req:Class 'recvAio' <environment: 0x12cb72700> 
  ..$ :List of 2
  .. ..$ ctx:<externalptr> 
  .. ..$ req:Class 'recvAio' <environment: 0x12cbc5708> 
 $ serverfree : logi [1:2] TRUE TRUE
 $ activestore: int [1:2] 0 0
 $ instance   : int [1:2] 0 0
 $ complete   : int [1:2] 0 0
 $ assigned   : int [1:2] 0 0
 $ basenames  : chr [1:2] "ws://127.0.0.1:5700/1" "ws://127.0.0.1:5700/2"
 $ servernames: chr [1:2] "ws://127.0.0.1:5700/1" "ws://127.0.0.1:5700/2"
 $ seq_n      : int [1:2] 1 2
 $ vectorised : logi FALSE
 $ auto       : logi FALSE
 $ cv         :Class 'conditionVariable' <externalptr> 
 $ sock       :Classes 'nanoSocket', 'nano' <externalptr> 
  ..- attr(*, "id")= int 1
  ..- attr(*, "state")= chr "opened"
  ..- attr(*, "protocol")= chr "rep"
  ..- attr(*, "dialer")=List of 1
  .. ..$ :Classes 'nanoDialer', 'nano' <externalptr> 
  .. .. ..- attr(*, "id")= int 1
  .. .. ..- attr(*, "url")= chr "ipc:///tmp/9848fd9af4e9f66881f6f44a"
  .. .. ..- attr(*, "state")= chr "started"
  .. .. ..- attr(*, "socket")= int 1
 $ usecall    : logi FALSE
 $ usefunc    : logi TRUE
 $ call       :  expression(f())
 $ func       :function (envir = parent.frame())  
 $ host       : chr "ipc:///tmp/9848fd9af4e9f66881f6f44a"
 $ url        : chr "ws://127.0.0.1:5700"
 $ n          : int 2
 $ asyncdial  : logi FALSE
 $ token      : logi FALSE
 $ lock       : logi FALSE
 $ tls        : NULL
 $ pass       : NULL
 $ monitor    : chr "ipc:///tmp/9848fd9af4e9f66881f6f44ac"
 $ rs         : NULL
 $ ext        :List of 2
  ..$ :function (envir = parent.frame())  
  ..$ :  expression(f())

Throttling

For performance, I would like to avoid calling the crew launcher every time the dispatcher wakes up. I was thinking of using later to implement a custom throttling mechanism:

1. The first time the dispatcher wakes up, use later to schedule the crew launcher to run e.g. 0.25 seconds later.
2. If the dispatcher wakes up before those 0.25 seconds elapse, just go back to sleep.
3. Let the later main loop run the crew launcher at the 0.25-second mark.
4. Repeat (1) the next time the dispatcher wakes up.

I haven't experimented with this part just yet, but I was wondering if you see any pitfalls at this stage.

[wlandau]

Re throttling: I had hoped the example below would write a new file every second, but I see no files being written.

library(mirai)
library(nanonext)

f <- function() {
  if (!is.finite(later::next_op_secs())) {
    later::later(
      func = \() writeLines(as.character(Sys.time()), basename(tempfile())),
      delay = 1
    )
  }
}

daemons(n = 2L, url = "ws://127.0.0.1:5700", ext = list(f, expression(f())))

now <- mclock()
while (mclock() - now < 3e4) {
  task <- mirai("task")
  msleep(10)
}

daemons(0L)

From the later README:

To avoid bugs due to reentrancy, by default, scheduled operations only run when there is no other R code present on the execution stack; i.e., when R is sitting at the top-level prompt.

That's probably the reason.

But maybe I can live without throttling if the launcher logic is fast enough. The original motivation was that in order to know how many new workers to launch, crew had to loop over all the tasks and check nanonext::.unresolved(). If instead crew can look up the number of undispatched tasks and use existing vectors like online and serverfree. Wouldn't suspect too much slowness from serverfree & (online | (launching & !instance)) even if those vectors are thousands of elements long. (launching is a vector crew creates itself based on the startup time.)

[wlandau]

One thing I forgot about: rotation. How would you recommend rotating a server listener from inside the dispatcher? E.g. a replacement for saisei() to be invoked from the function in ext.

[shikokuchuo]

A couple of things:

• Currently 'ext' takes a list of a function and/or call. It's intended you pick either of depending on what's easier to use. You don't need f and expression(f) - you can specify one as NULL.
• Is it possible to define the R6 object inside your function (not in the closure)? I'll need to take a closer look if there's a reason this doesn't work.
• Running later on dispatcher should not be necessary. Also as we're waiting it'll never run...
• (activevec & serverfree) I think that gets you the available workers i.e. connected and free. There should be enough logic not to call unresolved().
• To rotate listeners, perhaps copy these lines of code for now

  mirai/R/mirai.R

  Lines 326 to 330 in 7640107

  	reap(attr(servers[[i]], "listener")[[1L]])
  	attr(servers[[i]], "listener") <- NULL
  	data <- servernames[i] <- if (auto) auto_tokenized_url() else new_tokenized_url(basenames[i])
  	instance[i] <- -abs(instance[i])
  	listen(servers[[i]], url = data, tls = tls, error = TRUE)

  . This is a good point - eventually I will need to make an interface like nextget() for all the functions you call, also possibly the other variables so that this can remain stable.

[shikokuchuo]

I hope this isn't too discouraging, I know you already put work into https://github.com/shikokuchuo/mirai/tree/dispatcher. I wish I had thought of #78 (reply in thread) sooner.

Not at all discouraging, to be honest I haven't thought through the point about the R6 object needing to be passed through to dispatcher yet. If it proves too difficult, we can always try to jump straight to threaded dispatcher.

[wlandau]

Interesting case. I think the solution here is to use a later poll in crew on the host process, not from dispatcher. So it will keep checking until the connection happens - if it doesn't then it can trigger dispatcher to run again etc. The key is that you would need to trust later to run in the background - R has to be at the top level prompt for this to happen.

If only later could work inside functions and had the power to suspend the main event loop to run a quick task. Then the host process would be perfect and ext would have been moot from the beginning. Unfortunately, later conflicts with the major non-interactive use cases of crew, including targets.

Not at all discouraging, to be honest I haven't thought through the point about the R6 object needing to be passed through to dispatcher yet. If it proves too difficult, we can always try to jump straight to threaded dispatcher.

That's good to know. It's starting to look really challenging:

1. Special task-level condition variables on the dispatcher.
2. R6 objects in ext.
3. saisei() on the dispatcher.
4. How to detect crashes that happen prior to connecting.

All this makes me think the threaded dispatcher would be much more feasible. It's a challenge in and of itself, but at least crew is far less likely to add additional burdens like the 4 above.

[wlandau]

By the way, I started using nextget("cv") in crew, and it simplified so much that the major challenge was deprecating arguments and methods! crew is simpler than it has ever been, and design elements are falling into place so well. Which now makes me hesitate even more when it comes to using ext in crew.

[shikokuchuo]

Wow your crew backlog tests seem to run a lot faster!

[wlandau]

Yeah, I am really enjoying that!

Also, an update: I don't think I can entirely avoid .unresolved() just because of how controller$wait(mode = "one") works, but I no longer need it for auto-scaling.