Optimize binary_slice/2 with step > 1

[AlexGx]

Assisted-by: GPT-5.6

This avoids collector and per-byte binary allocations (into: <<first_byte>> invokes Collectable.BitString,
compiler only enables the native bitstring-comprehension path for into: <<>>).
New implementation scales approximately linearly with the number of output elements, with a nearly constant per-element cost.

Im not completely sure whether this is needed, or maybe compiler should be improved instead...

Case	Before	After	Runtime change	Allocated Mem	Reductions
16 B, step 2	181.58 μs	102.73 μs	1.77× faster	1.07 MB → 0.122 MB	39.31 K → 12.02 K
1 KiB, step 2	13.68 ms	1.90 ms	7.21× faster	70.78 MB → 0.160 MB	2.18 M → 0.52 M
64 KiB, step 100	17.91 ms	2.45 ms	7.30× faster	90.20 MB → 0.160 MB	2.78 M → 0.66 M
64 KiB, step 2	101.14 ms	11.06 ms	9.14× faster	450.04 MB → 0.016 MB	13.71 M → 3.28 M
1 KiB, step 1, full range	19.88 μs	20.03 μs	~same	0 B → 0 B	3.00 K → 3.00 K
64 KiB, step 1, full range	19.56 μs	19.85 μs	~same	0 B → 0 B	3.00 K → 3.00 K
16 B, step 100, 1 output byte*	46.41 μs	71.04 μs	1.53× slower	171.88 KB → 125 KB	14.03 K → 5.02 K
16 B, step 15, 2 output bytes*	75.22 μs	83.69 μs	1.11× slower	312.50 KB → 125 KB	19.05 K → 6.02 K

• new implementation is slightly slower for very small stepped slices that produce only one or two bytes output.

Bench

Mix.install([{:benchee, "== 1.5.1"}])

