尝试获得惰性评估以适用于无限流

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返回一个新流，该流仅按需过滤尾部。

例如

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。