Precompile all possible cities into a workers module for quick lookups by function clause

src/brc.erl

@@ -2,13 +2,12 @@
 
 -export([run/1, find_cities/2]).
 
--include("hash.hrl").
-
 run([File]) ->
+  LkupTable = find_cities(atom_to_list(File), 1024*1024*2), %% 2MB should be enough :-)
+  generate_brc_workers:generate_and_compile(LkupTable),
   Workers = brc_workers:spawn_workers(erlang:system_info(logical_processors)),
   {Pid, Ref} = erlang:spawn_monitor(fun() -> exit({normal, brc_processor:start(Workers)}) end),
   brc_reader:start(File, Pid),
-  LkupTable = find_cities(atom_to_list(File), 1024*1024*2), %% 2MB should be enough :-)
   receive
     {'DOWN', Ref, process, Pid, {normal, Result}} ->
       [ exit(P, kill) || P <- Workers ],
@@ -27,48 +26,27 @@ round_back(IntFloat) ->
 find_cities(File, Size) ->
   {ok, FD} = prim_file:open(File, [read, binary, raw, read_ahead]),
   {ok, Data} = prim_file:read(FD, Size),
-  create_lookup_table(Data, #{}).
+  Cities = create_lookup_table(Data, []),
+  maps:from_list(lists:zip(lists:seq(1, length(Cities)), Cities)).
 
 create_lookup_table(Bin, State) ->
-  create_lookup_table(Bin, State, {<<>>, ?INIT}).
-
-create_lookup_table(<<C:8, $;:8, Rest/binary>>, State, {Raw, Acc}) ->
-  do_create_lookup_table(Rest, State, {<<Raw/binary, C>>, ?HASH(Acc, C)});
-create_lookup_table(<<C:16, $;:8, Rest/binary>> = Bin, State, {Raw, Acc}) ->
-  <<C1:8, C2:8, _/binary>> = Bin,
-  do_create_lookup_table(Rest, State, {<<Raw/binary, C1, C2>>, ?HASH(Acc, C)});
-create_lookup_table(<<C:24, $;:8, Rest/binary>> = Bin, State, {Raw, Acc}) ->
-  <<C1:8, C2:8, C3:8, _/binary>> = Bin,
-  do_create_lookup_table(Rest, State, {<<Raw/binary, C1, C2, C3>>, ?HASH(Acc, C)});
-create_lookup_table(<<C:24, Rest/binary>> = Bin, State, {Raw, Acc}) ->
-  <<C1:8, C2:8, C3:8, _/binary>> = Bin,
-  create_lookup_table(Rest, State, {<<Raw/binary, C1, C2, C3>>, ?HASH(Acc, C)});
+  Cities0 = create_lookup_table(Bin, State, <<>>),
+  SortedCities = lists:foldl(fun(City, Acc) -> maps:update_with(City, fun(V) -> V + 1 end, 1, Acc) end, #{}, Cities0),
+  {Cities, _} = lists:unzip(lists:reverse(lists:keysort(2, maps:to_list(SortedCities)))),
+  Cities.
+
+create_lookup_table(<<C:8, $;:8, Rest/binary>>, State, City) ->
+  do_create_lookup_table(Rest, State, <<City/binary, C>>);
+create_lookup_table(<<C:8, Rest/binary>>, State, City) ->
+  create_lookup_table(Rest, State, <<City/binary, C>>);
 create_lookup_table(_, State, _) ->
-State.
-
-do_create_lookup_table(<<_:40, $\n:8, Rest/binary>>, State, {CityRaw, City}) ->
-  create_lookup_table(Rest, State#{City => CityRaw}, {<<>>, ?INIT});
-do_create_lookup_table(<<_:32, $\n:8, Rest/binary>>, State, {CityRaw, City}) ->
-  create_lookup_table(Rest, State#{City => CityRaw}, {<<>>, ?INIT});
-do_create_lookup_table(<<_:24, $\n:8, Rest/binary>>, State, {CityRaw, City}) ->
-  create_lookup_table(Rest, State#{City => CityRaw}, {<<>>, ?INIT});
-do_create_lookup_table(_, State, {CityRaw, City}) ->
-  State#{City => CityRaw}.
-
--ifdef(EUNIT).
-
--include_lib("eunit/include/eunit.hrl").
-
-all_cities_hasheable_test() ->
-  {CitiesL, Map} = create_lookup_table_from_all_cities(),
-  Keys = maps:keys(Map),
-  io:format("Smallest hash: ~p Biggest hash: ~p~n", [lists:min(Keys), lists:max(Keys)]),
-  io:format("Missing: ~p~n", [CitiesL -- maps:values(Map)]),
-  ?assertEqual(length(CitiesL), maps:size(Map)).
-
-create_lookup_table_from_all_cities() ->
-  {ok, CitiesBin} = file:read_file(code:lib_dir(brc) ++ "/test/" ++ "cities_with_population_1000.txt"),
-  CitiesL = binary:split(CitiesBin, <<"\n">>, [global]),
-  {CitiesL, lists:foldl(fun(City, Acc) -> create_lookup_table(<<City/binary, ";">>, Acc, {<<>>, ?INIT}) end, #{}, CitiesL)}.
-
--endif.
+  State.
+
+do_create_lookup_table(<<_:40, $\n:8, Rest/binary>>, State, CityRaw) ->
+  create_lookup_table(Rest, [CityRaw | State], <<>>);
+do_create_lookup_table(<<_:32, $\n:8, Rest/binary>>, State, CityRaw) ->
+  create_lookup_table(Rest, [CityRaw | State], <<>>);
+do_create_lookup_table(<<_:24, $\n:8, Rest/binary>>, State, CityRaw) ->
+  create_lookup_table(Rest, [CityRaw | State], <<>>);
+do_create_lookup_table(_, State, CityRaw) ->
+  [CityRaw | State].

src/brc_reader.erl

@@ -2,29 +2,46 @@
 
 -export([start/2]).
 
--define(TWO_MB, 1024*1024*2).
+-define(ONE_MB, 1024*1024*10).
 
-start(File, Pid) ->
+start(File0, Pid) ->
   spawn_link(fun() ->
-               read_loop(atom_to_list(File), Pid, 0, undefined, ?TWO_MB)
+                File = atom_to_list(File0),
+                FileSize = filelib:file_size(File),
+                read_loop(File, Pid, 0, undefined, {0, ?ONE_MB, FileSize})
              end).
 
-read_loop(File, Pid, 0, undefined, Size) ->
+read_loop(File, Pid, 0, undefined, {N, ChunkSize, FileSize}) ->
+  Start = erlang:monotonic_time(),
+  put(start, Start),
   {ok, FD} = prim_file:open(File, [read, binary, raw, read_ahead]),
-  read_loop(File, Pid, 0, FD, Size);
-read_loop(File, Pid, Offset, FD, Size) ->
-  {NewOffset, OffsetSizes} = lists:foldl(fun(_, {O, Acc}) -> {O + Size, [{O, Size} | Acc]} end, {Offset, []}, lists:seq(1,5)),
-  {ok, Data} = prim_file:pread(FD, lists:reverse(OffsetSizes)),
-  case send_data(Data, Pid) of
+  read_loop(File, Pid, 0, FD, {N, ChunkSize, FileSize});
+read_loop(File, Pid, Offset, FD, {N, ChunkSize, FileSize}) when N =< 10 ->
+  case read_and_send(FD, Offset, ChunkSize, Pid) of
     eof ->
-      prim_file:close(FD),
-      Pid ! done;
-    ok ->
-      read_loop(File, Pid, NewOffset, FD, Size)
+      ok;
+    continue ->
+      read_loop(File, Pid, Offset + ChunkSize, FD, {N + 1, ChunkSize, FileSize})
+  end;
+read_loop(File, Pid, Offset, FD, {N, _ChunkSize, FileSize}) when N > 10 ->
+  NewChunkSize = FileSize div 100,
+  case read_and_send(FD, Offset, NewChunkSize, Pid) of
+    eof ->
+      ok;
+    continue ->
+      read_loop(File, Pid, Offset + NewChunkSize, FD, {N + 1, NewChunkSize, FileSize})
   end.
 
-send_data([], _Pid) -> ok;
-send_data([eof | _], _Pid) -> eof;
-send_data([Data | Rest], Pid) ->
-  Pid ! {chunk, Data},
-  send_data(Rest, Pid).
+read_and_send(FD, Offset, Size, Pid) ->
+  case prim_file:pread(FD, Offset, Size) of
+    eof ->
+      prim_file:close(FD),
+      Now = erlang:monotonic_time(),
+      Start = get(start),
+      io:format("Sending all data took: ~p~n", [(Now - Start) / 1000_000_000.0]),
+      Pid ! done,
+      eof;
+    {ok, Data} ->
+      Pid ! {chunk, Data},
+      continue
+  end.

src/generate_brc_workers.erl

@@ -0,0 +1,88 @@
+-module(generate_brc_workers).
+
+-export([generate_and_compile/1]).
+
+generate_and_compile(LkupTable) ->
+  Start = init_module(),
+  ProcessLines = process_lines(LkupTable),
+  End = remaining_parts_of_module(),
+  Full = <<Start/binary, ProcessLines/binary, End/binary>>,
+  {ok, Tokens, _} = erl_scan:string(binary_to_list(Full)),
+  {_, ParsedForms} = lists:foldl(fun(Token, {TokenAcc, FormsAcc}) ->
+                                   case Token of
+                                     {dot,_} ->
+                                       FormTokens = lists:reverse([Token | TokenAcc]),
+                                       {ok, Forms} = erl_parse:parse_form(FormTokens),
+                                       {[], [Forms | FormsAcc]};
+                                     Token ->
+                                       {[Token | TokenAcc], FormsAcc}
+                                   end
+                                 end, {[], []}, Tokens),
+  {ok, brc_workers, Bin} = compile:forms(lists:reverse(ParsedForms)),
+  {module, brc_workers} = code:load_binary(brc_workers, "nofile", Bin).
+
+init_module() ->
+  <<"-module(brc_workers).\n",
+    "\n",
+    "-export([spawn_workers/1]).\n",
+    "\n",
+    "spawn_workers(N) ->\n",
+    "  Options = [link,\n",
+    "             {min_heap_size, 1024*1024},\n",
+    "             {min_bin_vheap_size, 1024*1024},\n",
+    "             {max_heap_size, (1 bsl 59) -1}\n",
+    "            ],\n",
+    "  spawn_workers(N, Options, []).\n",
+    "\n",
+    "spawn_workers(0, _Options, Acc) -> Acc;\n",
+    "spawn_workers(N, Options, Acc) ->\n",
+    "  Pid = erlang:spawn_opt(fun() ->\n",
+    "                            worker()\n",
+    "                          end,\n",
+    "                          Options),\n",
+    "  spawn_workers(N - 1, Options, [Pid | Acc]).\n",
+    "\n",
+    "worker() ->\n",
+    "  process_flag(message_queue_data, off_heap),\n",
+    "  process_flag(priority, high),\n",
+    "  receive\n",
+    "    {chunk, Chunk} ->\n",
+    "      process_lines(Chunk),\n",
+    "      worker();\n",
+    "    {Parent, no_more_work} ->\n",
+    "      Parent ! {self(), worker_done, maps:from_list(get())}\n",
+    "    end.\n"
+    "\n">>.
+
+process_lines(LkupTable) ->
+  Bin0 =
+    lists:foldl(fun({Hash, City}, Acc) ->
+                   <<Acc/binary,
+                     "process_lines(<<\"", City/binary, "\", $;:8, Rest/binary>>) ->\n"
+                     "  do_process_line(Rest, ", (integer_to_binary(Hash))/binary, ");\n">>
+                end,
+                <<>>,
+                maps:to_list(LkupTable)),
+  <<Bin0/binary, "process_lines(<<>>) ->\n  ok.\n\n">>.
+
+remaining_parts_of_module() ->
+  <<"%% Very specialized float-parser for floats with a single fractional\n"
+    "%% digit, and returns the result as an integer * 10.\n"
+    "do_process_line(<<$-, A:8, B:8, $., C:8, $\\n:8, Rest/binary>>, City) ->\n"
+    "  add_to_state(Rest, City, -1 * ((A - $0) * 100 + (B - $0) * 10 + (C - $0)));\n"
+    "do_process_line(<<$-, B:8, $., C:8, $\\n:8, Rest/binary>>, City) ->\n"
+    "  add_to_state(Rest, City, -1 * ((B - $0) * 10 + (C - $0)));\n"
+    "do_process_line(<<A:8, B:8, $., C:8, $\\n:8, Rest/binary>>, City) ->\n"
+    "  add_to_state(Rest, City, (A - $0) * 100 + (B - $0) * 10 + (C - $0));\n"
+    "do_process_line(<<B:8, $., C:8, $\\n:8, Rest/binary>>, City) ->\n"
+    "  add_to_state(Rest, City, (B - $0) * 10 + (C - $0)).\n"
+    "\n"
+    "add_to_state(<<Rest/binary>>, City, Measurement) ->\n"
+    "  case get(City) of\n"
+    "    undefined ->\n"
+    "      put(City, {Measurement, Measurement, Measurement, 1}),\n"
+    "      process_lines(Rest);\n"
+    "    {Min, Max, MeasurementAcc, N} ->\n"
+    "      put(City, {min(Min, Measurement), max(Max, Measurement), MeasurementAcc + Measurement, N + 1}),\n"
+    "      process_lines(Rest)\n"
+    "  end.\n">>.

test/cities_with_population_1000.txt

@@ -87726,7 +87726,6 @@ Poschiavo
 Posedarje
 Poselki
 Posëlok Mar’ino
-Poselok Turisticheskogo pansionata ""Klyazminskoe vodohranilische
 Posëlok Voskhod
 Posen
 Poseritz

test/run_proper_SUITE.erl

@@ -79,6 +79,7 @@ run([File]) ->
   %% Just read all the cities into LkupTable. PropEr generates some really weird
   %% City;Float pairs and it's not representative of what the Java generator does
   LkupTable = brc:find_cities(atom_to_list(File), filelib:file_size(File)),
+  generate_brc_workers:generate_and_compile(LkupTable),
   Workers = brc_workers:spawn_workers(erlang:system_info(logical_processors)),
   {Pid, Ref} = erlang:spawn_monitor(fun() -> exit({normal, brc_processor:start(Workers)}) end),
   brc_reader:start(File, Pid),