rugyoga

rugyoga

Using streams and concurrency to demonstrate Elixir's potential to colleagues

I’m trying to create a simple example to show colleague the potential of Elixir.
My contrived example was to find the most frequently occurring alphanumeric
in a very large data file. It ruby, it’s simply:

It took 23 minutes.

My attempt in Elixir was to use File.stream! with params to chunk it into 1M chunks and apply 6 cores.

To my horror it took 60 mins of cpu and 10 minutes of real time.
What am I doing wrong?

Marked As Solved

akash-akya

akash-akya

My attempt at it.
Showing benchmark comparison with previous best, AlphamaxOptimMutable by krstfk. Mostly similar to krstfk’s solution but has slight difference in details.

defmodule AlphaMaxFreq do
  @chars Enum.to_list(1..?z)

  defp collect(<<>>, acc), do: acc

  defp collect(<<char::size(8), rest::binary>>, ref)
       when (char >= ?0 and char <= ?9) or (char >= ?a and char <= ?z) or
              (char >= ?A and char <= ?Z) do
    :counters.add(ref, char, 1)
    collect(rest, ref)
  end

  defp collect(<<_::utf8, rest::binary>>, acc), do: collect(rest, acc)
  defp collect(<<_::size(8), rest::binary>>, acc), do: collect(rest, acc)

  def max(file) do
    sch = :erlang.system_info(:schedulers)
    ref = :counters.new(length(@chars), [:write_concurrency])

    File.stream!(file, [], 10 * 64 * 1024)
    |> Task.async_stream(&collect(&1, ref), max_concurrency: sch, ordered: false)
    |> Stream.run()

    char = Enum.max_by(@chars, &:counters.get(ref, &1))
    IO.inspect([<<char::size(8)>>, :counters.get(ref, char)])
  end
end

Benchmark

Operating System: macOS
CPU Information: Intel(R) Core(TM) i7-8750H CPU @ 2.20GHz
Number of Available Cores: 12
Available memory: 16 GB
Elixir 1.9.4
Erlang 22.1.1

Benchmark suite executing with the following configuration:
warmup: 2 s
time: 5 s
memory time: 0 ns
parallel: 1
inputs: none specified
Estimated total run time: 14 s

Benchmarking AlphaMaxFreq...
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
["P", 1054798]
Benchmarking AlphamaxOptimMutable...
{"P", 1054798}
{"P", 1054798}
{"P", 1054798}

Name                           ips        average  deviation         median         99th %
AlphaMaxFreq                  1.45         0.69 s    ±11.58%         0.65 s         0.83 s
AlphamaxOptimMutable          0.33         3.04 s     ±6.69%         3.04 s         3.18 s

Comparison:
AlphaMaxFreq                  1.45
AlphamaxOptimMutable          0.33 - 4.42x slower +2.35 s

Extended statistics:

Name                         minimum        maximum    sample size                     mode
AlphaMaxFreq                  0.62 s         0.83 s              8                     None
AlphamaxOptimMutable          2.90 s         3.18 s              2                     None

Also Liked

kokolegorille

kokolegorille

This might give You some hints…

krstfk

krstfk

While I agree with @akash-akya that making such a piece of code faster is not necessarily the most interesting feature of Elixir, in that case there is room for improvement.
I think a good tool to figure out where a piece of code is spending its time is Benchee.
My hypothesis is that the code is slow because of the list creations in count_alpha. On the reduce side of the code, the [encoding: :utf8] option passed to File.stream! might be problematic.

To test this hypothesis I wrote a quick and dirty version (probably not the fastest and certainly not the most idiomatic or elegant) :

