title: Avoiding memory leaks with Scalaz 7 zipWithIndex/group enumeratees
body: <p><strong>Background</strong></p>
  
  <p>As noted in <a href="https://stackoverflow.com/questions/19059831/asynchronous-iteratee-processing-in-scalaz">this question</a>, I'm using Scalaz 7 iteratees to process a large (i.e., unbounded) stream of data in constant heap space.</p>
  
  <p>My code looks like this:</p>
  
  <pre><code>type ErrorOrT[M[+_], A] = EitherT[M, Throwable, A]
  type ErrorOr[A] = ErrorOrT[IO, A]
  
  def processChunk(c: Chunk, idx: Long): Result
  
  def process(data: EnumeratorT[Chunk, ErrorOr]): IterateeT[Vector[(Chunk, Long)], ErrorOr, Vector[Result]] =
    Iteratee.fold[Vector[(Chunk, Long)], ErrorOr, Vector[Result]](Nil) { (rs, vs) =&gt;
      rs ++ vs map { 
        case (c, i) =&gt; processChunk(c, i) 
      }
    } &amp;= (data.zipWithIndex mapE Iteratee.group(P))
  </code></pre>
  
  <p><strong>The Problem</strong> </p>
  
  <p>I seem to have run into a memory leak, but I'm not familiar enough with Scalaz/FP to know whether the bug is in Scalaz or in my code. Intuitively, I expect this code to require only (on the order of) <em>P</em> times the <code>Chunk</code>-size space.</p>
  
  <p>Note: I found <a href="https://stackoverflow.com/questions/16228154/scalaz-7-iteratee-to-process-large-zip-file-outofmemoryerror">a similar question</a> in which an <code>OutOfMemoryError</code> was encountered, but my code is not using <code>consume</code>.</p>
  
  <p><strong>Testing</strong></p>
  
  <p>I ran some tests to try and isolate the problem. To summarize, the leak only appears to arise when both <code>zipWithIndex</code> and <code>group</code> are used.</p>
  
  <pre><code>// no zipping/grouping
  scala&gt; (i1 &amp;= enumArrs(1 &lt;&lt; 25, 128)).run.unsafePerformIO
  res47: Long = 4294967296
  
  // grouping only
  scala&gt; (i2 &amp;= (enumArrs(1 &lt;&lt; 25, 128) mapE Iteratee.group(4))).run.unsafePerformIO
  res49: Long = 4294967296
  
  // zipping and grouping
  scala&gt; (i3 &amp;= (enumArrs(1 &lt;&lt; 25, 128).zipWithIndex mapE Iteratee.group(4))).run.unsafePerformIO
  java.lang.OutOfMemoryError: Java heap space
  
  // zipping only
  scala&gt; (i4 &amp;= (enumArrs(1 &lt;&lt; 25, 128).zipWithIndex)).run.unsafePerformIO
  res51: Long = 4294967296
  
  // no zipping/grouping, larger arrays
  scala&gt; (i1 &amp;= enumArrs(1 &lt;&lt; 27, 128)).run.unsafePerformIO
  res53: Long = 17179869184
  
  // zipping only, larger arrays
  scala&gt; (i4 &amp;= (enumArrs(1 &lt;&lt; 27, 128).zipWithIndex)).run.unsafePerformIO
  res54: Long = 17179869184
  </code></pre>
  
  <p>Code for the tests:</p>
  
  <pre><code>import scalaz.iteratee._, scalaz.effect.IO, scalaz.std.vector._
  
  // define an enumerator that produces a stream of new, zero-filled arrays
  def enumArrs(sz: Int, n: Int) = 
    Iteratee.enumIterator[Array[Int], IO](
      Iterator.continually(Array.fill(sz)(0)).take(n))
  
  // define an iteratee that consumes a stream of arrays 
  // and computes its length
  val i1 = Iteratee.fold[Array[Int], IO, Long](0) { 
    (c, a) =&gt; c + a.length 
  }
  
  // define an iteratee that consumes a grouped stream of arrays 
  // and computes its length
  val i2 = Iteratee.fold[Vector[Array[Int]], IO, Long](0) { 
    (c, as) =&gt; c + as.map(_.length).sum 
  }
  
  // define an iteratee that consumes a grouped/zipped stream of arrays
  // and computes its length
  val i3 = Iteratee.fold[Vector[(Array[Int], Long)], IO, Long](0) {
    (c, vs) =&gt; c + vs.map(_._1.length).sum
  }
  
  // define an iteratee that consumes a zipped stream of arrays
  // and computes its length
  val i4 = Iteratee.fold[(Array[Int], Long), IO, Long](0) {
    (c, v) =&gt; c + v._1.length
  }
  </code></pre>
  
  <p><strong>Questions</strong></p>
  
  <ul>
  <li>Is the bug in my code?</li>
  <li>How can I make this work in constant heap space?</li>
  </ul>