r - 在 foreach 中为 .combine 构建函数

Question

我有一个要并行执行的过程，但由于某些原因而失败 strange error .现在我正在考虑合并，并计算主CPU上的失败任务。但是我不知道如何为 .combine 编写这样的函数。

应该怎么写？

我知道怎么写，例如 this answer 提供了一个例子，但它没有提供如何处理失败的任务，也没有在 master 上重复任务。

我会做这样的事情:

foreach(i=1:100, .combine = function(x, y){tryCatch(?)} %dopar% {
    long_process_which_fails_randomly(i)
}

但是，如何在 .combine 函数中使用该任务的输入(如果可以的话)？或者我应该在 %dopar%里面提供返回一个标志或一个列表来计算它？

Answer 1

要在 combine 函数中执行任务，您需要在 foreach 循环体返回的结果对象中包含额外信息。在这种情况下，这将是一个错误标志和值 i .有很多方法可以做到这一点，但这里有一个例子:

comb <- function(results, x) {
  i <- x$i
  result <- x$result
  if (x$error) {
    cat(sprintf('master computing failed task %d\n', i))
    # Could call function repeatedly until it succeeds,
    # but that could hang the master
    result <- try(fails_randomly(i))
  }
  results[i] <- list(result)  # guard against a NULL result
  results
}

r <- foreach(i=1:100, .combine='comb',
             .init=vector('list', 100)) %dopar% {
  tryCatch({
    list(error=FALSE, i=i, result=fails_randomly(i))
  },
  error=function(e) {
    list(error=TRUE, i=i, result=e)
  })
}

我很想通过重复执行并行循环来解决这个问题，直到计算出所有任务:

x <- rnorm(100)
results <- lapply(x, function(i) simpleError(''))

# Might want to put a limit on the number of retries
repeat {
  ix <- which(sapply(results, function(x) inherits(x, 'error')))
  if (length(ix) == 0)
    break

  cat(sprintf('computing tasks %s\n', paste(ix, collapse=',')))
  r <- foreach(i=x[ix], .errorhandling='pass') %dopar% {
    fails_randomly(i)
  }

  results[ix] <- r
}

请注意，此解决方案使用 .errorhandling如果可能发生错误，该选项非常有用。有关此选项的更多信息，请参阅 foreach 手册页。