Dirty read and Transactional write

dirty_read_and_transactional_write.erl

-module(dirty_read_and_transactional_write).
-author('[email protected]').

-export([run/3]).

-record(store, {key, value}).

run(TryCount, ProcessCount, ReadWriteCount) ->
  WriterPids = spawn_runner(ProcessCount),
  ReaderPids = spawn_runner(ProcessCount),
  mnesia:start(),
  do(TryCount, ReaderPids, WriterPids, ReadWriteCount),
  mnesia:stop(),
  stop_runner(ReaderPids ++ WriterPids),
  ok.

spawn_runner(Count) ->
  lists:map(
    fun (_) -> spawn_link(fun runner_process/0) end,
    lists:seq(1, Count)
  ).

stop_runner(Pids) ->
  lists:foreach(fun (Pid) -> Pid ! finish end, Pids).

runner_process() ->
  receive
    {run, Count, Function, ReturnPid} ->
      lists:foreach(fun (_) -> Function() end, lists:seq(1, Count)),
      ReturnPid ! {self(), ok};
    finish ->
      exit(normal);
    Other ->
      io:fwrite("Runner received unknown message: ~p~n", [Other])
  end,
  runner_process().

do(0, _, _, _) -> ok;
do(TryCount, ReaderPids, WriterPids, ReadWriteCount) ->
  io:fwrite("--<reduction:~p>--~n", [TryCount]),
  {ok, [RunTime, WallClock]} = read_and_write(ReaderPids, WriterPids, ReadWriteCount),
  io:fwrite("~p(~p)ms~n", [RunTime, WallClock]),
  do(TryCount-1, ReaderPids, WriterPids, ReadWriteCount).

run_process(Pids, Count, Function) ->
  lists:foreach(fun (Pid) ->
    Pid ! {run, Count, Function, self()}
  end, Pids),
  ok.

join_process(Pids) ->
  lists:foreach(fun (Pid) ->
    receive {Pid, _} -> ok end
  end, Pids),
  ok.

benchmark(TargetFunction) ->
  lists:foreach(fun statistics/1, [runtime, wall_clock]),
  Result = TargetFunction(),
  Times = lists:map(
    fun (Type) -> {_, T} = statistics(Type), T end,
    [runtime, wall_clock]
  ),
  {Result, Times}.

read_and_write(ReaderPids, WriterPids, Count) ->
  ok = init_mnesia(),
  benchmark(fun () ->
    run_process(WriterPids, Count, fun () ->
      mnesia:activity(transaction, fun () ->
        lists:foreach(fun (_) ->
          [{store, foo, N}] = mnesia:read(store, foo, sticky_write),
          ok = mnesia:s_write(#store{key = foo, value = N+1})
        end, [1, 2])
      end)
    end),
    run_process(ReaderPids, Count, fun () ->
      [{store, foo, N}] = mnesia:dirty_read(store, foo),
      if (N rem 2 =:= 0) -> true end
    end),
    join_process(WriterPids ++ ReaderPids)
  end).

init_mnesia() ->
  mnesia:delete_table(store),
  {atomic, ok} = mnesia:create_table(
    store,
    [
      {record_name, store},
      {type, set},
      {attributes, record_info(fields, store)}
    ]
  ),
  mnesia:activity(
    async_dirty,
    fun mnesia:write/1,
    [#store{key = foo, value = 0}]
  ).