Lookahead benchmarks - http://stackoverflow.com/questions/29136874/enumerable-stream-with-look-ahead

Benchmark results

Benchmarks completed on Windows 7 Lenovo i5 laptop with Elixir 1.0.3, Erlang 17.4

Elixir.PatrickSuspend.lookahead(1)           10000   104.90 µs/op
Elixir.PatrickSuspend.lookahead(3)           10000   107.20 µs/op
Elixir.PatrickSuspend.lookahead(2)           10000   114.30 µs/op
Elixir.PatrickSuspend.lookahead(10)          10000   119.60 µs/op
Elixir.PatrickSuspend.lookahead(5)           10000   123.50 µs/op
Elixir.Spits.lookahead(1)                    10000   174.00 µs/op
Elixir.Spits.lookahead(10)                   10000   186.80 µs/op
Elixir.Spits.lookahead(3)                    10000   189.60 µs/op
Elixir.Spits.lookahead(5)                    10000   191.40 µs/op
Elixir.Spits.lookahead(2)                    10000   207.80 µs/op
Elixir.PatrickSuspend.lookahead(50)          10000   220.80 µs/op
Elixir.PatrickChunk.lookahead(1)              5000   310.60 µs/op
Elixir.Spits.lookahead(50)                    5000   320.60 µs/op
Elixir.PatrickTransform.lookahead(1)          5000   357.00 µs/op
Elixir.PatrickTransform.lookahead(3)          5000   387.60 µs/op
Elixir.PatrickTransform.lookahead(2)          5000   392.60 µs/op
Elixir.PatrickChunk.lookahead(2)              5000   398.00 µs/op
Elixir.PatrickTransform.lookahead(10)         5000   403.20 µs/op
Elixir.Spits.lookahead(100)                   5000   409.00 µs/op
Elixir.PatrickTransform.lookahead(5)          5000   409.00 µs/op
Elixir.PatrickChunk.lookahead(3)              5000   448.20 µs/op
Elixir.PatrickTransform.lookahead(50)         5000   518.60 µs/op
Elixir.PatrickSuspend.lookahead(100)          5000   542.80 µs/op
Elixir.PatrickChunk.lookahead(5)              5000   607.00 µs/op
Elixir.Jose.lookahead(2)                      5000   643.40 µs/op
Elixir.Jose.lookahead(1)                      5000   647.60 µs/op
Elixir.PatrickTransform.lookahead(100)        2000   750.50 µs/op
Elixir.Jose.lookahead(3)                      2000   789.50 µs/op
Elixir.Jose.lookahead(10)                     2000   793.00 µs/op
Elixir.PatrickChunk.lookahead(10)             2000   823.50 µs/op
Elixir.Jose.lookahead(5)                      2000   852.00 µs/op
Elixir.Jose.lookahead(50)                     1000   1390.00 µs/op
Elixir.Jose.lookahead(100)                    1000   1947.00 µs/op
Elixir.PatrickChunk.lookahead(50)              500   3058.00 µs/op
Elixir.PatrickChunk.lookahead(100)             500   5092.00 µs/op
Elixir.PatrickUnfold.lookahead(100)              1   1241000.00 µs/op
Elixir.PatrickUnfold.lookahead(50)               1   1345000.00 µs/op
Elixir.PatrickUnfold.lookahead(1)                1   1484000.00 µs/op
Elixir.PatrickUnfold.lookahead(10)               1   1499000.00 µs/op
Elixir.PatrickUnfold.lookahead(3)                1   1542000.00 µs/op
Elixir.PatrickUnfold.lookahead(5)                1   1647000.00 µs/op
Elixir.PatrickUnfold.lookahead(2)                1   1804000.00 µs/op

lookahead.ex

defmodule Jose do
  def lookahead(enumerable, n) when n > 0 do
    enumerable
    |> Stream.chunk(n + 1, 1, [])
    |> Stream.flat_map(fn list ->
        length = length(list)
        if length < n + 1 do
          [list|Enum.scan(1..n-1, list, fn _, acc -> Enum.drop(acc, 1) end)]
        else
          [list]
        end
      end)
  end
end

defmodule PatrickResource do
  def lookahead(enumerable, n) when n > 0 do
    Stream.resource(
      fn -> Enum.split(enumerable, n+1) end,
      fn {buffer, rest} = acc ->
        case buffer do
          [_] ->
            {:halt, acc}
          _x ->
            curr = Enum.drop(buffer, 1) ++ Enum.take(rest, 1)
            {[curr], {curr, Enum.drop(rest, 1)}}
        end
      end,
      fn {result, _rest} -> result end
    )
  end
end