defmodule AlphamaxOptim do
  @valid_chars Enum.map(?a..?z, fn x -> <<x::utf8>> end) ++
                 Enum.map(?A..?Z, fn x -> <<x::utf8>> end) ++
                 Enum.map(?0..?9, fn x -> <<x::utf8>> end)

  @init_char_map Enum.map(@valid_chars, fn x -> {x, 0} end) |> Enum.into(%{})

  def count_alpha(string) do
    recursive_count(@init_char_map, string)
  end

  def recursive_count(acc, ""), do: acc

  for ch <- @valid_chars do
    def recursive_count(acc, unquote(ch) <> next) do
      recursive_count(%{acc | unquote(ch) => acc[unquote(ch)] + 1}, next)
    end
  end

  def recursive_count(acc, str) do
    case String.next_codepoint(str) do
      nil -> recursive_count(acc, "")
      {_, next} -> recursive_count(acc, next)
    end
  end

  def process(file) do
    File.stream!(file, [], 1_000_000)
    |> Task.async_stream(fn x -> count_alpha(IO.iodata_to_binary(x)) end,
      max_concurrency: 6,
      ordered: false
    )
    |> Stream.map(fn {:ok, x} -> x end)
    |> Enum.reduce(fn a, b -> Map.merge(a, b, fn _k, v1, v2 -> v1 + v2 end) end)
    |> Enum.max_by(fn {_k, v} -> v end)
    |> IO.inspect()
  end
end

I believe it is correct as it spits out the same result as the original. Ideally, it should be tested with something like proper, using the original as an anchor.

Then I benchmarked it against the original with a 100M file :

Benchmarking optimized100...
{"0", 8819253}
{"0", 8819253}
{"0", 8819253}
Benchmarking original100...
{"0", 8819253}
{"0", 8819253}

Name                   ips        average  deviation         median         99th %
optimized100          0.22         4.48 s     ±0.37%         4.48 s         4.49 s
original100         0.0183        54.79 s     ±0.00%        54.79 s        54.79 s

Comparison:
optimized100          0.22
original100         0.0183 - 12.23x slower +50.31 s

And with a 500M file (just to be sure) :

Benchmarking optimized500...
{"0", 31722406}
{"0", 31722406}
Benchmarking original500...
{"0", 31722406}
{"0", 31722406}

Name                   ips        average  deviation         median         99th %
optimized500        0.0404       0.41 min     ±0.00%       0.41 min       0.41 min
original500        0.00359       4.64 min     ±0.00%       4.64 min       4.64 min

Comparison:
optimized500        0.0404
original500        0.00359 - 11.25x slower +4.23 min

Then I wondered what would happen if we had mutability, only to remember that we did : ets tables or :counters. I’ve never used counters, so I tried it :

defmodule AlphamaxOptimMutable do
  @valid_chars Enum.map(?a..?z, fn x -> <<x::utf8>> end) ++
                 Enum.map(?A..?Z, fn x -> <<x::utf8>> end) ++
                 Enum.map(?0..?9, fn x -> <<x::utf8>> end)


  @indexed_chars Enum.with_index(@valid_chars, 1) |> Enum.into(%{})


  def count_alpha(ref, string) do
    recursive_count(ref, string)
  end

  def recursive_count(ref, ""), do: ref

  for ch <- @valid_chars do
    def recursive_count(ref, unquote(ch) <> next) do
      :counters.add(ref, @indexed_chars[unquote(ch)], 1)
      recursive_count(ref, next)
    end
  end

  def recursive_count(ref, str) do
    case String.next_codepoint(str) do
      nil -> recursive_count(ref, "")
      {_, next} -> recursive_count(ref, next)
    end
  end

  def process(file) do
    ref = :counters.new(length(@valid_chars), [:write_concurrency])

    File.stream!(file, [], 1_000_000)
    |> Task.async_stream(fn x -> count_alpha(ref, IO.iodata_to_binary(x)) end,
      max_concurrency: 6,
      ordered: false
    )
    |> Enum.map(fn x -> x end)

    @indexed_chars
    |> Enum.map(fn {ch, ix} -> {ch, :counters.get(ref, ix)} end)
    |> Enum.max_by(fn {_k, v} -> v end)
    |> IO.inspect()
  end
end

Benchmarked it again :

Name                           ips        average  deviation         median         99th %
optimized_mutable100          0.27         3.65 s     ±0.55%         3.65 s         3.66 s
original100                 0.0184        54.32 s     ±0.00%        54.32 s        54.32 s

Comparison:
optimized_mutable100          0.27
original100                 0.0184 - 14.88x slower +50.67 s

