试图让惰性评估适用于无限流

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 引起的，它将尾过滤器创建为惰性值，但随后立即访问它，从而强制评估该值。这会导致整个流被评估为递归调用炸毁堆栈。

惰性 val 延迟用于构造变量的表达式的执行，直到它被访问。所以需要延迟访问tail.filter(predicate)，直到流的用户访问尾部。

实现这一点的最简单和更实用的方法是使用视图实现过滤器。也就是说 filter 返回一个新的流，它只按需过滤尾部。

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)
  }
}

最后，出于多种原因，您应该考虑使用自己的类型和对象实现一个空流，而且如果您的generator 决定它想要终止一个无限流。

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

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