Future error when concurrent via plumber API

[afoiniotis001]

Our algorithm approximately wants 5 minutes to be executed, due to time consuming calculations and heavy load of data I use multiple futures to execute the code in parallel.
The plan strategy I use is plan(multisession). We have a plumber API endpoint and when we initiate concurrent calls to the API and hence to the algorithm I receive the following error :

FutureError: Cannot resolve MultisessionFuture (), because the connection to the worker is corrupt: Connection (connection: index=4, description="<-localhost:11803", class="sockconn", mode="a+b", text="binary", opened="opened", can read="yes", can write="yes", id=3711, raw_id="<pointer: 0xe7f>") is no longer valid.

After a quick search I found out that R assigns the same workers to the new API call, how can we avoid the worker corruption

[HenrikBengtsson]

What's your sessionInfo() after you get that error?

Is that the full error message? I'd expect another sentence to follow "... is no longer valid." If that's all, the make sure to update all your packages, especially parallelly and future because there should be more info in that error message.

[afoiniotis001]

We run the code in a container on Azure web service, therefore I cannot run the sessioninfo(). Here is the full message I received, it's very important for me to make the code run in multiple threads. As you understand we anticipate concurrent calls on the plumber API and each call splits the execution into 8 workers (via futures) which consume all CPU for a couple of minutes. Looking forward for your suggestion on how to make this work 2021-06-02T20:02:09.059690192Z <FutureError: Cannot resolve MultisessionFuture (<none>), because the connection to the worker is corrupt: Connection (connection: index=7, description="<-localhost:11945", class="sockconn", mode="a+b", text="binary", opened="opened", can read="yes", can write="yes", id=2078, raw_id="<pointer: 0x81e>") is no longer valid. It differ from the currently registered R connection with the same index 7 (connection: index=7, description="<-localhost:11945", class="sockconn", mode="a+b", text="binary", opened="opened", can read="yes", can write="yes", id=2720, raw_id="<pointer: 0xaa0>")> 2021-06-02T20:02:09.059789293Z 2021-06-02T20:02:09.059795393Z 2021-06-02T20:02:09.059799493Z DEBUG: BEGIN TROUBLESHOOTING HELP 2021-06-02T20:02:09.059803493Z Future involved: 2021-06-02T20:02:09.060093495Z function (expr, envir = parent.frame(), substitute = TRUE, lazy = FALSE, 2021-06-02T20:02:09.060756702Z seed = FALSE, globals = TRUE, packages = NULL, label = NULL, 2021-06-02T20:02:09.060849602Z gc = FALSE, ...) 2021-06-02T20:02:09.060858103Z { 2021-06-02T20:02:09.060987004Z if (substitute) 2021-06-02T20:02:09.061047704Z expr <- substitute(expr) 2021-06-02T20:02:09.061054404Z makeFuture <- plan("next") 2021-06-02T20:02:09.061059004Z future <- makeFuture(expr, substitute = FALSE, envir = envir, 2021-06-02T20:02:09.061063704Z lazy = lazy, seed = seed, globals = globals, packages = packages, 2021-06-02T20:02:09.061067904Z label = label, gc = gc, ...) 2021-06-02T20:02:09.061071804Z if (!inherits(future, "Future")) { 2021-06-02T20:02:09.061076005Z stop(FutureError("plan(\"next\") returned a function that does not return a Future object: ", 2021-06-02T20:02:09.061080505Z paste(sQuote(class(future)), collapse = ", "))) 2021-06-02T20:02:09.061084405Z } 2021-06-02T20:02:09.061088205Z future 2021-06-02T20:02:09.061091905Z } 2021-06-02T20:02:09.061103505Z <bytecode: 0x56370aa46660> 2021-06-02T20:02:09.061107805Z <environment: namespace:future> 2021-06-02T20:02:09.061111805Z attr(,"untweakable") 2021-06-02T20:02:09.061703710Z [1] "asynchronous" "conditions" "envir" "expr" "globals" 2021-06-02T20:02:09.061720110Z [6] "lazy" "local" "packages" "seed" "stdout" 2021-06-02T20:02:09.061726111Z [11] "substitute" "version" 2021-06-02T20:02:09.061730811Z attr(,"tweakable") 2021-06-02T20:02:09.061735211Z [1] "earlySignal" "split" 2021-06-02T20:02:09.061739911Z 2021-06-02T20:02:09.061743711Z DEBUG: END TROUBLESHOOTING HELP

[HenrikBengtsson]

FutureError: Cannot resolve MultisessionFuture (<none>), because the connection to the worker is corrupt: Connection (connection: index=7, description="<-localhost:11945", class="sockconn", mode="a+b", text="binary", opened="opened", can read="yes", can write="yes", id=2078, raw_id="<pointer: 0x81e>") is no longer valid. It differ from the currently registered R connection with the same index 7 (connection: index=7, description="<-localhost:11945", class="sockconn", mode="a+b", text="binary", opened="opened", can read="yes", can write="yes", id=2720, raw_id="<pointer: 0xaa0>")

With multisession, you get R workers that are so called PSOCK cluster nodes. Your main R session communicates with those using socket connections. This error message says that when your main R session tried to send a future to one of the workers, the socket connection with index 7 has been replaced by a new connection with the same index.

Without seeing your code (it would make it easier if you could share), I can only guess why this happens. It might happen because you terminate the workers (e.g. closeAllConnections() but also by other means), and create new ones. It could be some other code that you use that end up doing it. Regardless where it happens, it's a bug if done.

We run the code in a container on Azure web service, therefore I cannot run the sessioninfo().

I assume that you run R (>= 4.0.0). If so, add:

globalCallingHandlers(FutureError = function(e) { print(sessionInfo() })

on top of your script, and R should print the session info whenever there's a FutureError. (Assuming that is not caught and muffled elsewhere in the process).

PS. Please use the GitHub web interface when commenting; if you reply by email, we end up with odd newline wraps making it complicated to read. (I've had to reformat your error dump to make it readable).