Как написать не утечку хвостовой рекурсивной функции с помощью Stream.cons в Scala?

При написании функции, действующей на Stream (s), существуют разные понятия рекурсии. Первый простой смысл не является рекурсивным на уровне компилятора, так как хвост, если не оценивается мгновенно, поэтому функция возвращает сразу, но возвращаемый поток рекурсивный:

final def simpleRec[A](as: Stream[A]): Stream[B] = 
  if (a.isEmpty) Stream.empty              
  else someB(a.head) #:: simpleRec(a.tail)

Вышеупомянутое понятие рекурсии не вызывает никаких проблем. Второй по сути является хвостовым рекурсивным на уровне компилятора:

@tailrec
final def rec[A](as: Stream[A]): Stream[B] = 
  if (a.isEmpty) Stream.empty              // A) degenerated
  else if (someCond) rec(a.tail)           // B) tail recursion
  else someB(a.head) #:: rec(a.tail)       // C) degenerated

Проблема заключается в том, что случай C) обнаруживается компилятором как вызов без хвоста, даже если фактический вызов не выполняется. Этого можно избежать, разложив хвост потока в вспомогательную функцию:

@tailrec
final def rec[A](as: Stream[A]): Stream[B] = 
  if (a.isEmpty) Stream.empty              
  else if (someCond) rec(a.tail)          // B)
  else someB(a.head) #:: recHelp(a.tail)  

@tailrec
final def recHelp[A](as: Stream[A]): Stream[B] = 
  rec(as)

Пока он компилируется, этот подход в конечном итоге приводит к утечке памяти. Так как хвост-рекурсивный rec в конечном итоге вызывается из функции recHelp, кадр стека функции recHelp содержит ссылку на головку пара и не позволяет потоку собирать мусор до тех пор, пока rec возвращает вызовы, которые могут быть довольно длинными (с точки зрения шагов рекурсии) в зависимости от количества вызовов B).

Обратите внимание, что даже в случае без помощи, если компилятор разрешил @tailrec, утечка памяти может все еще присутствовать, поскольку хвост ленивого потока фактически создает анонимный объект, содержащий ссылку на головку потока.

Ответ 1

Проблема, как вы намекали, заключается в том, что в коде, который вы вставили, функция filterHelp сохраняет голову (отсюда ваше решение удаляет ее).

Лучший ответ - просто избегать этого удивительного поведения, использовать Scalaz EphemeralStream и видеть его как не oom, так и работать значительно быстрее, так как он намного приятнее для gc. Его не всегда так просто работать с, например, head is() = > A not A, без экстракторов и т.д., но все они ориентированы на одно объективное и надежное использование потока.

Функция filterHelper обычно не нуждается в том, чтобы сохранить ссылку:

import scalaz.EphemeralStream

@scala.annotation.tailrec
def filter[A](s: EphemeralStream[A], f: A => Boolean): EphemeralStream[A] = 
  if (s.isEmpty) 
    s
  else
    if (f(s.head())) 
      EphemeralStream.cons(s.head(), filterHelp(s.tail() , f) )
    else
      filter(s.tail(), f)

def filterHelp[A](s: EphemeralStream[A], f: A => Boolean) =
  filter(s, f)

def s1 = EphemeralStream.range(1, big)

Я бы сказал, что если у вас нет веских причин использовать Stream (другие зависимости библиотек и т.д.), тогда просто придерживайтесь EphemeralStream, там гораздо меньше сюрпризов.

Ответ 2

Возможное обходное решение заключается в том, чтобы метод recHelp не содержал ссылку на головку потока. Это может быть достигнуто путем передачи завернутого потока в него и мутации оболочки для удаления ссылки из нее:

@tailrec
final def rec[A](as: Stream[A]): Stream[B] = 
  if (a.isEmpty) Stream.empty              
  else if (someCond) rec(a.tail)          
  else {
    // don't inline and don't define as def,
    // or anonymous lazy wrapper object would hold reference
    val tailRef = new AtomicReference(a.tail)
    someB(a.head) #:: recHelp(tailRef)  
  }

@tailrec
final def recHelp[A](asRef: AtomicReference[Stream[A]]): Stream[B] = 
  // Note: don't put the content of the holder into a local variable
  rec(asRef.getAndSet(null))

AtomicReference просто удобство, атомарность в этом случае не требуется, любой простой объект-держатель будет делать.

Также обратите внимание, что поскольку recHelp завершается в хвост потока Cons, поэтому он будет оцениваться только один раз, а Cons также выполняет синхронизацию.