Merge branch 'maint'

6 files changed, 118 insertions, 18 deletions

diff --git a/erts/emulator/beam/erl_async.c b/erts/emulator/beam/erl_async.c
index 054d1a48f6..e6d72f569b 100644
--- a/erts/emulator/beam/erl_async.c
+++ b/erts/emulator/beam/erl_async.c
@@ -583,6 +583,20 @@ int erts_async_ready_clean(void *varq, void *val)
 #endif
 
 /*
+** Generate a fair async key prom an ErlDrvPort
+** The port data gives a fair distribution grom port pointer
+** to unsigned integer - to be used in key for driver_async below.
+*/
+unsigned int driver_async_port_key(ErlDrvPort port)
+{
+    ErlDrvTermData td = driver_mk_port(port);
+    if (td == (ErlDrvTermData) NIL) {
+	return 0;
+    }
+    return (unsigned int) (UWord) internal_port_data(td);
+}
+
+/*
 ** Schedule async_invoke on a worker thread
 ** NOTE will be syncrounous when threads are unsupported
 ** return values:
diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h
index f43f5a766e..d2a11cce05 100644
--- a/erts/emulator/beam/erl_driver.h
+++ b/erts/emulator/beam/erl_driver.h
@@ -133,7 +133,7 @@ typedef struct {
 
 #define ERL_DRV_EXTENDED_MARKER		(0xfeeeeeed)
 #define ERL_DRV_EXTENDED_MAJOR_VERSION	2
-#define ERL_DRV_EXTENDED_MINOR_VERSION	1
+#define ERL_DRV_EXTENDED_MINOR_VERSION	2
 
 /*
  * The emulator will refuse to load a driver with different major
@@ -644,6 +644,8 @@ EXTERN int erl_drv_send_term(ErlDrvTermData port,
 			     int len);
 
 /* Async IO functions */
+EXTERN unsigned int driver_async_port_key(ErlDrvPort port);
+
 EXTERN long driver_async(ErlDrvPort ix,
 			 unsigned int* key,
 			 void (*async_invoke)(void*), 
diff --git a/erts/emulator/drivers/common/efile_drv.c b/erts/emulator/drivers/common/efile_drv.c
index 9e95325c24..c997fe1bf9 100644
--- a/erts/emulator/drivers/common/efile_drv.c
+++ b/erts/emulator/drivers/common/efile_drv.c
@@ -772,6 +772,7 @@ file_init(void)
     return 0;
 }
 
+
 /*********************************************************************
  * Driver entry point -> start
  */
@@ -788,7 +789,7 @@ file_start(ErlDrvPort port, char* command)
     }
     desc->fd = FILE_FD_INVALID;
     desc->port = port;
-    desc->key = (unsigned int) (UWord) port;
+    desc->key = driver_async_port_key(port);
     desc->flags = 0;
     desc->invoke = NULL;
     desc->d = NULL;