defmodule PatrickUnfold do
  def lookahead(stream, n) when n > 0 do
    Stream.unfold split(stream, n+1), fn
      {[], _stream} ->
        nil
      {[_ | buffer] = current, stream} ->
        {value, stream} = split(stream, 1)
        {current, {buffer ++ value, stream}}
    end
  end

  defp split(stream, n) do
    {Enum.take(stream, n), Stream.drop(stream, n)}
  end
end

defmodule PatrickChunk do
  def lookahead(enum, n) do
    stop = make_ref
    enum
    |> Stream.concat(List.duplicate(stop, n))
    |> Stream.chunk(n+1, 1)
    |> Stream.map(&Enum.reject(&1, fn x -> x == stop end))
  end
end

defmodule PatrickTransform do
  def lookahead(enum, n) do
    stop = make_ref
    enum
    |> Stream.concat(List.duplicate(stop, n+1))
    |> Stream.transform([], fn val, acc ->
      case {val, acc} do
        {^stop, []}                         -> {[]   , []           }
        {^stop, [_|rest] = buf}             -> {[buf], rest         }
        {val  , buf} when length(buf) < n+1 -> {[]   , buf ++ [val] }
        {val  , [_|rest] = buf}             -> {[buf], rest ++ [val]}
      end
    end)
  end
end

defmodule PatrickSuspend do
  def lookahead(enum, n) do
    step = fn val, _acc -> {:suspend, val} end
    next = &Enumerable.reduce(enum, &1, step)
    &do_lookahead(n, :buffer, [], next, &1, &2)
  end

  # stream suspended
  defp do_lookahead(n, state, buf, next, {:suspend, acc}, fun) do
    {:suspended, acc, &do_lookahead(n, state, buf, next, &1, fun)}
  end

  # stream halted
  defp do_lookahead(_n, _state, _buf, _next, {:halt, acc}, _fun) do
    {:halted, acc}
  end

  # initial buffering
  defp do_lookahead(n, :buffer, buf, next, {:cont, acc}, fun) do
    case next.({:cont, []}) do
      {:suspended, val, next} ->
        new_state = if length(buf) < n, do: :buffer, else: :emit
        do_lookahead(n, new_state, buf ++ [val], next, {:cont, acc}, fun)
      {_, _} ->
        do_lookahead(n, :emit, buf, next, {:cont, acc}, fun)
    end
  end

  # emitting
  defp do_lookahead(n, :emit, [_|rest] = buf, next, {:cont, acc}, fun) do
    case next.({:cont, []}) do
      {:suspended, val, next} ->
        do_lookahead(n, :emit, rest ++ [val], next, fun.(buf, acc), fun)
      {_, _} ->
        do_lookahead(n, :emit, rest, next, fun.(buf, acc), fun)
    end
  end

  # buffer empty, halting
  defp do_lookahead(_n, :emit, [], _next, {:cont, acc}, _fun) do
    {:halted, acc}
  end
end

defmodule Spits do
  def lookahead(enum, n) when n >= 0 do
    reducer = fn -> Enumerable.reduce(enum, {:cont, {0, []}}, fn
      item, {c, list} when c < n  -> {:cont, {c+1, list ++ [item]}} # Build up the first list
      item, {c, list} when c == n -> {:suspend, {c+1, list ++ [item]}} # Suspend on first full list
      item, {c, [_|list]} -> {:suspend, {c, list ++ [item]}} # Remove the first item and emit
      end)
    end

    Stream.resource(reducer,
      fn
        {:suspended, {_, list} = acc , fun} -> {[list], fun.({:cont, acc})} 
        {:halted, _} = result -> lookahead_trail(n, result) # Emit the trailing items
        {:done, _} = result -> lookahead_trail(n, result) # Emit the trailing items
      end,
      fn
        {:suspended, acc, fun} -> fun.({:halt, acc}) # Ensure the reducer is halted after suspend
        _ ->
      end)
  end

  defp lookahead_trail(n, acc) do
    case acc do
      {action, {c, [_|rest]}} when c > n -> {[], {action, {c-1, rest}}} # List already emitted here
      {action, {c, [_|rest] = list}} -> {[list], {action, {c-1, rest}}} # Emit the next tail item
      acc -> {:halt, acc } # Finish of the stream
    end
  end
end

lookahead_bench.exs

defmodule LookaheadBench do
  use Benchfella

  @list Enum.to_list(1..500)

  [1, 2, 3, 5, 10, 50, 100] |> Enum.each fn n ->
    [Spits, PatrickSuspend, PatrickUnfold, PatrickChunk, PatrickTransform, Jose] |> Enum.each fn m ->
      @lookahead_n n
      @lookahead_m m

      bench "#{m}.lookahead(#{n})" do
        @list |> @lookahead_m.lookahead(@lookahead_n) |> Stream.run
      end
    end
  end
end