如何使用Repa或Accelerate处理无限序列？

Question

我刚刚开始学习Haskell，我的问题可能是一个琐碎的问题，所以很抱歉打扰你。

因此，假设我有一个真正的无限数据源(例如stdin)，并且我想使用Repa或Accelerate的并行工具来处理数据。它们都使用数组(有限数据结构)。在我看来，我的代码将处理数组序列，但我真的不知道该怎么做，然后我需要从一个数组和下一个数组中获得一些数据来计算一些东西。

所以，我的意思是，如何计算所有输入的y = phase(a[i] * conj a[i+1])，其中a是无限的复数流。

Answer 1

正如您提到的，Repa和Accelerate都是为处理固定大小的数组和矩阵而设计的。这些是面向批处理的算法，它们基于许多优化，这些优化只有在使用固定的内存位时才可用。

对于“无限”的数据，你需要一个流(或“在线”)算法。你所描述的方法是可流传输的，因为它只需要有限的窗口。您可以通过使用Repa为流进入的每一对输入(a[i], a[i+1])计算相位，将流算法转换为大量的小批量步骤，但由于将每个小块移动到连续的内存阵列所需的大量装箱和复制，您可能会失去速度。

相比较而言，您可以使用pipes之类的工具来生成快速流式处理算法。

import Pipes
import Data.Complex

type ComplexPair = (Complex Double, Complex Double)

pairOff :: Monad m => Pipe a (a, a) m r
pairOff = await >>= forever . go where
  go x = await >>= \y -> yield (x, y)

compute :: Monad m => Pipe ComplexPair Double m r
compute = do
  (ai, ai1) <- await
  yield (phase $ ai * conjugate ai1)

run :: Monad m => Pipe (Complex Double) Double m r
run = pairOff >-> compute

然后，run可以由无限的流输入源提供，并转储到处理管道的下一步中。这些源可以是一元的，如果是m ~ Identity的话，也可以是纯的。

Answer 2

据我所知，你不能直接使用Repa或Accelerate来处理无限的数据流。它被设计成对向量进行操作，这是一种不同的数据结构。有一些方法可以解决这个问题，您可以选择以下几种方法：

1. 读取固定数量的值，使用它构造一个向量，执行计算，然后再次向下推送这些值，一次迭代地处理数据块。你必须找出你想要的区块大小，但它会在计算过程中给你最优的性能。不幸的是，这意味着您将失去所有这些读取-构造步骤的效率。事实上，这将是我最担心的并行化步骤，因为你可能正在执行这些步骤，而前面的代码块正在使用一个矢量库进行processed.
2. Forget，并转向管道库。它被设计为在常量内存中处理数据流，并获得了相当好的性能。使用pipes-concurrency包，您可以编写读取数据并并行处理数据的流处理器。你不会有向量的效率，但在使用最少内存的情况下更容易有效地处理无限的数据流。
3. 首先使用列表实现它，懒惰地处理每个元素并使用常规的Haskell函数来实现。您不会获得速度或内存使用的好处，但它将很容易编写。一旦你理解了这个概念，你就可以决定是否需要矢量或流的效率，然后在尝试优化之前运行基准测试工具来确定瓶颈在哪里。

就我个人而言，我会从2.或3.开始，然后确定我的程序是否足够慢，足以保证使用向量。也许你可以从向量中获得额外的性能，但除非你使用的是GPU，否则这是不可能的。