defmodule Before do
  def binary_slice(binary, first..last//step)
      when is_binary(binary) and step > 0 do
    total = byte_size(binary)

    first = if first < 0, do: max(first + total, 0), else: first
    last = if last < 0, do: last + total, else: last

    amount = last - first + 1

    if first < total and amount > 0 do
      part = binary_part(binary, first, min(amount, total - first))

      if step == 1 do
        part
      else
        <<first_byte, rest::binary>> = part
        for <<_::size(^step - 1)-bytes, byte <- rest>>, into: <<first_byte>>, do: <<byte>>
      end
    else
      ""
    end
  end

  def binary_slice(binary, _.._//_ = range) when is_binary(binary) do
    raise ArgumentError,
          "binary_slice/2 does not accept ranges with negative steps, got: #{inspect(range)}"
  end
end

defmodule After do
  def binary_slice(binary, first..last//step)
      when is_binary(binary) and step > 0 do
    total = byte_size(binary)

    first = if first < 0, do: max(first + total, 0), else: first
    last = if last < 0, do: last + total, else: last

    amount = last - first + 1

    if first < total and amount > 0 do
      part = binary_part(binary, first, min(amount, total - first))

      if step == 1 do
        part
      else
        <<first_byte, rest::binary>> = part
        tail = for <<_::size(^step - 1)-bytes, byte <- rest>>, into: <<>>, do: <<byte>>
        <<first_byte, tail::binary>>
      end
    else
      ""
    end
  end

  def binary_slice(binary, _.._//_ = range) when is_binary(binary) do
    raise ArgumentError,
          "binary_slice/2 does not accept ranges with negative steps, got: #{inspect(range)}"
  end
end

defmodule Batch do
  # Both implementations use the same batch size for a given input.
  def run(slicer, {binary, range, iterations}), do: repeat(slicer, binary, range, iterations)

  defp repeat(_slicer, _binary, _range, 0), do: :ok

  defp repeat(slicer, binary, range, remaining) do
    slicer.binary_slice(binary, range)
    repeat(slicer, binary, range, remaining - 1)
  end
end

bytes = for byte <- 0..255, into: <<>>, do: <<byte>>
small = binary_part(bytes, 0, 16)
medium = :binary.copy(bytes, 4)
large = :binary.copy(bytes, 256)

inputs = %{
  "step 1: empty input" => {"", 0..10//1},
  "step 1: 16 bytes, full range" => {small, 0..-1//1},
  "step 1: 1 KiB, full range" => {medium, 0..-1//1},
  "step 1: 64 KiB, full range" => {large, 0..-1//1},
  "step 1: 64 KiB, middle 32 KiB" => {large, 16_384..49_151//1},
  "step 1: 64 KiB, middle 32 bytes" => {large, 100..131//1},
  "step 1: 64 KiB, negative indices (1 KiB)" => {large, -2048..-1025//1},
  "step 1: 64 KiB, clipped last index" => {large, 65_500..70_000//1},
  "step 1: 64 KiB, start beyond end" => {large, 65_536..70_000//1},
  "step 2: empty input" => {"", 0..10//2},
  "16 bytes, step 2" => {small, 0..-1//2},
  "16 bytes, step 15 (two output bytes)" => {small, 0..-1//15},
  "16 bytes, step 16 (first byte only)" => {small, 0..-1//16},
  "16 bytes, step 17 (first byte only)" => {small, 0..-1//17},
  "16 bytes, step 100 (first byte only)" => {small, 0..-1//100},
  "1 KiB, step 2" => {medium, 0..-1//2},
  "64 KiB, step 100" => {large, 0..-1//100},
  "64 KiB, step 2" => {large, 0..-1//2}
}

# Verify equivalence and choose batch sizes outside the timed samples.
# Large stepped outputs use 100 calls so that a 10-second timing window
# includes many samples. Fast paths retain 1,000 calls to reduce timer noise.
inputs =
  Map.new(inputs, fn {name, {binary, range}} ->
    expected = Before.binary_slice(binary, range)

    unless expected == After.binary_slice(binary, range) do
      raise "Results differ for #{name}"
    end

    iterations = if range.step > 1 and byte_size(expected) >= 1024, do: 100, else: 1_000
    {"#{name} [#{iterations} calls/sample]", {binary, range, iterations}}
  end)

{options, [], []} =
  OptionParser.parse(System.argv(), strict: [quick: :boolean, reverse: :boolean])

jobs = [
  {"before: Collectable.BitString", &Batch.run(Before, &1)},
  {"after: binary comprehension", &Batch.run(After, &1)}
]

jobs = if options[:reverse], do: Enum.reverse(jobs), else: jobs

jobs =
  jobs
  |> Enum.with_index(1)
  |> Map.new(fn {{name, fun}, index} -> {"#{index}. #{name}", fun} end)

timing =
  if options[:quick] do
    IO.puts("SMOKE CHECK ONLY: use a full run for performance comparisons.")
    [warmup: 0, time: 0.05, memory_time: 0.05, reduction_time: 0.05]
  else
    [warmup: 2, time: 10, memory_time: 1, reduction_time: 1]
  end

IO.puts("Timings, memory, and reductions are per batch; batch size appears in each input name.")
IO.puts("Divide by batch size for per-slice values. Memory excludes off-heap binary payloads.")

Benchee.run(
  jobs,
  timing ++
    [
      inputs: inputs,
      parallel: 1,
      formatters: [{Benchee.Formatters.Console, extended_statistics: true}]
    ]
)

[lukaszsamson]

Can't we apply this optimization to Collectable.BitString? It would improve all usages of into: bitstring not only this particular one

[lukaszsamson]

Something like this in lib/elixir/src/elixir_erl_for.erl:

%% Add before the generic build_into/7 clause.
build_into(Ann, Clauses, Expr, {bin, _, _} = Into, Uniq, InitVars, S) ->
  {Prefix, SP} = build_var(Ann, S),
  {Tail, ST} = build_inline(
    Ann, Clauses, Expr, {bin, Ann, []}, Uniq, InitVars, SP
  ),
  Result = {bin, Ann, [
    {bin_element, Ann, Prefix, default, [bitstring]},
    {bin_element, Ann, Tail, default, [bitstring]}
  ]},
  {{block, Ann, [
    {match, Ann, Prefix, Into},
    Result
  ]}, ST};

[josevalim]

Let’s try optimizing the compiler first, yeah, then we can revisit this!