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.

dimitarvp

dimitarvp

Well, I did. :wink: Here’s the code:

defmodule AlphaMaxDimi 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)

  defp collect_chunk(blob, ref) do
    subref = :counters.new(length(@chars), [])
    collect(blob, subref)
    Enum.each(@chars, fn c ->
      :counters.add(ref, c, :counters.get(subref, c))
    end)
  end

  def max(path) do
    ref = :counters.new(length(@chars), [:write_concurrency])

    path
    |> File.stream!([], 512 * 1024)
    |> Task.async_stream(&collect_chunk(&1, ref), ordered: false)
    |> Stream.run()

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

The code is almost the same as @akash-akya’s but I was not happy that each and every character gets to write to an atomic counter. So I batched those. Each 512KB buffer read from the file gets their own private counters which are updated serially (single thread, e.g. single Erlang process), and then they get coalesced on to the global counters. This improved speeds roughly from 20% to 25%, depending on file sizes.

Here are the comparisons between mine and @akash-akya’s code on three different files.

18MB file (2019-annual/vernacular.txt), benchmark time 5s:

Name                   ips        average  deviation         median         99th %
Dimi (18MB)           7.67      130.37 ms     ±4.49%      127.94 ms      145.92 ms
Akash (18MB)          6.01      166.37 ms     ±3.07%      165.58 ms      182.67 ms

Comparison:
Dimi (18MB)           7.67
Akash (18MB)          6.01 - 1.28x slower +36.01 ms

300MB file (2019-annual/reference.txt), benchmark time 20s:

Name                    ips        average  deviation         median         99th %
Dimi (300MB)           0.49         2.06 s     ±0.55%         2.06 s         2.08 s
Akash (300MB)          0.40         2.50 s     ±1.61%         2.52 s         2.53 s

Comparison:
Dimi (300MB)           0.49
Akash (300MB)          0.40 - 1.21x slower +0.44 s

1.6GB file (2019-annual/taxa.txt), benchmark time 60s:

Name                    ips        average  deviation         median         99th %
Dimi (1.6GB)         0.0925        10.81 s     ±0.30%        10.82 s        10.84 s
Akash (1.6GB)        0.0755        13.25 s     ±0.46%        13.25 s        13.31 s

Comparison:
Dimi (1.6GB)         0.0925
Akash (1.6GB)        0.0755 - 1.22x slower +2.43 s

Where Next?

Popular in Questions Top

sergio
In Ruby, I can go: User.find_by(email: "foobar@email.com").update(email: "hello@email.com") How can I do something similar in Elixir? ...
New
gshaw
What is the idiomatic way of matching for not nil in Elixir? E.g., First way: defp halt_if_not_signed_in(conn, signed_in_account) when...
New
_russellb
I want to try my hand at web scraping. What tools/libraries do I need to use. I’m hoping to turn this into something professional so don’...
New
Werner
Hi, I’m using Ubuntu 18.04 and after updating to OTP-24.0 yesterday i have this warning when I run “mix local.hex”: 14:57:30.512 [warn] ...
New
Jim
As a follow up to my earlier question: I have the code compiling and running but not getting a successful login from the rest server. ...
New
hariharasudhan94
I would like to know what is the best IDE for elixir development?
New
Fl4m3Ph03n1x
Background Let’s assume I have a typical GenServer that receives messages as requests, does some operation in a DB and returns responses....
New
WestKeys
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New
Mooodi
Given a string, how can I get access to its character by index? Enum.at("my_string", 2) doesn't work. Or rather, not char, but a substr...
New
joeerl
Hello again - after a longish gap I’ve decided I really must dig into Elixir and see what’s been happening here - so I have a few questio...
New

Other popular topics Top

senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New
sergio
In Ruby, I can go: User.find_by(email: "foobar@email.com").update(email: "hello@email.com") How can I do something similar in Elixir? ...
New
yawaramin
In the Dialyzer docs ( http://erlang.org/doc/man/dialyzer.html#requesting-or-suppressing-warnings-in-source-files ), there is a way to tu...
New
JorisKok
I have a server on AWS, and was running a load test using artillery. When looking at the Phoenix dashboard I see the Ports going to 100% ...
New
AstonJ
You’re a programmer, so you don’t need spoon feeding with the conventional drivel about “this is an integer.” No. You need to know what’s...
New
KronicDeth
Elixir plugin for JetBrain’s IntelliJ Platform (including Rubymine) This is a plugin that adds support for Elixir to JetBrains IntelliJ...
289 35421 110
New
danschultzer
None of the current solutions worked well for me, so I went ahead and built a user management system from scratch. This project took far...
548 27727 240
New
aadeshere1
I have a another noob question about loop. Since elixir is immutable, while loop is not directly possible. total = 10 while total != 0 ...
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
vrod
I am using the Starship cross-shell prompt – it seems pretty nice, but I get some errors: [WARN] - (starship::utils): Executing command ...
New

We're in Beta

About us Mission Statement