optimized_mutable500        0.0525       0.32 min     ±0.00%       0.32 min       0.32 min
original500                0.00360       4.63 min     ±0.00%       4.63 min       4.63 min

Comparison:
optimized_mutable500        0.0525
original500                0.00360 - 14.56x slower +4.31 min


There are probably much better ways to improve the original code (performance wise), but I think the methodology would be more or less similar.

akash-akya

akash-akya

Something similar is explored in https://www.youtube.com/watch?v=Y83p_VsvRFA.

Elixir is not that good with computation-intensive operations. Immutability does not help here either. That being said you can do a few things which might improve it.

The elixir code is not equivalent to ruby code

  1. you are using Enum, which is not lazy. It is essentially loading the whole file at once and passing whole new object between enum operations.
  2. you should merge many enum operations into one to avoid multiple intermediate object creation, you can replace all of it with an Enum.reduce. we can find max in one pass without creating map or intermediate list etc
  3. you can replace sort & fetch by max_by or min_by, ruby version does not sort
  4. you can use binary utf8 pattern matching (<<char::utf8, rest::binary>>) to avoid codepoints and regexp

you can see memory used by all objects in the VM using :observer.start()

I’m trying to create a simple example to show colleague the potential of Elixir.

I think you should focus on unique features elixir brings to the table such as fault tolerance and concurrency, rather than number crunching. Depending on the problem you are trying to solve, these things might be way more important than raw computation speed.

Where Next?

Popular in Questions Top

chokchit
** (DBConnection.ConnectionError) connection not available and request was dropped from queue after 2733ms. You can configure how long re...
New
New
Kurisu
For example for a current url like http://localhost:4000/cosmetic/products?_utf8=✓&amp;query=perfume&amp;page=2, I would like to get: ...
New
qwerescape
Is there a way to get the call stack or stack trace at any point in the code? Not from exceptions, but an expression that returns how the...
New
JulienCorb
I am trying to implement my new.html.eex file to create new posts on my website. new.html.eex: &lt;h1&gt;Create Post&lt;/h1&gt; &lt;%= ...
New
itssasanka
Hi all, Trying to get some more clarity over utc_datetime and naive_datetime for Ecto: The documentation above suggests that while ...
New
nobody
Hi! In PHP: $_SERVER[‘SERVER_ADDR’] - in Elixir? Searched the docs for ip address and the web, no good results. Thanks!
New
bsollish-terakeet
Credo is smart enough to check for (something like) this: assert length(the_list) == 0 with this response: Checking if an enum is empt...
New
komlanvi
Hi everyone, I was playing with phoenix liveView but I run into an issue. I have a form and want to validate each input text when the te...
New
vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
New

Other popular topics Top

hariharasudhan94
lets say i have a sample like a = 20; b = 10; if (a &gt; b) do {:ok, "a"} end if (a &lt; b) do {:ok, b} end if (a == b) do {:ok, "equa...
New
chrismccord
As promised, the first release candidate of Phoenix 1.3.0 is out! This release focuses on code generators with improved project structure...
New
lessless
I believe there are people here who are dealing with CSV files import on the daily basis, and since Excel is a really popular tool there ...
New
greenz1
I have a phoenix application from which a user can download multiple(5-6) files of size 1MB. I couldn’t find anything related to sending ...
New
belgoros
I’m not a pro in using Regex and can’t figure out why the following behaviour happens, especially if we take into account the difference ...
New
stefanchrobot
What’s the safe way to decode a JSON string into a struct? I want to avoid calling String.to_atom. Jason.decode can give me a map with st...
New
jay1
Why is it that the mnesia database isn’t the most preferred database for use in Elixir/Phoenix?
New
jason.o
In the code below, if the create action is not set to accept “extra_key” as an input, it errors out with a message shown above. Is there ...
New
shijith.k
I am trying to start a new phoenix project with elixir 1.9, but mix phx.new does not work. It says that ** (Mix) The task "phx.new" could...
New
jononomo
For some reason my phoenix channels are working for me in my local dev environment, but as soon as I deploy via Docker, I get a 403 error...
New

Latest on Elixir Forum

Elixir Forum

We're in Beta

About us Mission Statement