Передавать внешние значения между чанками в clusterApply (параллельный пакет)

Question

У меня есть воспроизводимый скрипт для использования кластера.Он не оптимизирован, но используется в качестве практического специалиста для изучения кластерного приложения.

Краткое описание задачи:

1. Существует вектор переменнойдлина;

2. Требуется преобразовать его определенным образом:

   начальный вектор - 235416749 ...

   ожидаемый результат - 235556779 ...

Преобразование выполняется с применением кластера (см. Скрипт ниже).

Выход кластера (пример) выглядит следующим образом:

str (chunkSplit) # начальные чанки Список из 4
$: int [1:24950] 56583 22166 49905 20040 60870 49899 85589 96478 81119 36474
$: int [1: 25050] 22652 31943 57699 58051 33846 27328 34429 84989 16295 29308
$: int [1: 25050] 42641 639 7499 8691978 55004 73069 1528
$: int [1: 24950] 57401 60632 6284 43612 40011 31096 9494 24453 81221 99553

str (cluster_Output) # обновленные чанки Список из 4
$: int [1:24950] 56583 56583 56583 56583 60870 60870 85589 96478 96478 96478
$: int [1: 25050] 22652 31943 57699 58051 58051 58051 58051 84989 84989 84989
$: int [1: 25050] 42644 8639 869991978 91978 91978 91978
$: int [1: 24950] 57401 60632 60632 60632 60632 60632 60632 60632 81221 99553

Преобразование внутри каждого блока соответствует логике, описанной выше.

Описание процесса

Скрипт рассчитывает генерировать общий вывод правильным образом (из других блоков).
Таким образом, каждый следующий блок должен быть преобразован с учетом результата преобразования предыдущего (первый элемент второго блока> = последний элемент первого блока).

Однако куски не были преобразованы таким образом.Потому что переменная max_val не передает правильное значение между кусками.

Вопрос:
Я полагаю, это потому, что каждый работник создает свою собственную среду.Я был бы очень признателен, если вы посоветуете мне, как решить проблему в кластере (передача данных между блоками внутри кластера).

Воспроизводимый сценарий

# Test cluster apllication
library(parallel)


# the function to be applied in the cluster
dochunks <- function(x) {

  # chunk counter
  step2 <<- step2 + 1
  xxx <- unlist(x)
  x_end <- length(xxx)

  # if it is the 2nd and higher chunks -> the 1st element of the vector
  # assigned as max_val
  if (max_val > xxx[1] & step2 >= 2) xxx[1] <- max_val

  i <- 1

  for (i in 1:x_end) {
    if ((i + 1) <= x_end & xxx[i] > xxx[i + 1]) {
      xxx[i + 1] <- xxx[i]
    }
  }

  max_val <<- xxx[x_end]
  return(xxx)
}


# the max value
size <- 100000
# test vector
myvector <- sample(100000, size, replace = TRUE)
# set up workers to process the vector
workers <- 4
# set up the cluster
cls <- makeCluster(workers)
# split test vector into chunks with the number of workers
chunkSplit <- clusterSplit(cls, myvector)

# max value of each chunk
max_val <<- 0
# temporary vector for each chunk
xxx <<- 0
# counter of the application of the each chunk
step2 <<- 0
# export vars into the cluster
clusterExport(cls, varlist = c("max_val", "step2"), envir = environment())
# run the cluster
cluster_Output <- clusterApply(cls, chunkSplit, dochunks)
# quit the cluster
stopCluster(cls)

# compare initial and updated chunks
str(chunkSplit) # initial chunks
str(cluster_Output) # output chunks