arrays - Repa 阵列上的并行 mapM

Question

在我最近的 work与 Gibbs sampling ，我一直在充分利用 RVar 在我看来，它为随机数生成提供了一个近乎理想的接口(interface)。遗憾的是，由于无法在 map 中使用 monadic 操作，我一直无法使用 Repa。

虽然显然一元映射一般不能并行化，但在我看来 RVar可能至少是一个 monad 的例子，其中的效果可以安全地并行化(至少在原则上；我对 RVar 的内部工作并不十分熟悉)。即，我想写如下内容，

drawClass :: Sample -> RVar Class
drawClass = ...

drawClasses :: Array U DIM1 Sample -> RVar (Array U DIM1 Class)
drawClasses samples = A.mapM drawClass samples

哪里 A.mapM看起来像，

mapM :: ParallelMonad m => (a -> m b) -> Array r sh a -> m (Array r sh b)

虽然很明显这将如何工作在很大程度上取决于 RVar 的实现及其底层证券 RandomSource ，原则上人们会认为这将涉及为每个产生的线程绘制一个新的随机种子并照常进行。

直觉上，这个想法似乎可以推广到其他一些 monad。

所以，我的问题是:可以构造一个类 ParallelMonad哪些 monad 可以安全地并行化其效果(大概至少由 RVar 居住)？

它可能是什么样子？其他哪些 monad 可能存在于这个类中？其他人是否考虑过这在 Repa 中如何运作的可能性？

最后，如果这种并行 monadic 操作的概念无法推广，那么在 RVar 的特定情况下，有没有人看到有什么好的方法可以使此工作正常进行？ (哪里会非常有用)？放弃 RVar因为并行性是一个非常困难的权衡。

Answer 1

问这个问题已经7年了，似乎仍然没有人想出一个很好的解决方案。 Repa 没有 mapM/traverse就像函数一样，甚至可以在没有并行化的情况下运行。此外，考虑到过去几年取得的进展，它似乎也不太可能发生。

由于 Haskell 中许多数组库的陈旧状态以及我对它们的功能集的总体不满，我已经在数组库中投入了几年的工作 massiv ，它借鉴了 Repa 的一些概念，但将其带到了一个完全不同的层次。介绍就够了。

在今天之前，在 massiv 中有三个类似一元映射的函数。 (不包括同义词之类的函数: imapM 、 forM 等):

mapM - 任意 Monad 中的常用映射.由于明显的原因不可并行化，而且速度也有点慢(按照通常的 mapM 列表慢)

traversePrim - 这里我们仅限于 PrimMonad ，明显快于 mapM ，但其原因在本次讨论中并不重要。

mapIO - 这个，顾名思义，仅限于 IO (或者更确切地说 MonadUnliftIO ，但这无关紧要)。因为我们在IO我们可以自动将数组拆分为与核心数量一样多的块，并使用单独的工作线程来映射 IO对这些块中的每个元素进行操作。不像纯fmap ，这也是可并行化的，我们必须在 IO这是因为调度的不确定性以及我们映射操作的副作用。

所以，当我读到这个问题时，我心想这个问题实际上在 massiv 中得到了解决。 ，但没有那么快。随机数生成器，例如 mwc-random 和其他人在 random-fu 不能在多个线程中使用相同的生成器。这意味着，我唯一缺少的拼图是:“为每个产生的线程绘制一个新的随机种子并照常进行”。换句话说，我需要两件事:

一个可以初始化与工作线程数量一样多的生成器的函数

以及一个抽象，可以根据操作在哪个线程中运行，无缝地为映射函数提供正确的生成器。

所以这正是我所做的。

首先我将使用特制的 randomArrayWS 来举例说明。和 initWorkerStates 函数，因为它们与问题更相关，后来转移到更一般的一元映射。以下是它们的类型签名:

randomArrayWS ::
     (Mutable r ix e, MonadUnliftIO m, PrimMonad m)
  => WorkerStates g -- ^ Use `initWorkerStates` to initialize you per thread generators
  -> Sz ix -- ^ Resulting size of the array
  -> (g -> m e) -- ^ Generate the value using the per thread generator.
  -> m (Array r ix e)

