make_cluster(): subsetting with [ not working

[HenrikBengtsson]

Low-level issue

I attempted:

cl <- mirai::make_cluster(2)
cl_kk <- cl[1]
print(cl_kk)
#> Error in ..[[attr(x, "id")]] : 
#>   wrong arguments for subsetting an environment

Looking at:

> str(cl_kk)
List of 1
 $ :Class 'miraiNode' <environment: 0x55e7611e18b8> 
  ..- attr(*, "node")= int 1
  ..- attr(*, "id")= chr "`11`"
 - attr(*, "class")= chr [1:2] "miraiCluster" "cluster"

I see that cl_kk lost the id attribute. To fix this, add:

`[.miraiCluster` <- function (cl, ...) {
    v <- NextMethod()
    attr(v, "id") <- attr(cl, "id")
    v
}

With this, the following works:

> cl[1]
< miraiCluster >
 - cluster ID: `1`
 - nodes: 1
 - active: TRUE

> str(cl[1])
List of 1
 $ :Class 'miraiNode' <environment: 0x563cf247df10> 
  ..- attr(*, "node")= int 1
  ..- attr(*, "id")= chr "`1`"
 - attr(*, "class")= chr [1:2] "miraiCluster" "cluster"
 - attr(*, "id")= chr "`1`"

> cl[2]
< miraiCluster >
 - cluster ID: `1`
 - nodes: 1
 - active: TRUE

> str(cl[2])
List of 1
 $ :Class 'miraiNode' <environment: 0x563cf247d998> 
  ..- attr(*, "node")= int 2
  ..- attr(*, "id")= chr "`1`"
 - attr(*, "class")= chr [1:2] "miraiCluster" "cluster"
 - attr(*, "id")= chr "`1`"

The bigger picture

I spotted the above while trying to add support for plan(cluster, workers = cl) with cl <- mirai::make_cluster(2) to future.

Part of the generic implementation of cluster futures is that it, regardless of actual implementation, communicates with a specific cluster node (cl_kk <- cl[kk]) multiple times. The gist is (it's done differently):

1. Export globals using clusterExport(cl_kk, globals)
2. Evaluate future expression using clusterEvalQ(cl_kk, <expr>)

While doing this, I realized that although I use the same cl_kk in both steps, they do not necessarily reach the same cluster node (R process). Here's a minimal reproducible example using just mirai:

cl <- mirai::make_cluster(2)

for (kk in seq_along(cl)) {
  message(sprintf("Subcluster with node %d ...", kk))
  pids <- sapply(1:10, FUN = function(void) {
    unlist(parallel::clusterEvalQ(cl[kk], Sys.getpid()))
  })
  print(pids)
}

If you run this a few times, you'll see that despite using the same cl[kk], clusterEvalQ() ends up running on different R processes;

Subcluster with node 1 ...
 [1] 442421 442421 442416 442421 442416 442421 442416 442421 442416 442421
Subcluster with node 2 ...
 [1] 442416 442421 442416 442421 442416 442421 442416 442421 442416 442421

I think this behavior does not meet the expectations on how parallel cluster:s should work. Of course, a first, conservative approach would set [.miraiCluster to give an error saying miraiCluster does not support subsetting, but I think the real goal should be to support being able to hit the same R process when using a single cluster node.

[shikokuchuo]

Hi Henrik, thanks for spotting this.

I would not otherwise have encountered [.cluster in the parallel package. I will look to add support for subsetting, probably similar to what you propose.

As for hitting the same node, I am not currently aware that this is a feature of clusters - functions such as clusterEvalQ() or clusterExport() seem to apply to a whole cluster. I am mentioning this as my understanding is that a parallel backend must support the public interface e.g. user calling parLapply() must get back what is expected. However, as to how that is actually implemented, there cannot be such strict stipulations, otherwise if things can only be done one way (one node = one socket connection for example), then we effectively have copies of the same thing, and the whole exercise of supporting different backends becomes not so meaningful. One immediate consequence of this is that out of the box, without any overt attempt to optimise on my behalf, the 'mirai' clusters already outperform base R according to my benchmarks.

However, the current state of affairs represents only the initial implementation, so it is possible that what you describe becomes possible down the line. And there are multiple ways to achieve this theoretically. In the meantime, one alternative I thought might work for you, is to do everything in one clusterEvalQ() call so as to avoid having to re-hit the same node again.

At the end of the day, I am thinking that it is probably best to focus on future.mirai - an adaptation of an adaptation, something is bound to get lost in translation! I still need to find time to focus on that - but if you have any questions or requests on that front please do feel free to let me know - it would be great to get that across the line.

[shikokuchuo]

Have taken the 'conservative approach' for now in b0f2ac9.

[HenrikBengtsson]

The way I interpret the parallel package and its documentation is that cluster:s have unique nodes and the different apply functions operate over the nodes in a deterministic order. Without this, for instance, clusterSetRNGStream() would not work; it won't be able to set up RNG streams such that following apply functions would return numerically reproducible results.

[shikokuchuo]

Yes thanks I am aware of that. The current behaviour obeys this as for clusters at the underlying level all nodes connect to a single NNG socket. In this case tasks are distributed in a deterministic fashion (round-robin).

[shikokuchuo]

Sorry, I made the change without running the manual tests. Turns out the [ method is actually used internally by parallel, which makes a lot of sense now (I was wondering why the method was implemented). So I've reverted, and instead fixed the print method so it doesn't produce the error you encountered. This of course doesn't alter the underlying behaviour.

[shikokuchuo]

Hi Henrik, coming back to this. I must have missed your opening line here and just jumped to the detail:

I spotted the above while trying to add support for plan(cluster, workers = cl) with cl <- mirai::make_cluster(2) to future.

I think that makes perfect sense to enable miraiCluster as a type of cluster future. I am missing at the moment the reason why connecting to a specific node is needed in this case.

I just tried something like:

library(mirai)
library(parallel)
cl <- make_cluster(2)
setDefaultCluster(cl)
getDefaultCluster()
#> < miraiCluster >
#> - cluster ID: `0`
#> - nodes: 2
#> - active: TRUE

library(future)
plan("cluster")
f <- future({
     a <- 7
     b <- 3
     c <- 2
     a * b * c
     })
value(f)
#> [1] 42

Do let me know what you think.

[HenrikBengtsson]

Sorry for the radio silence. To clarify, I posted this independently of Futureverse and plan(cluster) etc.

As for hitting the same node, I am not currently aware that this is a feature of clusters - functions such as clusterEvalQ() or clusterExport() seem to apply to a whole cluster.

Give that cluster object can be subsetted, e.g.

> cl <- parallel::makeCluster(3)
> print(cl)
socket cluster with 3 nodes on host 'localhost'

> cl2 <- cl[c(1,3)]
> print(cl2)
socket cluster with 2 nodes on host 'localhost'

I think the assumption that you cannot operating on parts of a cluster needs to be confirmed.

I see this differently, so my concern is that the cluster returned by mirai::make_cluster() does not fulfil the expectations/requirements for what a cluster object of the parallel package should provide. I think the culprit is that the parallel API for cluster objects rely on the concept of specific node:s, and that they can be indexed.

If you'd like to rely on this assumption, my recommendation would be to have [ for miraiCluster produce an informative error, e.g.

`[.miraiCluster` <- function(cl, ...) {
  stop("A ", class(cl)[1], " object cannot be subsetted")
}

That would protect anyone attempting to use a miraiCluster in a way it's not designed. It would give an early error, rather than a delayed error, for any code out there that rely on being able to operator on a subset of a cluster object. Issue #93 is just one example; I wouldn't be surprised if there's other code out there, possible on CRAN, Bioconductor, or elsewhere, that rely on this. In the worst case, this may result in silently incorrect results. I think adding the above gatekeeper could help lower the risk for this.

FWIW, @mschubert noticed this issue for clustermq, cf. mschubert/clustermq#119.

[HenrikBengtsson]

Ah, I missed that you tried to add such a gatekeeping, but had to revert it (#83 (reply in thread)).

[wlandau]

FWIW, @mschubert noticed this issue for clustermq, cf. mschubert/clustermq#119.

Yes, and I see he proposed:

Always pretend to have one parallel worker, but split work to the actual number of workers using Q

which sounds feasible. You rightly point out:

If so, one need to worry about the different "load-balancing" features of the "cluster" API, e.g. parLapply(..., chunk.size) and parLapplyLB().

I think it is totally reasonable to exclude R-specific load-balancing features like chunk.size. Otherwise, modern popular concurrency frameworks like ZeroMQ and NNG would not be pluggable, which would be a great loss for the parallel package.

[shikokuchuo]

Setting aside the further discussion for now, just a note to please disregard the initial request for comment by me.

I had mistakenly assumed that setting the default cluster in parallel meant that it would carry over when calling future::plan("cluster"). Turns out this isn't the case and 'workers' needs to be explicitly specified to use the cluster. This takes us back to the original problem and brings us no closer.

[HenrikBengtsson]

I think it is totally reasonable to exclude R-specific load-balancing features like chunk.size.
Otherwise, modern popular concurrency frameworks like ZeroMQ and NNG would not be pluggable, which would be a great loss for the parallel package.

I think one must be extremely careful when making such decisions. It's important to make sure one is not putting a square peg in a round hole in situations where we're gluing major frameworks together. To me, it's not clear if that is happening here and what the shape of the hole is, but it's important to know that before moving forward. For instance, incorrect use affects the maintainer and future of the parallel package, e.g. it might lock them in on a path where they didn't intend to go. And, as I said before, it can hit users negatively out there without their and our knowledge.

It could very well be that the parallel API needs refinement and clarification (I guess this discussion shows that). If parallel provides a round hole, and we have a square peg, then we might have to work with parallel to expand on its API. To be clear, I'm re-re-evaluating my view of the parallel API yet again.

[shikokuchuo]

Another attempt at this. The original issue was the print method for a subset cluster giving an error:

cl <- mirai::make_cluster(2)
cl_kk <- cl[1]
print(cl_kk)
#> Error in ..[[attr(x, "id")]] : 
#>   wrong arguments for subsetting an environment

I have now implemented in the print method for it to give a more meaningful error instead:

cl <- mirai::make_cluster(2)
cl_kk <- cl[1]
print(cl_kk)
#> Error in print.miraiCluster(cl_kk) : 
#>    cluster of type 'mirai' is not subsettable

This should be an improvement. Not much else I can do at this stage, as [ is used internally by the parallel package, so I can't disable it.

[shikokuchuo]

A possible solution is as I mentioned above - move all internal usage of [ to .subset() - which I've confirmed works, as that's exactly how I've implemented [.miraiCluster, and then we can document that each cluster back-end should supply it's own [ method if supported, or an appropriate error if not.

[shikokuchuo]

Just to spell out the obvious, but this puts the onus on the back-end provider to comply with the API, rather than all downstream users to check if something is supported - and I know this is relevant for you in future, but for the unknown / possible cases out there, it seems more troublesome to require an extra step from them.

[HenrikBengtsson]

I've created HenrikBengtsson/Wishlist-for-R#161 to continue "that" discussion.

Back to mirai per se. I see that each cluster node as a node attribute, that appears to be an index;

> cl <- mirai::make_cluster(2)
> str(cl[[1]])
Class 'miraiNode' <environment: 0x5d74f8b258b0> 
 - attr(*, "node")= int 1
 - attr(*, "id")= chr "`1`"
> str(cl[[2]])
Class 'miraiNode' <environment: 0x5d74f8b25f78> 
 - attr(*, "node")= int 2
 - attr(*, "id")= chr "`1`"

Couldn't this information be used so that sendCall(cl[[i]], ...) will create a mirai "task" that is only picked up by worker i? Likewise, recvData(cl[[i]], ...) would only receive that from worker i.

[shikokuchuo]

Thanks, that’s a great idea to continue that discussion there. I will post a reply once I have my thoughts together on this - especially in light of the new findings below.

As for mirai, I'm glad to say that this no longer applies. The commit I put through at the R dev day last week actually solves this issue by providing its own [.miraiCluster method. The objects retuned from this method are no longer classed ‘cluster’, hence will no longer work with functions taking a cluster.

library(mirai)
library(parallel)
cl <- make_cluster(2)
parLapply(cl, 1:2, rnorm)
#> [[1]]
#> [1] 0.5912369
#> 
#> [[2]]
#> [1] -0.3363421 -0.8519036
parLapply(cl[1], 1:2, rnorm)
#> Error in checkCluster(cl): not a valid cluster

^{Created on 2024-05-01 with reprex v2.1.0}

This addresses the most pressing concerns that you raised above.

The presence of the subset method was found thanks to Ivan Krylov’s excellent documentation work, and in the bugzilla contribution it is stated that back-ends should supply their own method where appropriate.

As to your question of whether a miraiCluster after using the [ operator could return a 'subset' cluster - well the spirit of the discussions at R Project Sprint were to allow novel approaches to back-ends, otherwise it would not really permit progress and innovation if everything is implemented in the same way, as long as the parallel interface is respected. As an example, it is of course possible to have separate NNG socket connections to each node, but this is not really a modern design. NNG sockets are massively scalable, supporting multiple connections in both directions, listeners and dialers. Not to make best use of this (as designed) is not something that I could honestly recommend.

To summarize, mirai fully supports the parallel interface, i.e. all the functions taking a cluster in the parallel package can take a miraiCluster. The subsetting with [ point is moot as that operation no longer produces a cluster / miraiCluster.

[HenrikBengtsson]

This addresses the most pressing concerns that you raised above.

Thanks for finding this "workaround"/"patch"/"hack" and thanks to Ivan. I like how it helps to prevent mistakes from propagating in several places, e.g.

> library(future)
> cl <- mirai::make_cluster(2)
> plan(cluster, workers = cl)
Error: 'inherits(cl, "cluster")' is not TRUE

Here, the error is triggered because the initialization calls parallel::clusterEvalQ(cl[kk], Sys.getpid()) internally. If it wouldn't do it there, it would most likely happen at a later stage.

... R dev day last week ...

It's the first time I heard of this event, but I now found the 2024-04-26 event on https://contributor.r-project.org/events/. I was aware of the R Dev Day on 2024-07-12 following the useR! 2024 conference. (Making these comments here for future references and to raise awareness)

The presence of the subset method was found thanks to Ivan Krylov’s excellent documentation work, and in the bugzilla contribution it is stated that back-ends should supply their own method where appropriate.

This is https://bugs.r-project.org/show_bug.cgi?id=18587, correct?

To summarize, mirai fully supports the parallel interface ...

Although one could say it's technically supporting all functions calls in parallel, I'd argue this stretching the definition of "parallel interface" and comes with a risk of other forked "interpretations" and future surprises (due to changes in parallel or requirements on CRAN or Bioconductor, etc.).

Personally, I still think the original intent of the parallel cluster API was that you can address individual cluster node. But it doesn't really matter what I think - all that matters is that there is more than one interpretation/view out there. Because of that, I mean that this needs to be clarified in the parallel package, and I think "lack of addressable" nodes needs its own explicit specification (HenrikBengtsson/Wishlist-for-R#161). If that's not done, there's a risk of things getting interpreted slightly different by different used. We really need Luke's input (and action) on this, because we already know what happens to when part of the R API is not 100% defined ...

To be clear, I don't want to slow down progress. On the contrary, by clarifying and formalizing things, we should be able to move forward faster in a more robust fashion.