Scalaz 7 Iteratee 处理大型 zip 文件 (OutOfMemoryError)

Question

我正在尝试使用 scalaz iteratee 包来处理恒定空间中的大型 zip 文件。我需要对 zip 文件中的每个文件执行一个长时间运行的过程。这些进程可以（并且应该）并行运行。

我创建了一个EnumeratorT，它将每个ZipEntry 膨胀成一个File 对象。签名看起来像：

def enumZipFile(f:File):EnumeratorT[IoExceptionOr[IO[File]], IO]

我想附加一个IterateeT，它将对每个文件执行长时间运行的过程。我基本上得到了类似的东西：

type IOE[A] = IoExceptionOr[A]

def action(f:File):IO[List[Promise[IOE[File]]]] = (
  consume[Promise[IOE[File]], IO, List] %=
  map[IOE[File], Promise[IOE[File]], IO](longRunningProcess) %=
  map[IOE[IO[File]], IOE[File], IO](_.unsafePerformIO) &=
  enumZipFile(f)
).run

def longRunningProcess:(iof:IOE[File]):Promise[IOE[File]] =
  Promise { Thread.sleep(5000); iof }

当我尝试运行它时：

action(new File("/really/big/file.zip")).unsafePerformIO.sequence.get

我收到java.lang.OutOfMemoryError: Java heap space 消息。这对我来说很有意义，因为它试图在所有这些 IO 和 Promise 对象的内存中建立一个庞大的列表。

几个问题：

• 有人对如何避免这种情况有任何想法吗？感觉就像我错误地解决了这个问题，因为我真的只关心longRunningProcess 的副作用。
• Enumerator 方法在这里是错误的方法吗？

我几乎没有想法，所以任何事情都会有所帮助。

谢谢！

更新 #1

这是堆栈跟踪：

[error] java.lang.OutOfMemoryError: Java heap space
[error]         at scalaz.Free.flatMap(Free.scala:46)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:61)
[error]         at scalaz.effect.IOFunctions$$anon$5.apply(IO.scala:222)
[error]         at scalaz.effect.IO$$anonfun$flatMap$1.apply(IO.scala:62)

我目前正在听取 nadavwr 的建议，以确保一切都像我想的那样。我会报告任何更新。

更新 #2

使用以下两个答案的想法，我找到了一个不错的解决方案。正如 huynhjl 建议的那样（我使用 nadavwr 的分析堆转储的建议进行了验证），consume 导致每个膨胀的ZipEntry 都保存在内存中，这就是进程内存不足的原因。我将consume 更改为foldM，并将长时间运行的进程更新为只返回Promise[IOE[Unit]] 而不是对文件的引用。这样我最后就有了所有 IoExceptions 的集合。这是可行的解决方案：

def action(f:File):IO[List[Promise[IOE[Unit]]]] = (
  foldM[Promise[IOE[Unit]], IO, List[Promise[IOE[Unit]]]](List.empty)((acc,x) => IO(x :: acc)) %=
  map[IOE[File], Promise[IOE[Unit]], IO](longRunningProcess) %=
  map[IOE[IO[File]], IOE[File], IO](_.unsafePerformIO) &=
  enumZipFile(f)
).run

def longRunningProcess:(iof:IOE[File]):Promise[IOE[Unit]] =
  Promise { Thread.sleep(5000); iof.map(println) }

此解决方案会在异步上传每个条目时对其进行膨胀。最后，我有一个包含任何错误的已完成 Promise 对象的庞大列表。我仍然不完全相信这是对 Iteratee 的正确使用，但我现在确实有几个可重用、可组合的部分，可以在我们系统的其他部分中使用（这对我们来说是一种非常常见的模式）。

感谢您的帮助！

Answer 1

不要使用consume。查看我最近的另一个答案：How to use IO with Scalaz7 Iteratees without overflowing the stack?

foldM 可能是更好的选择。

还可以尝试将文件映射到其他内容（如成功返回代码），以查看这是否允许 JVM 垃圾收集膨胀的 zip 条目。

Answer 2

多贵（就内存而言，您的longRunningProcess 是多少？文件压缩如何？它们的执行次数是否达到了您的预期？（一个简单的计数器会很有帮助）

堆栈跟踪将有助于确定压死骆驼的最后一根稻草——有时这就是罪魁祸首。

如果您想确定是什么占用了这么多内存，您可以使用-XX:+HeapDumpOnOutOfMemoryError JVM 参数，然后使用 VisualVM、Eclipse MAT 或其他堆分析器对其进行分析。

跟进

对我来说，您列举承诺似乎很奇怪。独立于枚举器和迭代器启动计算是违反直觉的。返回“惰性”元素而不是承诺的枚举器可能会更好地为基于迭代的解决方案提供服务。不幸的是，这将使您对单个文件的处理成为串行的，但这是您的迭代器——非阻塞流处理。

恕我直言，基于演员的解决方案更适合，但演员和迭代者（尤其是后者）对于您要完成的任务（至少是您共享的部分）似乎都过分了。

请考虑 Scala 2.10 的 scala.concurrent 包中的普通期货/承诺，并且一定要看看 Scala 的并行集合。在证明不够充分之前，我不会在代码中引入其他概念。尝试定义一个固定大小的 ExecutionContext 来限制您的并行性。

Answer 3

我在快速阅读后开始回答，不知何故，我的脑海中出现了“堆栈溢出”而不是“内存不足错误”......必须是 URL :-)

不过，依赖递归的函数式计算容易受到堆栈溢出的影响，因此我已经为任何偶然发现的人留下了答案，并承诺会尝试提出更相关的答案。

如果您得到的是堆栈溢出，那么您将需要一个“蹦床”，这是一种在递归之间将您的计算提升到堆栈之外的构造。

请参阅Learning Scalaz Day 18 中标题为“Stackless Scala with Free Monads”的部分，这是@eed3si9n 优秀系列文章的一部分。

另见@mpilquist 的this gist，演示了一个蹦床迭代器。