如何在 R 中将 `foreach` 和 `%dopar%` 与 `R6` 类一起使用？

我在尝试使用时遇到了问题%dopar% and foreach()与一个R6班级。四处搜索，我只能找到两个与此相关的资源，一个未答复所以问题和一个开放的GitHub问题 on the R6存储库。

在一条评论（即 GitHub 问题）中，建议通过重新分配parent_env类的SomeClass$parent_env <- environment()。我想了解到底是什么environment()指的是这个表达式（即SomeClass$parent_env <- environment()) 内被调用%dopar% of foreach?

这是一个最小的可重现示例：

Work <- R6::R6Class("Work",

    public = list(
        values = NULL,


        initialize = function() {
            self$values <- "some values"
        }
    )
)

现在，以下Task类使用Work构造函数中的类。

Task <- R6::R6Class("Task",
    private = list(
        ..work = NULL
    ),


    public = list(
        initialize = function(time) {
            private$..work <- Work$new()
            Sys.sleep(time)
        }
    ),


    active = list(
        work = function() {
            return(private$..work)
        }
    )
)

In the Factory类，该Task类被创建并且foreach实施于..m.thread().

Factory<- R6::R6Class("Factory",

    private = list(
        ..warehouse = list(),
        ..amount = NULL,
        ..parallel = NULL,


        ..m.thread = function(object, ...) {
            cluster <- parallel::makeCluster(parallel::detectCores() -  1)
            doParallel::registerDoParallel(cluster)

            private$..warehouse <- foreach::foreach(1:private$..amount, .export = c("Work")) %dopar% {
                # What exactly does `environment()` encapsulate in this context?
                object$parent_env <- environment()
                object$new(...) 
            }

            parallel::stopCluster(cluster)
        },


        ..s.thread = function(object, ...) {
            for (i in 1:private$..amount) {
               private$..warehouse[[i]] <- object$new(...)
            }
        },


        ..run = function(object, ...) {
            if(private$..parallel) {
                private$..m.thread(object, ...)
            } else {
                private$..s.thread(object, ...)
            }
        }
    ),


    public = list(
        initialize = function(object, ..., amount = 10, parallel = FALSE) {
            private$..amount = amount
            private$..parallel = parallel

            private$..run(object, ...)
        }
    ),


    active = list(
        warehouse = function() {
            return(private$..warehouse)
        }
    )
)

然后，它被称为：

library(foreach)

x = Factory$new(Task, time = 2, amount = 10, parallel = TRUE)

没有以下行object$parent_env <- environment()，它会抛出一个错误（即，如其他两个链接中提到的）：Error in { : task 1 failed - "object 'Work' not found".

我想知道，（1）分配时有哪些潜在的陷阱parent_env inside foreach（2）为什么它首先有效？

更新1：

• 我回来了environment()从内部foreach()，使得private$..warehouse捕捉这些环境
• using rlang::env_print()在调试会话中（即browser()声明紧随其后foreach已结束执行）它们的组成如下：

Browse[1]> env_print(private$..warehouse[[1]])

# <environment: 000000001A8332F0>
# parent: <environment: global>
# bindings:
#  * Work: <S3: R6ClassGenerator>
#  * ...: <...>

Browse[1]> env_print(environment())

# <environment: 000000001AC0F890>
# parent: <environment: 000000001AC20AF0>
# bindings:
#  * private: <env>
#  * cluster: <S3: SOCKcluster>
#  * ...: <...>

Browse[1]> env_print(parent.env(environment()))

# <environment: 000000001AC20AF0>
# parent: <environment: global>
# bindings:
#  * private: <env>
#  * self: <S3: Factory>

Browse[1]> env_print(parent.env(parent.env(environment())))

# <environment: global>
# parent: <environment: package:rlang>
# bindings:
#  * Work: <S3: R6ClassGenerator>
#  * .Random.seed: <int>
#  * Factory: <S3: R6ClassGenerator>
#  * Task: <S3: R6ClassGenerator>

免责声明：我在这里所说的很多内容都是基于我所知道的有根据的猜测和推论， 我不能保证一切都是100%正确的。

我觉得可能会有很多陷阱 哪一种适用实际上取决于你做什么。 我认为你的第二个问题更重要 因为如果你明白这一点， 您将能够自己评估一些陷阱。

题目比较复杂， 但你可能可以先阅读有关R 的词法作用域。 本质上，R 有一种环境层次结构， 当执行R代码时， 在当前环境中找不到其值的变量 （这是什么environment()回报） 正在寻求在parent环境 （不要与调用者环境混淆）。

根据您链接的 GitHub 问题，R6生成器保存对其父环境的“引用”， 他们希望他们的类可能需要的一切都可以在所述父级或环境层次结构中的某个位置找到， 从该父母开始并“向上”。

您使用的解决方法有效的原因是您正在将生成器的父环境替换为当前环境中的父环境foreach在并行工作者内部调用 （可能是不同的R进程，不一定是不同的线程）， 并且，考虑到你的.export规范可能会导出必要的值， 然后，R 的词法作用域可以从以下位置开始搜索缺失值：foreach在单独的线程/进程中调用。

对于您链接的具体示例， 我发现有一种更简单的方法可以让它发挥作用 （至少在我的 Linux 机器上） 是要做到以下几点：

library(doParallel)

cluster <- parallel::makeCluster(parallel::detectCores() -  1)
doParallel::registerDoParallel(cluster)
parallel::clusterExport(cluster, setdiff(ls(), "cluster"))

x = Factory$new(Task, time = 1, amount = 3)

but离开..m.thread功能为：

..m.thread = function(object, amount, ...) {
    private$..warehouse <- foreach::foreach(1:amount) %dopar% {
        object$new(...) 
    }
}

（并手动调用stopCluster完成后）。

The clusterExport调用应该具有类似于*的语义： 从主 R 进程的全局环境中获取所有内容，除了cluster, 并使其在每个并行工作人员的全局环境中可用。 这样，里面的任何代码foreach当词法作用域到达各自的全局环境时，调用可以使用生成器。foreach可以很聪明，自动导出一些变量 （如 GitHub 问题所示）， 但它有局限性， 并且词法作用域期间使用的层次结构可能会变得非常混乱。

*我说“类似于”是因为我不知道如果使用分叉，R 到底会做什么来区分（全局）环境， 但由于需要出口， 我认为它们确实是相互独立的。

PS：我会打电话给on.exit(parallel::stopCluster(cluster))如果您在函数调用中创建工作人员， 这样，您就可以避免留下进程，直到它们在发生错误时以某种方式停止。