Initial commit

Author: rrcook

.formatter.exs

@@ -0,0 +1,4 @@
+# Used by "mix format"
+[
+  inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"]
+]

.gitignore

@@ -0,0 +1,26 @@
+# The directory Mix will write compiled artifacts to.
+/_build/
+
+# If you run "mix test --cover", coverage assets end up here.
+/cover/
+
+# The directory Mix downloads your dependencies sources to.
+/deps/
+
+# Where third-party dependencies like ExDoc output generated docs.
+/doc/
+
+# Ignore .fetch files in case you like to edit your project deps locally.
+/.fetch
+
+# If the VM crashes, it generates a dump, let's ignore it too.
+erl_crash.dump
+
+# Also ignore archive artifacts (built via "mix archive.build").
+*.ez
+
+# Ignore package tarball (built via "mix hex.build").
+brc-*.tar
+
+# Temporary files, for example, from tests.
+/tmp/

README.md

@@ -0,0 +1,21 @@
+# Brc
+
+My attempt at the billion row challenge in Elixir. 
+Elixir 1.16.0 - important because the argument order for File.stream changed between 1.15 and 1.16.
+Vanilla elixir, no extra libraries to run. I am including eflambe to run & make flamegraphs to help tune performance.
+
+brc_city was my first attempt. It used a process for every city. Don't do this, most of your app's time will be spent in process sleeping.
+
+brc uses a pool of workers. Each worker receives a list of cities. Fewer workers with more work eliminates idle time.
+
+If using eflambe, run something like
+
+iex -S mix
+
+:eflambe.apply({Brc, :run_file_buf, ["measurements.txt"]}, [output_format: :brendan_gregg, open: :speedscope])
+
+
+## Installation
+
+After cloning, run 'mix deps.get' to get eflambe. Then 'mix escript.build', then './brc measurments.txt'
+

lib/brc.ex

@@ -0,0 +1,109 @@
+defmodule Brc do
+  use Agent
+
+  @pool_size 8
+  @job_size  10_000
+
+  # each process has its own partial map of the cities
+  # for each line, parse out the city name and temperature * 10, keeping the numbers as integers
+  # if a city is already in the map, to the min, max and counting on it
+  def process_lines(state_map, lines) do
+
+    Enum.reduce(lines, state_map, fn line, acc_map ->
+      [city, temperature_text] = :binary.split(line, ";")
+      [t1, t2] = :binary.split(temperature_text, ".")
+      temperature = :erlang.binary_to_integer(t1 <> t2)
+
+      Map.update(
+        acc_map,
+        city,
+        {temperature, 1, temperature, temperature},
+        fn {min_temp, count, sum, max_temp} ->
+          {min(min_temp, temperature), count + 1, sum + temperature, max(max_temp, temperature)}
+        end
+      )
+    end)
+  end
+
+  def test_file_buf(filename) do
+    file_stream = File.stream!(filename, [:read_ahead], 65_536)
+    # IO.inspect(file_stream)
+
+    Stream.transform(file_stream, <<>>, fn elem, acc ->
+      [new_acc | output_enum] = :binary.split(acc <> elem, <<"\n">>, [:global]) |> Enum.reverse()
+      {output_enum, new_acc}
+    end)
+    |> Stream.chunk_every(@job_size)
+    |> Enum.map(fn _ -> :ok end)
+  end
+
+
+  def run_file_buf(filename) do
+
+    :ets.new(:brc, [:public, :named_table])
+    :ets.insert(:brc, {:index, -1})
+
+    worker_pool =
+      Enum.map(1..@pool_size, fn _ ->
+        Agent.start_link(fn -> %{} end) |> elem(1)
+      end)
+
+    file_stream = File.stream!(filename, 65_536, [:read_ahead])
+    # IO.inspect(file_stream)
+
+    # transform the bundles of bytes into lines separated by \n
+    # this is faster than letting File.stream do :lines by itself
+    Stream.transform(file_stream, <<>>, fn elem, acc ->
+      [new_acc | output_enum] = :binary.split(acc <> elem, <<"\n">>, [:global]) |> Enum.reverse()
+      {output_enum, new_acc}
+    end)
+    |> Stream.chunk_every(@job_size)
+    |> Stream.each(fn job ->
+      # round robin through the pool of workers
+      index = :ets.update_counter(:brc, _key = :index, _increment_by = 1)
+      # this is basically using Agents as less-hassle GenServers
+      Agent.cast(Enum.at(worker_pool, rem(index, @pool_size)), Brc, :process_lines, [job])
+    end)
+    |> Stream.run
+
+    # synchronous here to make sure all of the workers are finished
+    pool_maps = Enum.map(worker_pool, fn pid -> Agent.get(pid, fn state -> state end) end)
+
+    # feeding all other maps into one, first one is the chosen one
+    [head | tail] = pool_maps
+
+    # if each map has a city, merge it into the chosen one
+    combined_map =
+      Enum.reduce(tail, head, fn elem, acc ->
+        Map.merge(acc, elem, fn _key, {min1, count1, sum1, max1}, {min2, count2, sum2, max2} ->
+          {min(min1, min2), count1 + count2, sum1 + sum2, max(max1, max2)}
+        end)
+      end)
+
+    # city for sorting, plus string we will output
+    keys_strings =
+      Map.keys(combined_map)
+      |> Enum.map(fn key ->
+        {min_temp, count, sum, max_temp} = Map.get(combined_map, key)
+        {key,
+         "#{key}=#{min_temp / 10}/#{:erlang.float_to_binary(sum / (count * 10), decimals: 1)}/#{max_temp / 10}"}
+      end)
+
+    # sort the strings by city/key then discard the key, keep the output
+    sorted_strings = Enum.sort_by(keys_strings, &elem(&1, 0)) |> Enum.map(&elem(&1, 1))
+
+    # output in brc format
+    IO.puts("{#{Enum.join(sorted_strings, ", ")}}")
+
+  end
+
+  def main(args) do
+    {uSec, :ok} =
+      :timer.tc(fn ->
+        run_file_buf(Enum.at(args, 0))
+        :ok
+      end)
+
+    IO.puts("It took #{uSec / 1000} milliseconds")
+  end
+end

