Switched from a file stream based approach

to a prim_file and buffer based approach.
Use erlang's prim_file to read in large buffers then do a read_line to reach
the next new line after the buffer.

Author: rrcook

lib/brc.ex

@@ -1,73 +1,84 @@
 defmodule Brc do
   use Agent
 
-  @pool_size 8
-  @job_size  10_000
+  @pool_size :erlang.system_info(:logical_processors)
+  @blob_size 100_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
+  def process_lines(state_map, buffer) do
+    lines = :binary.split(buffer, <<"\n">>, [:global])
 
     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
-      )
+      case line do
+        "" ->
+          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)
   end
 
-  def test_file_buf(filename) do
-    file_stream = File.stream!(filename, [:read_ahead], 65_536)
-    # IO.inspect(file_stream)
+  def process_file(file, worker_pool) do
+    case :prim_file.read(file, @blob_size) do
+      :eof ->
+        :ok
 
-    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
+      {:ok, buffer} ->
+        buffer =
+          case :prim_file.read_line(file) do
+            :eof ->
+              buffer
 
+            {:ok, line} ->
+              <<buffer::binary, line::binary>>
+          end
 
-  def run_file_buf(filename) do
+        index = :ets.update_counter(:brc, _key = :index, _increment_by = 1)
 
+        Agent.cast(:ets.lookup_element(:workers, rem(index, @pool_size), 2), Brc, :process_lines, [
+          buffer
+        ])
+
+        process_file(file, worker_pool)
+    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)
+    :ets.new(:workers, [:set, :public, :named_table])
 
-    file_stream = File.stream!(filename, 65_536, [:read_ahead])
-    # IO.inspect(file_stream)
+   worker_pool =
+     Enum.map(1..@pool_size, fn _ ->
+       Agent.start_link(fn -> %{} end) |> elem(1)
+     end)
 
-    # 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
+  Enum.each(Enum.zip(0..@pool_size - 1, worker_pool), fn w ->
+    :ets.insert(:workers, w)
+  end)
+
+    {:ok, file} = :prim_file.open(filename, [:binary, :read])
+
+    process_file(file, worker_pool)
 
     # 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)
+    pool_maps = Enum.map(worker_pool, fn pid -> Agent.get(pid, fn state -> state end, :infinity) end)
 
     # feeding all other maps into one, first one is the chosen one
     [head | tail] = pool_maps
@@ -85,6 +96,7 @@ defmodule Brc do
       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)
@@ -94,10 +106,10 @@ defmodule Brc do
 
     # output in brc format
     IO.puts("{#{Enum.join(sorted_strings, ", ")}}")
-
   end
 
   def main(args) do
+    IO.puts("Using prim_file")
     {uSec, :ok} =
       :timer.tc(fn ->
         run_file_buf(Enum.at(args, 0))