initWorkerStates :: MonadIO m => Comp -> (WorkerId -> m s) -> m (WorkerStates s)

对于不熟悉的人 massiv , Comp 参数是要使用的计算策略，值得注意的构造函数是:

Seq - 按顺序运行计算，不 fork 任何线程

Par - 启动尽可能多的线程，并使用这些线程来完成工作。

我会用 mwc-random 最初以包为例，后来移至 RVarT :

λ> import Data.Massiv.Array
λ> import System.Random.MWC (createSystemRandom, uniformR)
λ> import System.Random.MWC.Distributions (standard)
λ> gens <- initWorkerStates Par (\_ -> createSystemRandom)

上面我们使用系统随机性为每个线程初始化了一个单独的生成器，但我们也可以通过从 WorkerId 派生出每个线程的种子来使用它。争论，这仅仅是 Int worker 指数。现在我们可以使用这些生成器来创建一个具有随机值的数组:

λ> randomArrayWS gens (Sz2 2 3) standard :: IO (Array P Ix2 Double)
Array P Par (Sz (2 :. 3))
  [ [ -0.9066144845415213, 0.5264323240310042, -1.320943607597422 ]
  , [ -0.6837929005619592, -0.3041255565826211, 6.53353089112833e-2 ]
  ]

通过使用 Par策略 scheduler 库将在可用工作人员之间平均分配生成工作，每个工作人员将使用自己的生成器，从而使其线程安全。没有什么能阻止我们重用相同的 WorkerStates任意次数，只要不是同时进行，否则会导致异常:

λ> randomArrayWS gens (Sz1 10) (uniformR (0, 9)) :: IO (Array P Ix1 Int)
Array P Par (Sz1 10)
  [ 3, 6, 1, 2, 1, 7, 6, 0, 8, 8 ]

现在放 mwc-random另一方面，我们可以通过使用类似 generateArrayWS 的函数将相同的概念重用于其他可能的用例。 :

generateArrayWS ::
     (Mutable r ix e, MonadUnliftIO m, PrimMonad m)
  => WorkerStates s
  -> Sz ix --  ^ size of new array
  -> (ix -> s -> m e) -- ^ element generating action
  -> m (Array r ix e)

和 mapWS :

mapWS ::
     (Source r' ix a, Mutable r ix b, MonadUnliftIO m, PrimMonad m)
  => WorkerStates s
  -> (a -> s -> m b) -- ^ Mapping action
  -> Array r' ix a -- ^ Source array
  -> m (Array r ix b)

这是关于如何使用此功能的 promise 示例 rvar , random-fu 和 mersenne-random-pure64 图书馆。我们本可以使用 randomArrayWS这里也是，但为了举例，假设我们已经有一个具有不同 RVarT 的数组s，在这种情况下，我们需要一个 mapWS :

λ> import Data.Massiv.Array
λ> import Control.Scheduler (WorkerId(..), initWorkerStates)
λ> import Data.IORef
λ> import System.Random.Mersenne.Pure64 as MT
λ> import Data.RVar as RVar
λ> import Data.Random as Fu
λ> rvarArray = makeArrayR D Par (Sz2 3 9) (\ (i :. j) -> Fu.uniformT i j)
λ> mtState <- initWorkerStates Par (newIORef . MT.pureMT . fromIntegral . getWorkerId)
λ> mapWS mtState RVar.runRVarT rvarArray :: IO (Array P Ix2 Int)
Array P Par (Sz (3 :. 9))
  [ [ 0, 1, 2, 2, 2, 4, 5, 0, 3 ]
  , [ 1, 1, 1, 2, 3, 2, 6, 6, 2 ]
  , [ 0, 1, 2, 3, 4, 4, 6, 7, 7 ]
  ]

需要注意的是，尽管在上面的示例中使用了 Mersenne Twister 的纯实现，但我们无法逃避 IO。这是因为非确定性调度，这意味着我们永远不知道哪个工作人员将处理数组的哪个块，因此哪个生成器将用于数组的哪个部分。从好的方面来说，如果生成器是纯且可拆分的，例如 splitmix ，那么我们可以使用纯粹的、确定性的和可并行化的生成函数: randomArray ，但这已经是一个单独的故事了。