lib/brc_city.ex

@@ -0,0 +1,90 @@
+defmodule BrcCity do
+  use Agent
+
+  def dispatch_line(state, line) do
+    [city, temperature_text] = :binary.split(line, ";")
+    [t1, t2] = :binary.split(temperature_text, ".")
+    temperature = :erlang.binary_to_integer(t1 <> t2)
+
+    {new_state, pid} = case Map.get(state, city) do
+      nil ->
+        {:ok, pid} = Agent.start_link(fn -> {1000, 0, 0, -1000} end)
+        {Map.put(state, city, pid), pid}
+      existing_pid ->
+        {state, existing_pid}
+    end
+
+    Agent.cast(pid, Brc, :update_worker, [temperature])
+    new_state
+  end
+
+  def dispatch_get(state) do
+    keys_strings = Map.keys(state)
+    |> Enum.map(fn key ->
+      pid = Map.get(state, key)
+      Agent.get(pid, fn {min_temp, count, sum, max_temp} ->
+        {key, "#{key}=#{min_temp / 10}/#{:erlang.float_to_binary(sum / (count * 10), [decimals: 1])}/#{max_temp / 10}"}
+      end)
+    end)
+    sorted_strings = Enum.sort_by(keys_strings, &(elem(&1, 0))) |> Enum.map(&(elem(&1, 1)))
+
+    "{#{Enum.join(sorted_strings, ", ")}}"
+  end
+
+  @spec update_worker({integer(), integer(), integer(), integer()}, integer()) :: {integer(), integer(), integer(), integer()}
+  def update_worker({min_temp, count, sum, max_temp}, temperature) do
+    {
+      min(min_temp, temperature),
+      count + 1,
+      sum + temperature,
+      max(max_temp, temperature)
+    }
+  end
+
+  defp gather_lines(<<"\n", rest::binary>>, acc, dispatch_pid) do
+    # do things with acc
+    Agent.cast(dispatch_pid, Brc, :dispatch_line, [acc])
+    gather_lines(rest, <<>>, dispatch_pid)
+  end
+
+  defp gather_lines(<<c::binary-size(1), rest::binary>>, acc, dispatch_pid) do
+    gather_lines(rest, acc <> c, dispatch_pid)
+  end
+
+  defp gather_lines(<<>>, acc, _dispatch_pid) do
+    acc
+  end
+
+  def run_file_buf(filename) do
+
+    {:ok, dispatch_pid} = Agent.start_link(fn -> %{} end)
+
+    file_stream = File.stream!(filename, 16384, [:read_ahead])
+    IO.inspect(file_stream)
+
+    worker_stream = Stream.transform(file_stream, <<>>,
+      fn elem, acc ->
+        # IO.puts(elem)
+        new_acc = gather_lines(acc <> elem, <<>>, dispatch_pid)
+        {[], new_acc}
+      end
+    )
+
+    Stream.run(worker_stream)
+
+    my_list = Agent.get(dispatch_pid, &dispatch_get/1)
+    IO.puts(my_list)
+
+  end
+
+  def main(args) do
+
+    {uSec, :ok} = :timer.tc(
+      fn ->
+        run_file_buf(Enum.at(args, 0))
+        :ok
+      end
+    )
+    IO.puts("It took #{uSec / 1000} milliseconds")
+  end
+end

mix.exs

@@ -0,0 +1,30 @@
+defmodule Brc.MixProject do
+  use Mix.Project
+
+  def project do
+    [
+      app: :brc,
+      escript: [main_module: Brc],
+      version: "0.1.0",
+      elixir: "~> 1.15",
+      start_permanent: Mix.env() == :prod,
+      deps: deps()
+    ]
+  end
+
+  # Run "mix help compile.app" to learn about applications.
+  def application do
+    [
+      extra_applications: [:logger]
+    ]
+  end
+
+  # Run "mix help deps" to learn about dependencies.
+  defp deps do
+    [
+      {:eflambe, "~> 0.3.1"}
+      # {:dep_from_hexpm, "~> 0.3.0"},
+      # {:dep_from_git, git: "https://github.com/elixir-lang/my_dep.git", tag: "0.1.0"}
+    ]
+  end
+end

test/brc_test.exs

@@ -0,0 +1,8 @@
+defmodule BrcTest do
+  use ExUnit.Case
+  doctest Brc
+
+  test "greets the world" do
+    assert Brc.hello() == :world
+  end
+end

test/test_helper.exs

@@ -0,0 +1 @@
+ExUnit.start()