Julia:并行循环遍历分区迭代器

Question

所以我想遍历的东西分区的列表中,说1:n了一些n13和21,我非常想运行的代码看起来像这样之间:

valid_num = @parallel (+) for p in partitions(1:n)
  int(is_valid(p))
end

println(valid_num)

这将使用@parallel formap来减少我的问题.例如,将其与Julia文档中的示例进行比较:

nheads = @parallel (+) for i=1:200000000
  Int(rand(Bool))
end

但是,如果我尝试修改循环,我会收到以下错误:

ERROR: `getindex` has no method matching getindex(::SetPartitions{UnitRange{Int64}}, ::Int64)
 in anonymous at no file:1433
 in anonymous at multi.jl:1279
 in run_work_thunk at multi.jl:621
 in run_work_thunk at multi.jl:630
 in anonymous at task.jl:6

我想这是因为我想遍历的东西,是不是形式1:n(编辑:我想这是因为你不能说p[3]如果p=partitions(1:n)).

我已经尝试过pmap用来解决这个问题,但是因为分区的数量可以变得非常大,非常快(有超过250万个分区1:13,当我得到的1:21东西将是巨大的)时,构建这么大的数组变成了问题.我让它跑了一夜,但仍然没有完成.

有没有人对如何在朱莉娅有效地做到这一点有任何建议？我可以访问~30核心计算机,我的任务似乎很容易并行化,所以如果有人知道在Julia中做到这一点的好方法,我将非常感激.

非常感谢!

Answer 1

下面的代码给出了 511，即 10 个集合中大小为 2 的分区数。

using Iterators
s = [1,2,3,4,5,6,7,8,9,10]
is_valid(p) = length(p)==2
valid_num = @parallel (+) for i = 1:30
  sum(map(is_valid, takenth(chain(1:29,drop(partitions(s), i-1)), 30)))
end

该解决方案结合了 taketh、drop 和 chain 迭代器，以获得与下面先前答案下的 take_every 迭代器相同的效果。请注意，在此解决方案中，每个进程都必须计算每个分区。但是，由于每个进程使用不同的参数drop，因此任何两个进程都不会在同一分区上调用 is_valid 。

除非您想做大量数学运算来弄清楚如何实际跳过分区，否则无法避免在至少一个进程上顺序计算分区。我认为西蒙的答案是在一个进程上执行此操作并分配分区。我的要求每个工作进程自己计算分区，这意味着计算是重复的。然而，它是并行复制的，这（如果您实际上有 30 个处理器）不会浪费您的时间。

以下是有关如何实际计算分区迭代器的资源：http://www.informatik.uni-ulm.de/ni/Lehre/WS03/DMM/Software/partitions.pdf。

先前的答案（比必要的更复杂）

我在写我的答案时注意到西蒙的答案。我们的解决方案似乎与我相似，只是我的解决方案使用迭代器来避免在内存中存储分区。我不确定哪种尺寸的设置实际上会更快，但我认为同时拥有这两种选择是件好事。假设计算 is_valid 比计算分区本身花费的时间要长得多，您可以执行以下操作：

s = [1,2,3,4]
is_valid(p) = length(p)==2
valid_num = @parallel (+) for i = 1:30
  foldl((x,y)->(x + int(is_valid(y))), 0, take_every(partitions(s), i-1, 30))
end

这给出了 7，即一组 4 个大小为 2 的分区的数量。 take_every 函数返回一个迭代器，该迭代器返回从第 i 个分区开始的每 30 个分区。这是代码：

import Base: start, done, next
immutable TakeEvery{Itr}
  itr::Itr
  start::Any
  value::Any
  flag::Bool
  skip::Int64
end
function take_every(itr, offset, skip)
  value, state = Nothing, start(itr)
  for i = 1:(offset+1)
    if done(itr, state)
      return TakeEvery(itr, state, value, false, skip)
    end
    value, state = next(itr, state)
  end
  if done(itr, state)
    TakeEvery(itr, state, value, true, skip)
  else
    TakeEvery(itr, state, value, false, skip)
  end
end
function start{Itr}(itr::TakeEvery{Itr})
  itr.value, itr.start, itr.flag
end
function next{Itr}(itr::TakeEvery{Itr}, state)
  value, state_, flag = state
  for i=1:itr.skip
    if done(itr.itr, state_)
      return state[1], (value, state_, false)
    end
    value, state_ = next(itr.itr, state_)
  end
  if done(itr.itr, state_)
    state[1], (value, state_, !flag)
  else
    state[1], (value, state_, false)
  end
end
function done{Itr}(itr::TakeEvery{Itr}, state)
  done(itr.itr, state[2]) && !state[3]
end