scala - 如何在不溢出堆栈的情况下将 IO 与 Scalaz7 Iteratees 一起使用？

Question

考虑这个代码(取自 here 并修改为使用字节而不是字符行)。

import java.io.{ File, InputStream, BufferedInputStream, FileInputStream }
import scalaz._, Scalaz._, effect._, iteratee.{ Iteratee => I, _ }
import std.list._

object IterateeIOExample {
  type ErrorOr[+A] = EitherT[IO, Throwable, A]

  def openStream(f: File) = IO(new BufferedInputStream(new FileInputStream(f)))
  def readByte(s: InputStream) = IO(Some(s.read()).filter(_ != -1))
  def closeStream(s: InputStream) = IO(s.close())

  def tryIO[A, B](action: IO[B]) = I.iterateeT[A, ErrorOr, B] {
    EitherT(action.catchLeft).map(r => I.sdone(r, I.emptyInput))
  }

  def enumBuffered(r: => BufferedInputStream) = new EnumeratorT[Int, ErrorOr] {
    lazy val reader = r
    def apply[A] = (s: StepT[Int, ErrorOr, A]) => s.mapCont(k =>
      tryIO(readByte(reader)) flatMap {
        case None => s.pointI
        case Some(byte) => k(I.elInput(byte)) >>== apply[A]
      })
  }

  def enumFile(f: File) = new EnumeratorT[Int, ErrorOr] {
    def apply[A] = (s: StepT[Int, ErrorOr, A]) =>
      tryIO(openStream(f)).flatMap(stream => I.iterateeT[Int, ErrorOr, A](
        EitherT(
          enumBuffered(stream).apply(s).value.run.ensuring(closeStream(stream)))))
  }

  def main(args: Array[String]) {
    val action = (
      I.consume[Int, ErrorOr, List] &=
      enumFile(new File(args(0)))).run.run
    println(action.unsafePerformIO())
  }
}

在体面大小的文件 (8kb) 上运行此代码会产生 StackOverflowException。一些搜索发现可以通过使用 Trampoline monad 而不是 IO 来避免异常，但这似乎不是一个很好的解决方案 - 牺牲功能纯度来让程序完成。解决这个问题的显而易见的方法是使用 IO 或 Trampoline 作为 Monad Transformer 来包装另一个，但我找不到它们中任何一个的转换器版本的实现，而且我还没有足够的函数式编程大师知道如何编写我自己的(了解更多关于 FP 是这个项目的目的之一，但我怀疑创建新的 monad 转换器目前有点超出我的水平)。我想我可以围绕创建、运行和返回迭代的结果来包装一个大的 IO 操作，但这感觉更像是一种解决方法而不是解决方案。

大概有些 monad 无法转换为 monad 转换器，所以我想知道是否可以在不丢失 IO 或溢出堆栈的情况下处理大文件，如果是这样，如何？

额外的问题:我想不出任何方法让迭代对象在处理时发出信号，表明它在处理时遇到了错误，除了让它返回任一个，这使得组合它们变得不那么容易。上面的代码显示了如何使用eitherT 来处理枚举器中的错误，但是它对迭代器是如何工作的呢？

Answer 1

在您的代码的不同位置创建异常并打印它们的堆栈长度后，我觉得您的代码没有溢出。一切似乎都以恒定的堆栈大小运行。所以我找了其他地方。最终我复制了 consume 的实现并添加了一些堆栈深度打印并确认它在那里溢出。

所以这溢出:

(I.consume[Int, Id, List] &= EnumeratorT.enumStream(Stream.fill(10000)(1))).run

但是，我后来发现这不是:

(I.putStrTo[Int](System.out) &= EnumeratorT.enumStream(Stream.fill(10000)(1)))
  .run.unsafePerformIO()

putStrTo用途 foldM并且以某种方式不会导致溢出。所以我想知道是否 consume可以按照 foldM来实现.我只是从消费和调整中复制了一些东西，直到它编译:

def consume1[E, F[_]:Monad, A[_]:PlusEmpty:Applicative]: IterateeT[E, F, A[E]] = {
  I.foldM[E, F, A[E]](PlusEmpty[A].empty){ (acc: A[E], e: E) =>
    (Applicative[A].point(e) <+> acc).point[F]
  }
}

它奏效了!打印一长串整数。