Merge branch 'sf/erts_de_busy_limit' into dev

* sf/erts_de_busy_limit:
  Add flag-based setting for the distribution buffer busy limit

4 files changed, 36 insertions, 7 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();