scala - 尝试让惰性评估适用于无限流

Question

我正在尝试通过过滤操作实现无限流。我想通过对尾部使用惰性求值来使其不会因堆栈溢出错误而崩溃。

abstract class MyStream[+A] {
  def head: A
  def tail: MyStream[A]

  def #::[B >: A](element: B): MyStream[B] // prepend operator

  def filter(predicate: A => Boolean): MyStream[A]
}

class FiniteStream[+A](val head: A, val tail: MyStream[A]) extends MyStream[A] {    
  override def #::[B >: A](element: B): MyStream[B] = new FiniteStream[B](element, this)

  override def filter(predicate: A => Boolean): MyStream[A] = {
    lazy val filteredTail = tail.filter(predicate)
    if (predicate(head)) filteredTail
    else filteredTail
  }
}

class InfiniteStream[+A](override val head: A, generator: A => A) extends MyStream[A] {
  override def tail: MyStream[A] = {
    lazy val tail = new InfiniteStream[A](generator(head), generator)
    tail
  }

  override def #::[B >: A](element: B): MyStream[B] =
    new FiniteStream[B](element, this)

  override def filter(predicate: A => Boolean): MyStream[A] = {
    lazy val filteredTail = tail.filter(predicate)
    if (predicate(head)) head #:: filteredTail
    else filteredTail
  }
}

object MyStream {
    def from[A](start: A)(generator: A => A): MyStream[A] = new InfiniteStream[A](start, generator)
}

val stream: MyStream[Int] = MyStream.from(1)((n: Int) => n + 1)
val filtered = stream.filter(_ % 2 == 0)

但是这个程序确实因堆栈溢出错误而崩溃。看来我的惰性评估策略不起作用。尾部仍在评估中。为什么？

Answer 1

该问题是由 InfiniteStream.filter 引起的，它将尾部过滤器创建为惰性值，但随后立即访问它，从而强制对值进行求值。这会导致整个流被评估为递归调用，导致堆栈爆炸。

lazy val 会延迟用于构造变量的表达式的执行，直到它被访问为止。因此，您需要延迟对 tail.filter(predicate) 的访问，直到流的用户访问 tail。

实现这一点的最简单且更实用的方法是使用 View 实现过滤器。即过滤器返回一个新流，仅按需过滤尾部。

EG

class FilterStream[+A] private (predicate: predicate: A => Boolean, stream: MyStream) extends MyStream[A] {
  override def head: A = stream.head
  override def tail: MyStream[A] = FilterStream.dropWhile(!predicate(_), stream)
}


object FilterStream {
  def apply[A](predicate: predicate: A => Boolean, stream: MyStream[A]): MyStream[A] = {
    new FilterStream(predicate, dropWhile(!predicate(_), stream))
  }

  @tailrec
  def dropWhile[A](predicate: predicate: A => Boolean, stream: MyStream[A]): MyStream[A] = {
    if (stream.isEmpty || predicate(stream.head)) stream
    else dropWhile(predicate, stream.tail)
  }
}

最后，出于多种原因，您应该考虑实现一个具有自己的类型和对象的空流，而且这样您就可以在您的生成器决定需要时终止无限流。

object Nil extends MyStream[Nothing] {
  override def head: A = throw NoSuchElement
  override def tail: MyStream[A] = throw NoSuchElement
}

Head 和 tail 始终是不安全的方法，另一个改进是使用案例类来公开流的形状，然后用户将在流上进行模式匹配。这将保护您的用户不必使用 head 和 tail 等不安全的方法。