aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator
diff options
context:
space:
mode:
authorRickard Green <[email protected]>2010-11-02 11:52:22 +0100
committerRickard Green <[email protected]>2010-11-02 11:52:22 +0100
commit158ed71a5ddc5050809723a214a8d8c841022871 (patch)
tree13f2cfedca8745942abf0572dcbb9044e6ab3df1 /erts/emulator
parent96419fe8c795656c46ba50b1aac92c98acc73a0b (diff)
parent8faf1746ece60fc5fa634e5fd16e98df1ef7f3ba (diff)
downloadotp-158ed71a5ddc5050809723a214a8d8c841022871.tar.gz
otp-158ed71a5ddc5050809723a214a8d8c841022871.tar.bz2
otp-158ed71a5ddc5050809723a214a8d8c841022871.zip
Merge branch 'sf/erts_de_busy_limit' into dev
* sf/erts_de_busy_limit: Add flag-based setting for the distribution buffer busy limit
Diffstat (limited to 'erts/emulator')
-rw-r--r--erts/emulator/beam/dist.c8
-rw-r--r--erts/emulator/beam/dist.h3
-rw-r--r--erts/emulator/beam/erl_bif_info.c7
-rw-r--r--erts/emulator/beam/erl_init.c25
-rw-r--r--erts/emulator/test/distribution_SUITE.erl93
5 files changed, 127 insertions, 9 deletions
diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c
index 16b6aeac3f..4497e17d79 100644
--- a/erts/emulator/beam/dist.c
+++ b/erts/emulator/beam/dist.c
@@ -97,6 +97,8 @@ dist_msg_dbg(ErtsDistExternal *edep, char *what, byte *buf, int sz)
#define PASS_THROUGH 'p' /* This code should go */
int erts_is_alive; /* System must be blocked on change */
+int erts_dist_buf_busy_limit;
+
/* distribution trap functions */
Export* dsend2_trap = NULL;
@@ -1453,8 +1455,6 @@ int erts_net_message(Port *prt,
return -1;
}
-#define ERTS_DE_BUSY_LIMIT (128*1024)
-
static int
dsig_send(ErtsDSigData *dsdp, Eterm ctl, Eterm msg, int force_busy)
{
@@ -1540,7 +1540,7 @@ dsig_send(ErtsDSigData *dsdp, Eterm ctl, Eterm msg, int force_busy)
ErtsProcList *plp = NULL;
erts_smp_spin_lock(&dep->qlock);
dep->qsize += size_obuf(obuf);
- if (dep->qsize >= ERTS_DE_BUSY_LIMIT)
+ if (dep->qsize >= erts_dist_buf_busy_limit)
dep->qflgs |= ERTS_DE_QFLG_BUSY;
if (!force_busy && (dep->qflgs & ERTS_DE_QFLG_BUSY)) {
erts_smp_spin_unlock(&dep->qlock);
@@ -1911,7 +1911,7 @@ erts_dist_command(Port *prt, int reds_limit)
ASSERT(dep->qsize >= obufsize);
dep->qsize -= obufsize;
obufsize = 0;
- if (de_busy && !prt_busy && dep->qsize < ERTS_DE_BUSY_LIMIT) {
+ if (de_busy && !prt_busy && dep->qsize < erts_dist_buf_busy_limit) {
ErtsProcList *suspendees;
int resumed;
suspendees = get_suspended_on_de(dep, ERTS_DE_QFLG_BUSY);
diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h
index fa19c7fb45..28cdd05c3c 100644
--- a/erts/emulator/beam/dist.h
+++ b/erts/emulator/beam/dist.h
@@ -99,7 +99,8 @@ typedef struct {
#define ERTS_DE_IS_CONNECTED(DEP) \
(!ERTS_DE_IS_NOT_CONNECTED((DEP)))
-
+#define ERTS_DE_BUSY_LIMIT (128*1024)
+extern int erts_dist_buf_busy_limit;
extern int erts_is_alive;
/*
diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index 40d8dc097c..801263ec26 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -2533,6 +2533,13 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1)
BIF_RET(erts_nif_taints(BIF_P));
} else if (ERTS_IS_ATOM_STR("reader_groups_map", BIF_ARG_1)) {
BIF_RET(erts_get_reader_groups_map(BIF_P));
+ } else if (ERTS_IS_ATOM_STR("dist_buf_busy_limit", BIF_ARG_1)) {
+ Uint hsz = 0;
+
+ (void) erts_bld_uint(NULL, &hsz, erts_dist_buf_busy_limit);
+ hp = hsz ? HAlloc(BIF_P, hsz) : NULL;
+ res = erts_bld_uint(&hp, NULL, erts_dist_buf_busy_limit);
+ BIF_RET(res);
}
BIF_ERROR(BIF_P, BADARG);
diff --git a/erts/emulator/beam/erl_init.c b/erts/emulator/beam/erl_init.c
index 4ae656a3ad..36fefc5cba 100644
--- a/erts/emulator/beam/erl_init.c
+++ b/erts/emulator/beam/erl_init.c
@@ -535,7 +535,8 @@ void erts_usage(void)
erts_fprintf(stderr, "-W<i|w> set error logger warnings mapping,\n");
erts_fprintf(stderr, " see error_logger documentation for details\n");
-
+ erts_fprintf(stderr, "-zdbbl size set the distribution buffer busy limit in kilobytes\n");
+ erts_fprintf(stderr, " valid range is [1-%d]\n", INT_MAX/1024);
erts_fprintf(stderr, "\n");
erts_fprintf(stderr, "Note that if the emulator is started with erlexec (typically\n");
erts_fprintf(stderr, "from the erl script), these flags should be specified with +.\n");
@@ -818,7 +819,7 @@ early_init(int *argc, char **argv) /*
erl_sys_args(argc, argv);
erts_ets_realloc_always_moves = 0;
-
+ erts_dist_buf_busy_limit = ERTS_DE_BUSY_LIMIT;
}
#ifndef ERTS_SMP
@@ -1346,6 +1347,26 @@ erl_start(int argc, char **argv)
}
break;
+ case 'z': {
+ char *sub_param = argv[i]+2;
+ int new_limit;
+
+ if (has_prefix("dbbl", sub_param)) {
+ arg = get_arg(sub_param+4, argv[i+1], &i);
+ new_limit = atoi(arg);
+ if (new_limit < 1 || INT_MAX/1024 < new_limit) {
+ erts_fprintf(stderr, "Invalid dbbl limit: %d\n", new_limit);
+ erts_usage();
+ } else {
+ erts_dist_buf_busy_limit = new_limit*1024;
+ }
+ } else {
+ erts_fprintf(stderr, "bad -z option %s\n", argv[i]);
+ erts_usage();
+ }
+ break;
+ }
+
default:
erts_fprintf(stderr, "%s unknown flag %s\n", argv[0], argv[i]);
erts_usage();
diff --git a/erts/emulator/test/distribution_SUITE.erl b/erts/emulator/test/distribution_SUITE.erl
index 7c19274696..79252d0593 100644
--- a/erts/emulator/test/distribution_SUITE.erl
+++ b/erts/emulator/test/distribution_SUITE.erl
@@ -27,6 +27,7 @@
-export([all/1,
ping/1, bulk_send/1, bulk_send_small/1,
bulk_send_big/1,
+ bulk_send_bigbig/1,
local_send/1, local_send_small/1, local_send_big/1,
local_send_legal/1, link_to_busy/1, exit_to_busy/1,
lost_exit/1, link_to_dead/1, link_to_dead_new_node/1,
@@ -50,7 +51,8 @@
-export([sender/3, receiver2/2, dummy_waiter/0, dead_process/0,
roundtrip/1, bounce/1, do_dist_auto_connect/1, inet_rpc_server/1,
dist_parallel_sender/3, dist_parallel_receiver/0,
- dist_evil_parallel_receiver/0]).
+ dist_evil_parallel_receiver/0,
+ sendersender/4, sendersender2/4]).
all(suite) -> [
ping, bulk_send, local_send, link_to_busy, exit_to_busy,
@@ -121,7 +123,7 @@ bulk_send(doc) ->
"the time. This tests that a process that is suspended on a ",
"busy port will eventually be resumed."];
bulk_send(suite) ->
- [bulk_send_small, bulk_send_big].
+ [bulk_send_small, bulk_send_big, bulk_send_bigbig].
bulk_send_small(Config) when is_list(Config) ->
?line bulk_send(64, 32).
@@ -129,6 +131,9 @@ bulk_send_small(Config) when is_list(Config) ->
bulk_send_big(Config) when is_list(Config) ->
?line bulk_send(32, 64).
+bulk_send_bigbig(Config) when is_list(Config) ->
+ ?line bulk_sendsend(32*5, 4).
+
bulk_send(Terms, BinSize) ->
?line Dog = test_server:timetrap(test_server:seconds(30)),
@@ -145,6 +150,53 @@ bulk_send(Terms, BinSize) ->
?line test_server:timetrap_cancel(Dog),
{comment, integer_to_list(trunc(Size/1024/Elapsed+0.5)) ++ " K/s"}.
+bulk_sendsend(Terms, BinSize) ->
+ {Rate1, MonitorCount1} = bulk_sendsend2(Terms, BinSize, 5),
+ {Rate2, MonitorCount2} = bulk_sendsend2(Terms, BinSize, 995),
+ Ratio = if MonitorCount2 == 0 -> MonitorCount1 / 1.0;
+ true -> MonitorCount1 / MonitorCount2
+ end,
+ %% A somewhat arbitrary ratio, but hopefully one that will accomodate
+ %% a wide range of CPU speeds.
+ true = (Ratio > 8.0),
+ {comment,
+ integer_to_list(Rate1) ++ " K/s, " ++
+ integer_to_list(Rate2) ++ " K/s, " ++
+ integer_to_list(MonitorCount1) ++ " monitor msgs, " ++
+ integer_to_list(MonitorCount2) ++ " monitor msgs, " ++
+ float_to_list(Ratio) ++ " monitor ratio"}.
+
+bulk_sendsend2(Terms, BinSize, BusyBufSize) ->
+ ?line Dog = test_server:timetrap(test_server:seconds(30)),
+
+ ?line io:format("Sending ~w binaries, each of size ~w K",
+ [Terms, BinSize]),
+ ?line {ok, NodeRecv} = start_node(bulk_receiver),
+ ?line Recv = spawn(NodeRecv, erlang, apply, [fun receiver/2, [0, 0]]),
+ ?line Bin = list_to_binary(lists:duplicate(BinSize*1024, 253)),
+ ?line Size = Terms*size(Bin),
+
+ %% SLF LEFT OFF HERE.
+ %% When the caller uses small hunks, like 4k via
+ %% bulk_sendsend(32*5, 4), then (on my laptop at least), we get
+ %% zero monitor messages. But if we use "+zdbbl 5", then we
+ %% get a lot of monitor messages. So, if we can count up the
+ %% total number of monitor messages that we get when running both
+ %% default busy size and "+zdbbl 5", and if the 5 case gets
+ %% "many many more" monitor messages, then we know we're working.
+
+ ?line {ok, NodeSend} = start_node(bulk_sender, "+zdbbl " ++ integer_to_list(BusyBufSize)),
+ ?line _Send = spawn(NodeSend, erlang, apply, [fun sendersender/4, [self(), Recv, Bin, Terms]]),
+ ?line {Elapsed, {TermsN, SizeN}, MonitorCount} =
+ receive {sendersender, BigRes} ->
+ BigRes
+ end,
+ ?line stop_node(NodeRecv),
+ ?line stop_node(NodeSend),
+
+ ?line test_server:timetrap_cancel(Dog),
+ {trunc(SizeN/1024/Elapsed+0.5), MonitorCount}.
+
sender(To, _Bin, 0) ->
To ! {done, self()},
receive
@@ -155,6 +207,43 @@ sender(To, Bin, Left) ->
To ! {term, Bin},
sender(To, Bin, Left-1).
+%% Sender process to be run on a slave node
+
+sendersender(Parent, To, Bin, Left) ->
+ erlang:system_monitor(self(), [busy_dist_port]),
+ [spawn(fun() -> sendersender2(To, Bin, Left, false) end) ||
+ _ <- lists:seq(1,1)],
+ {USec, {Res, MonitorCount}} =
+ timer:tc(?MODULE, sendersender2, [To, Bin, Left, true]),
+ Parent ! {sendersender, {USec/1000000, Res, MonitorCount}}.
+
+sendersender2(To, Bin, Left, SendDone) ->
+ sendersender3(To, Bin, Left, SendDone, 0).
+
+sendersender3(To, _Bin, 0, SendDone, MonitorCount) ->
+ if SendDone ->
+ To ! {done, self()};
+ true ->
+ ok
+ end,
+ receive
+ {monitor, _Pid, _Type, _Info} = M ->
+ sendersender3(To, _Bin, 0, SendDone, MonitorCount + 1)
+ after 0 ->
+ if SendDone ->
+ receive
+ Any when is_tuple(Any), size(Any) == 2 ->
+ {Any, MonitorCount}
+ end;
+ true ->
+ exit(normal)
+ end
+ end;
+sendersender3(To, Bin, Left, SendDone, MonitorCount) ->
+ To ! {term, Bin},
+ %%timer:sleep(50),
+ sendersender3(To, Bin, Left-1, SendDone, MonitorCount).
+
%% Receiver process to be run on a slave node.
receiver(Terms, Size) ->