From 6d1f1d43aa0a9e0a2cb275ef760339c49518fc69 Mon Sep 17 00:00:00 2001
From: Patrik Nyblom <pan@erlang.org>
Date: Fri, 16 Aug 2013 17:14:25 +0200
Subject: Create better distribution of files over async threads

The actual port id is used to create a key from the
pointer value which is the ErlDrvPort. To do this
a new driver api function driver_async_port_key is
added and the driver API minor version is updated.

The documentation is updated and the faulty description of
how to spread ports over async threads is updated to
use the new API.

Testcase also added.
---
 erts/emulator/test/efile_SUITE.erl | 88 +++++++++++++++++++++++++++++++++++---
 1 file changed, 83 insertions(+), 5 deletions(-)

(limited to 'erts/emulator/test')

diff --git a/erts/emulator/test/efile_SUITE.erl b/erts/emulator/test/efile_SUITE.erl
index 65367eab98..f79bb761d1 100644
--- a/erts/emulator/test/efile_SUITE.erl
+++ b/erts/emulator/test/efile_SUITE.erl
@@ -19,16 +19,16 @@
 -module(efile_SUITE).
 -export([all/0, suite/0,groups/0,init_per_suite/1, end_per_suite/1, 
 	 init_per_group/2,end_per_group/2]).
--export([iter_max_files/1]).
+-export([iter_max_files/1, async_dist/1]).
 
--export([do_iter_max_files/2]).
+-export([do_iter_max_files/2, do_async_dist/1]).
 
 -include_lib("test_server/include/test_server.hrl").
 
 suite() -> [{ct_hooks,[ts_install_cth]}].
 
 all() -> 
-    [iter_max_files].
+    [iter_max_files, async_dist].
 
 groups() -> 
     [].
@@ -45,6 +45,84 @@ init_per_group(_GroupName, Config) ->
 end_per_group(_GroupName, Config) ->
     Config.
 
+do_async_dist(Dir) ->
+    X = 100,
+    AT = erlang:system_info(thread_pool_size),
+    Keys = file_keys(Dir,AT*X,[],[]),
+    Tab = ets:new(x,[ordered_set]),
+    [ ets:insert(Tab,{N,0}) || N <- lists:seq(0,AT-1) ],
+    [ ets:update_counter(Tab,(N rem AT),1) || N <- Keys ],
+    Res = [ V || {_,V} <- ets:tab2list(Tab) ],
+    ets:delete(Tab),
+    {Res, sdev(Res)/X}.
+
+sdev(List) ->
+    Len = length(List),
+    Mean = lists:sum(List)/Len,
+    math:sqrt(lists:sum([ (X - Mean) * (X - Mean) || X <- List ]) / Len).
+
+file_keys(_,0,FdList,FnList) ->
+    [ file:close(FD) || FD <- FdList ],
+    [ file:delete(FN) || FN <- FnList ],
+    [];
+file_keys(Dir,Num,FdList,FnList) ->
+    Name = "dummy"++integer_to_list(Num),
+    FN = filename:join([Dir,Name]),
+    case file:open(FN,[write,raw]) of
+	{ok,FD} ->
+	    {file_descriptor,prim_file,{Port,_}} = FD,
+	    <<X:32/integer-big>> = 
+		iolist_to_binary(erlang:port_control(Port,$K,[])),
+	    [X | file_keys(Dir,Num-1,[FD|FdList],[FN|FnList])];
+	{error,_} ->
+	    % Try freeing up FD's if there are any
+	    case FdList of 
+		[] ->
+		    exit({cannot_open_file,FN});
+		_ ->
+		    [ file:close(FD) || FD <- FdList ],
+		    [ file:delete(F) || F <- FnList ],
+		    file_keys(Dir,Num,[],[])
+	    end
+    end.
+
+async_dist(doc) ->
+    "Check that the distribution of files over async threads is fair";
+async_dist(Config) when is_list(Config) ->
+    DataDir = ?config(data_dir,Config),
+    TestFile = filename:join(DataDir, "existing_file"),
+    Dir = filename:dirname(code:which(?MODULE)),
+    AsyncSizes = [7,10,100,255,256,64,63,65],
+    Max = 0.5,
+
+    lists:foreach(fun(Size) ->
+			  {ok,Node} = 
+			      test_server:start_node
+				(test_iter_max_files,slave,
+				 [{args,
+				   "+A "++integer_to_list(Size)++
+				   " -pa " ++ Dir}]),
+			  {Distr,SD} = rpc:call(Node,?MODULE,do_async_dist,
+						[DataDir]),
+			  test_server:stop_node(Node),
+			  if
+			      SD > Max ->
+				  io:format("Bad async queue distribution for "
+					    "~p async threads:~n"
+					    "    Standard deviation is ~p~n"
+					    "    Key distribution:~n    ~lp~n",
+					    [Size,SD,Distr]),
+				  exit({bad_async_dist,Size,SD,Distr});
+			      true ->
+				  io:format("OK async queue distribution for "
+					    "~p async threads:~n"
+					    "    Standard deviation is ~p~n"
+					    "    Key distribution:~n    ~lp~n",
+					    [Size,SD,Distr]),
+				  ok
+			  end
+		  end, AsyncSizes),
+    ok.
 
 %%
 %% Open as many files as possible. Do this several times and check 
@@ -98,7 +176,7 @@ open_files(Name) ->
     ?line case file:open(Name, [read,raw]) of
 	      {ok, Fd} ->
 		  [Fd| open_files(Name)];
-	      {error, Reason} ->
-%		  io:format("Error reason: ~p", [Reason]),
+	      {error, _Reason} ->
+%		  io:format("Error reason: ~p", [_Reason]),
 		  []
 	  end.
-- 
cgit v1.2.3