   defp build_producer!(stream, pipes) do
     num_pipes = pipes |> Enum.count
 
-    Stream.transform stream, fn -> 0 end, fn line, index ->
-      pipe_index = index |> rem(num_pipes)
-      { line_receiver, relay } = pipes |> Enum.fetch!(pipe_index)
-      line_receiver |> send({ index, line })
+    stream
+     |> Stream.chunk(num_pipes, num_pipes, [])
+     |> Stream.transform fn -> 0 end, fn lines, index ->
+       mapped = lines |> Stream.with_index |> Enum.map fn { line, i } ->
+        pipe_index = (index + i) |> rem(num_pipes)
+        { line_receiver, relay } = pipes |> Enum.fetch!(pipe_index)
+        line_receiver |> send({ index + i, line })
 
-      { [{ relay, index }], index + 1 }
+        { relay, index + i }
+       end
+
+      { [mapped], index + num_pipes }
     end, fn index -> 
       pipes |> Enum.each fn { line_receiver, relay } ->
         relay |> send :halt

  alias CSV.Lexer, as: Lexer
  alias CSV.Relay, as: Relay

  @num_pipes 8

  @doc """
  Decode a stream of comma-separated lines into a table.

    * `:delimiter`   – The delimiter token to use, defaults to `\r\n`. Must be a string.
    * `:strip_cells` – When set to true, will strip whitespace from cells. Defaults to false.
    * `:num_pipes`   – The number of parallel operations to run when producing the stream.
    * `:headers`     – When set to `true`, will take the first row of the csv and use it as
      header values.
      When set to a list, will use the given list as header values.