@@ -3133,25 +3134,25 @@ file_flush(ErlDrvData e) {
 
 /*********************************************************************
  * Driver entry point -> control
+ * Only debug functionality...
  */
 static ErlDrvSSizeT
 file_control(ErlDrvData e, unsigned int command, 
 	     char* buf, ErlDrvSizeT len, char **rbuf, ErlDrvSizeT rlen) {
-    /*
-     *  warning: variable ‘desc’ set but not used 
-     *  [-Wunused-but-set-variable]
-     *  ... no kidding ...
-     *
-     *
     file_descriptor *desc = (file_descriptor *)e;
     switch (command) {
+    case 'K' :
+	if (rlen < 4) {
+	    *rbuf = EF_ALLOC(4);
+	}
+	(*rbuf)[0] = ((desc->key) >> 24) & 0xFF;
+	(*rbuf)[1] = ((desc->key) >> 16) & 0xFF;
+	(*rbuf)[2] = ((desc->key) >> 8) & 0xFF;
+	(*rbuf)[3] = (desc->key) & 0xFF;
+	return 4;
     default:
 	return 0;
-    } 
-    ASSERT(0);
-    desc = NULL; 
-    */
-    return 0;
+    }
 }
 
 /*********************************************************************
diff --git a/erts/emulator/drivers/win32/win_efile.c b/erts/emulator/drivers/win32/win_efile.c
index be3d86a1d2..b36a103f8e 100644
--- a/erts/emulator/drivers/win32/win_efile.c
+++ b/erts/emulator/drivers/win32/win_efile.c
@@ -772,6 +772,7 @@ efile_may_openfile(Efile_error* errInfo, char *name) {
     DWORD attr;
 
     if ((attr = GetFileAttributesW(wname)) == INVALID_FILE_ATTRIBUTES) {
+	errno = ENOENT;
 	return check_error(-1, errInfo);
     }
 
diff --git a/erts/emulator/sys/win32/erl_win_dyn_driver.h b/erts/emulator/sys/win32/erl_win_dyn_driver.h
index a7c53c904d..b9a9838a36 100644
--- a/erts/emulator/sys/win32/erl_win_dyn_driver.h
+++ b/erts/emulator/sys/win32/erl_win_dyn_driver.h
@@ -80,6 +80,7 @@ WDD_TYPEDEF(int, erl_drv_output_term, (ErlDrvTermData, ErlDrvTermData*, int));
 WDD_TYPEDEF(int, driver_output_term, (ErlDrvPort, ErlDrvTermData*, int));
 WDD_TYPEDEF(int, erl_drv_send_term, (ErlDrvTermData, ErlDrvTermData, ErlDrvTermData*, int));
 WDD_TYPEDEF(int, driver_send_term, (ErlDrvPort, ErlDrvTermData, ErlDrvTermData*, int));
+WDD_TYPEDEF(unsigned int, driver_async_port_key, (ErlDrvPort));
 WDD_TYPEDEF(long, driver_async, (ErlDrvPort,unsigned int*,void (*)(void*),void*,void (*)(void*)));
 WDD_TYPEDEF(int, driver_async_cancel, (unsigned int));
 WDD_TYPEDEF(int, driver_lock_driver, (ErlDrvPort));
@@ -197,6 +198,7 @@ typedef struct {
     WDD_FTYPE(driver_output_term) *driver_output_term;
     WDD_FTYPE(erl_drv_send_term) *erl_drv_send_term;
     WDD_FTYPE(driver_send_term) *driver_send_term;
+    WDD_FTYPE(driver_async_port_key) *driver_async_port_key;
     WDD_FTYPE(driver_async) *driver_async;
     WDD_FTYPE(driver_async_cancel) *driver_async_cancel;
     WDD_FTYPE(driver_lock_driver) *driver_lock_driver;
@@ -308,6 +310,7 @@ extern TWinDynDriverCallbacks WinDynDriverCallbacks;
 #define driver_output_term (WinDynDriverCallbacks.driver_output_term)
 #define erl_drv_send_term (WinDynDriverCallbacks.erl_drv_send_term)
 #define driver_send_term (WinDynDriverCallbacks.driver_send_term)
+#define driver_async_port_key (WinDynDriverCallbacks.driver_async_port_key)
 #define driver_async (WinDynDriverCallbacks.driver_async)
 #define driver_async_cancel (WinDynDriverCallbacks.driver_async_cancel)
 #define driver_lock_driver (WinDynDriverCallbacks.driver_lock_driver)
@@ -443,6 +446,7 @@ do {				                        \
 ((W).driver_output_term) = driver_output_term;		\
 ((W).erl_drv_send_term) = erl_drv_send_term;		\
 ((W).driver_send_term) = driver_send_term;		\
+((W).driver_async_port_key) = driver_async_port_key;	\
 ((W).driver_async) = driver_async;			\
 ((W).driver_async_cancel) = driver_async_cancel;	\
 ((W).driver_lock_driver) = driver_lock_driver;	       	\
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.